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;
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();
};
diff --git a/ext/native/thread/threadpool.cpp b/ext/native/thread/threadpool.cpp
index 40d4ad04df..a312748014 100644
--- a/ext/native/thread/threadpool.cpp
+++ b/ext/native/thread/threadpool.cpp
@@ -1,33 +1,35 @@
#include "base/logging.h"
#include "thread/threadpool.h"
#include "thread/threadutil.h"
+#include "Common/MakeUnique.h"
///////////////////////////// WorkerThread
-WorkerThread::WorkerThread() : active(true), started(false) {
- thread.reset(new std::thread(std::bind(&WorkerThread::WorkFunc, this)));
- while(!started) { };
-}
-
WorkerThread::~WorkerThread() {
- mutex.lock();
- active = false;
- signal.notify_one();
- mutex.unlock();
- thread->join();
+ {
+ std::lock_guard guard(mutex);
+ active = false;
+ signal.notify_one();
+ }
+ if (thread.joinable()) {
+ thread.join();
+ }
}
-void WorkerThread::Process(const std::function& work) {
- mutex.lock();
- work_ = work;
+void WorkerThread::StartUp() {
+ thread = std::thread(std::bind(&WorkerThread::WorkFunc, this));
+}
+
+void WorkerThread::Process(std::function work) {
+ std::lock_guard guard(mutex);
+ work_ = std::move(work);
jobsTarget = jobsDone + 1;
signal.notify_one();
- mutex.unlock();
}
void WorkerThread::WaitForCompletion() {
std::unique_lock guard(doneMutex);
- if (jobsDone < jobsTarget) {
+ while (jobsDone < jobsTarget) {
done.wait(guard);
}
}
@@ -35,27 +37,25 @@ void WorkerThread::WaitForCompletion() {
void WorkerThread::WorkFunc() {
setCurrentThreadName("Worker");
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();
- done.notify_one();
+
+ std::lock_guard doneGuard(doneMutex);
jobsDone++;
- doneMutex.unlock();
+ done.notify_one();
}
}
}
-LoopWorkerThread::LoopWorkerThread() : WorkerThread(true) {
- thread.reset(new std::thread(std::bind(&LoopWorkerThread::WorkFunc, this)));
- 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;
@@ -65,22 +65,25 @@ void LoopWorkerThread::Process(const std::function &work, int st
void LoopWorkerThread::WorkFunc() {
setCurrentThreadName("LoopWorker");
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();
- done.notify_one();
+
+ std::lock_guard doneGuard(doneMutex);
jobsDone++;
- doneMutex.unlock();
+ done.notify_one();
}
}
}
///////////////////////////// ThreadPool
-ThreadPool::ThreadPool(int numThreads) : workersStarted(false) {
+ThreadPool::ThreadPool(int numThreads) {
if (numThreads <= 0) {
numThreads_ = 1;
ILOG("ThreadPool: Bad number of threads %i", numThreads);
@@ -94,8 +97,11 @@ ThreadPool::ThreadPool(int numThreads) : workersStarted(false) {
void ThreadPool::StartWorkers() {
if (!workersStarted) {
- for(int i = 0; i < numThreads_; ++i) {
- workers.push_back(std::make_shared());
+ 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));
}
workersStarted = true;
}
@@ -111,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);
diff --git a/ext/native/thread/threadpool.h b/ext/native/thread/threadpool.h
index 23afbbda63..2064b31c28 100644
--- a/ext/native/thread/threadpool.h
+++ b/ext/native/thread/threadpool.h
@@ -12,41 +12,41 @@
// 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(const std::function& work);
+ void Process(std::function work);
// wait for a submitted work item to be completed
void WaitForCompletion();
protected:
- WorkerThread(bool ignored) : active(true), started(false) {}
- virtual void WorkFunc();
-
- 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
- volatile bool active, started;
+ bool active = true;
int jobsDone = 0;
int jobsTarget = 0;
private:
- std::function work_; // the work to be done by this thread
-
- WorkerThread(const WorkerThread& other); // prevent copies
- void operator =(const WorkerThread &other);
-};
-
-class LoopWorkerThread : public WorkerThread {
-public:
- LoopWorkerThread();
- void Process(const std::function &work, int start, int end);
-
-protected:
virtual void WorkFunc();
+ std::function work_; // the work to be done by this thread
+
+ WorkerThread(const WorkerThread& other) = delete; // prevent copies
+ void operator =(const WorkerThread &other) = delete;
+};
+
+class LoopWorkerThread final : public WorkerThread {
+public:
+ LoopWorkerThread() = default;
+ void Process(std::function work, int start, int end);
+
private:
+ virtual void WorkFunc() override;
+
int start_;
int end_;
std::function work_; // the work to be done by this thread
@@ -65,13 +65,13 @@ public:
private:
int numThreads_;
- std::vector> workers;
+ 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;
};