#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"); } }