MDEV-11296 - InnoDB stalls under OLTP RW on P8

Simplified away rw_lock_get_waiters(), rw_lock_set_waiter_flag(),
rw_lock_reset_waiter_flag(). Let waiters have predictable data type.

storage/innobase/include/sync0rw.h

@@ -484,14 +484,6 @@ ulint
 rw_lock_get_sx_lock_count(
 /*======================*/
 	const rw_lock_t*	lock);	/*!< in: rw-lock */
-/********************************************************************//**
-Check if there are threads waiting for the rw-lock.
-@return 1 if waiters, 0 otherwise */
-UNIV_INLINE
-ulint
-rw_lock_get_waiters(
-/*================*/
-	const rw_lock_t*	lock);	/*!< in: rw-lock */
 /******************************************************************//**
 Returns the write-status of the lock - this function made more sense
 with the old rw_lock implementation.
@@ -620,7 +612,7 @@ struct rw_lock_t
 	volatile lint	lock_word;
 
 	/** 1: there are waiters */
-	volatile ulint	waiters;
+	volatile uint32_t	waiters;
 
 	/** Default value FALSE which means the lock is non-recursive.
 	The value is typically set to TRUE making normal rw_locks recursive.

storage/innobase/include/sync0rw.ic

@@ -66,52 +66,6 @@ rw_lock_remove_debug_info(
 	ulint		lock_type);	/*!< in: lock type */
 #endif /* UNIV_DEBUG */
 
-/********************************************************************//**
-Check if there are threads waiting for the rw-lock.
-@return 1 if waiters, 0 otherwise */
-UNIV_INLINE
-ulint
-rw_lock_get_waiters(
-/*================*/
-	const rw_lock_t*	lock)	/*!< in: rw-lock */
-{
-	return(lock->waiters);
-}
-
-/********************************************************************//**
-Sets lock->waiters to 1. It is not an error if lock->waiters is already
-1. On platforms where ATOMIC builtins are used this function enforces a
-memory barrier. */
-UNIV_INLINE
-void
-rw_lock_set_waiter_flag(
-/*====================*/
-	rw_lock_t*	lock)	/*!< in/out: rw-lock */
-{
-#ifdef INNODB_RW_LOCKS_USE_ATOMICS
-	my_atomic_storelint(&lock->waiters, 1);
-#else /* INNODB_RW_LOCKS_USE_ATOMICS */
-	lock->waiters = 1;
-#endif /* INNODB_RW_LOCKS_USE_ATOMICS */
-}
-
-/********************************************************************//**
-Resets lock->waiters to 0. It is not an error if lock->waiters is already
-0. On platforms where ATOMIC builtins are used this function enforces a
-memory barrier. */
-UNIV_INLINE
-void
-rw_lock_reset_waiter_flag(
-/*======================*/
-	rw_lock_t*	lock)	/*!< in/out: rw-lock */
-{
-#ifdef INNODB_RW_LOCKS_USE_ATOMICS
-	my_atomic_storelint(&lock->waiters, 0);
-#else /* INNODB_RW_LOCKS_USE_ATOMICS */
-	lock->waiters = 0;
-#endif /* INNODB_RW_LOCKS_USE_ATOMICS */
-}
-
 /******************************************************************//**
 Returns the write-status of the lock - this function made more sense
 with the old rw_lock implementation.
@@ -555,7 +509,7 @@ rw_lock_x_unlock_func(
 		We do not need to signal wait_ex waiters, since they cannot
 		exist when there is a writer. */
 		if (lock->waiters) {
-			rw_lock_reset_waiter_flag(lock);
+			my_atomic_store32((int32*) &lock->waiters, 0);
 			os_event_set(lock->event);
 			sync_array_object_signalled();
 		}
@@ -606,7 +560,7 @@ rw_lock_sx_unlock_func(
 			since they cannot exist when there is an sx-lock
 			holder. */
 			if (lock->waiters) {
-				rw_lock_reset_waiter_flag(lock);
+				my_atomic_store32((int32*) &lock->waiters, 0);
 				os_event_set(lock->event);
 				sync_array_object_signalled();
 			}

storage/innobase/include/sync0types.h

@@ -1258,12 +1258,10 @@ enum rw_lock_flag_t {
 
 #ifdef _WIN64
 #define my_atomic_addlint(A,B) my_atomic_add64((int64*) (A), (B))
-#define my_atomic_storelint(A,B) my_atomic_store64((int64*) (A), (B))
 #define my_atomic_loadlint(A) my_atomic_load64((int64*) (A))
 #define my_atomic_caslint(A,B,C) my_atomic_cas64((int64*) (A), (int64*) (B), (C))
 #else
 #define my_atomic_addlint my_atomic_addlong
-#define my_atomic_storelint my_atomic_storelong
 #define my_atomic_loadlint my_atomic_loadlong
 #define my_atomic_caslint my_atomic_caslong
 #endif

storage/innobase/row/row0merge.cc

@@ -2090,8 +2090,7 @@ row_merge_read_clustered_index(
 			}
 
 			if (dbug_run_purge
-			    || rw_lock_get_waiters(
+			    || dict_index_get_lock(clust_index)->waiters) {
-				    dict_index_get_lock(clust_index))) {
 				/* There are waiters on the clustered
 				index tree lock, likely the purge
 				thread. Store and restore the cursor

storage/innobase/sync/sync0rw.cc

@@ -399,7 +399,7 @@ lock_loop:
 
 		/* Set waiters before checking lock_word to ensure wake-up
 		signal is sent. This may lead to some unnecessary signals. */
-		rw_lock_set_waiter_flag(lock);
+		my_atomic_store32((int32*) &lock->waiters, 1);
 
 		if (rw_lock_s_lock_low(lock, pass, file_name, line)) {
 
@@ -806,7 +806,7 @@ lock_loop:
 
 	/* Waiters must be set before checking lock_word, to ensure signal
 	is sent. This could lead to a few unnecessary wake-up signals. */
-	rw_lock_set_waiter_flag(lock);
+	my_atomic_store32((int32*) &lock->waiters, 1);
 
 	if (rw_lock_x_lock_low(lock, pass, file_name, line)) {
 		sync_array_free_cell(sync_arr, cell);
@@ -911,7 +911,7 @@ lock_loop:
 
 	/* Waiters must be set before checking lock_word, to ensure signal
 	is sent. This could lead to a few unnecessary wake-up signals. */
-	rw_lock_set_waiter_flag(lock);
+	my_atomic_store32((int32*) &lock->waiters, 1);
 
 	if (rw_lock_sx_lock_low(lock, pass, file_name, line)) {
 
@@ -950,16 +950,14 @@ rw_lock_validate(
 /*=============*/
 	const rw_lock_t*	lock)	/*!< in: rw-lock */
 {
-	ulint	waiters;
 	lint	lock_word;
 
 	ut_ad(lock);
 
-	waiters = rw_lock_get_waiters(lock);
 	lock_word = lock->lock_word;
 
 	ut_ad(lock->magic_n == RW_LOCK_MAGIC_N);
-	ut_ad(waiters == 0 || waiters == 1);
+	ut_ad(lock->waiters < 2);
 	ut_ad(lock_word > -(2 * X_LOCK_DECR));
 	ut_ad(lock_word <= X_LOCK_DECR);
 
@@ -1229,7 +1227,7 @@ rw_lock_list_print_info(
 
 			fprintf(file, "RW-LOCK: %p ", (void*) lock);
 
-			if (rw_lock_get_waiters(lock)) {
+			if (lock->waiters) {
 				fputs(" Waiters for the lock exist\n", file);
 			} else {
 				putc('\n', file);
@@ -1283,7 +1281,7 @@ rw_lock_print(
 
 	if (lock->lock_word != X_LOCK_DECR) {
 
-		if (rw_lock_get_waiters(lock)) {
+		if (lock->waiters) {
 			fputs(" Waiters for the lock exist\n", stderr);
 		} else {
 			putc('\n', stderr);