MDEV-21674 purge_sys.stop() fails to wait for purge workers to complete

Since commit 5e62b6a5e06eb02cbde1e34e95e26f42d87fce02 (MDEV-16264),
purge_sys_t::stop() no longer waited for all purge activity to stop.

This caused problems on FLUSH TABLES...FOR EXPORT because of
purge running concurrently with the buffer pool flush.
The assertion at the end of buf_flush_dirty_pages() could fail.

The, implemented by Vladislav Vaintroub, aims to eliminate race
conditions when stopping or resuming purge:

waitable_task::disable(): Wait for the task to complete, then replace
the task callback function with noop.

waitable_task::enable(): Restore the original task callback function
after disable().

purge_sys_t::stop(): Invoke purge_coordinator_task.disable().

purge_sys_t::resume(): Invoke purge_coordinator_task.enable().

purge_sys_t::running(): Add const qualifier, and clarify the comment.
The purge coordinator task will remain active as long as any purge
worker task is active.

purge_worker_callback(): Assert purge_sys.running().

srv_purge_wakeup(): Merge with the only caller purge_sys_t::resume().

purge_coordinator_task: Use static linkage.

2 files changed, 34 insertions, 6 deletions

diff --git a/tpool/task.cc b/tpool/task.cc
index c8384321e50..0b5253bc725 100644
--- a/tpool/task.cc
+++ b/tpool/task.cc
@@ -1,4 +1,4 @@
-/* Copyright(C) 2019 MariaDB Corporation.
+/* Copyright (C) 2019, 2020, MariaDB Corporation.
 
 This program is free software; you can redistribute itand /or modify
 it under the terms of the GNU General Public License as published by
@@ -57,7 +57,7 @@ void execute_after_task_callback()
 
   /* Task that provide wait() operation. */
   waitable_task::waitable_task(callback_func func, void* arg, task_group* group) :
-    task(func,arg, group),m_mtx(),m_cv(),m_ref_count(),m_waiter_count(){}
+    task(func,arg, group),m_mtx(),m_cv(),m_ref_count(),m_waiter_count(),m_original_func(){}
 
   void waitable_task::add_ref()
   {
@@ -72,13 +72,37 @@ void execute_after_task_callback()
     if (!m_ref_count && m_waiter_count)
       m_cv.notify_all();
   }
-  void waitable_task::wait()
+  void waitable_task::wait(std::unique_lock<std::mutex>& lk)
   {
-    std::unique_lock<std::mutex> lk(m_mtx);
     m_waiter_count++;
     while (m_ref_count)
       m_cv.wait(lk);
     m_waiter_count--;
   }
+  void waitable_task::wait()
+  {
+    std::unique_lock<std::mutex> lk(m_mtx);
+    wait(lk);
+  }
 
-}
\ No newline at end of file
+  static void noop(void*)
+  {
+  }
+  void waitable_task::disable()
+  {
+    std::unique_lock<std::mutex> lk(m_mtx);
+    if (m_func == noop)
+     return;
+    wait(lk);
+    m_original_func = m_func;
+    m_func = noop;
+  }
+  void waitable_task::enable()
+  {
+    std::unique_lock<std::mutex> lk(m_mtx);
+    if(m_func != noop)
+      return;
+    wait(lk);
+    m_func = m_original_func;
+  }
+}
diff --git a/tpool/tpool.h b/tpool/tpool.h
index b4fa7210350..403a7faba50 100644
--- a/tpool/tpool.h
+++ b/tpool/tpool.h
@@ -1,4 +1,4 @@
-/* Copyright(C) 2019 MariaDB
+/* Copyright (C) 2019, 2020, MariaDB Corporation.
 
 This program is free software; you can redistribute itand /or modify
 it under the terms of the GNU General Public License as published by
@@ -96,6 +96,8 @@ class waitable_task :public task
   std::condition_variable m_cv;
   int m_ref_count;
   int m_waiter_count;
+  callback_func m_original_func;
+  void wait(std::unique_lock<std::mutex>&lk);
 public:
   waitable_task(callback_func func, void* arg, task_group* group = nullptr);
   void add_ref() override;
@@ -103,6 +105,8 @@ public:
   TPOOL_SUPPRESS_TSAN bool is_running() { return get_ref_count() > 0; }
   TPOOL_SUPPRESS_TSAN int get_ref_count() {return m_ref_count;}
   void wait();
+  void disable();
+  void enable();
   virtual ~waitable_task() {};
 };
 enum class aio_opcode