Bug#11763573 - 56299: MUTEX DEADLOCK WITH COM_BINLOG_DUMP, BINLOG PURGE, AND PROCESSLIST/KILL

The bug case is similar to one fixed earlier bug_49536.
Deadlock involving LOCK_log appears to be possible because the purge running thread
is holding LOCK_log whereas there is no sense of doing that and which fact was
exploited by the earlier bug fixes.

Fixed with small reengineering of rotate_and_purge(), adding two new methods and
setting up a policy to execute those instead of the former
rotate_and_purge(RP_LOCK_LOG_IS_ALREADY_LOCKED).
The policy for using rotate(), purge() is that if the caller acquires LOCK_log itself,
it should call rotate(), release the mutex and run purge().

Side effect of this patch is refining error message of bug@11747416 to print
the whole path.

7 files changed, 106 insertions, 47 deletions

diff --git a/sql/log.cc b/sql/log.cc
index c6b41447d6a..77ba7aa6283 100644
--- a/sql/log.cc
+++ b/sql/log.cc
@@ -49,7 +49,7 @@
 
 #include "sql_plugin.h"
 #include "rpl_handler.h"
-
+#include "debug_sync.h"
 /* max size of the log message */
 #define MAX_LOG_BUFFER_SIZE 1024
 #define MAX_TIME_SIZE 32
@@ -5002,19 +5002,29 @@ err:
       {
         bool synced;
         if ((error= flush_and_sync(&synced)))
-          goto unlock;
-
-        if ((error= RUN_HOOK(binlog_storage, after_flush,
+        {
+          mysql_mutex_unlock(&LOCK_log);
+        }
+        else if ((error= RUN_HOOK(binlog_storage, after_flush,
                  (thd, log_file_name, file->pos_in_file, synced))))
         {
           sql_print_error("Failed to run 'after_flush' hooks");
-          goto unlock;
+          mysql_mutex_unlock(&LOCK_log);
+        } 
+        else
+        {
+          bool check_purge;
+          signal_update();
+          error= rotate(false, &check_purge);
+          mysql_mutex_unlock(&LOCK_log);
+          if (!error && check_purge)
+            purge();
         }
-        signal_update();
-        rotate_and_purge(RP_LOCK_LOG_IS_ALREADY_LOCKED);
       }
-unlock:
-      mysql_mutex_unlock(&LOCK_log);
+      else
+      {
+        mysql_mutex_unlock(&LOCK_log);
+      }
     }
 
     if (error)
@@ -5100,25 +5110,29 @@ bool general_log_write(THD *thd, enum enum_server_command command,
 }
 
 /**
+  The method executes rotation when LOCK_log is already acquired
+  by the caller.
+
+  @param force_rotate  caller can request the log rotation
+  @param check_purge   is set to true if rotation took place
+
   @note
     If rotation fails, for instance the server was unable 
     to create a new log file, we still try to write an 
     incident event to the current log.
 
   @retval
-    nonzero - error 
+    nonzero - error in rotating routine.
 */
-int MYSQL_BIN_LOG::rotate_and_purge(uint flags)
+int MYSQL_BIN_LOG::rotate(bool force_rotate, bool* check_purge)
 {
   int error= 0;
-  DBUG_ENTER("MYSQL_BIN_LOG::rotate_and_purge");
-#ifdef HAVE_REPLICATION
-  bool check_purge= false;
-#endif
-  if (!(flags & RP_LOCK_LOG_IS_ALREADY_LOCKED))
-    mysql_mutex_lock(&LOCK_log);
-  if ((flags & RP_FORCE_ROTATE) ||
-      (my_b_tell(&log_file) >= (my_off_t) max_size))
+  DBUG_ENTER("MYSQL_BIN_LOG::rotate");
+
+  //todo: fix the macro def and restore safe_mutex_assert_owner(&LOCK_log);
+  *check_purge= false;
+
+  if (force_rotate || (my_b_tell(&log_file) >= (my_off_t) max_size))
   {
     if ((error= new_file_without_locking()))
       /** 
@@ -5133,24 +5147,59 @@ int MYSQL_BIN_LOG::rotate_and_purge(uint flags)
       if (!write_incident(current_thd, FALSE))
         flush_and_sync(0);
 
-#ifdef HAVE_REPLICATION
-    check_purge= true;
-#endif
+    *check_purge= true;
   }
-  if (!(flags & RP_LOCK_LOG_IS_ALREADY_LOCKED))
-    mysql_mutex_unlock(&LOCK_log);
+  DBUG_RETURN(error);
+}
+
+/**
+  The method executes logs purging routine.
+
+  @retval
+    nonzero - error in rotating routine.
+*/
+void MYSQL_BIN_LOG::purge()
+{
 #ifdef HAVE_REPLICATION
-  /*
-    NOTE: Run purge_logs wo/ holding LOCK_log
-          as it otherwise will deadlock in ndbcluster_binlog_index_purge_file
-  */
-  if (!error && check_purge && expire_logs_days)
+  if (expire_logs_days)
   {
+    DEBUG_SYNC(current_thd, "at_purge_logs_before_date");
     time_t purge_time= my_time(0) - expire_logs_days*24*60*60;
     if (purge_time >= 0)
+    {
       purge_logs_before_date(purge_time);
+    }
   }
 #endif
+}
+
+/**
+  The method is a shortcut of @c rotate() and @c purge().
+  LOCK_log is acquired prior to rotate and is released after it.
+
+  @param force_rotate  caller can request the log rotation
+
+  @retval
+    nonzero - error in rotating routine.
+*/
+int MYSQL_BIN_LOG::rotate_and_purge(bool force_rotate)
+{
+  int error= 0;
+  DBUG_ENTER("MYSQL_BIN_LOG::rotate_and_purge");
+  bool check_purge= false;
+
+  //todo: fix the macro def and restore safe_mutex_assert_not_owner(&LOCK_log);
+  mysql_mutex_lock(&LOCK_log);
+  error= rotate(force_rotate, &check_purge);
+  /*
+    NOTE: Run purge_logs wo/ holding LOCK_log because it does not need
+          the mutex. Otherwise causes various deadlocks.
+  */
+  mysql_mutex_unlock(&LOCK_log);
+
+  if (!error && check_purge)
+    purge();
+
   DBUG_RETURN(error);
 }
 
@@ -5352,10 +5401,17 @@ bool MYSQL_BIN_LOG::write_incident(THD *thd, bool lock)
   {
     if (!error && !(error= flush_and_sync(0)))
     {
+      bool check_purge= false;
       signal_update();
-      error= rotate_and_purge(RP_LOCK_LOG_IS_ALREADY_LOCKED);
+      error= rotate(false, &check_purge);
+      mysql_mutex_unlock(&LOCK_log);
+      if (!error && check_purge)
+        purge();
+    }
+    else
+    {
+      mysql_mutex_unlock(&LOCK_log);
     }
-    mysql_mutex_unlock(&LOCK_log);
   }
   DBUG_RETURN(error);
 }
@@ -5388,11 +5444,13 @@ bool MYSQL_BIN_LOG::write(THD *thd, IO_CACHE *cache, Log_event *commit_event,
                           bool incident)
 {
   DBUG_ENTER("MYSQL_BIN_LOG::write(THD *, IO_CACHE *, Log_event *)");
-  mysql_mutex_lock(&LOCK_log);
 
   DBUG_ASSERT(is_open());
   if (likely(is_open()))                       // Should always be true
   {
+    bool check_purge;
+    
+    mysql_mutex_lock(&LOCK_log);
     /*
       We only bother to write to the binary log if there is anything
       to write.
@@ -5460,12 +5518,17 @@ bool MYSQL_BIN_LOG::write(THD *thd, IO_CACHE *cache, Log_event *commit_event,
       mysql_mutex_lock(&LOCK_prep_xids);
       prepared_xids++;
       mysql_mutex_unlock(&LOCK_prep_xids);
+      mysql_mutex_unlock(&LOCK_log);
     }
     else
-      if (rotate_and_purge(RP_LOCK_LOG_IS_ALREADY_LOCKED))
+    {
+      if (rotate(false, &check_purge))
         goto err;
+      mysql_mutex_unlock(&LOCK_log);
+      if (check_purge) 
+        purge();
+    }
   }
-  mysql_mutex_unlock(&LOCK_log);
 
   DBUG_RETURN(0);
 
diff --git a/sql/log.h b/sql/log.h
index 0f0f81dd402..481bd545960 100644
--- a/sql/log.h
+++ b/sql/log.h
@@ -455,7 +455,9 @@ public:
   void make_log_name(char* buf, const char* log_ident);
   bool is_active(const char* log_file_name);
   int update_log_index(LOG_INFO* linfo, bool need_update_threads);
-  int rotate_and_purge(uint flags);
+  int rotate(bool force_rotate, bool* check_purge);
+  void purge();
+  int rotate_and_purge(bool force_rotate);
   /**
      Flush binlog cache and synchronize to disk.
 
diff --git a/sql/rpl_injector.cc b/sql/rpl_injector.cc
index d4e3e0fd7b4..c5fdc521f11 100644
--- a/sql/rpl_injector.cc
+++ b/sql/rpl_injector.cc
@@ -237,7 +237,7 @@ int injector::record_incident(THD *thd, Incident incident)
   Incident_log_event ev(thd, incident);
   if (int error= mysql_bin_log.write(&ev))
     return error;
-  return mysql_bin_log.rotate_and_purge(RP_FORCE_ROTATE);
+  return mysql_bin_log.rotate_and_purge(true);
 }
 
 int injector::record_incident(THD *thd, Incident incident, LEX_STRING const message)
@@ -245,5 +245,5 @@ int injector::record_incident(THD *thd, Incident incident, LEX_STRING const mess
   Incident_log_event ev(thd, incident, message);
   if (int error= mysql_bin_log.write(&ev))
     return error;
-  return mysql_bin_log.rotate_and_purge(RP_FORCE_ROTATE);
+  return mysql_bin_log.rotate_and_purge(true);
 }
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 9f13908d31c..1a51b66f8a6 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -161,9 +161,6 @@ typedef struct st_user_var_events
   bool unsigned_flag;
 } BINLOG_USER_VAR_EVENT;
 
-#define RP_LOCK_LOG_IS_ALREADY_LOCKED 1
-#define RP_FORCE_ROTATE               2
-
 /*
   The COPY_INFO structure is used by INSERT/REPLACE code.
   The schema of the row counting by the INSERT/INSERT ... ON DUPLICATE KEY
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 4fac64749e9..e3da697ec78 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -2861,7 +2861,7 @@ end_with_restore_list:
       {
         Incident_log_event ev(thd, incident);
         (void) mysql_bin_log.write(&ev);        /* error is ignored */
-        if (mysql_bin_log.rotate_and_purge(RP_FORCE_ROTATE))
+        if (mysql_bin_log.rotate_and_purge(true))
         {
           res= 1;
           break;
diff --git a/sql/sql_reload.cc b/sql/sql_reload.cc
index b567e3a1f85..089077bb89c 100644
--- a/sql/sql_reload.cc
+++ b/sql/sql_reload.cc
@@ -141,7 +141,7 @@ bool reload_acl_and_cache(THD *thd, unsigned long options,
     tmp_write_to_binlog= 0;
     if (mysql_bin_log.is_open())
     {
-      if (mysql_bin_log.rotate_and_purge(RP_FORCE_ROTATE))
+      if (mysql_bin_log.rotate_and_purge(true))
         *write_to_binlog= -1;
     }
   }
diff --git a/sql/sql_repl.cc b/sql/sql_repl.cc
index 80fef42434c..ef7d35ecfaa 100644
--- a/sql/sql_repl.cc
+++ b/sql/sql_repl.cc
@@ -1030,12 +1030,9 @@ err:
        detailing the fatal error message with coordinates 
        of the last position read.
     */
-    char b_start[FN_REFLEN], b_end[FN_REFLEN];
-    fn_format(b_start, coord->file_name, "", "", MY_REPLACE_DIR);
-    fn_format(b_end,   log_file_name,    "", "", MY_REPLACE_DIR);
     my_snprintf(error_text, sizeof(error_text), fmt, errmsg,
-                b_start, (llstr(coord->pos, llbuff1), llbuff1),
-                b_end, (llstr(my_b_tell(&log), llbuff2), llbuff2));
+                coord->file_name, (llstr(coord->pos, llbuff1), llbuff1),
+                log_file_name, (llstr(my_b_tell(&log), llbuff2), llbuff2));
   }
   else
     strcpy(error_text, errmsg);