MDEV-7148 - Recurring: InnoDB: Failing assertion: !lock->recursive

Re-applied lost in the merge revision:
commit ed313e8a92a836422b9ae7b9ecf11c44ed4d5d66
Author: Sergey Vojtovich <svoj@mariadb.org>
Date:   Mon Dec 1 14:58:29 2014 +0400

    MDEV-7148 - Recurring: InnoDB: Failing assertion: !lock->recursive

    On PPC64 high-loaded server may crash due to assertion failure in InnoDB
    rwlocks code.

    This happened because load order between "recursive" and "writer_thread"
    wasn't properly enforced.

4 files changed, 28 insertions, 14 deletions

diff --git a/storage/innobase/include/sync0rw.ic b/storage/innobase/include/sync0rw.ic
index bb05ae7daf1..69251da6e35 100644
--- a/storage/innobase/include/sync0rw.ic
+++ b/storage/innobase/include/sync0rw.ic
@@ -386,6 +386,7 @@ rw_lock_x_lock_func_nowait(
 	ulint		line)	/*!< in: line where requested */
 {
 	ibool success;
+	ibool local_recursive= lock->recursive;
 
 #ifdef INNODB_RW_LOCKS_USE_ATOMICS
 	success = os_compare_and_swap_lint(&lock->lock_word, X_LOCK_DECR, 0);
@@ -400,10 +401,14 @@ rw_lock_x_lock_func_nowait(
 	mutex_exit(&(lock->mutex));
 
 #endif
+	/* Note: recursive must be loaded before writer_thread see
+	comment for rw_lock_set_writer_id_and_recursion_flag().
+	To achieve this we load it before os_compare_and_swap_lint(),
+	which implies full memory barrier in current implementation. */
 	if (success) {
 		rw_lock_set_writer_id_and_recursion_flag(lock, TRUE);
 
-	} else if (lock->recursive
+	} else if (local_recursive
 		   && os_thread_eq(lock->writer_thread,
 				   os_thread_get_curr_id())) {
 		/* Relock: this lock_word modification is safe since no other
diff --git a/storage/innobase/sync/sync0rw.cc b/storage/innobase/sync/sync0rw.cc
index 4ff330791a0..e77c7a9b396 100644
--- a/storage/innobase/sync/sync0rw.cc
+++ b/storage/innobase/sync/sync0rw.cc
@@ -545,6 +545,8 @@ rw_lock_x_lock_low(
 	const char*	file_name,/*!< in: file name where lock requested */
 	ulint		line)	/*!< in: line where requested */
 {
+	ibool local_recursive= lock->recursive;
+
 	if (rw_lock_lock_word_decr(lock, X_LOCK_DECR)) {
 
 		/* lock->recursive also tells us if the writer_thread
@@ -566,12 +568,12 @@ rw_lock_x_lock_low(
 	} else {
 		os_thread_id_t	thread_id = os_thread_get_curr_id();
 
-		if (!pass) {
-			os_rmb;
-		}
-
-		/* Decrement failed: relock or failed lock */
-		if (!pass && lock->recursive
+		/* Decrement failed: relock or failed lock
+		Note: recursive must be loaded before writer_thread see
+		comment for rw_lock_set_writer_id_and_recursion_flag().
+		To achieve this we load it before rw_lock_lock_word_decr(),
+		which implies full memory barrier in current implementation. */
+		if (!pass && local_recursive
 		    && os_thread_eq(lock->writer_thread, thread_id)) {
 			/* Relock */
 			if (lock->lock_word == 0) {
diff --git a/storage/xtradb/include/sync0rw.ic b/storage/xtradb/include/sync0rw.ic
index 8aadc406132..d481d14009b 100644
--- a/storage/xtradb/include/sync0rw.ic
+++ b/storage/xtradb/include/sync0rw.ic
@@ -511,6 +511,7 @@ rw_lock_x_lock_func_nowait(
 	ulint		line)	/*!< in: line where requested */
 {
 	ibool success;
+	ibool local_recursive= lock->recursive;
 
 #ifdef INNODB_RW_LOCKS_USE_ATOMICS
 	success = os_compare_and_swap_lint(&lock->lock_word, X_LOCK_DECR, 0);
@@ -525,10 +526,14 @@ rw_lock_x_lock_func_nowait(
 	mutex_exit(&(lock->mutex));
 
 #endif
+	/* Note: recursive must be loaded before writer_thread see
+	comment for rw_lock_set_writer_id_and_recursion_flag().
+	To achieve this we load it before os_compare_and_swap_lint(),
+	which implies full memory barrier in current implementation. */
 	if (success) {
 		rw_lock_set_writer_id_and_recursion_flag(lock, TRUE);
 
-	} else if (lock->recursive
+	} else if (local_recursive
 		   && os_thread_eq(lock->writer_thread,
 				   os_thread_get_curr_id())) {
 		/* Relock: this lock_word modification is safe since no other
diff --git a/storage/xtradb/sync/sync0rw.cc b/storage/xtradb/sync/sync0rw.cc
index 3296e2e74a7..51bfba9950b 100644
--- a/storage/xtradb/sync/sync0rw.cc
+++ b/storage/xtradb/sync/sync0rw.cc
@@ -694,6 +694,8 @@ rw_lock_x_lock_low(
 	const char*	file_name,/*!< in: file name where lock requested */
 	ulint		line)	/*!< in: line where requested */
 {
+	ibool local_recursive= lock->recursive;
+
 	if (rw_lock_lock_word_decr(lock, X_LOCK_DECR)) {
 
 		/* lock->recursive also tells us if the writer_thread
@@ -715,12 +717,12 @@ rw_lock_x_lock_low(
 	} else {
 		os_thread_id_t	thread_id = os_thread_get_curr_id();
 
-		if (!pass) {
-			os_rmb;
-		}
-
-		/* Decrement failed: relock or failed lock */
-		if (!pass && lock->recursive
+		/* Decrement failed: relock or failed lock
+		Note: recursive must be loaded before writer_thread see
+		comment for rw_lock_set_writer_id_and_recursion_flag().
+		To achieve this we load it before rw_lock_lock_word_decr(),
+		which implies full memory barrier in current implementation. */
+		if (!pass && local_recursive
 		    && os_thread_eq(lock->writer_thread, thread_id)) {
 			/* Relock */
 			if (lock->lock_word == 0) {