diff --git a/thread/threadpool.cpp b/thread/threadpool.cpp index 4524082e58..2b14500df8 100644 --- a/thread/threadpool.cpp +++ b/thread/threadpool.cpp @@ -31,9 +31,9 @@ void WorkerThread::WaitForCompletion() { void WorkerThread::WorkFunc() { mutex.lock(); started = true; - while(active) { + while (active) { signal.wait(mutex); - if(active) { + if (active) { work_(); doneMutex.lock(); done.notify_one(); @@ -42,6 +42,35 @@ void WorkerThread::WorkFunc() { } } +LoopWorkerThread::LoopWorkerThread() : WorkerThread(true) { + thread = new std::thread(std::bind(&LoopWorkerThread::WorkFunc, this)); + doneMutex.lock(); + while(!started) { }; +} + +void LoopWorkerThread::Process(const std::function &work, int start, int end) { + mutex.lock(); + work_ = work; + start_ = start; + end_ = end; + signal.notify_one(); + mutex.unlock(); +} + +void LoopWorkerThread::WorkFunc() { + mutex.lock(); + started = true; + while (active) { + signal.wait(mutex); + if (active) { + work_(start_, end_); + doneMutex.lock(); + done.notify_one(); + doneMutex.unlock(); + } + } +} + ///////////////////////////// ThreadPool ThreadPool::ThreadPool(int numThreads) : numThreads(numThreads), workersStarted(false) { @@ -50,13 +79,13 @@ ThreadPool::ThreadPool(int numThreads) : numThreads(numThreads), workersStarted( void ThreadPool::StartWorkers() { if(!workersStarted) { for(int i=0; i()); + workers.push_back(std::make_shared()); } workersStarted = true; } } -void ThreadPool::ParallelLoop(std::function loop, int lower, int upper) { +void ThreadPool::ParallelLoop(const std::function &loop, int lower, int upper) { int range = upper - lower; if (range >= numThreads * 2) { // don't parallelize tiny loops (this could be better, maybe add optional parameter that estimates work per iteration) lock_guard guard(mutex); @@ -67,7 +96,7 @@ void ThreadPool::ParallelLoop(std::function loop, int lower, int int chunk = range / numThreads; int s = lower; for (int i = 0; i < numThreads - 1; ++i) { - workers[i]->Process(std::bind(loop, s, s+chunk)); + workers[i]->Process(loop, s, s+chunk); s+=chunk; } // This is the final chunk. diff --git a/thread/threadpool.h b/thread/threadpool.h index e5b26df623..c24c30caa7 100644 --- a/thread/threadpool.h +++ b/thread/threadpool.h @@ -17,20 +17,37 @@ public: // wait for a submitted work item to be completed void WaitForCompletion(); -private: +protected: + WorkerThread(bool ignored) : active(true), started(false) {} + virtual void WorkFunc(); + std::thread *thread; // the worker thread ::condition_variable signal; // used to signal new work ::condition_variable done; // used to signal work completion ::recursive_mutex mutex, doneMutex; // associated with each respective condition variable volatile bool active, started; - std::function work_; // the work to be done by this thread - void WorkFunc(); +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(); + +private: + int start_; + int end_; + std::function work_; // the work to be done by this thread +}; + // A thread pool manages a set of worker threads, and allows the execution of parallel loops on them // individual parallel loops are fully sequentialized to simplify synchronization, which should not // be a problem as they should each use the entire system @@ -40,17 +57,17 @@ public: // don't need a destructor, "workers" is cleared on delete, // leading to the stopping and joining of all worker threads (RAII and all that) - void ParallelLoop(std::function loop, int lower, int upper); + void ParallelLoop(const std::function &loop, int lower, int upper); private: const int numThreads; - std::vector> workers; + std::vector> workers; ::recursive_mutex mutex; // used to sequentialize loop execution bool workersStarted; void StartWorkers(); ThreadPool(const ThreadPool& other); // prevent copies - void operator =(const WorkerThread &other); + void operator =(const ThreadPool &other); };