MDEV-6860 Parallel async replication hangs (#1400)

Instrumenting parallel slave worker thread with wsrep replication hooks.
Added mtr test for testing parallel slave support.
The test is based on the test attached in MDEV-6860 jira tracker.

4 files changed, 94 insertions, 0 deletions

diff --git a/mysql-test/suite/galera/r/MDEV-6860.result b/mysql-test/suite/galera/r/MDEV-6860.result
new file mode 100644
index 00000000000..72a743537c4
--- /dev/null
+++ b/mysql-test/suite/galera/r/MDEV-6860.result
@@ -0,0 +1,14 @@
+connection node_2;
+connection node_1;
+connection node_2;
+START SLAVE;
+connect node_3, 127.0.0.1, root, , test, $NODE_MYPORT_3;
+connection node_3;
+CREATE TABLE t1 (f1 int, f2 int) ENGINE=InnoDB;
+connection node_2;
+connection node_1;
+connection node_3;
+DROP TABLE t1;
+connection node_2;
+STOP SLAVE;
+RESET SLAVE ALL;
diff --git a/mysql-test/suite/galera/t/MDEV-6860.cnf b/mysql-test/suite/galera/t/MDEV-6860.cnf
new file mode 100644
index 00000000000..d8defd343ad
--- /dev/null
+++ b/mysql-test/suite/galera/t/MDEV-6860.cnf
@@ -0,0 +1,7 @@
+!include ../galera_2nodes_as_slave.cnf
+
+[mysqld.2]
+slave-parallel-threads=2
+slave-parallel-mode=optimistic
+[mysqld.1]
+wsrep-slave-threads=10
diff --git a/mysql-test/suite/galera/t/MDEV-6860.test b/mysql-test/suite/galera/t/MDEV-6860.test
new file mode 100644
index 00000000000..3a8c98f317f
--- /dev/null
+++ b/mysql-test/suite/galera/t/MDEV-6860.test
@@ -0,0 +1,42 @@
+--source include/have_innodb.inc
+
+--source include/galera_cluster.inc
+
+--connection node_2
+--disable_query_log
+--eval CHANGE MASTER TO  MASTER_HOST='127.0.0.1', MASTER_USER='root', MASTER_PORT=$NODE_MYPORT_3, MASTER_USE_GTID=slave_pos;
+--enable_query_log
+START SLAVE;
+
+--connect node_3, 127.0.0.1, root, , test, $NODE_MYPORT_3
+--connection node_3
+--let $inserts=1000
+CREATE TABLE t1 (f1 int, f2 int) ENGINE=InnoDB;
+
+--let $count=0
+--disable_query_log
+while($count < $inserts)
+{
+  --eval insert into t1 values ($count,1)
+  --inc $count
+}
+--enable_query_log
+
+--connection node_2
+
+--let $wait_condition = SELECT COUNT(*) = $inserts FROM t1
+--source include/wait_condition.inc
+
+--connection node_1
+--let $wait_condition = SELECT COUNT(*) = $inserts FROM t1
+--source include/wait_condition.inc
+
+--connection node_3
+DROP TABLE t1;
+
+--connection node_2
+--let $wait_condition = SELECT COUNT(*) = 0 FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA = 'test'
+--source include/wait_condition.inc
+
+STOP SLAVE;
+RESET SLAVE ALL;
diff --git a/sql/rpl_parallel.cc b/sql/rpl_parallel.cc
index 24db5f5fb5e..4313840119e 100644
--- a/sql/rpl_parallel.cc
+++ b/sql/rpl_parallel.cc
@@ -4,6 +4,10 @@
 #include "rpl_mi.h"
 #include "sql_parse.h"
 #include "debug_sync.h"
+#include "wsrep_mysqld.h"
+#ifdef WITH_WSREP
+#include "wsrep_trans_observer.h"
+#endif
 
 /*
   Code for optional parallel execution of replicated events on the slave.
@@ -35,6 +39,13 @@ rpt_handle_event(rpl_parallel_thread::queued_event *qev,
 
   DBUG_ASSERT(qev->typ == rpl_parallel_thread::queued_event::QUEUED_EVENT);
   ev= qev->ev;
+#ifdef WITH_WSREP
+  if (wsrep_before_statement(thd))
+  {
+    WSREP_WARN("Parallel slave failed at wsrep_before_statement() hook");
+    return(1);
+  }
+#endif /* WITH_WSREP */
 
   thd->system_thread_info.rpl_sql_info->rpl_filter = rli->mi->rpl_filter;
   ev->thd= thd;
@@ -50,6 +61,13 @@ rpt_handle_event(rpl_parallel_thread::queued_event *qev,
   err= apply_event_and_update_pos_for_parallel(ev, thd, rgi);
 
   thread_safe_increment64(&rli->executed_entries);
+#ifdef WITH_WSREP
+  if (wsrep_after_statement(thd))
+  {
+    WSREP_WARN("Parallel slave failed at wsrep_after_statement() hook");
+    err= 1;
+  }
+#endif /* WITH_WSREP */
   /* ToDo: error handling. */
   return err;
 }
@@ -1066,6 +1084,14 @@ handle_rpl_parallel_thread(void *arg)
   mysql_cond_signal(&rpt->COND_rpl_thread);
 
   thd->set_command(COM_SLAVE_WORKER);
+#ifdef WITH_WSREP
+  wsrep_open(thd);
+  if (wsrep_before_command(thd))
+  {
+    WSREP_WARN("Parallel slave failed at wsrep_before_command() hook");
+    rpt->stop = true;
+  }
+#endif /* WITH_WSREP */
   while (!rpt->stop)
   {
     uint wait_count= 0;
@@ -1436,6 +1462,11 @@ handle_rpl_parallel_thread(void *arg)
         rpt->pool->release_thread(rpt);
     }
   }
+#ifdef WITH_WSREP
+  wsrep_after_command_before_result(thd);
+  wsrep_after_command_after_result(thd);
+  wsrep_close(thd);
+#endif /* WITH_WSREP */
 
   rpt->thd= NULL;
   mysql_mutex_unlock(&rpt->LOCK_rpl_thread);