Condition variable abstraction.

3 files changed, 109 insertions, 21 deletions

diff --git a/common/os_thread.hpp b/common/os_thread.hpp
index c4bc130d..e5eb19d3 100644
--- a/common/os_thread.hpp
+++ b/common/os_thread.hpp
@@ -140,6 +140,98 @@ namespace os {
 
 
     /**
+     * Same interface as std::unique_lock;
+     */
+    template< class Mutex >
+    class unique_lock
+    {
+    public:
+        typedef Mutex mutex_type;
+
+        inline explicit
+        unique_lock(mutex_type & mutex) :
+            _mutex(&mutex)
+        {
+            _mutex->lock();
+        }
+
+        inline
+        ~unique_lock() {
+            _mutex->unlock();
+        }
+
+        inline void
+        lock() {
+            _mutex->lock();
+        }
+
+        inline void
+        unlock() {
+            _mutex->unlock();
+        }
+
+        mutex_type *
+        mutex() const {
+            return _mutex;
+        }
+
+    protected:
+        mutex_type *_mutex;
+    };
+
+
+    /**
+     * Same interface as std::condition_variable
+     */
+    class condition_variable
+    {
+    public:
+#ifdef _WIN32
+        /* FIXME */
+#else
+        typedef pthread_cond_t native_handle_type;
+#endif
+
+        condition_variable() {
+#ifdef _WIN32
+            /* FIXME */
+#else
+            pthread_cond_init(&_native_handle, NULL);
+#endif
+        }
+
+        ~condition_variable() {
+#ifdef _WIN32
+            /* FIXME */
+#else
+            pthread_cond_destroy(&_native_handle);
+#endif
+        }
+
+        inline void
+        signal(void) {
+#ifdef _WIN32
+            /* FIXME */
+#else
+            pthread_cond_signal(&_native_handle);
+#endif
+        }
+
+        inline void
+        wait(unique_lock<mutex> & lock) {
+#ifdef _WIN32
+            /* FIXME */
+#else
+            pthread_cond_wait(&_native_handle, &lock.mutex()->native_handle());
+#endif
+        }
+
+    protected:
+        native_handle_type _native_handle;
+    };
+
+
+    /**
      * Same interface as boost::thread_specific_ptr.
      */
     template <typename T>
diff --git a/common/os_workqueue.hpp b/common/os_workqueue.hpp
index e41d8b27..1c6798c1 100644
--- a/common/os_workqueue.hpp
+++ b/common/os_workqueue.hpp
@@ -28,10 +28,10 @@ class WorkQueue {
 
 	bool busy;
 	bool exit_workqueue;
-	pthread_cond_t wake_cond;
-	pthread_cond_t complete_cond;
+	os::condition_variable wake_cond;
+	os::condition_variable complete_cond;
 
-	os::mutex lock;
+	os::mutex mutex;
 
 	void wake_up_thread(void);
 	void thread_entry(void);
diff --git a/common/workqueue_posix.cpp b/common/workqueue_posix.cpp
index 9b02d6c0..f46bbd97 100644
--- a/common/workqueue_posix.cpp
+++ b/common/workqueue_posix.cpp
@@ -12,13 +12,13 @@ namespace os
  */
 int WorkQueue::run_tasks(void)
 {
-    lock.lock();
+    os::unique_lock<os::mutex> lock(mutex);
 
-    while (work_queue.empty() && !exit_workqueue)
-        pthread_cond_wait(&wake_cond, &lock.native_handle());
+    while (work_queue.empty() && !exit_workqueue) {
+        wake_cond.wait(lock);
+    }
 
     if (exit_workqueue) {
-        lock.unlock();
         return -1;
     }
 
@@ -41,9 +41,7 @@ int WorkQueue::run_tasks(void)
     lock.lock();
 
     busy = false;
-    pthread_cond_signal(&complete_cond);
-
-    lock.unlock();
+    complete_cond.signal();
 
     return 0;
 }
@@ -51,23 +49,23 @@ int WorkQueue::run_tasks(void)
 /* Must be called with WorkQueue::lock held */
 void WorkQueue::wake_up_thread(void)
 {
-    pthread_cond_signal(&wake_cond);
+    wake_cond.signal();
 }
 
 void WorkQueue::queue_work(WorkQueueWork *task)
 {
-    lock.lock();
+    mutex.lock();
     work_queue.push(task);
     wake_up_thread();
-    lock.unlock();
+    mutex.unlock();
 }
 
 void WorkQueue::flush(void)
 {
-    lock.lock();
-    while (!work_queue.empty() || busy)
-        pthread_cond_wait(&complete_cond, &lock.native_handle());
-    lock.unlock();
+    os::unique_lock<os::mutex> lock(mutex);
+    while (!work_queue.empty() || busy) {
+        complete_cond.wait(lock);
+    }
 }
 
 void WorkQueue::thread_entry(void)
@@ -81,10 +79,10 @@ void WorkQueue::thread_entry(void)
 
 void WorkQueue::destroy(void)
 {
-    lock.lock();
+    mutex.lock();
     exit_workqueue = true;
     wake_up_thread();
-    lock.unlock();
+    mutex.unlock();
 }
 
 extern "C"
@@ -102,8 +100,6 @@ WorkQueue::WorkQueue(void) :
 {
     int err;
 
-    pthread_cond_init(&wake_cond, NULL);
-    pthread_cond_init(&complete_cond, NULL);
     err = pthread_create(&handle, NULL, WorkQueue__entry_thunk, this);
     assert(!err);
 }