Fix a deadlock in thd_report_wait_for()

Unlike commit a54abf01 claimed,
the caller of THD::awake() may actually hold the InnoDB lock_sys->mutex.
That commit introduced a deadlock of threads in the replication slave
when running the test rpl.rpl_parallel_optimistic_nobinlog.

lock_trx_handle_wait(): Expect the callers to acquire and release
lock_sys->mutex and trx->mutex.

innobase_kill_query(): Restore the logic for conditionally acquiring
and releasing the mutexes. THD::awake() can be called from inside
InnoDB while holding one or both mutexes, via thd_report_wait_for() and
via wsrep_innobase_kill_one_trx().

storage/innobase/handler/ha_innodb.cc

@@ -4915,8 +4915,31 @@ static void innobase_kill_query(handlerton*, THD* thd, enum thd_kill_levels)
 
 	if (trx_t* trx = thd_to_trx(thd)) {
 		ut_ad(trx->mysql_thd == thd);
+
+		switch (trx->abort_type) {
+		case TRX_WSREP_ABORT:
+			break;
+		case TRX_SERVER_ABORT:
+			if (!wsrep_thd_is_BF(trx->mysql_thd, FALSE)) {
+				lock_mutex_enter();
+			}
+			/* fall through */
+		case TRX_REPLICATION_ABORT:
+			trx_mutex_enter(trx);
+		}
 		/* Cancel a pending lock request if there are any */
 		lock_trx_handle_wait(trx);
+		switch (trx->abort_type) {
+		case TRX_WSREP_ABORT:
+			break;
+		case TRX_SERVER_ABORT:
+			if (!wsrep_thd_is_BF(trx->mysql_thd, FALSE)) {
+				lock_mutex_exit();
+			}
+			/* fall through */
+		case TRX_REPLICATION_ABORT:
+			trx_mutex_exit(trx);
+		}
 	}
 
 	DBUG_VOID_RETURN;

storage/innobase/lock/lock0lock.cc

@@ -7977,26 +7977,19 @@ lock_trx_handle_wait(
 /*=================*/
 	trx_t*	trx)	/*!< in/out: trx lock state */
 {
-	dberr_t	err;
-
-	lock_mutex_enter();
-
-	trx_mutex_enter(trx);
+	ut_ad(lock_mutex_own());
+	ut_ad(trx_mutex_own(trx));
 
 	if (trx->lock.was_chosen_as_deadlock_victim) {
-		err = DB_DEADLOCK;
-	} else if (trx->lock.wait_lock != NULL) {
-		lock_cancel_waiting_and_release(trx->lock.wait_lock);
-		err = DB_LOCK_WAIT;
-	} else {
+		return DB_DEADLOCK;
+	}
+	if (!trx->lock.wait_lock) {
 		/* The lock was probably granted before we got here. */
-		err = DB_SUCCESS;
+		return DB_SUCCESS;
 	}
 
-	lock_mutex_exit();
-	trx_mutex_exit(trx);
-
-	return(err);
+	lock_cancel_waiting_and_release(trx->lock.wait_lock);
+	return DB_LOCK_WAIT;
 }
 
 /*********************************************************************//**

storage/innobase/row/row0sel.cc

@@ -4615,7 +4615,11 @@ row_search_for_mysql(
 			a deadlock and the transaction had to wait then
 			release the lock it is waiting on. */
 
+			lock_mutex_enter();
+			trx_mutex_enter(trx);
 			err = lock_trx_handle_wait(trx);
+			lock_mutex_exit();
+			trx_mutex_exit(trx);
 
 			switch (err) {
 			case DB_SUCCESS:

storage/xtradb/handler/ha_innodb.cc

@@ -5511,8 +5511,31 @@ static void innobase_kill_query(handlerton*, THD* thd, enum thd_kill_levels)
 #endif /* WITH_WSREP */
 	if (trx_t* trx = thd_to_trx(thd)) {
 		ut_ad(trx->mysql_thd == thd);
+
+		switch (trx->abort_type) {
+		case TRX_WSREP_ABORT:
+			break;
+		case TRX_SERVER_ABORT:
+			if (!wsrep_thd_is_BF(trx->mysql_thd, FALSE)) {
+				lock_mutex_enter();
+			}
+			/* fall through */
+		case TRX_REPLICATION_ABORT:
+			trx_mutex_enter(trx);
+		}
 		/* Cancel a pending lock request if there are any */
 		lock_trx_handle_wait(trx);
+		switch (trx->abort_type) {
+		case TRX_WSREP_ABORT:
+			break;
+		case TRX_SERVER_ABORT:
+			if (!wsrep_thd_is_BF(trx->mysql_thd, FALSE)) {
+				lock_mutex_exit();
+			}
+			/* fall through */
+		case TRX_REPLICATION_ABORT:
+			trx_mutex_exit(trx);
+		}
 	}
 
 	DBUG_VOID_RETURN;

storage/xtradb/lock/lock0lock.cc

@@ -8087,26 +8087,19 @@ lock_trx_handle_wait(
 /*=================*/
 	trx_t*	trx)	/*!< in/out: trx lock state */
 {
-	dberr_t	err;
-
-	lock_mutex_enter();
-
-	trx_mutex_enter(trx);
+	ut_ad(lock_mutex_own());
+	ut_ad(trx_mutex_own(trx));
 
 	if (trx->lock.was_chosen_as_deadlock_victim) {
-		err = DB_DEADLOCK;
-	} else if (trx->lock.wait_lock != NULL) {
-		lock_cancel_waiting_and_release(trx->lock.wait_lock);
-		err = DB_LOCK_WAIT;
-	} else {
+		return DB_DEADLOCK;
+	}
+	if (!trx->lock.wait_lock) {
 		/* The lock was probably granted before we got here. */
-		err = DB_SUCCESS;
+		return DB_SUCCESS;
 	}
 
-	lock_mutex_exit();
-	trx_mutex_exit(trx);
-
-	return(err);
+	lock_cancel_waiting_and_release(trx->lock.wait_lock);
+	return DB_LOCK_WAIT;
 }
 
 /*********************************************************************//**

storage/xtradb/row/row0sel.cc

@@ -4627,7 +4627,11 @@ row_search_for_mysql(
 			a deadlock and the transaction had to wait then
 			release the lock it is waiting on. */
 
+			lock_mutex_enter();
+			trx_mutex_enter(trx);
 			err = lock_trx_handle_wait(trx);
+			lock_mutex_exit();
+			trx_mutex_exit(trx);
 
 			switch (err) {
 			case DB_SUCCESS: