Adding upstream version 1:10.11.6.upstream/1%10.11.6upstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 3453 insertions, 0 deletions

diff --git a/sql/rpl_parallel.cc b/sql/rpl_parallel.cc
new file mode 100644
index 00000000..95a1234e
--- /dev/null
+++ b/sql/rpl_parallel.cc
@@ -0,0 +1,3453 @@
+#include "mariadb.h"
+#include "rpl_parallel.h"
+#include "slave.h"
+#include "rpl_mi.h"
+#include "sql_parse.h"
+#include "debug_sync.h"
+#include "sql_repl.h"
+#include "wsrep_mysqld.h"
+#ifdef WITH_WSREP
+#include "wsrep_trans_observer.h"
+#endif
+
+/*
+  Code for optional parallel execution of replicated events on the slave.
+*/
+
+
+/*
+  Maximum number of queued events to accumulate in a local free list, before
+  moving them to the global free list. There is additional a limit of how much
+  to accumulate based on opt_slave_parallel_max_queued.
+*/
+#define QEV_BATCH_FREE 200
+
+
+struct rpl_parallel_thread_pool global_rpl_thread_pool;
+
+static void signal_error_to_sql_driver_thread(THD *thd, rpl_group_info *rgi,
+                                              int err);
+static void
+register_wait_for_prior_event_group_commit(rpl_group_info *rgi,
+                                           rpl_parallel_entry *entry);
+
+static int
+rpt_handle_event(rpl_parallel_thread::queued_event *qev,
+                 struct rpl_parallel_thread *rpt)
+{
+  int err;
+  rpl_group_info *rgi= qev->rgi;
+  Relay_log_info *rli= rgi->rli;
+  THD *thd= rgi->thd;
+  Log_event *ev;
+
+  DBUG_ASSERT(qev->typ == rpl_parallel_thread::queued_event::QUEUED_EVENT);
+  ev= qev->ev;
+#ifdef WITH_WSREP
+  if (wsrep_before_statement(thd))
+  {
+    WSREP_WARN("Parallel slave failed at wsrep_before_statement() hook");
+    return(1);
+  }
+#endif /* WITH_WSREP */
+
+  thd->system_thread_info.rpl_sql_info->rpl_filter = rli->mi->rpl_filter;
+  ev->thd= thd;
+
+  strcpy(rgi->event_relay_log_name_buf, qev->event_relay_log_name);
+  rgi->event_relay_log_name= rgi->event_relay_log_name_buf;
+  rgi->event_relay_log_pos= qev->event_relay_log_pos;
+  rgi->future_event_relay_log_pos= qev->future_event_relay_log_pos;
+  strcpy(rgi->future_event_master_log_name, qev->future_event_master_log_name);
+  if (event_can_update_last_master_timestamp(ev))
+    rgi->last_master_timestamp= ev->when + (time_t)ev->exec_time;
+  err= apply_event_and_update_pos_for_parallel(ev, thd, rgi);
+
+  rli->executed_entries++;
+#ifdef WITH_WSREP
+  if (wsrep_after_statement(thd))
+  {
+    WSREP_WARN("Parallel slave failed at wsrep_after_statement() hook");
+    err= 1;
+  }
+#endif /* WITH_WSREP */
+  /* ToDo: error handling. */
+  return err;
+}
+
+
+static void
+handle_queued_pos_update(THD *thd, rpl_parallel_thread::queued_event *qev)
+{
+  int cmp;
+  Relay_log_info *rli;
+  rpl_parallel_entry *e;
+
+  /*
+    Events that are not part of an event group, such as Format Description,
+    Stop, GTID List and such, are executed directly in the driver SQL thread,
+    to keep the relay log state up-to-date. But the associated position update
+    is done here, in sync with other normal events as they are queued to
+    worker threads.
+  */
+  if ((thd->variables.option_bits & OPTION_BEGIN) &&
+      opt_using_transactions)
+    return;
+
+  /* Do not update position if an earlier event group caused an error abort. */
+  DBUG_ASSERT(qev->typ == rpl_parallel_thread::queued_event::QUEUED_POS_UPDATE);
+  rli= qev->rgi->rli;
+  e= qev->entry_for_queued;
+  if (e->stop_on_error_sub_id < (uint64)ULONGLONG_MAX ||
+      (e->force_abort && !rli->stop_for_until))
+    return;
+
+  mysql_mutex_lock(&rli->data_lock);
+  cmp= compare_log_name(rli->group_relay_log_name, qev->event_relay_log_name);
+  if (cmp < 0)
+  {
+    rli->group_relay_log_pos= qev->future_event_relay_log_pos;
+    strmake_buf(rli->group_relay_log_name, qev->event_relay_log_name);
+  } else if (cmp == 0 &&
+             rli->group_relay_log_pos < qev->future_event_relay_log_pos)
+    rli->group_relay_log_pos= qev->future_event_relay_log_pos;
+
+  cmp= compare_log_name(rli->group_master_log_name, qev->future_event_master_log_name);
+  if (cmp < 0)
+  {
+    strcpy(rli->group_master_log_name, qev->future_event_master_log_name);
+    rli->group_master_log_pos= qev->future_event_master_log_pos;
+  }
+  else if (cmp == 0
+           && rli->group_master_log_pos < qev->future_event_master_log_pos)
+    rli->group_master_log_pos= qev->future_event_master_log_pos;
+  mysql_mutex_unlock(&rli->data_lock);
+  mysql_cond_broadcast(&rli->data_cond);
+}
+
+
+/*
+  Wait for any pending deadlock kills. Since deadlock kills happen
+  asynchronously, we need to be sure they will be completed before starting a
+  new transaction. Otherwise the new transaction might suffer a spurious kill.
+*/
+static void
+wait_for_pending_deadlock_kill(THD *thd, rpl_group_info *rgi)
+{
+  PSI_stage_info old_stage;
+
+  mysql_mutex_lock(&thd->LOCK_wakeup_ready);
+  thd->ENTER_COND(&thd->COND_wakeup_ready, &thd->LOCK_wakeup_ready,
+                  &stage_waiting_for_deadlock_kill, &old_stage);
+  while (rgi->killed_for_retry == rpl_group_info::RETRY_KILL_PENDING)
+    mysql_cond_wait(&thd->COND_wakeup_ready, &thd->LOCK_wakeup_ready);
+  thd->EXIT_COND(&old_stage);
+}
+
+
+static void
+finish_event_group(rpl_parallel_thread *rpt, uint64 sub_id,
+                   rpl_parallel_entry *entry, rpl_group_info *rgi)
+{
+  THD *thd= rpt->thd;
+  wait_for_commit *wfc= &rgi->commit_orderer;
+  int err;
+
+  if (rgi->get_finish_event_group_called())
+    return;
+
+  thd->get_stmt_da()->set_overwrite_status(true);
+
+  if (unlikely(rgi->worker_error))
+  {
+    /*
+      In case a previous wait was killed, we need to re-register to be able to
+      repeat the wait.
+
+      And before doing that, we un-register any previous registration (in case
+      we got an error earlier and skipped waiting).
+    */
+    thd->wait_for_commit_ptr->unregister_wait_for_prior_commit();
+    mysql_mutex_lock(&entry->LOCK_parallel_entry);
+    register_wait_for_prior_event_group_commit(rgi, entry);
+    mysql_mutex_unlock(&entry->LOCK_parallel_entry);
+  }
+
+  /*
+    Remove any left-over registration to wait for a prior commit to
+    complete. Normally, such wait would already have been removed at
+    this point by wait_for_prior_commit() called from within COMMIT
+    processing.
+
+    However, in case of MyISAM and no binlog, we might not have any commit
+    processing, and so we need to do the wait here, before waking up any
+    subsequent commits, to preserve correct order of event execution.
+
+    Also, in the error case we might have skipped waiting and thus need to
+    remove it explicitly. Or the wait might have been killed and we need to
+    repeat the registration and the wait.
+
+    It is important in the non-error case to do a wait, not just an
+    unregister. Because we might be last in a group-commit that is
+    replicated in parallel, and the following event will then wait
+    for us to complete and rely on this also ensuring that any other
+    event in the group has completed.
+
+    And in the error case, correct GCO lifetime relies on the fact that once
+    the last event group in the GCO has executed wait_for_prior_commit(),
+    all earlier event groups have also committed; this way no more
+    mark_start_commit() calls can be made and it is safe to de-allocate
+    the GCO.
+
+    Thus this final wait is done with kill ignored during the wait. This is
+    fine, at this point there is no active query or transaction to abort, and
+    the thread will continue as soon as earlier event groups complete.
+
+    Note though, that in the non-error case there is no guarantee that
+    finish_event_group() will be run in-order. For example, a successful
+    binlog group commit will wakeup all participating event groups
+    simultaneously so only thread scheduling will decide the order in which
+    finish_event_group() calls acquire LOCK_parallel_entry.
+  */
+  err= wfc->wait_for_prior_commit(thd, false);
+  if (unlikely(err) && !rgi->worker_error)
+    signal_error_to_sql_driver_thread(thd, rgi, err);
+  thd->wait_for_commit_ptr= NULL;
+
+  mysql_mutex_lock(&entry->LOCK_parallel_entry);
+  /*
+    We need to mark that this event group started its commit phase, in case we
+    missed it before (otherwise we would deadlock the next event group that is
+    waiting for this). In most cases (normal DML), it will be a no-op.
+  */
+  rgi->mark_start_commit_no_lock();
+  rgi->commit_orderer.wakeup_blocked= false;
+
+  if (entry->last_committed_sub_id < sub_id)
+  {
+    /*
+      Record that this event group has finished (eg. transaction is
+      committed, if transactional), so other event groups will no longer
+      attempt to wait for us to commit. Once we have increased
+      entry->last_committed_sub_id, no other threads will execute
+      register_wait_for_prior_commit() against us. Thus, by doing one
+      extra (usually redundant) wakeup_subsequent_commits() we can ensure
+      that no register_wait_for_prior_commit() can ever happen without a
+      subsequent wakeup_subsequent_commits() to wake it up.
+
+      We can race here with the next transactions, but that is fine, as
+      long as we check that we do not decrease last_committed_sub_id. If
+      this commit is done, then any prior commits will also have been
+      done and also no longer need waiting for.
+    */
+    entry->last_committed_sub_id= sub_id;
+    if (entry->need_sub_id_signal)
+      mysql_cond_broadcast(&entry->COND_parallel_entry);
+
+    /* Now free any GCOs in which all transactions have committed. */
+    group_commit_orderer *tmp_gco= rgi->gco;
+    while (tmp_gco &&
+           (!tmp_gco->next_gco || tmp_gco->last_sub_id > sub_id ||
+            tmp_gco->next_gco->wait_count > entry->count_committing_event_groups))
+    {
+      /*
+        We must not free a GCO before the wait_count of the following GCO has
+        been reached and wakeup has been sent. Otherwise we will lose the
+        wakeup and hang (there were several such bugs in the past).
+
+        The intention is that this is ensured already since we only free when
+        the last event group in the GCO has committed
+        (tmp_gco->last_sub_id <= sub_id). However, if we have a bug, we have
+        extra check on next_gco->wait_count to hopefully avoid hanging; we
+        have here an assertion in debug builds that this check does not in
+        fact trigger.
+      */
+      DBUG_ASSERT(!tmp_gco->next_gco || tmp_gco->last_sub_id > sub_id);
+      tmp_gco= tmp_gco->prev_gco;
+    }
+    while (tmp_gco)
+    {
+      group_commit_orderer *prev_gco= tmp_gco->prev_gco;
+      tmp_gco->next_gco->prev_gco= NULL;
+      rpt->loc_free_gco(tmp_gco);
+      tmp_gco= prev_gco;
+    }
+  }
+
+  /*
+    If this event group got error, then any following event groups that have
+    not yet started should just skip their group, preparing for stop of the
+    SQL driver thread.
+  */
+  if (unlikely(rgi->worker_error) && entry->stop_on_error_sub_id > sub_id)
+    entry->stop_on_error_sub_id= sub_id;
+  mysql_mutex_unlock(&entry->LOCK_parallel_entry);
+#ifdef ENABLED_DEBUG_SYNC
+  DBUG_EXECUTE_IF("hold_worker_on_schedule", {
+      if (entry->stop_on_error_sub_id < (uint64)ULONGLONG_MAX)
+      {
+        debug_sync_set_action(thd, STRING_WITH_LEN("now SIGNAL continue_worker"));
+      }
+    });
+
+  DBUG_EXECUTE_IF("rpl_parallel_simulate_wait_at_retry", {
+      if (rgi->current_gtid.seq_no == 1000) {
+        DBUG_ASSERT(entry->stop_on_error_sub_id == sub_id);
+        debug_sync_set_action(thd,
+                              STRING_WITH_LEN("now WAIT_FOR proceed_by_1000"));
+      }
+    });
+  DBUG_EXECUTE_IF("hold_worker2_favor_worker3", {
+      if (rgi->current_gtid.seq_no == 2001) {
+        DBUG_ASSERT(!rgi->worker_error || entry->stop_on_error_sub_id == sub_id);
+        debug_sync_set_action(thd, STRING_WITH_LEN("now SIGNAL cont_worker3"));
+      }
+    });
+#endif
+
+  if (rgi->killed_for_retry == rpl_group_info::RETRY_KILL_PENDING)
+    wait_for_pending_deadlock_kill(thd, rgi);
+  thd->clear_error();
+  thd->reset_killed();
+  /*
+    Would do thd->get_stmt_da()->set_overwrite_status(false) here, but
+    reset_diagnostics_area() already does that.
+  */
+  thd->get_stmt_da()->reset_diagnostics_area();
+  wfc->wakeup_subsequent_commits(rgi->worker_error);
+  rgi->did_mark_start_commit= false;
+  rgi->set_finish_event_group_called(true);
+}
+
+
+static void
+signal_error_to_sql_driver_thread(THD *thd, rpl_group_info *rgi, int err)
+{
+  rgi->worker_error= err;
+  DBUG_EXECUTE_IF("hold_worker2_favor_worker3", {
+      if (rgi->current_gtid.seq_no == 2002) {
+        debug_sync_set_action(thd, STRING_WITH_LEN("now WAIT_FOR cont_worker2"));
+      }});
+
+  rgi->cleanup_context(thd, true);
+  rgi->rli->abort_slave= true;
+  rgi->rli->stop_for_until= false;
+  mysql_mutex_lock(rgi->rli->relay_log.get_log_lock());
+  rgi->rli->relay_log.signal_relay_log_update();
+  mysql_mutex_unlock(rgi->rli->relay_log.get_log_lock());
+}
+
+
+static void
+unlock_or_exit_cond(THD *thd, mysql_mutex_t *lock, bool *did_enter_cond,
+                    PSI_stage_info *old_stage)
+{
+  if (*did_enter_cond)
+  {
+    thd->EXIT_COND(old_stage);
+    *did_enter_cond= false;
+  }
+  else
+    mysql_mutex_unlock(lock);
+}
+
+
+static void
+register_wait_for_prior_event_group_commit(rpl_group_info *rgi,
+                                           rpl_parallel_entry *entry)
+{
+  mysql_mutex_assert_owner(&entry->LOCK_parallel_entry);
+  if (rgi->wait_commit_sub_id > entry->last_committed_sub_id)
+  {
+    /*
+      Register that the commit of this event group must wait for the
+      commit of the previous event group to complete before it may
+      complete itself, so that we preserve commit order.
+    */
+    wait_for_commit *waitee=
+      &rgi->wait_commit_group_info->commit_orderer;
+    rgi->commit_orderer.register_wait_for_prior_commit(waitee);
+  }
+}
+
+
+/*
+  Do not start parallel execution of this event group until all prior groups
+  have reached the commit phase that are not safe to run in parallel with.
+*/
+static void
+do_gco_wait(rpl_group_info *rgi, group_commit_orderer *gco,
+            bool *did_enter_cond, PSI_stage_info *old_stage)
+{
+  THD *thd= rgi->thd;
+  rpl_parallel_entry *entry= rgi->parallel_entry;
+  uint64 wait_count;
+
+  mysql_mutex_assert_owner(&entry->LOCK_parallel_entry);
+
+  if (!gco->installed)
+  {
+    group_commit_orderer *prev_gco= gco->prev_gco;
+    if (prev_gco)
+    {
+      prev_gco->last_sub_id= gco->prior_sub_id;
+      prev_gco->next_gco= gco;
+    }
+    gco->installed= true;
+  }
+  wait_count= gco->wait_count;
+  if (wait_count > entry->count_committing_event_groups)
+  {
+    DEBUG_SYNC(thd, "rpl_parallel_start_waiting_for_prior");
+    thd->ENTER_COND(&gco->COND_group_commit_orderer,
+                    &entry->LOCK_parallel_entry,
+                    &stage_waiting_for_prior_transaction_to_start_commit,
+                    old_stage);
+    *did_enter_cond= true;
+    thd->set_time_for_next_stage();
+    do
+    {
+      if (!rgi->worker_error && unlikely(thd->check_killed(1)))
+      {
+        DEBUG_SYNC(thd, "rpl_parallel_start_waiting_for_prior_killed");
+        thd->clear_error();
+        thd->get_stmt_da()->reset_diagnostics_area();
+        thd->send_kill_message();
+        slave_output_error_info(rgi, thd);
+        signal_error_to_sql_driver_thread(thd, rgi, 1);
+        /*
+          Even though we were killed, we need to continue waiting for the
+          prior event groups to signal that we can continue. Otherwise we
+          mess up the accounting for ordering. However, now that we have
+          marked the error, events will just be skipped rather than
+          executed, and things will progress quickly towards stop.
+        */
+      }
+      mysql_cond_wait(&gco->COND_group_commit_orderer,
+                      &entry->LOCK_parallel_entry);
+    } while (wait_count > entry->count_committing_event_groups);
+  }
+}
+
+
+static bool
+do_stop_handling(rpl_group_info *rgi)
+{
+  bool should_stop= false;
+  rpl_parallel_entry *entry=  rgi->parallel_entry;
+
+  mysql_mutex_assert_owner(&entry->LOCK_parallel_entry);
+
+  if (unlikely(entry->force_abort) && rgi->gtid_sub_id > entry->stop_sub_id)
+  {
+    /*
+      We are stopping (STOP SLAVE), and this event group need not be applied
+      before we can safely stop. So return a flag that will cause us to skip,
+      rather than execute, the following events. Once all queued events have
+      been skipped, the STOP SLAVE is complete (for this thread).
+    */
+    should_stop= true;
+  }
+
+  if (unlikely(entry->stop_on_error_sub_id <= rgi->wait_commit_sub_id))
+  {
+    rgi->worker_error= 1;
+    should_stop= true;
+  }
+
+  if (likely(!should_stop))
+  {
+    /*
+      Since we did not decide to stop, bump the largest_started_sub_id while
+      still holding LOCK_parallel_entry.
+    */
+    if (rgi->gtid_sub_id > entry->largest_started_sub_id)
+      entry->largest_started_sub_id= rgi->gtid_sub_id;
+  }
+
+  return should_stop;
+}
+
+
+static bool
+do_ftwrl_wait(rpl_group_info *rgi,
+              bool *did_enter_cond, PSI_stage_info *old_stage)
+{
+  THD *thd= rgi->thd;
+  rpl_parallel_entry *entry= rgi->parallel_entry;
+  uint64 sub_id= rgi->gtid_sub_id;
+  bool aborted= false;
+  DBUG_ENTER("do_ftwrl_wait");
+
+  mysql_mutex_assert_owner(&entry->LOCK_parallel_entry);
+
+  /*
+    If a FLUSH TABLES WITH READ LOCK (FTWRL) is pending, check if this
+    transaction is later than transactions that have priority to complete
+    before FTWRL. If so, wait here so that FTWRL can proceed and complete
+    first.
+
+    (entry->pause_sub_id is ULONGLONG_MAX if no FTWRL is pending, which makes
+    this test false as required).
+  */
+  if (unlikely(sub_id > entry->pause_sub_id))
+  {
+    thd->ENTER_COND(&entry->COND_parallel_entry, &entry->LOCK_parallel_entry,
+                    &stage_waiting_for_ftwrl, old_stage);
+    *did_enter_cond= true;
+    thd->set_time_for_next_stage();
+    do
+    {
+      if (entry->force_abort || rgi->worker_error)
+      {
+        aborted= true;
+        break;
+      }
+      if (unlikely(thd->check_killed()))
+      {
+        slave_output_error_info(rgi, thd);
+        signal_error_to_sql_driver_thread(thd, rgi, 1);
+        break;
+      }
+      mysql_cond_wait(&entry->COND_parallel_entry, &entry->LOCK_parallel_entry);
+    } while (sub_id > entry->pause_sub_id);
+
+    DBUG_EXECUTE_IF("delay_ftwrl_wait_gtid_0_x_100", {
+        if (rgi->current_gtid.domain_id == 0 &&
+            rgi->current_gtid.seq_no == 100) {
+          /*
+            Simulate delayed wakeup from the mysql_cond_wait(). To do this, we
+            need to have the LOCK_parallel_entry mutex released during the wait.
+          */
+          mysql_mutex_unlock(&entry->LOCK_parallel_entry);
+          debug_sync_set_action(thd,
+              STRING_WITH_LEN("now SIGNAL pause_wait_started WAIT_FOR pause_wait_continue"));
+          mysql_mutex_lock(&entry->LOCK_parallel_entry);
+        }
+      });
+    /*
+      We do not call EXIT_COND() here, as this will be done later by our
+      caller (since we set *did_enter_cond to true).
+    */
+  }
+
+  DBUG_RETURN(aborted);
+}
+
+
+static int
+pool_mark_busy(rpl_parallel_thread_pool *pool, THD *thd)
+{
+  PSI_stage_info old_stage;
+  int res= 0;
+
+  /*
+    Wait here while the queue is busy. This is done to make FLUSH TABLES WITH
+    READ LOCK work correctly, without incuring extra locking penalties in
+    normal operation. FLUSH TABLES WITH READ LOCK needs to lock threads in the
+    thread pool, and for this we need to make sure the pool will not go away
+    during the operation. The LOCK_rpl_thread_pool is not suitable for
+    this. It is taken by release_thread() while holding LOCK_rpl_thread; so it
+    must be released before locking any LOCK_rpl_thread lock, or a deadlock
+    can occur.
+
+    So we protect the infrequent operations of FLUSH TABLES WITH READ LOCK and
+    pool size changes with this condition wait.
+  */
+  DBUG_EXECUTE_IF("mark_busy_mdev_22370",my_sleep(1000000););
+  mysql_mutex_lock(&pool->LOCK_rpl_thread_pool);
+  if (thd)
+  {
+    thd->ENTER_COND(&pool->COND_rpl_thread_pool, &pool->LOCK_rpl_thread_pool,
+                    &stage_waiting_for_rpl_thread_pool, &old_stage);
+    thd->set_time_for_next_stage();
+  }
+  while (pool->busy)
+  {
+    if (thd && unlikely(thd->check_killed()))
+    {
+      res= 1;
+      break;
+    }
+    mysql_cond_wait(&pool->COND_rpl_thread_pool, &pool->LOCK_rpl_thread_pool);
+  }
+  if (!res)
+    pool->busy= true;
+  if (thd)
+    thd->EXIT_COND(&old_stage);
+  else
+    mysql_mutex_unlock(&pool->LOCK_rpl_thread_pool);
+
+  return res;
+}
+
+
+static void
+pool_mark_not_busy(rpl_parallel_thread_pool *pool)
+{
+  mysql_mutex_lock(&pool->LOCK_rpl_thread_pool);
+  DBUG_ASSERT(pool->busy);
+  pool->busy= false;
+  mysql_cond_broadcast(&pool->COND_rpl_thread_pool);
+  mysql_mutex_unlock(&pool->LOCK_rpl_thread_pool);
+}
+
+
+void
+rpl_unpause_after_ftwrl(THD *thd)
+{
+  uint32 i;
+  rpl_parallel_thread_pool *pool= &global_rpl_thread_pool;
+  DBUG_ENTER("rpl_unpause_after_ftwrl");
+
+  DBUG_ASSERT(pool->busy);
+
+  for (i= 0; i < pool->count; ++i)
+  {
+    rpl_parallel_entry *e;
+    rpl_parallel_thread *rpt= pool->threads[i];
+
+    mysql_mutex_lock(&rpt->LOCK_rpl_thread);
+    if (!rpt->current_owner)
+    {
+      mysql_mutex_unlock(&rpt->LOCK_rpl_thread);
+      continue;
+    }
+    e= rpt->current_entry;
+    mysql_mutex_lock(&e->LOCK_parallel_entry);
+    rpt->pause_for_ftwrl = false;
+    mysql_mutex_unlock(&rpt->LOCK_rpl_thread);
+    /*
+      Do not change pause_sub_id if force_abort is set.
+      force_abort is set in case of STOP SLAVE.
+
+      Reason: If pause_sub_id is not changed and force_abort_is set,
+      any parallel slave thread waiting in do_ftwrl_wait() will
+      on wakeup return from do_ftwrl_wait() with 1. This will set
+      skip_event_group to 1 in handle_rpl_parallel_thread() and the
+      parallel thread will abort at once.
+
+      If pause_sub_id is changed, the code in handle_rpl_parallel_thread()
+      would continue to execute the transaction in the queue, which would
+      cause some transactions to be lost.
+    */
+    if (!e->force_abort)
+      e->pause_sub_id= (uint64)ULONGLONG_MAX;
+    mysql_cond_broadcast(&e->COND_parallel_entry);
+    mysql_mutex_unlock(&e->LOCK_parallel_entry);
+  }
+
+  pool_mark_not_busy(pool);
+  DBUG_VOID_RETURN;
+}
+
+
+/*
+  .
+
+  Note: in case of error return, rpl_unpause_after_ftwrl() must _not_ be called.
+*/
+int
+rpl_pause_for_ftwrl(THD *thd)
+{
+  uint32 i;
+  rpl_parallel_thread_pool *pool= &global_rpl_thread_pool;
+  int err;
+  Dynamic_array<Master_info*> mi_arr(4, 4); // array of replication source mi:s
+  DBUG_ENTER("rpl_pause_for_ftwrl");
+
+  /*
+    While the count_pending_pause_for_ftwrl counter is non-zero, the pool
+    cannot be shutdown/resized, so threads are guaranteed to not disappear.
+
+    This is required to safely be able to access the individual threads below.
+    (We cannot lock an individual thread while holding LOCK_rpl_thread_pool,
+    as this can deadlock against release_thread()).
+  */
+  if ((err= pool_mark_busy(pool, thd)))
+    DBUG_RETURN(err);
+
+  for (i= 0; i < pool->count; ++i)
+  {
+    PSI_stage_info old_stage;
+    rpl_parallel_entry *e;
+    rpl_parallel_thread *rpt= pool->threads[i];
+
+    mysql_mutex_lock(&rpt->LOCK_rpl_thread);
+    if (!rpt->current_owner)
+    {
+      mysql_mutex_unlock(&rpt->LOCK_rpl_thread);
+      continue;
+    }
+    e= rpt->current_entry;
+    mysql_mutex_lock(&e->LOCK_parallel_entry);
+    /*
+      Setting the rpt->pause_for_ftwrl flag makes sure that the thread will not
+      de-allocate itself until signalled to do so by rpl_unpause_after_ftwrl().
+    */
+    rpt->pause_for_ftwrl = true;
+    mysql_mutex_unlock(&rpt->LOCK_rpl_thread);
+    ++e->need_sub_id_signal;
+    if (e->pause_sub_id == (uint64)ULONGLONG_MAX)
+    {
+      e->pause_sub_id= e->largest_started_sub_id;
+      DBUG_EXECUTE_IF("pause_for_ftwrl_wait", {
+          mysql_mutex_unlock(&e->LOCK_parallel_entry);
+          debug_sync_set_action(thd,
+                                STRING_WITH_LEN("now "
+                                                "SIGNAL pause_ftwrl_waiting "
+                                                "WAIT_FOR pause_ftwrl_cont"));
+          mysql_mutex_lock(&e->LOCK_parallel_entry);
+        });
+    }
+    thd->ENTER_COND(&e->COND_parallel_entry, &e->LOCK_parallel_entry,
+                    &stage_waiting_for_ftwrl_threads_to_pause, &old_stage);
+    thd->set_time_for_next_stage();
+    while (e->pause_sub_id < (uint64)ULONGLONG_MAX &&
+           e->last_committed_sub_id < e->pause_sub_id &&
+           !err)
+    {
+      if (unlikely(thd->check_killed()))
+      {
+        err= 1;
+        break;
+      }
+      mysql_cond_wait(&e->COND_parallel_entry, &e->LOCK_parallel_entry);
+    };
+    --e->need_sub_id_signal;
+
+    thd->EXIT_COND(&old_stage);
+    if (err)
+      break;
+    /*
+      Notify any source any domain waiting-for-master Start-Alter to give way.
+    */
+    Master_info *mi= e->rli->mi;
+    bool found= false;
+    for (uint i= 0; i < mi_arr.elements() && !found; i++)
+      found= mi_arr.at(i) == mi;
+    if (!found)
+    {
+      mi_arr.append(mi);
+      start_alter_info *info=NULL;
+      mysql_mutex_lock(&mi->start_alter_list_lock);
+      List_iterator<start_alter_info> info_iterator(mi->start_alter_list);
+      while ((info= info_iterator++))
+      {
+        mysql_mutex_lock(&mi->start_alter_lock);
+
+        DBUG_ASSERT(info->state == start_alter_state::REGISTERED);
+
+        info->state= start_alter_state::ROLLBACK_ALTER;
+        info->direct_commit_alter= true;
+        mysql_cond_broadcast(&info->start_alter_cond);
+        mysql_mutex_unlock(&mi->start_alter_lock);
+      }
+      mysql_mutex_unlock(&mi->start_alter_list_lock);
+    }
+  }
+
+  if (err)
+    rpl_unpause_after_ftwrl(thd);
+  DBUG_RETURN(err);
+}
+
+
+#ifndef DBUG_OFF
+static int
+dbug_simulate_tmp_error(rpl_group_info *rgi, THD *thd)
+{
+  if (rgi->current_gtid.domain_id == 0 && rgi->current_gtid.seq_no == 100 &&
+      rgi->retry_event_count == 4)
+  {
+    thd->clear_error();
+    thd->get_stmt_da()->reset_diagnostics_area();
+    my_error(ER_LOCK_DEADLOCK, MYF(0));
+    return 1;
+  }
+  return 0;
+}
+#endif
+
+
+/*
+  If we detect a deadlock due to eg. storage engine locks that conflict with
+  the fixed commit order, then the later transaction will be killed
+  asynchroneously to allow the former to complete its commit.
+
+  In this case, we convert the 'killed' error into a deadlock error, and retry
+  the later transaction.
+
+  If we are doing optimistic parallel apply of transactions not known to be
+  safe, we convert any error to a deadlock error, but then at retry we will
+  wait for prior transactions to commit first, so that the retries can be
+  done non-speculative.
+*/
+static void
+convert_kill_to_deadlock_error(rpl_group_info *rgi)
+{
+  THD *thd= rgi->thd;
+  int err_code;
+
+  if (!thd->get_stmt_da()->is_error())
+    return;
+  err_code= thd->get_stmt_da()->sql_errno();
+  if ((rgi->speculation == rpl_group_info::SPECULATE_OPTIMISTIC &&
+       err_code != ER_PRIOR_COMMIT_FAILED) ||
+      ((err_code == ER_QUERY_INTERRUPTED || err_code == ER_CONNECTION_KILLED) &&
+       rgi->killed_for_retry))
+  {
+    thd->clear_error();
+    my_error(ER_LOCK_DEADLOCK, MYF(0));
+    thd->reset_killed();
+  }
+}
+
+
+/*
+  Check if an event marks the end of an event group. Returns non-zero if so,
+  zero otherwise.
+
+  In addition, returns 1 if the group is committing, 2 if it is rolling back.
+*/
+static int
+is_group_ending(Log_event *ev, Log_event_type event_type)
+{
+  if (event_type == XID_EVENT || event_type == XA_PREPARE_LOG_EVENT)
+    return 1;
+  if (event_type == QUERY_EVENT)  // COMMIT/ROLLBACK are never compressed
+  {
+    Query_log_event *qev = (Query_log_event *)ev;
+    if (qev->is_commit() ||
+        !strncmp(qev->query, STRING_WITH_LEN("XA COMMIT")) ||
+        !strncmp(qev->query, STRING_WITH_LEN("XA ROLLBACK")))
+      return 1;
+    if (qev->is_rollback())
+      return 2;
+  }
+  return 0;
+}
+
+
+static int
+retry_event_group(rpl_group_info *rgi, rpl_parallel_thread *rpt,
+                  rpl_parallel_thread::queued_event *orig_qev)
+{
+  IO_CACHE rlog;
+  LOG_INFO linfo;
+  File fd= (File)-1;
+  const char *errmsg;
+  inuse_relaylog *ir= rgi->relay_log;
+  uint64 event_count;
+  uint64 events_to_execute= rgi->retry_event_count;
+  Relay_log_info *rli= rgi->rli;
+  int err;
+  ulonglong cur_offset, old_offset;
+  char log_name[FN_REFLEN];
+  THD *thd= rgi->thd;
+  rpl_parallel_entry *entry= rgi->parallel_entry;
+  ulong retries= 0;
+  Format_description_log_event *description_event= NULL;
+
+do_retry:
+  event_count= 0;
+  err= 0;
+  errmsg= NULL;
+#ifdef WITH_WSREP
+  thd->wsrep_cs().reset_error();
+  WSREP_DEBUG("retrying async replication event");
+#endif /* WITH_WSREP */
+
+  /*
+    If we already started committing before getting the deadlock (or other
+    error) that caused us to need to retry, we have already signalled
+    subsequent transactions that we have started committing. This is
+    potentially a problem, as now we will rollback, and if subsequent
+    transactions would start to execute now, they could see an unexpected
+    state of the database and get eg. key not found or duplicate key error.
+
+    However, to get a deadlock in the first place, there must have been
+    another earlier transaction that is waiting for us. Thus that other
+    transaction has _not_ yet started to commit, and any subsequent
+    transactions will still be waiting at this point.
+
+    So here, we decrement back the count of transactions that started
+    committing (if we already incremented it), undoing the effect of an
+    earlier mark_start_commit(). Then later, when the retry succeeds and we
+    commit again, we can do a new mark_start_commit() and eventually wake up
+    subsequent transactions at the proper time.
+
+    We need to do the unmark before the rollback, to be sure that the
+    transaction we deadlocked with will not signal that it started to commit
+    until after the unmark.
+  */
+  DBUG_EXECUTE_IF("inject_mdev8302", { my_sleep(20000);});
+  rgi->unmark_start_commit();
+  DEBUG_SYNC(thd, "rpl_parallel_retry_after_unmark");
+
+  /*
+    We might get the deadlock error that causes the retry during commit, while
+    sitting in wait_for_prior_commit(). If this happens, we will have a
+    pending error in the wait_for_commit object. So clear this by
+    unregistering (and later re-registering) the wait.
+  */
+  if(thd->wait_for_commit_ptr)
+    thd->wait_for_commit_ptr->unregister_wait_for_prior_commit();
+  DBUG_EXECUTE_IF("inject_mdev8031", {
+      /* Simulate that we get deadlock killed at this exact point. */
+      rgi->killed_for_retry= rpl_group_info::RETRY_KILL_KILLED;
+      thd->set_killed(KILL_CONNECTION);
+  });
+#ifdef ENABLED_DEBUG_SYNC
+  DBUG_EXECUTE_IF("rpl_parallel_simulate_wait_at_retry", {
+      if (rgi->current_gtid.seq_no == 1001) {
+        debug_sync_set_action(thd,
+                              STRING_WITH_LEN("rpl_parallel_simulate_wait_at_retry WAIT_FOR proceed_by_1001"));
+      }
+      DEBUG_SYNC(thd, "rpl_parallel_simulate_wait_at_retry");
+    });
+#endif
+
+  rgi->cleanup_context(thd, 1);
+  wait_for_pending_deadlock_kill(thd, rgi);
+  thd->reset_killed();
+  thd->clear_error();
+  rgi->killed_for_retry = rpl_group_info::RETRY_KILL_NONE;
+#ifdef ENABLED_DEBUG_SYNC
+    DBUG_EXECUTE_IF("hold_worker2_favor_worker3", {
+      if (rgi->current_gtid.seq_no == 2003) {
+        debug_sync_set_action(thd,
+                              STRING_WITH_LEN("now WAIT_FOR cont_worker3"));
+      }
+    });
+#endif
+  /*
+    If we retry due to a deadlock kill that occurred during the commit step, we
+    might have already updated (but not committed) an update of table
+    mysql.gtid_slave_pos, and cleared the gtid_pending flag. Now we have
+    rolled back any such update, so we must set the gtid_pending flag back to
+    true so that we will do a new update when/if we succeed with the retry.
+  */
+  rgi->gtid_pending= true;
+
+  mysql_mutex_lock(&rli->data_lock);
+  ++rli->retried_trans;
+  ++rpt->last_trans_retry_count;
+  statistic_increment(slave_retried_transactions, LOCK_status);
+  mysql_mutex_unlock(&rli->data_lock);
+
+  for (;;)
+  {
+    mysql_mutex_lock(&entry->LOCK_parallel_entry);
+    if (rgi->gtid_sub_id < entry->stop_on_error_sub_id ||
+        DBUG_IF("simulate_mdev_12746"))
+    {
+      register_wait_for_prior_event_group_commit(rgi, entry);
+    }
+    else
+    {
+      /*
+        A failure of a preceding "parent" transaction may not be
+        seen by the current one through its own worker_error.
+        Such induced error gets set by ourselves now.
+      */
+      err= rgi->worker_error= 1;
+      my_error(ER_PRIOR_COMMIT_FAILED, MYF(0));
+      mysql_mutex_unlock(&entry->LOCK_parallel_entry);
+
+      goto err;
+    }
+    mysql_mutex_unlock(&entry->LOCK_parallel_entry);
+
+    /*
+      Let us wait for all prior transactions to complete before trying again.
+      This way, we avoid repeatedly conflicting with and getting deadlock
+      killed by the same earlier transaction.
+    */
+    if (!(err= thd->wait_for_prior_commit()))
+    {
+      rgi->speculation = rpl_group_info::SPECULATE_WAIT;
+      break;
+    }
+
+    convert_kill_to_deadlock_error(rgi);
+    if (!has_temporary_error(thd))
+      goto err;
+    /*
+      If we get a temporary error such as a deadlock kill, we can safely
+      ignore it, as we already rolled back.
+
+      But we still want to retry the wait for the prior transaction to
+      complete its commit.
+    */
+    thd->clear_error();
+    thd->reset_killed();
+    if(thd->wait_for_commit_ptr)
+      thd->wait_for_commit_ptr->unregister_wait_for_prior_commit();
+    DBUG_EXECUTE_IF("inject_mdev8031", {
+        /* Inject a small sleep to give prior transaction a chance to commit. */
+        my_sleep(100000);
+    });
+  }
+
+  /*
+    Let us clear any lingering deadlock kill one more time, here after
+    wait_for_prior_commit() has completed. This should rule out any
+    possibility of an old deadlock kill lingering on beyond this point.
+  */
+  thd->reset_killed();
+#ifdef WITH_WSREP
+  if (wsrep_before_command(thd))
+  {
+    WSREP_WARN("Parallel slave worker failed at wsrep_before_command() hook");
+    err= 1;
+    goto err;
+  }
+  wsrep_start_trx_if_not_started(thd);
+  WSREP_DEBUG("parallel slave retry, after trx start");
+
+#endif /* WITH_WSREP */
+  strmake_buf(log_name, ir->name);
+  if ((fd= open_binlog(&rlog, log_name, &errmsg)) <0)
+  {
+    err= 1;
+    goto err;
+  }
+  cur_offset= rgi->retry_start_offset;
+  delete description_event;
+  description_event=
+    read_relay_log_description_event(&rlog, cur_offset, &errmsg);
+  if (!description_event)
+  {
+    err= 1;
+    goto err;
+  }
+  DBUG_EXECUTE_IF("inject_mdev8031", {
+      /* Simulate pending KILL caught in read_relay_log_description_event(). */
+      if (unlikely(thd->check_killed()))
+      {
+        err= 1;
+        goto err;
+      }
+  });
+  my_b_seek(&rlog, cur_offset);
+
+  do
+  {
+    Log_event_type event_type;
+    Log_event *ev;
+    rpl_parallel_thread::queued_event *qev;
+
+    /* The loop is here so we can try again the next relay log file on EOF. */
+    for (;;)
+    {
+      old_offset= cur_offset;
+      ev= Log_event::read_log_event(&rlog, description_event,
+                                    opt_slave_sql_verify_checksum);
+      cur_offset= my_b_tell(&rlog);
+
+      if (ev)
+        break;
+      if (unlikely(rlog.error < 0))
+      {
+        errmsg= "slave SQL thread aborted because of I/O error";
+        err= 1;
+        goto check_retry;
+      }
+      if (unlikely(rlog.error > 0))
+      {
+        sql_print_error("Slave SQL thread: I/O error reading "
+                        "event(errno: %d  cur_log->error: %d)",
+                        my_errno, rlog.error);
+        errmsg= "Aborting slave SQL thread because of partial event read";
+        err= 1;
+        goto err;
+      }
+      /* EOF. Move to the next relay log. */
+      end_io_cache(&rlog);
+      mysql_file_close(fd, MYF(MY_WME));
+      fd= (File)-1;
+
+      /* Find the next relay log file. */
+      if((err= rli->relay_log.find_log_pos(&linfo, log_name, 1)) ||
+         (err= rli->relay_log.find_next_log(&linfo, 1)))
+      {
+        char buff[22];
+        sql_print_error("next log error: %d  offset: %s  log: %s",
+                        err,
+                        llstr(linfo.index_file_offset, buff),
+                        log_name);
+        goto err;
+      }
+      strmake_buf(log_name ,linfo.log_file_name);
+
+      DBUG_EXECUTE_IF("inject_retry_event_group_open_binlog_kill", {
+          if (retries < 2)
+          {
+            /* Simulate that we get deadlock killed during open_binlog(). */
+            thd->reset_for_next_command();
+            rgi->killed_for_retry= rpl_group_info::RETRY_KILL_KILLED;
+            thd->set_killed(KILL_CONNECTION);
+            thd->send_kill_message();
+            fd= (File)-1;
+            err= 1;
+            goto check_retry;
+          }
+      });
+      if ((fd= open_binlog(&rlog, log_name, &errmsg)) <0)
+      {
+        err= 1;
+        goto check_retry;
+      }
+      description_event->reset_crypto();
+      /* Loop to try again on the new log file. */
+    }
+
+    event_type= ev->get_type_code();
+    if (event_type == FORMAT_DESCRIPTION_EVENT)
+    {
+      Format_description_log_event *newde= (Format_description_log_event*)ev;
+      newde->copy_crypto_data(description_event);
+      delete description_event;
+      description_event= newde;
+      continue;
+    }
+    else if (event_type == START_ENCRYPTION_EVENT)
+    {
+      description_event->start_decryption((Start_encryption_log_event*)ev);
+      delete ev;
+      continue;
+    }
+    else if (!Log_event::is_group_event(event_type))
+    {
+      delete ev;
+      continue;
+    }
+    ev->thd= thd;
+
+    mysql_mutex_lock(&rpt->LOCK_rpl_thread);
+    qev= rpt->retry_get_qev(ev, orig_qev, log_name, old_offset,
+                            cur_offset - old_offset);
+    mysql_mutex_unlock(&rpt->LOCK_rpl_thread);
+    if (!qev)
+    {
+      delete ev;
+      my_error(ER_OUT_OF_RESOURCES, MYF(0));
+      err= 1;
+      goto err;
+    }
+    if (is_group_ending(ev, event_type) == 1)
+      rgi->mark_start_commit();
+
+    err= rpt_handle_event(qev, rpt);
+    ++event_count;
+    mysql_mutex_lock(&rpt->LOCK_rpl_thread);
+    rpt->free_qev(qev);
+    mysql_mutex_unlock(&rpt->LOCK_rpl_thread);
+
+    delete_or_keep_event_post_apply(rgi, event_type, ev);
+    DBUG_EXECUTE_IF("rpl_parallel_simulate_double_temp_err_gtid_0_x_100",
+                    if (retries == 0) err= dbug_simulate_tmp_error(rgi, thd););
+    DBUG_EXECUTE_IF("rpl_parallel_simulate_infinite_temp_err_gtid_0_x_100",
+                    err= dbug_simulate_tmp_error(rgi, thd););
+    if (!err)
+      continue;
+
+check_retry:
+    convert_kill_to_deadlock_error(rgi);
+    if (has_temporary_error(thd))
+    {
+      ++retries;
+      if (retries < slave_trans_retries)
+      {
+        if (fd >= 0)
+        {
+          end_io_cache(&rlog);
+          mysql_file_close(fd, MYF(MY_WME));
+          fd= (File)-1;
+        }
+        goto do_retry;
+      }
+      sql_print_error("Slave worker thread retried transaction %lu time(s) "
+                      "in vain, giving up. Consider raising the value of "
+                      "the slave_transaction_retries variable.",
+                      slave_trans_retries);
+    }
+    goto err;
+
+  } while (event_count < events_to_execute);
+
+err:
+
+  if (description_event)
+    delete description_event;
+  if (fd >= 0)
+  {
+    end_io_cache(&rlog);
+    mysql_file_close(fd, MYF(MY_WME));
+  }
+  if (errmsg)
+    sql_print_error("Error reading relay log event: %s", errmsg);
+  return err;
+}
+
+
+pthread_handler_t
+handle_rpl_parallel_thread(void *arg)
+{
+  THD *thd;
+  PSI_stage_info old_stage;
+  struct rpl_parallel_thread::queued_event *events;
+  bool group_standalone= true;
+  bool in_event_group= false;
+  bool skip_event_group= false;
+  rpl_group_info *group_rgi= NULL;
+  group_commit_orderer *gco;
+  uint64 event_gtid_sub_id= 0;
+  rpl_sql_thread_info sql_info(NULL);
+  int err;
+
+  struct rpl_parallel_thread *rpt= (struct rpl_parallel_thread *)arg;
+
+  my_thread_init();
+  thd = new THD(next_thread_id());
+  thd->thread_stack = (char*)&thd;
+  server_threads.insert(thd);
+  set_current_thd(thd);
+  pthread_detach_this_thread();
+  thd->store_globals();
+  thd->init_for_queries();
+  thd->variables.binlog_annotate_row_events= 0;
+  init_thr_lock();
+  thd->system_thread= SYSTEM_THREAD_SLAVE_SQL;
+  thd->security_ctx->skip_grants();
+  thd->variables.max_allowed_packet= slave_max_allowed_packet;
+  /* Ensure that slave can exeute any alter table it gets from master */
+  thd->variables.alter_algorithm= (ulong) Alter_info::ALTER_TABLE_ALGORITHM_DEFAULT;
+  thd->slave_thread= 1;
+
+  set_slave_thread_options(thd);
+  thd->client_capabilities = CLIENT_LOCAL_FILES;
+  thd->net.reading_or_writing= 0;
+  thd_proc_info(thd, "Waiting for work from main SQL threads");
+  thd->variables.lock_wait_timeout= LONG_TIMEOUT;
+  thd->system_thread_info.rpl_sql_info= &sql_info;
+  /*
+    We need to use (at least) REPEATABLE READ isolation level. Otherwise
+    speculative parallel apply can run out-of-order and give wrong results
+    for statement-based replication.
+  */
+  thd->variables.tx_isolation= ISO_REPEATABLE_READ;
+
+  mysql_mutex_lock(&rpt->LOCK_rpl_thread);
+  rpt->thd= thd;
+  PSI_thread *psi= PSI_CALL_get_thread();
+  PSI_CALL_set_thread_os_id(psi);
+  PSI_CALL_set_thread_THD(psi, thd);
+  PSI_CALL_set_thread_id(psi, thd->thread_id);
+  rpt->thd->set_psi(psi);
+
+  while (rpt->delay_start)
+    mysql_cond_wait(&rpt->COND_rpl_thread, &rpt->LOCK_rpl_thread);
+
+  rpt->running= true;
+  mysql_cond_signal(&rpt->COND_rpl_thread);
+
+  thd->set_command(COM_SLAVE_WORKER);
+#ifdef WITH_WSREP
+  wsrep_open(thd);
+  if (wsrep_before_command(thd))
+  {
+    WSREP_WARN("Parallel slave failed at wsrep_before_command() hook");
+    rpt->stop = true;
+  }
+#endif /* WITH_WSREP */
+  while (!rpt->stop)
+  {
+    uint wait_count= 0;
+    rpl_parallel_thread::queued_event *qev, *next_qev;
+
+    rpt->start_time_tracker();
+    thd->ENTER_COND(&rpt->COND_rpl_thread, &rpt->LOCK_rpl_thread,
+                    &stage_waiting_for_work_from_sql_thread, &old_stage);
+    /*
+      There are 4 cases that should cause us to wake up:
+       - Events have been queued for us to handle.
+       - We have an owner, but no events and not inside event group -> we need
+         to release ourself to the thread pool
+       - SQL thread is stopping, and we have an owner but no events, and we are
+         inside an event group; no more events will be queued to us, so we need
+         to abort the group (force_abort==1).
+       - Thread pool shutdown (rpt->stop==1).
+    */
+    while (!( (events= rpt->event_queue) ||
+              (rpt->current_owner && !in_event_group) ||
+              (rpt->current_owner && group_rgi->parallel_entry->force_abort) ||
+              rpt->stop))
+    {
+      if (!wait_count++)
+        thd->set_time_for_next_stage();
+      mysql_cond_wait(&rpt->COND_rpl_thread, &rpt->LOCK_rpl_thread);
+    }
+    rpt->dequeue1(events);
+    thd->EXIT_COND(&old_stage);
+    rpt->add_to_worker_idle_time_and_reset();
+
+  more_events:
+    for (qev= events; qev; qev= next_qev)
+    {
+      Log_event_type event_type;
+      rpl_group_info *rgi= qev->rgi;
+      rpl_parallel_entry *entry= rgi->parallel_entry;
+      bool end_of_group;
+      int group_ending;
+
+      next_qev= qev->next;
+      if (qev->typ == rpl_parallel_thread::queued_event::QUEUED_POS_UPDATE)
+      {
+        handle_queued_pos_update(thd, qev);
+        rpt->loc_free_qev(qev);
+        continue;
+      }
+      else if (qev->typ ==
+               rpl_parallel_thread::queued_event::QUEUED_MASTER_RESTART)
+      {
+        if (in_event_group)
+        {
+          /*
+            Master restarted (crashed) in the middle of an event group.
+            So we need to roll back and discard that event group.
+          */
+          group_rgi->cleanup_context(thd, 1);
+          in_event_group= false;
+          finish_event_group(rpt, group_rgi->gtid_sub_id,
+                             qev->entry_for_queued, group_rgi);
+
+          rpt->loc_free_rgi(group_rgi);
+          thd->rgi_slave= group_rgi= NULL;
+        }
+
+        rpt->loc_free_qev(qev);
+        continue;
+      }
+      DBUG_ASSERT(qev->typ==rpl_parallel_thread::queued_event::QUEUED_EVENT);
+
+      thd->rgi_slave= rgi;
+      gco= rgi->gco;
+      /* Handle a new event group, which will be initiated by a GTID event. */
+      if ((event_type= qev->ev->get_type_code()) == GTID_EVENT)
+      {
+        rpt->last_trans_retry_count= 0;
+        rpt->last_seen_gtid= rgi->current_gtid;
+        rpt->channel_name_length= (uint)rgi->rli->mi->connection_name.length;
+        if (rpt->channel_name_length)
+          memcpy(rpt->channel_name, rgi->rli->mi->connection_name.str,
+                 rgi->rli->mi->connection_name.length);
+
+        bool did_enter_cond= false;
+        PSI_stage_info old_stage;
+
+        DBUG_EXECUTE_IF("rpl_parallel_delay_gtid_0_x_100_start", {
+            if (rgi->current_gtid.domain_id==0 &&
+                rgi->current_gtid.seq_no == 100)
+              my_sleep(10000);
+          });
+#ifdef ENABLED_DEBUG_SYNC
+        DBUG_EXECUTE_IF("hold_worker_on_schedule", {
+            if (rgi->current_gtid.domain_id == 0 &&
+                rgi->current_gtid.seq_no == 100) {
+                  debug_sync_set_action(thd,
+                STRING_WITH_LEN("now SIGNAL reached_pause WAIT_FOR continue_worker"));
+            }
+          });
+        DBUG_EXECUTE_IF("rpl_parallel_scheduled_gtid_0_x_100", {
+            if (rgi->current_gtid.domain_id == 0 &&
+                rgi->current_gtid.seq_no == 100) {
+              debug_sync_set_action(thd,
+                      STRING_WITH_LEN("now SIGNAL scheduled_gtid_0_x_100"));
+            }
+          });
+#endif
+
+        if(unlikely(thd->wait_for_commit_ptr) && group_rgi != NULL)
+        {
+          /*
+            This indicates that we get a new GTID event in the middle of
+            a not completed event group. This is corrupt binlog (the master
+            will never write such binlog), so it does not happen unless
+            someone tries to inject wrong crafted binlog, but let us still
+            try to handle it somewhat nicely.
+          */
+          group_rgi->cleanup_context(thd, true);
+          finish_event_group(rpt, group_rgi->gtid_sub_id,
+                             group_rgi->parallel_entry, group_rgi);
+          rpt->loc_free_rgi(group_rgi);
+        }
+
+        thd->tx_isolation= (enum_tx_isolation)thd->variables.tx_isolation;
+        in_event_group= true;
+        /*
+          If the standalone flag is set, then this event group consists of a
+          single statement (possibly preceeded by some Intvar_log_event and
+          similar), without any terminating COMMIT/ROLLBACK/XID.
+        */
+        group_standalone=
+          (0 != (static_cast<Gtid_log_event *>(qev->ev)->flags2 &
+                 Gtid_log_event::FL_STANDALONE));
+
+        event_gtid_sub_id= rgi->gtid_sub_id;
+        rgi->thd= thd;
+
+        DBUG_EXECUTE_IF("gco_wait_delay_gtid_0_x_99", {
+            if (rgi->current_gtid.domain_id == 0 && rgi->current_gtid.seq_no == 99) {
+              debug_sync_set_action(thd,
+                  STRING_WITH_LEN("now SIGNAL gco_wait_paused WAIT_FOR gco_wait_cont"));
+            } });
+
+        mysql_mutex_lock(&entry->LOCK_parallel_entry);
+        do_gco_wait(rgi, gco, &did_enter_cond, &old_stage);
+        skip_event_group= do_stop_handling(rgi);
+        if (likely(!skip_event_group))
+          skip_event_group= do_ftwrl_wait(rgi, &did_enter_cond, &old_stage);
+
+        /*
+          Register ourself to wait for the previous commit, if we need to do
+          such registration _and_ that previous commit has not already
+          occurred.
+        */
+        register_wait_for_prior_event_group_commit(rgi, entry);
+
+        unlock_or_exit_cond(thd, &entry->LOCK_parallel_entry,
+                            &did_enter_cond, &old_stage);
+
+        thd->wait_for_commit_ptr= &rgi->commit_orderer;
+
+        if (opt_gtid_ignore_duplicates &&
+            rgi->rli->mi->using_gtid != Master_info::USE_GTID_NO)
+        {
+          int res=
+            rpl_global_gtid_slave_state->check_duplicate_gtid(&rgi->current_gtid,
+                                                             rgi);
+          if (res < 0)
+          {
+            /* Error. */
+            slave_output_error_info(rgi, thd);
+            signal_error_to_sql_driver_thread(thd, rgi, 1);
+          }
+          else if (!res)
+          {
+            /* GTID already applied by another master connection, skip. */
+            skip_event_group= true;
+          }
+          else
+          {
+            /* We have to apply the event. */
+          }
+        }
+        /*
+          If we are optimistically running transactions in parallel, but this
+          particular event group should not run in parallel with what came
+          before, then wait now for the prior transaction to complete its
+          commit.
+        */
+        if (rgi->speculation == rpl_group_info::SPECULATE_WAIT &&
+            (err= thd->wait_for_prior_commit()))
+        {
+          slave_output_error_info(rgi, thd);
+          signal_error_to_sql_driver_thread(thd, rgi, 1);
+        }
+      }
+
+      group_rgi= rgi;
+      group_ending= is_group_ending(qev->ev, event_type);
+      /*
+        We do not unmark_start_commit() here in case of an explicit ROLLBACK
+        statement. Such events should be very rare, there is no real reason
+        to try to group commit them - on the contrary, it seems best to avoid
+        running them in parallel with following group commits, as with
+        ROLLBACK events we are already deep in dangerous corner cases with
+        mix of transactional and non-transactional tables or the like. And
+        avoiding the mark_start_commit() here allows us to keep an assertion
+        in ha_rollback_trans() that we do not rollback after doing
+        mark_start_commit().
+      */
+      if (group_ending == 1 && likely(!rgi->worker_error))
+      {
+        /*
+          Do an extra check for (deadlock) kill here. This helps prevent a
+          lingering deadlock kill that occurred during normal DML processing to
+          propagate past the mark_start_commit(). If we detect a deadlock only
+          after mark_start_commit(), we have to unmark, which has at least a
+          theoretical possibility of leaving a window where it looks like all
+          transactions in a GCO have started committing, while in fact one
+          will need to rollback and retry. This is not supposed to be possible
+          (since there is a deadlock, at least one transaction should be
+          blocked from reaching commit), but this seems a fragile ensurance,
+          and there were historically a number of subtle bugs in this area.
+        */
+        if (!thd->killed)
+        {
+          DEBUG_SYNC(thd, "rpl_parallel_before_mark_start_commit");
+          if (thd->lex->stmt_accessed_temp_table())
+          {
+            /*
+              Temporary tables are special, they require strict
+              single-threaded use as they have no locks protecting concurrent
+              access. Therefore, we cannot safely use the optimization of
+              overlapping the commit of this transaction with the start of the
+              following.
+              So we skip the early mark_start_commit() and also block any
+              wakeup_subsequent_commits() until this event group is fully
+              done, inside finish_event_group().
+            */
+            rgi->commit_orderer.wakeup_blocked= true;
+          }
+          else
+            rgi->mark_start_commit();
+          DEBUG_SYNC(thd, "rpl_parallel_after_mark_start_commit");
+        }
+      }
+
+      /*
+        If the SQL thread is stopping, we just skip execution of all the
+        following event groups. We still do all the normal waiting and wakeup
+        processing between the event groups as a simple way to ensure that
+        everything is stopped and cleaned up correctly.
+      */
+      if (likely(!rgi->worker_error) && !skip_event_group)
+      {
+        ++rgi->retry_event_count;
+#ifndef DBUG_OFF
+        err= 0;
+        DBUG_EXECUTE_IF("rpl_parallel_simulate_temp_err_xid",
+          if (event_type == XID_EVENT)
+          {
+            thd->clear_error();
+            thd->get_stmt_da()->reset_diagnostics_area();
+            my_error(ER_LOCK_DEADLOCK, MYF(0));
+            err= 1;
+            DEBUG_SYNC(thd, "rpl_parallel_simulate_temp_err_xid");
+          });
+        if (!err)
+#endif
+        {
+          if (unlikely(thd->check_killed()))
+          {
+            thd->clear_error();
+            thd->get_stmt_da()->reset_diagnostics_area();
+            thd->send_kill_message();
+            err= 1;
+          }
+          else
+            err= rpt_handle_event(qev, rpt);
+        }
+        delete_or_keep_event_post_apply(rgi, event_type, qev->ev);
+        DBUG_EXECUTE_IF("rpl_parallel_simulate_temp_err_gtid_0_x_100",
+                        err= dbug_simulate_tmp_error(rgi, thd););
+        if (unlikely(err))
+        {
+          ulong max_retries= slave_trans_retries;
+          convert_kill_to_deadlock_error(rgi);
+          DBUG_EXECUTE_IF("rpl_mdev31655_zero_retries",
+                          if ((rgi->current_gtid.seq_no % 1000) == 0)
+                            max_retries= 0;
+                          );
+          if (has_temporary_error(thd) && max_retries > 0)
+            err= retry_event_group(rgi, rpt, qev);
+        }
+      }
+      else
+      {
+        delete qev->ev;
+        thd->get_stmt_da()->set_overwrite_status(true);
+        err= thd->wait_for_prior_commit();
+        thd->get_stmt_da()->set_overwrite_status(false);
+      }
+
+      end_of_group=
+        in_event_group &&
+        ((group_standalone && !Log_event::is_part_of_group(event_type)) ||
+         group_ending);
+
+      rpt->loc_free_qev(qev);
+
+      if (unlikely(err))
+      {
+        if (!rgi->worker_error)
+        {
+          slave_output_error_info(rgi, thd);
+          signal_error_to_sql_driver_thread(thd, rgi, err);
+        }
+        thd->reset_killed();
+      }
+      if (end_of_group)
+      {
+        in_event_group= false;
+        finish_event_group(rpt, event_gtid_sub_id, entry, rgi);
+        rpt->loc_free_rgi(rgi);
+        thd->rgi_slave= group_rgi= rgi= NULL;
+        skip_event_group= false;
+        DEBUG_SYNC(thd, "rpl_parallel_end_of_group");
+      }
+    }
+
+    mysql_mutex_lock(&rpt->LOCK_rpl_thread);
+    /*
+      Now that we have the lock, we can move everything from our local free
+      lists to the real free lists that are also accessible from the SQL
+      driver thread.
+    */
+    rpt->batch_free();
+
+    if ((events= rpt->event_queue) != NULL)
+    {
+      /*
+        Take next group of events from the replication pool.
+        This is faster than having to wakeup the pool manager thread to give
+        us a new event.
+      */
+      rpt->dequeue1(events);
+      mysql_mutex_unlock(&rpt->LOCK_rpl_thread);
+      goto more_events;
+    }
+
+    rpt->inuse_relaylog_refcount_update();
+
+    if (in_event_group && group_rgi->parallel_entry->force_abort)
+    {
+      /*
+        We are asked to abort, without getting the remaining events in the
+        current event group.
+
+        We have to rollback the current transaction and update the last
+        sub_id value so that SQL thread will know we are done with the
+        half-processed event group.
+      */
+      mysql_mutex_unlock(&rpt->LOCK_rpl_thread);
+      signal_error_to_sql_driver_thread(thd, group_rgi, 1);
+      finish_event_group(rpt, group_rgi->gtid_sub_id,
+                         group_rgi->parallel_entry, group_rgi);
+      in_event_group= false;
+      mysql_mutex_lock(&rpt->LOCK_rpl_thread);
+      rpt->free_rgi(group_rgi);
+      thd->rgi_slave= group_rgi= NULL;
+      skip_event_group= false;
+    }
+    if (!in_event_group)
+    {
+      /* If we are in a FLUSH TABLES FOR READ LOCK, wait for it */
+      while (rpt->current_entry && rpt->pause_for_ftwrl)
+      {
+        /*
+          We are currently in the delicate process of pausing parallel
+          replication while FLUSH TABLES WITH READ LOCK is starting. We must
+          not de-allocate the thread (setting rpt->current_owner= NULL) until
+          rpl_unpause_after_ftwrl() has woken us up.
+        */
+        rpl_parallel_entry *e= rpt->current_entry;
+        /*
+          Wait for rpl_unpause_after_ftwrl() to wake us up.
+          Note that rpl_pause_for_ftwrl() may wait for 'e->pause_sub_id'
+          to change. This should happen eventually in finish_event_group()
+        */
+        mysql_mutex_lock(&e->LOCK_parallel_entry);
+        mysql_mutex_unlock(&rpt->LOCK_rpl_thread);
+        if (rpt->pause_for_ftwrl)
+          mysql_cond_wait(&e->COND_parallel_entry, &e->LOCK_parallel_entry);
+        mysql_mutex_unlock(&e->LOCK_parallel_entry);
+        mysql_mutex_lock(&rpt->LOCK_rpl_thread);
+      }
+
+      rpt->current_owner= NULL;
+      /* Tell wait_for_done() that we are done, if it is waiting. */
+      if (likely(rpt->current_entry) &&
+          unlikely(rpt->current_entry->force_abort))
+        mysql_cond_broadcast(&rpt->COND_rpl_thread_stop);
+
+      rpt->current_entry= NULL;
+      if (!rpt->stop)
+        rpt->pool->release_thread(rpt);
+    }
+  }
+#ifdef WITH_WSREP
+  wsrep_after_command_before_result(thd);
+  wsrep_after_command_after_result(thd);
+  wsrep_close(thd);
+#endif /* WITH_WSREP */
+
+  rpt->thd= NULL;
+  mysql_mutex_unlock(&rpt->LOCK_rpl_thread);
+
+  thd->clear_error();
+  thd->catalog= 0;
+  thd->reset_query();
+  thd->reset_db(&null_clex_str);
+  thd_proc_info(thd, "Slave worker thread exiting");
+  thd->temporary_tables= 0;
+
+  THD_CHECK_SENTRY(thd);
+  server_threads.erase(thd);
+  delete thd;
+
+  mysql_mutex_lock(&rpt->LOCK_rpl_thread);
+  rpt->running= false;
+  mysql_cond_signal(&rpt->COND_rpl_thread);
+  mysql_mutex_unlock(&rpt->LOCK_rpl_thread);
+
+  my_thread_end();
+
+  return NULL;
+}
+
+
+static void
+dealloc_gco(group_commit_orderer *gco)
+{
+  mysql_cond_destroy(&gco->COND_group_commit_orderer);
+  my_free(gco);
+}
+
+/**
+   Change thread count for global parallel worker threads
+
+   @param pool          parallel thread pool
+   @param new_count     Number of threads to be in pool. 0 in shutdown
+   @param force         Force thread count to new_count even if slave
+                        threads are running
+
+   By default we don't resize pool of there are running threads.
+   However during shutdown we will always do it.
+   This is needed as any_slave_sql_running() returns 1 during shutdown
+   as we don't want to access master_info while
+   Master_info_index::free_connections are running.
+*/
+
+static int
+rpl_parallel_change_thread_count(rpl_parallel_thread_pool *pool,
+                                 uint32 new_count, bool force)
+{
+  uint32 i;
+  rpl_parallel_thread **old_list= NULL;
+  rpl_parallel_thread **new_list= NULL;
+  rpl_parallel_thread *new_free_list= NULL;
+  rpl_parallel_thread *rpt_array= NULL;
+  int res;
+
+  if ((res= pool_mark_busy(pool, current_thd)))
+    return res;
+
+  /* Protect against parallel pool resizes */
+  if (pool->count == new_count)
+  {
+    pool_mark_not_busy(pool);
+    return 0;
+  }
+
+  /*
+    If we are about to delete pool, do an extra check that there are no new
+    slave threads running since we marked pool busy
+  */
+  if (!new_count && !force)
+  {
+    if (any_slave_sql_running(false))
+    {
+      DBUG_PRINT("warning",
+                 ("SQL threads running while trying to reset parallel pool"));
+      pool_mark_not_busy(pool);
+      return 0;                                 // Ok to not resize pool
+    }
+  }
+
+  /*
+    Allocate the new list of threads up-front.
+    That way, if we fail half-way, we only need to free whatever we managed
+    to allocate, and will not be left with a half-functional thread pool.
+  */
+  if (new_count &&
+      !my_multi_malloc(PSI_INSTRUMENT_ME, MYF(MY_WME|MY_ZEROFILL),
+                       &new_list, new_count*sizeof(*new_list),
+                       &rpt_array, new_count*sizeof(*rpt_array),
+                       NULL))
+  {
+    my_error(ER_OUTOFMEMORY, MYF(0), (int(new_count*sizeof(*new_list) +
+                                          new_count*sizeof(*rpt_array))));
+    goto err;
+  }
+
+  for (i= 0; i < new_count; ++i)
+  {
+    pthread_t th;
+
+    new_list[i]= &rpt_array[i];
+    new_list[i]->delay_start= true;
+    mysql_mutex_init(key_LOCK_rpl_thread, &new_list[i]->LOCK_rpl_thread,
+                     MY_MUTEX_INIT_SLOW);
+    mysql_cond_init(key_COND_rpl_thread, &new_list[i]->COND_rpl_thread, NULL);
+    mysql_cond_init(key_COND_rpl_thread_queue,
+                    &new_list[i]->COND_rpl_thread_queue, NULL);
+    mysql_cond_init(key_COND_rpl_thread_stop,
+                    &new_list[i]->COND_rpl_thread_stop, NULL);
+    new_list[i]->pool= pool;
+    if (mysql_thread_create(key_rpl_parallel_thread, &th, &connection_attrib,
+                            handle_rpl_parallel_thread, new_list[i]))
+    {
+      my_error(ER_OUT_OF_RESOURCES, MYF(0));
+      goto err;
+    }
+    new_list[i]->next= new_free_list;
+    new_free_list= new_list[i];
+  }
+
+  /*
+    Grab each old thread in turn, and signal it to stop.
+
+    Note that since we require all replication threads to be stopped before
+    changing the parallel replication worker thread pool, all the threads will
+    be already idle and will terminate immediately.
+  */
+  for (i= 0; i < pool->count; ++i)
+  {
+    rpl_parallel_thread *rpt;
+
+    mysql_mutex_lock(&pool->LOCK_rpl_thread_pool);
+    while ((rpt= pool->free_list) == NULL)
+      mysql_cond_wait(&pool->COND_rpl_thread_pool, &pool->LOCK_rpl_thread_pool);
+    pool->free_list= rpt->next;
+    mysql_mutex_unlock(&pool->LOCK_rpl_thread_pool);
+    mysql_mutex_lock(&rpt->LOCK_rpl_thread);
+    rpt->stop= true;
+    mysql_cond_signal(&rpt->COND_rpl_thread);
+    mysql_mutex_unlock(&rpt->LOCK_rpl_thread);
+  }
+
+  for (i= 0; i < pool->count; ++i)
+  {
+    rpl_parallel_thread *rpt= pool->threads[i];
+    mysql_mutex_lock(&rpt->LOCK_rpl_thread);
+    while (rpt->running)
+      mysql_cond_wait(&rpt->COND_rpl_thread, &rpt->LOCK_rpl_thread);
+    mysql_mutex_unlock(&rpt->LOCK_rpl_thread);
+    mysql_mutex_destroy(&rpt->LOCK_rpl_thread);
+    mysql_cond_destroy(&rpt->COND_rpl_thread);
+    while (rpt->qev_free_list)
+    {
+      rpl_parallel_thread::queued_event *next= rpt->qev_free_list->next;
+      my_free(rpt->qev_free_list);
+      rpt->qev_free_list= next;
+    }
+    while (rpt->rgi_free_list)
+    {
+      rpl_group_info *next= rpt->rgi_free_list->next;
+      delete rpt->rgi_free_list;
+      rpt->rgi_free_list= next;
+    }
+    while (rpt->gco_free_list)
+    {
+      group_commit_orderer *next= rpt->gco_free_list->next_gco;
+      dealloc_gco(rpt->gco_free_list);
+      rpt->gco_free_list= next;
+    }
+  }
+
+  old_list= pool->threads;
+  if (new_count < pool->count)
+    pool->count= new_count;
+  pool->threads= new_list;
+  if (new_count > pool->count)
+    pool->count= new_count;
+  my_free(old_list);
+  pool->free_list= new_free_list;
+  for (i= 0; i < pool->count; ++i)
+  {
+    mysql_mutex_lock(&pool->threads[i]->LOCK_rpl_thread);
+    pool->threads[i]->delay_start= false;
+    pool->threads[i]->current_start_alter_id= 0;
+    pool->threads[i]->current_start_alter_domain_id= 0;
+    pool->threads[i]->reserved_start_alter_thread= false;
+    mysql_cond_signal(&pool->threads[i]->COND_rpl_thread);
+    while (!pool->threads[i]->running)
+      mysql_cond_wait(&pool->threads[i]->COND_rpl_thread,
+                      &pool->threads[i]->LOCK_rpl_thread);
+    mysql_mutex_unlock(&pool->threads[i]->LOCK_rpl_thread);
+  }
+
+  pool_mark_not_busy(pool);
+
+  return 0;
+
+err:
+  if (new_list)
+  {
+    while (new_free_list)
+    {
+      mysql_mutex_lock(&new_free_list->LOCK_rpl_thread);
+      new_free_list->delay_start= false;
+      new_free_list->stop= true;
+      mysql_cond_signal(&new_free_list->COND_rpl_thread);
+      while (!new_free_list->running)
+        mysql_cond_wait(&new_free_list->COND_rpl_thread,
+                        &new_free_list->LOCK_rpl_thread);
+      while (new_free_list->running)
+        mysql_cond_wait(&new_free_list->COND_rpl_thread,
+                        &new_free_list->LOCK_rpl_thread);
+      mysql_mutex_unlock(&new_free_list->LOCK_rpl_thread);
+      new_free_list= new_free_list->next;
+    }
+    my_free(new_list);
+  }
+  pool_mark_not_busy(pool);
+  return 1;
+}
+
+/*
+  Deactivate the parallel replication thread pool, if there are now no more
+  SQL threads running.
+*/
+
+int rpl_parallel_resize_pool_if_no_slaves(void)
+{
+  /* master_info_index is set to NULL on shutdown */
+  if (opt_slave_parallel_threads > 0 && !any_slave_sql_running(false))
+    return rpl_parallel_inactivate_pool(&global_rpl_thread_pool);
+  return 0;
+}
+
+
+/**
+  Pool activation is preceeded by taking a "lock" of pool_mark_busy
+  which guarantees the number of running slaves drops to zero atomicly
+  with the number of pool workers.
+  This resolves race between the function caller thread and one
+  that may be attempting to deactivate the pool.
+*/
+int
+rpl_parallel_activate_pool(rpl_parallel_thread_pool *pool)
+{
+  int rc= 0;
+  struct pool_bkp_for_pfs* bkp= &pool->pfs_bkp;
+
+  if ((rc= pool_mark_busy(pool, current_thd)))
+    return rc;   // killed
+
+  if (!pool->count)
+  {
+    pool_mark_not_busy(pool);
+    rc= rpl_parallel_change_thread_count(pool, opt_slave_parallel_threads,
+                                         0);
+    if (!rc)
+    {
+      if (pool->count)
+      {
+        if (bkp->inited)
+        {
+          if (bkp->count != pool->count)
+          {
+            bkp->destroy();
+            bkp->init(pool->count);
+          }
+        }
+        else
+          bkp->init(pool->count);
+        bkp->is_valid= false; // Mark backup as stale during pool init
+      }
+    }
+
+  }
+  else
+  {
+    pool_mark_not_busy(pool);
+  }
+  return rc;
+}
+
+
+int
+rpl_parallel_inactivate_pool(rpl_parallel_thread_pool *pool)
+{
+  return rpl_parallel_change_thread_count(pool, 0, 0);
+}
+
+
+void
+rpl_parallel_thread::batch_free()
+{
+  mysql_mutex_assert_owner(&LOCK_rpl_thread);
+  if (loc_qev_list)
+  {
+    *loc_qev_last_ptr_ptr= qev_free_list;
+    qev_free_list= loc_qev_list;
+    loc_qev_list= NULL;
+    dequeue2(loc_qev_size);
+    /* Signal that our queue can now accept more events. */
+    mysql_cond_signal(&COND_rpl_thread_queue);
+    loc_qev_size= 0;
+    qev_free_pending= 0;
+  }
+  if (loc_rgi_list)
+  {
+    *loc_rgi_last_ptr_ptr= rgi_free_list;
+    rgi_free_list= loc_rgi_list;
+    loc_rgi_list= NULL;
+  }
+  if (loc_gco_list)
+  {
+    *loc_gco_last_ptr_ptr= gco_free_list;
+    gco_free_list= loc_gco_list;
+    loc_gco_list= NULL;
+  }
+}
+
+
+void
+rpl_parallel_thread::inuse_relaylog_refcount_update()
+{
+  inuse_relaylog *ir= accumulated_ir_last;
+  if (ir)
+  {
+    ir->dequeued_count+= accumulated_ir_count;
+    accumulated_ir_count= 0;
+    accumulated_ir_last= NULL;
+  }
+}
+
+
+rpl_parallel_thread::queued_event *
+rpl_parallel_thread::get_qev_common(Log_event *ev, ulonglong event_size)
+{
+  queued_event *qev;
+  mysql_mutex_assert_owner(&LOCK_rpl_thread);
+  if ((qev= qev_free_list))
+    qev_free_list= qev->next;
+  else if(!(qev= (queued_event *)my_malloc(PSI_INSTRUMENT_ME, sizeof(*qev), MYF(0))))
+  {
+    my_error(ER_OUTOFMEMORY, MYF(0), (int)sizeof(*qev));
+    return NULL;
+  }
+  qev->typ= rpl_parallel_thread::queued_event::QUEUED_EVENT;
+  qev->ev= ev;
+  qev->event_size= (size_t)event_size;
+  qev->next= NULL;
+  return qev;
+}
+
+
+rpl_parallel_thread::queued_event *
+rpl_parallel_thread::get_qev(Log_event *ev, ulonglong event_size,
+                             Relay_log_info *rli)
+{
+  queued_event *qev= get_qev_common(ev, event_size);
+  if (!qev)
+    return NULL;
+  strcpy(qev->event_relay_log_name, rli->event_relay_log_name);
+  qev->event_relay_log_pos= rli->event_relay_log_pos;
+  qev->future_event_relay_log_pos= rli->future_event_relay_log_pos;
+  strcpy(qev->future_event_master_log_name, rli->future_event_master_log_name);
+  return qev;
+}
+
+
+rpl_parallel_thread::queued_event *
+rpl_parallel_thread::retry_get_qev(Log_event *ev, queued_event *orig_qev,
+                                   const char *relay_log_name,
+                                   ulonglong event_pos, ulonglong event_size)
+{
+  queued_event *qev= get_qev_common(ev, event_size);
+  if (!qev)
+    return NULL;
+  qev->rgi= orig_qev->rgi;
+  strcpy(qev->event_relay_log_name, relay_log_name);
+  qev->event_relay_log_pos= event_pos;
+  qev->future_event_relay_log_pos= event_pos+event_size;
+  strcpy(qev->future_event_master_log_name,
+         orig_qev->future_event_master_log_name);
+  return qev;
+}
+
+
+void
+rpl_parallel_thread::loc_free_qev(rpl_parallel_thread::queued_event *qev)
+{
+  inuse_relaylog *ir= qev->ir;
+  inuse_relaylog *last_ir= accumulated_ir_last;
+  if (ir != last_ir)
+  {
+    if (last_ir)
+      inuse_relaylog_refcount_update();
+    accumulated_ir_last= ir;
+  }
+  ++accumulated_ir_count;
+  if (!loc_qev_list)
+    loc_qev_last_ptr_ptr= &qev->next;
+  else
+    qev->next= loc_qev_list;
+  loc_qev_list= qev;
+  loc_qev_size+= qev->event_size;
+  /*
+    We want to release to the global free list only occasionally, to avoid
+    having to take the LOCK_rpl_thread muted too many times.
+
+    However, we do need to release regularly. If we let the unreleased part
+    grow too large, then the SQL driver thread may go to sleep waiting for
+    the queue to drop below opt_slave_parallel_max_queued, and this in turn
+    can stall all other worker threads for more stuff to do.
+  */
+  if (++qev_free_pending >= QEV_BATCH_FREE ||
+      loc_qev_size >= opt_slave_parallel_max_queued/3)
+  {
+    mysql_mutex_lock(&LOCK_rpl_thread);
+    batch_free();
+    mysql_mutex_unlock(&LOCK_rpl_thread);
+  }
+}
+
+
+void
+rpl_parallel_thread::free_qev(rpl_parallel_thread::queued_event *qev)
+{
+  mysql_mutex_assert_owner(&LOCK_rpl_thread);
+  qev->next= qev_free_list;
+  qev_free_list= qev;
+}
+
+
+rpl_group_info*
+rpl_parallel_thread::get_rgi(Relay_log_info *rli, Gtid_log_event *gtid_ev,
+                             rpl_parallel_entry *e, ulonglong event_size)
+{
+  rpl_group_info *rgi;
+  mysql_mutex_assert_owner(&LOCK_rpl_thread);
+  if ((rgi= rgi_free_list))
+  {
+    rgi_free_list= rgi->next;
+    rgi->reinit(rli);
+  }
+  else
+  {
+    if(!(rgi= new rpl_group_info(rli)))
+    {
+      my_error(ER_OUTOFMEMORY, MYF(0), (int)sizeof(*rgi));
+      return NULL;
+    }
+    rgi->is_parallel_exec = true;
+  }
+  if ((rgi->deferred_events_collecting= rli->mi->rpl_filter->is_on()) &&
+      !rgi->deferred_events)
+    rgi->deferred_events= new Deferred_log_events(rli);
+  if (event_group_new_gtid(rgi, gtid_ev))
+  {
+    free_rgi(rgi);
+    my_error(ER_OUT_OF_RESOURCES, MYF(MY_WME));
+    return NULL;
+  }
+  rgi->parallel_entry= e;
+  rgi->relay_log= rli->last_inuse_relaylog;
+  rgi->retry_start_offset= rli->future_event_relay_log_pos-event_size;
+  rgi->retry_event_count= 0;
+  rgi->killed_for_retry= rpl_group_info::RETRY_KILL_NONE;
+  /* rgi is transaction specific so we need to move this value to rgi */
+  rgi->reserved_start_alter_thread= reserved_start_alter_thread;
+  rgi->rpt= this;
+  rgi->direct_commit_alter= false;
+  rgi->finish_event_group_called= false;
+
+  DBUG_ASSERT(!rgi->sa_info);
+  /*
+    We can remove the reserved_start_alter_thread flag.
+    If we get more concurrent alter handle_split_alter will
+    automatically set this flag again.
+  */
+  reserved_start_alter_thread= false;
+  return rgi;
+}
+
+
+void
+rpl_parallel_thread::loc_free_rgi(rpl_group_info *rgi)
+{
+  DBUG_ASSERT(rgi->commit_orderer.waitee == NULL);
+  rgi->free_annotate_event();
+  if (!loc_rgi_list)
+    loc_rgi_last_ptr_ptr= &rgi->next;
+  else
+    rgi->next= loc_rgi_list;
+  loc_rgi_list= rgi;
+}
+
+
+void
+rpl_parallel_thread::free_rgi(rpl_group_info *rgi)
+{
+  mysql_mutex_assert_owner(&LOCK_rpl_thread);
+  DBUG_ASSERT(rgi->commit_orderer.waitee == NULL);
+  rgi->free_annotate_event();
+  rgi->next= rgi_free_list;
+  rgi_free_list= rgi;
+}
+
+
+group_commit_orderer *
+rpl_parallel_thread::get_gco(uint64 wait_count, group_commit_orderer *prev,
+                             uint64 prior_sub_id)
+{
+  group_commit_orderer *gco;
+  mysql_mutex_assert_owner(&LOCK_rpl_thread);
+  if ((gco= gco_free_list))
+    gco_free_list= gco->next_gco;
+  else if(!(gco= (group_commit_orderer *)my_malloc(PSI_INSTRUMENT_ME, sizeof(*gco), MYF(0))))
+  {
+    my_error(ER_OUTOFMEMORY, MYF(0), (int)sizeof(*gco));
+    return NULL;
+  }
+  mysql_cond_init(key_COND_group_commit_orderer,
+                  &gco->COND_group_commit_orderer, NULL);
+  gco->wait_count= wait_count;
+  gco->prev_gco= prev;
+  gco->next_gco= NULL;
+  gco->prior_sub_id= prior_sub_id;
+  gco->installed= false;
+  gco->flags= 0;
+#ifndef DBUG_OFF
+  gco->gc_done= false;
+#endif
+  return gco;
+}
+
+
+void
+rpl_parallel_thread::loc_free_gco(group_commit_orderer *gco)
+{
+#ifndef DBUG_OFF
+  DBUG_ASSERT(!gco->gc_done);
+  gco->gc_done= true;
+#endif
+  if (!loc_gco_list)
+    loc_gco_last_ptr_ptr= &gco->next_gco;
+  else
+    gco->next_gco= loc_gco_list;
+  loc_gco_list= gco;
+}
+
+void rpl_group_info::finish_start_alter_event_group()
+{
+   finish_event_group(rpt, this->gtid_sub_id, this->parallel_entry, this);
+}
+
+rpl_parallel_thread::rpl_parallel_thread()
+  : channel_name_length(0), last_error_number(0), last_error_timestamp(0),
+    worker_idle_time(0), last_trans_retry_count(0), start_time(0)
+{
+}
+
+
+rpl_parallel_thread_pool::rpl_parallel_thread_pool()
+  : threads(0), free_list(0), count(0), inited(false),current_start_alters(0), busy(false),
+    pfs_bkp{0, false, false, NULL}
+{
+}
+
+
+int
+rpl_parallel_thread_pool::init(uint32 size)
+{
+  threads= NULL;
+  free_list= NULL;
+  count= 0;
+  busy= false;
+
+  mysql_mutex_init(key_LOCK_rpl_thread_pool, &LOCK_rpl_thread_pool,
+                   MY_MUTEX_INIT_SLOW);
+  mysql_cond_init(key_COND_rpl_thread_pool, &COND_rpl_thread_pool, NULL);
+  inited= true;
+
+  /*
+    The pool is initially empty. Threads will be spawned when a slave SQL
+    thread is started.
+  */
+
+  return 0;
+}
+
+
+void
+rpl_parallel_thread_pool::destroy()
+{
+  deactivate();
+  pfs_bkp.destroy();
+  destroy_cond_mutex();
+}
+
+void
+rpl_parallel_thread_pool::deactivate()
+{
+  if (!inited)
+    return;
+  rpl_parallel_change_thread_count(this, 0, 1);
+}
+
+void
+rpl_parallel_thread_pool::destroy_cond_mutex()
+{
+  if (!inited)
+    return;
+  mysql_mutex_destroy(&LOCK_rpl_thread_pool);
+  mysql_cond_destroy(&COND_rpl_thread_pool);
+  inited= false;
+}
+
+
+/*
+  Wait for a worker thread to become idle. When one does, grab the thread for
+  our use and return it.
+
+  Note that we return with the worker threads's LOCK_rpl_thread mutex locked.
+*/
+struct rpl_parallel_thread *
+rpl_parallel_thread_pool::get_thread(rpl_parallel_thread **owner,
+                                     rpl_parallel_entry *entry)
+{
+  rpl_parallel_thread *rpt;
+
+  DBUG_ASSERT(count > 0);
+  mysql_mutex_lock(&LOCK_rpl_thread_pool);
+  while (unlikely(busy) || !(rpt= free_list))
+    mysql_cond_wait(&COND_rpl_thread_pool, &LOCK_rpl_thread_pool);
+  free_list= rpt->next;
+  mysql_mutex_unlock(&LOCK_rpl_thread_pool);
+  mysql_mutex_lock(&rpt->LOCK_rpl_thread);
+  rpt->current_owner= owner;
+  rpt->current_entry= entry;
+
+  return rpt;
+}
+
+
+/*
+  Release a thread to the thread pool.
+  The thread should be locked, and should not have any work queued for it.
+*/
+void
+rpl_parallel_thread_pool::release_thread(rpl_parallel_thread *rpt)
+{
+  rpl_parallel_thread *list;
+
+  mysql_mutex_assert_owner(&rpt->LOCK_rpl_thread);
+  DBUG_ASSERT(rpt->current_owner == NULL);
+  mysql_mutex_lock(&LOCK_rpl_thread_pool);
+  list= free_list;
+  rpt->next= list;
+  free_list= rpt;
+  if (!list)
+    mysql_cond_broadcast(&COND_rpl_thread_pool);
+  mysql_mutex_unlock(&LOCK_rpl_thread_pool);
+}
+
+void
+rpl_parallel_thread_pool::copy_pool_for_pfs(Relay_log_info *rli)
+{
+  if (pfs_bkp.inited)
+  {
+    for(uint i=0; i<count;i++)
+    {
+      rpl_parallel_thread *rpt, *pfs_rpt;
+      rpt= threads[i];
+      pfs_rpt= pfs_bkp.rpl_thread_arr[i];
+      if (rpt->channel_name_length)
+      {
+        pfs_rpt->channel_name_length= rpt->channel_name_length;
+        strmake(pfs_rpt->channel_name, rpt->channel_name,
+                rpt->channel_name_length);
+      }
+      pfs_rpt->thd= rpt->thd;
+      pfs_rpt->last_seen_gtid= rpt->last_seen_gtid;
+      if (rli->err_thread_id && rpt->thd->thread_id == rli->err_thread_id)
+      {
+        pfs_rpt->last_error_number= rli->last_error().number;
+        strmake(pfs_rpt->last_error_message,
+            rli->last_error().message, sizeof(rli->last_error().message));
+        pfs_rpt->last_error_timestamp= rli->last_error().skr*1000000;
+      }
+      pfs_rpt->running= false;
+      pfs_rpt->worker_idle_time= rpt->get_worker_idle_time();
+      pfs_rpt->last_trans_retry_count= rpt->last_trans_retry_count;
+    }
+    pfs_bkp.is_valid= true;
+  }
+}
+
+/* 
+  START ALTER , COMMIT ALTER / ROLLBACK ALTER scheduling
+  
+  Steps:-
+    1. (For Gtid_log_event SA). Get the worker thread which is either
+      e->rpl_threads[i] is NULL means worker from poll has not been assigned yet
+      e->rpl_threads[i]->current_owner != &e->rpl_threads[i] 
+          Thread has been released, or about to //same as choose_thread logic
+      !e->rpl_threads[i]->current_start_alter_id is 0 , safe to schedule.
+      We dont want to schedule on worker which already have been scheduled SA
+      but CA/RA has not been scheduled yet. current_start_alter_id will indicate
+      this. If we dont do this we will get deadlock.
+    2. (For Gtid_log_event SA)
+      call choose_thread_internal so that e->rpl_threads[idx] is not null
+      update the current_start_alter_id
+    3. (For Gtid_log_event SA)
+      update local e->pending_start_alters(local) variable and 
+       pool->current_start_alters(global)
+      We need 2 status variable (global and local) because we can have 
+       slave_domain_parallel_threads != pool->threads.
+    4. (For CA/RA Gtid_log_event)
+      Update e->pending_start_alters and pool->current_start_alters
+      while holding mutex lock on pool (if SA is not assigned to 
+      reserved thread)
+
+
+    @returns
+      true  Worker allocated (choose_thread_internal called)
+      false Worker not allocated (choose_thread_internal not called)
+*/
+static bool handle_split_alter(rpl_parallel_entry *e,  
+                               Gtid_log_event *gtid_ev, uint32 *idx,
+                               //choose_thread_internal specific
+                               bool *did_enter_cond, rpl_group_info* rgi,
+                               PSI_stage_info *old_stage)
+{
+  uint16 flags_extra= gtid_ev->flags_extra;
+  bool thread_allocated= false;
+  //Step 1
+  if (flags_extra & Gtid_log_event::FL_START_ALTER_E1 ||
+      //This will arrange finding threads for CA/RA as well
+      //as concurrent DDL
+      e->pending_start_alters)
+  {
+    /*
+     j is needed for round robin scheduling, we will start with rpl_thread_idx
+     go till rpl_thread_max and then start with 0 to rpl_thread_idx
+    */
+    int j= e->rpl_thread_idx;
+    for(uint i= 0; i < e->rpl_thread_max; i++)
+    {
+      if (!e->rpl_threads[j] || e->rpl_threads[j]->current_owner
+          != &e->rpl_threads[j] || !e->rpl_threads[j]->current_start_alter_id)
+      {
+        //This condition will hit atleast one time no matter what happens
+        *idx= j;
+        DBUG_PRINT("info", ("Start alter id %d", j));
+        goto idx_found;
+      }
+      j++;
+      j= j % e->rpl_thread_max;
+    }
+    //We did not find and idx
+  DBUG_ASSERT(0);
+  return false;
+idx_found:
+    e->rpl_thread_idx= *idx;
+    e->choose_thread_internal(*idx, did_enter_cond, rgi, old_stage);
+    thread_allocated= true;
+    if (flags_extra & Gtid_log_event::FL_START_ALTER_E1)
+    {
+      mysql_mutex_assert_owner(&e->rpl_threads[*idx]->LOCK_rpl_thread);
+      e->rpl_threads[e->rpl_thread_idx]->current_start_alter_id= gtid_ev->seq_no;
+      e->rpl_threads[e->rpl_thread_idx]->current_start_alter_domain_id= 
+                                                            gtid_ev->domain_id;
+      /*
+       We are locking LOCK_rpl_thread_pool becuase we are going to update
+       current_start_alters
+      */
+      mysql_mutex_lock(&global_rpl_thread_pool.LOCK_rpl_thread_pool);
+      if (e->pending_start_alters < e->rpl_thread_max - 1 &&
+              global_rpl_thread_pool.current_start_alters
+              < global_rpl_thread_pool.count - 1)
+      {
+        e->pending_start_alters++;
+        global_rpl_thread_pool.current_start_alters++;
+      }
+      else
+      {
+        e->rpl_threads[*idx]->reserved_start_alter_thread= true;
+        e->rpl_threads[*idx]->current_start_alter_id= 0;
+        e->rpl_threads[*idx]->current_start_alter_domain_id= 0;
+      }
+      mysql_mutex_unlock(&global_rpl_thread_pool.LOCK_rpl_thread_pool);
+    }
+  }
+  if(flags_extra & (Gtid_log_event::FL_COMMIT_ALTER_E1 |
+                    Gtid_log_event::FL_ROLLBACK_ALTER_E1 ))
+  {
+    //Free the corrosponding rpt current_start_alter_id
+    for(uint i= 0; i < e->rpl_thread_max; i++)
+    {
+      if(e->rpl_threads[i] &&
+          e->rpl_threads[i]->current_start_alter_id == gtid_ev->sa_seq_no &&
+          e->rpl_threads[i]->current_start_alter_domain_id == gtid_ev->domain_id)
+      {
+        mysql_mutex_lock(&global_rpl_thread_pool.LOCK_rpl_thread_pool);
+        e->rpl_threads[i]->current_start_alter_id= 0;
+        e->rpl_threads[i]->current_start_alter_domain_id= 0;
+        global_rpl_thread_pool.current_start_alters--;
+        e->pending_start_alters--;
+        DBUG_PRINT("info", ("Commit/Rollback alter id %d", i));
+        mysql_mutex_unlock(&global_rpl_thread_pool.LOCK_rpl_thread_pool);
+        break;
+      }
+    }
+  }
+
+  return thread_allocated;
+
+}
+
+
+/*
+  Obtain a worker thread that we can queue an event to.
+
+  Each invocation allocates a new worker thread, to maximise
+  parallelism. However, only up to a maximum of
+  --slave-domain-parallel-threads workers can be occupied by a single
+  replication domain; after that point, we start re-using worker threads that
+  are still executing events that were queued earlier for this thread.
+
+  We never queue more than --rpl-parallel-wait-queue_max amount of events
+  for one worker, to avoid the SQL driver thread using up all memory with
+  queued events while worker threads are stalling.
+
+  Note that this function returns with rpl_parallel_thread::LOCK_rpl_thread
+  locked. Exception is if we were killed, in which case NULL is returned.
+
+  The *did_enter_cond flag is set true if we had to wait for a worker thread
+  to become free (with mysql_cond_wait()). If so, old_stage will also be set,
+  and the LOCK_rpl_thread must be released with THD::EXIT_COND() instead
+  of mysql_mutex_unlock.
+
+  When `gtid_ev' is not NULL the last worker thread will be returned again,
+  if it is still available. Otherwise a new worker thread is allocated.
+
+  A worker for XA transaction is determined through xid hashing which
+  ensure for a XA-complete to be scheduled to the same-xid XA-prepare worker.
+*/
+rpl_parallel_thread *
+rpl_parallel_entry::choose_thread(rpl_group_info *rgi, bool *did_enter_cond,
+                                  PSI_stage_info *old_stage,
+                                  Gtid_log_event *gtid_ev)
+{
+  uint32 idx;
+
+  idx= rpl_thread_idx;
+  if (gtid_ev)
+  {
+    if (++idx >= rpl_thread_max)
+      idx= 0;
+    //rpl_thread_idx will be updated handle_split_alter
+    if (handle_split_alter(this, gtid_ev, &idx, did_enter_cond, rgi, old_stage))
+      return rpl_threads[idx];
+    if (gtid_ev->flags2 &
+        (Gtid_log_event::FL_COMPLETED_XA | Gtid_log_event::FL_PREPARED_XA))
+    {     
+      idx= my_hash_sort(&my_charset_bin, gtid_ev->xid.key(),
+                        gtid_ev->xid.key_length()) % rpl_thread_max;
+    }
+    rpl_thread_idx= idx;
+  }
+  return choose_thread_internal(idx, did_enter_cond, rgi, old_stage);
+}
+
+rpl_parallel_thread * rpl_parallel_entry::choose_thread_internal(uint idx,
+                                  bool *did_enter_cond, rpl_group_info *rgi,
+                                  PSI_stage_info *old_stage)
+{
+  rpl_parallel_thread* thr= rpl_threads[idx];
+  Relay_log_info *rli= rgi->rli;
+  if (thr)
+  {
+    *did_enter_cond= false;
+    mysql_mutex_lock(&thr->LOCK_rpl_thread);
+    for (;;)
+    {
+      if (thr->current_owner != &rpl_threads[idx])
+      {
+        /*
+          The worker thread became idle, and returned to the free list and
+          possibly was allocated to a different request. So we should allocate
+          a new worker thread.
+        */
+        unlock_or_exit_cond(rli->sql_driver_thd, &thr->LOCK_rpl_thread,
+                            did_enter_cond, old_stage);
+        thr= NULL;
+        break;
+      }
+      else if (thr->queued_size <= opt_slave_parallel_max_queued)
+      {
+        /* The thread is ready to queue into. */
+        break;
+      }
+      else
+      {
+        /*
+          We have reached the limit of how much memory we are allowed to use
+          for queuing events, so wait for the thread to consume some of its
+          queue.
+        */
+        if (!*did_enter_cond)
+        {
+          /*
+            We need to do the debug_sync before ENTER_COND().
+            Because debug_sync changes the thd->mysys_var->current_mutex,
+            and this can cause THD::awake to use the wrong mutex.
+          */
+#ifdef ENABLED_DEBUG_SYNC
+          DBUG_EXECUTE_IF("rpl_parallel_wait_queue_max",
+            {
+              debug_sync_set_action(rli->sql_driver_thd,
+                        STRING_WITH_LEN("now SIGNAL wait_queue_ready"));
+            };);
+#endif
+          rli->sql_driver_thd->ENTER_COND(&thr->COND_rpl_thread_queue,
+                                          &thr->LOCK_rpl_thread,
+                                          &stage_waiting_for_room_in_worker_thread,
+                                          old_stage);
+          *did_enter_cond= true;
+        }
+
+        if (unlikely(rli->sql_driver_thd->check_killed(1)))
+        {
+          unlock_or_exit_cond(rli->sql_driver_thd, &thr->LOCK_rpl_thread,
+                              did_enter_cond, old_stage);
+          my_error(ER_CONNECTION_KILLED, MYF(0));
+#ifdef ENABLED_DEBUG_SYNC
+          DBUG_EXECUTE_IF("rpl_parallel_wait_queue_max",
+            {
+              debug_sync_set_action(rli->sql_driver_thd,
+                        STRING_WITH_LEN("now SIGNAL wait_queue_killed"));
+            };);
+#endif
+          slave_output_error_info(rgi, rli->sql_driver_thd);
+          return NULL;
+        }
+
+        mysql_cond_wait(&thr->COND_rpl_thread_queue, &thr->LOCK_rpl_thread);
+      }
+    }
+  }
+  if (!thr)
+    rpl_threads[idx]= thr= global_rpl_thread_pool.get_thread(&rpl_threads[idx],
+                                                             this);
+
+  return thr;
+}
+
+static void
+free_rpl_parallel_entry(void *element)
+{
+  rpl_parallel_entry *e= (rpl_parallel_entry *)element;
+  while (e->current_gco)
+  {
+    group_commit_orderer *prev_gco= e->current_gco->prev_gco;
+    dealloc_gco(e->current_gco);
+    e->current_gco= prev_gco;
+  }
+  mysql_cond_destroy(&e->COND_parallel_entry);
+  mysql_mutex_destroy(&e->LOCK_parallel_entry);
+  my_free(e);
+}
+
+
+rpl_parallel::rpl_parallel() :
+  current(NULL), sql_thread_stopping(false)
+{
+  my_hash_init(PSI_INSTRUMENT_ME, &domain_hash, &my_charset_bin, 32,
+               offsetof(rpl_parallel_entry, domain_id), sizeof(uint32),
+               NULL, free_rpl_parallel_entry, HASH_UNIQUE);
+}
+
+
+void
+rpl_parallel::reset()
+{
+  my_hash_reset(&domain_hash);
+  current= NULL;
+  sql_thread_stopping= false;
+}
+
+
+rpl_parallel::~rpl_parallel()
+{
+  my_hash_free(&domain_hash);
+}
+
+
+rpl_parallel_entry *
+rpl_parallel::find(uint32 domain_id, Relay_log_info *rli)
+{
+  struct rpl_parallel_entry *e;
+
+  if (!(e= (rpl_parallel_entry *)my_hash_search(&domain_hash,
+                                                (const uchar *)&domain_id,
+                                                sizeof(domain_id))))
+  {
+    /* Allocate a new, empty one. */
+    ulong count= opt_slave_domain_parallel_threads;
+    if (count == 0 || count > opt_slave_parallel_threads)
+      count= opt_slave_parallel_threads;
+    rpl_parallel_thread **p;
+    if (!my_multi_malloc(PSI_INSTRUMENT_ME, MYF(MY_WME|MY_ZEROFILL),
+                         &e, sizeof(*e),
+                         &p, count*sizeof(*p),
+                         NULL))
+    {
+      my_error(ER_OUTOFMEMORY, MYF(0), (int)(sizeof(*e)+count*sizeof(*p)));
+      return NULL;
+    }
+    e->rpl_threads= p;
+    e->rpl_thread_max= count;
+    e->domain_id= domain_id;
+    e->stop_on_error_sub_id= (uint64)ULONGLONG_MAX;
+    e->pause_sub_id= (uint64)ULONGLONG_MAX;
+    e->pending_start_alters= 0;
+    e->rli= rli;
+    mysql_mutex_init(key_LOCK_parallel_entry, &e->LOCK_parallel_entry,
+                     MY_MUTEX_INIT_FAST);
+    mysql_cond_init(key_COND_parallel_entry, &e->COND_parallel_entry, NULL);
+    if (my_hash_insert(&domain_hash, (uchar *)e))
+    {
+      free_rpl_parallel_entry(e);
+      return NULL;
+    }
+  }
+  else
+  {
+    DBUG_ASSERT(rli == e->rli);
+
+    e->force_abort= false;
+  }
+
+  return e;
+}
+
+/**
+  Wait until all sql worker threads has stopped processing
+
+  This is called when sql thread has been killed/stopped
+*/
+
+void
+rpl_parallel::wait_for_done(THD *thd, Relay_log_info *rli)
+{
+  struct rpl_parallel_entry *e;
+  rpl_parallel_thread *rpt;
+  uint32 i, j;
+  Master_info *mi= rli->mi;
+  /*
+    First signal all workers that they must force quit; no more events will
+    be queued to complete any partial event groups executed.
+  */
+  for (i= 0; i < domain_hash.records; ++i)
+  {
+    e= (struct rpl_parallel_entry *)my_hash_element(&domain_hash, i);
+    mysql_mutex_lock(&e->LOCK_parallel_entry);
+    /*
+      We want the worker threads to stop as quickly as is safe. If the slave
+      SQL threads are behind, we could have significant amount of events
+      queued for the workers, and we want to stop without waiting for them
+      all to be applied first. But if any event group has already started
+      executing in a worker, we want to be sure that all prior event groups
+      are also executed, so that we stop at a consistent point in the binlog
+      stream (per replication domain).
+
+      At this point, we are holding LOCK_parallel_entry, and we know that no
+      event group after e->largest_started_sub_id has started running yet. We
+      record this value in e->stop_sub_id, and then each event group can check
+      their own sub_id against it. If their sub_id is strictly larger, then
+      that event group will be skipped.
+
+      If we stop due to reaching the START SLAVE UNTIL condition, then we
+      need to continue executing any queued events up to that point.
+    */
+    e->force_abort= true;
+    e->stop_sub_id= rli->stop_for_until ?
+      e->current_sub_id : e->largest_started_sub_id;
+    mysql_mutex_unlock(&e->LOCK_parallel_entry);
+    for (j= 0; j < e->rpl_thread_max; ++j)
+    {
+      if ((rpt= e->rpl_threads[j]))
+      {
+        mysql_mutex_lock(&rpt->LOCK_rpl_thread);
+        if (rpt->current_owner == &e->rpl_threads[j])
+          mysql_cond_signal(&rpt->COND_rpl_thread);
+        mysql_mutex_unlock(&rpt->LOCK_rpl_thread);
+      }
+    }
+  }
+#ifdef ENABLED_DEBUG_SYNC
+  DBUG_EXECUTE_IF("rpl_parallel_wait_for_done_trigger",
+  {
+    debug_sync_set_action(thd,
+                          STRING_WITH_LEN("now SIGNAL wait_for_done_waiting"));
+  };);
+#endif
+
+  global_rpl_thread_pool.copy_pool_for_pfs(rli);
+  /*
+    Shutdown SA alter threads through marking their execution states
+    to force their early post-SA execution exit. Upon that the affected SA threads
+    change their state to COMPLETED, notify any waiting CA|RA and this thread.
+  */
+  start_alter_info *info=NULL;
+  mysql_mutex_lock(&mi->start_alter_list_lock);
+  List_iterator<start_alter_info> info_iterator(mi->start_alter_list);
+  mi->is_shutdown= true;   // a sign to stop in concurrently coming in new SA:s
+  while ((info= info_iterator++))
+  {
+    mysql_mutex_lock(&mi->start_alter_lock);
+    if (info->state == start_alter_state::COMPLETED)
+    {
+      mysql_mutex_unlock(&mi->start_alter_lock);
+      continue;
+    }
+    info->state= start_alter_state::ROLLBACK_ALTER;
+    // Any possible CA that is (will be) waiting will complete this ALTER instance
+    info->direct_commit_alter= true;
+    mysql_cond_broadcast(&info->start_alter_cond); // notify SA:s
+    mysql_mutex_unlock(&mi->start_alter_lock);
+
+    // await SA in the COMPLETED state
+    mysql_mutex_lock(&mi->start_alter_lock);
+    while(info->state == start_alter_state::ROLLBACK_ALTER)
+      mysql_cond_wait(&info->start_alter_cond, &mi->start_alter_lock);
+
+    DBUG_ASSERT(info->state == start_alter_state::COMPLETED);
+
+    mysql_mutex_unlock(&mi->start_alter_lock);
+  }
+  mysql_mutex_unlock(&mi->start_alter_list_lock);
+
+  DBUG_EXECUTE_IF("rpl_slave_stop_CA_before_binlog",
+    {
+      debug_sync_set_action(thd, STRING_WITH_LEN("now signal proceed_CA_1"));
+    });
+
+  for (i= 0; i < domain_hash.records; ++i)
+  {
+    e= (struct rpl_parallel_entry *)my_hash_element(&domain_hash, i);
+    for (j= 0; j < e->rpl_thread_max; ++j)
+    {
+      if ((rpt= e->rpl_threads[j]))
+      {
+        mysql_mutex_lock(&rpt->LOCK_rpl_thread);
+        while (rpt->current_owner == &e->rpl_threads[j])
+          mysql_cond_wait(&rpt->COND_rpl_thread_stop, &rpt->LOCK_rpl_thread);
+        mysql_mutex_unlock(&rpt->LOCK_rpl_thread);
+      }
+    }
+  }
+  // Now that all threads are docked, remained alter states are safe to destroy
+  mysql_mutex_lock(&mi->start_alter_list_lock);
+  info_iterator.rewind();
+  while ((info= info_iterator++))
+  {
+    info_iterator.remove();
+    mysql_cond_destroy(&info->start_alter_cond);
+    my_free(info);
+  }
+  mi->is_shutdown= false;
+  mysql_mutex_unlock(&mi->start_alter_list_lock);
+}
+
+
+/*
+  This function handles the case where the SQL driver thread reached the
+  START SLAVE UNTIL position; we stop queueing more events but continue
+  processing remaining, already queued events; then use executes manual
+  STOP SLAVE; then this function signals to worker threads that they
+  should stop the processing of any remaining queued events.
+*/
+void
+rpl_parallel::stop_during_until()
+{
+  struct rpl_parallel_entry *e;
+  uint32 i;
+
+  for (i= 0; i < domain_hash.records; ++i)
+  {
+    e= (struct rpl_parallel_entry *)my_hash_element(&domain_hash, i);
+    mysql_mutex_lock(&e->LOCK_parallel_entry);
+    if (e->force_abort)
+      e->stop_sub_id= e->largest_started_sub_id;
+    mysql_mutex_unlock(&e->LOCK_parallel_entry);
+  }
+}
+
+
+bool
+rpl_parallel::workers_idle()
+{
+  struct rpl_parallel_entry *e;
+  uint32 i, max_i;
+
+  max_i= domain_hash.records;
+  for (i= 0; i < max_i; ++i)
+  {
+    bool active;
+    e= (struct rpl_parallel_entry *)my_hash_element(&domain_hash, i);
+    mysql_mutex_lock(&e->LOCK_parallel_entry);
+    active= e->current_sub_id > e->last_committed_sub_id;
+    mysql_mutex_unlock(&e->LOCK_parallel_entry);
+    if (active)
+      break;
+  }
+  return (i == max_i);
+}
+
+
+int
+rpl_parallel_entry::queue_master_restart(rpl_group_info *rgi,
+                                         Format_description_log_event *fdev)
+{
+  uint32 idx;
+  rpl_parallel_thread *thr;
+  rpl_parallel_thread::queued_event *qev;
+  Relay_log_info *rli= rgi->rli;
+
+  /*
+    We only need to queue the server restart if we still have a thread working
+    on a (potentially partial) event group.
+
+    If the last thread we queued for has finished, then it cannot have any
+    partial event group that needs aborting.
+
+    Thus there is no need for the full complexity of choose_thread(). We only
+    need to check if we have a current worker thread, and queue for it if so.
+  */
+  idx= rpl_thread_idx;
+  thr= rpl_threads[idx];
+  if (!thr)
+    return 0;
+  mysql_mutex_lock(&thr->LOCK_rpl_thread);
+  if (thr->current_owner != &rpl_threads[idx])
+  {
+    /* No active worker thread, so no need to queue the master restart. */
+    mysql_mutex_unlock(&thr->LOCK_rpl_thread);
+    return 0;
+  }
+
+  if (!(qev= thr->get_qev(fdev, 0, rli)))
+  {
+    mysql_mutex_unlock(&thr->LOCK_rpl_thread);
+    return 1;
+  }
+
+  qev->rgi= rgi;
+  qev->typ= rpl_parallel_thread::queued_event::QUEUED_MASTER_RESTART;
+  qev->entry_for_queued= this;
+  qev->ir= rli->last_inuse_relaylog;
+  ++qev->ir->queued_count;
+  thr->enqueue(qev);
+  mysql_cond_signal(&thr->COND_rpl_thread);
+  mysql_mutex_unlock(&thr->LOCK_rpl_thread);
+  return 0;
+}
+
+
+int
+rpl_parallel::wait_for_workers_idle(THD *thd)
+{
+  uint32 i, max_i;
+
+  /*
+    The domain_hash is only accessed by the SQL driver thread, so it is safe
+    to iterate over without a lock.
+  */
+  max_i= domain_hash.records;
+  for (i= 0; i < max_i; ++i)
+  {
+    PSI_stage_info old_stage;
+    struct rpl_parallel_entry *e;
+    int err= 0;
+
+    e= (struct rpl_parallel_entry *)my_hash_element(&domain_hash, i);
+    mysql_mutex_lock(&e->LOCK_parallel_entry);
+    ++e->need_sub_id_signal;
+    thd->ENTER_COND(&e->COND_parallel_entry, &e->LOCK_parallel_entry,
+                    &stage_waiting_for_workers_idle, &old_stage);
+    while (e->current_sub_id > e->last_committed_sub_id)
+    {
+      if (unlikely(thd->check_killed()))
+      {
+        err= 1;
+        break;
+      }
+      mysql_cond_wait(&e->COND_parallel_entry, &e->LOCK_parallel_entry);
+    }
+    --e->need_sub_id_signal;
+    thd->EXIT_COND(&old_stage);
+    if (err)
+      return err;
+  }
+  return 0;
+}
+
+
+/*
+  Handle seeing a GTID during slave restart in GTID mode. If we stopped with
+  different replication domains having reached different positions in the relay
+  log, we need to skip event groups in domains that are further progressed.
+
+  Updates the state with the seen GTID, and returns true if this GTID should
+  be skipped, false otherwise.
+*/
+bool
+process_gtid_for_restart_pos(Relay_log_info *rli, rpl_gtid *gtid)
+{
+  slave_connection_state::entry *gtid_entry;
+  slave_connection_state *state= &rli->restart_gtid_pos;
+
+  if (likely(state->count() == 0) ||
+      !(gtid_entry= state->find_entry(gtid->domain_id)))
+    return false;
+  if (gtid->server_id == gtid_entry->gtid.server_id)
+  {
+    uint64 seq_no= gtid_entry->gtid.seq_no;
+    if (gtid->seq_no >= seq_no)
+    {
+      /*
+        This domain has reached its start position. So remove it, so that
+        further events will be processed normally.
+      */
+      state->remove(&gtid_entry->gtid);
+    }
+    return gtid->seq_no <= seq_no;
+  }
+  else
+    return true;
+}
+
+
+/*
+  This is used when we get an error during processing in do_event();
+  We will not queue any event to the thread, but we still need to wake it up
+  to be sure that it will be returned to the pool.
+*/
+static void
+abandon_worker_thread(THD *thd, rpl_parallel_thread *cur_thread,
+                      bool *did_enter_cond, PSI_stage_info *old_stage)
+{
+  unlock_or_exit_cond(thd, &cur_thread->LOCK_rpl_thread,
+                      did_enter_cond, old_stage);
+  mysql_cond_signal(&cur_thread->COND_rpl_thread);
+}
+
+
+/*
+  do_event() is executed by the sql_driver_thd thread.
+  It's main purpose is to find a thread that can execute the query.
+
+  @retval  0    ok, event was accepted
+  @retval  1    error
+  @retval -1    event should be executed serially, in the sql driver thread
+*/
+
+int
+rpl_parallel::do_event(rpl_group_info *serial_rgi, Log_event *ev,
+                       ulonglong event_size)
+{
+  rpl_parallel_entry *e;
+  rpl_parallel_thread *cur_thread;
+  rpl_parallel_thread::queued_event *qev;
+  rpl_group_info *rgi= NULL;
+  Relay_log_info *rli= serial_rgi->rli;
+  enum Log_event_type typ;
+  bool is_group_event;
+  bool did_enter_cond= false;
+  PSI_stage_info old_stage;
+
+  DBUG_EXECUTE_IF("slave_crash_if_parallel_apply", DBUG_SUICIDE(););
+  /* Handle master log name change, seen in Rotate_log_event. */
+  typ= ev->get_type_code();
+  if (unlikely(typ == ROTATE_EVENT))
+  {
+    Rotate_log_event *rev= static_cast<Rotate_log_event *>(ev);
+    if ((rev->server_id != global_system_variables.server_id ||
+         rli->replicate_same_server_id) &&
+        !rev->is_relay_log_event() &&
+        !rli->is_in_group())
+    {
+      memcpy(rli->future_event_master_log_name,
+             rev->new_log_ident, rev->ident_len+1);
+      rli->notify_group_master_log_name_update();
+    }
+  }
+
+  /*
+    Execute queries non-parallel if slave_skip_counter is set, as it's is
+    easier to skip queries in single threaded mode.
+  */
+  if (rli->slave_skip_counter)
+    return -1;
+
+  /* Execute pre-10.0 event, which have no GTID, in single-threaded mode. */
+  is_group_event= Log_event::is_group_event(typ);
+  if (unlikely(!current) && typ != GTID_EVENT &&
+      !(unlikely(rli->gtid_skip_flag != GTID_SKIP_NOT) && is_group_event))
+    return -1;
+
+  /* Note: rli->data_lock is released by sql_delay_event(). */
+  if (sql_delay_event(ev, rli->sql_driver_thd, serial_rgi))
+  {
+    /*
+      If sql_delay_event() returns non-zero, it means that the wait timed out
+      due to slave stop. We should not queue the event in this case, it must
+      not be applied yet.
+    */
+    delete ev;
+    return 1;
+  }
+
+  if (unlikely(typ == FORMAT_DESCRIPTION_EVENT))
+  {
+    Format_description_log_event *fdev=
+      static_cast<Format_description_log_event *>(ev);
+    if (fdev->created)
+    {
+      /*
+        This format description event marks a new binlog after a master server
+        restart. We are going to close all temporary tables to clean up any
+        possible left-overs after a prior master crash.
+
+        Thus we need to wait for all prior events to execute to completion,
+        in case they need access to any of the temporary tables.
+
+        We also need to notify the worker thread running the prior incomplete
+        event group (if any), as such event group signifies an incompletely
+        written group cut short by a master crash, and must be rolled back.
+      */
+      if (current->queue_master_restart(serial_rgi, fdev) ||
+          wait_for_workers_idle(rli->sql_driver_thd))
+      {
+        delete ev;
+        return 1;
+      }
+    }
+  }
+  else if (unlikely(typ == GTID_LIST_EVENT))
+  {
+    Gtid_list_log_event *glev= static_cast<Gtid_list_log_event *>(ev);
+    rpl_gtid *list= glev->list;
+    uint32 count= glev->count;
+    rli->update_relay_log_state(list, count);
+    while (count)
+    {
+      process_gtid_for_restart_pos(rli, list);
+      ++list;
+      --count;
+    }
+  }
+
+  /*
+    Stop queueing additional event groups once the SQL thread is requested to
+    stop.
+
+    We have to queue any remaining events of any event group that has already
+    been partially queued, but after that we will just ignore any further
+    events the SQL driver thread may try to queue, and eventually it will stop.
+  */
+  if ((typ == GTID_EVENT || !is_group_event) && rli->abort_slave)
+    sql_thread_stopping= true;
+  if (sql_thread_stopping)
+  {
+    delete ev;
+    /*
+      Return "no error"; normal stop is not an error, and otherwise the error
+      has already been recorded.
+    */
+    return 0;
+  }
+
+  if (unlikely(rli->gtid_skip_flag != GTID_SKIP_NOT) && is_group_event)
+  {
+    if (typ == GTID_EVENT)
+      rli->gtid_skip_flag= GTID_SKIP_NOT;
+    else
+    {
+      if (rli->gtid_skip_flag == GTID_SKIP_STANDALONE)
+      {
+        if (!Log_event::is_part_of_group(typ))
+          rli->gtid_skip_flag= GTID_SKIP_NOT;
+      }
+      else
+      {
+        DBUG_ASSERT(rli->gtid_skip_flag == GTID_SKIP_TRANSACTION);
+        if (typ == XID_EVENT || typ == XA_PREPARE_LOG_EVENT ||
+            (typ == QUERY_EVENT &&  // COMMIT/ROLLBACK are never compressed
+             (((Query_log_event *)ev)->is_commit() ||
+              ((Query_log_event *)ev)->is_rollback())))
+          rli->gtid_skip_flag= GTID_SKIP_NOT;
+      }
+      delete_or_keep_event_post_apply(serial_rgi, typ, ev);
+      return 0;
+    }
+  }
+
+  Gtid_log_event *gtid_ev= NULL;
+  if (typ == GTID_EVENT)
+  {
+    rpl_gtid gtid;
+    gtid_ev= static_cast<Gtid_log_event *>(ev);
+    uint32 domain_id= (rli->mi->using_gtid == Master_info::USE_GTID_NO ||
+                       rli->mi->parallel_mode <= SLAVE_PARALLEL_MINIMAL ?
+                       0 : gtid_ev->domain_id);
+    if (!(e= find(domain_id, rli)))
+    {
+      my_error(ER_OUT_OF_RESOURCES, MYF(MY_WME));
+      delete ev;
+      return 1;
+    }
+    current= e;
+
+    gtid.domain_id= gtid_ev->domain_id;
+    gtid.server_id= gtid_ev->server_id;
+    gtid.seq_no= gtid_ev->seq_no;
+    rli->update_relay_log_state(&gtid, 1);
+    serial_rgi->gtid_ev_flags_extra= gtid_ev->flags_extra;
+    if (process_gtid_for_restart_pos(rli, &gtid))
+    {
+      /*
+        This domain has progressed further into the relay log before the last
+        SQL thread restart. So we need to skip this event group to not doubly
+        apply it.
+      */
+      rli->gtid_skip_flag= ((gtid_ev->flags2 & Gtid_log_event::FL_STANDALONE) ?
+                            GTID_SKIP_STANDALONE : GTID_SKIP_TRANSACTION);
+      delete_or_keep_event_post_apply(serial_rgi, typ, ev);
+      return 0;
+    }
+  }
+  else
+    e= current;
+
+  /*
+    Find a worker thread to queue the event for.
+    Prefer a new thread, so we maximise parallelism (at least for the group
+    commit). But do not exceed a limit of --slave-domain-parallel-threads;
+    instead re-use a thread that we queued for previously.
+  */
+  cur_thread=
+    e->choose_thread(serial_rgi, &did_enter_cond, &old_stage, gtid_ev);
+  if (!cur_thread)
+  {
+    /* This means we were killed. The error is already signalled. */
+    delete ev;
+    return 1;
+  }
+
+  if (!(qev= cur_thread->get_qev(ev, event_size, rli)))
+  {
+    abandon_worker_thread(rli->sql_driver_thd, cur_thread,
+                          &did_enter_cond, &old_stage);
+    delete ev;
+    return 1;
+  }
+
+  if (typ == GTID_EVENT)
+  {
+    bool new_gco;
+    enum_slave_parallel_mode mode= rli->mi->parallel_mode;
+    uchar gtid_flags= gtid_ev->flags2;
+    group_commit_orderer *gco;
+    uint8 force_switch_flag;
+    enum rpl_group_info::enum_speculation speculation;
+
+    if (!(rgi= cur_thread->get_rgi(rli, gtid_ev, e, event_size)))
+    {
+      cur_thread->free_qev(qev);
+      abandon_worker_thread(rli->sql_driver_thd, cur_thread,
+                            &did_enter_cond, &old_stage);
+      delete ev;
+      return 1;
+    }
+
+    /*
+      We queue the event group in a new worker thread, to run in parallel
+      with previous groups.
+
+      To preserve commit order within the replication domain, we set up
+      rgi->wait_commit_sub_id to make the new group commit only after the
+      previous group has committed.
+
+      Event groups that group-committed together on the master can be run
+      in parallel with each other without restrictions. But one batch of
+      group-commits may not start before all groups in the previous batch
+      have initiated their commit phase; we set up rgi->gco to ensure that.
+    */
+    rgi->wait_commit_sub_id= e->current_sub_id;
+    rgi->wait_commit_group_info= e->current_group_info;
+
+    speculation= rpl_group_info::SPECULATE_NO;
+    new_gco= true;
+    force_switch_flag= 0;
+    gco= e->current_gco;
+    if (likely(gco))
+    {
+      uint8 flags= gco->flags;
+
+      if (mode <= SLAVE_PARALLEL_MINIMAL ||
+          !(gtid_flags & Gtid_log_event::FL_GROUP_COMMIT_ID) ||
+          e->last_commit_id != gtid_ev->commit_id ||
+          /*
+            MULTI_BATCH is also set when the current gtid even being a member
+            of a commit group is flagged as DDL which disallows parallel.
+          */
+          (gtid_flags & Gtid_log_event::FL_DDL))
+        flags|= group_commit_orderer::MULTI_BATCH;
+      /* Make sure we do not attempt to run DDL in parallel speculatively. */
+      if (gtid_flags & Gtid_log_event::FL_DDL)
+        flags|= (force_switch_flag= group_commit_orderer::FORCE_SWITCH);
+
+      if (!(flags & group_commit_orderer::MULTI_BATCH))
+      {
+        /*
+          Still the same batch of event groups that group-committed together
+          on the master, so we can run in parallel.
+        */
+        new_gco= false;
+      }
+      else if ((mode >= SLAVE_PARALLEL_OPTIMISTIC) &&
+               !(flags & group_commit_orderer::FORCE_SWITCH))
+      {
+        /*
+          In transactional parallel mode, we optimistically attempt to run
+          non-DDL in parallel. In case of conflicts, we catch the conflict as
+          a deadlock or other error, roll back and retry serially.
+
+          The assumption is that only a few event groups will be
+          non-transactional or otherwise unsuitable for parallel apply. Those
+          transactions are still scheduled in parallel, but we set a flag that
+          will make the worker thread wait for everything before to complete
+          before starting.
+        */
+        new_gco= false;
+        if (!(gtid_flags & Gtid_log_event::FL_TRANSACTIONAL) ||
+            ( (!(gtid_flags & Gtid_log_event::FL_ALLOW_PARALLEL) ||
+               (gtid_flags & Gtid_log_event::FL_WAITED)) &&
+              (mode < SLAVE_PARALLEL_AGGRESSIVE)))
+        {
+          /*
+            This transaction should not be speculatively run in parallel with
+            what came before, either because it cannot safely be rolled back in
+            case of a conflict, or because it was marked as likely to conflict
+            and require expensive rollback and retry.
+
+            Here we mark it as such, and then the worker thread will do a
+            wait_for_prior_commit() before starting it. We do not introduce a
+            new group_commit_orderer, since we still want following transactions
+            to run in parallel with transactions prior to this one.
+          */
+          speculation= rpl_group_info::SPECULATE_WAIT;
+        }
+        else
+          speculation= rpl_group_info::SPECULATE_OPTIMISTIC;
+      }
+      gco->flags= flags;
+    }
+    else
+    {
+      if (gtid_flags & Gtid_log_event::FL_DDL)
+        force_switch_flag= group_commit_orderer::FORCE_SWITCH;
+    }
+    rgi->speculation= speculation;
+
+    if (gtid_flags & Gtid_log_event::FL_GROUP_COMMIT_ID)
+      e->last_commit_id= gtid_ev->commit_id;
+    else
+      e->last_commit_id= 0;
+
+    if (new_gco)
+    {
+      /*
+        Do not run this event group in parallel with what came before; instead
+        wait for everything prior to at least have started its commit phase, to
+        avoid any risk of performing any conflicting action too early.
+
+        Remember the count that marks the end of the previous batch of event
+        groups that run in parallel, and allocate a new gco.
+      */
+      uint64 count= e->count_queued_event_groups;
+
+      if (!(gco= cur_thread->get_gco(count, gco, e->current_sub_id)))
+      {
+        cur_thread->free_rgi(rgi);
+        cur_thread->free_qev(qev);
+        abandon_worker_thread(rli->sql_driver_thd, cur_thread,
+                              &did_enter_cond, &old_stage);
+        delete ev;
+        return 1;
+      }
+      gco->flags|= force_switch_flag;
+      e->current_gco= gco;
+    }
+    rgi->gco= gco;
+
+    qev->rgi= e->current_group_info= rgi;
+    e->current_sub_id= rgi->gtid_sub_id;
+    ++e->count_queued_event_groups;
+  }
+  else if (!is_group_event)
+  {
+    int err;
+    bool tmp;
+    /*
+      Events like ROTATE and FORMAT_DESCRIPTION. Do not run in worker thread.
+      Same for events not preceeded by GTID (we should not see those normally,
+      but they might be from an old master).
+    */
+    qev->rgi= serial_rgi;
+
+    tmp= serial_rgi->is_parallel_exec;
+    serial_rgi->is_parallel_exec= true;
+    err= rpt_handle_event(qev, NULL);
+    serial_rgi->is_parallel_exec= tmp;
+    if (ev->is_relay_log_event())
+      qev->future_event_master_log_pos= 0;
+    else if (typ == ROTATE_EVENT)
+      qev->future_event_master_log_pos=
+        (static_cast<Rotate_log_event *>(ev))->pos;
+    else
+      qev->future_event_master_log_pos= ev->log_pos;
+    delete_or_keep_event_post_apply(serial_rgi, typ, ev);
+
+    if (err)
+    {
+      cur_thread->free_qev(qev);
+      abandon_worker_thread(rli->sql_driver_thd, cur_thread,
+                            &did_enter_cond, &old_stage);
+      return 1;
+    }
+    /*
+      Queue a position update, so that the position will be updated in a
+      reasonable way relative to other events:
+
+       - If the currently executing events are queued serially for a single
+         thread, the position will only be updated when everything before has
+         completed.
+
+       - If we are executing multiple independent events in parallel, then at
+         least the position will not be updated until one of them has reached
+         the current point.
+    */
+    qev->typ= rpl_parallel_thread::queued_event::QUEUED_POS_UPDATE;
+    qev->entry_for_queued= e;
+  }
+  else
+  {
+    qev->rgi= e->current_group_info;
+  }
+
+  /*
+    Queue the event for processing.
+  */
+  qev->ir= rli->last_inuse_relaylog;
+  ++qev->ir->queued_count;
+  cur_thread->enqueue(qev);
+  unlock_or_exit_cond(rli->sql_driver_thd, &cur_thread->LOCK_rpl_thread,
+                      &did_enter_cond, &old_stage);
+  mysql_cond_signal(&cur_thread->COND_rpl_thread);
+
+  return 0;
+}