MDEV-24114 : Crash: WSREP: invalid state ROLLED_BACK (FATAL)10.2-MDEV-25114

Reverts fix for MDEV-23328 i.e. commit 29bbcac0ee841

In this fix we try to modify normal SQL KILL in a following way:

trx_id= thd_to_trx(victim_thd)->id;
mutex_unlock(victim_thd->LOCK_thd_data);
mutex_unlock(victim_thd->LOCK_thd_kill);
lock_mutex_enter
trx=find_and_lock_trx_by_id(trx_id)
mytex_lock(trx->mysql_thd->LOCK_thd_kill);
mutex_lock(trx->mysql_thd->LOCK_thd_data)

For THD::awake() we use:

mutex_lock(thd->LOCK_thd_kill);
mutex_lock(thd->LOCK_thd_data);
thd->awake();
mutex_unlock(thd->LOCK_thd_data);
mutex_unlock(thd->LOCK_thd_kill);

For THD::set_killed in most cases we use

mutex_lock(thd->LOCK_thd_kill);
mutex_lock(thd->LOCK_thd_data);
thd->set_killed_no_mutex(...);
mutex_unlock(thd->LOCK_thd_data);
mutex_unlock(thd->LOCK_thd_kill);

14 files changed, 204 insertions, 187 deletions

diff --git a/include/mysql/service_wsrep.h b/include/mysql/service_wsrep.h
index 8432526a78b..7a04defde96 100644
--- a/include/mysql/service_wsrep.h
+++ b/include/mysql/service_wsrep.h
@@ -1,5 +1,5 @@
 #ifndef MYSQL_SERVICE_WSREP_INCLUDED
-/* Copyright (c) 2015 MariaDB Corporation Ab
+/* Copyright (c) 2015-2021 MariaDB Corporation Ab
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
diff --git a/sql/event_scheduler.cc b/sql/event_scheduler.cc
index 423f7e6ea2b..799a689e1c6 100644
--- a/sql/event_scheduler.cc
+++ b/sql/event_scheduler.cc
@@ -650,7 +650,8 @@ Event_scheduler::stop()
     state= STOPPING;
     DBUG_PRINT("info", ("Scheduler thread has id %lu",
                         (ulong) scheduler_thd->thread_id));
-    /* Lock from delete */
+    /* Lock from concurrent usage and delete/kill */
+    mysql_mutex_lock(&scheduler_thd->LOCK_thd_kill);
     mysql_mutex_lock(&scheduler_thd->LOCK_thd_data);
     /* This will wake up the thread if it waits on Queue's conditional */
     sql_print_information("Event Scheduler: Killing the scheduler thread, "
@@ -658,6 +659,7 @@ Event_scheduler::stop()
                           (ulong) scheduler_thd->thread_id);
     scheduler_thd->awake(KILL_CONNECTION);
     mysql_mutex_unlock(&scheduler_thd->LOCK_thd_data);
+    mysql_mutex_unlock(&scheduler_thd->LOCK_thd_kill);
 
     /* thd could be 0x0, when shutting down */
     sql_print_information("Event Scheduler: "
diff --git a/sql/handler.cc b/sql/handler.cc
index 87592beb5d3..5290c1f3a1c 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -838,7 +838,6 @@ static my_bool kill_handlerton(THD *thd, plugin_ref plugin,
 {
   handlerton *hton= plugin_hton(plugin);
 
-  mysql_mutex_assert_owner(&thd->LOCK_thd_data);
   if (hton->state == SHOW_OPTION_YES && hton->kill_query &&
       thd_get_ha_data(thd, hton))
     hton->kill_query(hton, thd, *(enum thd_kill_levels *) level);
diff --git a/sql/slave.cc b/sql/slave.cc
index 761fdbe807a..5ff53d4d2bd 100644
--- a/sql/slave.cc
+++ b/sql/slave.cc
@@ -1,5 +1,5 @@
 /* Copyright (c) 2000, 2017, Oracle and/or its affiliates.
-   Copyright (c) 2009, 2020, MariaDB Corporation
+   Copyright (c) 2009, 2021, MariaDB Corporation
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -317,9 +317,11 @@ static void bg_rpl_load_gtid_slave_state(void *)
 static void bg_slave_kill(void *victim)
 {
   THD *to_kill= (THD *)victim;
+  mysql_mutex_lock(&to_kill->LOCK_thd_kill);
   mysql_mutex_lock(&to_kill->LOCK_thd_data);
   to_kill->awake(KILL_CONNECTION);
   mysql_mutex_unlock(&to_kill->LOCK_thd_data);
+  mysql_mutex_unlock(&to_kill->LOCK_thd_kill);
   mysql_mutex_lock(&to_kill->LOCK_wakeup_ready);
   to_kill->rgi_slave->killed_for_retry= rpl_group_info::RETRY_KILL_KILLED;
   mysql_cond_broadcast(&to_kill->COND_wakeup_ready);
@@ -754,6 +756,7 @@ terminate_slave_thread(THD *thd,
     int error __attribute__((unused));
     DBUG_PRINT("loop", ("killing slave thread"));
 
+    mysql_mutex_lock(&thd->LOCK_thd_kill);
     mysql_mutex_lock(&thd->LOCK_thd_data);
 #ifndef DONT_USE_THR_ALARM
     /*
@@ -767,7 +770,7 @@ terminate_slave_thread(THD *thd,
     thd->awake(NOT_KILLED);
 
     mysql_mutex_unlock(&thd->LOCK_thd_data);
-
+    mysql_mutex_unlock(&thd->LOCK_thd_kill);
     /*
       There is a small chance that slave thread might miss the first
       alarm. To protect againts it, resend the signal until it reacts
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index 5ada018e540..c72d192efa5 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -593,10 +593,12 @@ handle_condition(THD *thd,
 extern "C" void thd_kill_timeout(THD* thd)
 {
   thd->status_var.max_statement_time_exceeded++;
+  mysql_mutex_lock(&thd->LOCK_thd_kill);
   mysql_mutex_lock(&thd->LOCK_thd_data);
   /* Kill queries that can't cause data corruptions */
   thd->awake(KILL_TIMEOUT);
   mysql_mutex_unlock(&thd->LOCK_thd_data);
+  mysql_mutex_unlock(&thd->LOCK_thd_kill);
 }
 
 
@@ -1693,6 +1695,7 @@ void THD::awake(killed_state state_to_set)
   DBUG_PRINT("enter", ("this: %p current_thd: %p  state: %d",
                        this, current_thd, (int) state_to_set));
   THD_CHECK_SENTRY(this);
+  mysql_mutex_assert_owner(&LOCK_thd_kill);
   mysql_mutex_assert_owner(&LOCK_thd_data);
 
   print_aborted_warning(3, "KILLED");
@@ -1705,7 +1708,6 @@ void THD::awake(killed_state state_to_set)
     state_to_set= killed;
 
   /* Set the 'killed' flag of 'this', which is the target THD object. */
-  mysql_mutex_lock(&LOCK_thd_kill);
   set_killed_no_mutex(state_to_set);
 
   if (state_to_set >= KILL_CONNECTION || state_to_set == NOT_KILLED)
@@ -1792,7 +1794,6 @@ void THD::awake(killed_state state_to_set)
     }
     mysql_mutex_unlock(&mysys_var->mutex);
   }
-  mysql_mutex_unlock(&LOCK_thd_kill);
   DBUG_VOID_RETURN;
 }
 
@@ -1808,9 +1809,10 @@ void THD::disconnect()
 {
   Vio *vio= NULL;
 
+  mysql_mutex_lock(&LOCK_thd_kill);
   mysql_mutex_lock(&LOCK_thd_data);
 
-  set_killed(KILL_CONNECTION);
+  set_killed_no_mutex(KILL_CONNECTION);
 
 #ifdef SIGNAL_WITH_VIO_CLOSE
   /*
@@ -1828,6 +1830,7 @@ void THD::disconnect()
   net.thd= 0;                                   // Don't collect statistics
 
   mysql_mutex_unlock(&LOCK_thd_data);
+  mysql_mutex_unlock(&LOCK_thd_kill);
 }
 
 
@@ -1844,9 +1847,10 @@ bool THD::notify_shared_lock(MDL_context_owner *ctx_in_use,
   {
     /* This code is similar to kill_delayed_threads() */
     DBUG_PRINT("info", ("kill delayed thread"));
+    mysql_mutex_lock(&in_use->LOCK_thd_kill);
     mysql_mutex_lock(&in_use->LOCK_thd_data);
     if (in_use->killed < KILL_CONNECTION)
-      in_use->set_killed(KILL_CONNECTION);
+      in_use->set_killed_no_mutex(KILL_CONNECTION);
     if (in_use->mysys_var)
     {
       mysql_mutex_lock(&in_use->mysys_var->mutex);
@@ -1858,11 +1862,14 @@ bool THD::notify_shared_lock(MDL_context_owner *ctx_in_use,
       mysql_mutex_unlock(&in_use->mysys_var->mutex);
     }
     mysql_mutex_unlock(&in_use->LOCK_thd_data);
+    mysql_mutex_unlock(&in_use->LOCK_thd_kill);
     signalled= TRUE;
   }
 
   if (needs_thr_lock_abort)
   {
+    bool need_mutex_release= true;
+    mysql_mutex_lock(&in_use->LOCK_thd_kill);
     mysql_mutex_lock(&in_use->LOCK_thd_data);
     /* If not already dying */
     if (in_use->killed != KILL_CONNECTION_HARD)
@@ -1879,18 +1886,24 @@ bool THD::notify_shared_lock(MDL_context_owner *ctx_in_use,
           thread can see those instances (e.g. see partitioning code).
         */
         if (!thd_table->needs_reopen())
-        {
           signalled|= mysql_lock_abort_for_thread(this, thd_table);
-          if (WSREP(this) && wsrep_thd_is_BF(this, FALSE))
-          {
-            WSREP_DEBUG("remove_table_from_cache: %llu",
-                        (unsigned long long) this->real_id);
-            wsrep_abort_thd((void *)this, (void *)in_use, FALSE);
-          }
-        }
       }
+
+#ifdef WITH_WSREP
+      if (WSREP(this) && wsrep_thd_is_BF(this, FALSE))
+      {
+        WSREP_DEBUG("remove_table_from_cache: %llu",
+                    (unsigned long long) this->real_id);
+        wsrep_abort_thd((void *)this, (void *)in_use, FALSE);
+	need_mutex_release= false;
+      }
+#endif /* WITH_WSREP */
+    }
+    if (need_mutex_release)
+    {
+      mysql_mutex_unlock(&in_use->LOCK_thd_data);
+      mysql_mutex_unlock(&in_use->LOCK_thd_kill);
     }
-    mysql_mutex_unlock(&in_use->LOCK_thd_data);
   }
   DBUG_RETURN(signalled);
 }
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 50ab3c56ca9..9defbc4da03 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -1,6 +1,6 @@
 /*
    Copyright (c) 2000, 2016, Oracle and/or its affiliates.
-   Copyright (c) 2009, 2017, MariaDB Corporation.
+   Copyright (c) 2009, 2021, MariaDB Corporation.
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -3727,7 +3727,7 @@ public:
     mysql_mutex_unlock(&LOCK_thd_kill);
   }
   /*
-    This is only used by THD::awake where we need to keep the lock mutex
+    This is only used where we need to keep the lock mutex
     locked over some time.
     It's ok to have this inline, as in most cases killed_errno_arg will
     be a constant 0 and most of the function will disappear.
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index 51b2c84cfea..f6135e8b168 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -2752,9 +2752,10 @@ void kill_delayed_threads(void)
   Delayed_insert *di;
   while ((di= it++))
   {
+    mysql_mutex_lock(&di->thd.LOCK_thd_kill);
     mysql_mutex_lock(&di->thd.LOCK_thd_data);
     if (di->thd.killed < KILL_CONNECTION)
-      di->thd.set_killed(KILL_CONNECTION);
+      di->thd.set_killed_no_mutex(KILL_CONNECTION);
     if (di->thd.mysys_var)
     {
       mysql_mutex_lock(&di->thd.mysys_var->mutex);
@@ -2773,6 +2774,7 @@ void kill_delayed_threads(void)
       mysql_mutex_unlock(&di->thd.mysys_var->mutex);
     }
     mysql_mutex_unlock(&di->thd.LOCK_thd_data);
+    mysql_mutex_unlock(&di->thd.LOCK_thd_kill);
   }
   mysql_mutex_unlock(&LOCK_delayed_insert); // For unlink from list
   DBUG_VOID_RETURN;
@@ -3139,10 +3141,12 @@ pthread_handler_t handle_delayed_insert(void *arg)
       THD::notify_shared_lock() dosn't try to access open tables after
       this.
     */
+    mysql_mutex_lock(&thd->LOCK_thd_kill);
     mysql_mutex_lock(&thd->LOCK_thd_data);
-    thd->set_killed(KILL_CONNECTION_HARD);	        // If error
+    thd->set_killed_no_mutex(KILL_CONNECTION_HARD);	        // If error
     thd->mdl_context.set_needs_thr_lock_abort(0);
     mysql_mutex_unlock(&thd->LOCK_thd_data);
+    mysql_mutex_unlock(&thd->LOCK_thd_kill);
 
     close_thread_tables(thd);			// Free the table
     thd->release_transactional_locks();
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index f782c5c25e7..55f463abc34 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -1,5 +1,5 @@
 /* Copyright (c) 2000, 2017, Oracle and/or its affiliates.
-   Copyright (c) 2008, 2020, MariaDB
+   Copyright (c) 2008, 2021, MariaDB
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -8907,7 +8907,8 @@ THD *find_thread_by_id(longlong id, bool query_id)
       continue;
     if (id == (query_id ? tmp->query_id : (longlong) tmp->thread_id))
     {
-      mysql_mutex_lock(&tmp->LOCK_thd_data);    // Lock from delete
+      mysql_mutex_lock(&tmp->LOCK_thd_kill); // Lock from delete
+      mysql_mutex_lock(&tmp->LOCK_thd_data); // Lock from concurrent usage
       break;
     }
   }
@@ -8970,6 +8971,7 @@ kill_one_thread(THD *thd, longlong id, killed_state kill_signal, killed_type typ
       error= (type == KILL_TYPE_QUERY ? ER_KILL_QUERY_DENIED_ERROR :
                                         ER_KILL_DENIED_ERROR);
     mysql_mutex_unlock(&tmp->LOCK_thd_data);
+    mysql_mutex_unlock(&tmp->LOCK_thd_kill);
   }
   DBUG_PRINT("exit", ("%d", error));
   DBUG_RETURN(error);
diff --git a/sql/sql_plugin_services.ic b/sql/sql_plugin_services.ic
index 358052132a0..0b7ab13427e 100644
--- a/sql/sql_plugin_services.ic
+++ b/sql/sql_plugin_services.ic
@@ -1,5 +1,5 @@
 /* Copyright (c) 2009, 2010, Oracle and/or its affiliates.
-   Copyright (c) 2012, 2014, Monty Program Ab
+   Copyright (c) 2012, 2021, MariaDB Corporation Ab
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
diff --git a/sql/sql_repl.cc b/sql/sql_repl.cc
index 7ff0e27b008..637820308cb 100644
--- a/sql/sql_repl.cc
+++ b/sql/sql_repl.cc
@@ -3398,7 +3398,8 @@ void kill_zombie_dump_threads(uint32 slave_server_id)
     if (tmp->get_command() == COM_BINLOG_DUMP &&
        tmp->variables.server_id == slave_server_id)
     {
-      mysql_mutex_lock(&tmp->LOCK_thd_data);    // Lock from delete
+      mysql_mutex_lock(&tmp->LOCK_thd_kill);    // Lock from delete
+      mysql_mutex_lock(&tmp->LOCK_thd_data);    // Lock from concurrent
       break;
     }
   }
@@ -3412,6 +3413,7 @@ void kill_zombie_dump_threads(uint32 slave_server_id)
     */
     tmp->awake(KILL_SLAVE_SAME_ID);
     mysql_mutex_unlock(&tmp->LOCK_thd_data);
+    mysql_mutex_unlock(&tmp->LOCK_thd_kill);
   }
 }
 
diff --git a/sql/wsrep_dummy.cc b/sql/wsrep_dummy.cc
index f5e88aeac1f..f58b2ad5236 100644
--- a/sql/wsrep_dummy.cc
+++ b/sql/wsrep_dummy.cc
@@ -1,4 +1,4 @@
-/* Copyright (C) 2014, 2020, MariaDB
+/* Copyright (C) 2014, 2021, MariaDB
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
diff --git a/sql/wsrep_mysqld.cc b/sql/wsrep_mysqld.cc
index f22d8bf0f5a..6dd45446ddf 100644
--- a/sql/wsrep_mysqld.cc
+++ b/sql/wsrep_mysqld.cc
@@ -1,4 +1,5 @@
-/* Copyright 2008-2015 Codership Oy <http://www.codership.com>
+/* Copyright 2008-2021 Codership Oy <http://www.codership.com>
+   Copyright 2008-2021 MariaDB Corporation Ab
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -838,13 +839,6 @@ void wsrep_thr_init()
 void wsrep_init_startup (bool first)
 {
   if (wsrep_init()) unireg_abort(1);
-
-  wsrep_thr_lock_init(
-     (wsrep_thd_is_brute_force_fun)wsrep_thd_is_BF,
-     (wsrep_abort_thd_fun)wsrep_abort_thd,
-     wsrep_debug, wsrep_convert_LOCK_to_trx,
-     (wsrep_on_fun)wsrep_on);
-
   /*
     Pre-initialize global_system_variables.table_plugin with a dummy engine
     (placeholder) required during the initialization of wsrep threads (THDs).
@@ -1649,10 +1643,10 @@ static bool wsrep_can_run_in_toi(THD *thd, const char *db, const char *table,
 }
 
 /*
-  returns: 
+  returns:
    0: statement was replicated as TOI
    1: TOI replication was skipped
-  -1: TOI replication failed 
+  -1: TOI replication failed
  */
 static int wsrep_TOI_begin(THD *thd, char *db_, char *table_,
                            const TABLE_LIST* table_list,
@@ -1980,10 +1974,11 @@ bool wsrep_grant_mdl_exception(MDL_context *requestor_ctx,
                                MDL_ticket *ticket,
                                const MDL_key *key)
 {
+  THD *request_thd= requestor_ctx->get_thd();
+
   /* Fallback to the non-wsrep behaviour */
-  if (!WSREP_ON) return FALSE;
+  if (!WSREP(request_thd)) return FALSE;
 
-  THD *request_thd= requestor_ctx->get_thd();
   THD *granted_thd= ticket->get_ctx()->get_thd();
   bool ret= false;
 
@@ -2020,6 +2015,7 @@ bool wsrep_grant_mdl_exception(MDL_context *requestor_ctx,
     ticket->wsrep_report(wsrep_debug);
 
     mysql_mutex_lock(&granted_thd->LOCK_thd_data);
+
     if (granted_thd->wsrep_exec_mode == TOTAL_ORDER ||
         granted_thd->wsrep_exec_mode == REPL_RECV)
     {
@@ -2058,8 +2054,8 @@ bool wsrep_grant_mdl_exception(MDL_context *requestor_ctx,
         ticket->wsrep_report(true);
       }
 
-      mysql_mutex_unlock(&granted_thd->LOCK_thd_data);
       wsrep_abort_thd((void *) request_thd, (void *) granted_thd, 1);
+      mysql_mutex_unlock(&granted_thd->LOCK_thd_data);
       ret= false;
     }
   }
@@ -2403,13 +2399,15 @@ void wsrep_close_client_connections(my_bool wait_to_end, THD *except_caller_thd)
     /*
       instead of wsrep_close_thread() we do now  soft kill by THD::awake
      */
+    mysql_mutex_lock(&tmp->LOCK_thd_kill);
     mysql_mutex_lock(&tmp->LOCK_thd_data);
 
     tmp->awake(KILL_CONNECTION);
 
     mysql_mutex_unlock(&tmp->LOCK_thd_data);
-
+    mysql_mutex_unlock(&tmp->LOCK_thd_kill);
   }
+
   mysql_mutex_unlock(&LOCK_thread_count);
 
   if (thread_count)
@@ -2633,11 +2631,11 @@ enum wsrep_exec_mode wsrep_thd_exec_mode(THD *thd)
 
 const char *wsrep_thd_exec_mode_str(THD *thd)
 {
-  return 
+  return
     (!thd) ? "void" :
     (thd->wsrep_exec_mode == LOCAL_STATE)  ? "local"         :
     (thd->wsrep_exec_mode == REPL_RECV)    ? "applier"       :
-    (thd->wsrep_exec_mode == TOTAL_ORDER)  ? "total order"   : 
+    (thd->wsrep_exec_mode == TOTAL_ORDER)  ? "total order"   :
     (thd->wsrep_exec_mode == LOCAL_COMMIT) ? "local commit"  : "void";
 }
 
@@ -2650,12 +2648,12 @@ enum wsrep_query_state wsrep_thd_query_state(THD *thd)
 
 const char *wsrep_thd_query_state_str(THD *thd)
 {
-  return 
-    (!thd) ? "void" : 
+  return
+    (!thd) ? "void" :
     (thd->wsrep_query_state == QUERY_IDLE)        ? "idle"          :
     (thd->wsrep_query_state == QUERY_EXEC)        ? "executing"     :
     (thd->wsrep_query_state == QUERY_COMMITTING)  ? "committing"    :
-    (thd->wsrep_query_state == QUERY_EXITING)     ? "exiting"       : 
+    (thd->wsrep_query_state == QUERY_EXITING)     ? "exiting"       :
     (thd->wsrep_query_state == QUERY_ROLLINGBACK) ? "rolling back"  : "void";
 }
 
@@ -2668,14 +2666,14 @@ enum wsrep_conflict_state wsrep_thd_get_conflict_state(THD *thd)
 
 const char *wsrep_thd_conflict_state_str(THD *thd)
 {
-  return 
+  return
     (!thd) ? "void" :
     (thd->wsrep_conflict_state == NO_CONFLICT)      ? "no conflict"  :
     (thd->wsrep_conflict_state == MUST_ABORT)       ? "must abort"   :
     (thd->wsrep_conflict_state == ABORTING)         ? "aborting"     :
-    (thd->wsrep_conflict_state == MUST_REPLAY)      ? "must replay"  : 
-    (thd->wsrep_conflict_state == REPLAYING)        ? "replaying"    : 
-    (thd->wsrep_conflict_state == RETRY_AUTOCOMMIT) ? "retrying"     : 
+    (thd->wsrep_conflict_state == MUST_REPLAY)      ? "must replay"  :
+    (thd->wsrep_conflict_state == REPLAYING)        ? "replaying"    :
+    (thd->wsrep_conflict_state == RETRY_AUTOCOMMIT) ? "retrying"     :
     (thd->wsrep_conflict_state == CERT_FAILURE)     ? "cert failure" : "void";
 }
 
@@ -2688,6 +2686,7 @@ wsrep_ws_handle_t* wsrep_thd_ws_handle(THD *thd)
 
 void wsrep_thd_LOCK(THD *thd)
 {
+  mysql_mutex_lock(&thd->LOCK_thd_kill);
   mysql_mutex_lock(&thd->LOCK_thd_data);
 }
 
@@ -2695,16 +2694,16 @@ void wsrep_thd_LOCK(THD *thd)
 void wsrep_thd_UNLOCK(THD *thd)
 {
   mysql_mutex_unlock(&thd->LOCK_thd_data);
+  mysql_mutex_unlock(&thd->LOCK_thd_kill);
 }
 
-
-extern "C" time_t wsrep_thd_query_start(THD *thd) 
+extern "C" time_t wsrep_thd_query_start(THD *thd)
 {
   return thd->query_start();
 }
 
 
-extern "C" uint32 wsrep_thd_wsrep_rand(THD *thd) 
+extern "C" uint32 wsrep_thd_wsrep_rand(THD *thd)
 {
   return thd->wsrep_rand;
 }
@@ -2715,7 +2714,7 @@ longlong wsrep_thd_trx_seqno(THD *thd)
 }
 
 
-extern "C" query_id_t wsrep_thd_query_id(THD *thd) 
+extern "C" query_id_t wsrep_thd_query_id(THD *thd)
 {
   return thd->query_id;
 }
@@ -2746,13 +2745,13 @@ const char *wsrep_thd_query(THD *thd)
 }
 
 
-extern "C" query_id_t wsrep_thd_wsrep_last_query_id(THD *thd) 
+extern "C" query_id_t wsrep_thd_wsrep_last_query_id(THD *thd)
 {
   return thd->wsrep_last_query_id;
 }
 
 
-extern "C" void wsrep_thd_set_wsrep_last_query_id(THD *thd, query_id_t id) 
+extern "C" void wsrep_thd_set_wsrep_last_query_id(THD *thd, query_id_t id)
 {
   thd->wsrep_last_query_id= id;
 }
@@ -2763,6 +2762,8 @@ extern "C" void wsrep_thd_awake(THD *thd, my_bool signal)
   if (signal)
   {
     thd->awake(KILL_QUERY);
+    mysql_mutex_unlock(&thd->LOCK_thd_data);
+    mysql_mutex_unlock(&thd->LOCK_thd_kill);
   }
   else
   {
@@ -2842,7 +2843,7 @@ bool wsrep_create_like_table(THD* thd, TABLE_LIST* table,
   if (create_info->tmp_table())
   {
     /* CREATE TEMPORARY TABLE LIKE must be skipped from replication */
-    WSREP_DEBUG("CREATE TEMPORARY TABLE LIKE... skipped replication\n %s", 
+    WSREP_DEBUG("CREATE TEMPORARY TABLE LIKE... skipped replication\n %s",
                 thd->query());
   }
   else if (!(thd->find_temporary_table(src_table)))
@@ -2852,7 +2853,7 @@ bool wsrep_create_like_table(THD* thd, TABLE_LIST* table,
   }
   else
   {
-    /* here we have CREATE TABLE LIKE <temporary table> 
+    /* here we have CREATE TABLE LIKE <temporary table>
        the temporary table definition will be needed in slaves to
        enable the create to succeed
      */
diff --git a/sql/wsrep_thd.cc b/sql/wsrep_thd.cc
index ef8c0e132f7..e4b073d6e46 100644
--- a/sql/wsrep_thd.cc
+++ b/sql/wsrep_thd.cc
@@ -808,6 +808,10 @@ int wsrep_abort_thd(void *bf_thd_ptr, void *victim_thd_ptr, my_bool signal)
   THD *bf_thd     = (THD *) bf_thd_ptr;
   DBUG_ENTER("wsrep_abort_thd");
 
+  /* We need to hold THD::LOCK_thd_data for below wsrep checks. */
+  if (victim_thd)
+    mysql_mutex_assert_owner(&victim_thd->LOCK_thd_data);
+
   if ( (WSREP(bf_thd) ||
          ( (WSREP_ON || bf_thd->variables.wsrep_OSU_method == WSREP_OSU_RSU) &&
            bf_thd->wsrep_exec_mode == TOTAL_ORDER) )                         &&
@@ -826,7 +830,12 @@ int wsrep_abort_thd(void *bf_thd_ptr, void *victim_thd_ptr, my_bool signal)
 
     WSREP_DEBUG("wsrep_abort_thd, by: %llu, victim: %llu", (bf_thd) ?
                 (long long)bf_thd->real_id : 0, (long long)victim_thd->real_id);
+    /* We need to release THD::LOCK_thd_data because below
+    innobase_kill_query will later acquire it and makes sanity
+    checks. */
+    mysql_mutex_unlock(&victim_thd->LOCK_thd_data);
     ha_abort_transaction(bf_thd, victim_thd, signal);
+    mysql_mutex_lock(&victim_thd->LOCK_thd_data);
   }
   else
   {
diff --git a/storage/innobase/handler/ha_innodb.cc b/storage/innobase/handler/ha_innodb.cc
index 00c9595aa87..2a250ac4f24 100644
--- a/storage/innobase/handler/ha_innodb.cc
+++ b/storage/innobase/handler/ha_innodb.cc
@@ -60,7 +60,6 @@ this program; if not, write to the Free Software Foundation, Inc.,
 
 #include <my_service_manager.h>
 #include <key.h>
-#include <sql_manager.h>
 
 /* Include necessary InnoDB headers */
 #include "btr0btr.h"
@@ -5234,30 +5233,60 @@ UNIV_INTERN void lock_cancel_waiting_and_release(lock_t* lock);
 @sa THD::awake() @sa ha_kill_query() */
 static void innobase_kill_query(handlerton*, THD* thd, enum thd_kill_levels)
 {
-	DBUG_ENTER("innobase_kill_query");
+  DBUG_ENTER("innobase_kill_query");
+
 #ifdef WITH_WSREP
-	if (wsrep_thd_get_conflict_state(thd) != NO_CONFLICT) {
-		/* if victim has been signaled by BF thread and/or aborting
-		   is already progressing, following query aborting is not necessary
-		   any more.
-		   Also, BF thread should own trx mutex for the victim, which would
-		   conflict with trx_mutex_enter() below
-		*/
-		DBUG_VOID_RETURN;
-	}
+  if (!thd || wsrep_thd_get_conflict_state(thd) != NO_CONFLICT)
+  {
+    /* if victim has been signaled by BF thread and/or aborting
+    is already progressing, following query aborting is not necessary
+    any more.
+    */
+    DBUG_VOID_RETURN;
+  }
 #endif /* WITH_WSREP */
 
+  /* Here we are holding THD::LOCK_thd_kill to protect from
+  concurrent kill and THD::LOCK_thd_data from concurrent
+  usage.*/
   if (trx_t* trx= thd_to_trx(thd))
   {
-    ut_ad(trx->mysql_thd == thd);
-    lock_mutex_enter();
-    if (lock_t *lock= trx->lock.wait_lock)
+    if (trx_id_t trx_id= trx->id)
     {
-      trx_mutex_enter(trx);
-      lock_cancel_waiting_and_release(lock);
-      trx_mutex_exit(trx);
+      ut_ad(trx->mysql_thd == thd);
+      /* We need to release THD::LOCK_thd_kill and THD::LOCK_thd_data
+      to follow mutexing ordering rules here. Order should be:
+      lock_sys => trx_mutex => THD::LOCK_thd_kill => THD::LOCK_thd_data */
+      wsrep_thd_UNLOCK(thd);
+      /* At this point we are not holding any mutexes that would
+      protect victim thd and victim trx. */
+      lock_mutex_enter();
+      /* Find victim trx by using it's id. This is required because
+      victim could have been rolled back or committed meanwhile and
+      trx pointer deleted. */
+      if (trx_t* victim_trx= trx_rw_is_active(trx_id, NULL, true))
+      {
+        trx_mutex_enter(victim_trx);
+        ut_ad(victim_trx->mysql_thd);
+        wsrep_thd_LOCK(victim_trx->mysql_thd);
+        /* Make sure victim transaction is not meanwhile rolled back
+        or committed. */
+        if (victim_trx->mysql_thd == thd
+            && victim_trx->state == TRX_STATE_ACTIVE)
+        {
+          /* Cancel waiting lock and release it. */
+          if (lock_t *lock= victim_trx->lock.wait_lock)
+            lock_cancel_waiting_and_release(lock);
+        }
+
+        victim_trx->release_reference();
+        trx_mutex_exit(victim_trx);
+        /* Note that THD::LOCK_thd_kill and THD::LOCK_thd_data
+        will be released on kill_one_thread or on
+        wsrep_thd_awake */
+      }
+      lock_mutex_exit();
     }
-    lock_mutex_exit();
   }
 
   DBUG_VOID_RETURN;
@@ -19535,68 +19564,62 @@ wsrep_abort_slave_trx(
 		(long long)bf_seqno, (long long)victim_seqno);
 	abort();
 }
-
-struct bg_wsrep_kill_trx_arg {
-	my_thread_id	thd_id;
-	trx_id_t	trx_id;
-	int64_t		bf_seqno;
-	ibool		signal;
-};
-
-static void bg_wsrep_kill_trx(
-	void *void_arg)
+/*******************************************************************//**
+This function is used to kill one transaction in BF. */
+UNIV_INTERN
+void
+wsrep_innobase_kill_one_trx(
+/*========================*/
+	MYSQL_THD const bf_thd,
+	const trx_t * const bf_trx,
+	trx_t *victim_trx,
+	ibool signal)
 {
-	bg_wsrep_kill_trx_arg *arg = (bg_wsrep_kill_trx_arg*)void_arg;
-	THD *thd 		   = find_thread_by_id(arg->thd_id, false);
-	trx_t *victim_trx 	   = NULL;
-	bool awake 		   = false;
-	DBUG_ENTER("bg_wsrep_kill_trx");
+        ut_ad(bf_thd);
+        ut_ad(victim_trx);
+        ut_ad(lock_mutex_own());
+        ut_ad(trx_mutex_own(victim_trx));
 
-	if (thd) {
-		victim_trx= thd_to_trx(thd);
-		/* Victim trx might not exist e.g. on MDL-conflict. */
-		if (victim_trx) {
-			lock_mutex_enter();
-			trx_mutex_enter(victim_trx);
-			if (victim_trx->id != arg->trx_id ||
-			    victim_trx->state == TRX_STATE_COMMITTED_IN_MEMORY)
-			{
-				/* Victim was meanwhile rolled back or
-				committed */
-				trx_mutex_exit(victim_trx);
-				lock_mutex_exit();
-				wsrep_thd_UNLOCK(thd);
-				victim_trx= NULL;
-			}
-		} else {
-			/* find_thread_by_id locked
-			THD::LOCK_thd_data */
-			wsrep_thd_UNLOCK(thd);
-		}
-	}
+	DBUG_ENTER("wsrep_innobase_kill_one_trx");
+	THD *thd          = (THD *) victim_trx->mysql_thd;
+	int64_t bf_seqno  = wsrep_thd_trx_seqno(bf_thd);
 
-	if (!victim_trx) {
-		/* Victim trx might not exist (MDL-conflict) or victim
-		was meanwhile rolled back or committed because of
-		a KILL statement or a disconnect. */
-		goto ret;
+	if (!thd) {
+		DBUG_PRINT("wsrep", ("no thd for conflicting lock"));
+		WSREP_WARN("no THD for trx: " TRX_ID_FMT, victim_trx->id);
+		DBUG_VOID_RETURN;
 	}
 
+	WSREP_LOG_CONFLICT(bf_thd, thd, TRUE);
+
 	WSREP_DEBUG("BF kill (" ULINTPF ", seqno: " INT64PF
 		    "), victim: (%lu) trx: " TRX_ID_FMT,
-		    arg->signal, arg->bf_seqno,
+		    signal, bf_seqno,
 		    thd_get_thread_id(thd),
 		    victim_trx->id);
 
 	WSREP_DEBUG("Aborting query: %s conf %d trx: %" PRId64,
-		    (wsrep_thd_query(thd)) ? wsrep_thd_query(thd) : "void",
+		    (thd && wsrep_thd_query(thd)) ? wsrep_thd_query(thd) : "void",
 		    wsrep_thd_conflict_state(thd, FALSE),
 		    wsrep_thd_ws_handle(thd)->trx_id);
 
+	/* Here we will lock THD::LOCK_thd_kill and THD::LOCK_thd_data */
+	wsrep_thd_LOCK(thd);
+
+        DBUG_EXECUTE_IF("sync.wsrep_after_BF_victim_lock",
+	{
+		const char act[]=
+			"now "
+			"wait_for signal.wsrep_after_BF_victim_lock";
+		DBUG_ASSERT(!debug_sync_set_action(bf_thd,
+			    STRING_WITH_LEN(act)));
+	};);
+
 	if (wsrep_thd_query_state(thd) == QUERY_EXITING) {
 		WSREP_DEBUG("kill trx EXITING for " TRX_ID_FMT,
 			    victim_trx->id);
-		goto ret_unlock;
+		wsrep_thd_UNLOCK(thd);
+		DBUG_VOID_RETURN;
 	}
 
 	if (wsrep_thd_exec_mode(thd) != LOCAL_STATE) {
@@ -19612,13 +19635,17 @@ static void bg_wsrep_kill_trx(
         case MUST_ABORT:
 		WSREP_DEBUG("victim " TRX_ID_FMT " in MUST ABORT state",
 			    victim_trx->id);
-		goto ret_awake;
+		wsrep_thd_awake(thd, signal);
+		DBUG_VOID_RETURN;
+		break;
 	case ABORTED:
 	case ABORTING: // fall through
 	default:
 		WSREP_DEBUG("victim " TRX_ID_FMT " in state %d",
 			    victim_trx->id, wsrep_thd_get_conflict_state(thd));
-		goto ret_unlock;
+		wsrep_thd_UNLOCK(thd);
+		DBUG_VOID_RETURN;
+		break;
 	}
 
 	switch (wsrep_thd_query_state(thd)) {
@@ -19631,12 +19658,12 @@ static void bg_wsrep_kill_trx(
 			    victim_trx->id);
 
 		if (wsrep_thd_exec_mode(thd) == REPL_RECV) {
-			wsrep_abort_slave_trx(arg->bf_seqno,
+			wsrep_abort_slave_trx(bf_seqno,
 					      wsrep_thd_trx_seqno(thd));
 		} else {
 			wsrep_t *wsrep= get_wsrep();
 			rcode = wsrep->abort_pre_commit(
-				wsrep, arg->bf_seqno,
+				wsrep, bf_seqno,
 				(wsrep_trx_id_t)wsrep_thd_ws_handle(thd)->trx_id
 			);
 
@@ -19645,7 +19672,9 @@ static void bg_wsrep_kill_trx(
 				WSREP_DEBUG("cancel commit warning: "
 					    TRX_ID_FMT,
 					    victim_trx->id);
-				goto ret_awake;
+				wsrep_thd_awake(thd, signal);
+				DBUG_VOID_RETURN;
+				break;
 			case WSREP_OK:
 				break;
 			default:
@@ -19658,9 +19687,11 @@ static void bg_wsrep_kill_trx(
 				 * kill the lock holder first.
 				 */
 				abort();
+				break;
 			}
 		}
-		goto ret_awake;
+		wsrep_thd_awake(thd, signal);
+		break;
 	case QUERY_EXEC:
 		/* it is possible that victim trx is itself waiting for some
 		 * other lock. We need to cancel this waiting
@@ -19681,20 +19712,22 @@ static void bg_wsrep_kill_trx(
 				lock_cancel_waiting_and_release(wait_lock);
 			}
 
+			wsrep_thd_awake(thd, signal);
 		} else {
 			/* abort currently executing query */
 			DBUG_PRINT("wsrep",("sending KILL_QUERY to: %lu",
                                             thd_get_thread_id(thd)));
 			WSREP_DEBUG("kill query for: %ld",
 				thd_get_thread_id(thd));
+			wsrep_thd_awake(thd, signal);
 
 			/* for BF thd, we need to prevent him from committing */
 			if (wsrep_thd_exec_mode(thd) == REPL_RECV) {
-				wsrep_abort_slave_trx(arg->bf_seqno,
+				wsrep_abort_slave_trx(bf_seqno,
 						    wsrep_thd_trx_seqno(thd));
 			}
 		}
-		goto ret_awake;
+		break;
 	case QUERY_IDLE:
 	{
 		WSREP_DEBUG("kill IDLE for " TRX_ID_FMT, victim_trx->id);
@@ -19702,9 +19735,9 @@ static void bg_wsrep_kill_trx(
 		if (wsrep_thd_exec_mode(thd) == REPL_RECV) {
 			WSREP_DEBUG("kill BF IDLE, seqno: %lld",
 				    (long long)wsrep_thd_trx_seqno(thd));
-			wsrep_abort_slave_trx(arg->bf_seqno,
+			wsrep_abort_slave_trx(bf_seqno,
 					      wsrep_thd_trx_seqno(thd));
-			goto ret_unlock;
+			DBUG_VOID_RETURN;
 		}
                 /* This will lock thd from proceeding after net_read() */
 		wsrep_thd_set_conflict_state(thd, ABORTING);
@@ -19725,67 +19758,17 @@ static void bg_wsrep_kill_trx(
 		DBUG_PRINT("wsrep",("signalling wsrep rollbacker"));
 		WSREP_DEBUG("signaling aborter");
 		wsrep_unlock_rollback();
-		goto ret_unlock;
+		wsrep_thd_UNLOCK(thd);
+
+		break;
 	}
 	default:
 		WSREP_WARN("bad wsrep query state: %d",
 			  wsrep_thd_query_state(thd));
-		goto ret_unlock;
+		wsrep_thd_UNLOCK(thd);
+		break;
 	}
 
-ret_awake:
-	awake= true;
-
-ret_unlock:
-	trx_mutex_exit(victim_trx);
-	lock_mutex_exit();
-	if (awake)
-		wsrep_thd_awake(thd, arg->signal);
-	wsrep_thd_UNLOCK(thd);
-
-ret:
-	free(arg);
-	DBUG_VOID_RETURN;
-
-}
-
-/*******************************************************************//**
-This function is used to kill one transaction in BF. */
-UNIV_INTERN
-void
-wsrep_innobase_kill_one_trx(
-/*========================*/
-	MYSQL_THD const bf_thd,
-	const trx_t * const bf_trx,
-	trx_t *victim_trx,
-	ibool signal)
-{
-	ut_ad(bf_thd);
-	ut_ad(victim_trx);
-	ut_ad(lock_mutex_own());
-	ut_ad(trx_mutex_own(victim_trx));
-
-	bg_wsrep_kill_trx_arg *arg = (bg_wsrep_kill_trx_arg*)malloc(sizeof(*arg));
-	arg->thd_id	= thd_get_thread_id(victim_trx->mysql_thd);
-	arg->trx_id	= victim_trx->id;
-	arg->bf_seqno	= wsrep_thd_trx_seqno((THD*)bf_thd);
-	arg->signal	= signal;
-
-	DBUG_ENTER("wsrep_innobase_kill_one_trx");
-
-	WSREP_LOG_CONFLICT(bf_thd, victim_trx->mysql_thd, TRUE);
-
-	DBUG_EXECUTE_IF("sync.wsrep_after_BF_victim_lock",
-		{
-		  const char act[]=
-		    "now "
-		    "wait_for signal.wsrep_after_BF_victim_lock";
-		  DBUG_ASSERT(!debug_sync_set_action(bf_thd,
-						     STRING_WITH_LEN(act)));
-		};);
-
-
-	mysql_manager_submit(bg_wsrep_kill_trx, arg);
 	DBUG_VOID_RETURN;
 }
 
@@ -19821,7 +19804,6 @@ wsrep_abort_transaction(
 		wsrep_thd_LOCK(victim_thd);
 		wsrep_thd_set_conflict_state(victim_thd, MUST_ABORT);
 		wsrep_thd_awake(victim_thd, signal);
-		wsrep_thd_UNLOCK(victim_thd);
 	}
 
 	DBUG_VOID_RETURN;