Mercurial

comparison seobeo/tests/seobeo_worker_test.c @ 250:745fd127b2a1

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
[seobeo] Add bounded worker interface Co-authored-by: Copilot <[email protected]>
author MrJuneJune <me@mrjunejune.com>
date Tue, 04 Aug 2026 06:23:37 -0700
parents
children
comparison
equal deleted inserted replaced
249:c5129452493e 250:745fd127b2a1
1 #include "seobeo/seobeo_worker.h"
2
3 #include <assert.h>
4 #include <stdatomic.h>
5 #include <stdlib.h>
6 #include <unistd.h>
7
8 typedef struct {
9 atomic_int executed;
10 atomic_int cleaned;
11 atomic_int entered;
12 atomic_int release;
13 atomic_int cleanup_entered;
14 atomic_int hold_cleanup;
15 } Worker_Test_Context;
16
17 typedef struct {
18 Seobeo_Worker_Pool *p_pool;
19 Seobeo_Worker_Result result;
20 } Pool_Shutdown_Context;
21
22 typedef struct {
23 Seobeo_Worker_Pool *p_pool;
24 atomic_int returned;
25 } Pool_Wait_Context;
26
27 typedef struct {
28 Seobeo_Worker_Pool *p_pool;
29 Seobeo_Worker_Result wait_result;
30 Seobeo_Worker_Result shutdown_result;
31 atomic_int cleaned;
32 } Recursive_Cleanup_Context;
33
34 static void count_task(void *p_context)
35 {
36 Worker_Test_Context *p_test = p_context;
37 atomic_fetch_add(&p_test->executed, 1);
38 }
39
40 static void no_op_task(void *p_context)
41 {
42 (void)p_context;
43 }
44
45 static void blocking_task(void *p_context)
46 {
47 Worker_Test_Context *p_test = p_context;
48 atomic_fetch_add(&p_test->executed, 1);
49 atomic_store(&p_test->entered, 1);
50 while (!atomic_load(&p_test->release))
51 usleep(1000);
52 }
53
54 static void cleanup_task(void *p_context)
55 {
56 Worker_Test_Context *p_test = p_context;
57 if (atomic_load(&p_test->hold_cleanup))
58 {
59 atomic_store(&p_test->cleanup_entered, 1);
60 while (!atomic_load(&p_test->release))
61 usleep(1000);
62 }
63 atomic_fetch_add(&p_test->cleaned, 1);
64 }
65
66 static void shutdown_pool_task(void *p_context)
67 {
68 Pool_Shutdown_Context *p_shutdown = p_context;
69 p_shutdown->result =
70 Seobeo_Worker_Pool_Shutdown(p_shutdown->p_pool, FALSE);
71 }
72
73 static void wait_pool_task(void *p_context)
74 {
75 Pool_Wait_Context *p_wait = p_context;
76 assert(Seobeo_Worker_Pool_Wait(p_wait->p_pool) == SEOBEO_WORKER_OK);
77 atomic_store(&p_wait->returned, 1);
78 }
79
80 static void recursive_pool_cleanup(void *p_context)
81 {
82 Recursive_Cleanup_Context *p_cleanup = p_context;
83 p_cleanup->wait_result =
84 Seobeo_Worker_Pool_Wait(p_cleanup->p_pool);
85 p_cleanup->shutdown_result =
86 Seobeo_Worker_Pool_Shutdown(p_cleanup->p_pool, FALSE);
87 Seobeo_Worker_Pool_Destroy(p_cleanup->p_pool);
88 atomic_store(&p_cleanup->cleaned, 1);
89 }
90
91 static void wait_for_value(atomic_int *p_value, int expected)
92 {
93 for (int i = 0; i < 2000; i++)
94 {
95 if (atomic_load(p_value) == expected)
96 return;
97 usleep(1000);
98 }
99 assert(FALSE && "worker operation timed out");
100 }
101
102 static void test_joinable_thread(void)
103 {
104 Worker_Test_Context context = {0};
105 Seobeo_Thread *p_thread =
106 Seobeo_Thread_Start(count_task, &context, cleanup_task);
107 assert(p_thread);
108 assert(Seobeo_Thread_Join(p_thread) == SEOBEO_WORKER_OK);
109 assert(atomic_load(&context.executed) == 1);
110 assert(atomic_load(&context.cleaned) == 1);
111 }
112
113 static void test_detached_thread(void)
114 {
115 Worker_Test_Context context = {0};
116 assert(Seobeo_Thread_Start_Detached(
117 count_task,
118 &context,
119 cleanup_task) == SEOBEO_WORKER_OK);
120 wait_for_value(&context.cleaned, 1);
121 assert(atomic_load(&context.executed) == 1);
122 }
123
124 static void test_pool_drain(void)
125 {
126 Worker_Test_Context context = {0};
127 Seobeo_Worker_Pool *p_pool =
128 Seobeo_Worker_Pool_Create(2, 8);
129 assert(p_pool);
130 for (int i = 0; i < 8; i++)
131 {
132 assert(Seobeo_Worker_Pool_Submit(
133 p_pool,
134 count_task,
135 &context,
136 cleanup_task) == SEOBEO_WORKER_OK);
137 }
138 assert(Seobeo_Worker_Pool_Wait(p_pool) == SEOBEO_WORKER_OK);
139 assert(atomic_load(&context.executed) == 8);
140 assert(atomic_load(&context.cleaned) == 8);
141 assert(Seobeo_Worker_Pool_Pending(p_pool) == 0);
142 assert(Seobeo_Worker_Pool_Active(p_pool) == 0);
143 assert(Seobeo_Worker_Pool_Shutdown(p_pool, TRUE) == SEOBEO_WORKER_OK);
144 assert(Seobeo_Worker_Pool_Submit(
145 p_pool,
146 count_task,
147 &context,
148 cleanup_task) == SEOBEO_WORKER_STOPPED);
149 Seobeo_Worker_Pool_Destroy(p_pool);
150 }
151
152 static void test_pool_queue_limit(void)
153 {
154 Worker_Test_Context context = {0};
155 Seobeo_Worker_Pool *p_pool =
156 Seobeo_Worker_Pool_Create(1, 1);
157 assert(p_pool);
158 assert(Seobeo_Worker_Pool_Submit(
159 p_pool,
160 blocking_task,
161 &context,
162 cleanup_task) == SEOBEO_WORKER_OK);
163 wait_for_value(&context.entered, 1);
164 assert(Seobeo_Worker_Pool_Submit(
165 p_pool,
166 count_task,
167 &context,
168 cleanup_task) == SEOBEO_WORKER_OK);
169 assert(Seobeo_Worker_Pool_Submit(
170 p_pool,
171 count_task,
172 &context,
173 cleanup_task) == SEOBEO_WORKER_QUEUE_FULL);
174 atomic_store(&context.release, 1);
175 assert(Seobeo_Worker_Pool_Wait(p_pool) == SEOBEO_WORKER_OK);
176 assert(atomic_load(&context.executed) == 2);
177 assert(atomic_load(&context.cleaned) == 2);
178 Seobeo_Worker_Pool_Destroy(p_pool);
179 }
180
181 static void test_pool_cancel_queue(void)
182 {
183 Worker_Test_Context context = {0};
184 Seobeo_Worker_Pool *p_pool =
185 Seobeo_Worker_Pool_Create(1, 4);
186 assert(p_pool);
187 assert(Seobeo_Worker_Pool_Submit(
188 p_pool,
189 blocking_task,
190 &context,
191 cleanup_task) == SEOBEO_WORKER_OK);
192 wait_for_value(&context.entered, 1);
193 assert(Seobeo_Worker_Pool_Submit(
194 p_pool,
195 count_task,
196 &context,
197 cleanup_task) == SEOBEO_WORKER_OK);
198 Recursive_Cleanup_Context recursive = {
199 .p_pool = p_pool,
200 .wait_result = SEOBEO_WORKER_OK,
201 .shutdown_result = SEOBEO_WORKER_OK,
202 };
203 assert(Seobeo_Worker_Pool_Submit(
204 p_pool,
205 no_op_task,
206 &recursive,
207 recursive_pool_cleanup) == SEOBEO_WORKER_OK);
208
209 atomic_store(&context.hold_cleanup, 1);
210 Pool_Shutdown_Context shutdown = {
211 .p_pool = p_pool,
212 .result = SEOBEO_WORKER_THREAD_ERROR,
213 };
214 Seobeo_Thread *p_shutdown = Seobeo_Thread_Start(
215 shutdown_pool_task,
216 &shutdown,
217 NULL);
218 assert(p_shutdown);
219 wait_for_value(&context.cleanup_entered, 1);
220
221 Pool_Wait_Context wait = {
222 .p_pool = p_pool,
223 };
224 Seobeo_Thread *p_wait = Seobeo_Thread_Start(
225 wait_pool_task,
226 &wait,
227 NULL);
228 assert(p_wait);
229 usleep(20000);
230 assert(atomic_load(&wait.returned) == 0);
231
232 atomic_store(&context.release, 1);
233 assert(Seobeo_Thread_Join(p_shutdown) == SEOBEO_WORKER_OK);
234 assert(Seobeo_Thread_Join(p_wait) == SEOBEO_WORKER_OK);
235 assert(shutdown.result == SEOBEO_WORKER_OK);
236 assert(atomic_load(&context.executed) == 1);
237 assert(atomic_load(&context.cleaned) == 2);
238 assert(atomic_load(&recursive.cleaned) == 1);
239 assert(recursive.wait_result == SEOBEO_WORKER_INVALID_ARGUMENT);
240 assert(recursive.shutdown_result == SEOBEO_WORKER_INVALID_ARGUMENT);
241 Seobeo_Worker_Pool_Destroy(p_pool);
242 }
243
244 int main(void)
245 {
246 assert(Seobeo_Thread_Current_Id() != 0);
247 assert(Seobeo_Thread_Start(NULL, NULL, NULL) == NULL);
248 assert(Seobeo_Thread_Start_Detached(
249 NULL,
250 NULL,
251 NULL) == SEOBEO_WORKER_INVALID_ARGUMENT);
252 assert(Seobeo_Worker_Pool_Create(0, 1) == NULL);
253 assert(Seobeo_Worker_Pool_Create(1, 0) == NULL);
254
255 test_joinable_thread();
256 test_detached_thread();
257 test_pool_drain();
258 test_pool_queue_limit();
259 test_pool_cancel_queue();
260 return 0;
261 }