Bug#26336636 USE STD::ATOMIC RATHER THAN VOLATILE WHEN INSPECTING STA…

…TUS OF OTHER THREADS

Summary:
Backport Bug#26336636 USE STD::ATOMIC RATHER THAN VOLATILE WHEN
INSPECTING STATUS OF OTHER THREADS to make THD::killed thread safe

Also include changes to MyRocks that check the killed state.

Binlog commit is modified to check the kill flag without checking
shutdown_in_progress, which is not atomic.

Also add new testcase to test semisync shutdown disconnect path.

Porting notes: when porting to 8.0.22+ the changes for MyRocks and
test cases should be kept.

Reviewed By: yizhang82

Differential Revision: D25773954

fbshipit-source-id: 722a5262b62

mysql-test/r/disconnect_on_shutdown_semisync.result

@@ -0,0 +1,32 @@
+set @save_master_timeout=@@global.rpl_semi_sync_master_timeout;
+set @save_master_wait_no_slave=@@global.rpl_semi_sync_master_wait_no_slave;
+set @save_master_enabled=@@global.rpl_semi_sync_master_enabled;
+call mtr.add_suppression("SEMISYNC: Forced shutdown. Some updates might not be replicated.");
+CREATE TABLE t1 (a INT) ENGINE = INNODB;
+INSERT INTO t1 VALUES (1), (2);
+[ enable semi-sync on master ]
+set global rpl_semi_sync_master_timeout= 600000 /* 600s */;
+set global rpl_semi_sync_master_enabled = 1;
+show variables like 'rpl_semi_sync_master_enabled';
+Variable_name	Value
+rpl_semi_sync_master_enabled	ON
+[ status of semi-sync on master should be ON even without any semi-sync slaves ]
+show status like 'Rpl_semi_sync_master_clients';
+Variable_name	Value
+Rpl_semi_sync_master_clients	0
+show status like 'Rpl_semi_sync_master_status';
+Variable_name	Value
+Rpl_semi_sync_master_status	ON
+show status like 'Rpl_semi_sync_master_yes_tx';
+Variable_name	Value
+Rpl_semi_sync_master_yes_tx	0
+Should wait for semisync ack
+INSERT INTO t1 VALUES (3);
+Checking for thread to wait
+Shutting down the server
+ERROR HY000: Lost connection to MySQL server during query
+#
+# Clean up
+#
+# restart the server
+DROP TABLE t1;

mysql-test/t/disconnect_on_shutdown_semisync-master.opt

@@ -0,0 +1,2 @@
+$SEMISYNC_PLUGIN_OPT
+--force-restart

mysql-test/t/disconnect_on_shutdown_semisync.test

@@ -0,0 +1,56 @@
+source include/have_semisync.inc;
+source include/not_embedded.inc;
+source include/have_innodb.inc;
+source include/have_log_bin.inc;
+
+set @save_master_timeout=@@global.rpl_semi_sync_master_timeout;
+set @save_master_wait_no_slave=@@global.rpl_semi_sync_master_wait_no_slave;
+set @save_master_enabled=@@global.rpl_semi_sync_master_enabled;
+
+call mtr.add_suppression("SEMISYNC: Forced shutdown. Some updates might not be replicated.");
+CREATE TABLE t1 (a INT) ENGINE = INNODB;
+INSERT INTO t1 VALUES (1), (2);
+
+echo [ enable semi-sync on master ];
+set global rpl_semi_sync_master_timeout= 600000 /* 600s */;
+set global rpl_semi_sync_master_enabled = 1;
+show variables like 'rpl_semi_sync_master_enabled';
+
+echo [ status of semi-sync on master should be ON even without any semi-sync slaves ];
+show status like 'Rpl_semi_sync_master_clients';
+show status like 'Rpl_semi_sync_master_status';
+show status like 'Rpl_semi_sync_master_yes_tx';
+
+connect (con1,localhost,root,,);
+connection con1;
+--echo Should wait for semisync ack
+--send INSERT INTO t1 VALUES (3)
+
+connection default;
+--echo Checking for thread to wait
+let $wait_condition=
+select count(*)>0 from information_schema.processlist where state='Waiting for semi-sync ACK from slave';
+source include/wait_condition.inc;
+
+--echo Shutting down the server
+--exec echo "wait" > $MYSQLTEST_VARDIR/tmp/mysqld.1.expect
+--exec $MYSQL -e "shutdown;" 2>&1
+--source include/wait_until_disconnected.inc
+
+connection con1;
+--error 2013
+--reap
+disconnect con1;
+
+--echo #
+--echo # Clean up
+--echo #
+
+connection default;
+
+--echo # restart the server
+--exec echo "restart:" > $MYSQLTEST_VARDIR/tmp/mysqld.1.expect
+--enable_reconnect
+--source include/wait_until_connected_again.inc
+
+DROP TABLE t1;

sql/binlog.cc

@@ -10022,13 +10022,7 @@ MYSQL_BIN_LOG::finish_commit(THD *thd, bool async)
     transactions that are in the commit pipeline
   */
   DEBUG_SYNC(thd, "commit_wait_for_shutdown");
-  DBUG_EXECUTE_IF("commit_on_shutdown_testing", {
-    /* For testing, set killed flag if shutdown is in progress */
-    if (shutdown_in_progress)
-      thd->killed = THD::KILL_CONNECTION;
-  };);
-  if (shutdown_in_progress && !thd->slave_thread &&
-      thd->killed == THD::KILL_CONNECTION) {
+  if (!thd->slave_thread && thd->killed == THD::KILL_CONNECTION) {
     thd->disconnect();
   }
 

sql/debug_sync.cc

@@ -1887,7 +1887,7 @@ static void debug_sync_execute(THD *thd, st_debug_sync_action *action)
                    if (thd->killed)
                      DBUG_PRINT("debug_sync_exec",
                                 ("killed %d from '%s'  at: '%s'",
-                                 thd->killed, sig_wait, dsp_name));
+                                 thd->killed.load(), sig_wait, dsp_name));
                    else
                      DBUG_PRINT("debug_sync_exec",
                                 ("%s from '%s'  at: '%s'",

sql/event_queue.cc

@@ -577,7 +577,7 @@ Event_queue::get_top_for_execution_if_time(THD *thd,
     /* Break loop if thd has been killed */
     if (thd->killed)
     {
-      DBUG_PRINT("info", ("thd->killed=%d", thd->killed));
+      DBUG_PRINT("info", ("thd->killed=%d", thd->killed.load()));
       goto end;
     }
 

sql/filesort.cc

@@ -765,7 +765,7 @@ static ha_rows find_all_keys(Sort_param *param, SQL_SELECT *select,
   my_off_t record;
   TABLE *sort_form;
   THD *thd= current_thd;
-  volatile THD::killed_state *killed= &thd->killed;
+  std::atomic<THD::killed_state> *killed= &thd->killed;
   handler *file;
   MY_BITMAP *save_read_set, *save_write_set;
   bool skip_record;
@@ -1677,14 +1677,14 @@ int merge_buffers(Sort_param *param, IO_CACHE *from_file,
   QUEUE queue;
   qsort2_cmp cmp;
   void *first_cmp_arg;
-  volatile THD::killed_state *killed= &current_thd->killed;
-  THD::killed_state not_killable;
+  std::atomic<THD::killed_state> *killed= &current_thd->killed;
+  std::atomic<THD::killed_state> not_killable;
   DBUG_ENTER("merge_buffers");
 
   current_thd->inc_status_sort_merge_passes();
   if (param->not_killable)
   {
-    killed= &not_killable;
+   killed= &not_killable;
     not_killable= THD::NOT_KILLED;
   }
 

sql/mysqld.cc

@@ -2155,6 +2155,16 @@ static void close_connections(void)
   }
   mysql_mutex_unlock(&LOCK_connection_count);
 
+  DBUG_EXECUTE_IF("commit_on_shutdown_testing", {
+    THD *thd = new THD();
+    thd->thread_stack = reinterpret_cast<char *>(&(thd));
+    thd->store_globals();
+    const char act[] = "now signal shutdown_started";
+    DBUG_ASSERT(opt_debug_sync_timeout > 0);
+    DBUG_ASSERT(!debug_sync_set_action(current_thd, STRING_WITH_LEN(act)));
+    thd->restore_globals();
+    delete thd;
+  });
   close_active_mi();
   DBUG_PRINT("quit",("close_connections thread"));
   DBUG_VOID_RETURN;
@@ -2256,12 +2266,6 @@ void kill_mysql(void)
       sql_print_error("Can't create thread to kill server (errno= %d).", error);
   }
 #endif
-  DBUG_EXECUTE_IF("commit_on_shutdown_testing", {
-    const char act[] = "now signal shutdown_started";
-    DBUG_ASSERT(opt_debug_sync_timeout > 0);
-    DBUG_ASSERT(!debug_sync_set_action(current_thd, STRING_WITH_LEN(act)));
-  };);
-
   DBUG_VOID_RETURN;
 }
 

sql/rpl_rli_pdb.cc

@@ -2774,7 +2774,7 @@ int slave_worker_exec_job(Slave_worker *worker, Relay_log_info *rli)
     if (log_warnings > 1)
       sql_print_information("Worker %lu is exiting: killed %i, error %i, "
                             "running_status %d",
-                            worker->id, thd->killed, thd->is_error(),
+                            worker->id, thd->killed.load(), thd->is_error(),
                             worker->running_status.load());
     worker->slave_worker_ends_group(ev, error, temporary_error);
   }

sql/signal_handler.cc

@@ -163,7 +163,7 @@ extern "C" sig_handler handle_fatal_signal(int sig)
   if (thd)
   {
     const char *kreason= "UNKNOWN";
-    switch (thd->killed) {
+    switch (thd->killed.load()) {
     case THD::NOT_KILLED:
       kreason= "NOT_KILLED";
       break;

sql/sp_head.cc

@@ -756,7 +756,7 @@ bool sp_head::execute(THD *thd, bool merge_da_on_success)
 
  done:
   DBUG_PRINT("info", ("err_status: %d  killed: %d  is_slave_error: %d  report_error: %d",
-                      err_status, thd->killed, thd->is_slave_error,
+                      err_status, thd->killed.load(), thd->is_slave_error,
                       thd->is_error()));
 
   if (thd->killed)

sql/sql_class.h

@@ -3902,7 +3902,7 @@ class THD :public MDL_context_owner,
     ABORT_QUERY=ER_QUERY_EXCEEDED_ROWS_EXAMINED_LIMIT,
     KILLED_NO_VALUE      /* means neither of the states */
   };
-  killed_state volatile killed;
+  std::atomic<killed_state> killed;
 
   /* scramble - random string sent to client on handshake */
   char	     scramble[SCRAMBLE_LENGTH+1];

sql/sql_delete.cc

@@ -1051,7 +1051,7 @@ bool multi_delete::send_eof()
 
   /* compute a total error to know if something failed */
   local_error= local_error || error;
-  killed_status= (local_error == 0)? THD::NOT_KILLED : thd->killed;
+  killed_status= (local_error == 0)? THD::NOT_KILLED : thd->killed.load();
   /* reset used flags */
   THD_STAGE_INFO(thd, stage_end);
 

sql/sql_load.cc

@@ -590,7 +590,7 @@ int mysql_load(THD *thd,sql_exchange *ex,TABLE_LIST *table_list,
                   };);
 
 #ifndef EMBEDDED_LIBRARY
-  killed_status= (error == 0) ? THD::NOT_KILLED : thd->killed;
+  killed_status= (error == 0) ? THD::NOT_KILLED : thd->killed.load();
 #endif
 
   /*

sql/sql_update.cc

@@ -2449,7 +2449,7 @@ bool multi_update::send_eof()
     if local_error is not set ON until after do_updates() then
     later carried out killing should not affect binlogging.
   */
-  killed_status= (local_error == 0)? THD::NOT_KILLED : thd->killed;
+  killed_status= (local_error == 0)? THD::NOT_KILLED : thd->killed.load();
   THD_STAGE_INFO(thd, stage_end);
 
   /* We must invalidate the query cache before binlog writing and

storage/rocksdb/ha_rocksdb.cc

@@ -12456,7 +12456,7 @@ static int calculate_cardinality_table_scan(
         &to_recalc,
     std::unordered_map<GL_INDEX_ID, Rdb_index_stats> *stats,
     table_cardinality_scan_type scan_type, uint64_t max_num_rows_scanned,
-    THD::killed_state volatile *killed) {
+    std::atomic<THD::killed_state> *killed) {
   DBUG_ENTER_FUNC();
 
   DBUG_ASSERT(scan_type != SCAN_TYPE_NONE);
@@ -12674,7 +12674,8 @@ static int read_stats_from_ssts(
 static int calculate_stats(
     const std::unordered_map<GL_INDEX_ID, std::shared_ptr<const Rdb_key_def>>
         &to_recalc,
-    table_cardinality_scan_type scan_type, THD::killed_state volatile *killed) {
+    table_cardinality_scan_type scan_type,
+    std::atomic<THD::killed_state> *killed) {
   DBUG_ENTER_FUNC();
 
   std::unordered_map<GL_INDEX_ID, Rdb_index_stats> stats;
@@ -12711,7 +12712,7 @@ static int calculate_stats(
 
 static int calculate_stats_for_table(
     const std::string &tbl_name, table_cardinality_scan_type scan_type,
-    THD::killed_state volatile *killed = nullptr) {
+    std::atomic<THD::killed_state> *killed = nullptr) {
   DBUG_ENTER_FUNC();
   std::unordered_map<GL_INDEX_ID, std::shared_ptr<const Rdb_key_def>> to_recalc;
   std::vector<GL_INDEX_ID> indexes;

storage/rocksdb/rdb_threads.h

@@ -53,9 +53,7 @@ class Rdb_thread {
   mysql_mutex_t m_signal_mutex;
   mysql_cond_t m_signal_cond;
 
-  // TODO: When porting to 8.0 we should move to std::atomic
-  // instead of volatile
-  THD::killed_state volatile m_killed;
+  std::atomic<THD::killed_state> m_killed;
 
  public:
   Rdb_thread() : m_run_once(false), m_killed(THD::NOT_KILLED) {}