test/queue.c


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
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");
    }
}