MDEV-18450 Slaves wait shutdown

The patches features an optional shutdown behavior to hold on until after all connected slaves have been sent the last binlogged event. The connected slave is one whose START SLAVE has been acknowledged and that was not stopped since that though it could be technically reconnecting in background. The solution therefore disallows killing the dump thread until is has found EOF of the latest binlog file. It is up to the shutdown requester (DBA) to set up a sufficiently large shutdown timeout value for shudown to wait patiently until lagging behind slaves have been synchronized. On the other hand if a specific slave needs exclusion from synchronization the DBA would have to stop it manually which would terminate its dump thread. `mysqladmin shutdown' is extended with a `--wait_for_all_slaves' option which translates to `SHUTDOW WAIT FOR ALL SLAVES' sql query to enable the feature on the client side. The patch also performs a small refactoring of the server shutdown around close_connections() to introduce kill thread phases which are two as of current.
2025-07-29 05:21:33 +03:00 · 2019-03-07 08:12:26 +04:00
parent e450527938
commit 3568427d11
19 changed files with 459 additions and 54 deletions
--- a/client/client_priv.h
+++ b/client/client_priv.h
@ -100,6 +100,7 @@ enum options_client
  OPT_SKIP_ANNOTATE_ROWS_EVENTS,
  OPT_SSL_CRL, OPT_SSL_CRLPATH,
  OPT_PRINT_ROW_COUNT, OPT_PRINT_ROW_EVENT_POSITIONS,
+  OPT_SHUTDOWN_WAIT_FOR_SLAVES,
  OPT_MAX_CLIENT_OPTION /* should be always the last */
 };

--- a/client/mysqladmin.cc
+++ b/client/mysqladmin.cc
@ -40,7 +40,8 @@ ulonglong last_values[MAX_MYSQL_VAR+100];
 static int interval=0;
 static my_bool option_force=0,interrupted=0,new_line=0,
               opt_compress= 0, opt_local= 0, opt_relative= 0, opt_verbose= 0,
-               opt_vertical= 0, tty_password= 0, opt_nobeep;
+               opt_vertical= 0, tty_password= 0, opt_nobeep,
+               opt_shutdown_wait_for_slaves= 0;
 static my_bool debug_info_flag= 0, debug_check_flag= 0;
 static uint tcp_port = 0, option_wait = 0, option_silent=0, nr_iterations;
 static uint opt_count_iterations= 0, my_end_arg;
@ -218,6 +219,11 @@ static struct my_option my_long_options[] =
  {"shutdown_timeout", OPT_SHUTDOWN_TIMEOUT, "", &opt_shutdown_timeout,
   &opt_shutdown_timeout, 0, GET_ULONG, REQUIRED_ARG,
   SHUTDOWN_DEF_TIMEOUT, 0, 3600*12, 0, 1, 0},
+  {"wait_for_all_slaves", OPT_SHUTDOWN_WAIT_FOR_SLAVES,
+   "Defers shutdown until after all binlogged events have been sent to "
+   "all connected slaves", &opt_shutdown_wait_for_slaves,
+   &opt_shutdown_wait_for_slaves, 0, GET_BOOL, NO_ARG, 0, 0, 0,
+   0, 0, 0},
  {"plugin_dir", OPT_PLUGIN_DIR, "Directory for client-side plugins.",
    &opt_plugin_dir, &opt_plugin_dir, 0,
   GET_STR, REQUIRED_ARG, 0, 0, 0, 0, 0, 0},
@ -693,7 +699,17 @@ static int execute_commands(MYSQL *mysql,int argc, char **argv)
 	  !stat(pidfile, &pidfile_status))
 	last_modified= pidfile_status.st_mtime;

-      if (mysql_shutdown(mysql, SHUTDOWN_DEFAULT))
+      if (opt_shutdown_wait_for_slaves)
+      {
+        sprintf(buff, "SHUTDOWN WAIT FOR ALL SLAVES");
+        if (mysql_query(mysql, buff))
+        {
+          my_printf_error(0, "%s failed; error: '%-.200s'",
+                          error_flags, buff, mysql_error(mysql));
+          return -1;
+        }
+      }
+      else if (mysql_shutdown(mysql, SHUTDOWN_DEFAULT))
      {
 	my_printf_error(0, "shutdown failed; error: '%s'", error_flags,
 			mysql_error(mysql));
--- a/mysql-test/mysql-test-run.pl
+++ b/mysql-test/mysql-test-run.pl
@ -4550,7 +4550,8 @@ sub extract_warning_lines ($$) {
     qr/InnoDB: See also */,
     qr/InnoDB: Cannot open .*ib_buffer_pool.* for reading: No such file or directory*/,
     qr/InnoDB: Table .*mysql.*innodb_table_stats.* not found./,
-     qr/InnoDB: User stopword table .* does not exist./
+     qr/InnoDB: User stopword table .* does not exist./,
+     qr/Dump thread [0-9]+ last sent to server [0-9]+ binlog file:pos .+/

    );

--- a/mysql-test/suite/rpl/include/rpl_shutdown_wait_slaves.inc
+++ b/mysql-test/suite/rpl/include/rpl_shutdown_wait_slaves.inc
@ -0,0 +1,83 @@
+--connection server_1
+
+CREATE TABLE t1 (a INT) ENGINE=innodb;
+
+--save_master_pos
+
+--connection server_2
+--sync_with_master
+
+--connection server_3
+--sync_with_master
+
+--connection server_4
+--source include/stop_slave.inc
+
+--connection server_1
+--disable_query_log
+--let $count=1000
+while ($count)
+{
+ INSERT INTO t1 SET a=1;
+ --dec $count
+}
+--enable_query_log
+--save_master_pos
+
+# Shutdown master and restart server_4 who will be waiting for the master
+# to start replication at its shutdown beginning phase.
+# The being forked out server_4 dump thread must relate to a record
+# in slave_list, and it won't start sending out binlog events
+# until has received a signal from the shutdown thread.
+# This also proves delivery to a  started-in-middle-of-shutdown slave.
+--connection server_1
+SET @@GLOBAL.debug_dbug="+d,simulate_delay_at_shutdown";
+
+--connection server_4
+--source include/start_slave.inc
+
+--connection server_1
+--write_file $MYSQLTEST_VARDIR/tmp/mysqld.1.expect
+wait
+EOF
+# --shutdown_server 60
+--send SHUTDOWN WAIT FOR ALL SLAVES
+--reap
+--source include/wait_until_disconnected.inc
+#
+# MDEV-18450 liveness condition:
+# Despite shutdown even "late" slave #4 is in sync
+#
+--connection server_4
+--sync_with_master
+
+--connection server_3
+--sync_with_master
+
+--connection server_2
+--sync_with_master
+
+--connection server_1
+--append_file $MYSQLTEST_VARDIR/tmp/mysqld.1.expect
+restart
+EOF
+
+--connection default
+--enable_reconnect
+--source include/wait_until_connected_again.inc
+--connection server_1
+--enable_reconnect
+--source include/wait_until_connected_again.inc
+
+#
+# Cleanup
+#
+--connection server_1
+DROP TABLE t1;
+
+--connection server_2
+--source include/start_slave.inc
+--connection server_3
+--source include/start_slave.inc
+--connection server_4
+--source include/start_slave.inc
--- a/mysql-test/suite/rpl/r/rpl_shutdown_wait_semisync_slaves.result
+++ b/mysql-test/suite/rpl/r/rpl_shutdown_wait_semisync_slaves.result
@ -0,0 +1,58 @@
+include/rpl_init.inc [topology=1->2, 1->3, 1->4]
+connection server_1;
+call mtr.add_suppression("Timeout waiting for reply of binlog");
+SET @@GLOBAL.rpl_semi_sync_master_enabled = 1;
+connection server_2;
+set global rpl_semi_sync_slave_enabled = 1;
+include/stop_slave.inc
+include/start_slave.inc
+set global rpl_semi_sync_slave_enabled = 1;
+connection server_3;
+set global rpl_semi_sync_slave_enabled = 1;
+include/stop_slave.inc
+include/start_slave.inc
+set global rpl_semi_sync_slave_enabled = 1;
+connection server_1;
+CREATE TABLE t1 (a INT) ENGINE=innodb;
+connection server_2;
+connection server_3;
+connection server_4;
+include/stop_slave.inc
+connection server_1;
+connection server_1;
+SET @@GLOBAL.debug_dbug="+d,simulate_delay_at_shutdown";
+connection server_4;
+include/start_slave.inc
+connection server_1;
+SHUTDOWN WAIT FOR ALL SLAVES;
+connection server_4;
+connection server_3;
+connection server_2;
+connection server_1;
+connection default;
+connection server_1;
+connection server_1;
+DROP TABLE t1;
+connection server_2;
+include/start_slave.inc
+Warnings:
+Note	1254	Slave is already running
+connection server_3;
+include/start_slave.inc
+Warnings:
+Note	1254	Slave is already running
+connection server_4;
+include/start_slave.inc
+Warnings:
+Note	1254	Slave is already running
+connection server_2;
+include/stop_slave.inc
+include/start_slave.inc
+SET @@GLOBAL.rpl_semi_sync_slave_enabled = 0;;
+connection server_3;
+include/stop_slave.inc
+include/start_slave.inc
+SET @@GLOBAL.rpl_semi_sync_slave_enabled = 0;;
+connection server_1;
+SET @@GLOBAL.rpl_semi_sync_master_enabled = 0;
+include/rpl_end.inc
--- a/mysql-test/suite/rpl/r/rpl_shutdown_wait_slaves.result
+++ b/mysql-test/suite/rpl/r/rpl_shutdown_wait_slaves.result
@ -0,0 +1,35 @@
+include/rpl_init.inc [topology=1->2, 1->3, 1->4]
+connection server_1;
+CREATE TABLE t1 (a INT) ENGINE=innodb;
+connection server_2;
+connection server_3;
+connection server_4;
+include/stop_slave.inc
+connection server_1;
+connection server_1;
+SET @@GLOBAL.debug_dbug="+d,simulate_delay_at_shutdown";
+connection server_4;
+include/start_slave.inc
+connection server_1;
+SHUTDOWN WAIT FOR ALL SLAVES;
+connection server_4;
+connection server_3;
+connection server_2;
+connection server_1;
+connection default;
+connection server_1;
+connection server_1;
+DROP TABLE t1;
+connection server_2;
+include/start_slave.inc
+Warnings:
+Note	1254	Slave is already running
+connection server_3;
+include/start_slave.inc
+Warnings:
+Note	1254	Slave is already running
+connection server_4;
+include/start_slave.inc
+Warnings:
+Note	1254	Slave is already running
+include/rpl_end.inc
--- a/mysql-test/suite/rpl/t/rpl_shutdown_wait_semisync_slaves.cnf
+++ b/mysql-test/suite/rpl/t/rpl_shutdown_wait_semisync_slaves.cnf
@ -0,0 +1,16 @@
+!include ../my.cnf
+
+[mysqld.1]
+log_warnings=3
+[mysqld.2]
+
+[mysqld.3]
+
+[mysqld.4]
+
+[ENV]
+SERVER_MYPORT_3=		@mysqld.3.port
+SERVER_MYSOCK_3=		@mysqld.3.socket
+
+SERVER_MYPORT_4=		@mysqld.4.port
+SERVER_MYSOCK_4=		@mysqld.4.socket
--- a/mysql-test/suite/rpl/t/rpl_shutdown_wait_semisync_slaves.test
+++ b/mysql-test/suite/rpl/t/rpl_shutdown_wait_semisync_slaves.test
@ -0,0 +1,46 @@
+#
+# MDEV-18450 "Slow" shutdown to wait for slaves that are to be fed
+# with everything in the master binlog before shutdown completes.
+#
+# This is a semisync version of basic tests.
+--source include/have_innodb.inc
+--source include/have_debug.inc
+--let $rpl_topology=1->2, 1->3, 1->4
+--source include/rpl_init.inc
+
+--connection server_1
+call mtr.add_suppression("Timeout waiting for reply of binlog");
+--let $sav_enabled_master=`SELECT @@GLOBAL.rpl_semi_sync_master_enabled`
+SET @@GLOBAL.rpl_semi_sync_master_enabled = 1;
+
+--let slaves= 3
+--let i= 2
+while (`SELECT $i <= $slaves`)
+{
+  --connection server_$i
+  --let $sav_enabled_slave=`SELECT @@GLOBAL.rpl_semi_sync_slave_enabled`
+  set global rpl_semi_sync_slave_enabled = 1;
+
+  source include/stop_slave.inc;
+  source include/start_slave.inc;
+  set global rpl_semi_sync_slave_enabled = 1;
+
+  --inc $i
+}
+
+--source include/rpl_shutdown_wait_slaves.inc
+--let i= 2
+while (`SELECT $i <= $slaves`)
+{
+  --connection server_$i
+  source include/stop_slave.inc;
+  source include/start_slave.inc;
+  --eval SET @@GLOBAL.rpl_semi_sync_slave_enabled = $sav_enabled_slave;
+
+  --inc $i
+}
+
+--connection server_1
+--eval SET @@GLOBAL.rpl_semi_sync_master_enabled = $sav_enabled_master
+
+--source include/rpl_end.inc
--- a/mysql-test/suite/rpl/t/rpl_shutdown_wait_slaves.cnf
+++ b/mysql-test/suite/rpl/t/rpl_shutdown_wait_slaves.cnf
@ -0,0 +1,16 @@
+!include ../my.cnf
+
+[mysqld.1]
+log_warnings=3
+[mysqld.2]
+
+[mysqld.3]
+
+[mysqld.4]
+
+[ENV]
+SERVER_MYPORT_3=		@mysqld.3.port
+SERVER_MYSOCK_3=		@mysqld.3.socket
+
+SERVER_MYPORT_4=		@mysqld.4.port
+SERVER_MYSOCK_4=		@mysqld.4.socket
--- a/mysql-test/suite/rpl/t/rpl_shutdown_wait_slaves.test
+++ b/mysql-test/suite/rpl/t/rpl_shutdown_wait_slaves.test
@ -0,0 +1,11 @@
+#
+# MDEV-18450 "Slow" shutdown to wait for slaves that are to be fed
+# with everything in the master binlog before shutdown completes.
+#
+--source include/have_innodb.inc
+--source include/have_debug.inc
+--let $rpl_topology=1->2, 1->3, 1->4
+--source include/rpl_init.inc
+
+--source include/rpl_shutdown_wait_slaves.inc
+--source include/rpl_end.inc
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@ -466,6 +466,7 @@ uint protocol_version;
 uint lower_case_table_names;
 ulong tc_heuristic_recover= 0;
 Atomic_counter<uint32_t> thread_count;
+bool shutdown_wait_for_slaves;
 int32 slave_open_temp_tables;
 ulong thread_created;
 ulong back_log, connect_timeout, concurrency, server_id;
@ -1519,19 +1520,9 @@ static void end_ssl();
 ** Code to end mysqld
 ****************************************************************************/

-static my_bool kill_all_threads(THD *thd, void *)
+/* common callee of two shutdown phases */
+static void kill_thread(THD *thd)
 {
-  DBUG_PRINT("quit", ("Informing thread %ld that it's time to die",
-                      (ulong) thd->thread_id));
-  /* We skip slave threads on this first loop through. */
-  if (thd->slave_thread)
-    return 0;
-
-  if (DBUG_EVALUATE_IF("only_kill_system_threads", !thd->system_thread, 0))
-    return 0;
-
-  thd->set_killed(KILL_SERVER_HARD);
-  MYSQL_CALLBACK(thread_scheduler, post_kill_notification, (thd));
  if (WSREP(thd)) mysql_mutex_lock(&thd->LOCK_thd_data);
  mysql_mutex_lock(&thd->LOCK_thd_kill);
  if (thd->mysys_var)
@ -1557,12 +1548,52 @@ static my_bool kill_all_threads(THD *thd, void *)
  }
  mysql_mutex_unlock(&thd->LOCK_thd_kill);
  if (WSREP(thd)) mysql_mutex_unlock(&thd->LOCK_thd_data);
+}
+
+
+/**
+  First shutdown everything but slave threads and binlog dump connections
+*/
+static my_bool kill_thread_phase_1(THD *thd, void *)
+{
+  DBUG_PRINT("quit", ("Informing thread %ld that it's time to die",
+                      (ulong) thd->thread_id));
+  if (thd->slave_thread || thd->is_binlog_dump_thread())
+    return 0;
+
+  if (DBUG_EVALUATE_IF("only_kill_system_threads", !thd->system_thread, 0))
+    return 0;
+
+  thd->set_killed(KILL_SERVER_HARD);
+  MYSQL_CALLBACK(thread_scheduler, post_kill_notification, (thd));
+  kill_thread(thd);
  return 0;
 }


-static my_bool warn_threads_still_active(THD *thd, void *)
+/**
+  Last shutdown binlog dump connections
+*/
+static my_bool kill_thread_phase_2(THD *thd, void *)
 {
+  if (shutdown_wait_for_slaves)
+  {
+    thd->set_killed(KILL_SERVER);
+  }
+  else
+  {
+    thd->set_killed(KILL_SERVER_HARD);
+    MYSQL_CALLBACK(thread_scheduler, post_kill_notification, (thd));
+  }
+  kill_thread(thd);
+  return 0;
+}
+
+
+/* associated with the kill thread phase 1 */
+static my_bool warn_threads_active_after_phase_1(THD *thd, void *)
+{
+  if (!thd->is_binlog_dump_thread())
    sql_print_warning("%s: Thread %llu (user : '%s') did not exit\n", my_progname,
                      (ulonglong) thd->thread_id,
                      (thd->main_security_ctx.user ?
@ -1571,6 +1602,23 @@ static my_bool warn_threads_still_active(THD *thd, void *)
 }


+/* associated with the kill thread phase 2 */
+static my_bool warn_threads_active_after_phase_2(THD *thd, void *)
+{
+  mysql_mutex_lock(&thd->LOCK_thd_data);
+  // dump thread may not have yet (or already) current_linfo set
+  sql_print_warning("Dump thread %llu last sent to server %lu "
+                    "binlog file:pos %s:%llu",
+                    thd->thread_id, thd->variables.server_id,
+                    thd->current_linfo ?
+                    my_basename(thd->current_linfo->log_file_name) : "NULL",
+                    thd->current_linfo ? thd->current_linfo->pos : 0);
+  mysql_mutex_unlock(&thd->LOCK_thd_data);
+
+  return 0;
+}
+
+
 /**
  Kills main thread.

@ -1650,6 +1698,21 @@ void kill_mysql(THD *thd)
  {
    my_free(user);
  }
+
+  DBUG_EXECUTE_IF("mysql_admin_shutdown_wait_for_slaves",
+                  thd->lex->is_shutdown_wait_for_slaves= true;);
+  DBUG_EXECUTE_IF("simulate_delay_at_shutdown",
+                  {
+                    DBUG_ASSERT(binlog_dump_thread_count == 3);
+                    const char act[]=
+                      "now "
+                      "SIGNAL greetings_from_kill_mysql";
+                    DBUG_ASSERT(!debug_sync_set_action(thd,
+                                                       STRING_WITH_LEN(act)));
+                  };);
+
+  if (thd->lex->is_shutdown_wait_for_slaves)
+    shutdown_wait_for_slaves= true;
  break_connect_loop();
 }

@ -1693,7 +1756,7 @@ static void close_connections(void)
    This will give the threads some time to gracefully abort their
    statements and inform their clients that the server is about to die.
  */
-  server_threads.iterate(kill_all_threads);
+  server_threads.iterate(kill_thread_phase_1);

  Events::deinit();
  slave_prepare_for_shutdown();
@ -1716,11 +1779,11 @@ static void close_connections(void)
  */
  DBUG_PRINT("info", ("thread_count: %u", uint32_t(thread_count)));

-  for (int i= 0; thread_count && i < 1000; i++)
+  for (int i= 0; (thread_count - binlog_dump_thread_count) && i < 1000; i++)
    my_sleep(20000);

  if (global_system_variables.log_warnings)
-    server_threads.iterate(warn_threads_still_active);
+    server_threads.iterate(warn_threads_active_after_phase_1);

 #ifdef WITH_WSREP
  if (wsrep_inited == 1)
@ -1732,9 +1795,23 @@ static void close_connections(void)
  DBUG_PRINT("quit", ("Waiting for threads to die (count=%u)",
                      uint32_t(thread_count)));

-  while (thread_count)
+  while (thread_count - binlog_dump_thread_count)
    my_sleep(1000);

+  /* Kill phase 2 */
+  server_threads.iterate(kill_thread_phase_2);
+  for (uint64 i= 0; thread_count; i++)
+  {
+    /*
+      This time the warnings are emitted within the loop to provide a
+      dynamic view on the shutdown status through the errorlog.
+    */
+    if (global_system_variables.log_warnings > 2 && i % 60000 == 0)
+      server_threads.iterate(warn_threads_active_after_phase_2);
+    my_sleep(1000);
+  }
+  /* End of kill phase 2 */
+
  DBUG_PRINT("quit",("close_connections thread"));
  DBUG_VOID_RETURN;
 }
--- a/sql/repl_failsafe.cc
+++ b/sql/repl_failsafe.cc
@ -147,11 +147,11 @@ int THD::register_slave(uchar *packet, size_t packet_length)
  if (!(si->master_id= uint4korr(p)))
    si->master_id= global_system_variables.server_id;

-  binlog_dump_thread_count++;
  unregister_slave();
  mysql_mutex_lock(&LOCK_thd_data);
  slave_info= si;
  mysql_mutex_unlock(&LOCK_thd_data);
+  binlog_dump_thread_count++;
  return 0;

 err:
@ -161,6 +161,16 @@ err:
 }


+bool THD::is_binlog_dump_thread()
+{
+  mysql_mutex_lock(&LOCK_thd_data);
+  bool res= slave_info != NULL;
+  mysql_mutex_unlock(&LOCK_thd_data);
+
+  return res;
+}
+
+
 static my_bool show_slave_hosts_callback(THD *thd, Protocol *protocol)
 {
  my_bool res= FALSE;
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@ -644,6 +644,7 @@ THD::THD(my_thread_id id, bool is_wsrep_applier, bool skip_global_sys_var_lock)
   m_tmp_tables_locked(false)
 #ifdef HAVE_REPLICATION
   ,
+   current_linfo(0),
   slave_info(0)
 #endif
 #ifdef WITH_WSREP
@ -751,7 +752,6 @@ THD::THD(my_thread_id id, bool is_wsrep_applier, bool skip_global_sys_var_lock)
  progress.arena= 0;
  progress.report_to_client= 0;
  progress.max_counter= 0;
-  current_linfo =  0;
  slave_thread = 0;
  connection_name.str= 0;
  connection_name.length= 0;
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@ -3194,12 +3194,6 @@ public:
  /** number of name_const() substitutions, see sp_head.cc:subst_spvars() */
  uint       query_name_consts;

-  /*
-    If we do a purge of binary logs, log index info of the threads
-    that are currently reading it needs to be adjusted. To do that
-    each thread that is using LOG_INFO needs to adjust the pointer to it
-  */
-  LOG_INFO*  current_linfo;
  NET*       slave_net;			// network connection from slave -> m.

  /*
@ -4813,10 +4807,20 @@ private:

 public:
 #ifdef HAVE_REPLICATION
+  /*
+    If we do a purge of binary logs, log index info of the threads
+    that are currently reading it needs to be adjusted. To do that
+    each thread that is using LOG_INFO needs to adjust the pointer to it
+  */
+  LOG_INFO *current_linfo;
  Slave_info *slave_info;

+  void set_current_linfo(LOG_INFO *linfo);
+  void reset_current_linfo() { set_current_linfo(0); }
+
  int register_slave(uchar *packet, size_t packet_length);
  void unregister_slave();
+  bool is_binlog_dump_thread();
 #endif

  inline ulong wsrep_binlog_format() const
@ -4975,18 +4979,6 @@ public:
       (THD_TRANS::DID_WAIT | THD_TRANS::CREATED_TEMP_TABLE |
        THD_TRANS::DROPPED_TEMP_TABLE | THD_TRANS::DID_DDL));
  }
-  /*
-    Reset current_linfo
-    Setting current_linfo to 0 needs to be done with LOCK_thd_data to
-    ensure that adjust_linfo_offsets doesn't use a structure that may
-    be deleted.
-  */
-  inline void reset_current_linfo()
-  {
-    mysql_mutex_lock(&LOCK_thd_data);
-    current_linfo= 0;
-    mysql_mutex_unlock(&LOCK_thd_data);
-  }


  uint get_net_wait_timeout()
--- a/sql/sql_lex.h
+++ b/sql/sql_lex.h
@ -3226,6 +3226,7 @@ public:
  /* The following is used by KILL */
  killed_state kill_signal;
  killed_type  kill_type;
+  bool is_shutdown_wait_for_slaves;
  /*
    This variable is used in post-parse stage to declare that sum-functions,
    or functions which have sense only if GROUP BY is present, are allowed.
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@ -2168,6 +2168,7 @@ bool dispatch_command(enum enum_server_command command, THD *thd,
    */
    enum mysql_enum_shutdown_level level;
    level= (enum mysql_enum_shutdown_level) (uchar) packet[0];
+    thd->lex->is_shutdown_wait_for_slaves= false;  // "deferred" cleanup
    if (level == SHUTDOWN_DEFAULT)
      level= SHUTDOWN_WAIT_ALL_BUFFERS; // soon default will be configurable
    else if (level != SHUTDOWN_WAIT_ALL_BUFFERS)
--- a/sql/sql_repl.cc
+++ b/sql/sql_repl.cc
@ -506,6 +506,22 @@ static enum enum_binlog_checksum_alg get_binlog_checksum_value_at_connect(THD *
  DBUG_RETURN(ret);
 }

+
+/**
+  Set current_linfo
+
+  Setting current_linfo needs to be done with LOCK_thd_data to ensure that
+  adjust_linfo_offsets doesn't use a structure that may be deleted.
+*/
+
+void THD::set_current_linfo(LOG_INFO *linfo)
+{
+  mysql_mutex_lock(&LOCK_thd_data);
+  current_linfo= linfo;
+  mysql_mutex_unlock(&LOCK_thd_data);
+}
+
+
 /*
  Adjust the position pointer in the binary log file for all running slaves

@ -2125,9 +2141,8 @@ static int init_binlog_sender(binlog_send_info *info,

  // set current pos too
  linfo->pos= *pos;
-
  // note: publish that we use file, before we open it
-  thd->current_linfo= linfo;
+  thd->set_current_linfo(linfo);

  if (check_start_offset(info, linfo->log_file_name, *pos))
    return 1;
@ -2365,12 +2380,13 @@ static int send_format_descriptor_event(binlog_send_info *info, IO_CACHE *log,
  DBUG_RETURN(0);
 }

-static bool should_stop(binlog_send_info *info)
+static bool should_stop(binlog_send_info *info, bool kill_server_check= false)
 {
  return
    info->net->error ||
    info->net->vio == NULL ||
-      info->thd->killed ||
+    (info->thd->killed &&
+     (info->thd->killed != KILL_SERVER || kill_server_check)) ||
    info->error != 0 ||
    info->should_stop;
 }
@ -2393,7 +2409,7 @@ static int wait_new_events(binlog_send_info *info,         /* in */
                        &stage_master_has_sent_all_binlog_to_slave,
                        &old_stage);

-  while (!should_stop(info))
+  while (!should_stop(info, true))
  {
    *end_pos_ptr= mysql_bin_log.get_binlog_end_pos(binlog_end_pos_filename);
    if (strcmp(linfo->log_file_name, binlog_end_pos_filename) != 0)
@ -2745,6 +2761,14 @@ void mysql_binlog_send(THD* thd, char* log_ident, my_off_t pos,
    info->error= ER_UNKNOWN_ERROR;
    goto err;
  }
+  DBUG_EXECUTE_IF("simulate_delay_at_shutdown",
+                 {
+                   const char act[]=
+                     "now "
+                     "WAIT_FOR greetings_from_kill_mysql";
+                   DBUG_ASSERT(!debug_sync_set_action(thd,
+                                                      STRING_WITH_LEN(act)));
+                 };);

  /*
    heartbeat_period from @master_heartbeat_period user variable
@ -3952,7 +3976,7 @@ bool mysql_show_binlog_events(THD* thd)
      goto err;
    }

-    thd->current_linfo= &linfo;
+    thd->set_current_linfo(&linfo);

    if ((file=open_binlog(&log, linfo.log_file_name, &errmsg)) < 0)
      goto err;
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@ -14717,6 +14717,15 @@ kill_expr:

 shutdown:
        SHUTDOWN { Lex->sql_command= SQLCOM_SHUTDOWN; }
+        shutdown_option {}
+        ;
+
+shutdown_option:
+        /*  Empty */    { Lex->is_shutdown_wait_for_slaves= false; }
+        | WAIT_SYM FOR_SYM ALL SLAVES
+        {
+          Lex->is_shutdown_wait_for_slaves= true;
+        }
        ;

 /* change database */
--- a/sql/sql_yacc_ora.yy
+++ b/sql/sql_yacc_ora.yy
@ -14862,8 +14862,16 @@ kill_expr:

 shutdown:
        SHUTDOWN { Lex->sql_command= SQLCOM_SHUTDOWN; }
+        shutdown_option {}
        ;

+shutdown_option:
+        /*  Empty */    { Lex->is_shutdown_wait_for_slaves= false; }
+        | WAIT_SYM FOR_SYM ALL SLAVES
+        {
+          Lex->is_shutdown_wait_for_slaves= true;
+        }
+        ;
 /* change database */

 use: