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.

1 files changed, 65 insertions, 22 deletions

diff --git a/storage/innobase/srv/srv0srv.cc b/storage/innobase/srv/srv0srv.cc
index a16d37e7e32..60ff945fc12 100644
--- a/storage/innobase/srv/srv0srv.cc
+++ b/storage/innobase/srv/srv0srv.cc
@@ -3,7 +3,7 @@
 Copyright (c) 1995, 2017, Oracle and/or its affiliates. All Rights Reserved.
 Copyright (c) 2008, 2009 Google Inc.
 Copyright (c) 2009, Percona Inc.
-Copyright (c) 2013, 2019, MariaDB Corporation.
+Copyright (c) 2013, 2020, MariaDB Corporation.
 
 Portions of this file contain modifications contributed and copyrighted by
 Google, Inc. Those modifications are gratefully acknowledged and are described
@@ -589,10 +589,6 @@ struct purge_coordinator_state
 };
 
 static purge_coordinator_state purge_state;
-extern tpool::waitable_task purge_coordinator_task;
-
-/** @return whether the purge coordinator thread is active */
-bool purge_sys_t::running() { return purge_coordinator_task.is_running(); }
 
 /** threadpool timer for srv_error_monitor_task(). */
 std::unique_ptr<tpool::timer> srv_error_monitor_timer;
@@ -1590,9 +1586,8 @@ static tpool::task_group purge_task_group;
 tpool::waitable_task purge_worker_task(purge_worker_callback, nullptr,
                                        &purge_task_group);
 static tpool::task_group purge_coordinator_task_group(1);
-tpool::waitable_task purge_coordinator_task(purge_coordinator_callback,
-                                            nullptr,
-                                            &purge_coordinator_task_group);
+static tpool::waitable_task purge_coordinator_task
+  (purge_coordinator_callback, nullptr, &purge_coordinator_task_group);
 
 static tpool::timer *purge_coordinator_timer;
 
@@ -1611,6 +1606,66 @@ srv_wake_purge_thread_if_not_active()
 	}
 }
 
+/** @return whether the purge tasks are active */
+bool purge_sys_t::running() const
+{
+  return purge_coordinator_task.is_running();
+}
+
+/** Stop purge during FLUSH TABLES FOR EXPORT */
+void purge_sys_t::stop()
+{
+  rw_lock_x_lock(&latch);
+
+  if (!enabled())
+  {
+    /* Shutdown must have been initiated during FLUSH TABLES FOR EXPORT. */
+    ut_ad(!srv_undo_sources);
+    rw_lock_x_unlock(&latch);
+    return;
+  }
+
+  ut_ad(srv_n_purge_threads > 0);
+
+  const auto paused= m_paused++;
+
+  rw_lock_x_unlock(&latch);
+
+  if (!paused)
+  {
+    ib::info() << "Stopping purge";
+    MONITOR_ATOMIC_INC(MONITOR_PURGE_STOP_COUNT);
+    purge_coordinator_task.disable();
+  }
+}
+
+/** Resume purge at UNLOCK TABLES after FLUSH TABLES FOR EXPORT */
+void purge_sys_t::resume()
+{
+   if (!enabled())
+   {
+     /* Shutdown must have been initiated during FLUSH TABLES FOR EXPORT. */
+     ut_ad(!srv_undo_sources);
+     return;
+   }
+   ut_ad(!srv_read_only_mode);
+   ut_ad(srv_force_recovery < SRV_FORCE_NO_BACKGROUND);
+   ut_ad(!sync_check_iterate(sync_check()));
+   purge_coordinator_task.enable();
+   rw_lock_x_lock(&latch);
+   int32_t paused= m_paused--;
+   ut_a(paused);
+
+   if (paused == 1)
+   {
+     ib::info() << "Resuming purge";
+     purge_state.m_running = 0;
+     srv_wake_purge_thread_if_not_active();
+     MONITOR_ATOMIC_INC(MONITOR_PURGE_RESUME_COUNT);
+   }
+   rw_lock_x_unlock(&latch);
+}
+
 /** Wake up the master thread if it is suspended or being suspended. */
 void
 srv_wake_master_thread()
@@ -2182,7 +2237,8 @@ static void purge_worker_callback(void*)
   ut_ad(srv_force_recovery < SRV_FORCE_NO_BACKGROUND);
   void *ctx;
   THD *thd= acquire_thd(&ctx);
-  while (srv_task_execute()) {}
+  while (srv_task_execute())
+    ut_ad(purge_sys.running());
   release_thd(thd,ctx);
 }
 
@@ -2287,19 +2343,6 @@ ulint srv_get_task_queue_length()
 }
 #endif
 
-/** Wake up the purge coordinator. */
-void
-srv_purge_wakeup()
-{
-	ut_ad(!srv_read_only_mode);
-	if (srv_force_recovery >= SRV_FORCE_NO_BACKGROUND) {
-		return;
-	}
-	ut_a(purge_sys.enabled() && !purge_sys.paused());
-	purge_state.m_running = 0;
-	srv_wake_purge_thread_if_not_active();
-}
-
 /** Shut down the purge threads. */
 void srv_purge_shutdown()
 {