Stop slave immediately in MTS if partial trx in the relay log can be …

…rollbacked

Summary:
In MTS stop slave can take a minute to complete if the last transaction
is partially downloaded from the master. The slave waits for a minute for the
master to send the rest of the trx and then finally gives up. Strictly, this
wait is only required if the partial trx cannot be rollbacked safely (e.g. trx
on non-transactional engine, DDLs etc.). This change checks if there are no jobs
queued in the worker threads and if the partial trx can be rollbacked, if yes,
it immediately stops the slave.

Reviewed By: tianx

Differential Revision: D5130797

fbshipit-source-id: 4f1f0b6

mysql-test/suite/rpl/r/rpl_stop_slave_partial_trx.result

@@ -0,0 +1,111 @@
+include/master-slave.inc
+Warnings:
+Note	####	Sending passwords in plain text without SSL/TLS is extremely insecure.
+Note	####	Storing MySQL user name or password information in the master info repository is not secure and is therefore not recommended. Please consider using the USER and PASSWORD connection options for START SLAVE; see the 'START SLAVE Syntax' in the MySQL Manual for more information.
+[connection master]
+call mtr.add_suppression("The slave coordinator and worker threads are stopped, possibly leaving data in inconsistent state");
+create database d1;
+create database d2;
+create database d3;
+create table d1.t1 (a int) engine=innodb;
+create table d2.t2 (a int) engine=myisam;
+create table d3.t3 (a int) engine=innodb;
+lock tables d1.t1 read;
+insert into d1.t1 values(1);
+insert into d1.t1 values(1);
+insert into d1.t1 values(1);
+insert into d1.t1 values(1);
+insert into d1.t1 values(1);
+set global debug= '+d,dump_thread_wait_after_send_write_rows';
+insert into d2.t2 values(1);
+unlock tables;
+set @start=now();
+stop slave;
+select timestampdiff(SECOND, @start, now()) >= 60;
+timestampdiff(SECOND, @start, now()) >= 60
+1
+start slave;
+SET DEBUG_SYNC= 'now SIGNAL signal.continue';
+SET DEBUG_SYNC= 'RESET';
+set @@global.debug= '-d,dump_thread_wait_after_send_write_rows';
+"Tables on master:"
+connection master
+select * from d1.t1;
+a
+1
+1
+1
+1
+1
+select * from d2.t2;
+a
+1
+select * from d3.t3;
+a
+"Tables on slave:"
+connection slave
+select * from d1.t1;
+a
+1
+1
+1
+1
+1
+select * from d2.t2;
+a
+1
+select * from d3.t3;
+a
+delete from d1.t1;
+delete from d2.t2;
+delete from d3.t3;
+lock tables d1.t1 read;
+insert into d1.t1 values(1);
+insert into d1.t1 values(1);
+insert into d1.t1 values(1);
+insert into d1.t1 values(1);
+insert into d1.t1 values(1);
+set global debug= '+d,dump_thread_wait_after_send_write_rows';
+insert into d3.t3 values(1);
+unlock tables;
+set @start=now();
+stop slave;
+select timestampdiff(SECOND, @start, now()) < 60;
+timestampdiff(SECOND, @start, now()) < 60
+1
+start slave;
+SET DEBUG_SYNC= 'now SIGNAL signal.continue';
+SET DEBUG_SYNC= 'RESET';
+set @@global.debug= '-d,dump_thread_wait_after_send_write_rows';
+"Tables on master:"
+connection master
+select * from d1.t1;
+a
+1
+1
+1
+1
+1
+select * from d2.t2;
+a
+select * from d3.t3;
+a
+1
+"Tables on slave:"
+connection slave
+select * from d1.t1;
+a
+1
+1
+1
+1
+1
+select * from d2.t2;
+a
+select * from d3.t3;
+a
+1
+drop database d1;
+drop database d2;
+drop database d3;
+include/rpl_end.inc

mysql-test/suite/rpl/t/rpl_stop_slave_partial_trx-master.opt

@@ -0,0 +1,2 @@
+--slave_parallel_workers=8 --enforce-gtid-consistency --gtid-mode=ON --log-bin
+--log-slave-updates

mysql-test/suite/rpl/t/rpl_stop_slave_partial_trx-slave.opt

@@ -0,0 +1,2 @@
+--slave_parallel_workers=8 --enforce-gtid-consistency --gtid-mode=ON --log-bin
+--log-slave-updates

mysql-test/suite/rpl/t/rpl_stop_slave_partial_trx.test

@@ -0,0 +1,156 @@
+source include/master-slave.inc;
+source include/have_debug.inc;
+source include/have_debug_sync.inc;
+source include/have_binlog_format_row.inc;
+
+call mtr.add_suppression("The slave coordinator and worker threads are stopped, possibly leaving data in inconsistent state");
+
+# Create schema
+connection master;
+create database d1;
+create database d2;
+create database d3;
+create table d1.t1 (a int) engine=innodb;
+create table d2.t2 (a int) engine=myisam; # non-transactional engine
+create table d3.t3 (a int) engine=innodb; # transactional engine
+sync_slave_with_master;
+
+
+
+## Test 1: STOP SLAVE when there are pending jobs in the worker queues and there
+## is a partial transaction on a non-transactional table. The slave should wait
+## for 1 minute to complete the partial transaction before giving up.
+
+# Block all d1.t1 transactions on the slave
+connection slave;
+lock tables d1.t1 read;
+
+# Generate some load, all of these will be blocked in the slave worker queue
+connection master;
+let $num_inserts=5;
+while ($num_inserts)
+{
+  insert into d1.t1 values(1);
+  dec $num_inserts;
+}
+
+# This will stop the dump thread before sending the entire group
+set global debug= '+d,dump_thread_wait_after_send_write_rows';
+insert into d2.t2 values(1);
+
+# wait for the dump thread reach the sync point
+--let $wait_condition= select count(*)=1 from information_schema.processlist where state LIKE '%debug sync point%' and command like 'Binlog Dump%'
+--source include/wait_condition.inc
+
+connection slave;
+# unblock d1.t1
+unlock tables;
+
+# This should take at least a minute because the trx on t2 is not completely downloaded and cannot be rollbacked safely
+set @start=now();
+stop slave;
+select timestampdiff(SECOND, @start, now()) >= 60;
+start slave;
+
+connection master;
+SET DEBUG_SYNC= 'now SIGNAL signal.continue';
+# wait for the dump thread to come out of the waiting phase before resetting the signals
+--let $wait_condition= select count(*)=0 from information_schema.processlist where state LIKE '%debug sync point%' and command='Binlog Dump'
+--source include/wait_condition.inc
+SET DEBUG_SYNC= 'RESET';
+
+connection master;
+set @@global.debug= '-d,dump_thread_wait_after_send_write_rows';
+sync_slave_with_master;
+
+# Verification
+connection master;
+echo "Tables on master:"
+connection master;
+select * from d1.t1;
+select * from d2.t2;
+select * from d3.t3;
+echo "Tables on slave:"
+connection slave;
+select * from d1.t1;
+select * from d2.t2;
+select * from d3.t3;
+
+# cleanup
+connection master;
+delete from d1.t1;
+delete from d2.t2;
+delete from d3.t3;
+sync_slave_with_master;
+
+
+
+## Test 2: STOP SLAVE when there are pending jobs in the worker queues and there
+## is a partial transaction on a transactional table. The slave should stop
+## immidiately after completing all pending full transactions.
+
+# Block all d1.t1 transactions on the slave
+connection slave;
+lock tables d1.t1 read;
+
+# Generate some load, all of these will be blocked in the slave worker queue
+connection master;
+let $num_inserts=5;
+while ($num_inserts)
+{
+  insert into d1.t1 values(1);
+  dec $num_inserts;
+}
+
+# This will stop the dump thread before sending the entire group
+set global debug= '+d,dump_thread_wait_after_send_write_rows';
+insert into d3.t3 values(1);
+
+# wait for the dump thread reach the sync point
+--let $wait_condition= select count(*)=1 from information_schema.processlist where state LIKE '%debug sync point%' and command like 'Binlog Dump%'
+--source include/wait_condition.inc
+
+connection slave;
+# unblock d1.t1
+unlock tables;
+
+# Since the partial transaction is on a transactional table the slave should
+# stop as soon as is completes all pending full transactions
+set @start=now();
+stop slave;
+select timestampdiff(SECOND, @start, now()) < 60;
+start slave;
+
+connection master;
+SET DEBUG_SYNC= 'now SIGNAL signal.continue';
+# wait for the dump thread to come out of the waiting phase before resetting the signals
+--let $wait_condition= select count(*)=0 from information_schema.processlist where state LIKE '%debug sync point%' and command='Binlog Dump'
+--source include/wait_condition.inc
+SET DEBUG_SYNC= 'RESET';
+
+connection master;
+set @@global.debug= '-d,dump_thread_wait_after_send_write_rows';
+sync_slave_with_master;
+
+# Verification
+connection master;
+echo "Tables on master:"
+connection master;
+select * from d1.t1;
+select * from d2.t2;
+select * from d3.t3;
+echo "Tables on slave:"
+connection slave;
+select * from d1.t1;
+select * from d2.t2;
+select * from d3.t3;
+
+# final cleanup
+connection master;
+drop database d1;
+drop database d2;
+drop database d3;
+sync_slave_with_master;
+
+
+source include/rpl_end.inc;

sql/rpl_master.cc

@@ -1742,6 +1742,20 @@ void mysql_binlog_send(THD* thd, char* log_ident, my_off_t pos,
         set_timespec_nsec(last_event_sent_ts, 0);
       }
 
+      DBUG_EXECUTE_IF("dump_thread_wait_after_send_write_rows",
+                      {
+                        if (event_type == WRITE_ROWS_EVENT)
+                        {
+                          net_flush(net);
+                          const char act[]=
+                            "now "
+                            "wait_for signal.continue";
+                          DBUG_ASSERT(opt_debug_sync_timeout > 0);
+                          DBUG_ASSERT(!debug_sync_set_action(current_thd,
+                                STRING_WITH_LEN(act)));
+                        }
+                      });
+
       DBUG_EXECUTE_IF("dump_thread_wait_before_send_xid",
                       {
                         if (event_type == XID_EVENT)

sql/rpl_rli.cc

@@ -417,6 +417,39 @@ bool Relay_log_info::mts_finalize_recovery()
   DBUG_RETURN(ret);
 }
 
+bool Relay_log_info::mts_workers_queue_empty()
+{
+  Slave_worker *worker= NULL;
+  ulong ret= 0;
+  for (ulong i= 0; i< workers.elements; ++i)
+  {
+    worker= *dynamic_element(&workers, i, Slave_worker**);
+    mysql_mutex_lock(&worker->jobs_lock);
+    ret+= worker->curr_jobs;
+    mysql_mutex_unlock(&worker->jobs_lock);
+  }
+  return ret == 0;
+}
+
+/* Checks if all in-flight stmts/trx can be safely rollbacked */
+bool Relay_log_info::cannot_safely_rollback()
+{
+  if (!is_parallel_exec())
+    return info_thd->transaction.all.cannot_safely_rollback();
+
+  bool ret= false;
+  Slave_worker *worker= NULL;
+
+  for (ulong i= 0; i< workers.elements; ++i)
+  {
+    worker= *dynamic_element(&workers, i, Slave_worker**);
+    mysql_mutex_lock(&worker->jobs_lock);
+    ret= ret || worker->info_thd->transaction.all.cannot_safely_rollback();
+    mysql_mutex_unlock(&worker->jobs_lock);
+  }
+  return ret;
+}
+
 static inline int add_relay_log(Relay_log_info* rli,LOG_INFO* linfo)
 {
   MY_STAT s;

sql/rpl_rli.h

@@ -707,6 +707,9 @@ class Relay_log_info : public Rpl_info
       mts_group_status == MTS_IN_GROUP;
   }
 
+  bool mts_workers_queue_empty();
+  bool cannot_safely_rollback();
+
   /**
      While a group is executed by a Worker the relay log can change.
      Coordinator notifies Workers about this event. Worker is supposed

sql/rpl_slave.cc

@@ -1838,8 +1838,13 @@ static bool sql_slave_killed(THD* thd, Relay_log_info* rli)
   if (abort_loop || thd->killed || rli->abort_slave)
   {
     rli->sql_thread_kill_accepted= true;
+    /* NOTE: In MTS mode if all workers are done and if the partial trx
+       (if any) can be rollbacked safely we can accept the kill */
+    bool can_rollback= !rli->is_mts_in_group() ||
+                       (rli->mts_workers_queue_empty() &&
+                        !rli->cannot_safely_rollback());
     is_parallel_warn= (rli->is_parallel_exec() &&
-                       (rli->is_mts_in_group() || thd->killed));
+                       (!can_rollback || thd->killed));
     /*
       Slave can execute stop being in one of two MTS or Single-Threaded mode.
       The modes define different criteria to accept the stop.