BUG#34397106 GR secondary node crashes during relay log purge

Problem
-------
Group Replication secondary node crashed while purging of the relay log.

Analysis / Root-cause analysis
------------------------------
Server receives SIGSEGV signal when Global_THD_manager iterates over
threads using do_for_all_thd method.

Solution
--------
Threads which execute operations on log files register their
operations on global mysql_bin_log object. Next when log_in_use
or adjust_linfo_offsets functions are executed all registred
LOG_INFO objects are searched or adjusted. When a thread no longer
executes operations on a log it unregisters LOG_INFO from mysql_bin_log
object.

Change-Id: I0e229433278e8997ea4339eefc506dbfc47314d9

Author: Slawomir Maludzinski

sql/binlog.cc

@@ -2922,101 +2922,6 @@ static bool binlog_savepoint_rollback_can_release_mdl(handlerton *, THD *thd) {
   return !trans_cannot_safely_rollback(thd);
 }
 
-/**
-  Adjust log offset in the binary log file for all running slaves
-  This class implements call back function for do_for_all_thd().
-  It is called for each thd in thd list to adjust offset.
-*/
-class Adjust_offset : public Do_THD_Impl {
- public:
-  Adjust_offset(my_off_t value) : m_purge_offset(value) {}
-  void operator()(THD *thd) override {
-    LOG_INFO *linfo;
-    mysql_mutex_lock(&thd->LOCK_thd_data);
-    if ((linfo = thd->current_linfo)) {
-      /*
-        Index file offset can be less that purge offset only if
-        we just started reading the index file. In that case
-        we have nothing to adjust.
-      */
-      if (linfo->index_file_offset < m_purge_offset)
-        linfo->fatal = (linfo->index_file_offset != 0);
-      else
-        linfo->index_file_offset -= m_purge_offset;
-    }
-    mysql_mutex_unlock(&thd->LOCK_thd_data);
-  }
-
- private:
-  my_off_t m_purge_offset;
-};
-
-/*
-  Adjust the position pointer in the binary log file for all running slaves.
-
-  SYNOPSIS
-    adjust_linfo_offsets()
-    purge_offset	Number of bytes removed from start of log index file
-
-  NOTES
-    - This is called when doing a PURGE when we delete lines from the
-      index log file.
-
-  REQUIREMENTS
-    - Before calling this function, we have to ensure that no threads are
-      using any binary log file before purge_offset.
-
-  TODO
-    - Inform the slave threads that they should sync the position
-      in the binary log file with flush_relay_log_info.
-      Now they sync is done for next read.
-*/
-static void adjust_linfo_offsets(my_off_t purge_offset) {
-  Adjust_offset adjust_offset(purge_offset);
-  Global_THD_manager::get_instance()->do_for_all_thd(&adjust_offset);
-}
-
-/**
-  This class implements Call back function for do_for_all_thd().
-  It is called for each thd in thd list to count
-  threads using bin log file
-*/
-
-class Log_in_use : public Do_THD_Impl {
- public:
-  Log_in_use(const char *value) : m_log_name(value), m_count(0) {
-    m_log_name_len = strlen(m_log_name) + 1;
-  }
-  void operator()(THD *thd) override {
-    LOG_INFO *linfo;
-    mysql_mutex_lock(&thd->LOCK_thd_data);
-    if ((linfo = thd->current_linfo)) {
-      if (!strncmp(m_log_name, linfo->log_file_name, m_log_name_len)) {
-        LogErr(WARNING_LEVEL, ER_BINLOG_FILE_BEING_READ_NOT_PURGED, m_log_name,
-               thd->thread_id());
-        m_count++;
-      }
-    }
-    mysql_mutex_unlock(&thd->LOCK_thd_data);
-  }
-  int get_count() { return m_count; }
-
- private:
-  const char *m_log_name;
-  size_t m_log_name_len;
-  int m_count;
-};
-
-static int log_in_use(const char *log_name) {
-  Log_in_use log_in_use(log_name);
-#ifndef NDEBUG
-  if (current_thd)
-    DEBUG_SYNC(current_thd, "purge_logs_after_lock_index_before_thread_count");
-#endif
-  Global_THD_manager::get_instance()->do_for_all_thd(&log_in_use);
-  return log_in_use.get_count();
-}
-
 static bool purge_error_message(THD *thd, int res) {
   uint errcode;
 
@@ -3361,9 +3266,8 @@ bool show_binlog_events(THD *thd, MYSQL_BIN_LOG *binary_log) {
       goto err;
     }
 
-    mysql_mutex_lock(&thd->LOCK_thd_data);
-    thd->current_linfo = &linfo;
-    mysql_mutex_unlock(&thd->LOCK_thd_data);
+    linfo.thread_id = thd->thread_id();
+    binary_log->register_log_info(&linfo);
 
     BINLOG_FILE_READER binlog_file_reader(
         opt_source_verify_checksum,
@@ -3470,9 +3374,8 @@ bool show_binlog_events(THD *thd, MYSQL_BIN_LOG *binary_log) {
   } else
     my_eof(thd);
 
-  mysql_mutex_lock(&thd->LOCK_thd_data);
-  thd->current_linfo = nullptr;
-  mysql_mutex_unlock(&thd->LOCK_thd_data);
+  binary_log->unregister_log_info(&linfo);
+
   return !errmsg.empty();
 }
 
@@ -3553,6 +3456,7 @@ void MYSQL_BIN_LOG::cleanup() {
     mysql_mutex_destroy(&LOCK_sync);
     mysql_mutex_destroy(&LOCK_binlog_end_pos);
     mysql_mutex_destroy(&LOCK_xids);
+    mysql_mutex_destroy(&LOCK_log_info);
     mysql_cond_destroy(&update_cond);
     mysql_cond_destroy(&m_prep_xids_cond);
     if (!is_relay_log) {
@@ -3577,6 +3481,7 @@ void MYSQL_BIN_LOG::init_pthread_objects() {
   mysql_mutex_init(m_key_LOCK_binlog_end_pos, &LOCK_binlog_end_pos,
                    MY_MUTEX_INIT_FAST);
   mysql_mutex_init(m_key_LOCK_xids, &LOCK_xids, MY_MUTEX_INIT_FAST);
+  mysql_mutex_init(m_key_LOCK_log_info, &LOCK_log_info, MY_MUTEX_INIT_FAST);
   mysql_cond_init(m_key_update_cond, &update_cond);
   mysql_cond_init(m_key_prep_xids_cond, &m_prep_xids_cond);
   if (!is_relay_log) {
@@ -7316,6 +7221,64 @@ int MYSQL_BIN_LOG::get_gtid_executed(Sid_map *sid_map, Gtid_set *gtid_set) {
   return error;
 }
 
+void MYSQL_BIN_LOG::register_log_info(LOG_INFO *log_info) {
+  DBUG_TRACE;
+  MUTEX_LOCK(lock, &LOCK_log_info);
+  log_info_set.insert(log_info);
+}
+
+void MYSQL_BIN_LOG::unregister_log_info(LOG_INFO *log_info) {
+  DBUG_TRACE;
+  MUTEX_LOCK(lock, &LOCK_log_info);
+  log_info_set.erase(log_info);
+}
+
+int MYSQL_BIN_LOG::log_in_use(const char *log_name) {
+  DBUG_TRACE;
+#ifndef NDEBUG
+  if (current_thd)
+    DEBUG_SYNC(current_thd, "purge_logs_after_lock_index_before_thread_count");
+#endif
+
+  mysql_mutex_assert_owner(&LOCK_index);
+  MUTEX_LOCK(lock_log_info, &LOCK_log_info);
+
+  int count = 0;
+  int log_name_len = strlen(log_name) + 1;
+
+  std::for_each(
+      log_info_set.cbegin(), log_info_set.cend(),
+      [log_name, log_name_len, &count](LOG_INFO *log_info) {
+        if (!strncmp(log_name, log_info->log_file_name, log_name_len)) {
+          LogErr(WARNING_LEVEL, ER_BINLOG_FILE_BEING_READ_NOT_PURGED, log_name,
+                 log_info->thread_id);
+          count++;
+        }
+      });
+
+  return count;
+}
+
+void MYSQL_BIN_LOG::adjust_linfo_offsets(my_off_t purge_offset) {
+  DBUG_TRACE;
+
+  mysql_mutex_assert_owner(&LOCK_index);
+  MUTEX_LOCK(lock_log_info, &LOCK_log_info);
+
+  std::for_each(log_info_set.cbegin(), log_info_set.cend(),
+                [purge_offset](LOG_INFO *log_info) {
+                  /*
+                    Index file offset can be less that purge offset only if
+                    we just started reading the index file. In that case
+                    we have nothing to adjust.
+                  */
+                  if (log_info->index_file_offset < purge_offset)
+                    log_info->fatal = (log_info->index_file_offset != 0);
+                  else
+                    log_info->index_file_offset -= purge_offset;
+                });
+}
+
 /**
   Write the contents of the given IO_CACHE to the binary log.
 

sql/binlog.h

@@ -111,25 +111,22 @@ struct Binlog_user_var_event {
 #define LOG_CLOSE_TO_BE_OPENED 2
 #define LOG_CLOSE_STOP_EVENT 4
 
-/*
-  Note that we destroy the lock mutex in the destructor here.
-  This means that object instances cannot be destroyed/go out of scope
-  until we have reset thd->current_linfo to NULL;
- */
 struct LOG_INFO {
   char log_file_name[FN_REFLEN] = {0};
   my_off_t index_file_offset, index_file_start_offset;
   my_off_t pos;
   bool fatal;       // if the purge happens to give us a negative offset
   int entry_index;  // used in purge_logs(), calculatd in find_log_pos().
   int encrypted_header_size;
+  my_thread_id thread_id;
   LOG_INFO()
       : index_file_offset(0),
         index_file_start_offset(0),
         pos(0),
         fatal(false),
         entry_index(0),
-        encrypted_header_size(0) {
+        encrypted_header_size(0),
+        thread_id(0) {
     memset(log_file_name, 0, FN_REFLEN);
   }
 };
@@ -185,6 +182,8 @@ class MYSQL_BIN_LOG : public TC_LOG {
   PSI_mutex_key m_key_LOCK_sync;
   /** The instrumentation key to use for @ LOCK_xids. */
   PSI_mutex_key m_key_LOCK_xids;
+  /** The instrumentation key to use for @ m_key_LOCK_log_info. */
+  PSI_mutex_key m_key_LOCK_log_info;
   /** The instrumentation key to use for @ update_cond. */
   PSI_cond_key m_key_update_cond;
   /** The instrumentation key to use for @ prep_xids_cond. */
@@ -368,6 +367,7 @@ class MYSQL_BIN_LOG : public TC_LOG {
       PSI_mutex_key key_LOCK_flush_queue, PSI_mutex_key key_LOCK_log,
       PSI_mutex_key key_LOCK_binlog_end_pos, PSI_mutex_key key_LOCK_sync,
       PSI_mutex_key key_LOCK_sync_queue, PSI_mutex_key key_LOCK_xids,
+      PSI_mutex_key key_LOCK_log_info,
       PSI_mutex_key key_LOCK_wait_for_group_turn, PSI_cond_key key_COND_done,
       PSI_cond_key key_COND_flush_queue, PSI_cond_key key_update_cond,
       PSI_cond_key key_prep_xids_cond,
@@ -390,6 +390,7 @@ class MYSQL_BIN_LOG : public TC_LOG {
     m_key_LOCK_after_commit = key_LOCK_after_commit;
     m_key_LOCK_sync = key_LOCK_sync;
     m_key_LOCK_xids = key_LOCK_xids;
+    m_key_LOCK_log_info = key_LOCK_log_info;
     m_key_update_cond = key_update_cond;
     m_key_prep_xids_cond = key_prep_xids_cond;
     m_key_file_log = key_file_log;
@@ -1023,6 +1024,51 @@ class MYSQL_BIN_LOG : public TC_LOG {
     True while rotating binlog, which is caused by logging Incident_log_event.
   */
   bool is_rotating_caused_by_incident;
+
+ public:
+  /**
+    Register LOG_INFO so that log_in_use and adjust_linfo_offsets can
+    operate on all logs. Note that register_log_info, unregister_log_info,
+    log_in_use, adjust_linfo_offsets are is used on global mysql_bin_log object.
+    @param log_info pointer to LOG_INFO which is registred
+  */
+  void register_log_info(LOG_INFO *log_info);
+  /**
+    Unregister LOG_INFO when it is no longer needed.
+    @param log_info pointer to LOG_INFO which is registred
+  */
+  void unregister_log_info(LOG_INFO *log_info);
+  /**
+    Check if any threads use log name.
+    @note This method expects the LOCK_index to be taken so there are no
+    concurrent edits against linfo objects being iterated
+    @param log_name name of a log which is checked for usage
+
+  */
+  int log_in_use(const char *log_name);
+  /**
+    Adjust the position pointer in the binary log file for all running replicas.
+    SYNOPSIS
+      adjust_linfo_offsets()
+      purge_offset Number of bytes removed from start of log index file
+    NOTES
+      - This is called when doing a PURGE when we delete lines from the
+        index log file. This method expects the LOCK_index to be taken so there
+    are no concurrent edits against linfo objects being iterated. REQUIREMENTS
+      - Before calling this function, we have to ensure that no threads are
+        using any binary log file before purge_offset.
+    TODO
+      - Inform the replica threads that they should sync the position
+        in the binary log file with flush_relay_log_info.
+        Now they sync is done for next read.
+  */
+  void adjust_linfo_offsets(my_off_t purge_offset);
+
+ private:
+  mysql_mutex_t LOCK_log_info;
+  // Set of log info objects that are in usage and might prevent some other
+  // operations from executing.
+  std::set<LOG_INFO *> log_info_set;
 };
 
 struct LOAD_FILE_INFO {

sql/mysqld.cc

@@ -1089,6 +1089,7 @@ static PSI_mutex_key key_BINLOG_LOCK_binlog_end_pos;
 static PSI_mutex_key key_BINLOG_LOCK_sync;
 static PSI_mutex_key key_BINLOG_LOCK_sync_queue;
 static PSI_mutex_key key_BINLOG_LOCK_xids;
+static PSI_mutex_key key_BINLOG_LOCK_log_info;
 static PSI_mutex_key key_BINLOG_LOCK_wait_for_group_turn;
 static PSI_rwlock_key key_rwlock_global_sid_lock;
 PSI_rwlock_key key_rwlock_gtid_mode_lock;
@@ -4800,8 +4801,8 @@ int init_common_variables() {
       key_BINLOG_LOCK_flush_queue, key_BINLOG_LOCK_log,
       key_BINLOG_LOCK_binlog_end_pos, key_BINLOG_LOCK_sync,
       key_BINLOG_LOCK_sync_queue, key_BINLOG_LOCK_xids,
-      key_BINLOG_LOCK_wait_for_group_turn, key_BINLOG_COND_done,
-      key_BINLOG_COND_flush_queue, key_BINLOG_update_cond,
+      key_BINLOG_LOCK_log_info, key_BINLOG_LOCK_wait_for_group_turn,
+      key_BINLOG_COND_done, key_BINLOG_COND_flush_queue, key_BINLOG_update_cond,
       key_BINLOG_prep_xids_cond, key_BINLOG_COND_wait_for_group_turn,
       key_file_binlog, key_file_binlog_index, key_file_binlog_cache,
       key_file_binlog_index_cache);

sql/rpl_binlog_sender.cc

@@ -261,9 +261,8 @@ void Binlog_sender::init() {
   init_heartbeat_period();
   m_last_event_sent_ts = now_in_nanosecs();
 
-  mysql_mutex_lock(&thd->LOCK_thd_data);
-  thd->current_linfo = &m_linfo;
-  mysql_mutex_unlock(&thd->LOCK_thd_data);
+  m_linfo.thread_id = thd->thread_id();
+  mysql_bin_log.register_log_info(&m_linfo);
 
   /* Initialize the buffer only once. */
   m_packet.mem_realloc(PACKET_MIN_SIZE);  // size of the buffer
@@ -364,9 +363,7 @@ void Binlog_sender::cleanup() {
   if (m_transmit_started)
     (void)RUN_HOOK(binlog_transmit, transmit_stop, (thd, m_flag));
 
-  mysql_mutex_lock(&thd->LOCK_thd_data);
-  thd->current_linfo = nullptr;
-  mysql_mutex_unlock(&thd->LOCK_thd_data);
+  mysql_bin_log.unregister_log_info(&m_linfo);
 
   thd->variables.max_allowed_packet =
       global_system_variables.max_allowed_packet;
@@ -380,6 +377,7 @@ void Binlog_sender::cleanup() {
 
 void Binlog_sender::run() {
   DBUG_TRACE;
+
   init();
 
   unsigned int max_event_size =

sql/rpl_rli.cc

@@ -208,9 +208,9 @@ Relay_log_info::Relay_log_info(bool is_slave_recovery,
       PSI_NOT_INSTRUMENTED, key_RELAYLOG_LOCK_log,
       key_RELAYLOG_LOCK_log_end_pos, key_RELAYLOG_LOCK_sync,
       PSI_NOT_INSTRUMENTED, key_RELAYLOG_LOCK_xids, PSI_NOT_INSTRUMENTED,
-      PSI_NOT_INSTRUMENTED, PSI_NOT_INSTRUMENTED, key_RELAYLOG_update_cond,
-      PSI_NOT_INSTRUMENTED, PSI_NOT_INSTRUMENTED, key_file_relaylog,
-      key_file_relaylog_index, key_file_relaylog_cache,
+      PSI_NOT_INSTRUMENTED, PSI_NOT_INSTRUMENTED, PSI_NOT_INSTRUMENTED,
+      key_RELAYLOG_update_cond, PSI_NOT_INSTRUMENTED, PSI_NOT_INSTRUMENTED,
+      key_file_relaylog, key_file_relaylog_index, key_file_relaylog_cache,
       key_file_relaylog_index_cache);
 #endif
 

sql/sql_class.cc

@@ -739,7 +739,6 @@ THD::THD(bool enable_plugins)
   lex->thd = nullptr;
   lex->set_current_query_block(nullptr);
   m_lock_usec = 0L;
-  current_linfo = nullptr;
   slave_thread = false;
   memset(&variables, 0, sizeof(variables));
   m_thread_id = Global_THD_manager::reserved_thread_id;

sql/sql_class.h

@@ -2729,12 +2729,6 @@ class THD : public MDL_context_owner,
   /** number of name_const() substitutions, see sp_head.cc:subst_spvars() */
   uint query_name_consts;
 
-  /*
-    If we do a purge of binary logs, log index info of the threads
-    that are currently reading it needs to be adjusted. To do that
-    each thread that is using LOG_INFO needs to adjust the pointer to it
-  */
-  LOG_INFO *current_linfo;
   /* Used by the sys_var class to store temporary values */
   union {
     bool bool_value;