From c5986e9b260b681f33f48a44f02235e1bc412e43 Mon Sep 17 00:00:00 2001
From: sjaakola <seppo.jaakola@iki.fi>
Date: Thu, 19 Sep 2019 17:07:06 +0300
Subject: [PATCH] MDEV-6860 Parallel async replication hangs

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.
---
 mysql-test/suite/galera/r/MDEV-6860.result | 14 ++++++++
 mysql-test/suite/galera/t/MDEV-6860.cnf    |  7 ++++
 mysql-test/suite/galera/t/MDEV-6860.test   | 42 ++++++++++++++++++++++
 sql/rpl_parallel.cc                        | 31 ++++++++++++++++
 4 files changed, 94 insertions(+)
 create mode 100644 mysql-test/suite/galera/r/MDEV-6860.result
 create mode 100644 mysql-test/suite/galera/t/MDEV-6860.cnf
 create mode 100644 mysql-test/suite/galera/t/MDEV-6860.test

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 0000000000000..72a743537c457
--- /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 0000000000000..d8defd343adb7
--- /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 0000000000000..3a8c98f317f4a
--- /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 24db5f5fb5e80..4313840119ebe 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);