Merging Timer class as well as std::promise

based (blocking) ThreadPool::Finish, and also a better ParallelFor.
See merged commit logs for details.

Author: Chris Sullivan

README.md

@@ -4,26 +4,37 @@ 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);
 
-        cin.get();
+        // cold start for timing purposes
+        pool.ParallelFor(0,N,scale,a,b);
+
+        int ntrials = 10;
+        double tperformance = 0.0;
+        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";
+
+
+
         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,43 @@
 
 #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;
+        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";
+
+
+
         return 0;
 }

include/ThreadPool.hh

@@ -1,10 +1,13 @@
 #ifndef THREADPOOL_H
 #define THREADPOOL_H
 
-#include <thread>
+#include <iostream>
 #include <chrono>
+#include <thread>
 #include <functional>
 #include <vector>
+#include <utility>
+#include <future>
 
 #include "ThreadsafeQueue.hh"
 
@@ -17,50 +20,57 @@ public:
         ThreadPool(uint32_t numthreads);
         ~ThreadPool();
 
+        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) {
+                        m_promises.emplace_back();
+                        int mypromise = m_promises.size() - 1;
                         m_taskQueue.push([=]{
-                                int threadstart = begin + i*chunk;
-                                int threadstop = (i == m_nthreads - 1) ? end : threadstart + chunk;
-                                for (int it = threadstart; it < threadstop; ++it) {
-                                        ClassFunction::Serial(it);
-                                }
-                        });
+                                        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;
+        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_kernel) class name { public: static void Serial(const int& i) { function_kernel; } };
-
 #endif /* end of include guard: THREADPOOL_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

@@ -27,3 +27,7 @@ void ThreadPool::Worker() {
 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();
+}