Merge branch '10.4' into 10.5

1 files changed, 51 insertions, 5 deletions

diff --git a/sql/log.cc b/sql/log.cc
index 2a887a68606..921d68b9feb 100644
--- a/sql/log.cc
+++ b/sql/log.cc
@@ -6574,11 +6574,26 @@ bool MYSQL_BIN_LOG::write(Log_event *event_info, my_bool *with_annotate)
 
     if (direct)
     {
+      /* We come here only for incident events */
       int res;
       uint64 commit_id= 0;
+      MDL_request mdl_request;
       DBUG_PRINT("info", ("direct is set"));
+      DBUG_ASSERT(!thd->backup_commit_lock);
+
+      MDL_REQUEST_INIT(&mdl_request, MDL_key::BACKUP, "", "", MDL_BACKUP_COMMIT,
+                     MDL_EXPLICIT);
+      thd->mdl_context.acquire_lock(&mdl_request,
+                                    thd->variables.lock_wait_timeout);
+      thd->backup_commit_lock= &mdl_request;
+
       if ((res= thd->wait_for_prior_commit()))
+      {
+        if (mdl_request.ticket)
+          thd->mdl_context.release_lock(mdl_request.ticket);
+        thd->backup_commit_lock= 0;
         DBUG_RETURN(res);
+      }
       file= &log_file;
       my_org_b_tell= my_b_tell(file);
       mysql_mutex_lock(&LOCK_log);
@@ -6593,7 +6608,11 @@ bool MYSQL_BIN_LOG::write(Log_event *event_info, my_bool *with_annotate)
                                              commit_name.length);
           commit_id= entry->val_int(&null_value);
         });
-      if (write_gtid_event(thd, true, using_trans, commit_id))
+      res= write_gtid_event(thd, true, using_trans, commit_id);
+      if (mdl_request.ticket)
+        thd->mdl_context.release_lock(mdl_request.ticket);
+      thd->backup_commit_lock= 0;
+      if (res)
         goto err;
     }
     else
@@ -7657,7 +7676,11 @@ MYSQL_BIN_LOG::queue_for_group_commit(group_commit_entry *orig_entry)
   group_commit_entry *entry, *orig_queue, *last;
   wait_for_commit *cur;
   wait_for_commit *wfc;
+  bool backup_lock_released= 0;
+  int result= 0;
+  THD *thd= orig_entry->thd;
   DBUG_ENTER("MYSQL_BIN_LOG::queue_for_group_commit");
+  DBUG_ASSERT(thd == current_thd);
 
   /*
     Check if we need to wait for another transaction to commit before us.
@@ -7689,6 +7712,21 @@ MYSQL_BIN_LOG::queue_for_group_commit(group_commit_entry *orig_entry)
     {
       PSI_stage_info old_stage;
 
+        /*
+          Release MDL_BACKUP_COMMIT LOCK while waiting for other threads to
+          commit.
+          This is needed to avoid deadlock between the other threads (which not
+          yet have the MDL_BACKUP_COMMIT_LOCK) and any threads using
+          BACKUP LOCK BLOCK_COMMIT.
+        */
+      if (thd->backup_commit_lock && thd->backup_commit_lock->ticket &&
+          !backup_lock_released)
+      {
+        backup_lock_released= 1;
+        thd->mdl_context.release_lock(thd->backup_commit_lock->ticket);
+        thd->backup_commit_lock->ticket= 0;
+      }
+
       /*
         By setting wfc->opaque_pointer to our own entry, we mark that we are
         ready to commit, but waiting for another transaction to commit before
@@ -7749,7 +7787,8 @@ MYSQL_BIN_LOG::queue_for_group_commit(group_commit_entry *orig_entry)
             wfc->wakeup_error= ER_QUERY_INTERRUPTED;
           my_message(wfc->wakeup_error,
                      ER_THD(orig_entry->thd, wfc->wakeup_error), MYF(0));
-          DBUG_RETURN(-1);
+          result= -1;
+          goto end;
         }
       }
       orig_entry->thd->EXIT_COND(&old_stage);
@@ -7763,12 +7802,13 @@ MYSQL_BIN_LOG::queue_for_group_commit(group_commit_entry *orig_entry)
     then there is nothing else to do.
   */
   if (orig_entry->queued_by_other)
-    DBUG_RETURN(0);
+    goto end;
 
   if (wfc && wfc->wakeup_error)
   {
     my_error(ER_PRIOR_COMMIT_FAILED, MYF(0));
-    DBUG_RETURN(-1);
+    result= -1;
+    goto end;
   }
 
   /* Now enqueue ourselves in the group commit queue. */
@@ -7929,7 +7969,13 @@ MYSQL_BIN_LOG::queue_for_group_commit(group_commit_entry *orig_entry)
 
   DBUG_PRINT("info", ("Queued for group commit as %s",
                       (orig_queue == NULL) ? "leader" : "participant"));
-  DBUG_RETURN(orig_queue == NULL);
+  result= orig_queue == NULL;
+
+end:
+  if (backup_lock_released)
+    thd->mdl_context.acquire_lock(thd->backup_commit_lock,
+                                  thd->variables.lock_wait_timeout);
+  DBUG_RETURN(result);
 }
 
 bool