MDEV-22370 safe_mutex: Trying to lock uninitialized mutex at /data/src/10.4-bug/sql/rpl_parallel.cc, line 470 upon shutdown during FTWRL

Problem:- When we issue FTWRL with shutdown in parallel, there is race between
FTWRL and shutdown. Shutdown might destroy the mutex (pool->LOCK_rpl_thread_pool)
before FTWRL can lock it. So we can get crash on FTWRL thread

Solution:- mysql_mutex_destroy(pool->LOCK_rpl_thread_pool) should wait for
FTWRL thread to complete its work , and then destroy.
So slave_prepare_for_shutdown will just deactivate the pool, and mutex is destroyed
later in end_slave()

mysql-test/r/mdev_22370.result

@@ -0,0 +1,4 @@
+connect  con1,localhost,root,,;
+SET DEBUG_DBUG='+d,mark_busy_mdev_22370';
+FLUSH TABLES WITH READ LOCK;
+connection default;

mysql-test/t/mdev_22370.test

@@ -0,0 +1,17 @@
+#
+# MDEV-22370 safe_mutex: Trying to lock uninitialized mutex at
+#   /data/src/10.4-bug/sql/rpl_parallel.cc, line 470 upon shutdown during FTWRL
+#
+# Purpose of this test case to test crash while FTWRL and shutdown is in race
+# condition
+# Shutdown can execute first and destroy the mutex making mutex_lock in pool_mark_busy
+# to crash
+
+--source include/have_debug.inc
+--connect (con1,localhost,root,,)
+SET DEBUG_DBUG='+d,mark_busy_mdev_22370';
+--send
+  FLUSH TABLES WITH READ LOCK;
+
+--connection default
+--source include/restart_mysqld.inc

sql/rpl_parallel.cc

@@ -455,6 +455,7 @@ pool_mark_busy(rpl_parallel_thread_pool *pool, THD *thd)
     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)
   {
@@ -2001,10 +2002,24 @@ rpl_parallel_thread_pool::init(uint32 size)
 
 void
 rpl_parallel_thread_pool::destroy()
+{
+  deactivate();
+  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;

sql/rpl_parallel.h

@@ -240,6 +240,8 @@ struct rpl_parallel_thread_pool {
   rpl_parallel_thread_pool();
   int init(uint32 size);
   void destroy();
+  void deactivate();
+  void destroy_cond_mutex();
   struct rpl_parallel_thread *get_thread(rpl_parallel_thread **owner,
                                          rpl_parallel_entry *entry);
   void release_thread(rpl_parallel_thread *rpt);

sql/slave.cc

@@ -977,7 +977,7 @@ void slave_prepare_for_shutdown()
   mysql_mutex_unlock(&LOCK_active_mi);
   // It's safe to destruct worker pool now when
   // all driver threads are gone.
-  global_rpl_thread_pool.destroy();
+  global_rpl_thread_pool.deactivate();
 }
 
 /*