ring buffer tests

6 files changed, 116 insertions, 11 deletions

diff --git a/include/kpl/process.h b/include/kpl/process.h
index b47e122..7d13daf 100644
--- a/include/kpl/process.h
+++ b/include/kpl/process.h
@@ -35,6 +35,8 @@ typedef struct _kpl_task {
 } kpl_task;
 
 typedef struct {
+    int32_t thread_id;
+    kpl_env *env;
     kpl_task_queue task_queue;
     _Atomic ssize_t priority;
     // TODO
diff --git a/include/kpl/queue.h b/include/kpl/queue.h
index 3d77e13..977a32d 100644
--- a/include/kpl/queue.h
+++ b/include/kpl/queue.h
@@ -20,6 +20,11 @@ void kpl_task_queue_add(kpl_task_queue *task_queue, kpl_task *task);
 
 kpl_task *kpl_task_queue_next(kpl_task_queue *task_queue);
 
+typedef enum : int8_t {
+    KPL_RING_QUEUE_MIN_POW2     = 4,
+    KPL_RING_QUEUE_MAX_POW2     = 30
+} kpl_ring_queue_pow2_sizes;
+
 typedef struct {
     _Atomic ssize_t index;
     void *data;
@@ -31,7 +36,7 @@ typedef struct {
     kpl_ring_queue_item *items;
 } kpl_ring_queue;
 
-void kpl_ring_queue_init(kpl_ring_queue *ring_queue, int8_t size_pow);
+bool kpl_ring_queue_init(kpl_ring_queue *ring_queue, int8_t size_pow2);
 
 void kpl_ring_queue_free(kpl_ring_queue *ring_queue);
 
diff --git a/source/queue.c b/source/queue.c
index 519aa1f..6f36ce8 100644
--- a/source/queue.c
+++ b/source/queue.c
@@ -40,3 +40,60 @@ kpl_task *kpl_task_queue_next(kpl_task_queue *task_queue) {
         __atomic_store_n(&task->next, NULL, __ATOMIC_RELAXED);
     return task;
 }
+
+bool kpl_ring_queue_init(kpl_ring_queue *ring_queue, int8_t size_pow2) {
+    if (size_pow2 < KPL_RING_QUEUE_MIN_POW2 || size_pow2 > KPL_RING_QUEUE_MAX_POW2)
+        return false;
+    size_t size = 1L << size_pow2;
+    ring_queue->items = calloc(1, sizeof(kpl_ring_queue_item) * size);
+    ring_queue->items_mask = size - 1;
+    __atomic_store_n(&ring_queue->head_index, 0, __ATOMIC_RELAXED);
+    __atomic_store_n(&ring_queue->tail_index, 0, __ATOMIC_RELAXED);
+    for (size_t index = 0; index < size; index++)
+        __atomic_store_n(&ring_queue->items[index].index, index, __ATOMIC_RELAXED);
+    return true;
+}
+
+void kpl_ring_queue_free(kpl_ring_queue *ring_queue) {
+    free(ring_queue->items);
+}
+
+bool kpl_ring_queue_add(kpl_ring_queue *ring_queue, void *data) {
+    for (;;) {
+        ssize_t tail_index = __atomic_load_n(&ring_queue->tail_index, __ATOMIC_RELAXED);
+        kpl_ring_queue_item *item = &ring_queue->items[tail_index & ring_queue->items_mask];
+        ssize_t item_index = __atomic_load_n(&item->index, __ATOMIC_ACQUIRE), diff = item_index - tail_index;
+        if (diff > 0)
+            continue;
+        if (diff < 0)
+            break;
+        if (!__atomic_compare_exchange_n(&ring_queue->tail_index, &tail_index, tail_index + 1, true,
+                __ATOMIC_RELAXED, __ATOMIC_RELAXED))
+            continue;
+        item->data = data;
+        __atomic_store_n(&item->index, tail_index + 1, __ATOMIC_RELEASE);
+        return true;
+    }
+    return false;
+}
+
+void *kpl_ring_queue_next(kpl_ring_queue *ring_queue) {
+    for (;;) {
+        ssize_t head_index = __atomic_load_n(&ring_queue->head_index, __ATOMIC_RELAXED);
+        kpl_ring_queue_item *item = &ring_queue->items[head_index & ring_queue->items_mask];
+        ssize_t item_index = __atomic_load_n(&item->index, __ATOMIC_ACQUIRE),
+            diff = item_index - (head_index + 1);
+        if (diff > 0)
+            continue;
+        if (diff < 0)
+            break;
+        if (!__atomic_compare_exchange_n(&ring_queue->head_index, &head_index, head_index + 1, true,
+                __ATOMIC_RELAXED, __ATOMIC_RELAXED))
+            continue;
+         void *data = item->data;
+         item->data = NULL;
+         __atomic_store_n(&item->index, head_index + ring_queue->items_mask + 1, __ATOMIC_RELEASE);
+         return data;
+    }
+    return NULL;
+}
diff --git a/test/config.c b/test/config.c
index a871243..978e003 100644
--- a/test/config.c
+++ b/test/config.c
@@ -1,7 +1,7 @@
 
 #include "kpl_test.h"
 
-KPL_TEST(config_show_help) {
+KPL_TEST(show_help) {
     kpl_config config;
     const int32_t argc = 2;
     const char *argv[] = { "./kpl",  "--help", "not_used" };
@@ -10,7 +10,7 @@ KPL_TEST(config_show_help) {
     KPL_ASSERT(code == KPL_EXIT_CODE_HELP, "Invalid help exit code");
 }
 
-KPL_TEST(config_change_module_extension) {
+KPL_TEST(change_module_extension) {
     kpl_config config;
     const int32_t argc = 3;
     const char *argv[] = { "./kpl", "--module-extension=.l", "./file.l" };
@@ -20,7 +20,7 @@ KPL_TEST(config_change_module_extension) {
     KPL_ASSERT(strcmp(config.module_extension, ".l") == 0, "Invalid new module extension");
 }
 
-KPL_TEST(config_io_queue_pow2) {
+KPL_TEST(io_queue_pow2) {
     kpl_config config;
     const int32_t argc = 3;
     const char *argv[] = { "./kpl", "--io-queue-pow2=14", "./file.kpl" };
@@ -30,7 +30,7 @@ KPL_TEST(config_io_queue_pow2) {
     KPL_ASSERT(config.io.queue_pow2 == 14, "Invalid io queue power of two");
 }
 
-KPL_TEST(config_thread_count) {
+KPL_TEST(thread_count) {
     kpl_config config;
     const int32_t argc = 3;
     const char *argv[] = { "./kpl", "--thread-count=1", "./file.kpl" };
@@ -40,7 +40,7 @@ KPL_TEST(config_thread_count) {
     KPL_ASSERT(config.thread.count == 1, "Invalid thread count");
 }
 
-KPL_TEST(config_memory_pool_pow2) {
+KPL_TEST(memory_pool_pow2) {
     kpl_config config;
     const int32_t argc = 3;
     const char *argv[] = { "./kpl", "--memory-pool-pow2=6,7", "./file.kpl" };
diff --git a/test/queue.c b/test/queue.c
index 9886225..b011725 100644
--- a/test/queue.c
+++ b/test/queue.c
@@ -16,7 +16,7 @@ static const int32_t kpl_task_queue_total_increments = TASK_QUEUE_THREAD_COUNT *
 
 static kpl_task_queue_thread_test task_queue_threads[TASK_QUEUE_THREAD_COUNT] = {};
 
-static void *queue_task_queue_thread_fn(void *arg) {
+static void *task_queue_thread_fn(void *arg) {
     const int32_t thread_id = (int32_t) (intptr_t) arg;
     for (int32_t round = 0; round < TASK_QUEUE_PER_THREAD_INCREMENTS; round++) {
         for (int32_t id = 0; id < TASK_QUEUE_THREAD_COUNT; id++) {
@@ -35,11 +35,11 @@ static void *queue_task_queue_thread_fn(void *arg) {
     return NULL;
 }
 
-KPL_TEST(queue_task_queue_threads) {
+KPL_TEST(task_queue_threads) {
     for (int32_t id = 0; id < TASK_QUEUE_THREAD_COUNT; id++)
         kpl_task_queue_init(&task_queue_threads[id].task_queue);
     for (int32_t id = 0; id < TASK_QUEUE_THREAD_COUNT; id++)
-        pthread_create(&task_queue_threads[id].pthread, NULL, queue_task_queue_thread_fn, (void*) (intptr_t) id);
+        pthread_create(&task_queue_threads[id].pthread, NULL, task_queue_thread_fn, (void*) (intptr_t) id);
     for (int32_t id = 0; id < TASK_QUEUE_THREAD_COUNT; id++)
         pthread_join(task_queue_threads[id].pthread, NULL);
     for (int32_t id = 0; id < TASK_QUEUE_THREAD_COUNT; id++) {
@@ -53,3 +53,44 @@ KPL_TEST(queue_task_queue_threads) {
                 "Invalid number of increments");
     }
 }
+
+#define RING_QUEUE_THREAD_COUNT 3
+
+#define RING_QUEUE_SLOTS_PER_THREAD 1000
+
+#define RING_QUEUE_SLOT_LENGTH (RING_QUEUE_THREAD_COUNT * RING_QUEUE_SLOTS_PER_THREAD)
+
+static pthread_t ring_queue_threads[RING_QUEUE_THREAD_COUNT] = {};
+
+static bool ring_queue_test_slots[RING_QUEUE_SLOT_LENGTH] = {};
+
+static kpl_ring_queue ring_queue;
+
+static void *ring_queue_thread_fn(void *arg) {
+    const int32_t thread_id = (int32_t) (intptr_t) arg;
+    int32_t slot_index;
+    for (int32_t round = 1; round < RING_QUEUE_SLOTS_PER_THREAD; round++) {
+        slot_index = thread_id * RING_QUEUE_SLOTS_PER_THREAD + round;
+        if (!kpl_ring_queue_add(&ring_queue, (void*) (intptr_t) slot_index))
+            break;
+        if ((slot_index = (int32_t) (intptr_t) kpl_ring_queue_next(&ring_queue)))
+            ring_queue_test_slots[slot_index] = true;
+    }
+    while ((slot_index = (int32_t) (intptr_t) kpl_ring_queue_next(&ring_queue)))
+        ring_queue_test_slots[slot_index] = true;
+    return NULL;
+}
+
+KPL_TEST(ring_queue_threads) {
+    for (int32_t skip_round = 0; skip_round < RING_QUEUE_THREAD_COUNT; skip_round++)
+        ring_queue_test_slots[skip_round * RING_QUEUE_SLOTS_PER_THREAD] = true;
+    kpl_ring_queue_init(&ring_queue, 12);
+    for (int32_t id = 0; id < RING_QUEUE_THREAD_COUNT; id++)
+        pthread_create(&ring_queue_threads[id], NULL, ring_queue_thread_fn, (void*) (intptr_t) id);
+    for (int32_t id = 0; id < RING_QUEUE_THREAD_COUNT; id++)
+        pthread_join(ring_queue_threads[id], NULL);
+    for (int32_t slot_index = 0; slot_index < RING_QUEUE_SLOT_LENGTH; slot_index++) {
+        KPL_ASSERT(ring_queue_test_slots[slot_index], "Slot should have been set");
+    }
+    kpl_ring_queue_free(&ring_queue);
+}
diff --git a/test/test.c b/test/test.c
index a4a6281..04cade6 100644
--- a/test/test.c
+++ b/test/test.c
@@ -140,12 +140,12 @@ int32_t main(const int32_t argc, const char *argv[]) {
     return kpl_test_run_by_name(argc, argv, arg_index);
 }
 
-KPL_TEST(test_pass_by_default) {
+KPL_TEST(pass_by_default) {
     // Pass by default
     // Uncomment below to f ail
     // KPL_FAIL("Failure message");
 }
 
-KPL_TEST(test_counting) {
+KPL_TEST(counting) {
     KPL_ASSERT(_kpl_test_count > 0, "Test counter not incrementing");
 }