wsrep_sst.cc-refactoring

diff --git a/sql/log.cc b/sql/log.cc
index a938dcb..953990f 100644
--- a/sql/log.cc
+++ b/sql/log.cc
@@ -3690,7 +3690,10 @@ bool MYSQL_BIN_LOG::open(const char *log_name,
     new_xid_list_entry->binlog_id= current_binlog_id;
     /* Remove any initial entries with no pending XIDs.  */
     while ((b= binlog_xid_count_list.head()) && b->xid_count == 0)
+    {
       my_free(binlog_xid_count_list.get());
+    }
+    mysql_cond_broadcast(&COND_xid_list);
     binlog_xid_count_list.push_back(new_xid_list_entry);
     mysql_mutex_unlock(&LOCK_xid_list);
 
@@ -4227,6 +4230,7 @@ bool MYSQL_BIN_LOG::reset_logs(THD *thd, bool create_new_log,
       DBUG_ASSERT(b->xid_count == 0);
       my_free(binlog_xid_count_list.get());
     }
+    mysql_cond_broadcast(&COND_xid_list);
     reset_master_pending--;
     mysql_mutex_unlock(&LOCK_xid_list);
   }
@@ -4237,6 +4241,19 @@ bool MYSQL_BIN_LOG::reset_logs(THD *thd, bool create_new_log,
 }
 
 
+void MYSQL_BIN_LOG::wait_for_last_checkpoint_event()
+{
+  mysql_mutex_lock(&LOCK_xid_list);
+  for (;;)
+  {
+    if (binlog_xid_count_list.is_last(binlog_xid_count_list.head()))
+      break;
+    mysql_cond_wait(&COND_xid_list, &LOCK_xid_list);
+  }
+  mysql_mutex_unlock(&LOCK_xid_list);
+}
+
+
 /**
   Delete relay log files prior to rli->group_relay_log_name
   (i.e. all logs which are not involved in a non-finished group
@@ -9394,7 +9411,7 @@ void TC_LOG_BINLOG::close()
   */
   if (unlikely(reset_master_pending))
   {
-    mysql_cond_signal(&COND_xid_list);
+    mysql_cond_broadcast(&COND_xid_list);
     mysql_mutex_unlock(&LOCK_xid_list);
     DBUG_VOID_RETURN;
   }
@@ -9432,8 +9449,7 @@ void TC_LOG_BINLOG::close()
   mysql_mutex_lock(&LOCK_log);
   mysql_mutex_lock(&LOCK_xid_list);
   --mark_xid_done_waiting;
-  if (unlikely(reset_master_pending))
-    mysql_cond_signal(&COND_xid_list);
+  mysql_cond_broadcast(&COND_xid_list);
   /* We need to reload current_binlog_id due to release/re-take of lock. */
   current= current_binlog_id;
 
diff --git a/sql/log.h b/sql/log.h
index 9bf80d6..d81c4b0 100644
--- a/sql/log.h
+++ b/sql/log.h
@@ -788,6 +788,7 @@ class MYSQL_BIN_LOG: public TC_LOG, private MYSQL_LOG
   bool reset_logs(THD* thd, bool create_new_log,
                   rpl_gtid *init_state, uint32 init_state_len,
                   ulong next_log_number);
+  void wait_for_last_checkpoint_event();
   void close(uint exiting);
   void clear_inuse_flag_when_closing(File file);
 
diff --git a/sql/wsrep_sst.cc b/sql/wsrep_sst.cc
index c2a1ab5..ade4414 100644
--- a/sql/wsrep_sst.cc
+++ b/sql/wsrep_sst.cc
@@ -949,36 +949,54 @@ static int sst_create_file(const char *name, const char *content)
 
 static int run_sql_command(THD *thd, const char *query)
 {
+  int err= 0;
+  CHARSET_INFO *current_charset= thd->variables.character_set_client;
+
+  if (!is_supported_parser_charset(current_charset))
+  {
+    /* Do not use non-supported parser character sets */
+    WSREP_WARN("Current client character set is non-supported parser "
+               "character set: %s", current_charset->csname);
+
+    thd->variables.character_set_client= &my_charset_latin1;
+
+    WSREP_WARN("For SST temporally setting character set to : %s",
+               my_charset_latin1.csname);
+  }
+
   thd->set_query((char *)query, strlen(query));
 
   Parser_state ps;
   if (ps.init(thd, thd->query(), thd->query_length()))
   {
     WSREP_ERROR("SST query: %s failed", query);
-    return -1;
+    err= -1;
+    goto end;
   }
 
   mysql_parse(thd, thd->query(), thd->query_length(), &ps);
+
   if (thd->is_error())
   {
-    int const err= thd->get_stmt_da()->sql_errno();
+    int const sql_errno= thd->get_stmt_da()->sql_errno();
     WSREP_WARN ("Error executing '%s': %d (%s)%s",
-                query, err, thd->get_stmt_da()->message(),
-                err == ER_UNKNOWN_SYSTEM_VARIABLE ?
+                query, sql_errno, thd->get_stmt_da()->message(),
+                sql_errno == ER_UNKNOWN_SYSTEM_VARIABLE ?
                 ". Was mysqld built with --with-innodb-disallow-writes ?" : "");
     thd->clear_error();
-    return -1;
+    err= -1;
   }
-  return 0;
+
+end:
+  /* Restore the local character_set_client. */
+  thd->variables.character_set_client= current_charset;
+
+  return err;
 }
 
 
 static int sst_flush_tables(THD* thd)
 {
-  WSREP_INFO("Flushing tables for SST...");
-
-  int err= 0;
-  int not_used;
   /*
     Files created to notify the SST script about the outcome of table flush
     operation.
@@ -986,95 +1004,92 @@ static int sst_flush_tables(THD* thd)
   const char *flush_success= "tables_flushed";
   const char *flush_error= "sst_error";
 
-  CHARSET_INFO *current_charset= thd->variables.character_set_client;
+  int err= -1;
+  int not_used;
 
-  if (!is_supported_parser_charset(current_charset))
-  {
-      /* Do not use non-supported parser character sets */
-      WSREP_WARN("Current client character set is non-supported parser character set: %s", current_charset->csname);
-      thd->variables.character_set_client = &my_charset_latin1;
-      WSREP_WARN("For SST temporally setting character set to : %s",
-	      my_charset_latin1.csname);
-  }
+  WSREP_INFO("Flushing tables for SST...");
 
   if (run_sql_command(thd, "FLUSH TABLES WITH READ LOCK"))
   {
-    err= -1;
+    goto err1;
   }
-  else
+
+  /*
+    Make sure logs are flushed after global read lock acquired. In case
+    reload fails, we must also release the acquired FTWRL.
+  */
+  if (reload_acl_and_cache(thd, REFRESH_ENGINE_LOG, (TABLE_LIST *) 0,
+                           &not_used))
   {
-    /*
-      Make sure logs are flushed after global read lock acquired. In case
-      reload fails, we must also release the acquired FTWRL.
-    */
-    if (reload_acl_and_cache(thd, REFRESH_ENGINE_LOG | REFRESH_BINARY_LOG,
-                             (TABLE_LIST*) 0, &not_used))
-    {
-      thd->global_read_lock.unlock_global_read_lock(thd);
-      err= -1;
-    }
+    goto err2;
   }
 
-  thd->variables.character_set_client = current_charset;
-
-  if (err)
+  if (run_sql_command(thd, "SET GLOBAL innodb_disallow_writes=1"))
   {
-    WSREP_ERROR("Failed to flush and lock tables");
-
-    /*
-      The SST must be aborted as the flush tables failed. Notify this to SST
-      script by creating the error file.
-    */
-    int tmp;
-    if ((tmp= sst_create_file(flush_error, NULL))) {
-      err= tmp;
-    }
+    WSREP_ERROR("Failed to disallow InnoDB writes.");
+    goto err2;
   }
-  else
+
+  if (mysql_bin_log.is_open())
   {
-    WSREP_INFO("Tables flushed.");
+    if (reload_acl_and_cache(thd, REFRESH_BINARY_LOG, (TABLE_LIST *) 0,
+                             &not_used))
+      goto err3;
 
     /*
-      Tables have been flushed. Create a file with cluster state ID and
-      wsrep_gtid_domain_id.
+      Wait for binlog checkpoint event for current binary log file.
+      LOCK_xid_list and LOCK_log are chained, so the LOCK_log will only be
+      obtained after mark_xid_done() has written the last checkpoint
+      event. We keep LOCK_log for the duration of file transfer to prevent
+      statements that modify binary logs (like RESET LOGS, RESET MASTER) from
+      proceeding until the files have been transferred to the joiner node.
     */
-    char content[100];
-    snprintf(content, sizeof(content), "%s:%lld %d\n", wsrep_cluster_state_uuid,
-             (long long)wsrep_locked_seqno, wsrep_gtid_domain_id);
-    err= sst_create_file(flush_success, content);
+    mysql_bin_log.wait_for_last_checkpoint_event();
+    mysql_mutex_lock(mysql_bin_log.get_log_lock());
   }
 
-  return err;
-}
+  WSREP_INFO("Tables flushed.");
 
+  /*
+    Tables have been flushed. Create a file with cluster state ID and
+    wsrep_gtid_domain_id.
+  */
+  char content[100];
+  snprintf(content, sizeof(content), "%s:%lld %d\n", wsrep_cluster_state_uuid,
+           (long long)wsrep_locked_seqno, wsrep_gtid_domain_id);
+  err= sst_create_file(flush_success, content);
+  if (err == 0)
+  {
+    return 0;
+  }
 
-static void sst_disallow_writes (THD* thd, bool yes)
-{
-  char query_str[64] = { 0, };
-  ssize_t const query_max = sizeof(query_str) - 1;
-  CHARSET_INFO *current_charset;
+err3:
+  /* Switch innodb_disallow_writes back to 0. */
+  run_sql_command(thd, "SET GLOBAL innodb_disallow_writes=0");
 
-  current_charset = thd->variables.character_set_client;
+err2:
+  /* Release the acquired FTWRL. */
+  thd->global_read_lock.unlock_global_read_lock(thd);
 
-  if (!is_supported_parser_charset(current_charset))
-  {
-      /* Do not use non-supported parser character sets */
-      WSREP_WARN("Current client character set is non-supported parser character set: %s", current_charset->csname);
-      thd->variables.character_set_client = &my_charset_latin1;
-      WSREP_WARN("For SST temporally setting character set to : %s",
-	      my_charset_latin1.csname);
-  }
+err1:
+  WSREP_ERROR("Failed to flush and lock tables");
 
-  snprintf (query_str, query_max, "SET GLOBAL innodb_disallow_writes=%d",
-            yes ? 1 : 0);
+  err= -1;
 
-  if (run_sql_command(thd, query_str))
+  /*
+    The SST must be aborted as the flush tables failed. Notify this to SST
+    script by creating the error file.
+  */
+  int tmp;
+  if ((tmp= sst_create_file(flush_error, NULL)))
   {
-    WSREP_ERROR("Failed to disallow InnoDB writes");
+    err= tmp;
   }
-  thd->variables.character_set_client = current_charset;
+
+  return err;
 }
 
+
 static void* sst_donor_thread (void* a)
 {
   sst_thread_arg* arg= (sst_thread_arg*)a;
@@ -1119,7 +1134,6 @@ static void* sst_donor_thread (void* a)
         err= sst_flush_tables (thd.ptr);
         if (!err)
         {
-          sst_disallow_writes (thd.ptr, true);
           locked= true;
           goto wait_signal;
         }
@@ -1128,7 +1142,12 @@ static void* sst_donor_thread (void* a)
       {
         if (locked)
         {
-          sst_disallow_writes (thd.ptr, false);
+          if (mysql_bin_log.is_open())
+          {
+            mysql_mutex_assert_owner(mysql_bin_log.get_log_lock());
+            mysql_mutex_unlock(mysql_bin_log.get_log_lock());
+          }
+          run_sql_command(thd.ptr, "SET GLOBAL innodb_disallow_writes=0");
           thd.ptr->global_read_lock.unlock_global_read_lock (thd.ptr);
           locked= false;
         }
@@ -1160,7 +1179,13 @@ static void* sst_donor_thread (void* a)
 
   if (locked) // don't forget to unlock server before return
   {
-    sst_disallow_writes (thd.ptr, false);
+    if (mysql_bin_log.is_open())
+    {
+      mysql_mutex_assert_owner(mysql_bin_log.get_log_lock());
+      mysql_mutex_unlock(mysql_bin_log.get_log_lock());
+    }
+
+    run_sql_command(thd.ptr, "SET GLOBAL innodb_disallow_writes=0");
     thd.ptr->global_read_lock.unlock_global_read_lock (thd.ptr);
   }