Merge branch 'lockfree_merge' into lockfree

Author: Chris Sullivan

include/LockfreeThreadPool.hh

@@ -0,0 +1,78 @@
+#ifndef LOCKFREETHREADPOOL_H
+#define LOCKFREETHREADPOOL_H
+
+#include <iostream>
+#include <chrono>
+#include <thread>
+#include <functional>
+#include <vector>
+#include <utility>
+#include <future>
+
+#include "concurrentqueue.h"
+
+using namespace std;
+using namespace moodycamel;
+
+typedef std::function<void()> WorkType;
+
+class LockfreeThreadPool {
+public:
+        LockfreeThreadPool(uint32_t numthreads);
+        ~LockfreeThreadPool();
+
+        void Enqueue(function<void()> task);
+        void Worker();
+        void JoinAll();
+        void Finish();
+
+        template <typename T, typename... Params>
+        void ParallelFor(uint32_t begin, uint32_t end, int32_t n_tasks, T SerialFunction, Params&&... params) {
+
+                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();
+        }
+        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.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();
+                });
+            }
+            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;
+};
+
+#endif /* end of include guard: THREADPOOL_H */

include/ThreadPool.hh

@@ -9,10 +9,9 @@
 #include <utility>
 #include <future>
 
-#include "concurrentqueue.h"
+#include "ThreadsafeQueue.hh"
 
 using namespace std;
-using namespace moodycamel;
 
 typedef std::function<void()> WorkType;
 
@@ -27,23 +26,22 @@ public:
         void Finish();
 
         template <typename T, typename... Params>
-        void ParallelFor(uint32_t begin, uint32_t end, T SerialFunction, Params&&... params) {
+        void ParallelFor(uint32_t begin, uint32_t end, int32_t n_tasks, T SerialFunction, Params&&... params) {
 
-                int chunk = (end - begin) / m_nthreads;
-                for (int i = 0; i < m_nthreads; ++i) {
+                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([=]{
+                        m_taskQueue.push([=]{
                                         uint32_t threadstart = begin + i*chunk;
-                                        uint32_t threadstop = (i == m_nthreads - 1) ? end : threadstart + 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();
                                 });
                 }
-                WorkType work;
-                if (m_taskQueue.try_dequeue(work)) work(); // master thread is also a worker
                 Finish();
         }
         template<typename InputIt, typename T>
@@ -53,7 +51,7 @@ public:
             for (int i = 0; i < m_nthreads; i++) {
                 m_promises.emplace_back();
                 int mypromise = m_promises.size() - 1;
-                m_taskQueue.enqueue([=]{
+                m_taskQueue.push([=]{
                     InputIt threadBegin = begin + i*chunkSize;
                     InputIt threadOutput = outputBegin + i*chunkSize;
                     InputIt threadEnd = (i == m_nthreads - 1) ? end : threadBegin + chunkSize;
@@ -63,8 +61,6 @@ public:
                     m_promises[mypromise].set_value();
                 });
             }
-            WorkType work;
-            if (m_taskQueue.try_dequeue(work)) work(); // master thread is also a worker
             Finish();
         }
 
@@ -75,7 +71,7 @@ private:
         vector<promise<void>> m_promises;
         bool m_stopWorkers;
 
-        ConcurrentQueue<WorkType> m_taskQueue;
+        ThreadsafeQueue<WorkType> m_taskQueue;
 };
 
 #endif /* end of include guard: THREADPOOL_H */

src/LockfreeThreadPool.cc

@@ -0,0 +1,26 @@
+#include "LockfreeThreadPool.hh"
+
+LockfreeThreadPool::LockfreeThreadPool(uint32_t numthreads) : m_nthreads(numthreads), m_stopWorkers(false) {
+    for (uint32_t i=0; i<numthreads;i++) {
+        m_workers.emplace_back(&LockfreeThreadPool::Worker, this);
+    }
+}
+LockfreeThreadPool::~LockfreeThreadPool() {
+    m_stopWorkers = true;
+    JoinAll();
+}
+
+void LockfreeThreadPool::Worker() {
+    while(true) {
+        function<void()> work;
+        if (m_taskQueue.try_dequeue(work)) work();
+        if (m_stopWorkers) return;
+    }
+}
+void LockfreeThreadPool::JoinAll() {
+    for (auto& worker : m_workers) { worker.join(); }
+}
+void LockfreeThreadPool::Finish() {
+    for (auto& promise : m_promises) promise.get_future().get();
+    m_promises.clear();
+}

src/ThreadPool.cc

@@ -1,20 +1,28 @@
 #include "ThreadPool.hh"
 
 ThreadPool::ThreadPool(uint32_t numthreads) : m_nthreads(numthreads), m_stopWorkers(false) {
-    for (uint32_t i=0; i<numthreads-1;i++) { // -1 b/c main thread is also a worker
+    //for (uint32_t i=0; i<numthreads-1;i++) { // -1 b/c main thread is also a worker
+    for (uint32_t i=0; i<numthreads;i++) {
         m_workers.emplace_back(&ThreadPool::Worker, this);
     }
 }
 ThreadPool::~ThreadPool() {
     m_stopWorkers = true;
+    m_taskQueue.join();
     JoinAll();
 }
 
 void ThreadPool::Worker() {
     while(true) {
         function<void()> work;
-        if (m_taskQueue.try_dequeue(work)) work();
-        if (m_stopWorkers) return;
+        try {
+            work = m_taskQueue.pop();
+            work();
+        }
+        catch (const ThreadsafeQueue<WorkType>::QueueFinished&)
+        {
+            return;
+        }
     }
 }
 void ThreadPool::JoinAll() {