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
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
|
#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_ASSERT(kpl_ring_queue_init(&ring_queue, 12), "Ring queue should init");
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);
}
|
