1 files changed, 97 insertions, 34 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);