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

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