Added blocking ThreadPool::Finish() which waits until workers finish the task queue.

Added a scoped Timer class for performance measurements. Also added variadic template ParallelFor function
which accepts any callable type and forwards the parameter pack to the callable type. At the moment the
variadic parameter pack is captured by value, capture by reference would be much nicer, but unfortunately
C++14 lambdas can't yet capture by move or forward semantics. I understand that this can possibly be handled
by creating a wraper class that holds something like a std::ref, but I haven't dug into it much yet, since
the parameters only are copied once for each task, and so the overhead is not too large. But something to
add to the TODO list.

Author: Chris Sullivan

example.cc

@@ -1,5 +1,8 @@
 
 #include "ThreadPool.hh"
+#include "Timer.hh"
+
+#include <iostream>
 
 static double* a;
 static double* b;
@@ -13,17 +16,41 @@ void raw_scale(int i, double* a, double* b) {
 
 int main () {
 
-        ThreadPool pool(4);
+        ThreadPool pool(8);
 
-        int N = 1e7;
+        int N = 1e9;
         a = (double*)calloc(N,sizeof(double));
         b = (double*)calloc(N,sizeof(double));
         for (int i=0; i<N; i++) { b[i] = i; }
 
         SERIAL_OPERATION(static_scale, a[i]=4*b[i]);
-        pool.ParallelFor<static_scale>(0,N);
-
-        pool.ParallelFor(0,N,raw_scale,a,b);
+        {
+                // This is a cold start loop, it is not timed
+                pool.ParallelFor<static_scale>(0,N);
+        }
+
+        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<static_scale>(0,N);
+        }
+        cout << "Average: " << tperformance*1e-6 / ntrials << " ms\n\n";
+
+        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,raw_scale,a,b);
+        }
+        cout << "Average: " << tperformance*1e-6 / ntrials << " ms\n\n";
 
 
         cin.get();

include/ThreadPool.hh

@@ -9,6 +9,8 @@
 #include <mutex>
 #include <condition_variable>
 #include <utility>
+#include <atomic>
+#include <future>
 
 using namespace std;
 
@@ -26,42 +28,57 @@ public:
 
                 int chunk = (end - begin) / m_nthreads;
                 for (int i = 0; i < m_nthreads; ++i) {
+                        m_promises.emplace_back();
+                        int mypromise = m_promises.size() - 1;
                         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);
+                                                ClassFunction::SerialFunction(it);
                                         }
+                                        m_promises[mypromise].set_value();
                                 });
                 }
+                Finish();
         }
         template <typename T, typename... Params>
-        void ParallelFor(uint32_t begin, uint32_t end, T func, Params&&... 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;
                         Enqueue([=]{
                                         uint32_t threadstart = begin + i*chunk;
                                         uint32_t threadstop = (i == m_nthreads - 1) ? end : threadstart + chunk;
                                         for (uint32_t it = threadstart; it < threadstop; ++it) {
-                                                func(it, params...);
+                                                SerialFunction(it, params...);
                                         }
+                                        m_promises[mypromise].set_value();
                                 });
                 }
+                Finish();
+        }
+
+        void Finish() {
+                for (auto& promise : m_promises) promise.get_future().get();
+                m_promises.clear();
         }
 
 private:
         // threads and task queue
         int m_nthreads;
         vector<thread> m_workers;
         queue<function<void()>> m_tasks;
+        vector<promise<void>> m_promises;
 
         // thread synchronization members
         bool stop_workers;
         condition_variable m_condition;
         mutable mutex m_taskmutex;
+        atomic<int> m_nrunning;
 
 };
 
 // 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; } };
+#define SERIAL_OPERATION(name, function_kernal) class name { public: static void SerialFunction(const int& i) { function_kernal; } };

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,5 +1,6 @@
 #include "ThreadPool.hh"
-ThreadPool::ThreadPool(uint32_t numthreads) : m_nthreads(numthreads), stop_workers(false) {
+ThreadPool::ThreadPool(uint32_t numthreads) : m_nthreads(numthreads), stop_workers(false),
+                                              m_nrunning(0) {
 
         for (uint32_t i=0; i<numthreads;i++) {
                 m_workers.emplace_back(&ThreadPool::Worker, this);
@@ -11,6 +12,7 @@ ThreadPool::~ThreadPool() {
         m_condition.notify_all();
         JoinAll();
 }
+// TODO: add futures and promises for timing purposes
 void ThreadPool::Enqueue(function<void()> task) {
         {
                 unique_lock<mutex> lock(m_taskmutex);
@@ -23,12 +25,12 @@ void ThreadPool::Worker() {
         while(true) {
                 {
                         unique_lock<mutex> lock(m_taskmutex);
-                        cout << "Worker waiting..." << endl;
+                        //cout << "Worker waiting..." << endl;
                         while(!stop_workers && m_tasks.empty()) {
                                 m_condition.wait(lock);
                         }
                         if (stop_workers) { return; }
-                        cout << "Booting up..." << endl;
+                        //cout << "Booting up..." << endl;
                         work = m_tasks.front();
                         m_tasks.pop();
                 }