WorkQueueThread: rework Cancel/Shutdown workflow

- Cancel doesn't shut down anymore.
   Allowing it to be used multiple times thoughout the life of
   the WorkQueue
 - Remove Clear, so we only have Cancel semantics
 - Add IsCancelling so work items can abort early if cancelling
 - Replace m_cancelled and m_thread.joinable() guars with m_shutdown.
 - Rename Flush to WaitForCompletion (As it's ambiguous if a function
   called flush should be blocking or not)
 - Add documentation
This commit is contained in:
Scott Mansell 2023-02-04 12:32:58 +13:00
parent acdb0c5be1
commit 6594532f10
2 changed files with 55 additions and 72 deletions

View File

@ -3,13 +3,13 @@
#pragma once #pragma once
#include <atomic>
#include <condition_variable>
#include <functional> #include <functional>
#include <queue> #include <queue>
#include <string> #include <string>
#include <thread> #include <thread>
#include "Common/Event.h"
#include "Common/Flag.h"
#include "Common/Thread.h" #include "Common/Thread.h"
// A thread that executes the given function for every item placed into its queue. // A thread that executes the given function for every item placed into its queue.
@ -26,120 +26,104 @@ public:
Reset(std::move(function)); Reset(std::move(function));
} }
~WorkQueueThread() { Shutdown(); } ~WorkQueueThread() { 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(std::function<void(T)> function) void Reset(std::function<void(T)> function)
{ {
Shutdown(); Shutdown();
std::lock_guard lg(m_lock); std::lock_guard lg(m_lock);
m_cancelled = false; m_shutdown = false;
m_function = std::move(function); m_function = std::move(function);
m_thread = std::thread(&WorkQueueThread::ThreadLoop, this); m_thread = std::thread(&WorkQueueThread::ThreadLoop, this);
} }
// Adds an item to the work queue
template <typename... Args> template <typename... Args>
void EmplaceItem(Args&&... args) void EmplaceItem(Args&&... args)
{ {
std::lock_guard lg(m_lock); std::lock_guard lg(m_lock);
if (m_shutdown) if (m_shutdown)
{
return; return;
}
m_items.emplace(std::forward<Args>(args)...); m_items.emplace(std::forward<Args>(args)...);
m_idle = false; m_idle = false;
m_worker_cond_var.notify_one(); m_worker_cond_var.notify_one();
} }
// Adds an item to the work queue
void Push(T&& item) void Push(T&& item)
{ {
std::lock_guard lg(m_lock); std::lock_guard lg(m_lock);
if (m_cancelled) if (m_shutdown)
{
return; return;
}
m_items.push(item); m_items.push(item);
m_idle = false; m_idle = false;
m_worker_cond_var.notify_one(); m_worker_cond_var.notify_one();
} }
// Adds an item to the work queue
void Push(const T& item) void Push(const T& item)
{ {
std::lock_guard lg(m_lock); std::lock_guard lg(m_lock);
if (m_cancelled) if (m_shutdown)
{ return;
return;
} m_items.push(item);
m_items.push(item); m_idle = false;
m_idle = false; m_worker_cond_var.notify_one();
m_worker_cond_var.notify_one();
} }
void Clear() // Empties the queue
// If the worker polls IsCanceling(), it can abort it's work when Cancelling
void Cancel()
{ {
std::lock_guard lg(m_lock); std::unique_lock lg(m_lock);
if (m_shutdown)
return;
m_cancelling = true;
m_items = std::queue<T>(); m_items = std::queue<T>();
m_worker_cond_var.notify_one(); m_worker_cond_var.notify_one();
} }
void Cancel() // 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
void Shutdown(bool cancel = false)
{ {
if (!m_thread.joinable())
{
return;
}
{ {
std::unique_lock lg(m_lock); std::unique_lock lg(m_lock);
m_items = std::queue<T>(); if (m_shutdown || !m_thread.joinable())
m_cancelled = true; return;
if (cancel)
{
m_cancelling = true;
m_items = std::queue<T>();
}
m_shutdown = true; m_shutdown = true;
m_worker_cond_var.notify_one(); m_worker_cond_var.notify_one();
} }
m_thread.join(); m_thread.join();
} }
void Shutdown() // Blocks until all items in the queue have been processed (or cancelled)
void WaitForCompletion()
{ {
if (!m_thread.joinable())
{
return;
}
{
std::unique_lock lg(m_lock);
m_shutdown = true;
m_worker_cond_var.notify_one();
}
m_thread.join();
}
// Doesn't return until the most recent function invocation has finished.
void ClearAndFlush()
{
if (!m_thread.joinable())
{
return;
}
std::unique_lock lg(m_lock); std::unique_lock lg(m_lock);
m_items = std::queue<T>(); if (m_idle) // Only check m_idle, we want this to work even another thread called Shutdown
m_wait_cond_var.wait(lg, [&] {
return m_idle;
});
}
// Doesn't return until the most recent function invocation has finished.
void Flush()
{
if (!m_thread.joinable())
{
return; return;
}
std::unique_lock lg(m_lock); m_wait_cond_var.wait(lg, [&] { return m_idle; });
m_wait_cond_var.wait(lg, [&] {
return m_idle;
});
} }
// For the worker to check if it should abort it's work early.
bool IsCancelling() const { return m_cancelling.load(); }
private: private:
void ThreadLoop() void ThreadLoop()
{ {
@ -151,14 +135,13 @@ private:
if (m_items.empty()) if (m_items.empty())
{ {
m_idle = true; m_idle = true;
m_cancelling = false;
m_wait_cond_var.notify_all(); m_wait_cond_var.notify_all();
m_worker_cond_var.wait(lg, [&] { m_worker_cond_var.wait(
return m_shutdown || !m_items.empty(); lg, [&] { return m_shutdown || m_cancelling.load() || !m_items.empty(); });
});
if (m_shutdown) if (m_shutdown)
{
break; break;
}
continue; continue;
} }
T item{std::move(m_items.front())}; T item{std::move(m_items.front())};
@ -176,9 +159,9 @@ private:
std::queue<T> m_items; std::queue<T> m_items;
std::condition_variable m_wait_cond_var; std::condition_variable m_wait_cond_var;
std::condition_variable m_worker_cond_var; std::condition_variable m_worker_cond_var;
std::atomic<bool> m_cancelling = false;
bool m_idle = true; bool m_idle = true;
bool m_shutdown = false; bool m_shutdown = false;
bool m_cancelled = false;
}; };
} // namespace Common } // namespace Common

View File

@ -37,7 +37,7 @@ GameTracker::GameTracker(QObject* parent) : QFileSystemWatcher(parent)
connect(qApp, &QApplication::aboutToQuit, this, [this] { connect(qApp, &QApplication::aboutToQuit, this, [this] {
m_processing_halted = true; m_processing_halted = true;
m_load_thread.Cancel(); m_load_thread.Shutdown(true);
}); });
connect(this, &QFileSystemWatcher::directoryChanged, this, &GameTracker::UpdateDirectory); connect(this, &QFileSystemWatcher::directoryChanged, this, &GameTracker::UpdateDirectory);
connect(this, &QFileSystemWatcher::fileChanged, this, &GameTracker::UpdateFile); connect(this, &QFileSystemWatcher::fileChanged, this, &GameTracker::UpdateFile);
@ -203,7 +203,7 @@ void GameTracker::RemoveDirectory(const QString& dir)
void GameTracker::RefreshAll() void GameTracker::RefreshAll()
{ {
m_processing_halted = true; m_processing_halted = true;
m_load_thread.Clear(); m_load_thread.Cancel();
m_load_thread.EmplaceItem(Command{CommandType::ResumeProcessing, {}}); m_load_thread.EmplaceItem(Command{CommandType::ResumeProcessing, {}});
if (m_needs_purge) if (m_needs_purge)