Disallow KILL for wsrep transactions in committing state

When wsrep transaction has reached committing state,
it must become immune for KILL. Accompanied MTR test
can be used to reproduce assertion failure if the
transaction is not immune.

Implemented additional protection in wsrep_kill_thd()
against KILL in committing states.

mysql-test/suite/galera/t/galera_kill_group_commit.cnf

@@ -0,0 +1,5 @@
+!include ../galera_2nodes.cnf
+
+[mysqld]
+log-bin
+log-slave-updates

mysql-test/suite/galera/t/galera_kill_group_commit.test

@@ -0,0 +1,72 @@
+#
+# Verify that transaction which has reached group commit queue
+# cannot be killed. If the kill succeeds, assertion for
+# wsrep transaction state will fail.
+#
+# If the bug is present, i.e. wsrep transaction gets killed during
+# group commit wait, this test is enough to reproduce the crash
+# most of the time.
+#
+
+--source include/have_innodb.inc
+--source include/have_debug_sync.inc
+--source include/galera_cluster.inc
+
+# Connection for KILL commands
+--connect node_1_kill, 127.0.0.1, root, , test, $NODE_MYPORT_1
+# Connection for sync point control
+--connect node_1_ctrl, 127.0.0.1, root, , test, $NODE_MYPORT_1
+SET SESSION wsrep_sync_wait = 0;
+# Connection for group commit follower
+--connect node_1_follower, 127.0.0.1, root, , test, $NODE_MYPORT_1
+# Need to disable sync wait to reach commit queue when leader
+# is blocked.
+SET SESSION wsrep_sync_wait = 0;
+--let $follower_id = `SELECT CONNECTION_ID()`
+
+--connection node_1
+CREATE TABLE t1 (f1 INT PRIMARY KEY) ENGINE=InnoDB;
+--let $binlog_group_commits_orig = `SELECT VARIABLE_VALUE FROM information_schema.global_status WHERE VARIABLE_NAME = 'Binlog_group_commits'`
+
+SET SESSION DEBUG_SYNC = "commit_before_enqueue SIGNAL leader_before_enqueue_reached WAIT_FOR leader_before_enqueue_continue";
+--send INSERT INTO t1 VALUES (1)
+
+--connection node_1_ctrl
+SET DEBUG_SYNC = "now WAIT_FOR leader_before_enqueue_reached";
+
+--connection node_1_follower
+# SET SESSION DEBUG_SYNC = "group_commit_waiting_for_prior SIGNAL follower_waiting_for_prior_reached WAIT_FOR follower_waiting_for_prior_continue";
+--send INSERT INTO t1 VALUES (2);
+
+--connection node_1_ctrl
+# TODO: Is it possible to use sync points to enforce group commit to happen?
+# The leader will hold commit monitor in commit_before_enqueue sync point,
+# which prevents the follower to reach the group commit wait state.
+# We now sleep and expect the follower to reach group commit, but this
+# may cause false negatives.
+--sleep 1
+
+--connection node_1_kill
+--echo # Execute KILL QUERY for group commit follower
+--disable_query_log
+--disable_result_log
+# Because it is currently impossible to verify that the
+# follower has reached group commit queue, the KILL may
+# sometimes return success.
+--error 0,ER_KILL_DENIED_ERROR
+--eval KILL QUERY $follower_id
+--enable_result_log
+--enable_query_log
+
+SET DEBUG_SYNC = "now SIGNAL leader_before_enqueue_continue";
+--connection node_1_follower
+--reap
+
+--connection node_1
+--reap
+SELECT * FROM t1;
+
+SET DEBUG_SYNC = "RESET";
+DROP TABLE t1;
+
+--eval SELECT VARIABLE_VALUE - $binlog_group_commits_orig AS expect_2 FROM information_schema.global_status WHERE VARIABLE_NAME = 'Binlog_group_commits'

sql/sql_parse.cc

@@ -9328,17 +9328,9 @@ kill_one_thread(THD *thd, my_thread_id id, killed_state kill_signal, killed_type
 #ifdef WITH_WSREP
         if (WSREP(tmp))
         {
-          /*
-            Mark killed thd with kill_signal so that awake_no_mutex does
-            not dive into storage engine. We use ha_abort_transaction()
-            to do the storage engine part for wsrep THDs.
-          */
-          tmp->wsrep_abort_by_kill= kill_signal;
-          tmp->awake_no_mutex(kill_signal);
-          /* ha_abort_transaction() releases tmp->LOCK_thd_kill, so tmp
-             is not safe to access anymore. */
-          ha_abort_transaction(thd, tmp, 1);
-          DBUG_RETURN(0);
+          /* Object tmp is not guaranteed to exist after wsrep_kill_thd()
+             returns, so do early return from this function. */
+          DBUG_RETURN(wsrep_kill_thd(thd, tmp, kill_signal, type));
         }
         else
 #endif /* WITH_WSREP */

sql/wsrep_thd.cc

@@ -426,6 +426,33 @@ bool wsrep_bf_abort(THD* bf_thd, THD* victim_thd)
   return wsrep_bf_abort_low(bf_thd, victim_thd);
 }
 
+uint wsrep_kill_thd(THD *thd, THD *victim_thd, killed_state kill_signal, killed_type type)
+{
+  DBUG_ENTER("wsrep_kill_thd");
+  DBUG_ASSERT(WSREP(victim_thd));
+  mysql_mutex_assert_owner(&victim_thd->LOCK_thd_kill);
+  mysql_mutex_assert_owner(&victim_thd->LOCK_thd_data);
+  auto trx_state= victim_thd->wsrep_trx().state();
+  if (trx_state == wsrep::transaction::state::s_committing ||
+      trx_state == wsrep::transaction::state::s_ordered_commit) {
+    mysql_mutex_unlock(&victim_thd->LOCK_thd_data);
+    mysql_mutex_unlock(&victim_thd->LOCK_thd_kill);
+    DBUG_RETURN(type == KILL_TYPE_QUERY ? ER_KILL_QUERY_DENIED_ERROR :
+                ER_KILL_DENIED_ERROR);
+  }
+  /*
+    Mark killed victim_thd with kill_signal so that awake_no_mutex does
+    not dive into storage engine. We use ha_abort_transaction()
+    to do the storage engine part for wsrep THDs.
+  */
+  victim_thd->wsrep_abort_by_kill= kill_signal;
+  victim_thd->awake_no_mutex(kill_signal);
+  /* ha_abort_transaction() releases tmp->LOCK_thd_kill, so tmp
+     is not safe to access anymore. */
+  ha_abort_transaction(thd, victim_thd, 1);
+  DBUG_RETURN(0);
+}
+
 int wsrep_create_threadvars()
 {
   int ret= 0;

sql/wsrep_thd.h

@@ -99,6 +99,24 @@ void wsrep_abort_thd(THD *bf_thd,
                      THD *victim_thd,
                      my_bool signal) __attribute__((nonnull(1,2)));
 
+/**
+  Kill wsrep connection with kill_signal. Object thd is not
+  guaranteed to exist anymore when this function returns.
+
+  Asserts that the caller holds victim_thd->LOCK_thd_kill,
+  victim_thd->LOCK_thd_data.
+
+  Releases victim_thd->LOCK_thd_kill, victim_thd->LOCK_thd_data.
+
+  @param thd THD object for connection that executes the KILL.
+  @param victim_thd THD object for connection to be killed.
+  @param kill_signal Kill signal.
+  @param type Killed type
+
+  Return zero if the kill was successful, otherwise non-zero error code.
+ */
+uint wsrep_kill_thd(THD *thd, THD *victim_thd, killed_state kill_signal, killed_type type);
+
 /*
   Helper methods to deal with thread local storage.
   The purpose of these methods is to hide the details of thread

storage/innobase/handler/ha_innodb.cc

@@ -4154,7 +4154,6 @@ innobase_commit_low(
 	if (is_wsrep) {
 		tmp = thd_proc_info(thd, "innobase_commit_low()");
 	}
-	DEBUG_SYNC(thd, "wsrep_innobase_commit_low");
 #endif /* WITH_WSREP */
 	if (trx_is_started(trx)) {
 		trx_commit_for_mysql(trx);