amend 88722a7c

This will be pushed separately as MDEV-24142 once it works.

rw_lock_sx_unlock_func(): Wake up an X-latch waiter if needed.

sync_array_wait_event(): Wake up even if a pending SX-latch
request exists.

rw_lock_x_unlock_func(): Clean up.

FIXME: rw_lock_x_lock_wait_func() can still get stuck in
sync_array_wait_event() with lock_word==-X_LOCK_HALF_DECR.

storage/innobase/include/sync0rw.ic

@@ -390,36 +390,34 @@ rw_lock_x_unlock_func(
 #endif /* UNIV_DEBUG */
 	rw_lock_t*	lock)	/*!< in/out: rw-lock */
 {
-	int32_t lock_word = lock->lock_word;
-
-	if (lock_word == 0) {
-		/* Last caller in a possible recursive chain. */
-		lock->writer_thread = 0;
-	}
-
-	ut_d(rw_lock_remove_debug_info(lock, pass, RW_LOCK_X));
-
-	if (lock_word == 0 || lock_word == -X_LOCK_HALF_DECR) {
-		/* Last X-lock owned by this thread, it may still hold SX-locks.
-		ACQ_REL due to...
-		RELEASE: we release rw-lock
-		ACQUIRE: we want waiters to be loaded after lock_word is stored */
-		lock->lock_word.fetch_add(X_LOCK_DECR,
-					  std::memory_order_acq_rel);
-		if (lock->waiters) {
-			lock->wakeup_waiters();
-		}
-	} else if (lock_word == -X_LOCK_DECR
-		   || lock_word == -(X_LOCK_DECR + X_LOCK_HALF_DECR)) {
-		/* There are 2 x-locks */
-		lock->lock_word.fetch_add(X_LOCK_DECR);
-	} else {
-		/* There are more than 2 x-locks. */
-		ut_ad(lock_word < -X_LOCK_DECR);
-		lock->lock_word.fetch_add(1);
-	}
-
-	ut_ad(rw_lock_validate(lock));
+  ut_d(rw_lock_remove_debug_info(lock, pass, RW_LOCK_X));
+
+  switch (int32_t lock_word= lock->lock_word) {
+  case 0:
+    /* Last caller in a possible recursive chain. */
+    lock->writer_thread= 0;
+    /* fall through */
+  case -X_LOCK_HALF_DECR:
+    /* Last X-lock owned by this thread, it may still hold SX-locks.
+    ACQ_REL due to...
+    RELEASE: we release rw-lock
+    ACQUIRE: we want waiters to be loaded after lock_word is stored */
+    lock->lock_word.fetch_add(X_LOCK_DECR, std::memory_order_acq_rel);
+    if (lock->waiters)
+      lock->wakeup_waiters();
+    break;
+  case -X_LOCK_DECR:
+  case -X_LOCK_DECR - X_LOCK_HALF_DECR:
+    /* There are 2 x-locks */
+    lock->lock_word.fetch_add(X_LOCK_DECR);
+    break;
+  default:
+    /* There are more than 2 x-locks. */
+    ut_ad(lock_word < -X_LOCK_DECR);
+    lock->lock_word.fetch_add(1);
+  }
+
+  ut_ad(rw_lock_validate(lock));
 }
 
 /******************************************************************//**
@@ -436,12 +434,9 @@ rw_lock_sx_unlock_func(
 {
 	ut_ad(rw_lock_get_sx_lock_count(lock));
 	ut_ad(lock->sx_recursive > 0);
-
-	--lock->sx_recursive;
-
 	ut_d(rw_lock_remove_debug_info(lock, pass, RW_LOCK_SX));
 
-	if (lock->sx_recursive == 0) {
+	if (--lock->sx_recursive == 0) {
 		int32_t lock_word = lock->lock_word;
 		/* Last caller in a possible recursive chain. */
 		if (lock_word > 0) {
@@ -465,7 +460,11 @@ rw_lock_sx_unlock_func(
 			/* still has x-lock */
 			ut_ad(lock_word == -X_LOCK_HALF_DECR ||
 			      lock_word <= -(X_LOCK_DECR + X_LOCK_HALF_DECR));
-			lock->lock_word.fetch_add(X_LOCK_HALF_DECR);
+			switch (lock->lock_word.fetch_add(X_LOCK_HALF_DECR)) {
+			case -X_LOCK_HALF_DECR * 2:
+			case -X_LOCK_HALF_DECR:
+				lock->wakeup_wait_ex();
+			}
 		}
 	}
 

storage/innobase/sync/sync0arr.cc

@@ -414,9 +414,16 @@ sync_array_wait_event(
 	rw_lock_t *lock= cell->latch;
 	const int32_t lock_word = lock->lock_word;
 	if (cell->request_type == RW_LOCK_X_WAIT) {
-		if (lock_word) {
+		switch (lock_word) {
+		case 0:
+		case -X_LOCK_DECR:
+			break;
+		default:
 			mysql_mutex_lock(&lock->wait_mutex);
-			while (lock->lock_word) {
+			while (const int32_t l = lock->lock_word) {
+				if (l == -X_LOCK_DECR) {
+					break;
+				}
 				mysql_cond_wait(&lock->wait_ex_cond,
 						&lock->wait_mutex);
 			}

storage/innobase/sync/sync0rw.cc

@@ -129,7 +129,8 @@ wait_ex_cond:	A thread may only wait on the wait_ex_cond after it has
 		   (2) Verify that lock_word < 0.
 		These restrictions force the above ordering.
 		Immediately before sending the wake-up signal, we should:
-		   Verify lock_word == 0 (waiting thread holds x_lock)
+		   Verify lock_word == 0 || lock_word == -X_LOCK_HALF_DECR
+		   (waiting thread holds x_lock)
 */
 
 rw_lock_stats_t		rw_lock_stats;