From 0fcf809d375597a003839bc16e0fb58a1e8583b7 Mon Sep 17 00:00:00 2001 From: Peter Thoman Date: Wed, 1 May 2013 13:18:25 +0200 Subject: [PATCH] Added ThreadPool with parallel loop functionality --- native.vcxproj | 25 ++------------ native.vcxproj.filters | 6 ++++ thread/threadpool.cpp | 74 ++++++++++++++++++++++++++++++++++++++++++ thread/threadpool.h | 57 ++++++++++++++++++++++++++++++++ 4 files changed, 140 insertions(+), 22 deletions(-) create mode 100644 thread/threadpool.cpp create mode 100644 thread/threadpool.h diff --git a/native.vcxproj b/native.vcxproj index dee6f91a1a..11b90a4b8f 100644 --- a/native.vcxproj +++ b/native.vcxproj @@ -245,6 +245,7 @@ + @@ -295,28 +296,7 @@ - - AnySuitable - true - Speed - AnySuitable - Speed - AnySuitable - true - Speed - false - AnySuitable - Speed - false - StreamingSIMDExtensions2 - Fast - StreamingSIMDExtensions2 - Fast - MaxSpeed - MaxSpeed - Default - Default - + @@ -358,6 +338,7 @@ + diff --git a/native.vcxproj.filters b/native.vcxproj.filters index af6190b45d..da9f3d708f 100644 --- a/native.vcxproj.filters +++ b/native.vcxproj.filters @@ -251,6 +251,9 @@ file + + thread + @@ -444,6 +447,9 @@ file + + thread + diff --git a/thread/threadpool.cpp b/thread/threadpool.cpp new file mode 100644 index 0000000000..86794c6faa --- /dev/null +++ b/thread/threadpool.cpp @@ -0,0 +1,74 @@ +#include "threadpool.h" + +///////////////////////////// WorkerThread + +WorkerThread::WorkerThread() : active(true), started(false) { + thread = new std::thread([&]() { WorkFunc(); }); + doneMutex.lock(); + while(!started) { }; +} + +WorkerThread::~WorkerThread() { + mutex.lock(); + active = false; + signal.notify_one(); + mutex.unlock(); + thread->join(); + delete thread; +} + +void WorkerThread::Process(const std::function& work) { + mutex.lock(); + work_ = work; + signal.notify_one(); + mutex.unlock(); +} + +void WorkerThread::WaitForCompletion() { + done.wait(doneMutex); +} + +void WorkerThread::WorkFunc() { + mutex.lock(); + started = true; + while(active) { + signal.wait(mutex); + if(active) work_(); + doneMutex.lock(); + done.notify_one(); + doneMutex.unlock(); + } +} + +///////////////////////////// ThreadPool + +ThreadPool::ThreadPool(int numThreads) : numThreads(numThreads), workersStarted(false) { +} + +void ThreadPool::StartWorkers() { + if(!workersStarted) { + for(int i=0; i()); + } + workersStarted = true; + } +} + +void ThreadPool::ParallelLoop(std::function loop, int lower, int upper) { + StartWorkers(); + int range = upper-lower; + if(range >= numThreads*2) { // don't parallelize tiny loops + // could do slightly better load balancing for the generic case, + // but doesn't matter since all our loops are power of 2 + int chunk = range/numThreads; + for(int s=lower, i=0; iProcess(std::bind(loop, s, std::min(s+chunk,upper))); + } + for(int i=0; iWaitForCompletion(); + } + } else { + loop(lower, upper); + } +} + diff --git a/thread/threadpool.h b/thread/threadpool.h new file mode 100644 index 0000000000..19b563c9d7 --- /dev/null +++ b/thread/threadpool.h @@ -0,0 +1,57 @@ +#pragma once + +#include "thread.h" +#include "base/mutex.h" + +#include +#include + +// This is the simplest possible worker implementation I can think of +// but entirely sufficient for the given purpose. +// Only handles a single item of work at a time. +class WorkerThread { +public: + WorkerThread(); + ~WorkerThread(); + + // submit a new work item + void Process(const std::function& work); + // wait for a submitted work item to be completed + void WaitForCompletion(); + +private: + 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(); + + WorkerThread(const WorkerThread& other); // prevent copies + void operator =(const WorkerThread &other); +}; + +// A thread pool manages a set of worker threads, and allows the execution of parallel loops on them +// individual parallel loops are sequentialized, which should not be a problem as they should each +// use the entire system +class ThreadPool { +public: + ThreadPool(int numThreads); + // 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); + +private: + const int numThreads; + 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); +};