[PATCH v6 4/6] mt retrace: add workqueue support

[Imre Deak]

Signed-off-by: Imre Deak <imre.deak at intel.com>
---
 CMakeLists.txt             |    3 ++
 common/os_workqueue.hpp    |   49 +++++++++++++++++++
 common/workqueue_posix.cpp |  117 ++++++++++++++++++++++++++++++++++++++++++++
 common/workqueue_win32.cpp |   48 ++++++++++++++++++
 4 files changed, 217 insertions(+)
 create mode 100644 common/os_workqueue.hpp
 create mode 100644 common/workqueue_posix.cpp
 create mode 100644 common/workqueue_win32.cpp

diff --git a/CMakeLists.txt b/CMakeLists.txt
index 9fe2002..cea74f0 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -269,8 +269,10 @@ include_directories (
 if (WIN32)
     set (os os_win32.cpp)
     set (glws_os glws_wgl.cpp)
+    set (workqueue workqueue_win32.cpp)
 else ()
     set (os os_posix.cpp)
+    set (workqueue workqueue_posix.cpp)
     if (APPLE)
         set (glws_os glws_cocoa.mm)
     else ()
@@ -301,6 +303,7 @@ add_library (common STATIC
     common/image_pnm.cpp
     common/image_png.cpp
     common/${os}
+    common/${workqueue}
 )
 
 set_target_properties (common PROPERTIES
diff --git a/common/os_workqueue.hpp b/common/os_workqueue.hpp
new file mode 100644
index 0000000..e6b77d4
--- /dev/null
+++ b/common/os_workqueue.hpp
@@ -0,0 +1,49 @@
+#ifndef _OS_WORKQUEUE_HPP_
+#define _OS_WORKQUEUE_HPP_
+
+#include <queue>
+
+namespace os
+{
+
+class WorkQueue;
+
+class WorkQueueWork {
+protected:
+	friend class WorkQueue;
+
+public:
+	virtual void run(void) = 0;
+	virtual ~WorkQueueWork(void) { }
+};
+
+extern "C"
+void *WorkQueue__entry_thunk(void *data);
+
+class WorkQueue {
+	pthread_t handle;
+	std::queue<WorkQueueWork *> work_queue;
+
+	bool busy;
+	bool exit_workqueue;
+	pthread_cond_t wake_cond;
+	pthread_cond_t complete_cond;
+
+	pthread_mutex_t lock;
+
+	void wake_up_thread(void);
+	void thread_entry(void);
+	int run_tasks(void);
+	friend void *WorkQueue__entry_thunk(void *data);
+public:
+	void queue_work(WorkQueueWork *work);
+	void flush(void);
+	void destroy(void);
+
+	WorkQueue(void);
+	~WorkQueue();
+};
+
+}
+
+#endif
diff --git a/common/workqueue_posix.cpp b/common/workqueue_posix.cpp
new file mode 100644
index 0000000..dbcb82e
--- /dev/null
+++ b/common/workqueue_posix.cpp
@@ -0,0 +1,117 @@
+#include <pthread.h>
+#include <queue>
+#include <assert.h>
+
+#include "os_workqueue.hpp"
+
+namespace os
+{
+
+/**
+ * return 0 on batch complete, -1 on thread exit request.
+ */
+int WorkQueue::run_tasks(void)
+{
+    pthread_mutex_lock(&lock);
+
+    while (work_queue.empty() && !exit_workqueue)
+        pthread_cond_wait(&wake_cond, &lock);
+
+    if (exit_workqueue) {
+        pthread_mutex_unlock(&lock);
+        return -1;
+    }
+
+    std::queue<WorkQueueWork *> batch;
+    std::swap(work_queue, batch);
+    busy = true;
+
+    pthread_mutex_unlock(&lock);
+
+    assert(!batch.empty());
+    while (!batch.empty()) {
+        WorkQueueWork *task;
+
+        task = batch.front();
+        task->run();
+        batch.pop();
+        delete task;
+    }
+
+    pthread_mutex_lock(&lock);
+
+    busy = false;
+    pthread_cond_signal(&complete_cond);
+
+    pthread_mutex_unlock(&lock);
+
+    return 0;
+}
+
+/* Must be called with WorkQueue::lock held */
+void WorkQueue::wake_up_thread(void)
+{
+    pthread_cond_signal(&wake_cond);
+}
+
+void WorkQueue::queue_work(WorkQueueWork *task)
+{
+    pthread_mutex_lock(&lock);
+    work_queue.push(task);
+    wake_up_thread();
+    pthread_mutex_unlock(&lock);
+}
+
+void WorkQueue::flush(void)
+{
+    pthread_mutex_lock(&lock);
+    while (!work_queue.empty() || busy)
+        pthread_cond_wait(&complete_cond, &lock);
+    pthread_mutex_unlock(&lock);
+}
+
+void WorkQueue::thread_entry(void)
+{
+    int err;
+
+    do {
+        err = run_tasks();
+    } while (!err);
+}
+
+void WorkQueue::destroy(void)
+{
+    pthread_mutex_lock(&lock);
+    exit_workqueue = true;
+    wake_up_thread();
+    pthread_mutex_unlock(&lock);
+}
+
+extern "C"
+void *WorkQueue__entry_thunk(void *data)
+{
+    WorkQueue *thread = static_cast<WorkQueue *>(data);
+
+    thread->thread_entry();
+
+    return NULL;
+}
+
+WorkQueue::WorkQueue(void) :
+    busy(false), exit_workqueue(false)
+{
+    int err;
+
+    pthread_cond_init(&wake_cond, NULL);
+    pthread_cond_init(&complete_cond, NULL);
+    pthread_mutex_init(&lock, NULL);
+    err = pthread_create(&handle, NULL, WorkQueue__entry_thunk, this);
+    assert(!err);
+}
+
+WorkQueue::~WorkQueue(void)
+{
+    pthread_join(handle, NULL);
+}
+
+}
diff --git a/common/workqueue_win32.cpp b/common/workqueue_win32.cpp
new file mode 100644
index 0000000..cec6693
--- /dev/null
+++ b/common/workqueue_win32.cpp
@@ -0,0 +1,48 @@
+#include <pthread.h>
+#include <queue>
+#include <assert.h>
+
+#include "os.hpp"
+#include "os_workqueue.hpp"
+
+namespace os
+{
+
+/**
+ * return 0 on batch complete, -1 on thread exit request.
+ */
+int WorkQueue::run_tasks(void)
+{
+    return 0;
+}
+
+void WorkQueue::queue_work(WorkQueueWork *task)
+{
+    task->run();
+}
+
+void WorkQueue::flush(void)
+{
+}
+
+void WorkQueue::destroy(void)
+{
+}
+
+WorkQueue::WorkQueue(void) :
+	busy(false), exit_workqueue(false)
+{
+    static bool warned;
+
+    if (!warned) {
+        warned = true;
+        os::log("%s: no workqueue implementation, running in single-threaded mode\n",
+                __func__);
+    }
+}
+
+WorkQueue::~WorkQueue(void)
+{
+}
+
+}
-- 
1.7.9.5