Merge MDEV-4506: Parallel replication into 10.0-base.

1 files changed, 157 insertions, 1 deletions

diff --git a/sql/sql_class.h b/sql/sql_class.h
index 13a3adf5909..de3357212a3 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -47,6 +47,7 @@
 
 class Reprepare_observer;
 class Relay_log_info;
+struct rpl_group_info;
 class Rpl_filter;
 
 class Query_log_event;
@@ -1556,6 +1557,120 @@ private:
 };
 
 
+/*
+  Class to facilitate the commit of one transactions waiting for the commit of
+  another transaction to complete first.
+
+  This is used during (parallel) replication, to allow different transactions
+  to be applied in parallel, but still commit in order.
+
+  The transaction that wants to wait for a prior commit must first register
+  to wait with register_wait_for_prior_commit(waitee). Such registration
+  must be done holding the waitee->LOCK_wait_commit, to prevent the other
+  THD from disappearing during the registration.
+
+  Then during commit, if a THD is registered to wait, it will call
+  wait_for_prior_commit() as part of ha_commit_trans(). If no wait is
+  registered, or if the waitee for has already completed commit, then
+  wait_for_prior_commit() returns immediately.
+
+  And when a THD that may be waited for has completed commit (more precisely
+  commit_ordered()), then it must call wakeup_subsequent_commits() to wake
+  up any waiters. Note that this must be done at a point that is guaranteed
+  to be later than any waiters registering themselves. It is safe to call
+  wakeup_subsequent_commits() multiple times, as waiters are removed from
+  registration as part of the wakeup.
+
+  The reason for separate register and wait calls is that this allows to
+  register the wait early, at a point where the waited-for THD is known to
+  exist. And then the actual wait can be done much later, where the
+  waited-for THD may have been long gone. By registering early, the waitee
+  can signal before disappearing.
+*/
+struct wait_for_commit
+{
+  /*
+    The LOCK_wait_commit protects the fields subsequent_commits_list and
+    wakeup_subsequent_commits_running (for a waitee), and the flag
+    waiting_for_commit and associated COND_wait_commit (for a waiter).
+  */
+  mysql_mutex_t LOCK_wait_commit;
+  mysql_cond_t COND_wait_commit;
+  /* List of threads that did register_wait_for_prior_commit() on us. */
+  wait_for_commit *subsequent_commits_list;
+  /* Link field for entries in subsequent_commits_list. */
+  wait_for_commit *next_subsequent_commit;
+  /* Our waitee, if we did register_wait_for_prior_commit(), else NULL. */
+  wait_for_commit *waitee;
+  /*
+    Generic pointer for use by the transaction coordinator to optimise the
+    waiting for improved group commit.
+
+    Currently used by binlog TC to signal that a waiter is ready to commit, so
+    that the waitee can grab it and group commit it directly. It is free to be
+    used by another transaction coordinator for similar purposes.
+  */
+  void *opaque_pointer;
+  /*
+    The waiting_for_commit flag is cleared when a waiter has been woken
+    up. The COND_wait_commit condition is signalled when this has been
+    cleared.
+  */
+  bool waiting_for_commit;
+  /* The wakeup error code from the waitee. 0 means no error. */
+  int wakeup_error;
+  /*
+    Flag set when wakeup_subsequent_commits_running() is active, see comments
+    on that function for details.
+  */
+  bool wakeup_subsequent_commits_running;
+
+  void register_wait_for_prior_commit(wait_for_commit *waitee);
+  int wait_for_prior_commit()
+  {
+    /*
+      Quick inline check, to avoid function call and locking in the common case
+      where no wakeup is registered, or a registered wait was already signalled.
+    */
+    if (waiting_for_commit)
+      return wait_for_prior_commit2();
+    else
+      return wakeup_error;
+  }
+  void wakeup_subsequent_commits(int wakeup_error)
+  {
+    /*
+      Do the check inline, so only the wakeup case takes the cost of a function
+      call for every commmit.
+
+      Note that the check is done without locking. It is the responsibility of
+      the user of the wakeup facility to ensure that no waiters can register
+      themselves after the last call to wakeup_subsequent_commits().
+
+      This avoids having to take another lock for every commit, which would be
+      pointless anyway - even if we check under lock, there is nothing to
+      prevent a waiter from arriving just after releasing the lock.
+    */
+    if (subsequent_commits_list)
+      wakeup_subsequent_commits2(wakeup_error);
+  }
+  void unregister_wait_for_prior_commit()
+  {
+    if (waiting_for_commit)
+      unregister_wait_for_prior_commit2();
+  }
+
+  void wakeup(int wakeup_error);
+
+  int wait_for_prior_commit2();
+  void wakeup_subsequent_commits2(int wakeup_error);
+  void unregister_wait_for_prior_commit2();
+
+  wait_for_commit();
+  ~wait_for_commit();
+};
+
+
 extern "C" void my_message_sql(uint error, const char *str, myf MyFlags);
 
 class THD;
@@ -1590,8 +1705,9 @@ public:
 
   /* Used to execute base64 coded binlog events in MySQL server */
   Relay_log_info* rli_fake;
+  rpl_group_info* rgi_fake;
   /* Slave applier execution context */
-  Relay_log_info* rli_slave;
+  rpl_group_info* rgi_slave;
 
   /* Used to SLAVE SQL thread */
   Rpl_filter* rpl_filter;
@@ -3207,6 +3323,25 @@ public:
   void wait_for_wakeup_ready();
   /* Wake this thread up from wait_for_wakeup_ready(). */
   void signal_wakeup_ready();
+
+  wait_for_commit *wait_for_commit_ptr;
+  int wait_for_prior_commit()
+  {
+    if (wait_for_commit_ptr)
+    {
+      int err= wait_for_commit_ptr->wait_for_prior_commit();
+      if (err)
+        my_error(ER_PRIOR_COMMIT_FAILED, MYF(0));
+      return err;
+    }
+    return 0;
+  }
+  void wakeup_subsequent_commits(int wakeup_error)
+  {
+    if (wait_for_commit_ptr)
+      wait_for_commit_ptr->wakeup_subsequent_commits(wakeup_error);
+  }
+
 private:
 
   /** The current internal error handler for this thread, or NULL. */
@@ -3260,6 +3395,27 @@ private:
   bool wakeup_ready;
   mysql_mutex_t LOCK_wakeup_ready;
   mysql_cond_t COND_wakeup_ready;
+
+  /* Protect against add/delete of temporary tables in parallel replication */
+  void rgi_lock_temporary_tables();
+  void rgi_unlock_temporary_tables();
+  bool rgi_have_temporary_tables();
+public:
+  inline void lock_temporary_tables()
+  {
+    if (rgi_slave)
+      rgi_lock_temporary_tables();
+  }
+  inline void unlock_temporary_tables()
+  {
+    if (rgi_slave)
+      rgi_unlock_temporary_tables();
+  }    
+  inline bool have_temporary_tables()
+  {
+    return (temporary_tables ||
+            (rgi_slave && rgi_have_temporary_tables()));
+  }
 };