Better SHOW PROCESSLIST for replication

- When waiting for events, start time is now counted from start of wait
- Instead of having "Connect" as "Command" for all replication threads we
  now have:
  - Slave_IO for Slave thread reading relay log
  - Slave_SQL for slave executing SQL commands or distribution queries to
    Slave workers
  - Slave_worker for slave threads executin SQL commands in parallel replication

1 files changed, 12 insertions, 1 deletions

diff --git a/sql/rpl_parallel.cc b/sql/rpl_parallel.cc
index 18c83608cd3..a389dac1bba 100644
--- a/sql/rpl_parallel.cc
+++ b/sql/rpl_parallel.cc
@@ -307,6 +307,7 @@ do_gco_wait(rpl_group_info *rgi, group_commit_orderer *gco,
                     &stage_waiting_for_prior_transaction_to_start_commit,
                     old_stage);
     *did_enter_cond= true;
+    thd->set_time_for_next_stage();
     do
     {
       if (thd->check_killed() && !rgi->worker_error)
@@ -369,6 +370,7 @@ do_ftwrl_wait(rpl_group_info *rgi,
     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)
@@ -417,8 +419,11 @@ pool_mark_busy(rpl_parallel_thread_pool *pool, THD *thd)
   */
   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 && thd->check_killed())
@@ -534,6 +539,7 @@ rpl_pause_for_ftwrl(THD *thd)
       e->pause_sub_id= e->largest_started_sub_id;
     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)
@@ -995,7 +1001,6 @@ handle_rpl_parallel_thread(void *arg)
   */
   thd->variables.tx_isolation= ISO_REPEATABLE_READ;
 
-
   mysql_mutex_lock(&rpt->LOCK_rpl_thread);
   rpt->thd= thd;
 
@@ -1005,8 +1010,10 @@ handle_rpl_parallel_thread(void *arg)
   rpt->running= true;
   mysql_cond_signal(&rpt->COND_rpl_thread);
 
+  thd->set_command(COM_SLAVE_WORKER);
   while (!rpt->stop)
   {
+    uint wait_count= 0;
     rpl_parallel_thread::queued_event *qev, *next_qev;
 
     thd->ENTER_COND(&rpt->COND_rpl_thread, &rpt->LOCK_rpl_thread,
@@ -1025,7 +1032,11 @@ handle_rpl_parallel_thread(void *arg)
               (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);