Bug#36423593 Purging ndb_binlog_index affects ndb binlog [part 2]

Problem:
This patch is for the second part of the problem where PURGE BINARY LOGS
holds the LOCK_index mutex for long duration and thus affects SHOW
BINARY LOGS as well as blocking purge of both relay and binlog.

Analysis:
The ha_binlog_index_purge_file() callback is called with the LOCK_index
mutex held. When the callback implementation need to do work that takes
some time it will thus block above mentioned commands. While the prior
patch managed to remove lots of work from the callback, the problem
still exists for PURGE BINARY LOGS which should not return until work is
complete.

Solution:
To allow PURGE BINARY LOGS to wait after having released LOCK_index a
new ha_binlog_index_purge_wait() callback is added. The new callback is
only called for PURGE BINARY LOGS and thus making it possible to
preserve the existing synchronous behaviour of the command. Ideally the
ha_binlog_index_purge_file() callback could have been moved to be called
outside of LOCK_index, but the context regarding which file(s) to purge
would then not be available in all code paths. Adding a new callback for
waiting is a much smaller change and affects only the two code paths for
PURGE BINARY LOGS TO and PURGE BINARY LOGS BEFORE. The new callback has
no
effect for storage engines not installing the handlerton::binlog_func
callback.

In addition the mystery code and comment stating that the
ha_binlog_index_purge_file() callback can only be called when LOCK_index
is not already held is removed. The callback should always be called and
then it's up to the storage engine implementor to decide what to do.

Change-Id: I4d19bd9b0a4bfa9820d11ff6d23a4bc4b0ea7e06
(cherry picked from commit 11f4459a1668fc2c0f2b59f29f4c4701785c610b)

Author: blaudden

sql/binlog.cc

@@ -3356,6 +3356,7 @@ bool purge_source_logs_to_file(THD *thd, const char *to_log) {
   auto purge_error = mysql_bin_log.purge_logs(
       search_file_name, include_to_log, need_index_lock, need_update_threads,
       nullptr, auto_purge);
+  ha_binlog_index_purge_wait(thd);
   return purge_error_message(thd, purge_error);
 }
 
@@ -3391,6 +3392,7 @@ bool purge_source_logs_before_date(THD *thd, time_t purge_time) {
   constexpr auto auto_purge{false};
   auto purge_error =
       mysql_bin_log.purge_logs_before_date(purge_time, auto_purge);
+  ha_binlog_index_purge_wait(thd);
   return purge_error_message(thd, purge_error);
 }
 
@@ -6370,17 +6372,9 @@ int MYSQL_BIN_LOG::purge_index_entry(THD *thd, ulonglong *decrease_log_space,
         }
 
         error = 0;
-        if (!need_lock_index) {
-          /*
-            This is to avoid triggering an error in NDB.
-
-            @todo: This is weird, what does NDB errors have to do with
-            need_lock_index? Explain better or refactor /Sven
-          */
-          if (!is_relay_log) {
-            // Only called when purging binlog
-            ha_binlog_index_purge_file(current_thd, log_info.log_file_name);
-          }
+        if (!is_relay_log) {
+          // Only called when purging binlog
+          ha_binlog_index_purge_file(thd, log_info.log_file_name);
         }
 
         DBUG_PRINT("info", ("purging %s", log_info.log_file_name));

sql/handler.cc

@@ -5910,6 +5910,12 @@ int ha_binlog_index_purge_file(THD *thd, const char *file) {
   return 0;
 }
 
+void ha_binlog_index_purge_wait(THD *thd) {
+  assert(thd);
+  binlog_func_st bfn = {BFN_BINLOG_PURGE_WAIT, nullptr};
+  binlog_func_foreach(thd, &bfn);
+}
+
 struct binlog_log_query_st {
   enum_binlog_command binlog_command;
   const char *query;

sql/handler.h

@@ -698,7 +698,8 @@ enum enum_binlog_func {
   BFN_RESET_SLAVE = 2,
   BFN_BINLOG_WAIT = 3,
   BFN_BINLOG_END = 4,
-  BFN_BINLOG_PURGE_FILE = 5
+  BFN_BINLOG_PURGE_FILE = 5,
+  BFN_BINLOG_PURGE_WAIT = 6
 };
 
 enum enum_binlog_command {
@@ -7582,7 +7583,42 @@ void trans_register_ha(THD *thd, bool all, handlerton *ht,
                        const ulonglong *trxid);
 
 int ha_reset_logs(THD *thd);
+
+/**
+  Inform storage engine(s) that a binary log file will be purged and any
+  references to it should be removed.
+
+  The function is called for all purged files, regardless if it is an explicit
+  PURGE BINARY LOGS statement, or an automatic purge performed by the server.
+
+  @note Since function is called with the LOCK_index mutex held the work
+  performed in this callback should be kept at minimum. One way to defer work is
+  to schedule work and use the `ha_binlog_index_purge_wait` callback to wait for
+  completion.
+
+  @param thd Thread handle of session purging file. The nullptr value indicates
+  that purge is done at server startup.
+  @param file Name of file being purged.
+  @return Always 0, return value are ignored by caller.
+*/
 int ha_binlog_index_purge_file(THD *thd, const char *file);
+
+/**
+  Request the storage engine to complete any operations that were initiated
+  by `ha_binlog_index_purge_file` and which need to complete
+  before PURGE BINARY LOGS completes.
+
+  The function is called only from PURGE BINARY LOGS. Each PURGE BINARY LOGS
+  statement will result in 0, 1 or more calls to `ha_binlog_index_purge_file`,
+  followed by exactly 1 call to `ha_binlog_index_purge_wait`.
+
+  @note This function is called without LOCK_index mutex held and thus any
+  waiting performed will only affect the current session.
+
+  @param thd Thread handle of session.
+*/
+void ha_binlog_index_purge_wait(THD *thd);
+
 void ha_reset_slave(THD *thd);
 void ha_binlog_log_query(THD *thd, handlerton *db_type,
                          enum_binlog_command binlog_command, const char *query,

storage/ndb/plugin/ha_ndbcluster_binlog.cc

@@ -320,6 +320,14 @@ static Ndb_binlog_purger ndb_binlog_purger(&opt_bin_log,
      binlog on the local server, if the auto purge conditions are exceeded at
      that time the applier will trigger a purge of the binlog.
 
+  @note This callback is normally called with the `LOCK_index` mutex held, this
+  means that time consuming work should not be performed as the mutex is
+  needed for other functionality and are assumed to not be held for long.
+  Instead, the work to remove references to the purged file are started
+  asynchronously in the background and waited for completion from
+  ndbcluster_binlog_index_purge_wait() which is then called after releasing
+  the mutex.
+
   @param thd Thread handle
   @param filename Name of the binlog file which has been removed
 */
@@ -332,7 +340,7 @@ static void ndbcluster_binlog_index_purge_file(THD *thd, const char *filename) {
     // No thd indicates auto purge at server startup, just submit the purged
     // file to purger who will remove the rows when server has started properly.
     ndb_log_info("Purging binlog file at server startup: '%s'", filename);
-    ndb_binlog_purger.submit_purge_binlog_file(filename);
+    ndb_binlog_purger.submit_purge_binlog_file(nullptr, filename);
     return;
   }
 
@@ -342,17 +350,22 @@ static void ndbcluster_binlog_index_purge_file(THD *thd, const char *filename) {
   }
 
   ndb_log_info("Purging binlog file: '%s'", filename);
-  ndb_binlog_purger.submit_purge_binlog_file(filename);
+  ndb_binlog_purger.submit_purge_binlog_file(thd, filename);
+}
+
+static void ndbcluster_binlog_index_purge_wait(THD *thd) {
+  DBUG_TRACE;
+  assert(thd);
+  // Only PURGE BINARY LOGS TO and PURGE BINARY LOGS BEFORE calls wait
+  assert(thd_sql_command(thd) == SQLCOM_PURGE ||
+         thd_sql_command(thd) == SQLCOM_PURGE_BEFORE);
 
-  if (thd_sql_command(thd) == SQLCOM_PURGE ||
-      thd_sql_command(thd) == SQLCOM_PURGE_BEFORE) {
-    // When the ndb binlog thread triggers a purge it should never wait
-    assert(!ndb_thd_is_binlog_thread(thd));
+  // When the ndb binlog thread triggers a purge it should never wait
+  assert(!ndb_thd_is_binlog_thread(thd));
 
-    // Wait until purger has removed all rows for the file
-    ndb_binlog_purger.wait_purge_binlog_file(filename);
-  }
-  return;
+  // Wait until purger has removed all files requested by this session
+  ndb_log_info("Waiting for purge to complete");
+  ndb_binlog_purger.wait_purge_completed_for_session(thd);
 }
 
 /*
@@ -613,6 +626,9 @@ static int ndbcluster_binlog_func(handlerton *, THD *thd, enum_binlog_func fn,
     case BFN_BINLOG_PURGE_FILE:
       ndbcluster_binlog_index_purge_file(thd, (const char *)arg);
       break;
+    case BFN_BINLOG_PURGE_WAIT:
+      ndbcluster_binlog_index_purge_wait(thd);
+      break;
   }
   return 0;
 }

storage/ndb/plugin/ndb_binlog_purger.cc

@@ -66,7 +66,8 @@ Ndb_binlog_purger::Ndb_binlog_purger(bool *binlogging_on, ulong *log_purge_rate)
 
 Ndb_binlog_purger::~Ndb_binlog_purger() {}
 
-void Ndb_binlog_purger::submit_purge_binlog_file(const std::string &filename) {
+void Ndb_binlog_purger::submit_purge_binlog_file(void *session,
+                                                 const std::string &filename) {
   if (is_stop_requested()) {
     // Does not happen, but better not accept new work
     log_error("Binlog file '%s' submitted while stopping", filename.c_str());
@@ -76,26 +77,27 @@ void Ndb_binlog_purger::submit_purge_binlog_file(const std::string &filename) {
   std::unique_lock files_lock(m_purge_files_lock);
 
   // Don't allow adding existing filename
-  if (std::ranges::any_of(m_purge_files,
-                          [filename](const auto f) { return f == filename; })) {
+  if (std::ranges::any_of(m_purge_files, [filename](const auto &request) {
+        return request.filename == filename;
+      })) {
     log_info("Binlog file '%s' was already submitted for purge, skipping",
              filename.c_str());
     return;
   }
 
-  m_purge_files.push_back(std::move(filename));
+  m_purge_files.push_back({std::move(filename), session});
   purge_queue_size++;
   assert(purge_queue_size == static_cast<long long>(m_purge_files.size()));
   m_purge_file_added_cond.notify_one();
 }
 
-void Ndb_binlog_purger::wait_purge_binlog_file(const std::string &filename) {
+void Ndb_binlog_purger::wait_purge_completed_for_session(void *session) {
   std::unique_lock files_lock(m_purge_files_lock);
 
-  m_purge_files_finished_cond.wait(files_lock, [this, filename]() {
+  m_purge_files_finished_cond.wait(files_lock, [this, session]() {
     return is_stop_requested() ||
-           std::ranges::none_of(m_purge_files, [filename](const auto f) {
-             return f == filename;
+           std::ranges::none_of(m_purge_files, [session](const auto &request) {
+             return request.session == session;
            });
   });
 }
@@ -137,7 +139,7 @@ void Ndb_binlog_purger::find_and_delete_orphan_purged_rows() {
   for (const auto &file : ndb_binlog_index_files_not_existing) {
     log_info("Found row(s) for '%s' which has been purged, removing it",
              file.c_str());
-    submit_purge_binlog_file(file);
+    submit_purge_binlog_file(nullptr, file);
   }
 }
 
@@ -229,7 +231,7 @@ void Ndb_binlog_purger::process_purge_first_file_completed(
   }
 #endif
   std::unique_lock files_lock(m_purge_files_lock);
-  assert(m_purge_files[0] == filename);
+  assert(m_purge_files[0].filename == filename);
   m_purge_files.erase(m_purge_files.begin());
   purged_files_count++;
   purge_queue_size--;
@@ -242,7 +244,7 @@ bool Ndb_binlog_purger::process_purge_first_file(Ndb_local_connection &con) {
   if (m_purge_files.empty()) return false;
 
   // Start processing file to remove.
-  const auto filename = m_purge_files[0];
+  const auto filename = m_purge_files[0].filename;
   log_info("Start purging binlog file: '%s'", filename.c_str());
 
   files_lock.unlock();
@@ -315,7 +317,7 @@ void Ndb_binlog_purger::process_purge_files_list() {
       log_error(
           "Got too many errors when removing rows for '%s' from "
           "ndb_binlog_index, skipping...",
-          m_purge_files[0].c_str());
+          m_purge_files[0].filename.c_str());
       m_purge_files.erase(m_purge_files.begin());
       purge_queue_size--;
       assert(purge_queue_size == static_cast<long long>(m_purge_files.size()));

storage/ndb/plugin/ndb_binlog_purger.h

@@ -46,8 +46,14 @@ class Ndb_binlog_purger : public Ndb_component {
   THD *m_thd{nullptr};
   void create_thd(void *stackptr);
 
+  // Name of one file to purge and the session which requested it
+  struct PurgeRequest {
+    std::string filename;
+    void *session;
+  };
+
   // List of purged files whose rows need to be removed
-  std::vector<std::string> m_purge_files;
+  std::vector<PurgeRequest> m_purge_files;
   // Mutex protecting the list of files to purge
   std::mutex m_purge_files_lock;
   // Condition used by purger to wait until there are new files to
@@ -91,16 +97,18 @@ class Ndb_binlog_purger : public Ndb_component {
     @brief Submit the name of a purged binlog file for aynchrounous removal
     of corresponding rows from the ndb_binlog_index table.
 
+    @param session Identifies the session requesting purge. Used for being able
+    to wait for purge to complete, use nullptr when there is no need to wait.
     @param filename Name of the binlog file which has been purged
   */
-  void submit_purge_binlog_file(const std::string &filename);
+  void submit_purge_binlog_file(void *session, const std::string &filename);
 
   /*
-    @brief Wait until removal of rows for the given filename has completed.
+    @brief Wait until removal of files for the given session has completed.
 
-    @param filename Name of the binlog file to wait for removal of.
+    @param session Identifies the session requesting purge..
   */
-  void wait_purge_binlog_file(const std::string &filename);
+  void wait_purge_completed_for_session(void *session);
 
  private:
   virtual int do_init() override;