GUI: per-chan osc multi-threading!

src/engine/workPool.cpp

@@ -17,4 +17,136 @@
  * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  */
 
- #include "workPool.h"
+#include "workPool.h"
+#include "../ta-log.h"
+#include <thread>
+
+void* _workThread(void* inst) {
+  ((DivWorkThread*)inst)->run();
+  return NULL;
+}
+
+void DivWorkThread::run() {
+  std::unique_lock<std::mutex> unique(selfLock);
+  DivPendingTask task;
+
+  logV("running work thread");
+
+  while (true) {
+    lock.lock();
+    if (tasks.empty()) {
+      lock.unlock();
+      isBusy=false;
+      parent->notify.notify_one();
+      if (terminate) {
+        break;
+      }
+      notify.wait(unique);
+      continue;
+    } else {
+      task=tasks.front();
+      tasks.pop();
+      lock.unlock();
+
+      task.func(task.funcArg);
+
+      parent->busyCount--;
+      parent->notify.notify_one();
+    }
+  }
+}
+
+bool DivWorkThread::assign(const std::function<void(void*)>& what, void* arg) {
+  lock.lock();
+  if (tasks.size()>=30) {
+    lock.unlock();
+    return false;
+  }
+  tasks.push(DivPendingTask(what,arg));
+  parent->busyCount++;
+  parent->notify.notify_one();
+  isBusy=true;
+  lock.unlock();
+  notify.notify_one();
+  return true;
+}
+
+void DivWorkThread::wait() {
+  if (!isBusy) return;
+}
+
+bool DivWorkThread::busy() {
+  return isBusy;
+}
+
+void DivWorkThread::finish() {
+  lock.lock();
+  terminate=true;
+  lock.unlock();
+  notify.notify_one();
+  thread->join();
+}
+
+void DivWorkThread::init(DivWorkPool* p) {
+  parent=p;
+  thread=new std::thread(_workThread,this);
+}
+
+void DivWorkPool::push(const std::function<void(void*)>& what, void* arg) {
+  //logV("submitting work");
+  // if no work threads, just execute
+  if (!threaded) {
+    what(arg);
+    return;
+  }
+
+  if (pos>=count) pos=0;
+
+  for (unsigned int tryCount=0; tryCount<count; tryCount++) {
+    if (workThreads[pos++].assign(what,arg)) return;
+  }
+
+  // all threads are busy
+  logV("all busy");
+  what(arg);
+}
+
+bool DivWorkPool::busy() {
+  if (!threaded) return false;
+  for (unsigned int i=0; i<count; i++) {
+    if (workThreads[i].busy()) return true;
+  }
+  return false;
+}
+
+void DivWorkPool::wait() {
+  if (!threaded) return;
+  std::unique_lock<std::mutex> unique(selfLock);
+  while (busyCount!=0) {
+    notify.wait_for(unique,std::chrono::milliseconds(100));
+  }
+}
+
+DivWorkPool::DivWorkPool(unsigned int threads):
+  threaded(threads>0),
+  count(threads),
+  pos(0),
+  busyCount(0) {
+  if (threaded) {
+    workThreads=new DivWorkThread[threads];
+    for (unsigned int i=0; i<count; i++) {
+      workThreads[i].init(this);
+    }
+  } else {
+    workThreads=NULL;
+  }
+}
+
+DivWorkPool::~DivWorkPool() {
+  if (threaded) {
+    for (unsigned int i=0; i<count; i++) {
+      workThreads[i].finish();
+    }
+    delete[] workThreads;
+  }
+}

src/engine/workPool.h

@@ -22,18 +22,46 @@
 
 #include <thread>
 #include <mutex>
+#include <atomic>
 #include <functional>
 #include <condition_variable>
 
+#include "fixedQueue.h"
+
+class DivWorkPool;
+
+struct DivPendingTask {
+  std::function<void(void*)> func;
+  void* funcArg;
+  DivPendingTask(std::function<void(void*)> f, void* arg):
+    func(f),
+    funcArg(arg) {}
+  DivPendingTask():
+    func(NULL),
+    funcArg(NULL) {}
+};
+
 struct DivWorkThread {
+  DivWorkPool* parent;
   std::mutex lock;
+  std::mutex selfLock;
   std::thread* thread;
   std::condition_variable notify;
-  bool busy, terminate;
+  FixedQueue<DivPendingTask,32> tasks;
+  std::atomic<bool> isBusy;
+  bool terminate;
 
   void run();
+  bool assign(const std::function<void(void*)>& what, void* arg);
+  void wait();
+  bool busy();
+  void finish();
+
+  void init(DivWorkPool* p);
   DivWorkThread():
-    busy(false) {}
+    parent(NULL),
+    isBusy(false),
+    terminate(false) {}
 };
 
 /**
@@ -41,13 +69,20 @@ struct DivWorkThread {
  * it is highly recommended to use `new` when allocating a DivWorkPool.
  */
 class DivWorkPool {
+  bool threaded;
+  std::mutex selfLock;
+  unsigned int count;
+  unsigned int pos;
   DivWorkThread* workThreads;
   public:
+    std::condition_variable notify;
+    std::atomic<int> busyCount;
+    
     /**
      * push a new job to this work pool.
      * if all work threads are busy, this will block until one is free.
      */
-    bool push();
+    void push(const std::function<void(void*)>& what, void* arg);
     
     /**
      * check whether this work pool is busy.
@@ -57,7 +92,7 @@ class DivWorkPool {
     /**
      * wait for all work threads to finish.
      */
-    bool wait();
+    void wait();
 
     DivWorkPool(unsigned int threads=0);
     ~DivWorkPool();