add task queues

1 files changed, 55 insertions, 0 deletions

diff --git a/test/queue.c b/test/queue.c
new file mode 100644
index 0000000..9886225
--- /dev/null
+++ b/test/queue.c
@@ -0,0 +1,55 @@
+
+#include "kpl_test.h"
+
+typedef struct {
+    _Atomic int32_t increments;
+    bool pop_error;
+    kpl_task_queue task_queue;
+    pthread_t pthread;
+} kpl_task_queue_thread_test;
+
+#define TASK_QUEUE_THREAD_COUNT 3
+
+#define TASK_QUEUE_PER_THREAD_INCREMENTS 10
+
+static const int32_t kpl_task_queue_total_increments = TASK_QUEUE_THREAD_COUNT * TASK_QUEUE_PER_THREAD_INCREMENTS;
+
+static kpl_task_queue_thread_test task_queue_threads[TASK_QUEUE_THREAD_COUNT] = {};
+
+static void *queue_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++) {
+            kpl_task *task = calloc(1, sizeof(kpl_task));
+            kpl_task_queue_add(&task_queue_threads[id].task_queue, task);
+        }
+        kpl_task *task = kpl_task_queue_next(&task_queue_threads[thread_id].task_queue);
+        if (task) {
+            task_queue_threads[thread_id].increments++;
+            free(task);
+        } else {
+            task_queue_threads[thread_id].pop_error = true;
+            break;
+        }
+    }
+    return NULL;
+}
+
+KPL_TEST(queue_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);
+    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++) {
+        kpl_task *task = NULL;
+        while ((task = kpl_task_queue_next(&task_queue_threads[id].task_queue))) {
+            task_queue_threads[id].increments++;
+            free(task);
+        }
+        KPL_ASSERT(!task_queue_threads[id].pop_error, "Task queue pop failed");
+        KPL_ASSERT(task_queue_threads[id].increments == kpl_task_queue_total_increments,
+                "Invalid number of increments");
+    }
+}