[PATCH v2 16/18] 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 |  116 ++++++++++++++++++++++++++++++++++++++++++++
 common/workqueue_win32.cpp |   48 ++++++++++++++++++
 4 files changed, 216 insertions(+), 0 deletions(-)
 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 37b2b27..5551cf7 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -261,8 +261,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 ()
@@ -292,6 +294,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..8b36327
--- /dev/null
+++ b/common/workqueue_posix.cpp
@@ -0,0 +1,116 @@
+#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();
+	}
+
+	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.5.4