Mercurial

comparison seobeo/s_worker.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 <pthread.h>
4 #include <stdlib.h>
5 #include <string.h>
6
7 typedef struct {
8 Seobeo_Work_Function function;
9 void *p_context;
10 Seobeo_Work_Cleanup cleanup;
11 } Seobeo_Work_Item;
12
13 struct Seobeo_Thread {
14 pthread_t thread;
15 Seobeo_Work_Item item;
16 };
17
18 struct Seobeo_Worker_Pool {
19 pthread_t *p_threads;
20 Seobeo_Work_Item *p_queue;
21 uint32 worker_count;
22 uint32 queue_capacity;
23 uint32 queue_head;
24 uint32 queue_count;
25 uint32 active_count;
26 uint32 cleanup_count;
27 boolean accepting;
28 boolean stopping;
29 boolean drain;
30 boolean shutdown_started;
31 boolean joined;
32 pthread_mutex_t mutex;
33 pthread_cond_t work_available;
34 pthread_cond_t idle;
35 };
36
37 static _Thread_local Seobeo_Worker_Pool *g_current_worker_pool = NULL;
38
39 static void Seobeo_Work_Item_Run(Seobeo_Work_Item *p_item)
40 {
41 p_item->function(p_item->p_context);
42 if (p_item->cleanup)
43 p_item->cleanup(p_item->p_context);
44 }
45
46 static void *Seobeo_Thread_Run(void *p_argument)
47 {
48 Seobeo_Thread *p_thread = p_argument;
49 Seobeo_Work_Item_Run(&p_thread->item);
50 return NULL;
51 }
52
53 Seobeo_Thread *Seobeo_Thread_Start(
54 Seobeo_Work_Function function,
55 void *p_context,
56 Seobeo_Work_Cleanup cleanup)
57 {
58 if (!function)
59 return NULL;
60
61 Seobeo_Thread *p_thread = calloc(1, sizeof(*p_thread));
62 if (!p_thread)
63 return NULL;
64 p_thread->item = (Seobeo_Work_Item){
65 .function = function,
66 .p_context = p_context,
67 .cleanup = cleanup,
68 };
69 if (pthread_create(
70 &p_thread->thread,
71 NULL,
72 Seobeo_Thread_Run,
73 p_thread) != 0)
74 {
75 free(p_thread);
76 return NULL;
77 }
78 return p_thread;
79 }
80
81 Seobeo_Worker_Result Seobeo_Thread_Join(Seobeo_Thread *p_thread)
82 {
83 if (!p_thread)
84 return SEOBEO_WORKER_INVALID_ARGUMENT;
85 if (pthread_equal(pthread_self(), p_thread->thread))
86 return SEOBEO_WORKER_INVALID_ARGUMENT;
87
88 int result = pthread_join(p_thread->thread, NULL);
89 free(p_thread);
90 return result == 0
91 ? SEOBEO_WORKER_OK
92 : SEOBEO_WORKER_THREAD_ERROR;
93 }
94
95 static void *Seobeo_Thread_Run_Detached(void *p_argument)
96 {
97 Seobeo_Work_Item *p_item = p_argument;
98 Seobeo_Work_Item_Run(p_item);
99 free(p_item);
100 return NULL;
101 }
102
103 Seobeo_Worker_Result Seobeo_Thread_Start_Detached(
104 Seobeo_Work_Function function,
105 void *p_context,
106 Seobeo_Work_Cleanup cleanup)
107 {
108 if (!function)
109 return SEOBEO_WORKER_INVALID_ARGUMENT;
110
111 Seobeo_Work_Item *p_item = malloc(sizeof(*p_item));
112 if (!p_item)
113 return SEOBEO_WORKER_OUT_OF_MEMORY;
114 *p_item = (Seobeo_Work_Item){
115 .function = function,
116 .p_context = p_context,
117 .cleanup = cleanup,
118 };
119
120 pthread_attr_t attributes;
121 if (pthread_attr_init(&attributes) != 0)
122 {
123 free(p_item);
124 return SEOBEO_WORKER_THREAD_ERROR;
125 }
126 int result = pthread_attr_setdetachstate(
127 &attributes,
128 PTHREAD_CREATE_DETACHED);
129 pthread_t thread;
130 if (result == 0)
131 {
132 result = pthread_create(
133 &thread,
134 &attributes,
135 Seobeo_Thread_Run_Detached,
136 p_item);
137 }
138 pthread_attr_destroy(&attributes);
139 if (result != 0)
140 {
141 free(p_item);
142 return SEOBEO_WORKER_THREAD_ERROR;
143 }
144 return SEOBEO_WORKER_OK;
145 }
146
147 uint64 Seobeo_Thread_Current_Id(void)
148 {
149 pthread_t thread = pthread_self();
150 const uint8 *p_bytes = (const uint8 *)&thread;
151 uint64 hash = 1469598103934665603ULL;
152 for (size_t i = 0; i < sizeof(thread); i++)
153 {
154 hash ^= p_bytes[i];
155 hash *= 1099511628211ULL;
156 }
157 return hash;
158 }
159
160 static boolean Seobeo_Worker_Pool_Is_Current_Thread(
161 Seobeo_Worker_Pool *p_pool)
162 {
163 pthread_t current = pthread_self();
164 for (uint32 i = 0; i < p_pool->worker_count; i++)
165 {
166 if (pthread_equal(current, p_pool->p_threads[i]))
167 return TRUE;
168 }
169 return FALSE;
170 }
171
172 static void *Seobeo_Worker_Pool_Run(void *p_argument)
173 {
174 Seobeo_Worker_Pool *p_pool = p_argument;
175 while (TRUE)
176 {
177 pthread_mutex_lock(&p_pool->mutex);
178 while (p_pool->queue_count == 0 && !p_pool->stopping)
179 pthread_cond_wait(&p_pool->work_available, &p_pool->mutex);
180
181 if (p_pool->stopping &&
182 (!p_pool->drain || p_pool->queue_count == 0))
183 {
184 pthread_mutex_unlock(&p_pool->mutex);
185 break;
186 }
187
188 Seobeo_Work_Item item = p_pool->p_queue[p_pool->queue_head];
189 p_pool->queue_head =
190 (p_pool->queue_head + 1) % p_pool->queue_capacity;
191 p_pool->queue_count--;
192 p_pool->active_count++;
193 pthread_mutex_unlock(&p_pool->mutex);
194
195 Seobeo_Worker_Pool *p_previous_pool = g_current_worker_pool;
196 g_current_worker_pool = p_pool;
197 Seobeo_Work_Item_Run(&item);
198 g_current_worker_pool = p_previous_pool;
199
200 pthread_mutex_lock(&p_pool->mutex);
201 p_pool->active_count--;
202 if (p_pool->queue_count == 0 &&
203 p_pool->active_count == 0 &&
204 p_pool->cleanup_count == 0)
205 pthread_cond_broadcast(&p_pool->idle);
206 pthread_mutex_unlock(&p_pool->mutex);
207 }
208 return NULL;
209 }
210
211 Seobeo_Worker_Pool *Seobeo_Worker_Pool_Create(
212 uint32 worker_count,
213 uint32 queue_capacity)
214 {
215 if (worker_count == 0 || queue_capacity == 0)
216 return NULL;
217
218 Seobeo_Worker_Pool *p_pool = calloc(1, sizeof(*p_pool));
219 if (!p_pool)
220 return NULL;
221 p_pool->p_threads = calloc(worker_count, sizeof(*p_pool->p_threads));
222 p_pool->p_queue = calloc(queue_capacity, sizeof(*p_pool->p_queue));
223 if (!p_pool->p_threads || !p_pool->p_queue)
224 {
225 free(p_pool->p_threads);
226 free(p_pool->p_queue);
227 free(p_pool);
228 return NULL;
229 }
230
231 p_pool->worker_count = worker_count;
232 p_pool->queue_capacity = queue_capacity;
233 p_pool->accepting = TRUE;
234 boolean mutex_initialized = FALSE;
235 boolean work_condition_initialized = FALSE;
236 boolean idle_condition_initialized = FALSE;
237 if (pthread_mutex_init(&p_pool->mutex, NULL) == 0)
238 mutex_initialized = TRUE;
239 if (mutex_initialized &&
240 pthread_cond_init(&p_pool->work_available, NULL) == 0)
241 work_condition_initialized = TRUE;
242 if (work_condition_initialized &&
243 pthread_cond_init(&p_pool->idle, NULL) == 0)
244 idle_condition_initialized = TRUE;
245 if (!idle_condition_initialized)
246 {
247 if (work_condition_initialized)
248 pthread_cond_destroy(&p_pool->work_available);
249 if (mutex_initialized)
250 pthread_mutex_destroy(&p_pool->mutex);
251 free(p_pool->p_threads);
252 free(p_pool->p_queue);
253 free(p_pool);
254 return NULL;
255 }
256
257 uint32 created = 0;
258 for (; created < worker_count; created++)
259 {
260 if (pthread_create(
261 &p_pool->p_threads[created],
262 NULL,
263 Seobeo_Worker_Pool_Run,
264 p_pool) != 0)
265 break;
266 }
267 if (created != worker_count)
268 {
269 pthread_mutex_lock(&p_pool->mutex);
270 p_pool->stopping = TRUE;
271 pthread_cond_broadcast(&p_pool->work_available);
272 pthread_mutex_unlock(&p_pool->mutex);
273 for (uint32 i = 0; i < created; i++)
274 pthread_join(p_pool->p_threads[i], NULL);
275 pthread_cond_destroy(&p_pool->idle);
276 pthread_cond_destroy(&p_pool->work_available);
277 pthread_mutex_destroy(&p_pool->mutex);
278 free(p_pool->p_threads);
279 free(p_pool->p_queue);
280 free(p_pool);
281 return NULL;
282 }
283 return p_pool;
284 }
285
286 Seobeo_Worker_Result Seobeo_Worker_Pool_Submit(
287 Seobeo_Worker_Pool *p_pool,
288 Seobeo_Work_Function function,
289 void *p_context,
290 Seobeo_Work_Cleanup cleanup)
291 {
292 if (!p_pool || !function)
293 return SEOBEO_WORKER_INVALID_ARGUMENT;
294
295 pthread_mutex_lock(&p_pool->mutex);
296 if (!p_pool->accepting)
297 {
298 pthread_mutex_unlock(&p_pool->mutex);
299 return SEOBEO_WORKER_STOPPED;
300 }
301 if (p_pool->queue_count == p_pool->queue_capacity)
302 {
303 pthread_mutex_unlock(&p_pool->mutex);
304 return SEOBEO_WORKER_QUEUE_FULL;
305 }
306
307 uint32 tail =
308 (p_pool->queue_head + p_pool->queue_count) %
309 p_pool->queue_capacity;
310 p_pool->p_queue[tail] = (Seobeo_Work_Item){
311 .function = function,
312 .p_context = p_context,
313 .cleanup = cleanup,
314 };
315 p_pool->queue_count++;
316 pthread_cond_signal(&p_pool->work_available);
317 pthread_mutex_unlock(&p_pool->mutex);
318 return SEOBEO_WORKER_OK;
319 }
320
321 Seobeo_Worker_Result Seobeo_Worker_Pool_Wait(
322 Seobeo_Worker_Pool *p_pool)
323 {
324 if (!p_pool ||
325 Seobeo_Worker_Pool_Is_Current_Thread(p_pool) ||
326 g_current_worker_pool == p_pool)
327 return SEOBEO_WORKER_INVALID_ARGUMENT;
328
329 pthread_mutex_lock(&p_pool->mutex);
330 while (p_pool->queue_count > 0 ||
331 p_pool->active_count > 0 ||
332 p_pool->cleanup_count > 0)
333 pthread_cond_wait(&p_pool->idle, &p_pool->mutex);
334 pthread_mutex_unlock(&p_pool->mutex);
335 return SEOBEO_WORKER_OK;
336 }
337
338 Seobeo_Worker_Result Seobeo_Worker_Pool_Shutdown(
339 Seobeo_Worker_Pool *p_pool,
340 boolean drain)
341 {
342 if (!p_pool ||
343 Seobeo_Worker_Pool_Is_Current_Thread(p_pool) ||
344 g_current_worker_pool == p_pool)
345 return SEOBEO_WORKER_INVALID_ARGUMENT;
346
347 pthread_mutex_lock(&p_pool->mutex);
348 if (p_pool->joined)
349 {
350 pthread_mutex_unlock(&p_pool->mutex);
351 return SEOBEO_WORKER_OK;
352 }
353 if (p_pool->shutdown_started)
354 {
355 while (!p_pool->joined)
356 pthread_cond_wait(&p_pool->idle, &p_pool->mutex);
357 pthread_mutex_unlock(&p_pool->mutex);
358 return SEOBEO_WORKER_OK;
359 }
360 p_pool->shutdown_started = TRUE;
361 p_pool->accepting = FALSE;
362 p_pool->stopping = TRUE;
363 p_pool->drain = drain;
364
365 if (!drain)
366 {
367 while (p_pool->queue_count > 0)
368 {
369 Seobeo_Work_Item item = p_pool->p_queue[p_pool->queue_head];
370 p_pool->queue_head =
371 (p_pool->queue_head + 1) % p_pool->queue_capacity;
372 p_pool->queue_count--;
373 p_pool->cleanup_count++;
374 pthread_mutex_unlock(&p_pool->mutex);
375 if (item.cleanup)
376 {
377 Seobeo_Worker_Pool *p_previous_pool = g_current_worker_pool;
378 g_current_worker_pool = p_pool;
379 item.cleanup(item.p_context);
380 g_current_worker_pool = p_previous_pool;
381 }
382 pthread_mutex_lock(&p_pool->mutex);
383 p_pool->cleanup_count--;
384 if (p_pool->queue_count == 0 &&
385 p_pool->active_count == 0 &&
386 p_pool->cleanup_count == 0)
387 pthread_cond_broadcast(&p_pool->idle);
388 }
389 }
390 pthread_cond_broadcast(&p_pool->work_available);
391 pthread_mutex_unlock(&p_pool->mutex);
392
393 for (uint32 i = 0; i < p_pool->worker_count; i++)
394 {
395 if (pthread_join(p_pool->p_threads[i], NULL) != 0)
396 return SEOBEO_WORKER_THREAD_ERROR;
397 }
398
399 pthread_mutex_lock(&p_pool->mutex);
400 p_pool->joined = TRUE;
401 if (p_pool->active_count == 0 &&
402 p_pool->cleanup_count == 0)
403 pthread_cond_broadcast(&p_pool->idle);
404 pthread_mutex_unlock(&p_pool->mutex);
405 return SEOBEO_WORKER_OK;
406 }
407
408 void Seobeo_Worker_Pool_Destroy(Seobeo_Worker_Pool *p_pool)
409 {
410 if (!p_pool)
411 return;
412 if (Seobeo_Worker_Pool_Is_Current_Thread(p_pool) ||
413 g_current_worker_pool == p_pool)
414 return;
415 if (!p_pool->joined)
416 (void)Seobeo_Worker_Pool_Shutdown(p_pool, TRUE);
417 pthread_cond_destroy(&p_pool->idle);
418 pthread_cond_destroy(&p_pool->work_available);
419 pthread_mutex_destroy(&p_pool->mutex);
420 free(p_pool->p_threads);
421 free(p_pool->p_queue);
422 free(p_pool);
423 }
424
425 uint32 Seobeo_Worker_Pool_Pending(Seobeo_Worker_Pool *p_pool)
426 {
427 if (!p_pool)
428 return 0;
429 pthread_mutex_lock(&p_pool->mutex);
430 uint32 count = p_pool->queue_count;
431 pthread_mutex_unlock(&p_pool->mutex);
432 return count;
433 }
434
435 uint32 Seobeo_Worker_Pool_Active(Seobeo_Worker_Pool *p_pool)
436 {
437 if (!p_pool)
438 return 0;
439 pthread_mutex_lock(&p_pool->mutex);
440 uint32 count = p_pool->active_count;
441 pthread_mutex_unlock(&p_pool->mutex);
442 return count;
443 }