Fixed additional problem with kernel_mutex. Kernel mutex is also

held on lock_rec_other_has_conflicting that will also call (eventually)
wsrep_innobase_kill_one_trx. Added a new parameter have_kernel_mutex
to mark do we already own kernel mutex or not.

storage/innobase/handler/ha_innodb.cc

@@ -12541,14 +12541,25 @@ wsrep_abort_slave_trx(wsrep_seqno_t bf_seqno, wsrep_seqno_t victim_seqno)
 		(long long)bf_seqno, (long long)victim_seqno);
 	abort();
 }
+/*******************************************************************//**
+This function is used to kill one transaction in BF. */
 int
-wsrep_innobase_kill_one_trx(void *bf_thd_ptr, trx_t *bf_trx, trx_t *victim_trx, ibool signal)
+wsrep_innobase_kill_one_trx(
+        void  *bf_thd_ptr,       /*!< in: BF thd */
+	trx_t *bf_trx,           /*!< in: BF trx */
+	trx_t *victim_trx,       /*!< in: victim trx */
+	ibool signal,            /*!< in: signal to be used */
+	ibool have_kernel_mutex) /*!<in: do we own kernel mutex */
 {
 	DBUG_ENTER("wsrep_innobase_kill_one_trx");
 	THD *bf_thd 	  = (THD *)bf_thd_ptr;
 	THD *thd          = (THD *) victim_trx->mysql_thd;
 	int64_t bf_seqno  = (bf_thd) ? wsrep_thd_trx_seqno(bf_thd) : 0;
 
+	if (have_kernel_mutex) {
+		ut_ad(mutex_own(&kernel_mutex));
+	}
+
 	if (!bf_thd) bf_thd = (bf_trx) ? (THD *)bf_trx->mysql_thd : NULL;
 
 	if (!thd) {
@@ -12662,10 +12673,17 @@ wsrep_innobase_kill_one_trx(void *bf_thd_ptr, trx_t *bf_trx, trx_t *victim_trx,
 			if (wait_lock) {
 				WSREP_DEBUG("canceling wait lock");
 				victim_trx->was_chosen_as_deadlock_victim= TRUE;
+				if (!have_kernel_mutex) {
 					mutex_enter(&kernel_mutex);
+				}
 				lock_cancel_waiting_and_release(wait_lock);
+				/* If we already have kernel mutex when we
+				arrived to this function, do not yet release
+				it */
+				if (!have_kernel_mutex) {
 					mutex_exit(&kernel_mutex);
 				}
+			}
 
 			wsrep_thd_awake(bf_thd, thd, signal); 
 		} else {
@@ -12749,6 +12767,9 @@ wsrep_abort_transaction(handlerton* hton, THD *bf_thd, THD *victim_thd,
 	DBUG_ENTER("wsrep_innobase_abort_thd");
 	trx_t* victim_trx = thd_to_trx(victim_thd);
 	trx_t* bf_trx     = (bf_thd) ? thd_to_trx(bf_thd) : NULL;
+
+	ut_ad(!mutex_own(&kernel_mutex));
+
 	WSREP_DEBUG("abort transaction: BF: %s victim: %s", 
 		    wsrep_thd_query(bf_thd),
 		    wsrep_thd_query(victim_thd));
@@ -12756,7 +12777,7 @@ wsrep_abort_transaction(handlerton* hton, THD *bf_thd, THD *victim_thd,
 	if (victim_trx)
 	{
 		int rcode = wsrep_innobase_kill_one_trx(
-			bf_thd, bf_trx, victim_trx, signal);
+			bf_thd, bf_trx, victim_trx, signal, FALSE);
 		wsrep_srv_conc_cancel_wait(victim_trx);
 		DBUG_RETURN(rcode);
 	} else {

storage/innobase/include/ha_prototypes.h

@@ -286,9 +286,8 @@ thd_set_lock_wait_time(
         ulint   value);	/*!< in: time waited for the lock */
 
 #ifdef WITH_WSREP
-UNIV_INTERN
-int
-wsrep_innobase_kill_one_trx(void *bf_thd, trx_t *bf_trx, trx_t *victim_trx, ibool signal);
+UNIV_INTERN int wsrep_innobase_kill_one_trx(void *thd, trx_t *bf_trx, trx_t *victim_trx,
+	ibool signal, ibool have_kernel_mutex);
 int wsrep_thd_is_brute_force(void *thd_ptr);
 int wsrep_trx_order_before(void *thd1, void *thd2);
 void wsrep_innobase_mysql_sort(int mysql_type, uint charset_number,

storage/innobase/lock/lock0lock.c

@@ -1560,7 +1560,7 @@ wsrep_kill_victim(trx_t *trx, lock_t *lock) {
 				}
 			}
 			wsrep_innobase_kill_one_trx(
-				trx->mysql_thd, trx, lock->trx, TRUE);
+				trx->mysql_thd, trx, lock->trx, TRUE, TRUE);
 		}
 	}
 }

storage/xtradb/handler/ha_innodb.cc

@@ -13873,14 +13873,24 @@ wsrep_abort_slave_trx(wsrep_seqno_t bf_seqno, wsrep_seqno_t victim_seqno)
 		(long long)bf_seqno, (long long)victim_seqno);
 	abort();
 }
+/*******************************************************************//**
+This function is used to kill one transaction in BF. */
 int
-wsrep_innobase_kill_one_trx(trx_t *bf_trx, trx_t *victim_trx, ibool signal)
+wsrep_innobase_kill_one_trx(
+	trx_t *bf_trx,           /*!< in: BF trx */
+	trx_t *victim_trx,       /*!< in: victim trx */
+	ibool signal,            /*!< in: signal to be used */
+	ibool have_kernel_mutex) /*!<in: do we own kernel mutex */
 {
 	DBUG_ENTER("wsrep_innobase_kill_one_trx");
 	THD *thd          = (THD *) victim_trx->mysql_thd;
 	THD *bf_thd       = (bf_trx) ? (THD *)bf_trx->mysql_thd : NULL;
 	int64_t bf_seqno  = (bf_thd) ? wsrep_thd_trx_seqno(bf_thd) : 0;
 
+	if (have_kernel_mutex) {
+		ut_ad(mutex_own(&kernel_mutex));
+	}
+
 	if (!thd) {
 		DBUG_PRINT("wsrep", ("no thd for conflicting lock"));
 		WSREP_WARN("no THD for trx: %llu", victim_trx->id);
@@ -13985,10 +13995,17 @@ wsrep_innobase_kill_one_trx(trx_t *bf_trx, trx_t *victim_trx, ibool signal)
 			if (wait_lock) {
 				WSREP_DEBUG("canceling wait lock");
 				victim_trx->was_chosen_as_deadlock_victim= TRUE;
+				if (!have_kernel_mutex) {
 					mutex_enter(&kernel_mutex);
+				}
 				lock_cancel_waiting_and_release(wait_lock);
+				/* If we already have kernel mutex when we
+				arrived to this function, do not yet release
+				it */
+				if (!have_kernel_mutex) {
 					mutex_exit(&kernel_mutex);
 				}
+			}
 
 			wsrep_thd_awake(bf_thd, thd, signal); 
 		} else {
@@ -14072,6 +14089,9 @@ wsrep_abort_transaction(handlerton* hton, THD *bf_thd, THD *victim_thd,
 	DBUG_ENTER("wsrep_innobase_abort_thd");
 	trx_t* victim_trx = thd_to_trx(victim_thd);
 	trx_t* bf_trx     = (bf_thd) ? thd_to_trx(bf_thd) : NULL;
+
+	ut_ad(!mutex_own(&kernel_mutex));
+
 	WSREP_DEBUG("abort transaction: BF: %s victim: %s", 
 		    wsrep_thd_query(bf_thd),
 		    wsrep_thd_query(victim_thd));
@@ -14079,7 +14099,7 @@ wsrep_abort_transaction(handlerton* hton, THD *bf_thd, THD *victim_thd,
 	if (victim_trx)
 	{
 		int rcode = wsrep_innobase_kill_one_trx(bf_trx, victim_trx,
-							signal);
+			signal, FALSE);
 		wsrep_srv_conc_cancel_wait(victim_trx);
 		DBUG_RETURN(rcode);
 	} else {

storage/xtradb/include/ha_prototypes.h

@@ -301,9 +301,8 @@ thd_flush_log_at_trx_commit(
 	void*	thd);
 
 #ifdef WITH_WSREP
-UNIV_INTERN
-int
-wsrep_innobase_kill_one_trx(trx_t *bf_trx, trx_t *victim_trx, ibool signal);
+UNIV_INTERN int wsrep_innobase_kill_one_trx(trx_t *bf_trx, trx_t *victim_trx,
+	ibool signal, ibool have_kernel_mutex);
 int wsrep_thd_is_brute_force(void *thd_ptr);
 int wsrep_trx_order_before(void *thd1, void *thd2);
 void wsrep_innobase_mysql_sort(int mysql_type, uint charset_number,

storage/xtradb/lock/lock0lock.c

@@ -1533,7 +1533,7 @@ wsrep_kill_victim(trx_t *trx, lock_t *lock) {
 			/* cannot release lock, until our lock
 			is in the queue*/
 		} else if (lock->trx != trx) {
-			wsrep_innobase_kill_one_trx(trx, lock->trx, TRUE);
+			wsrep_innobase_kill_one_trx(trx, lock->trx, TRUE, TRUE);
 		}
 	}
 }
@@ -4245,7 +4245,7 @@ lock_table_other_has_incompatible(
 				} else {
                                   if (bf_this && bf_other)
 					wsrep_innobase_kill_one_trx(
-						(trx_t *)trx, lock->trx, TRUE);
+						(trx_t *)trx, lock->trx, TRUE, TRUE);
 					return(lock);
 				}
 			} else {