branches/zip:

  1. We add a vector of locks to trx_t. This array contains the autoinc
  locks granted to a transaction. There is one per table.

  2. We enforce releasing of these locks in the reverse order from the
  one in which they are acquired. The assumption is that since the
  AUTOINC locks are statement level locks. Nested statements introduced
  by triggers are stacked it should hold.

There was some cleanup done to the vector code too by adding const and
some new functions. Rename dict_table_t::auto_inc_lock to autoinc_lock.

Fix Bug#26316 Triggers create duplicate entries on auto-increment columns
rb://22

dict/dict0mem.c

@@ -58,7 +58,7 @@ dict_mem_table_create(
 	table->cols = mem_heap_alloc(heap, (n_cols + DATA_N_SYS_COLS)
 				     * sizeof(dict_col_t));
 
-	table->auto_inc_lock = mem_heap_alloc(heap, lock_get_size());
+	table->autoinc_lock = mem_heap_alloc(heap, lock_get_size());
 
 	mutex_create(&table->autoinc_mutex, SYNC_DICT_AUTOINC_MUTEX);
 

include/dict0mem.h

@@ -24,6 +24,7 @@ Created 1/8/1996 Heikki Tuuri
 #include "lock0types.h"
 #include "hash0hash.h"
 #include "que0types.h"
+#include "trx0types.h"
 
 /* Type flags of an index: OR'ing of the flags is allowed to define a
 combination of types */
@@ -381,13 +382,6 @@ struct dict_table_struct{
 				on the table: we cannot drop the table while
 				there are foreign key checks running on
 				it! */
-	lock_t*		auto_inc_lock;/* a buffer for an auto-inc lock
-				for this table: we allocate the memory here
-				so that individual transactions can get it
-				and release it without a need to allocate
-				space from the lock heap of the trx:
-				otherwise the lock heap would grow rapidly
-				if we do a large insert from a select */
 	dulint		query_cache_inv_trx_id;
 				/* transactions whose trx id < than this
 				number are not allowed to store to the MySQL
@@ -438,12 +432,33 @@ struct dict_table_struct{
 				any latch, because this is only used for
 				heuristics */
 	/*----------------------*/
+				/* The following fields are used by the
+				AUTOINC code.  The actual collection of
+				tables locked during AUTOINC read/write is
+				kept in trx_t. In order to quickly determine
+				whether a transaction has locked the AUTOINC
+				lock we keep a pointer to the transaction
+				here in the autoinc_trx variable. This is to
+				avoid acquiring the kernel mutex and scanning
+				the vector in trx_t.
+
+				When an AUTOINC lock has to wait, the
+				corresponding lock instance is created on
+				the trx lock heap rather than use the
+				pre-allocated instance in autoinc_lock below.*/
+	lock_t*		autoinc_lock;
+				/* a buffer for an AUTOINC lock
+				for this table: we allocate the memory here
+				so that individual transactions can get it
+				and release it without a need to allocate
+				space from the lock heap of the trx:
+				otherwise the lock heap would grow rapidly
+				if we do a large insert from a select */
 	mutex_t		autoinc_mutex;
 				/* mutex protecting the autoincrement
 				counter */
 	ib_uint64_t	autoinc;/* autoinc counter value to give to the
 				next inserted row */
-	/*----------------------*/
 	ulong		n_waiting_or_granted_auto_inc_locks;
 				/* This counter is used to track the number
 				of granted and pending autoinc locks on this
@@ -453,6 +468,9 @@ struct dict_table_struct{
 				acquired the AUTOINC lock or not. Of course
 				only one transaction can be granted the
 				lock but there can be multiple waiters. */
+	const trx_t*		autoinc_trx;
+				/* The transaction that currently holds the
+				the AUTOINC lock on this table. */
 	/*----------------------*/
 
 #ifdef UNIV_DEBUG

include/lock0lock.h

@@ -18,6 +18,7 @@ Created 5/7/1996 Heikki Tuuri
 #include "lock0types.h"
 #include "read0types.h"
 #include "hash0hash.h"
+#include "ut0vec.h"
 
 #ifdef UNIV_DEBUG
 extern ibool	lock_print_waits;
@@ -490,14 +491,6 @@ lock_table_unlock(
 /*==============*/
 	lock_t*	lock);	/* in: lock */
 /*************************************************************************
-Releases an auto-inc lock a transaction possibly has on a table.
-Releases possible other transactions waiting for this lock. */
-UNIV_INTERN
-void
-lock_table_unlock_auto_inc(
-/*=======================*/
-	trx_t*	trx);	/* in: transaction */
-/*************************************************************************
 Releases transaction locks, and releases possible other transactions waiting
 because of these locks. */
 UNIV_INTERN
@@ -653,6 +646,13 @@ ulint
 lock_number_of_rows_locked(
 /*=======================*/
 	trx_t*	trx);	/* in: transaction */
+/***********************************************************************
+Release all the transaction's autoinc locks. */
+UNIV_INTERN
+void
+lock_release_autoinc_locks(
+/*=======================*/
+	trx_t*		trx);		/* in/out: transaction */
 
 /***********************************************************************
 Gets the type of a lock. Non-inline version for using outside of the

include/row0mysql.h

@@ -163,12 +163,12 @@ row_update_prebuilt_trx(
 					handle */
 	trx_t*		trx);		/* in: transaction handle */
 /*************************************************************************
-Unlocks an AUTO_INC type lock possibly reserved by trx. */
+Unlocks AUTO_INC type locks that were possibly reserved by a trx. */
 UNIV_INTERN
 void
 row_unlock_table_autoinc_for_mysql(
 /*===============================*/
-	trx_t*	trx);	/* in: transaction */
+	trx_t*	trx);			/* in/out: transaction */
 /*************************************************************************
 Sets an AUTO_INC type lock on the table mentioned in prebuilt. The
 AUTO_INC lock gives exclusive access to the auto-inc counter of the

include/trx0trx.h

@@ -18,6 +18,7 @@ Created 3/26/1996 Heikki Tuuri
 #include "read0types.h"
 #include "dict0types.h"
 #include "trx0xa.h"
+#include "ut0vec.h"
 
 /* Dummy session used currently in MySQL interface */
 extern sess_t*	trx_dummy_sess;
@@ -601,9 +602,6 @@ struct trx_struct{
 					to srv_conc_innodb_enter, if the value
 					here is > 0, we decrement this by 1 */
 	/*------------------------------*/
-	lock_t*		auto_inc_lock;	/* possible auto-inc lock reserved by
-					the transaction; note that it is also
-					in the lock list trx_locks */
 	dict_index_t*	new_rec_locks[2];/* these are normally NULL; if
 					srv_locks_unsafe_for_binlog is TRUE
 					or session is using READ COMMITTED
@@ -735,9 +733,15 @@ struct trx_struct{
 	trx_undo_arr_t*	undo_no_arr;	/* array of undo numbers of undo log
 					records which are currently processed
 					by a rollback operation */
+	/*------------------------------*/
 	ulint		n_autoinc_rows;	/* no. of AUTO-INC rows required for
 					an SQL statement. This is useful for
 					multi-row INSERTs */
+	ib_vector_t*    autoinc_locks;  /* AUTOINC locks held by this
+					transaction. Note that these are
+					also in the lock list trx_locks. This
+					vector needs to be freed explicitly
+					when the trx_t instance is desrtoyed */
 	/*------------------------------*/
 	char detailed_error[256];	/* detailed error message for last
 					error, or empty. */

include/univ.i

@@ -165,6 +165,8 @@ operations (very slow); also UNIV_DEBUG must be defined */
 #define UNIV_BTR_DEBUG				/* check B-tree links */
 #define UNIV_LIGHT_MEM_DEBUG			/* light memory debugging */
 
+#define UNIV_DEBUG
+
 #ifdef HAVE_purify
 /* The following sets all new allocated memory to zero before use:
 this can be used to eliminate unnecessary Purify warnings, but note that

include/ut0vec.h

@@ -46,7 +46,16 @@ ulint
 ib_vector_size(
 /*===========*/
 				/* out: number of elements in vector */
-	ib_vector_t*	vec);	/* in: vector */
+	const ib_vector_t*	vec);	/* in: vector */
+
+/********************************************************************
+Test whether a vector is empty or not. */
+UNIV_INLINE
+ibool
+ib_vector_is_empty(
+/*===============*/
+				/* out: TRUE if empty */
+	const ib_vector_t*	vec);	/* in: vector */
 
 /********************************************************************
 Get the n'th element. */
@@ -58,6 +67,23 @@ ib_vector_get(
 	ib_vector_t*	vec,	/* in: vector */
 	ulint		n);	/* in: element index to get */
 
+/********************************************************************
+Remove the last element from the vector. */
+UNIV_INLINE
+void*
+ib_vector_pop(
+/*==========*/
+	ib_vector_t*	vec);	/* in: vector */
+
+/********************************************************************
+Free the underlying heap of the vector. Note that vec is invalid
+after this call. */
+UNIV_INLINE
+void
+ib_vector_free(
+/*===========*/
+	ib_vector_t*	vec);	/* in,own: vector */
+
 /* See comment at beginning of file. */
 struct ib_vector_struct {
 	mem_heap_t*	heap;	/* heap */

include/ut0vec.ic

@@ -5,7 +5,7 @@ ulint
 ib_vector_size(
 /*===========*/
 				/* out: number of elements in vector */
-	ib_vector_t*	vec)	/* in: vector */
+	const ib_vector_t*	vec)	/* in: vector */
 {
 	return(vec->used);
 }
@@ -24,3 +24,47 @@ ib_vector_get(
 
 	return(vec->data[n]);
 }
+
+/********************************************************************
+Remove the last element from the vector. */
+UNIV_INLINE
+void*
+ib_vector_pop(
+/*==========*/
+				/* out: last vector element */
+	ib_vector_t*    vec)    /* in/out: vector */
+{
+	void*           elem;
+
+	ut_a(vec->used > 0);
+	--vec->used;
+	elem = vec->data[vec->used];
+
+	ut_d(vec->data[vec->used] = NULL);
+	UNIV_MEM_INVALID(&vec->data[vec->used], sizeof(*vec->data));
+
+	return(elem);
+}
+
+/********************************************************************
+Free the underlying heap of the vector. Note that vec is invalid
+after this call. */
+UNIV_INLINE
+void
+ib_vector_free(
+/*===========*/
+	ib_vector_t*    vec)    /* in, own: vector */
+{
+	mem_heap_free(vec->heap);
+}
+
+/********************************************************************
+Test whether a vector is empty or not. */
+UNIV_INLINE
+ibool
+ib_vector_is_empty(
+/*===============*/			/* out: TRUE if empty else FALSE */
+	const ib_vector_t*	vec)	/* in vector to test */
+{
+	return(ib_vector_size(vec) == 0);
+}

lock/lock0lock.c

@@ -2166,24 +2166,25 @@ static
 void
 lock_grant(
 /*=======*/
-	lock_t*	lock)	/* in: waiting lock request */
+	lock_t*	lock)	/* in/out: waiting lock request */
 {
 	ut_ad(mutex_own(&kernel_mutex));
 
 	lock_reset_lock_and_trx_wait(lock);
 
 	if (lock_get_mode(lock) == LOCK_AUTO_INC) {
+		trx_t*		trx = lock->trx;
+		dict_table_t*	table = lock->un_member.tab_lock.table;
 
-		if (lock->trx->auto_inc_lock != NULL) {
+		if (table->autoinc_trx == trx) {
 			fprintf(stderr,
 				"InnoDB: Error: trx already had"
 				" an AUTO-INC lock!\n");
-		}
-
-		/* Store pointer to lock to trx so that we know to
-		release it at the end of the SQL statement */
+		} else {
+			table->autoinc_trx = trx;
 
-		lock->trx->auto_inc_lock = lock;
+			ib_vector_push(trx->autoinc_locks, lock);
+		}
 	}
 
 #ifdef UNIV_DEBUG
@@ -3531,15 +3532,16 @@ lock_table_create(
 		++table->n_waiting_or_granted_auto_inc_locks;
 	}
 
+	/* For AUTOINC locking we reuse the lock instance only if
+	there is no wait involved else we allocate the waiting lock
+	from the transaction lock heap. */
 	if (type_mode == LOCK_AUTO_INC) {
-		/* Only one trx can have the lock on the table
-		at a time: we may use the memory preallocated
-		to the table object */
 
-		lock = table->auto_inc_lock;
+		lock = table->autoinc_lock;
+
+		table->autoinc_trx = trx;
 
-		ut_a(trx->auto_inc_lock == NULL);
-		trx->auto_inc_lock = lock;
+		ib_vector_push(trx->autoinc_locks, lock);
 	} else {
 		lock = mem_heap_alloc(trx->lock_heap, sizeof(lock_t));
 	}
@@ -3571,16 +3573,39 @@ lock_table_remove_low(
 /*==================*/
 	lock_t*	lock)	/* in: table lock */
 {
-	dict_table_t*	table;
 	trx_t*		trx;
+	dict_table_t*	table;
 
 	ut_ad(mutex_own(&kernel_mutex));
 
-	table = lock->un_member.tab_lock.table;
 	trx = lock->trx;
+	table = lock->un_member.tab_lock.table;
+
+	/* Remove the table from the transaction's AUTOINC vector, if
+	the lock that is being release is an AUTOINC lock. */
+	if (lock_get_mode(lock) == LOCK_AUTO_INC) {
+
+		/* The table's AUTOINC lock can get transferred to
+		another transaction before we get here. */
+		if (table->autoinc_trx == trx) {
+			table->autoinc_trx = NULL;
+		}
+
+		/* The locks must be freed in the reverse order from
+		the one in which they were acquired. This is to avoid
+		traversing the AUTOINC lock vector unnecessarily. 
+
+		We only store locks that were granted in the
+		trx->autoinc_locks vector (see lock_table_create()
+		and lock_grant()). Therefore it can be empty and we
+		need to check for that. */
 
-	if (lock == trx->auto_inc_lock) {
-		trx->auto_inc_lock = NULL;
+		if (!ib_vector_is_empty(trx->autoinc_locks)) {
+			lock_t*	autoinc_lock;
+
+			autoinc_lock = ib_vector_pop(trx->autoinc_locks);
+			ut_a(autoinc_lock == lock);
+		}
 
 		ut_a(table->n_waiting_or_granted_auto_inc_locks > 0);
 		--table->n_waiting_or_granted_auto_inc_locks;
@@ -3955,24 +3980,6 @@ lock_table_unlock(
 	mutex_exit(&kernel_mutex);
 }
 
-/*************************************************************************
-Releases an auto-inc lock a transaction possibly has on a table.
-Releases possible other transactions waiting for this lock. */
-UNIV_INTERN
-void
-lock_table_unlock_auto_inc(
-/*=======================*/
-	trx_t*	trx)	/* in: transaction */
-{
-	if (trx->auto_inc_lock) {
-		mutex_enter(&kernel_mutex);
-
-		lock_table_dequeue(trx->auto_inc_lock);
-
-		mutex_exit(&kernel_mutex);
-	}
-}
-
 /*************************************************************************
 Releases transaction locks, and releases possible other transactions waiting
 because of these locks. */
@@ -4032,9 +4039,9 @@ lock_release_off_kernel(
 		lock = UT_LIST_GET_LAST(trx->trx_locks);
 	}
 
-	mem_heap_empty(trx->lock_heap);
+	ut_a(ib_vector_size(trx->autoinc_locks) == 0);
 
-	ut_a(trx->auto_inc_lock == NULL);
+	mem_heap_empty(trx->lock_heap);
 }
 
 /*************************************************************************
@@ -4054,6 +4061,11 @@ lock_cancel_waiting_and_release(
 	} else {
 		ut_ad(lock_get_type_low(lock) & LOCK_TABLE);
 
+		if (lock->trx->autoinc_locks != NULL) {
+			/* Release the transaction's AUTOINC locks/ */
+			lock_release_autoinc_locks(lock->trx);
+		}
+
 		lock_table_dequeue(lock);
 	}
 
@@ -5386,6 +5398,60 @@ lock_clust_rec_read_check_and_lock_alt(
 	return(ret);
 }
 
+/***********************************************************************
+Release the last lock from the transaction's autoinc locks. */
+UNIV_INLINE
+void
+lock_release_autoinc_last_lock(
+/*===========================*/
+	ib_vector_t*	autoinc_locks)	/* in/out: vector of AUTOINC locks */
+{
+	ulint		last;
+	lock_t*		lock;
+
+	ut_ad(mutex_own(&kernel_mutex));
+	ut_a(!ib_vector_is_empty(autoinc_locks));
+
+	/* The lock to be release must be the last lock acquired. */
+	last = ib_vector_size(autoinc_locks) - 1;
+	lock = ib_vector_get(autoinc_locks, last);
+
+	/* Should have only AUTOINC locks in the vector. */
+	ut_a(lock_get_mode(lock) == LOCK_AUTO_INC);
+	ut_a(lock_get_type(lock) == LOCK_TABLE);
+
+	ut_a(lock->un_member.tab_lock.table != NULL);
+
+	/* This will remove the lock from the trx autoinc_locks too. */
+	lock_table_dequeue(lock);
+}
+
+/***********************************************************************
+Release all the transaction's autoinc locks. */
+UNIV_INTERN
+void
+lock_release_autoinc_locks(
+/*=======================*/
+	trx_t*		trx)		/* in/out: transaction */
+{
+	ut_ad(mutex_own(&kernel_mutex));
+
+	ut_a(trx->autoinc_locks != NULL);
+
+	/* We release the locks in the reverse order. This is to
+	avoid searching the vector for the element to delete at
+	the lower level. See (lock_table_remove_low()) for details. */
+	while (!ib_vector_is_empty(trx->autoinc_locks)) {
+
+		/* lock_table_remove_low() will also remove the lock from
+		the transaction's autoinc_locks vector. */
+		lock_release_autoinc_last_lock(trx->autoinc_locks);
+	}
+
+	/* Should release all locks. */
+	ut_a(ib_vector_is_empty(trx->autoinc_locks));
+}
+
 /***********************************************************************
 Gets the type of a lock. Non-inline version for using outside of the
 lock module. */

row/row0mysql.c

@@ -852,19 +852,18 @@ row_update_statistics_if_needed(
 }
 
 /*************************************************************************
-Unlocks an AUTO_INC type lock possibly reserved by trx. */
+Unlocks AUTO_INC type locks that were possibly reserved by a trx. */
 UNIV_INTERN
 void
 row_unlock_table_autoinc_for_mysql(
 /*===============================*/
-	trx_t*	trx)	/* in: transaction */
+	trx_t*	trx)	/* in/out: transaction */
 {
-	if (!trx->auto_inc_lock) {
+	mutex_enter(&kernel_mutex);
 
-		return;
-	}
+	lock_release_autoinc_locks(trx);
 
-	lock_table_unlock_auto_inc(trx);
+	mutex_exit(&kernel_mutex);
 }
 
 /*************************************************************************
@@ -881,16 +880,20 @@ row_lock_table_autoinc_for_mysql(
 	row_prebuilt_t*	prebuilt)	/* in: prebuilt struct in the MySQL
 					table handle */
 {
-	trx_t*		trx		= prebuilt->trx;
-	ins_node_t*	node		= prebuilt->ins_node;
-	que_thr_t*	thr;
-	ulint		err;
-	ibool		was_lock_wait;
+	trx_t*			trx	= prebuilt->trx;
+	ins_node_t*		node	= prebuilt->ins_node;
+	const dict_table_t*	table	= prebuilt->table;
+	que_thr_t*		thr;
+	ulint			err;
+	ibool			was_lock_wait;
 
 	ut_ad(trx);
 	ut_ad(trx->mysql_thread_id == os_thread_get_curr_id());
 
-	if (trx->auto_inc_lock) {
+	/* If we already hold an AUTOINC lock on the table then do nothing.
+        Note: We peek at the value of the current owner without acquiring
+	the kernel mutex. **/
+	if (trx == table->autoinc_trx) {
 
 		return(DB_SUCCESS);
 	}

trx/trx0trx.c

@@ -169,8 +169,6 @@ trx_create(
 	trx->declared_to_be_inside_innodb = FALSE;
 	trx->n_tickets_to_enter_innodb = 0;
 
-	trx->auto_inc_lock = NULL;
-
 	trx->global_read_view_heap = mem_heap_create(256);
 	trx->global_read_view = NULL;
 	trx->read_view = NULL;
@@ -181,6 +179,10 @@ trx_create(
 
 	trx->n_autoinc_rows = 0;
 
+	/* Remember to free the vector explicitly. */
+	trx->autoinc_locks = ib_vector_create(
+		mem_heap_create(sizeof(ib_vector_t) + sizeof(void*) * 4), 4);
+
 	trx_reset_new_rec_lock_info(trx);
 
 	return(trx);
@@ -305,7 +307,6 @@ trx_free(
 	ut_a(UT_LIST_GET_LEN(trx->wait_thrs) == 0);
 
 	ut_a(!trx->has_search_latch);
-	ut_a(!trx->auto_inc_lock);
 
 	ut_a(trx->dict_operation_lock_mode == 0);
 
@@ -323,6 +324,10 @@ trx_free(
 
 	ut_a(trx->read_view == NULL);
 
+	ut_a(ib_vector_is_empty(trx->autoinc_locks));
+	/* We allocated a dedicated heap for the vector. */
+	ib_vector_free(trx->autoinc_locks);
+
 	mem_free(trx);
 }