MDEV-20612: Speed up lock_table_other_has_incompatible()bb-10.6-MDEV-24671

dict_table_t::n_lock_x_or_s: Keep track of LOCK_S or LOCK_X on the table.

lock_table_other_has_incompatible(): In the likely case that no
transaction is waiting for or holding LOCK_S or LOCK_X on the table,
return early: conflicts cannot exist.

This is based on the idea of Zhai Weixiang, who reported MySQL Bug #72948.

lock_table_has_to_wait_in_queue(), lock_table_dequeue():
Extend the optimization, inspired by
mysql/mysql-server@bb7191d6cbe47e15923143e194c03406cff9024b
by Jakub Łopuszański.

2 files changed, 72 insertions, 37 deletions

diff --git a/storage/innobase/include/dict0mem.h b/storage/innobase/include/dict0mem.h
index 637bb860146..11611b1110c 100644
--- a/storage/innobase/include/dict0mem.h
+++ b/storage/innobase/include/dict0mem.h
@@ -2279,20 +2279,22 @@ public:
 	/** Autoinc counter value to give to the next inserted row. */
 	ib_uint64_t				autoinc;
 
-	/** This counter is used to track the number of granted and pending
-	autoinc locks on this table. This value is set after acquiring the
-	lock_sys_t::mutex but we peek the contents to determine whether other
-	transactions have acquired the AUTOINC lock or not. Of course only one
-	transaction can be granted the lock but there can be multiple
-	waiters. */
-	ulong					n_waiting_or_granted_auto_inc_locks;
-
 	/** The transaction that currently holds the the AUTOINC lock on this
 	table. Protected by lock_sys.mutex. */
 	const trx_t*				autoinc_trx;
 
+  /** Number of granted or pending autoinc_lock on this table. This
+  value is set after acquiring lock_sys.mutex but
+  in innodb_autoinc_lock_mode=1 (the default),
+  ha_innobase::innobase_lock_autoinc() will perform a dirty read
+  to determine whether other transactions have acquired the autoinc_lock. */
+  uint32_t n_waiting_or_granted_auto_inc_locks;
+
 	/* @} */
 
+  /** Number of granted or pending LOCK_S or LOCK_X on the table */
+  uint32_t n_lock_x_or_s;
+
 	/** FTS specific state variables. */
 	fts_t*					fts;
 
diff --git a/storage/innobase/lock/lock0lock.cc b/storage/innobase/lock/lock0lock.cc
index eede0deec93..37e1a9781ec 100644
--- a/storage/innobase/lock/lock0lock.cc
+++ b/storage/innobase/lock/lock0lock.cc
@@ -3048,31 +3048,35 @@ lock_table_create(
 
 	check_trx_state(trx);
 
-	if ((type_mode & LOCK_MODE_MASK) == LOCK_AUTO_INC) {
+	switch (LOCK_MODE_MASK & type_mode) {
+	case LOCK_AUTO_INC:
 		++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) {
-
-		lock = table->autoinc_lock;
-
-		table->autoinc_trx = trx;
-
-		ib_vector_push(trx->autoinc_locks, &lock);
+		/* 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) {
+			lock = table->autoinc_lock;
 
-	} else if (trx->lock.table_cached
-		   < UT_ARR_SIZE(trx->lock.table_pool)) {
-		lock = &trx->lock.table_pool[trx->lock.table_cached++];
-	} else {
+			ut_ad(!table->autoinc_trx);
+			table->autoinc_trx = trx;
 
-		lock = static_cast<lock_t*>(
-			mem_heap_alloc(trx->lock.lock_heap, sizeof(*lock)));
+			ib_vector_push(trx->autoinc_locks, &lock);
+			goto allocated;
+		}
 
+		break;
+	case LOCK_X:
+	case LOCK_S:
+		++table->n_lock_x_or_s;
+		break;
 	}
 
+	lock = trx->lock.table_cached < array_elements(trx->lock.table_pool)
+		? &trx->lock.table_pool[trx->lock.table_cached++]
+		: static_cast<lock_t*>(
+			mem_heap_alloc(trx->lock.lock_heap, sizeof *lock));
+
+allocated:
 	lock->type_mode = ib_uint32_t(type_mode | LOCK_TABLE);
 	lock->trx = trx;
 
@@ -3231,7 +3235,8 @@ lock_table_remove_low(
 
 	/* Remove the table from the transaction's AUTOINC vector, if
 	the lock that is being released is an AUTOINC lock. */
-	if (lock->mode() == LOCK_AUTO_INC) {
+	switch (lock->mode()) {
+	case LOCK_AUTO_INC:
 		ut_ad((table->autoinc_trx == trx) == !lock->is_waiting());
 
 		if (table->autoinc_trx == trx) {
@@ -3246,8 +3251,16 @@ lock_table_remove_low(
 			lock_table_remove_autoinc_lock(lock, trx);
 		}
 
-		ut_a(table->n_waiting_or_granted_auto_inc_locks > 0);
-		table->n_waiting_or_granted_auto_inc_locks--;
+		ut_ad(table->n_waiting_or_granted_auto_inc_locks);
+		--table->n_waiting_or_granted_auto_inc_locks;
+		break;
+	case LOCK_X:
+	case LOCK_S:
+		ut_ad(table->n_lock_x_or_s);
+		--table->n_lock_x_or_s;
+		break;
+	default:
+		break;
 	}
 
 	UT_LIST_REMOVE(trx->lock.trx_locks, lock);
@@ -3355,12 +3368,17 @@ lock_table_other_has_incompatible(
 	const dict_table_t*	table,	/*!< in: table */
 	lock_mode		mode)	/*!< in: lock mode */
 {
-	lock_t*	lock;
-
 	lock_sys.mutex_assert_locked();
 
-	for (lock = UT_LIST_GET_LAST(table->locks);
-	     lock != NULL;
+	static_assert(LOCK_IS == 0, "compatibility");
+	static_assert(LOCK_IX == 1, "compatibility");
+
+	if (UNIV_LIKELY(mode <= LOCK_IX && !table->n_lock_x_or_s)) {
+		return(NULL);
+	}
+
+	for (lock_t* lock = UT_LIST_GET_LAST(table->locks);
+	     lock;
 	     lock = UT_LIST_GET_PREV(un_member.tab_lock.locks, lock)) {
 
 		if (lock->trx != trx
@@ -3544,7 +3562,15 @@ lock_table_has_to_wait_in_queue(
 	dict_table_t *table = wait_lock->un_member.tab_lock.table;
 	lock_sys.mutex_assert_locked();
 
-	for (const lock_t* lock = UT_LIST_GET_FIRST(table->locks);
+	static_assert(LOCK_IS == 0, "compatibility");
+	static_assert(LOCK_IX == 1, "compatibility");
+
+	if (UNIV_LIKELY(wait_lock->mode() <= LOCK_IX
+			&& !table->n_lock_x_or_s)) {
+		return(false);
+	}
+
+	for (const lock_t *lock = UT_LIST_GET_FIRST(table->locks);
 	     lock != wait_lock;
 	     lock = UT_LIST_GET_NEXT(un_member.tab_lock.locks, lock)) {
 
@@ -3569,14 +3595,21 @@ lock_table_dequeue(
 			behind will get their lock requests granted, if
 			they are now qualified to it */
 {
-	lock_sys.mutex_assert_locked();
 	mysql_mutex_assert_owner(&lock_sys.wait_mutex);
 	ut_a(in_lock->is_table());
-
+	const dict_table_t* table = in_lock->un_member.tab_lock.table;
+	lock_sys.mutex_assert_locked();
 	lock_t*	lock = UT_LIST_GET_NEXT(un_member.tab_lock.locks, in_lock);
 
 	lock_table_remove_low(in_lock);
 
+	static_assert(LOCK_IS == 0, "compatibility");
+	static_assert(LOCK_IX == 1, "compatibility");
+
+	if (UNIV_LIKELY(in_lock->mode() <= LOCK_IX && !table->n_lock_x_or_s)) {
+		return;
+	}
+
 	/* Check if waiting locks in the queue can now be granted: grant
 	locks if there are no conflicting locks ahead. */