Merge pull request #13579 from jordan-woyak/work-queue-thread-cleanup

WorkQueueThread: Implement in terms of WaitableSPSCQueue. Add unit tests.

1 files changed, 107 insertions, 107 deletions

diff --git a/Source/Core/Common/WorkQueueThread.h b/Source/Core/Common/WorkQueueThread.h
index be723299b2..190afcabf2 100644
--- a/Source/Core/Common/WorkQueueThread.h
+++ b/Source/Core/Common/WorkQueueThread.h
@@ -4,169 +4,169 @@
 #pragma once
 
 #include <atomic>
-#include <condition_variable>
 #include <functional>
-#include <queue>
+#include <mutex>
 #include <string>
-#include <string_view>
 #include <thread>
 
+#include "Common/Event.h"
+#include "Common/SPSCQueue.h"
 #include "Common/Thread.h"
 
 // A thread that executes the given function for every item placed into its queue.
 
 namespace Common
 {
-template <typename T>
-class WorkQueueThread
+namespace detail
+{
+template <typename T, bool IsSingleProducer>
+class WorkQueueThreadBase final
 {
 public:
-  WorkQueueThread() = default;
-  WorkQueueThread(const std::string_view name, std::function<void(T)> function)
+  WorkQueueThreadBase() = default;
+  WorkQueueThreadBase(std::string name, std::function<void(T)> function)
   {
-    Reset(name, std::move(function));
+    Reset(std::move(name), std::move(function));
   }
-  ~WorkQueueThread() { Shutdown(); }
+  ~WorkQueueThreadBase() { Shutdown(); }
 
   // Shuts the current work thread down (if any) and starts a new thread with the given function
   // Note: Some consumers of this API push items to the queue before starting the thread.
-  void Reset(const std::string_view name, std::function<void(T)> function)
+  void Reset(std::string name, std::function<void(T)> function)
   {
+    auto lg = GetLockGuard();
     Shutdown();
-    std::lock_guard lg(m_lock);
-    m_thread_name = name;
-    m_shutdown = false;
-    m_function = std::move(function);
-    m_thread = std::thread(&WorkQueueThread::ThreadLoop, this);
+    m_run_thread.store(true, std::memory_order_relaxed);
+    m_thread = std::thread(std::bind_front(&WorkQueueThreadBase::ThreadLoop, this), std::move(name),
+                           std::move(function));
   }
 
   // Adds an item to the work queue
   template <typename... Args>
   void EmplaceItem(Args&&... args)
   {
-    std::lock_guard lg(m_lock);
-    if (m_shutdown)
-      return;
-
-    m_items.emplace(std::forward<Args>(args)...);
-    m_idle = false;
-    m_worker_cond_var.notify_one();
+    auto lg = GetLockGuard();
+    m_items.Emplace(std::forward<Args>(args)...);
+    m_event.Set();
   }
+  void Push(T&& item) { EmplaceItem(std::move(item)); }
+  void Push(const T& item) { EmplaceItem(item); }
 
-  // Adds an item to the work queue
-  void Push(T&& item)
+  // Empties the queue, skipping all work.
+  // Blocks until the current work is cancelled.
+  void Cancel()
   {
-    std::lock_guard lg(m_lock);
-    if (m_shutdown)
-      return;
-
-    m_items.push(std::move(item));
-    m_idle = false;
-    m_worker_cond_var.notify_one();
+    auto lg = GetLockGuard();
+    if (IsRunning())
+    {
+      m_skip_work.store(true, std::memory_order_relaxed);
+      WaitForCompletion();
+      m_skip_work.store(false, std::memory_order_relaxed);
+    }
+    else
+    {
+      m_items.Clear();
+    }
   }
 
-  // Adds an item to the work queue
-  void Push(const T& item)
-  {
-    std::lock_guard lg(m_lock);
-    if (m_shutdown)
-      return;
+  // Tells the worker thread to stop when its queue is empty.
+  // Blocks until the worker thread exits. Does nothing if thread isn't running.
+  void Shutdown() { StopThread(true); }
 
-    m_items.push(item);
-    m_idle = false;
-    m_worker_cond_var.notify_one();
-  }
+  // Tells the worker thread to stop immediately, potentially leaving work in the queue.
+  // Blocks until the worker thread exits. Does nothing if thread isn't running.
+  void Stop() { StopThread(false); }
 
-  // Empties the queue
-  // If the worker polls IsCanceling(), it can abort it's work when Cancelling
-  void Cancel()
+  // Stops the worker thread ASAP and empties the queue.
+  void StopAndCancel()
   {
-    std::unique_lock lg(m_lock);
-    if (m_shutdown)
-      return;
+    auto lg = GetLockGuard();
+    Stop();
+    Cancel();
+  }
 
-    m_cancelling = true;
-    m_items = std::queue<T>();
-    m_worker_cond_var.notify_one();
+  // Blocks until all items in the queue have been processed (or cancelled)
+  // Does nothing if thread isn't running.
+  void WaitForCompletion()
+  {
+    auto lg = GetLockGuard();
+    if (IsRunning())
+      m_items.WaitForEmpty();
   }
 
-  // Tells the worker to shut down when it's queue is empty
-  // Blocks until the worker thread exits.
-  // If cancel is true, will Cancel before before telling the worker to exit
-  // Otherwise, all currently queued items will complete before the worker exits
-  void Shutdown(bool cancel = false)
+private:
+  void StopThread(bool wait_for_completion)
   {
-    {
-      std::unique_lock lg(m_lock);
-      if (m_shutdown || !m_thread.joinable())
-        return;
+    auto lg = GetLockGuard();
 
-      if (cancel)
-      {
-        m_cancelling = true;
-        m_items = std::queue<T>();
-      }
+    if (wait_for_completion)
+      WaitForCompletion();
 
-      m_shutdown = true;
-      m_worker_cond_var.notify_one();
+    if (m_run_thread.exchange(false, std::memory_order_relaxed))
+    {
+      m_event.Set();
+      m_thread.join();
     }
-
-    m_thread.join();
   }
 
-  // Blocks until all items in the queue have been processed (or cancelled)
-  void WaitForCompletion()
+  auto GetLockGuard()
   {
-    std::unique_lock lg(m_lock);
-    // don't check m_shutdown, because it gets set to request a shutdown, and we want to wait until
-    // after the shutdown completes.
-    // We also check m_cancelling, because we want to ensure the worker acknowledges our cancel.
-    if (m_idle && !m_cancelling.load())
-      return;
-
-    m_wait_cond_var.wait(lg, [&] { return m_idle && !m_cancelling; });
+    struct DummyLockGuard
+    {
+      // Silences unused variable warning.
+      ~DummyLockGuard() { void(); }
+    };
+
+    if constexpr (IsSingleProducer)
+      return DummyLockGuard{};
+    else
+      return std::lock_guard{m_mutex};
   }
 
-  // If the worker polls IsCanceling(), it can abort its work when Cancelling
-  bool IsCancelling() const { return m_cancelling.load(); }
+  bool IsRunning() { return m_thread.joinable(); }
 
-private:
-  void ThreadLoop()
+  void ThreadLoop(const std::string& thread_name, const std::function<void(T)>& function)
   {
-    Common::SetCurrentThreadName(m_thread_name.c_str());
+    Common::SetCurrentThreadName(thread_name.c_str());
 
-    while (true)
+    while (m_run_thread.load(std::memory_order_relaxed))
     {
-      std::unique_lock lg(m_lock);
-      while (m_items.empty())
+      if (m_items.Empty())
+      {
+        m_event.Wait();
+        continue;
+      }
+
+      if (m_skip_work.load(std::memory_order_relaxed))
       {
-        m_idle = true;
-        m_cancelling = false;
-        m_wait_cond_var.notify_all();
-        if (m_shutdown)
-          return;
-
-        m_worker_cond_var.wait(
-            lg, [&] { return !m_items.empty() || m_shutdown || m_cancelling.load(); });
+        m_items.Clear();
+        continue;
       }
-      T item{std::move(m_items.front())};
-      m_items.pop();
-      lg.unlock();
 
-      m_function(std::move(item));
+      function(std::move(m_items.Front()));
+      m_items.Pop();
     }
   }
 
-  std::function<void(T)> m_function;
-  std::string m_thread_name;
   std::thread m_thread;
-  std::mutex m_lock;
-  std::queue<T> m_items;
-  std::condition_variable m_wait_cond_var;
-  std::condition_variable m_worker_cond_var;
-  std::atomic<bool> m_cancelling = false;
-  bool m_idle = true;
-  bool m_shutdown = false;
+  Common::WaitableSPSCQueue<T> m_items;
+  Common::Event m_event;
+  std::atomic_bool m_skip_work = false;
+  std::atomic_bool m_run_thread = false;
+
+  using DummyMutex = std::type_identity<void>;
+  using ProducerMutex = std::conditional_t<IsSingleProducer, DummyMutex, std::recursive_mutex>;
+  ProducerMutex m_mutex;
 };
+}  // namespace detail
+
+// Multiple threads may use the public interface.
+template <typename T>
+using WorkQueueThread = detail::WorkQueueThreadBase<T, false>;
+
+// A "Single Producer" WorkQueueThread.
+// It uses no mutex but only one thread can safely manipulate the queue.
+template <typename T>
+using WorkQueueThreadSP = detail::WorkQueueThreadBase<T, true>;
 
 }  // namespace Common