MDEV-4506: Parallel replication: Intermediate commit.

Fix a number of failures in the test suite.

12 files changed, 140 insertions, 25 deletions

diff --git a/mysql-test/r/mysqld--help.result b/mysql-test/r/mysqld--help.result
index d9cdf3c3240..f6f03b42270 100644
--- a/mysql-test/r/mysqld--help.result
+++ b/mysql-test/r/mysqld--help.result
@@ -41,6 +41,17 @@ The following options may be given as the first argument:
  Type of BINLOG_CHECKSUM_ALG. Include checksum for log
  events in the binary log. Possible values are NONE and
  CRC32; default is NONE.
+ --binlog-commit-wait-count=# 
+ If non-zero, binlog write will wait at most
+ binlog_commit_wait_usec microseconds for at least this
+ many commits to queue up for group commit to the binlog.
+ This can reduce I/O on the binlog and provide increased
+ opportunity for parallel apply on the slave, but too high
+ a value will decrease commit throughput.
+ --binlog-commit-wait-usec=# 
+ Maximum time, in microseconds, to wait for more commits
+ to queue up for binlog group commit. Only takes effect if
+ the value of binlog_commit_wait_count is non-zero.
  --binlog-direct-non-transactional-updates 
  Causes updates to non-transactional engines using
  statement format to be written directly to binary log.
@@ -783,6 +794,11 @@ The following options may be given as the first argument:
  --slave-net-timeout=# 
  Number of seconds to wait for more data from any
  master/slave connection before aborting the read
+ --slave-parallel-threads=# 
+ If non-zero, number of threads to spawn to apply in
+ parallel events on the slave that were group-committed on
+ the master or were logged with GTID in different
+ replication domains.
  --slave-skip-errors=name 
  Tells the slave thread to continue replication when a
  query event returns an error from the provided list
@@ -922,6 +938,8 @@ bind-address (No default value)
 binlog-annotate-row-events FALSE
 binlog-cache-size 32768
 binlog-checksum NONE
+binlog-commit-wait-count 0
+binlog-commit-wait-usec 100000
 binlog-direct-non-transactional-updates FALSE
 binlog-format STATEMENT
 binlog-optimize-thread-scheduling TRUE
@@ -1130,6 +1148,7 @@ slave-compressed-protocol FALSE
 slave-exec-mode STRICT
 slave-max-allowed-packet 1073741824
 slave-net-timeout 3600
+slave-parallel-threads 0
 slave-skip-errors (No default value)
 slave-sql-verify-checksum TRUE
 slave-transaction-retries 10
diff --git a/mysql-test/suite/perfschema/r/all_instances.result b/mysql-test/suite/perfschema/r/all_instances.result
index 8ecb8dfe602..f338461f5cd 100644
--- a/mysql-test/suite/perfschema/r/all_instances.result
+++ b/mysql-test/suite/perfschema/r/all_instances.result
@@ -61,6 +61,7 @@ wait/synch/mutex/sql/LOCK_prepared_stmt_count
 wait/synch/mutex/sql/LOCK_prepare_ordered
 wait/synch/mutex/sql/LOCK_rpl_gtid_state
 wait/synch/mutex/sql/LOCK_rpl_status
+wait/synch/mutex/sql/LOCK_rpl_thread_pool
 wait/synch/mutex/sql/LOCK_server_started
 wait/synch/mutex/sql/LOCK_slave_list
 wait/synch/mutex/sql/LOCK_slave_state
@@ -122,8 +123,10 @@ wait/synch/cond/mysys/COND_alarm
 wait/synch/cond/mysys/my_thread_var::suspend
 wait/synch/cond/mysys/THR_COND_threads
 wait/synch/cond/sql/COND_flush_thread_cache
+wait/synch/cond/sql/COND_prepare_ordered
 wait/synch/cond/sql/COND_queue_state
 wait/synch/cond/sql/COND_rpl_status
+wait/synch/cond/sql/COND_rpl_thread_pool
 wait/synch/cond/sql/COND_server_started
 wait/synch/cond/sql/COND_thread_cache
 wait/synch/cond/sql/COND_thread_count
diff --git a/mysql-test/suite/perfschema/r/dml_setup_instruments.result b/mysql-test/suite/perfschema/r/dml_setup_instruments.result
index e1a5fdd11f6..38c02cc2bf4 100644
--- a/mysql-test/suite/perfschema/r/dml_setup_instruments.result
+++ b/mysql-test/suite/perfschema/r/dml_setup_instruments.result
@@ -38,14 +38,14 @@ order by name limit 10;
 NAME	ENABLED	TIMED
 wait/synch/cond/sql/COND_flush_thread_cache	YES	YES
 wait/synch/cond/sql/COND_manager	YES	YES
+wait/synch/cond/sql/COND_parallel_entry	YES	YES
+wait/synch/cond/sql/COND_prepare_ordered	YES	YES
 wait/synch/cond/sql/COND_queue_state	YES	YES
 wait/synch/cond/sql/COND_rpl_status	YES	YES
+wait/synch/cond/sql/COND_rpl_thread	YES	YES
+wait/synch/cond/sql/COND_rpl_thread_pool	YES	YES
 wait/synch/cond/sql/COND_server_started	YES	YES
 wait/synch/cond/sql/COND_thread_cache	YES	YES
-wait/synch/cond/sql/COND_thread_count	YES	YES
-wait/synch/cond/sql/Delayed_insert::cond	YES	YES
-wait/synch/cond/sql/Delayed_insert::cond_client	YES	YES
-wait/synch/cond/sql/Event_scheduler::COND_state	YES	YES
 select * from performance_schema.setup_instruments
 where name='Wait';
 select * from performance_schema.setup_instruments
diff --git a/mysql-test/suite/sys_vars/r/binlog_commit_wait_count_basic.result b/mysql-test/suite/sys_vars/r/binlog_commit_wait_count_basic.result
new file mode 100644
index 00000000000..6837489311a
--- /dev/null
+++ b/mysql-test/suite/sys_vars/r/binlog_commit_wait_count_basic.result
@@ -0,0 +1,13 @@
+SET @save_binlog_commit_wait_count= @@GLOBAL.binlog_commit_wait_count;
+SELECT @@GLOBAL.binlog_commit_wait_count as 'must be zero because of default';
+must be zero because of default
+0
+SELECT @@SESSION.binlog_commit_wait_count  as 'no session var';
+ERROR HY000: Variable 'binlog_commit_wait_count' is a GLOBAL variable
+SET GLOBAL binlog_commit_wait_count= 0;
+SET GLOBAL binlog_commit_wait_count= DEFAULT;
+SET GLOBAL binlog_commit_wait_count= 10;
+SELECT @@GLOBAL.binlog_commit_wait_count;
+@@GLOBAL.binlog_commit_wait_count
+10
+SET GLOBAL binlog_commit_wait_count = @save_binlog_commit_wait_count;
diff --git a/mysql-test/suite/sys_vars/r/binlog_commit_wait_usec_basic.result b/mysql-test/suite/sys_vars/r/binlog_commit_wait_usec_basic.result
new file mode 100644
index 00000000000..b85af0bc9c7
--- /dev/null
+++ b/mysql-test/suite/sys_vars/r/binlog_commit_wait_usec_basic.result
@@ -0,0 +1,13 @@
+SET @save_binlog_commit_wait_usec= @@GLOBAL.binlog_commit_wait_usec;
+SELECT @@GLOBAL.binlog_commit_wait_usec as 'check default';
+check default
+100000
+SELECT @@SESSION.binlog_commit_wait_usec  as 'no session var';
+ERROR HY000: Variable 'binlog_commit_wait_usec' is a GLOBAL variable
+SET GLOBAL binlog_commit_wait_usec= 0;
+SET GLOBAL binlog_commit_wait_usec= DEFAULT;
+SET GLOBAL binlog_commit_wait_usec= 10000;
+SELECT @@GLOBAL.binlog_commit_wait_usec;
+@@GLOBAL.binlog_commit_wait_usec
+10000
+SET GLOBAL binlog_commit_wait_usec = @save_binlog_commit_wait_usec;
diff --git a/mysql-test/suite/sys_vars/r/slave_parallel_threads_basic.result b/mysql-test/suite/sys_vars/r/slave_parallel_threads_basic.result
new file mode 100644
index 00000000000..2956d04c065
--- /dev/null
+++ b/mysql-test/suite/sys_vars/r/slave_parallel_threads_basic.result
@@ -0,0 +1,13 @@
+SET @save_slave_parallel_threads= @@GLOBAL.slave_parallel_threads;
+SELECT @@GLOBAL.slave_parallel_threads as 'must be zero because of default';
+must be zero because of default
+0
+SELECT @@SESSION.slave_parallel_threads  as 'no session var';
+ERROR HY000: Variable 'slave_parallel_threads' is a GLOBAL variable
+SET GLOBAL slave_parallel_threads= 0;
+SET GLOBAL slave_parallel_threads= DEFAULT;
+SET GLOBAL slave_parallel_threads= 10;
+SELECT @@GLOBAL.slave_parallel_threads;
+@@GLOBAL.slave_parallel_threads
+10
+SET GLOBAL slave_parallel_threads = @save_slave_parallel_threads;
diff --git a/mysql-test/suite/sys_vars/t/binlog_commit_wait_count_basic.test b/mysql-test/suite/sys_vars/t/binlog_commit_wait_count_basic.test
new file mode 100644
index 00000000000..ebce0da77fe
--- /dev/null
+++ b/mysql-test/suite/sys_vars/t/binlog_commit_wait_count_basic.test
@@ -0,0 +1,14 @@
+--source include/not_embedded.inc
+
+SET @save_binlog_commit_wait_count= @@GLOBAL.binlog_commit_wait_count;
+
+SELECT @@GLOBAL.binlog_commit_wait_count as 'must be zero because of default';
+--error ER_INCORRECT_GLOBAL_LOCAL_VAR
+SELECT @@SESSION.binlog_commit_wait_count  as 'no session var';
+
+SET GLOBAL binlog_commit_wait_count= 0;
+SET GLOBAL binlog_commit_wait_count= DEFAULT;
+SET GLOBAL binlog_commit_wait_count= 10;
+SELECT @@GLOBAL.binlog_commit_wait_count;
+
+SET GLOBAL binlog_commit_wait_count = @save_binlog_commit_wait_count;
diff --git a/mysql-test/suite/sys_vars/t/binlog_commit_wait_usec_basic.test b/mysql-test/suite/sys_vars/t/binlog_commit_wait_usec_basic.test
new file mode 100644
index 00000000000..ad9b6c99630
--- /dev/null
+++ b/mysql-test/suite/sys_vars/t/binlog_commit_wait_usec_basic.test
@@ -0,0 +1,14 @@
+--source include/not_embedded.inc
+
+SET @save_binlog_commit_wait_usec= @@GLOBAL.binlog_commit_wait_usec;
+
+SELECT @@GLOBAL.binlog_commit_wait_usec as 'check default';
+--error ER_INCORRECT_GLOBAL_LOCAL_VAR
+SELECT @@SESSION.binlog_commit_wait_usec  as 'no session var';
+
+SET GLOBAL binlog_commit_wait_usec= 0;
+SET GLOBAL binlog_commit_wait_usec= DEFAULT;
+SET GLOBAL binlog_commit_wait_usec= 10000;
+SELECT @@GLOBAL.binlog_commit_wait_usec;
+
+SET GLOBAL binlog_commit_wait_usec = @save_binlog_commit_wait_usec;
diff --git a/mysql-test/suite/sys_vars/t/slave_parallel_threads_basic.test b/mysql-test/suite/sys_vars/t/slave_parallel_threads_basic.test
new file mode 100644
index 00000000000..8e987489d86
--- /dev/null
+++ b/mysql-test/suite/sys_vars/t/slave_parallel_threads_basic.test
@@ -0,0 +1,14 @@
+--source include/not_embedded.inc
+
+SET @save_slave_parallel_threads= @@GLOBAL.slave_parallel_threads;
+
+SELECT @@GLOBAL.slave_parallel_threads as 'must be zero because of default';
+--error ER_INCORRECT_GLOBAL_LOCAL_VAR
+SELECT @@SESSION.slave_parallel_threads  as 'no session var';
+
+SET GLOBAL slave_parallel_threads= 0;
+SET GLOBAL slave_parallel_threads= DEFAULT;
+SET GLOBAL slave_parallel_threads= 10;
+SELECT @@GLOBAL.slave_parallel_threads;
+
+SET GLOBAL slave_parallel_threads = @save_slave_parallel_threads;
diff --git a/sql/rpl_parallel.cc b/sql/rpl_parallel.cc
index 63066d8d7c0..2bb5083a4f3 100644
--- a/sql/rpl_parallel.cc
+++ b/sql/rpl_parallel.cc
@@ -248,7 +248,7 @@ handle_rpl_parallel_thread(void *arg)
     if (!in_event_group)
     {
       rpt->current_entry= NULL;
-      if (!rpt->free)
+      if (!rpt->stop && !rpt->free)
       {
         mysql_mutex_lock(&rpt->pool->LOCK_rpl_thread_pool);
         list= rpt->pool->free_list;
@@ -262,9 +262,27 @@ handle_rpl_parallel_thread(void *arg)
     }
   }
 
+  rpt->thd= NULL;
+  mysql_mutex_unlock(&rpt->LOCK_rpl_thread);
+
+  thd->clear_error();
+  thd->catalog= 0;
+  thd->reset_query();
+  thd->reset_db(NULL, 0);
+  thd_proc_info(thd, "Slave worker thread exiting");
+  thd->temporary_tables= 0;
+  mysql_mutex_lock(&LOCK_thread_count);
+  THD_CHECK_SENTRY(thd);
+  delete thd;
+  mysql_mutex_unlock(&LOCK_thread_count);
+
+  mysql_mutex_lock(&rpt->LOCK_rpl_thread);
   rpt->running= false;
+  mysql_cond_signal(&rpt->COND_rpl_thread);
   mysql_mutex_unlock(&rpt->LOCK_rpl_thread);
 
+  my_thread_end();
+
   return NULL;
 }
 
@@ -344,6 +362,7 @@ rpl_parallel_change_thread_count(rpl_parallel_thread_pool *pool,
   {
     rpl_parallel_thread *rpt= pool->get_thread(NULL);
     rpt->stop= true;
+    mysql_cond_signal(&rpt->COND_rpl_thread);
     mysql_mutex_unlock(&rpt->LOCK_rpl_thread);
   }
 
@@ -354,7 +373,9 @@ rpl_parallel_change_thread_count(rpl_parallel_thread_pool *pool,
     while (rpt->running)
       mysql_cond_wait(&rpt->COND_rpl_thread, &rpt->LOCK_rpl_thread);
     mysql_mutex_unlock(&rpt->LOCK_rpl_thread);
-    delete rpt;
+    mysql_mutex_destroy(&rpt->LOCK_rpl_thread);
+    mysql_cond_destroy(&rpt->COND_rpl_thread);
+    my_free(rpt);
   }
 
   my_free(pool->threads);
@@ -386,6 +407,7 @@ err:
       mysql_mutex_lock(&new_free_list->LOCK_rpl_thread);
       new_free_list->delay_start= false;
       new_free_list->stop= true;
+      mysql_cond_signal(&new_free_list->COND_rpl_thread);
       while (!new_free_list->running)
         mysql_cond_wait(&new_free_list->COND_rpl_thread,
                         &new_free_list->LOCK_rpl_thread);
diff --git a/sql/sql_binlog.cc b/sql/sql_binlog.cc
index bef9a4c3475..df6aab88200 100644
--- a/sql/sql_binlog.cc
+++ b/sql/sql_binlog.cc
@@ -44,7 +44,6 @@
 
 void mysql_client_binlog_statement(THD* thd)
 {
-  struct rpl_group_info *rgi;
   DBUG_ENTER("mysql_client_binlog_statement");
   DBUG_PRINT("info",("binlog base64: '%*s'",
                      (int) (thd->lex->comment.length < 2048 ?
@@ -100,6 +99,7 @@ void mysql_client_binlog_statement(THD* thd)
   const char *error= 0;
   char *buf= (char *) my_malloc(decoded_len, MYF(MY_WME));
   Log_event *ev = 0;
+  struct rpl_group_info rgi(rli);
 
   /*
     Out of memory check
@@ -197,17 +197,8 @@ void mysql_client_binlog_statement(THD* thd)
         }
       }
 
-      if (!(rgi= rli->group_info))
-      {
-        if (!(rgi= rli->group_info= (struct rpl_group_info *)
-              my_malloc(sizeof(*rgi), MYF(0))))
-        {
-          my_error(ER_OUTOFMEMORY, MYF(0), sizeof(*rgi));
-          goto end;
-        }
-        bzero(rgi, sizeof(*rgi));
-      }
-      rgi->rli= rli;
+      rgi.rli= rli;
+      rgi.thd= thd;
       ev= Log_event::read_log_event(bufptr, event_len, &error,
                                     rli->relay_log.description_event_for_exec,
                                     0);
@@ -244,7 +235,7 @@ void mysql_client_binlog_statement(THD* thd)
         (ev->flags & LOG_EVENT_SKIP_REPLICATION_F ?
          OPTION_SKIP_REPLICATION : 0);
 
-      err= ev->apply_event(rgi);
+      err= ev->apply_event(&rgi);
 
       thd->variables.option_bits=
         (thd->variables.option_bits & ~OPTION_SKIP_REPLICATION) |
diff --git a/sql/sys_vars.cc b/sql/sys_vars.cc
index 1273bff1750..91f13bebd12 100644
--- a/sql/sys_vars.cc
+++ b/sql/sys_vars.cc
@@ -1442,11 +1442,9 @@ check_slave_parallel_threads(sys_var *self, THD *thd, set_var *var)
 {
   bool running;
 
-  mysql_mutex_unlock(&LOCK_global_system_variables);
   mysql_mutex_lock(&LOCK_active_mi);
   running= master_info_index->give_error_if_slave_running();
   mysql_mutex_unlock(&LOCK_active_mi);
-  mysql_mutex_lock(&LOCK_global_system_variables);
   if (running)
     return true;
 
@@ -1457,17 +1455,18 @@ static bool
 fix_slave_parallel_threads(sys_var *self, THD *thd, enum_var_type type)
 {
   bool running;
+  bool err= false;
 
   mysql_mutex_unlock(&LOCK_global_system_variables);
   mysql_mutex_lock(&LOCK_active_mi);
   running= master_info_index->give_error_if_slave_running();
   mysql_mutex_unlock(&LOCK_active_mi);
-  mysql_mutex_lock(&LOCK_global_system_variables);
   if (running || rpl_parallel_change_thread_count(&global_rpl_thread_pool,
                                                   opt_slave_parallel_threads))
-    return true;
+    err= true;
+  mysql_mutex_lock(&LOCK_global_system_variables);
 
-  return false;
+  return err;
 }
 
 
@@ -1497,7 +1496,7 @@ static Sys_var_ulong Sys_binlog_commit_wait_count(
 static Sys_var_ulong Sys_binlog_commit_wait_usec(
        "binlog_commit_wait_usec",
        "Maximum time, in microseconds, to wait for more commits to queue up "
-       " for binlog group commit. Only takes effect if the value of "
+       "for binlog group commit. Only takes effect if the value of "
        "binlog_commit_wait_count is non-zero.",
        GLOBAL_VAR(opt_binlog_commit_wait_usec), CMD_LINE(REQUIRED_ARG),
        VALID_RANGE(0, ULONG_MAX), DEFAULT(100000), BLOCK_SIZE(1));