From cb99f65c96846f368626becac831c6d66b8ddf11 Mon Sep 17 00:00:00 2001 From: Silent Date: Sun, 11 Aug 2019 13:27:46 +0200 Subject: [PATCH 1/9] Add MakeUnique.h so make_unique is available without C++14 Includes make_unique from C++14 and make_unique_default_init from C++20, because it doesn't require compiler support --- Common/Common.vcxproj | 1 + Common/MakeUnique.h | 42 ++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 43 insertions(+) create mode 100644 Common/MakeUnique.h diff --git a/Common/Common.vcxproj b/Common/Common.vcxproj index 2146cbb8b1..49eeb14169 100644 --- a/Common/Common.vcxproj +++ b/Common/Common.vcxproj @@ -431,6 +431,7 @@ + diff --git a/Common/MakeUnique.h b/Common/MakeUnique.h new file mode 100644 index 0000000000..367e37f52b --- /dev/null +++ b/Common/MakeUnique.h @@ -0,0 +1,42 @@ +#pragma once + +#include +#include + +// Custom make_unique so that C++14 support will not be necessary for compilation +template::value, int>::type = 0> +std::unique_ptr make_unique(Args&&... args) +{ + return std::unique_ptr(new T(std::forward(args)...)); +} + +template::value && std::extent::value == 0, int>::type = 0> +std::unique_ptr make_unique(std::size_t size) +{ + return std::unique_ptr(new typename std::remove_extent::type[size]()); +} + +template::value != 0, int>::type = 0> +void make_unique(Args&&... args) = delete; + + +template::value, int>::type = 0> + std::unique_ptr make_unique_default_init() +{ + return std::unique_ptr(new T); +} + +template::value && std::extent::value == 0, int>::type = 0> + std::unique_ptr make_unique_default_init(std::size_t size) +{ + return std::unique_ptr(new typename std::remove_extent::type[size]); +} + +template::value != 0, int>::type = 0> + void make_unique_default_init(Args&&... args) = delete; From 1add9324a4834b1f79473276981a0f5b22623b9f Mon Sep 17 00:00:00 2001 From: Silent Date: Sat, 10 Aug 2019 19:10:14 +0200 Subject: [PATCH 2/9] Clean up ThreadPool and WorkerThread constructors with C++11 features, removes variable instantiation from constructor initializer lists --- ext/native/thread/threadpool.cpp | 4 ++-- ext/native/thread/threadpool.h | 24 ++++++++++++------------ 2 files changed, 14 insertions(+), 14 deletions(-) diff --git a/ext/native/thread/threadpool.cpp b/ext/native/thread/threadpool.cpp index 40d4ad04df..558d2d5c30 100644 --- a/ext/native/thread/threadpool.cpp +++ b/ext/native/thread/threadpool.cpp @@ -4,7 +4,7 @@ ///////////////////////////// WorkerThread -WorkerThread::WorkerThread() : active(true), started(false) { +WorkerThread::WorkerThread() { thread.reset(new std::thread(std::bind(&WorkerThread::WorkFunc, this))); while(!started) { }; } @@ -80,7 +80,7 @@ void LoopWorkerThread::WorkFunc() { ///////////////////////////// ThreadPool -ThreadPool::ThreadPool(int numThreads) : workersStarted(false) { +ThreadPool::ThreadPool(int numThreads) { if (numThreads <= 0) { numThreads_ = 1; ILOG("ThreadPool: Bad number of threads %i", numThreads); diff --git a/ext/native/thread/threadpool.h b/ext/native/thread/threadpool.h index 23afbbda63..294b2bcac3 100644 --- a/ext/native/thread/threadpool.h +++ b/ext/native/thread/threadpool.h @@ -21,32 +21,32 @@ public: void WaitForCompletion(); protected: - WorkerThread(bool ignored) : active(true), started(false) {} - virtual void WorkFunc(); + WorkerThread(bool ignored) {} std::unique_ptr thread; // the worker thread std::condition_variable signal; // used to signal new work std::condition_variable done; // used to signal work completion std::mutex mutex, doneMutex; // associated with each respective condition variable - volatile bool active, started; + volatile bool active = true, started = false; int jobsDone = 0; int jobsTarget = 0; private: + virtual void WorkFunc(); + std::function work_; // the work to be done by this thread - WorkerThread(const WorkerThread& other); // prevent copies - void operator =(const WorkerThread &other); + WorkerThread(const WorkerThread& other) = delete; // prevent copies + void operator =(const WorkerThread &other) = delete; }; -class LoopWorkerThread : public WorkerThread { +class LoopWorkerThread final : public WorkerThread { public: LoopWorkerThread(); void Process(const std::function &work, int start, int end); -protected: - virtual void WorkFunc(); - private: + virtual void WorkFunc() override; + int start_; int end_; std::function work_; // the work to be done by this thread @@ -68,10 +68,10 @@ private: std::vector> workers; std::mutex mutex; // used to sequentialize loop execution - bool workersStarted; + bool workersStarted = false; void StartWorkers(); - ThreadPool(const ThreadPool& other); // prevent copies - void operator =(const ThreadPool &other); + ThreadPool(const ThreadPool& other) = delete; // prevent copies + void operator =(const ThreadPool &other) = delete; }; From 6ac41a5767335ea3a5e0254736b218c06c10b5ad Mon Sep 17 00:00:00 2001 From: Silent Date: Sat, 10 Aug 2019 19:32:54 +0200 Subject: [PATCH 3/9] Use move semantics for worker thread functors --- ext/native/thread/threadpool.cpp | 8 ++++---- ext/native/thread/threadpool.h | 4 ++-- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/ext/native/thread/threadpool.cpp b/ext/native/thread/threadpool.cpp index 558d2d5c30..f821fc121d 100644 --- a/ext/native/thread/threadpool.cpp +++ b/ext/native/thread/threadpool.cpp @@ -17,9 +17,9 @@ WorkerThread::~WorkerThread() { thread->join(); } -void WorkerThread::Process(const std::function& work) { +void WorkerThread::Process(std::function work) { mutex.lock(); - work_ = work; + work_ = std::move(work); jobsTarget = jobsDone + 1; signal.notify_one(); mutex.unlock(); @@ -53,9 +53,9 @@ LoopWorkerThread::LoopWorkerThread() : WorkerThread(true) { while (!started) { }; } -void LoopWorkerThread::Process(const std::function &work, int start, int end) { +void LoopWorkerThread::Process(std::function work, int start, int end) { std::lock_guard guard(mutex); - work_ = work; + work_ = std::move(work); start_ = start; end_ = end; jobsTarget = jobsDone + 1; diff --git a/ext/native/thread/threadpool.h b/ext/native/thread/threadpool.h index 294b2bcac3..ee4bcfef33 100644 --- a/ext/native/thread/threadpool.h +++ b/ext/native/thread/threadpool.h @@ -16,7 +16,7 @@ public: virtual ~WorkerThread(); // submit a new work item - void Process(const std::function& work); + void Process(std::function work); // wait for a submitted work item to be completed void WaitForCompletion(); @@ -42,7 +42,7 @@ private: class LoopWorkerThread final : public WorkerThread { public: LoopWorkerThread(); - void Process(const std::function &work, int start, int end); + void Process(std::function work, int start, int end); private: virtual void WorkFunc() override; From 0639ffcfb52c86722fc718fc854ce0818711fc34 Mon Sep 17 00:00:00 2001 From: Silent Date: Sat, 10 Aug 2019 19:38:22 +0200 Subject: [PATCH 4/9] Handle spurious wakeups on worker thread condition variables properly --- ext/native/thread/threadpool.cpp | 14 +++++++++++--- 1 file changed, 11 insertions(+), 3 deletions(-) diff --git a/ext/native/thread/threadpool.cpp b/ext/native/thread/threadpool.cpp index f821fc121d..cdb3db217a 100644 --- a/ext/native/thread/threadpool.cpp +++ b/ext/native/thread/threadpool.cpp @@ -27,7 +27,7 @@ void WorkerThread::Process(std::function work) { void WorkerThread::WaitForCompletion() { std::unique_lock guard(doneMutex); - if (jobsDone < jobsTarget) { + while (jobsDone < jobsTarget) { done.wait(guard); } } @@ -37,7 +37,11 @@ void WorkerThread::WorkFunc() { std::unique_lock guard(mutex); started = true; while (active) { - signal.wait(guard); + // 'active == false' is one of the conditions for signaling, + // do not "optimize" it + while (active && jobsTarget <= jobsDone) { + signal.wait(guard); + } if (active) { work_(); doneMutex.lock(); @@ -67,7 +71,11 @@ void LoopWorkerThread::WorkFunc() { std::unique_lock guard(mutex); started = true; while (active) { - signal.wait(guard); + // 'active == false' is one of the conditions for signaling, + // do not "optimize" it + while (active && jobsTarget <= jobsDone) { + signal.wait(guard); + } if (active) { work_(start_, end_); doneMutex.lock(); From 400a6f30576956b7b57222d65fa8f97659660ec1 Mon Sep 17 00:00:00 2001 From: Silent Date: Sat, 10 Aug 2019 19:49:24 +0200 Subject: [PATCH 5/9] Remove 'started' bool from worker threads, as fixes for spurious wakeups made it obsolete --- ext/native/thread/threadpool.cpp | 4 ---- ext/native/thread/threadpool.h | 2 +- 2 files changed, 1 insertion(+), 5 deletions(-) diff --git a/ext/native/thread/threadpool.cpp b/ext/native/thread/threadpool.cpp index cdb3db217a..37a99f737d 100644 --- a/ext/native/thread/threadpool.cpp +++ b/ext/native/thread/threadpool.cpp @@ -6,7 +6,6 @@ WorkerThread::WorkerThread() { thread.reset(new std::thread(std::bind(&WorkerThread::WorkFunc, this))); - while(!started) { }; } WorkerThread::~WorkerThread() { @@ -35,7 +34,6 @@ void WorkerThread::WaitForCompletion() { void WorkerThread::WorkFunc() { setCurrentThreadName("Worker"); std::unique_lock guard(mutex); - started = true; while (active) { // 'active == false' is one of the conditions for signaling, // do not "optimize" it @@ -54,7 +52,6 @@ void WorkerThread::WorkFunc() { LoopWorkerThread::LoopWorkerThread() : WorkerThread(true) { thread.reset(new std::thread(std::bind(&LoopWorkerThread::WorkFunc, this))); - while (!started) { }; } void LoopWorkerThread::Process(std::function work, int start, int end) { @@ -69,7 +66,6 @@ void LoopWorkerThread::Process(std::function work, int start, in void LoopWorkerThread::WorkFunc() { setCurrentThreadName("LoopWorker"); std::unique_lock guard(mutex); - started = true; while (active) { // 'active == false' is one of the conditions for signaling, // do not "optimize" it diff --git a/ext/native/thread/threadpool.h b/ext/native/thread/threadpool.h index ee4bcfef33..581302d7d2 100644 --- a/ext/native/thread/threadpool.h +++ b/ext/native/thread/threadpool.h @@ -27,7 +27,7 @@ protected: std::condition_variable signal; // used to signal new work std::condition_variable done; // used to signal work completion std::mutex mutex, doneMutex; // associated with each respective condition variable - volatile bool active = true, started = false; + bool active = true; int jobsDone = 0; int jobsTarget = 0; private: From 34109afb2d686d81a7a35c5d31b1900ba50e48bf Mon Sep 17 00:00:00 2001 From: Silent Date: Sat, 10 Aug 2019 19:52:57 +0200 Subject: [PATCH 6/9] Cleanup mutex usage in worker threads --- ext/native/thread/threadpool.cpp | 24 ++++++++++++------------ 1 file changed, 12 insertions(+), 12 deletions(-) diff --git a/ext/native/thread/threadpool.cpp b/ext/native/thread/threadpool.cpp index 37a99f737d..d48fd6a994 100644 --- a/ext/native/thread/threadpool.cpp +++ b/ext/native/thread/threadpool.cpp @@ -9,19 +9,19 @@ WorkerThread::WorkerThread() { } WorkerThread::~WorkerThread() { - mutex.lock(); - active = false; - signal.notify_one(); - mutex.unlock(); + { + std::lock_guard guard(mutex); + active = false; + signal.notify_one(); + } thread->join(); } void WorkerThread::Process(std::function work) { - mutex.lock(); + std::lock_guard guard(mutex); work_ = std::move(work); jobsTarget = jobsDone + 1; signal.notify_one(); - mutex.unlock(); } void WorkerThread::WaitForCompletion() { @@ -42,10 +42,10 @@ void WorkerThread::WorkFunc() { } if (active) { work_(); - doneMutex.lock(); - done.notify_one(); + + std::lock_guard doneGuard(doneMutex); jobsDone++; - doneMutex.unlock(); + done.notify_one(); } } } @@ -74,10 +74,10 @@ void LoopWorkerThread::WorkFunc() { } if (active) { work_(start_, end_); - doneMutex.lock(); - done.notify_one(); + + std::lock_guard doneGuard(doneMutex); jobsDone++; - doneMutex.unlock(); + done.notify_one(); } } } From 1a992bf023f386f8973ead747abe4f9726bf0a55 Mon Sep 17 00:00:00 2001 From: Silent Date: Sat, 10 Aug 2019 20:25:16 +0200 Subject: [PATCH 7/9] Store std::thread in worker thread class by value and explicitly start it up, and don't use shared_ptr in thread pools --- ext/native/thread/threadpool.cpp | 21 +++++++++++---------- ext/native/thread/threadpool.h | 12 ++++++------ 2 files changed, 17 insertions(+), 16 deletions(-) diff --git a/ext/native/thread/threadpool.cpp b/ext/native/thread/threadpool.cpp index d48fd6a994..cdf98fd1de 100644 --- a/ext/native/thread/threadpool.cpp +++ b/ext/native/thread/threadpool.cpp @@ -1,20 +1,23 @@ #include "base/logging.h" #include "thread/threadpool.h" #include "thread/threadutil.h" +#include "Common/MakeUnique.h" ///////////////////////////// WorkerThread -WorkerThread::WorkerThread() { - thread.reset(new std::thread(std::bind(&WorkerThread::WorkFunc, this))); -} - WorkerThread::~WorkerThread() { { std::lock_guard guard(mutex); active = false; signal.notify_one(); } - thread->join(); + if (thread.joinable()) { + thread.join(); + } +} + +void WorkerThread::StartUp() { + thread = std::thread(std::bind(&WorkerThread::WorkFunc, this)); } void WorkerThread::Process(std::function work) { @@ -50,10 +53,6 @@ void WorkerThread::WorkFunc() { } } -LoopWorkerThread::LoopWorkerThread() : WorkerThread(true) { - thread.reset(new std::thread(std::bind(&LoopWorkerThread::WorkFunc, this))); -} - void LoopWorkerThread::Process(std::function work, int start, int end) { std::lock_guard guard(mutex); work_ = std::move(work); @@ -99,7 +98,9 @@ ThreadPool::ThreadPool(int numThreads) { void ThreadPool::StartWorkers() { if (!workersStarted) { for(int i = 0; i < numThreads_; ++i) { - workers.push_back(std::make_shared()); + auto workerPtr = make_unique(); + workerPtr->StartUp(); + workers.push_back(std::move(workerPtr)); } workersStarted = true; } diff --git a/ext/native/thread/threadpool.h b/ext/native/thread/threadpool.h index 581302d7d2..2064b31c28 100644 --- a/ext/native/thread/threadpool.h +++ b/ext/native/thread/threadpool.h @@ -12,18 +12,18 @@ // Only handles a single item of work at a time. class WorkerThread { public: - WorkerThread(); + WorkerThread() = default; virtual ~WorkerThread(); + void StartUp(); + // submit a new work item void Process(std::function work); // wait for a submitted work item to be completed void WaitForCompletion(); protected: - WorkerThread(bool ignored) {} - - std::unique_ptr thread; // the worker thread + std::thread thread; // the worker thread std::condition_variable signal; // used to signal new work std::condition_variable done; // used to signal work completion std::mutex mutex, doneMutex; // associated with each respective condition variable @@ -41,7 +41,7 @@ private: class LoopWorkerThread final : public WorkerThread { public: - LoopWorkerThread(); + LoopWorkerThread() = default; void Process(std::function work, int start, int end); private: @@ -65,7 +65,7 @@ public: private: int numThreads_; - std::vector> workers; + std::vector> workers; std::mutex mutex; // used to sequentialize loop execution bool workersStarted = false; From 4da4e6a97a7152801e5886d44a2e079f06bb6310 Mon Sep 17 00:00:00 2001 From: Silent Date: Sat, 10 Aug 2019 20:50:56 +0200 Subject: [PATCH 8/9] Start as many worker threads as actually used (used to start more) --- ext/native/thread/threadpool.cpp | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/ext/native/thread/threadpool.cpp b/ext/native/thread/threadpool.cpp index cdf98fd1de..a312748014 100644 --- a/ext/native/thread/threadpool.cpp +++ b/ext/native/thread/threadpool.cpp @@ -97,7 +97,8 @@ ThreadPool::ThreadPool(int numThreads) { void ThreadPool::StartWorkers() { if (!workersStarted) { - for(int i = 0; i < numThreads_; ++i) { + workers.reserve(numThreads_ - 1); + for(int i = 0; i < numThreads_ - 1; ++i) { // create one less worker thread as the thread calling ParallelLoop will also do work auto workerPtr = make_unique(); workerPtr->StartUp(); workers.push_back(std::move(workerPtr)); @@ -116,14 +117,14 @@ void ThreadPool::ParallelLoop(const std::function &loop, int lowe // but doesn't matter since all our loops are power of 2 int chunk = range / numThreads_; int s = lower; - for (int i = 0; i < numThreads_ - 1; ++i) { - workers[i]->Process(loop, s, s+chunk); + for (auto& worker : workers) { + worker->Process(loop, s, s+chunk); s+=chunk; } // This is the final chunk. loop(s, upper); - for (int i = 0; i < numThreads_ - 1; ++i) { - workers[i]->WaitForCompletion(); + for (auto& worker : workers) { + worker->WaitForCompletion(); } } else { loop(lower, upper); From 86a887d7b09358cb6a535b001aff7a8cfc73d781 Mon Sep 17 00:00:00 2001 From: Silent Date: Sat, 10 Aug 2019 20:55:52 +0200 Subject: [PATCH 9/9] Init GlobalThreadPool in a thread safe manner --- Common/ThreadPools.cpp | 12 +++++------- Common/ThreadPools.h | 4 ++-- 2 files changed, 7 insertions(+), 9 deletions(-) diff --git a/Common/ThreadPools.cpp b/Common/ThreadPools.cpp index 53bfe12dfa..9af420cf37 100644 --- a/Common/ThreadPools.cpp +++ b/Common/ThreadPools.cpp @@ -1,18 +1,16 @@ #include "ThreadPools.h" #include "../Core/Config.h" +#include "Common/MakeUnique.h" -std::shared_ptr GlobalThreadPool::pool; -bool GlobalThreadPool::initialized = false; +std::unique_ptr GlobalThreadPool::pool; +std::once_flag GlobalThreadPool::init_flag; void GlobalThreadPool::Loop(const std::function& loop, int lower, int upper) { - Inititialize(); + std::call_once(init_flag, Inititialize); pool->ParallelLoop(loop, lower, upper); } void GlobalThreadPool::Inititialize() { - if(!initialized) { - pool = std::make_shared(g_Config.iNumWorkerThreads); - initialized = true; - } + pool = make_unique(g_Config.iNumWorkerThreads); } diff --git a/Common/ThreadPools.h b/Common/ThreadPools.h index 2ae46c2c00..e04303a9d2 100644 --- a/Common/ThreadPools.h +++ b/Common/ThreadPools.h @@ -9,7 +9,7 @@ public: static void Loop(const std::function& loop, int lower, int upper); private: - static std::shared_ptr pool; - static bool initialized; + static std::unique_ptr pool; + static std::once_flag init_flag; static void Inititialize(); };