MDEV-4506: Parallel replication. Intermediate commit.

Hook in the wait-for-prior-commit logic (not really tested yet).
Clean up some resource maintenance around rpl_group_info (may still be some
smaller issues there though).
Add a ToDo list at the top of rpl_parallel.cc

sql/log_event.h

@@ -1376,6 +1376,26 @@ public:
     }
   }
 
+  static bool is_group_event(enum Log_event_type ev_type)
+  {
+    switch (ev_type)
+    {
+    case START_EVENT_V3:
+    case STOP_EVENT:
+    case ROTATE_EVENT:
+    case SLAVE_EVENT:
+    case FORMAT_DESCRIPTION_EVENT:
+    case INCIDENT_EVENT:
+    case HEARTBEAT_LOG_EVENT:
+    case BINLOG_CHECKPOINT_EVENT:
+    case GTID_LIST_EVENT:
+      return false;
+
+    default:
+      return true;
+    }
+  }
+  
 protected:
 
   /**

sql/mysqld.cc

@@ -772,7 +772,7 @@ PSI_mutex_key key_BINLOG_LOCK_index, key_BINLOG_LOCK_xid_list,
   key_PARTITION_LOCK_auto_inc;
 PSI_mutex_key key_RELAYLOG_LOCK_index;
 PSI_mutex_key key_LOCK_slave_state, key_LOCK_binlog_state,
-  key_LOCK_rpl_thread, key_LOCK_rpl_thread_pool;
+  key_LOCK_rpl_thread, key_LOCK_rpl_thread_pool, key_LOCK_parallel_entry;
 
 PSI_mutex_key key_LOCK_stats,
   key_LOCK_global_user_client_stats, key_LOCK_global_table_stats,
@@ -850,7 +850,8 @@ static PSI_mutex_info all_server_mutexes[]=
   { &key_LOCK_slave_state, "LOCK_slave_state", 0},
   { &key_LOCK_binlog_state, "LOCK_binlog_state", 0},
   { &key_LOCK_rpl_thread, "LOCK_rpl_thread", 0},
-  { &key_LOCK_rpl_thread_pool, "LOCK_rpl_thread_pool", 0}
+  { &key_LOCK_rpl_thread_pool, "LOCK_rpl_thread_pool", 0},
+  { &key_LOCK_parallel_entry, "LOCK_parallel_entry", 0}
 };
 
 PSI_rwlock_key key_rwlock_LOCK_grant, key_rwlock_LOCK_logger,

sql/mysqld.h

@@ -249,7 +249,7 @@ extern PSI_mutex_key key_BINLOG_LOCK_index, key_BINLOG_LOCK_xid_list,
   key_LOCK_error_messages, key_LOCK_thread_count, key_PARTITION_LOCK_auto_inc;
 extern PSI_mutex_key key_RELAYLOG_LOCK_index;
 extern PSI_mutex_key key_LOCK_slave_state, key_LOCK_binlog_state,
-  key_LOCK_rpl_thread, key_LOCK_rpl_thread_pool;
+  key_LOCK_rpl_thread, key_LOCK_rpl_thread_pool, key_LOCK_parallel_entry;
 
 extern PSI_mutex_key key_LOCK_stats,
   key_LOCK_global_user_client_stats, key_LOCK_global_table_stats,

sql/rpl_parallel.h

@@ -25,7 +25,6 @@ struct rpl_parallel_thread {
     Log_event *ev;
     struct rpl_group_info *rgi;
   } *event_queue, *last_in_queue;
-  rpl_parallel_thread *wait_for; /* ToDo: change this ... */
 };
 
 
@@ -52,6 +51,14 @@ struct rpl_parallel_entry {
   uint64 last_commit_id;
   bool active;
   rpl_parallel_thread *rpl_thread;
+  /*
+    The sub_id of the last transaction to commit within this domain_id.
+    Must be accessed under LOCK_parallel_entry protection.
+  */
+  uint64 last_committed_sub_id;
+  mysql_mutex_t LOCK_parallel_entry;
+  uint64 current_sub_id;
+  struct rpl_group_info *current_group_info;
 };
 struct rpl_parallel {
   HASH domain_hash;
@@ -60,7 +67,7 @@ struct rpl_parallel {
   rpl_parallel();
   ~rpl_parallel();
   rpl_parallel_entry *find(uint32 domain_id);
-  bool do_event(Relay_log_info *rli, Log_event *ev, THD *thd);
+  bool do_event(struct rpl_group_info *serial_rgi, Log_event *ev, THD *thd);
 };
 
 

sql/rpl_rli.cc

@@ -113,8 +113,6 @@ Relay_log_info::~Relay_log_info()
   mysql_cond_destroy(&sleep_cond);
   relay_log.cleanup();
   free_annotate_event();
-  if (group_info)
-    my_free(group_info);
   DBUG_VOID_RETURN;
 }
 
@@ -1532,4 +1530,28 @@ end:
   DBUG_RETURN(err);
 }
 
+
+rpl_group_info::rpl_group_info(Relay_log_info *rli_)
+  : rli(rli_), gtid_sub_id(0), wait_commit_sub_id(0), wait_commit_group_info(0),
+    parallel_entry(0)
+{
+  bzero(&current_gtid, sizeof(current_gtid));
+}
+
+
+int
+event_group_new_gtid(rpl_group_info *rgi, Gtid_log_event *gev)
+{
+  uint64 sub_id= rpl_global_gtid_slave_state.next_subid(gev->domain_id);
+  if (!sub_id)
+  {
+    return 1;
+  }
+  rgi->gtid_sub_id= sub_id;
+  rgi->current_gtid.server_id= gev->server_id;
+  rgi->current_gtid.domain_id= gev->domain_id;
+  rgi->current_gtid.seq_no= gev->seq_no;
+  return 0;
+}
+
 #endif

sql/rpl_rli.h

@@ -314,7 +314,7 @@ public:
   char slave_patternload_file[FN_REFLEN]; 
   size_t slave_patternload_file_size;  
 
-  /* Various data related to the currently executing event group. */
+  /* ToDo: We need to remove this, always use the per-transaction one to work with parallel replication. */
   struct rpl_group_info *group_info;
   rpl_parallel parallel;
 
@@ -610,6 +610,30 @@ struct rpl_group_info
   */
   uint64 gtid_sub_id;
   rpl_gtid current_gtid;
+  /*
+    This is used to keep transaction commit order.
+    We will signal this when we commit, and can register it to wait for the
+    commit_orderer of the previous commit to signal us.
+  */
+  wait_for_commit commit_orderer;
+  /*
+    If non-zero, the sub_id of a prior event group whose commit we have to wait
+    for before committing ourselves. Then wait_commit_group_info points to the
+    event group to wait for.
+
+    Before using this, rpl_parallel_entry::last_committed_sub_id should be
+    compared against wait_commit_sub_id. Only if last_committed_sub_id is
+    smaller than wait_commit_sub_id must the wait be done (otherwise the
+    waited-for transaction is already committed, so we would otherwise wait
+    for the wrong commit).
+  */
+  uint64 wait_commit_sub_id;
+  struct rpl_group_info *wait_commit_group_info;
+
+  struct rpl_parallel_entry *parallel_entry;
+
+  rpl_group_info(Relay_log_info *rli);
+  ~rpl_group_info() { };
 };
 
 
@@ -620,5 +644,6 @@ int init_relay_log_info(Relay_log_info* rli, const char* info_fname);
 extern struct rpl_slave_state rpl_global_gtid_slave_state;
 
 int rpl_load_gtid_slave_state(THD *thd);
+int event_group_new_gtid(rpl_group_info *rgi, Gtid_log_event *gev);
 
 #endif /* RPL_RLI_H */

sql/slave.cc

@@ -3177,7 +3177,8 @@ int apply_event_and_update_pos(Log_event* ev, THD* thd,
 
   @retval 1 The event was not applied.
 */
-static int exec_relay_log_event(THD* thd, Relay_log_info* rli)
+static int exec_relay_log_event(THD* thd, Relay_log_info* rli,
+                                rpl_group_info *serial_rgi)
 {
   DBUG_ENTER("exec_relay_log_event");
 
@@ -3201,6 +3202,7 @@ static int exec_relay_log_event(THD* thd, Relay_log_info* rli)
   if (ev)
   {
     int exec_res;
+    Log_event_type typ= ev->get_type_code();
 
     /*
       This tests if the position of the beginning of the current event
@@ -3230,8 +3232,8 @@ static int exec_relay_log_event(THD* thd, Relay_log_info* rli)
          read hanging if the realy log does not have any more events.
       */
       DBUG_EXECUTE_IF("incomplete_group_in_relay_log",
-                      if ((ev->get_type_code() == XID_EVENT) ||
-                          ((ev->get_type_code() == QUERY_EVENT) &&
+                      if ((typ == XID_EVENT) ||
+                          ((typ == QUERY_EVENT) &&
                            strcmp("COMMIT", ((Query_log_event *) ev)->query) == 0))
                       {
                         DBUG_ASSERT(thd->transaction.all.modified_non_trans_table);
@@ -3244,11 +3246,25 @@ static int exec_relay_log_event(THD* thd, Relay_log_info* rli)
     }
 
     if (opt_slave_parallel_threads > 0)
-      DBUG_RETURN(rli->parallel.do_event(rli, ev, thd));
+      DBUG_RETURN(rli->parallel.do_event(serial_rgi, ev, thd));
+
+    /*
+      For GTID, allocate a new sub_id for the given domain_id.
+      The sub_id must be allocated in increasing order of binlog order.
+    */
+    if (typ == GTID_EVENT &&
+        event_group_new_gtid(serial_rgi, static_cast<Gtid_log_event *>(ev)))
+    {
+      sql_print_error("Error reading relay log event: %s",
+                      "slave SQL thread aborted because of out-of-memory error");
+      mysql_mutex_unlock(&rli->data_lock);
+      delete ev;
+      DBUG_RETURN(1);
+    }
 
     exec_res= apply_event_and_update_pos(ev, thd, rli->group_info, NULL);
 
-    switch (ev->get_type_code()) {
+    switch (typ) {
       case FORMAT_DESCRIPTION_EVENT:
         /*
           Format_description_log_event should not be deleted because it
@@ -4001,6 +4017,7 @@ pthread_handler_t handle_slave_sql(void *arg)
   Master_info *mi= ((Master_info*)arg);
   Relay_log_info* rli = &mi->rli;
   const char *errmsg;
+  rpl_group_info serial_rgi(rli);
 
   // needs to call my_thread_init(), otherwise we get a coredump in DBUG_ stuff
   my_thread_init();
@@ -4205,6 +4222,13 @@ log '%s' at position %s, relay log '%s' position: %s%s", RPL_LOG_NAME,
   }
   mysql_mutex_unlock(&rli->data_lock);
 
+  /*
+    ToDo: Get rid of this, all accesses to rpl_group_info must be made
+    per-worker-thread to work with parallel replication.
+  */
+  if (opt_slave_parallel_threads <= 0)
+    rli->group_info= &serial_rgi;
+
   /* Read queries from the IO/THREAD until this thread is killed */
 
   while (!sql_slave_killed(thd,rli))
@@ -4227,7 +4251,7 @@ log '%s' at position %s, relay log '%s' position: %s%s", RPL_LOG_NAME,
       saved_skip= 0;
     }
     
-    if (exec_relay_log_event(thd,rli))
+    if (exec_relay_log_event(thd, rli, &serial_rgi))
     {
       DBUG_PRINT("info", ("exec_relay_log_event() failed"));
       // do not scare the user if SQL thread was simply killed or stopped
@@ -5736,7 +5760,6 @@ static Log_event* next_event(Relay_log_info* rli)
   mysql_mutex_t *log_lock = rli->relay_log.get_log_lock();
   const char* errmsg=0;
   THD* thd = rli->sql_thd;
-  struct rpl_group_info *rgi;
   DBUG_ENTER("next_event");
 
   DBUG_ASSERT(thd != 0);
@@ -5824,45 +5847,12 @@ static Log_event* next_event(Relay_log_info* rli)
                                        opt_slave_sql_verify_checksum)))
 
     {
-      if (!(rgi= rli->group_info))
-      {
-        if (!(rgi= rli->group_info= (struct rpl_group_info *)
-              my_malloc(sizeof(*rgi), MYF(0))))
-        {
-          errmsg = "slave SQL thread aborted because of out-of-memory error";
-          if (hot_log)
-            mysql_mutex_unlock(log_lock);
-          goto err;
-        }
-        bzero(rgi, sizeof(*rgi));
-      }
-      rgi->rli= rli;
       DBUG_ASSERT(thd==rli->sql_thd);
       /*
         read it while we have a lock, to avoid a mutex lock in
         inc_event_relay_log_pos()
       */
       rli->future_event_relay_log_pos= my_b_tell(cur_log);
-      /*
-        For GTID, allocate a new sub_id for the given domain_id.
-        The sub_id must be allocated in increasing order of binlog order.
-      */
-      if (ev->get_type_code() == GTID_EVENT)
-      {
-        Gtid_log_event *gev= static_cast<Gtid_log_event *>(ev);
-        uint64 sub_id= rpl_global_gtid_slave_state.next_subid(gev->domain_id);
-        if (!sub_id)
-        {
-          errmsg = "slave SQL thread aborted because of out-of-memory error";
-          if (hot_log)
-            mysql_mutex_unlock(log_lock);
-          goto err;
-        }
-        rgi->gtid_sub_id= sub_id;
-        rgi->current_gtid.server_id= gev->server_id;
-        rgi->current_gtid.domain_id= gev->domain_id;
-        rgi->current_gtid.seq_no= gev->seq_no;
-      }
 
       if (hot_log)
         mysql_mutex_unlock(log_lock);

sql/sql_class.cc

@@ -5602,6 +5602,13 @@ wait_for_commit::wait_for_commit()
 }
 
 
+wait_for_commit::~wait_for_commit()
+{
+  mysql_mutex_destroy(&LOCK_wait_commit);
+  mysql_cond_destroy(&COND_wait_commit);
+}
+
+
 void
 wait_for_commit::wakeup()
 {

sql/sql_class.h

@@ -1659,6 +1659,7 @@ struct wait_for_commit
   void unregister_wait_for_prior_commit2();
 
   wait_for_commit();
+  ~wait_for_commit();
 };