Refactored thread pools with base classes.

- Created ThreadPoolBase class
- Inherited from it for LockfreeThreadPool and ThreadPool
- Removed ParallelMap since we weren't using it

Author: Joshua Bradt

include/LockfreeThreadPool.hh

@@ -9,69 +9,44 @@
 #include <utility>
 #include <future>
 
+#include "ThreadPoolBase.hh"
 #include "concurrentqueue.h"
 
 using namespace std;
 using namespace moodycamel;
 
 typedef std::function<void()> WorkType;
 
-class LockfreeThreadPool {
+class LockfreeThreadPool : public ThreadPoolBase {
 public:
-        LockfreeThreadPool(uint32_t numthreads);
-        ~LockfreeThreadPool();
 
-        void Enqueue(function<void()> task);
-        void Worker();
-        void JoinAll();
-        void Finish();
+        LockfreeThreadPool(uint32_t num_threads)
+        {
+            for (uint32_t i=0; i<num_threads;i++) {
+                m_workers.emplace_back(&LockfreeThreadPool::Worker, this);
+            }
+        }
 
-        template <typename T, typename... Params>
-        void ParallelFor(uint32_t begin, uint32_t end, int32_t n_tasks, T SerialFunction, Params&&... params) {
+        ~LockfreeThreadPool() {
+            CleanUp();
+            JoinAll();
+        }
 
-                n_tasks = (n_tasks >= m_nthreads) ? n_tasks : m_nthreads;
-                int chunk = (end - begin) / n_tasks;
-                for (int i = 0; i < n_tasks; ++i) {
-                        m_promises.emplace_back();
-                        int mypromise = m_promises.size() - 1;
-                        m_taskQueue.enqueue([=]{
-                                        uint32_t threadstart = begin + i*chunk;
-                                        uint32_t threadstop = (i == n_tasks - 1) ? end : threadstart + chunk;
-                                        for (uint32_t it = threadstart; it < threadstop; ++it) {
-                                                SerialFunction(it, params...);
-                                        }
-                                        m_promises[mypromise].set_value();
-                                });
-                }
-                Finish();
+        void AddTask(WorkType task) override
+        {
+            m_taskQueue.enqueue(task);
         }
-        template<typename InputIt, typename T>
-        void ParallelMap(InputIt begin, InputIt end, InputIt outputBegin, const std::function<T(T)>& func)
+
+        void Worker() override
         {
-            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.enqueue([=]{
-                    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();
-                });
+            while(true) {
+                function<void()> work;
+                if (m_taskQueue.try_dequeue(work)) work();
+                if (m_stopWorkers) return;
             }
-            Finish();
         }
 
 private:
-        // threads and task queue
-        int m_nthreads;
-        vector<thread> m_workers;
-        vector<promise<void>> m_promises;
-        bool m_stopWorkers;
-
         ConcurrentQueue<WorkType> m_taskQueue;
 };
 

include/ThreadPool.hh

@@ -9,68 +9,55 @@
 #include <utility>
 #include <future>
 
+#include "ThreadPoolBase.hh"
+
 #include "ThreadsafeQueue.hh"
 
 using namespace std;
 
 typedef std::function<void()> WorkType;
 
-class ThreadPool {
+class ThreadPool : public ThreadPoolBase {
 public:
-        ThreadPool(uint32_t numthreads);
-        ~ThreadPool();
 
-        void Enqueue(function<void()> task);
-        void Worker();
-        void JoinAll();
-        void Finish();
+        ThreadPool(uint32_t num_threads)
+        {
+            for (uint32_t i=0; i<num_threads;i++) {
+                m_workers.emplace_back(&ThreadPool::Worker, this);
+            }
+        }
 
-        template <typename T, typename... Params>
-        void ParallelFor(uint32_t begin, uint32_t end, int32_t n_tasks, T SerialFunction, Params&&... params) {
+        ~ThreadPool() {
+            CleanUp();
+            JoinAll();
+        }
 
-                n_tasks = (n_tasks >= m_nthreads) ? n_tasks : m_nthreads;
-                int chunk = (end - begin) / n_tasks;
-                for (int i = 0; i < n_tasks; ++i) {
-                        m_promises.emplace_back();
-                        int mypromise = m_promises.size() - 1;
-                        m_taskQueue.push([=]{
-                                        uint32_t threadstart = begin + i*chunk;
-                                        uint32_t threadstop = (i == n_tasks - 1) ? end : threadstart + chunk;
-                                        for (uint32_t it = threadstart; it < threadstop; ++it) {
-                                                SerialFunction(it, params...);
-                                        }
-                                        m_promises[mypromise].set_value();
-                                });
-                }
-                Finish();
+        void AddTask(WorkType task) override
+        {
+            m_taskQueue.push(task);
         }
-        template<typename InputIt, typename T>
-        void ParallelMap(InputIt begin, InputIt end, InputIt outputBegin, const std::function<T(T)>& func)
+
+        void CleanUp() override
         {
-            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();
-                });
+            m_stopWorkers = true;
+            m_taskQueue.join();
+        }
+
+        void Worker() override {
+            while(true) {
+                WorkType work;
+                try {
+                    work = m_taskQueue.pop();
+                    work();
+                }
+                catch (const ThreadsafeQueue<WorkType>::QueueFinished&)
+                {
+                    return;
+                }
             }
-            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;
 };
 

include/ThreadPoolBase.hh

@@ -0,0 +1,53 @@
+#ifndef THREADPOOLBASE_H
+#define THREADPOOLBASE_H
+
+#include <iostream>
+#include <chrono>
+#include <thread>
+#include <functional>
+#include <vector>
+#include <utility>
+#include <future>
+
+using namespace std;
+
+typedef std::function<void()> WorkType;
+
+class ThreadPoolBase {
+public:
+        ThreadPoolBase() = default;
+
+        virtual void AddTask(function<void()> task) = 0;
+
+        virtual void Worker() = 0;
+        virtual void CleanUp() { m_stopWorkers = true; }
+        void JoinAll();
+        void Finish();
+
+        template <typename T, typename... Params>
+        void ParallelFor(uint32_t begin, uint32_t end, uint32_t n_tasks, T SerialFunction, Params&&... params) {
+
+                n_tasks = (n_tasks >= m_workers.size()) ? n_tasks : m_workers.size();
+                m_tasksRemaining = n_tasks;
+                int chunk = (end - begin) / n_tasks;
+                for (int i = 0; i < n_tasks; ++i) {
+                        AddTask([=]{
+                                        uint32_t threadstart = begin + i*chunk;
+                                        uint32_t threadstop = (i == n_tasks - 1) ? end : threadstart + chunk;
+                                        for (uint32_t it = threadstart; it < threadstop; ++it) {
+                                                SerialFunction(it, params...);
+                                        }
+                                        m_tasksRemaining--;
+                                });
+                }
+                Finish();
+        }
+
+protected:
+        // threads and task queue
+        vector<thread> m_workers;
+        atomic<int> m_tasksRemaining;
+        bool m_stopWorkers;
+};
+
+#endif /* end of include guard: THREADPOOL_H */

src/LockfreeThreadPool.cc

@@ -0,0 +1,8 @@
+#include "ThreadPoolBase.hh"
+
+void ThreadPoolBase::JoinAll() {
+    for (auto& worker : m_workers) { worker.join(); }
+}
+void ThreadPoolBase::Finish() {
+    while (m_tasksRemaining > 0) {}
+}