Added a message passing system based on lock free queues for inter-thread communication. Channels now buffer on the sending side, and no longer require blocking when sending. Lots of other refactoring and bug fixes.

6 files changed, 119 insertions, 16 deletions

diff --git a/src/rt/sync/condition_variable.cpp b/src/rt/sync/condition_variable.cpp
new file mode 100644
index 00000000..d4032572
--- /dev/null
+++ b/src/rt/sync/condition_variable.cpp
@@ -0,0 +1,66 @@
+#include "../globals.h"
+
+/*
+ * Conditional variable. Implemented using pthreads condition variables, and
+ * using events on windows.
+ */
+
+#include "condition_variable.h"
+
+// #define TRACE
+
+condition_variable::condition_variable() {
+#if defined(__WIN32__)
+    _event = CreateEvent(NULL, FALSE, FALSE, NULL);
+#else
+    pthread_cond_init(&_cond, NULL);
+    pthread_mutex_init(&_mutex, NULL);
+#endif
+}
+
+condition_variable::~condition_variable() {
+#if defined(__WIN32__)
+    CloseHandle(_event);
+#else
+    pthread_cond_destroy(&_cond);
+    pthread_mutex_destroy(&_mutex);
+#endif
+}
+
+/**
+ * Wait indefinitely until condition is signaled.
+ */
+void condition_variable::wait() {
+#ifdef TRACE
+    printf("waiting on condition_variable: 0x%" PRIxPTR "\n",
+           (uintptr_t)this);
+#endif
+#if defined(__WIN32__)
+    WaitForSingleObject(_event, INFINITE);
+#else
+    pthread_mutex_lock(&_mutex);
+    pthread_cond_wait(&_cond, &_mutex);
+    pthread_mutex_unlock(&_mutex);
+#endif
+#ifdef TRACE
+    printf("resumed on condition_variable: 0x%" PRIxPTR "\n",
+           (uintptr_t)this);
+#endif
+}
+
+/**
+ * Signal condition, and resume the waiting thread.
+ */
+void condition_variable::signal() {
+#if defined(__WIN32__)
+    SetEvent(_event);
+#else
+    pthread_mutex_lock(&_mutex);
+    pthread_cond_signal(&_cond);
+    pthread_mutex_unlock(&_mutex);
+#endif
+#ifdef TRACE
+    printf("signal  condition_variable: 0x%" PRIxPTR "\n",
+           (uintptr_t)this);
+#endif
+}
diff --git a/src/rt/sync/condition_variable.h b/src/rt/sync/condition_variable.h
new file mode 100644
index 00000000..f847ef99
--- /dev/null
+++ b/src/rt/sync/condition_variable.h
@@ -0,0 +1,19 @@
+#ifndef CONDITION_VARIABLE_H
+#define CONDITION_VARIABLE_H
+
+class condition_variable {
+#if defined(__WIN32__)
+    HANDLE _event;
+#else
+    pthread_cond_t _cond;
+    pthread_mutex_t _mutex;
+#endif
+public:
+    condition_variable();
+    virtual ~condition_variable();
+
+    void wait();
+    void signal();
+};
+
+#endif /* CONDITION_VARIABLE_H */
diff --git a/src/rt/sync/lock_free_queue.cpp b/src/rt/sync/lock_free_queue.cpp
index 9d1081de..69241ece 100644
--- a/src/rt/sync/lock_free_queue.cpp
+++ b/src/rt/sync/lock_free_queue.cpp
@@ -5,33 +5,46 @@
  * dequeue() is not allowed to interrupt itself.
  */
 
+#include "../globals.h"
 #include "lock_free_queue.h"
 
-lock_free_queue::lock_free_queue() :
-    tail(this) {
+lock_free_queue_node::lock_free_queue_node() : next(NULL) {
+
+}
+
+lock_free_queue::lock_free_queue() : _tail(this) {
+
 }
 
-void lock_free_queue::enqueue(lock_free_queue_node *item) {
-    item->next = (lock_free_queue_node *) 0;
-    lock_free_queue_node *last = tail;
-    tail = item;
-    while (last->next)
+void
+lock_free_queue::enqueue(lock_free_queue_node *item) {
+    item->next = (lock_free_queue_node *) NULL;
+    lock_free_queue_node *last = _tail;
+    _tail = item;
+    while (last->next) {
         last = last->next;
+    }
     last->next = item;
 }
 
-lock_free_queue_node *lockfree_queue::dequeue() {
+lock_free_queue_node *
+lock_free_queue::dequeue() {
     lock_free_queue_node *item = next;
     if (item && !(next = item->next)) {
-        tail = (lock_free_queue_node *) this;
+        _tail = (lock_free_queue_node *) this;
         if (item->next) {
             lock_free_queue_node *lost = item->next;
             lock_free_queue_node *help;
             do {
                 help = lost->next;
                 enqueue(lost);
-            } while ((lost = help) != (lock_free_queue_node *) 0);
+            } while ((lost = help) != (lock_free_queue_node *) NULL);
         }
     }
     return item;
 }
+
+bool
+lock_free_queue::is_empty() {
+    return next == NULL;
+}
diff --git a/src/rt/sync/lock_free_queue.h b/src/rt/sync/lock_free_queue.h
index fba4aa9a..1f09ec52 100644
--- a/src/rt/sync/lock_free_queue.h
+++ b/src/rt/sync/lock_free_queue.h
@@ -2,14 +2,18 @@
 #define LOCK_FREE_QUEUE_H
 
 class lock_free_queue_node {
+public:
     lock_free_queue_node *next;
+    lock_free_queue_node();
 };
 
-class lock_free_queue {
+class lock_free_queue : lock_free_queue_node {
+    lock_free_queue_node *_tail;
 public:
     lock_free_queue();
     void enqueue(lock_free_queue_node *item);
     lock_free_queue_node *dequeue();
+    bool is_empty();
 };
 
 #endif /* LOCK_FREE_QUEUE_H */
diff --git a/src/rt/sync/spin_lock.cpp b/src/rt/sync/spin_lock.cpp
index 11a5cb20..4a113d1a 100644
--- a/src/rt/sync/spin_lock.cpp
+++ b/src/rt/sync/spin_lock.cpp
@@ -1,9 +1,10 @@
+#include "../globals.h"
+#include "spin_lock.h"
+
 /*
  * Your average spin lock.
  */
 
-#include "globals.h"
-
 // #define TRACE
 
 spin_lock::spin_lock() {
diff --git a/src/rt/sync/spin_lock.h b/src/rt/sync/spin_lock.h
index 3684c23a..f15416a2 100644
--- a/src/rt/sync/spin_lock.h
+++ b/src/rt/sync/spin_lock.h
@@ -1,5 +1,5 @@
-#ifndef UNFAIR_TICKET_LOCK_H
-#define UNFAIR_TICKET_LOCK_H
+#ifndef SPIN_LOCK_H
+#define SPIN_LOCK_H
 
 class spin_lock {
     unsigned ticket;
@@ -11,4 +11,4 @@ public:
     void unlock();
 };
 
-#endif /* UNFAIR_TICKET_LOCK_H */
+#endif /* SPIN_LOCK_H */