Initial Commit

ThreadPool.cpp

@@ -0,0 +1,143 @@
+#include "thread_pool.h"
+using namespace std;
+
+void _global_thread_worker_main(ThreadWorkerData* ptdd)
+{
+    unique_lock<mutex> ulk(ptdd->m);
+    while(ptdd->running)
+    {
+        ptdd->started=true;
+        ptdd->func();
+        ptdd->finished=true;
+        ptdd->cond.wait(ulk); /// wait for new work. (will unlock ptdd->m while waiting...)
+    }
+}
+
+ThreadPool::ThreadPool(int n) : left(0),busy(0),total(n)
+{
+
+}
+
+ThreadPool::~ThreadPool()
+{
+    for(int i=0;i<left+busy;i++)
+    {
+        //cout<<"Waiting... thread "<<tvec[i]->get_id()<<endl;
+
+        /// wait for it finish.
+        while(true)
+        {
+            dvec[i]->m.lock();
+            if(dvec[i]->finished)
+            {
+                dvec[i]->m.unlock();
+                break;
+            }
+            else
+            {
+                dvec[i]->m.unlock();
+                this_thread::sleep_for(chrono::seconds(1));
+            }
+        }
+
+        dvec[i]->running=false;
+        /// notify it
+        dvec[i]->cond.notify_all();
+        /// wait for it finish.
+        tvec[i]->join();
+        /// release resource
+        delete dvec[i];
+        dvec[i]=nullptr;
+        delete tvec[i];
+        tvec[i]=nullptr;
+    }
+}
+
+int ThreadPool::start(const function<void()>& func)
+{
+    if(left>0)
+    {
+        bool done=false;
+        int idx=-1;
+
+        for(int i=0;i<left+busy;i++)
+        {
+            if(!dvec[i]->m.try_lock())
+            {
+                /// Failed to lock.
+                continue;
+            }
+            if(dvec[i]->finished)
+            {
+                /// Found a finished work! Now we reuse it.
+                dvec[i]->func=func;
+                dvec[i]->started=false;
+                dvec[i]->finished=false;
+                dvec[i]->running=true;
+                dvec[i]->cond.notify_all();
+
+                done=true;
+                idx=i;
+
+                dvec[i]->m.unlock();
+                break;
+            }
+            else
+            {
+                dvec[i]->m.unlock();
+            }
+        }
+
+        if(done)
+        {
+            left--;
+            busy++;
+            return idx;
+        }
+        else
+        {
+            return -2;
+        }
+    }
+    else
+    {
+        if(busy<total)
+        {
+            /// Still can create thread
+            ThreadWorkerData* ptdd=new ThreadWorkerData;
+            ptdd->func=func;
+            ptdd->started=false;
+            ptdd->finished=false;
+            ptdd->running=true;
+            dvec.push_back(ptdd);
+
+            thread* t=new thread(_global_thread_worker_main,ptdd);
+            tvec.push_back(t);
+
+            busy++;
+
+            return tvec.size()-1;
+        }
+        else
+        {
+            /// Check if we can reuse finished thread
+            for(int i=0;i<total;i++)
+            {
+                if(!dvec[i]->m.try_lock())
+                {
+                    /// Failed to lock. continue.
+                    continue;
+                }
+                if(dvec[i]->finished)
+                {
+                    left++;
+                    busy--;
+                }
+                dvec[i]->m.unlock();
+            }
+
+            if(left>0) return start(func);
+            else return -1;
+        }
+    }
+}

ThreadPool.h

@@ -0,0 +1,34 @@
+#pragma once
+#include <thread>
+#include <functional>
+#include <vector>
+#include <mutex>
+#include <condition_variable>
+
+struct ThreadWorkerData
+{
+    std::function<void()> func;
+    std::mutex m;
+    std::condition_variable cond;
+    bool started;
+    bool finished;
+    bool running;
+};
+
+void _global_thread_worker_main(ThreadWorkerData* ptdd);
+
+
+class ThreadPool
+{
+public:
+    ThreadPool(int n);
+    ~ThreadPool();
+
+    int start(const std::function<void()>& func);
+private:
+    std::vector<ThreadWorkerData*> dvec;
+    std::vector<std::thread*> tvec;
+    int left;
+    int busy;
+    int total;
+};