Merge pull request #1 from ExtremeScale/master

Adding ParallelFor templated over any callable type with variadic templated arguments.
Credits: C. Sullivan and J. Bradt

Author: Chris Sullivan

CMakeLists.txt

@@ -0,0 +1,14 @@
+project(ThreadPool)
+cmake_minimum_required(VERSION 3.1)
+
+set(CMAKE_CXX_STANDARD 11)
+
+include_directories(include)
+add_definitions(-Wall -Wextra -pedantic)
+
+add_library(ThreadPool SHARED src/ThreadPool.cc)
+install(TARGETS ThreadPool LIBRARY DESTINATION lib)
+install(FILES include/ThreadPool.hh DESTINATION include)
+
+add_executable(test_ThreadPool test/test_ParallelMap.cc test/catch_main.cc)
+target_link_libraries(test_ThreadPool ThreadPool)

README.md

@@ -4,26 +4,25 @@ A simple std::thread based thread pool with a parallel-for loop implementation.
 
 Example of how to perform STREAM benchmark scale operation in parallel:
 ```C++
-#include "ThreadPool.hh"
-
-static double* a;
-static double* b;
-
-using namespace std;
+void scale(int i, double* a, double* b) {
+        a[i]=4*b[i];
+}
 
 int main () {
 
-        ThreadPool pool(4);
+        ThreadPool pool(8);
 
-        int N = 1e7;
-        a = (double*)calloc(N,sizeof(double));
-        b = (double*)calloc(N,sizeof(double));
+        int N = 1e9;
+        auto a = (double*)calloc(N,sizeof(double));
+        auto b = (double*)calloc(N,sizeof(double));
         for (int i=0; i<N; i++) { b[i] = i; }
 
-        SERIAL_OPERATION(scale, a[i]=4*b[i]);
-        pool.ParallelFor<scale>(0,N);
+        {
+            Timer timer([&](int elapsed) { cout << elapsed*1e-6 << " ms\n"; });
+            pool.ParallelFor(0,N,scale,a,b);
+        }
+
 
-        cin.get();
         return 0;
 }
 

default.inc

@@ -64,7 +64,7 @@ CFLAGS_EXTRA =
 
 # Mandatory arguments to the C++ compiler.  These arguments will be
 # passed even if CXXFLAGS has been overridden by command-line arguments.
-CXXFLAGS_EXTRA = -std=c++11
+CXXFLAGS_EXTRA = -std=c++1y
 
 # Mandatory arguments to the linker, before the listing of object
 # files.  These arguments will be passed even if LDFLAGS has been

example.cc

@@ -1,23 +1,54 @@
 
 #include "ThreadPool.hh"
+#include "Timer.hh"
+
+#include <iostream>
 
-static double* a;
-static double* b;
 
 using namespace std;
 
+
+void scale(int i, double* a, double* b) {
+        a[i]=4*b[i];
+}
+
 int main () {
 
-        ThreadPool pool(4);
+        ThreadPool pool(8);
 
-        int N = 1e7;
-        a = (double*)calloc(N,sizeof(double));
-        b = (double*)calloc(N,sizeof(double));
+        int N = 1e9;
+        auto a = (double*)calloc(N,sizeof(double));
+        auto b = (double*)calloc(N,sizeof(double));
         for (int i=0; i<N; i++) { b[i] = i; }
 
-        SERIAL_OPERATION(scale, a[i]=4*b[i]);
-        pool.ParallelFor<scale>(0,N);
 
-        cin.get();
+        // cold start for timing purposes
+        pool.ParallelFor(0,N,scale,a,b);
+
+        int ntrials = 10;
+        double tperformance = 0.0;
+        cout << "Callable: c-function pointer\n";
+        for (int i=0; i<ntrials; i++)
+        {
+                Timer timer([&](int elapsed){
+                                cout << "Trial " << i << ": "<< elapsed*1e-6 << " ms\n";
+                                tperformance+=elapsed;
+                        });
+                pool.ParallelFor(0,N,scale,a,b);
+        }
+        cout << "Average: " << tperformance*1e-6 / ntrials << " ms\n\n";
+
+        tperformance = 0.0;
+        cout << "Callable: lambda function (without capture) \n";
+        for (int i=0; i<ntrials; i++)
+        {
+                Timer timer([&](int elapsed){
+                                cout << "Trial " << i << ": "<< elapsed*1e-6 << " ms\n";
+                                tperformance+=elapsed;
+                        });
+                pool.ParallelFor(0,N,[](int k, double* a, double* b) {return a[k] = 4*b[k];},a,b);
+        }
+        cout << "Average: " << tperformance*1e-6 / ntrials << " ms\n\n";
+
         return 0;
 }

include/ThreadPool.hh

@@ -1,16 +1,20 @@
-#include <thread>
+#ifndef THREADPOOL_H
+#define THREADPOOL_H
+
 #include <iostream>
-#include <string>
 #include <chrono>
 #include <thread>
 #include <functional>
 #include <vector>
-#include <queue>
-#include <mutex>
-#include <condition_variable>
+#include <utility>
+#include <future>
+
+#include "ThreadsafeQueue.hh"
 
 using namespace std;
 
+typedef std::function<void()> WorkType;
+
 class ThreadPool {
 public:
         ThreadPool(uint32_t numthreads);
@@ -19,34 +23,54 @@ public:
         void Enqueue(function<void()> task);
         void Worker();
         void JoinAll();
+        void Finish();
 
-        template <typename ClassFunction>
-        void ParallelFor(uint32_t begin, uint32_t end) {
+        template <typename T, typename... Params>
+        void ParallelFor(uint32_t begin, uint32_t end, T SerialFunction, Params&&... params) {
 
                 int chunk = (end - begin) / m_nthreads;
                 for (int i = 0; i < m_nthreads; ++i) {
-                        Enqueue([=]{
-                                        int threadstart = begin + i*chunk;
-                                        int threadstop = (i == m_nthreads - 1) ? end : threadstart + chunk;
-                                        for (int it = threadstart; it < threadstop; ++it) {
-                                                ClassFunction::Serial(it);
+                        m_promises.emplace_back();
+                        int mypromise = m_promises.size() - 1;
+                        m_taskQueue.push([=]{
+                                        uint32_t threadstart = begin + i*chunk;
+                                        uint32_t threadstop = (i == m_nthreads - 1) ? end : threadstart + chunk;
+                                        for (uint32_t it = threadstart; it < threadstop; ++it) {
+                                                SerialFunction(it, params...);
                                         }
+                                        m_promises[mypromise].set_value();
                                 });
                 }
+                Finish();
+        }
+        template<typename InputIt, typename T>
+        void ParallelMap(InputIt begin, InputIt end, InputIt outputBegin, const std::function<T(T)>& func)
+        {
+            int chunkSize = (end - begin) / m_nthreads;
+            for (int i = 0; i < m_nthreads; i++) {
+                m_promises.emplace_back();
+                int mypromise = m_promises.size() - 1;
+                m_taskQueue.push([=]{
+                    InputIt threadBegin = begin + i*chunkSize;
+                    InputIt threadOutput = outputBegin + i*chunkSize;
+                    InputIt threadEnd = (i == m_nthreads - 1) ? end : threadBegin + chunkSize;
+                    while (threadBegin != threadEnd) {
+                        *(threadOutput++) = func(*(threadBegin++));
+                    }
+                    m_promises[mypromise].set_value();
+                });
+            }
+            Finish();
         }
 
 private:
         // threads and task queue
         int m_nthreads;
         vector<thread> m_workers;
-        queue<function<void()>> m_tasks;
-
-        // thread synchronization members
-        bool stop_workers;
-        condition_variable m_condition;
-        mutable mutex m_taskmutex;
+        vector<promise<void>> m_promises;
+        bool m_stopWorkers;
 
+        ThreadsafeQueue<WorkType> m_taskQueue;
 };
 
-// Macro to define a static class function which can be called via ThreadPool::ParallelFor<T>
-#define SERIAL_OPERATION(name, function_kernal) class name { public: static void Serial(const int& i) { function_kernal; } };
+#endif /* end of include guard: THREADPOOL_H */

include/ThreadsafeQueue.hh

@@ -0,0 +1,67 @@
+#ifndef THREADSAFEQUEUE_H
+#define THREADSAFEQUEUE_H
+
+#include <thread>
+#include <queue>
+#include <mutex>
+#include <condition_variable>
+
+template<typename T>
+class ThreadsafeQueue
+{
+public:
+    ThreadsafeQueue() : m_finished(false), m_maxSize(20) {}
+    ThreadsafeQueue(const size_t maxSize) : m_finished(false), m_maxSize(maxSize) {}
+    ThreadsafeQueue(const ThreadsafeQueue&) = delete;
+    ThreadsafeQueue(ThreadsafeQueue&&) = delete;
+
+    void push(const T& task)
+    {
+        std::unique_lock<std::mutex> lock(m_qmtx);
+        m_cond.wait(lock, [this]{ return m_contents.size() < m_maxSize; });
+        m_contents.push(task);
+        m_cond.notify_all();
+    }
+
+    void push(T&& task)
+    {
+        std::unique_lock<std::mutex> lock(m_qmtx);
+        m_cond.wait(lock, [this]{ return m_contents.size() < m_maxSize; });
+        m_contents.push(std::move(task));
+        m_cond.notify_all();
+    }
+
+    T pop()
+    {
+        std::unique_lock<std::mutex> lock(m_qmtx);
+        m_cond.wait(lock, [this]{ return !m_contents.empty() || m_finished; });
+        if (m_finished) throw QueueFinished();
+        T item = std::move(m_contents.front());
+        m_contents.pop();
+        m_cond.notify_all();  // I'm not sure if this can be called before the return...
+        return item;
+    }
+
+    void join()
+    {
+        std::unique_lock<std::mutex> lock(m_qmtx);
+        m_cond.wait(lock, [this]{ return m_contents.empty(); });
+        m_finished = true;
+        m_cond.notify_all();
+    }
+
+    class QueueFinished : public std::exception
+    {
+    public:
+        virtual const char* what() const noexcept { return "Queue has been joined"; };
+    };
+
+private:
+    std::queue<T> m_contents;
+    std::mutex m_qmtx;
+    std::condition_variable m_cond;
+    bool m_finished;
+    size_t m_maxSize;
+};
+
+#endif /* end of include guard: THREADSAFEQUEUE_H */

include/Timer.hh

@@ -0,0 +1,22 @@
+#ifndef TIMER_H
+#define TIMER_H
+#include <functional>
+#include <chrono>
+
+using namespace std;
+
+struct Timer {
+        Timer(function<void(int)> callback)
+                : callback(callback)
+                , t0(chrono::high_resolution_clock::now()) { ; }
+        ~Timer(void) {
+                auto  t1 = chrono::high_resolution_clock::now();
+                auto elapsed = chrono::duration_cast<chrono::nanoseconds>(t1-t0).count();
+                callback(elapsed);
+        }
+        function<void(int)> callback;
+        chrono::high_resolution_clock::time_point t0;
+
+};
+
+#endif

src/ThreadPool.cc

@@ -1,40 +1,33 @@
 #include "ThreadPool.hh"
-ThreadPool::ThreadPool(uint32_t numthreads) : m_nthreads(numthreads), stop_workers(false) {
 
+ThreadPool::ThreadPool(uint32_t numthreads) : m_nthreads(numthreads), m_stopWorkers(false) {
         for (uint32_t i=0; i<numthreads;i++) {
                 m_workers.emplace_back(&ThreadPool::Worker, this);
         }
 }
 ThreadPool::~ThreadPool() {
-
-        stop_workers = true;
-        m_condition.notify_all();
+        m_stopWorkers = true;
+        m_taskQueue.join();
         JoinAll();
 }
-void ThreadPool::Enqueue(function<void()> task) {
-        {
-                unique_lock<mutex> lock(m_taskmutex);
-                m_tasks.push(task);
-        }
-        m_condition.notify_one();
-}
+
 void ThreadPool::Worker() {
+    while(true) {
         function<void()> work;
-        while(true) {
-                {
-                        unique_lock<mutex> lock(m_taskmutex);
-                        cout << "Worker waiting..." << endl;
-                        while(!stop_workers && m_tasks.empty()) {
-                                m_condition.wait(lock);
-                        }
-                        if (stop_workers) { return; }
-                        cout << "Booting up..." << endl;
-                        work = m_tasks.front();
-                        m_tasks.pop();
-                }
-                work();
+        try {
+            work = m_taskQueue.pop();
+            work();
+        }
+        catch (const ThreadsafeQueue<WorkType>::QueueFinished&)
+        {
+            return;
         }
+    }
 }
 void ThreadPool::JoinAll() {
         for (auto& worker : m_workers) { worker.join(); }
 }
+void ThreadPool::Finish() {
+    for (auto& promise : m_promises) promise.get_future().get();
+    m_promises.clear();
+}