diff --git a/ext/native/thread/prioritizedworkqueue.cpp b/ext/native/thread/prioritizedworkqueue.cpp index 80c20136a1..f3b6d3f0e7 100644 --- a/ext/native/thread/prioritizedworkqueue.cpp +++ b/ext/native/thread/prioritizedworkqueue.cpp @@ -110,6 +110,7 @@ PrioritizedWorkQueueItem *PrioritizedWorkQueue::Pop() { static std::thread *workThread; static void threadfunc(PrioritizedWorkQueue *wq) { + setCurrentThreadName("PrioritizedWorkQueue"); while (true) { PrioritizedWorkQueueItem *item = wq->Pop(); if (!item) { diff --git a/ext/native/thread/threadpool.cpp b/ext/native/thread/threadpool.cpp index 9ba70fa39a..40d4ad04df 100644 --- a/ext/native/thread/threadpool.cpp +++ b/ext/native/thread/threadpool.cpp @@ -1,11 +1,11 @@ #include "base/logging.h" -#include "threadpool.h" +#include "thread/threadpool.h" +#include "thread/threadutil.h" ///////////////////////////// WorkerThread WorkerThread::WorkerThread() : active(true), started(false) { - thread = new std::thread(std::bind(&WorkerThread::WorkFunc, this)); - doneMutex.lock(); + thread.reset(new std::thread(std::bind(&WorkerThread::WorkFunc, this))); while(!started) { }; } @@ -15,22 +15,25 @@ WorkerThread::~WorkerThread() { signal.notify_one(); mutex.unlock(); thread->join(); - delete thread; } void WorkerThread::Process(const std::function& work) { mutex.lock(); work_ = work; + jobsTarget = jobsDone + 1; signal.notify_one(); mutex.unlock(); } void WorkerThread::WaitForCompletion() { std::unique_lock guard(doneMutex); - done.wait(guard); + if (jobsDone < jobsTarget) { + done.wait(guard); + } } void WorkerThread::WorkFunc() { + setCurrentThreadName("Worker"); std::unique_lock guard(mutex); started = true; while (active) { @@ -39,15 +42,15 @@ void WorkerThread::WorkFunc() { work_(); doneMutex.lock(); done.notify_one(); + jobsDone++; doneMutex.unlock(); } } } LoopWorkerThread::LoopWorkerThread() : WorkerThread(true) { - thread = new std::thread(std::bind(&LoopWorkerThread::WorkFunc, this)); - doneMutex.lock(); - while(!started) { }; + thread.reset(new std::thread(std::bind(&LoopWorkerThread::WorkFunc, this))); + while (!started) { }; } void LoopWorkerThread::Process(const std::function &work, int start, int end) { @@ -55,11 +58,12 @@ void LoopWorkerThread::Process(const std::function &work, int st work_ = work; start_ = start; end_ = end; + jobsTarget = jobsDone + 1; signal.notify_one(); - mutex.unlock(); } void LoopWorkerThread::WorkFunc() { + setCurrentThreadName("LoopWorker"); std::unique_lock guard(mutex); started = true; while (active) { @@ -68,6 +72,7 @@ void LoopWorkerThread::WorkFunc() { work_(start_, end_); doneMutex.lock(); done.notify_one(); + jobsDone++; doneMutex.unlock(); } } diff --git a/ext/native/thread/threadpool.h b/ext/native/thread/threadpool.h index 220d9aa82e..23afbbda63 100644 --- a/ext/native/thread/threadpool.h +++ b/ext/native/thread/threadpool.h @@ -24,12 +24,13 @@ protected: WorkerThread(bool ignored) : active(true), started(false) {} virtual void WorkFunc(); - std::thread *thread; // the worker thread + 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; - + int jobsDone = 0; + int jobsTarget = 0; private: std::function work_; // the work to be done by this thread