Mercurial
changeset 250:745fd127b2a1
[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 | c5129452493e |
| children | 117c4d53c9a4 |
| files | .claude/skills/zenbu-seobeo-networking/SKILL.md mrjunejune/main.c seobeo/BUILD seobeo/README.md seobeo/s_worker.c seobeo/seobeo.h seobeo/seobeo_worker.h seobeo/tests/BUILD seobeo/tests/seobeo_worker_test.c |
| diffstat | 9 files changed, 978 insertions(+), 63 deletions(-) [+] |
line wrap: on
line diff
--- a/.claude/skills/zenbu-seobeo-networking/SKILL.md Tue Aug 04 04:16:45 2026 -0700 +++ b/.claude/skills/zenbu-seobeo-networking/SKILL.md Tue Aug 04 06:23:37 2026 -0700 @@ -64,6 +64,19 @@ Seobeo_WebSocket_Server_Register("/chat", Chat_Handler, NULL); ``` +Background work: + +```c +Seobeo_Worker_Pool *pool = Seobeo_Worker_Pool_Create(2, 16); +Seobeo_Worker_Pool_Submit(pool, Process_File, context, free); +Seobeo_Worker_Pool_Shutdown(pool, TRUE); +Seobeo_Worker_Pool_Destroy(pool); +``` + +Use `Seobeo_Thread_Start` + `Seobeo_Thread_Join` for one joinable task, or +`Seobeo_Thread_Start_Detached` for fire-and-forget work. Prefer a bounded pool +for request-triggered FFmpeg, upload, or other expensive jobs. + ## Build targets Choose the smallest library variant that matches the feature: @@ -75,6 +88,7 @@ - `//seobeo:seobeo_tcp_client_ws`: HTTP client with WebSocket. - `//seobeo:seobeo`: full combined library. - `//seobeo:seobeo_debug`: full library with debug logging. +- `//seobeo:seobeo_worker`: standalone thread/worker-pool API. ## Tests
--- a/mrjunejune/main.c Tue Aug 04 04:16:45 2026 -0700 +++ b/mrjunejune/main.c Tue Aug 04 06:23:37 2026 -0700 @@ -16,8 +16,9 @@ volatile sig_atomic_t stop_server = 0; static _Atomic uint32 counter = 0; static _Atomic boolean g_latex_rendering = FALSE; +static Seobeo_Worker_Pool *g_media_worker_pool = NULL; -// Media Processing Context for background threads +// Media processing context owned by a background worker. typedef struct { int64 media_id; char s3_key_original[512]; @@ -31,6 +32,7 @@ typedef struct { char *input_path; char *output_path; + int result; } File_Converter_Config; // Server configuration (loaded from .config) @@ -407,8 +409,8 @@ return resp; } -// Background thread function for media processing -void *Simple_WebpConverter_Background(void *arg) +// Joinable worker function for local image conversion. +void Simple_WebpConverter_Background(void *arg) { File_Converter_Config *configuration = (File_Converter_Config *)arg; @@ -416,16 +418,21 @@ snprintf(cmd, sizeof(cmd), "ffmpeg -y -i %s -quality 80 %s 2>/tmp/error_log", configuration->input_path, configuration->output_path); Seobeo_Log(SEOBEO_INFO, "[MEDIA] Running FFmpeg: %s\n", cmd); - int ffmpeg_result = system(cmd); + configuration->result = system(cmd); - Seobeo_Log(SEOBEO_INFO, "[MEDIA] FFmpeg result: %d\n", ffmpeg_result); - if (ffmpeg_result != 0) + Seobeo_Log( + SEOBEO_INFO, + "[MEDIA] FFmpeg result: %d\n", + configuration->result); + if (configuration->result != 0) { Seobeo_Log(SEOBEO_ERROR, "[MEDIA] ERROR: FFmpeg conversion failed\n"); - return NULL; + return; } - Seobeo_Log(SEOBEO_INFO, "[MEDIA] Successfully converted to webp: %s\n"); - return NULL; + Seobeo_Log( + SEOBEO_INFO, + "[MEDIA] Successfully converted to webp: %s\n", + configuration->output_path); } Seobeo_Request_Entry *ConvertImageToWebP(Seobeo_Request_Entry *req, Dowa_Arena *arena) @@ -481,7 +488,10 @@ int open_flags = O_RDWR | O_CREAT | O_EXCL; char *uuid4 = (char *)Dowa_Arena_Allocate(arena, UUID_LEN); - uint32 seed = (uint32)time(NULL) ^ (uint32)pthread_self() ^ counter++; + uint32 seed = + (uint32)time(NULL) ^ + (uint32)Seobeo_Thread_Current_Id() ^ + counter++; Dowa_String_UUID(seed, uuid4); char *input_path = Dowa_Arena_Allocate(arena, TMP_FILE_LENGTH);; snprintf(input_path, TMP_FILE_LENGTH, "/tmp/%s", uuid4); @@ -499,7 +509,10 @@ uuid4 = (char *)Dowa_Arena_Allocate(arena, UUID_LEN); - seed = (uint32)time(NULL) ^ (uint32)pthread_self() ^ counter++; + seed = + (uint32)time(NULL) ^ + (uint32)Seobeo_Thread_Current_Id() ^ + counter++; Dowa_String_UUID(seed, uuid4); char *output_path = (char *)Dowa_Arena_Allocate(arena, TMP_FILE_LENGTH);; snprintf(output_path, TMP_FILE_LENGTH, "/tmp/%s.webp", uuid4); @@ -523,11 +536,15 @@ File_Converter_Config *configuration = Dowa_Arena_Allocate(arena, sizeof(File_Converter_Config)); configuration->input_path = input_path; configuration->output_path = output_path; + configuration->result = -1; - pthread_t thread_id; - int thread_result = pthread_create(&thread_id, NULL, Simple_WebpConverter_Background, (void *)configuration); - - if (thread_result != 0) + Seobeo_Thread *p_worker = Seobeo_Thread_Start( + Simple_WebpConverter_Background, + configuration, + NULL); + if (!p_worker || + Seobeo_Thread_Join(p_worker) != SEOBEO_WORKER_OK || + configuration->result != 0) { unlink(input_path); unlink(output_path); @@ -537,14 +554,7 @@ Dowa_HashMap_Push_Arena(resp, "body", error_msg, arena); return resp; } - else - { - // Detach thread so it cleans up automatically when done - pthread_detach(thread_id); - Seobeo_Log(SEOBEO_INFO, "[MEDIA] Successfully spawned and detached thread\n"); - } - size_t converted_size = 0; FILE *out_file = fopen(output_path, "rb"); if (!out_file) { @@ -607,7 +617,10 @@ int open_flags = O_RDWR | O_CREAT | O_EXCL; char *uuid4 = (char *)Dowa_Arena_Allocate(arena, UUID_LEN); - uint32 seed = (uint32)time(NULL) ^ (uint32)pthread_self() ^ counter++; + uint32 seed = + (uint32)time(NULL) ^ + (uint32)Seobeo_Thread_Current_Id() ^ + counter++; Dowa_String_UUID(seed, uuid4); char *input_path = Dowa_Arena_Allocate(arena, TMP_FILE_LENGTH); snprintf(input_path, TMP_FILE_LENGTH, "/tmp/%s", uuid4); @@ -627,7 +640,10 @@ write(input_fd, file_data, file_size); close(input_fd); - seed = (uint32)time(NULL) ^ (uint32)pthread_self() ^ counter++; + seed = + (uint32)time(NULL) ^ + (uint32)Seobeo_Thread_Current_Id() ^ + counter++; Dowa_String_UUID(seed, uuid4); char *output_path = (char *)Dowa_Arena_Allocate(arena, TMP_FILE_LENGTH);; snprintf(output_path, TMP_FILE_LENGTH, "/tmp/%s.mp4", uuid4); @@ -953,7 +969,10 @@ // Generate unique S3 key with timestamp char s3_key[512]; char *uuid = Dowa_Arena_Allocate(arena, UUID_LEN); - uint32 seed = (uint32)time(NULL) ^ (uint32)pthread_self() ^ counter++; + uint32 seed = + (uint32)time(NULL) ^ + (uint32)Seobeo_Thread_Current_Id() ^ + counter++; Dowa_String_UUID(seed, uuid); snprintf(s3_key, sizeof(s3_key), "uploads/%s/%s", uuid, filename); @@ -1336,7 +1355,10 @@ // Generate UUID for this upload char *uuid = Dowa_Arena_Allocate(arena, UUID_LEN); - uint32 seed = (uint32)time(NULL) ^ (uint32)pthread_self() ^ counter++; + uint32 seed = + (uint32)time(NULL) ^ + (uint32)Seobeo_Thread_Current_Id() ^ + counter++; Dowa_String_UUID(seed, uuid); // Generate S3 keys @@ -1423,23 +1445,22 @@ return resp; } -// Background thread function for media processing -void *Media_Process_Background(void *arg) +// Worker-pool task for S3 media processing. +void Media_Process_Background(void *arg) { Media_Processing_Context *ctx = (Media_Processing_Context *)arg; - Seobeo_Log(SEOBEO_INFO, "[MEDIA] Background thread started for media_id=%lld\n", (long long)ctx->media_id); + Seobeo_Log(SEOBEO_INFO, "[MEDIA] Background worker started for media_id=%lld\n", (long long)ctx->media_id); Seobeo_Log(SEOBEO_INFO, "[MEDIA] S3 key original: %s\n", ctx->s3_key_original); Seobeo_Log(SEOBEO_INFO, "[MEDIA] S3 key processed: %s\n", ctx->s3_key_processed); Seobeo_Log(SEOBEO_INFO, "[MEDIA] DB path: %s\n", ctx->db_path); - // Open thread-local DB connection + // Open a worker-local DB connection. Deita_Connection *db_conn = Deita_Connection_Create(DEITA_DATABASE_TYPE_SQLITE3, ctx->db_path); if (!db_conn || !Deita_Connection_Is_Open(db_conn)) { - Seobeo_Log(SEOBEO_ERROR, "[MEDIA] Thread ERROR: Failed to open database for media_id=%lld\n", (long long)ctx->media_id); - free(ctx); - return NULL; + Seobeo_Log(SEOBEO_ERROR, "[MEDIA] Worker ERROR: Failed to open database for media_id=%lld\n", (long long)ctx->media_id); + return; } // Update status to 'processing' @@ -1465,8 +1486,7 @@ Deita_Query_Execute_Update_Prepared(db_conn, update_error, 2, error_params); S3_Presigned_URL_Destroy(&download_url); Deita_Connection_Close(db_conn); - free(ctx); - return NULL; + return; } Seobeo_Log(SEOBEO_INFO, "[MEDIA] Generated presigned URL: %.100s...\n", download_url.url); @@ -1475,8 +1495,14 @@ char tmp_output[256]; char *uuid_input = malloc(UUID_LEN); char *uuid_output = malloc(UUID_LEN); - uint32 seed1 = (uint32)time(NULL) ^ (uint32)pthread_self() ^ counter++; - uint32 seed2 = (uint32)time(NULL) ^ (uint32)pthread_self() ^ counter++; + uint32 seed1 = + (uint32)time(NULL) ^ + (uint32)Seobeo_Thread_Current_Id() ^ + counter++; + uint32 seed2 = + (uint32)time(NULL) ^ + (uint32)Seobeo_Thread_Current_Id() ^ + counter++; Dowa_String_UUID(seed1, uuid_input); Dowa_String_UUID(seed2, uuid_output); snprintf(tmp_input, sizeof(tmp_input), "/tmp/%s", uuid_input); @@ -1505,8 +1531,7 @@ if (download_resp) Seobeo_Client_Response_Destroy(download_resp); unlink(tmp_input); Deita_Connection_Close(db_conn); - free(ctx); - return NULL; + return; } Seobeo_Log(SEOBEO_INFO, "[MEDIA] Successfully downloaded file to %s\n", tmp_input); @@ -1535,8 +1560,7 @@ unlink(tmp_input); unlink(tmp_output); Deita_Connection_Close(db_conn); - free(ctx); - return NULL; + return; } Seobeo_Log(SEOBEO_INFO, "[MEDIA] Successfully converted to webp: %s\n", tmp_output); @@ -1557,8 +1581,7 @@ unlink(tmp_input); unlink(tmp_output); Deita_Connection_Close(db_conn); - free(ctx); - return NULL; + return; } Seobeo_Log(SEOBEO_INFO, "[MEDIA] Successfully uploaded processed file to S3\n"); @@ -1574,9 +1597,6 @@ unlink(tmp_input); unlink(tmp_output); Deita_Connection_Close(db_conn); - free(ctx); - - return NULL; } // Media Upload API - Mark uploaded @@ -1684,13 +1704,20 @@ Seobeo_Log(SEOBEO_INFO, "[MEDIA] Content type for media_id=%lld: '%s'\n", (long long)media_id, content_type_copy); - // If content_type starts with "image/", spawn background processing thread + // Images are processed asynchronously by the bounded media pool. if (strncmp(content_type_copy, "image/", 6) == 0) { - Seobeo_Log(SEOBEO_INFO, "[MEDIA] Detected image type, preparing to spawn background thread for media_id=%lld\n", (long long)media_id); + Seobeo_Log(SEOBEO_INFO, "[MEDIA] Queueing image processing for media_id=%lld\n", (long long)media_id); - // Create context for background thread (heap allocated) + // The pool owns this context after a successful submission. Media_Processing_Context *ctx = malloc(sizeof(Media_Processing_Context)); + if (!ctx) + { + Dowa_HashMap_Push_Arena(resp, "status", "500", arena); + Dowa_HashMap_Push_Arena(resp, "content-type", "application/json", arena); + Dowa_HashMap_Push_Arena(resp, "body", "{\"error\":\"Unable to allocate media work\"}", arena); + return resp; + } ctx->media_id = media_id; strncpy(ctx->s3_key_original, s3_key_original_copy, sizeof(ctx->s3_key_original) - 1); strncpy(ctx->s3_key_processed, s3_key_processed_copy, sizeof(ctx->s3_key_processed) - 1); @@ -1704,23 +1731,42 @@ ctx->db_path[sizeof(ctx->db_path) - 1] = '\0'; ctx->s3_config = g_s3_config; - Seobeo_Log(SEOBEO_INFO, "[MEDIA] Creating pthread for media_id=%lld\n", (long long)media_id); - - // Spawn detached thread - pthread_t thread_id; - int thread_result = pthread_create(&thread_id, NULL, Media_Process_Background, ctx); - - if (thread_result != 0) + Seobeo_Worker_Result worker_result = + g_media_worker_pool + ? Seobeo_Worker_Pool_Submit( + g_media_worker_pool, + Media_Process_Background, + ctx, + free) + : SEOBEO_WORKER_STOPPED; + if (worker_result != SEOBEO_WORKER_OK) { - Seobeo_Log(SEOBEO_ERROR, "[MEDIA] ERROR: pthread_create failed with result=%d for media_id=%lld\n", thread_result, (long long)media_id); + Seobeo_Log( + SEOBEO_ERROR, + "[MEDIA] Worker submission failed with result=%d for media_id=%lld\n", + worker_result, + (long long)media_id); free(ctx); + const char *update_error = + "UPDATE media_uploads SET status='error', error_message=?, updated_at=strftime('%s','now') WHERE id=?"; + const char *error_params[] = { + "Media worker queue is unavailable", + media_id_str, + }; + Deita_Query_Execute_Update_Prepared( + g_db_connection, + update_error, + 2, + error_params); + Dowa_HashMap_Push_Arena(resp, "status", "503", arena); + Dowa_HashMap_Push_Arena(resp, "content-type", "application/json", arena); + Dowa_HashMap_Push_Arena(resp, "body", "{\"error\":\"Media worker queue is unavailable\"}", arena); + return resp; } - else - { - // Detach thread so it cleans up automatically when done - pthread_detach(thread_id); - Seobeo_Log(SEOBEO_INFO, "[MEDIA] Successfully spawned and detached thread for media_id=%lld\n", (long long)media_id); - } + Seobeo_Log( + SEOBEO_INFO, + "[MEDIA] Submitted media_id=%lld to the worker pool\n", + (long long)media_id); } else { @@ -1952,6 +1998,14 @@ // Initialize database init_database(); + g_media_worker_pool = Seobeo_Worker_Pool_Create(2, 16); + if (!g_media_worker_pool) + { + Seobeo_Log( + SEOBEO_ERROR, + "[MEDIA] Unable to initialize the media worker pool\n"); + } + Seobeo_Router_Init(); Seobeo_Router_Register("GET", "/", GetHomePage); @@ -2015,4 +2069,6 @@ if (!server_port || server_port[0] == '\0') server_port = "6969"; Seobeo_Web_Server_Start("mrjunejune/src", server_port, SEOBEO_MODE_EDGE, 4); + Seobeo_Worker_Pool_Destroy(g_media_worker_pool); + g_media_worker_pool = NULL; }
--- a/seobeo/BUILD Tue Aug 04 04:16:45 2026 -0700 +++ b/seobeo/BUILD Tue Aug 04 06:23:37 2026 -0700 @@ -6,11 +6,21 @@ srcs = [ "seobeo.h", "seobeo_internal.h", + "seobeo_worker.h", "snapshot_creator.h", ], visibility = ["//visibility:public"], ) +cc_library( + name = "seobeo_worker", + srcs = ["s_worker.c"], + hdrs = ["seobeo_worker.h"], + deps = ["//dowa:dowa"], + linkopts = ["-lpthread"], + visibility = ["//visibility:public"], +) + # Minimal TCP/SSL handling only (no HTTP, no WebSocket) alias( name = "seobeo_min", @@ -305,6 +315,7 @@ ], hdrs = [":seobeo_hdrs"], deps = [ + ":seobeo_worker", "//dowa:dowa", "@openssl//:ssl", ], @@ -331,6 +342,7 @@ ], hdrs = [":seobeo_hdrs"], deps = [ + ":seobeo_worker", "//dowa:dowa", "@openssl//:ssl", ], @@ -442,6 +454,7 @@ ], hdrs = [":seobeo_hdrs"], deps = [ + ":seobeo_worker", "//dowa:dowa", "@openssl//:ssl", ], @@ -468,6 +481,7 @@ ], hdrs = [":seobeo_hdrs"], deps = [ + ":seobeo_worker", "//dowa:dowa", "@openssl//:ssl", ],
--- a/seobeo/README.md Tue Aug 04 04:16:45 2026 -0700 +++ b/seobeo/README.md Tue Aug 04 06:23:37 2026 -0700 @@ -7,6 +7,7 @@ - HTTP/HTTPS client - SSL/TLS support - Async networking with libuv +- Joinable/detached tasks and bounded worker pools - Snapshot testing utilities ## Files @@ -19,6 +20,8 @@ | `s_network.c` | Network utilities | | `s_ssl.c` | SSL/TLS handling | | `s_logging.c` | Logging utilities | +| `s_worker.c` | Thread and worker-pool implementation | +| `seobeo_worker.h` | Public worker API | | `snapshot_creator.c/h` | Snapshot testing | | `docs/` | Documentation | | `examples/` | Usage examples | @@ -43,6 +46,48 @@ bazel test //seobeo:seobeo_test ``` +## Background workers + +Use a detached task for simple fire-and-forget work: + +```c +void Convert_Image(void *p_context) +{ + Conversion *p_conversion = p_context; + Run_FFmpeg(p_conversion); +} + +Seobeo_Worker_Result result = + Seobeo_Thread_Start_Detached( + Convert_Image, + p_conversion, + free); +if (result != SEOBEO_WORKER_OK) + free(p_conversion); +``` + +Use a bounded pool when requests can enqueue expensive work: + +```c +Seobeo_Worker_Pool *p_pool = + Seobeo_Worker_Pool_Create(2, 16); + +Seobeo_Worker_Result result = + Seobeo_Worker_Pool_Submit( + p_pool, + Convert_Image, + p_conversion, + free); + +Seobeo_Worker_Pool_Shutdown(p_pool, TRUE); +Seobeo_Worker_Pool_Destroy(p_pool); +``` + +Submitting transfers context ownership only when it returns +`SEOBEO_WORKER_OK`. Cleanup runs after successful work and for queued tasks +discarded by a non-draining shutdown. User callbacks never run while the pool +mutex is held. + ## Dependencies - libuv (via //third_party/libuv)
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/seobeo/s_worker.c Tue Aug 04 06:23:37 2026 -0700 @@ -0,0 +1,443 @@ +#include "seobeo/seobeo_worker.h" + +#include <pthread.h> +#include <stdlib.h> +#include <string.h> + +typedef struct { + Seobeo_Work_Function function; + void *p_context; + Seobeo_Work_Cleanup cleanup; +} Seobeo_Work_Item; + +struct Seobeo_Thread { + pthread_t thread; + Seobeo_Work_Item item; +}; + +struct Seobeo_Worker_Pool { + pthread_t *p_threads; + Seobeo_Work_Item *p_queue; + uint32 worker_count; + uint32 queue_capacity; + uint32 queue_head; + uint32 queue_count; + uint32 active_count; + uint32 cleanup_count; + boolean accepting; + boolean stopping; + boolean drain; + boolean shutdown_started; + boolean joined; + pthread_mutex_t mutex; + pthread_cond_t work_available; + pthread_cond_t idle; +}; + +static _Thread_local Seobeo_Worker_Pool *g_current_worker_pool = NULL; + +static void Seobeo_Work_Item_Run(Seobeo_Work_Item *p_item) +{ + p_item->function(p_item->p_context); + if (p_item->cleanup) + p_item->cleanup(p_item->p_context); +} + +static void *Seobeo_Thread_Run(void *p_argument) +{ + Seobeo_Thread *p_thread = p_argument; + Seobeo_Work_Item_Run(&p_thread->item); + return NULL; +} + +Seobeo_Thread *Seobeo_Thread_Start( + Seobeo_Work_Function function, + void *p_context, + Seobeo_Work_Cleanup cleanup) +{ + if (!function) + return NULL; + + Seobeo_Thread *p_thread = calloc(1, sizeof(*p_thread)); + if (!p_thread) + return NULL; + p_thread->item = (Seobeo_Work_Item){ + .function = function, + .p_context = p_context, + .cleanup = cleanup, + }; + if (pthread_create( + &p_thread->thread, + NULL, + Seobeo_Thread_Run, + p_thread) != 0) + { + free(p_thread); + return NULL; + } + return p_thread; +} + +Seobeo_Worker_Result Seobeo_Thread_Join(Seobeo_Thread *p_thread) +{ + if (!p_thread) + return SEOBEO_WORKER_INVALID_ARGUMENT; + if (pthread_equal(pthread_self(), p_thread->thread)) + return SEOBEO_WORKER_INVALID_ARGUMENT; + + int result = pthread_join(p_thread->thread, NULL); + free(p_thread); + return result == 0 + ? SEOBEO_WORKER_OK + : SEOBEO_WORKER_THREAD_ERROR; +} + +static void *Seobeo_Thread_Run_Detached(void *p_argument) +{ + Seobeo_Work_Item *p_item = p_argument; + Seobeo_Work_Item_Run(p_item); + free(p_item); + return NULL; +} + +Seobeo_Worker_Result Seobeo_Thread_Start_Detached( + Seobeo_Work_Function function, + void *p_context, + Seobeo_Work_Cleanup cleanup) +{ + if (!function) + return SEOBEO_WORKER_INVALID_ARGUMENT; + + Seobeo_Work_Item *p_item = malloc(sizeof(*p_item)); + if (!p_item) + return SEOBEO_WORKER_OUT_OF_MEMORY; + *p_item = (Seobeo_Work_Item){ + .function = function, + .p_context = p_context, + .cleanup = cleanup, + }; + + pthread_attr_t attributes; + if (pthread_attr_init(&attributes) != 0) + { + free(p_item); + return SEOBEO_WORKER_THREAD_ERROR; + } + int result = pthread_attr_setdetachstate( + &attributes, + PTHREAD_CREATE_DETACHED); + pthread_t thread; + if (result == 0) + { + result = pthread_create( + &thread, + &attributes, + Seobeo_Thread_Run_Detached, + p_item); + } + pthread_attr_destroy(&attributes); + if (result != 0) + { + free(p_item); + return SEOBEO_WORKER_THREAD_ERROR; + } + return SEOBEO_WORKER_OK; +} + +uint64 Seobeo_Thread_Current_Id(void) +{ + pthread_t thread = pthread_self(); + const uint8 *p_bytes = (const uint8 *)&thread; + uint64 hash = 1469598103934665603ULL; + for (size_t i = 0; i < sizeof(thread); i++) + { + hash ^= p_bytes[i]; + hash *= 1099511628211ULL; + } + return hash; +} + +static boolean Seobeo_Worker_Pool_Is_Current_Thread( + Seobeo_Worker_Pool *p_pool) +{ + pthread_t current = pthread_self(); + for (uint32 i = 0; i < p_pool->worker_count; i++) + { + if (pthread_equal(current, p_pool->p_threads[i])) + return TRUE; + } + return FALSE; +} + +static void *Seobeo_Worker_Pool_Run(void *p_argument) +{ + Seobeo_Worker_Pool *p_pool = p_argument; + while (TRUE) + { + pthread_mutex_lock(&p_pool->mutex); + while (p_pool->queue_count == 0 && !p_pool->stopping) + pthread_cond_wait(&p_pool->work_available, &p_pool->mutex); + + if (p_pool->stopping && + (!p_pool->drain || p_pool->queue_count == 0)) + { + pthread_mutex_unlock(&p_pool->mutex); + break; + } + + Seobeo_Work_Item item = p_pool->p_queue[p_pool->queue_head]; + p_pool->queue_head = + (p_pool->queue_head + 1) % p_pool->queue_capacity; + p_pool->queue_count--; + p_pool->active_count++; + pthread_mutex_unlock(&p_pool->mutex); + + Seobeo_Worker_Pool *p_previous_pool = g_current_worker_pool; + g_current_worker_pool = p_pool; + Seobeo_Work_Item_Run(&item); + g_current_worker_pool = p_previous_pool; + + pthread_mutex_lock(&p_pool->mutex); + p_pool->active_count--; + if (p_pool->queue_count == 0 && + p_pool->active_count == 0 && + p_pool->cleanup_count == 0) + pthread_cond_broadcast(&p_pool->idle); + pthread_mutex_unlock(&p_pool->mutex); + } + return NULL; +} + +Seobeo_Worker_Pool *Seobeo_Worker_Pool_Create( + uint32 worker_count, + uint32 queue_capacity) +{ + if (worker_count == 0 || queue_capacity == 0) + return NULL; + + Seobeo_Worker_Pool *p_pool = calloc(1, sizeof(*p_pool)); + if (!p_pool) + return NULL; + p_pool->p_threads = calloc(worker_count, sizeof(*p_pool->p_threads)); + p_pool->p_queue = calloc(queue_capacity, sizeof(*p_pool->p_queue)); + if (!p_pool->p_threads || !p_pool->p_queue) + { + free(p_pool->p_threads); + free(p_pool->p_queue); + free(p_pool); + return NULL; + } + + p_pool->worker_count = worker_count; + p_pool->queue_capacity = queue_capacity; + p_pool->accepting = TRUE; + boolean mutex_initialized = FALSE; + boolean work_condition_initialized = FALSE; + boolean idle_condition_initialized = FALSE; + if (pthread_mutex_init(&p_pool->mutex, NULL) == 0) + mutex_initialized = TRUE; + if (mutex_initialized && + pthread_cond_init(&p_pool->work_available, NULL) == 0) + work_condition_initialized = TRUE; + if (work_condition_initialized && + pthread_cond_init(&p_pool->idle, NULL) == 0) + idle_condition_initialized = TRUE; + if (!idle_condition_initialized) + { + if (work_condition_initialized) + pthread_cond_destroy(&p_pool->work_available); + if (mutex_initialized) + pthread_mutex_destroy(&p_pool->mutex); + free(p_pool->p_threads); + free(p_pool->p_queue); + free(p_pool); + return NULL; + } + + uint32 created = 0; + for (; created < worker_count; created++) + { + if (pthread_create( + &p_pool->p_threads[created], + NULL, + Seobeo_Worker_Pool_Run, + p_pool) != 0) + break; + } + if (created != worker_count) + { + pthread_mutex_lock(&p_pool->mutex); + p_pool->stopping = TRUE; + pthread_cond_broadcast(&p_pool->work_available); + pthread_mutex_unlock(&p_pool->mutex); + for (uint32 i = 0; i < created; i++) + pthread_join(p_pool->p_threads[i], NULL); + pthread_cond_destroy(&p_pool->idle); + pthread_cond_destroy(&p_pool->work_available); + pthread_mutex_destroy(&p_pool->mutex); + free(p_pool->p_threads); + free(p_pool->p_queue); + free(p_pool); + return NULL; + } + return p_pool; +} + +Seobeo_Worker_Result Seobeo_Worker_Pool_Submit( + Seobeo_Worker_Pool *p_pool, + Seobeo_Work_Function function, + void *p_context, + Seobeo_Work_Cleanup cleanup) +{ + if (!p_pool || !function) + return SEOBEO_WORKER_INVALID_ARGUMENT; + + pthread_mutex_lock(&p_pool->mutex); + if (!p_pool->accepting) + { + pthread_mutex_unlock(&p_pool->mutex); + return SEOBEO_WORKER_STOPPED; + } + if (p_pool->queue_count == p_pool->queue_capacity) + { + pthread_mutex_unlock(&p_pool->mutex); + return SEOBEO_WORKER_QUEUE_FULL; + } + + uint32 tail = + (p_pool->queue_head + p_pool->queue_count) % + p_pool->queue_capacity; + p_pool->p_queue[tail] = (Seobeo_Work_Item){ + .function = function, + .p_context = p_context, + .cleanup = cleanup, + }; + p_pool->queue_count++; + pthread_cond_signal(&p_pool->work_available); + pthread_mutex_unlock(&p_pool->mutex); + return SEOBEO_WORKER_OK; +} + +Seobeo_Worker_Result Seobeo_Worker_Pool_Wait( + Seobeo_Worker_Pool *p_pool) +{ + if (!p_pool || + Seobeo_Worker_Pool_Is_Current_Thread(p_pool) || + g_current_worker_pool == p_pool) + return SEOBEO_WORKER_INVALID_ARGUMENT; + + pthread_mutex_lock(&p_pool->mutex); + while (p_pool->queue_count > 0 || + p_pool->active_count > 0 || + p_pool->cleanup_count > 0) + pthread_cond_wait(&p_pool->idle, &p_pool->mutex); + pthread_mutex_unlock(&p_pool->mutex); + return SEOBEO_WORKER_OK; +} + +Seobeo_Worker_Result Seobeo_Worker_Pool_Shutdown( + Seobeo_Worker_Pool *p_pool, + boolean drain) +{ + if (!p_pool || + Seobeo_Worker_Pool_Is_Current_Thread(p_pool) || + g_current_worker_pool == p_pool) + return SEOBEO_WORKER_INVALID_ARGUMENT; + + pthread_mutex_lock(&p_pool->mutex); + if (p_pool->joined) + { + pthread_mutex_unlock(&p_pool->mutex); + return SEOBEO_WORKER_OK; + } + if (p_pool->shutdown_started) + { + while (!p_pool->joined) + pthread_cond_wait(&p_pool->idle, &p_pool->mutex); + pthread_mutex_unlock(&p_pool->mutex); + return SEOBEO_WORKER_OK; + } + p_pool->shutdown_started = TRUE; + p_pool->accepting = FALSE; + p_pool->stopping = TRUE; + p_pool->drain = drain; + + if (!drain) + { + while (p_pool->queue_count > 0) + { + Seobeo_Work_Item item = p_pool->p_queue[p_pool->queue_head]; + p_pool->queue_head = + (p_pool->queue_head + 1) % p_pool->queue_capacity; + p_pool->queue_count--; + p_pool->cleanup_count++; + pthread_mutex_unlock(&p_pool->mutex); + if (item.cleanup) + { + Seobeo_Worker_Pool *p_previous_pool = g_current_worker_pool; + g_current_worker_pool = p_pool; + item.cleanup(item.p_context); + g_current_worker_pool = p_previous_pool; + } + pthread_mutex_lock(&p_pool->mutex); + p_pool->cleanup_count--; + if (p_pool->queue_count == 0 && + p_pool->active_count == 0 && + p_pool->cleanup_count == 0) + pthread_cond_broadcast(&p_pool->idle); + } + } + pthread_cond_broadcast(&p_pool->work_available); + pthread_mutex_unlock(&p_pool->mutex); + + for (uint32 i = 0; i < p_pool->worker_count; i++) + { + if (pthread_join(p_pool->p_threads[i], NULL) != 0) + return SEOBEO_WORKER_THREAD_ERROR; + } + + pthread_mutex_lock(&p_pool->mutex); + p_pool->joined = TRUE; + if (p_pool->active_count == 0 && + p_pool->cleanup_count == 0) + pthread_cond_broadcast(&p_pool->idle); + pthread_mutex_unlock(&p_pool->mutex); + return SEOBEO_WORKER_OK; +} + +void Seobeo_Worker_Pool_Destroy(Seobeo_Worker_Pool *p_pool) +{ + if (!p_pool) + return; + if (Seobeo_Worker_Pool_Is_Current_Thread(p_pool) || + g_current_worker_pool == p_pool) + return; + if (!p_pool->joined) + (void)Seobeo_Worker_Pool_Shutdown(p_pool, TRUE); + pthread_cond_destroy(&p_pool->idle); + pthread_cond_destroy(&p_pool->work_available); + pthread_mutex_destroy(&p_pool->mutex); + free(p_pool->p_threads); + free(p_pool->p_queue); + free(p_pool); +} + +uint32 Seobeo_Worker_Pool_Pending(Seobeo_Worker_Pool *p_pool) +{ + if (!p_pool) + return 0; + pthread_mutex_lock(&p_pool->mutex); + uint32 count = p_pool->queue_count; + pthread_mutex_unlock(&p_pool->mutex); + return count; +} + +uint32 Seobeo_Worker_Pool_Active(Seobeo_Worker_Pool *p_pool) +{ + if (!p_pool) + return 0; + pthread_mutex_lock(&p_pool->mutex); + uint32 count = p_pool->active_count; + pthread_mutex_unlock(&p_pool->mutex); + return count; +}
--- a/seobeo/seobeo.h Tue Aug 04 04:16:45 2026 -0700 +++ b/seobeo/seobeo.h Tue Aug 04 06:23:37 2026 -0700 @@ -9,6 +9,7 @@ */ #include "seobeo/seobeo_internal.h" +#include "seobeo/seobeo_worker.h" #include <stdarg.h> #include <unistd.h>
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/seobeo/seobeo_worker.h Tue Aug 04 06:23:37 2026 -0700 @@ -0,0 +1,72 @@ +#ifndef SEOBEO_WORKER_H +#define SEOBEO_WORKER_H + +#include "dowa/dowa.h" + +typedef void (*Seobeo_Work_Function)(void *p_context); +typedef void (*Seobeo_Work_Cleanup)(void *p_context); + +typedef enum { + SEOBEO_WORKER_OK = 0, + SEOBEO_WORKER_INVALID_ARGUMENT, + SEOBEO_WORKER_OUT_OF_MEMORY, + SEOBEO_WORKER_THREAD_ERROR, + SEOBEO_WORKER_QUEUE_FULL, + SEOBEO_WORKER_STOPPED, +} Seobeo_Worker_Result; + +typedef struct Seobeo_Thread Seobeo_Thread; +typedef struct Seobeo_Worker_Pool Seobeo_Worker_Pool; + +/* + * Start one joinable task. Join consumes and destroys the thread handle. + * The optional cleanup function runs after the work function. + */ +Seobeo_Thread *Seobeo_Thread_Start( + Seobeo_Work_Function function, + void *p_context, + Seobeo_Work_Cleanup cleanup); +Seobeo_Worker_Result Seobeo_Thread_Join(Seobeo_Thread *p_thread); + +/* + * Start a detached task. On success, Seobeo owns the context until the work + * and optional cleanup functions finish. + */ +Seobeo_Worker_Result Seobeo_Thread_Start_Detached( + Seobeo_Work_Function function, + void *p_context, + Seobeo_Work_Cleanup cleanup); + +uint64 Seobeo_Thread_Current_Id(void); + +/* + * Create a bounded reusable pool. Submit is non-blocking and transfers context + * ownership only when it returns SEOBEO_WORKER_OK. + */ +Seobeo_Worker_Pool *Seobeo_Worker_Pool_Create( + uint32 worker_count, + uint32 queue_capacity); +Seobeo_Worker_Result Seobeo_Worker_Pool_Submit( + Seobeo_Worker_Pool *p_pool, + Seobeo_Work_Function function, + void *p_context, + Seobeo_Work_Cleanup cleanup); +Seobeo_Worker_Result Seobeo_Worker_Pool_Wait( + Seobeo_Worker_Pool *p_pool); + +/* + * Stop accepting work. When drain is TRUE, queued tasks finish. When FALSE, + * queued tasks are discarded and their cleanup functions run. + * Wait, Shutdown, and Destroy must not be called from a task or cleanup + * callback belonging to the same pool. Wait and Shutdown reject those calls; + * Destroy leaves the pool unchanged. + */ +Seobeo_Worker_Result Seobeo_Worker_Pool_Shutdown( + Seobeo_Worker_Pool *p_pool, + boolean drain); +void Seobeo_Worker_Pool_Destroy(Seobeo_Worker_Pool *p_pool); + +uint32 Seobeo_Worker_Pool_Pending(Seobeo_Worker_Pool *p_pool); +uint32 Seobeo_Worker_Pool_Active(Seobeo_Worker_Pool *p_pool); + +#endif
--- a/seobeo/tests/BUILD Tue Aug 04 04:16:45 2026 -0700 +++ b/seobeo/tests/BUILD Tue Aug 04 06:23:37 2026 -0700 @@ -73,3 +73,12 @@ timeout = "short", visibility = ["//visibility:public"], ) + +cc_test( + name = "seobeo_worker_test", + srcs = ["seobeo_worker_test.c"], + deps = ["//seobeo:seobeo_worker"], + size = "small", + timeout = "short", + visibility = ["//visibility:public"], +)
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/seobeo/tests/seobeo_worker_test.c Tue Aug 04 06:23:37 2026 -0700 @@ -0,0 +1,261 @@ +#include "seobeo/seobeo_worker.h" + +#include <assert.h> +#include <stdatomic.h> +#include <stdlib.h> +#include <unistd.h> + +typedef struct { + atomic_int executed; + atomic_int cleaned; + atomic_int entered; + atomic_int release; + atomic_int cleanup_entered; + atomic_int hold_cleanup; +} Worker_Test_Context; + +typedef struct { + Seobeo_Worker_Pool *p_pool; + Seobeo_Worker_Result result; +} Pool_Shutdown_Context; + +typedef struct { + Seobeo_Worker_Pool *p_pool; + atomic_int returned; +} Pool_Wait_Context; + +typedef struct { + Seobeo_Worker_Pool *p_pool; + Seobeo_Worker_Result wait_result; + Seobeo_Worker_Result shutdown_result; + atomic_int cleaned; +} Recursive_Cleanup_Context; + +static void count_task(void *p_context) +{ + Worker_Test_Context *p_test = p_context; + atomic_fetch_add(&p_test->executed, 1); +} + +static void no_op_task(void *p_context) +{ + (void)p_context; +} + +static void blocking_task(void *p_context) +{ + Worker_Test_Context *p_test = p_context; + atomic_fetch_add(&p_test->executed, 1); + atomic_store(&p_test->entered, 1); + while (!atomic_load(&p_test->release)) + usleep(1000); +} + +static void cleanup_task(void *p_context) +{ + Worker_Test_Context *p_test = p_context; + if (atomic_load(&p_test->hold_cleanup)) + { + atomic_store(&p_test->cleanup_entered, 1); + while (!atomic_load(&p_test->release)) + usleep(1000); + } + atomic_fetch_add(&p_test->cleaned, 1); +} + +static void shutdown_pool_task(void *p_context) +{ + Pool_Shutdown_Context *p_shutdown = p_context; + p_shutdown->result = + Seobeo_Worker_Pool_Shutdown(p_shutdown->p_pool, FALSE); +} + +static void wait_pool_task(void *p_context) +{ + Pool_Wait_Context *p_wait = p_context; + assert(Seobeo_Worker_Pool_Wait(p_wait->p_pool) == SEOBEO_WORKER_OK); + atomic_store(&p_wait->returned, 1); +} + +static void recursive_pool_cleanup(void *p_context) +{ + Recursive_Cleanup_Context *p_cleanup = p_context; + p_cleanup->wait_result = + Seobeo_Worker_Pool_Wait(p_cleanup->p_pool); + p_cleanup->shutdown_result = + Seobeo_Worker_Pool_Shutdown(p_cleanup->p_pool, FALSE); + Seobeo_Worker_Pool_Destroy(p_cleanup->p_pool); + atomic_store(&p_cleanup->cleaned, 1); +} + +static void wait_for_value(atomic_int *p_value, int expected) +{ + for (int i = 0; i < 2000; i++) + { + if (atomic_load(p_value) == expected) + return; + usleep(1000); + } + assert(FALSE && "worker operation timed out"); +} + +static void test_joinable_thread(void) +{ + Worker_Test_Context context = {0}; + Seobeo_Thread *p_thread = + Seobeo_Thread_Start(count_task, &context, cleanup_task); + assert(p_thread); + assert(Seobeo_Thread_Join(p_thread) == SEOBEO_WORKER_OK); + assert(atomic_load(&context.executed) == 1); + assert(atomic_load(&context.cleaned) == 1); +} + +static void test_detached_thread(void) +{ + Worker_Test_Context context = {0}; + assert(Seobeo_Thread_Start_Detached( + count_task, + &context, + cleanup_task) == SEOBEO_WORKER_OK); + wait_for_value(&context.cleaned, 1); + assert(atomic_load(&context.executed) == 1); +} + +static void test_pool_drain(void) +{ + Worker_Test_Context context = {0}; + Seobeo_Worker_Pool *p_pool = + Seobeo_Worker_Pool_Create(2, 8); + assert(p_pool); + for (int i = 0; i < 8; i++) + { + assert(Seobeo_Worker_Pool_Submit( + p_pool, + count_task, + &context, + cleanup_task) == SEOBEO_WORKER_OK); + } + assert(Seobeo_Worker_Pool_Wait(p_pool) == SEOBEO_WORKER_OK); + assert(atomic_load(&context.executed) == 8); + assert(atomic_load(&context.cleaned) == 8); + assert(Seobeo_Worker_Pool_Pending(p_pool) == 0); + assert(Seobeo_Worker_Pool_Active(p_pool) == 0); + assert(Seobeo_Worker_Pool_Shutdown(p_pool, TRUE) == SEOBEO_WORKER_OK); + assert(Seobeo_Worker_Pool_Submit( + p_pool, + count_task, + &context, + cleanup_task) == SEOBEO_WORKER_STOPPED); + Seobeo_Worker_Pool_Destroy(p_pool); +} + +static void test_pool_queue_limit(void) +{ + Worker_Test_Context context = {0}; + Seobeo_Worker_Pool *p_pool = + Seobeo_Worker_Pool_Create(1, 1); + assert(p_pool); + assert(Seobeo_Worker_Pool_Submit( + p_pool, + blocking_task, + &context, + cleanup_task) == SEOBEO_WORKER_OK); + wait_for_value(&context.entered, 1); + assert(Seobeo_Worker_Pool_Submit( + p_pool, + count_task, + &context, + cleanup_task) == SEOBEO_WORKER_OK); + assert(Seobeo_Worker_Pool_Submit( + p_pool, + count_task, + &context, + cleanup_task) == SEOBEO_WORKER_QUEUE_FULL); + atomic_store(&context.release, 1); + assert(Seobeo_Worker_Pool_Wait(p_pool) == SEOBEO_WORKER_OK); + assert(atomic_load(&context.executed) == 2); + assert(atomic_load(&context.cleaned) == 2); + Seobeo_Worker_Pool_Destroy(p_pool); +} + +static void test_pool_cancel_queue(void) +{ + Worker_Test_Context context = {0}; + Seobeo_Worker_Pool *p_pool = + Seobeo_Worker_Pool_Create(1, 4); + assert(p_pool); + assert(Seobeo_Worker_Pool_Submit( + p_pool, + blocking_task, + &context, + cleanup_task) == SEOBEO_WORKER_OK); + wait_for_value(&context.entered, 1); + assert(Seobeo_Worker_Pool_Submit( + p_pool, + count_task, + &context, + cleanup_task) == SEOBEO_WORKER_OK); + Recursive_Cleanup_Context recursive = { + .p_pool = p_pool, + .wait_result = SEOBEO_WORKER_OK, + .shutdown_result = SEOBEO_WORKER_OK, + }; + assert(Seobeo_Worker_Pool_Submit( + p_pool, + no_op_task, + &recursive, + recursive_pool_cleanup) == SEOBEO_WORKER_OK); + + atomic_store(&context.hold_cleanup, 1); + Pool_Shutdown_Context shutdown = { + .p_pool = p_pool, + .result = SEOBEO_WORKER_THREAD_ERROR, + }; + Seobeo_Thread *p_shutdown = Seobeo_Thread_Start( + shutdown_pool_task, + &shutdown, + NULL); + assert(p_shutdown); + wait_for_value(&context.cleanup_entered, 1); + + Pool_Wait_Context wait = { + .p_pool = p_pool, + }; + Seobeo_Thread *p_wait = Seobeo_Thread_Start( + wait_pool_task, + &wait, + NULL); + assert(p_wait); + usleep(20000); + assert(atomic_load(&wait.returned) == 0); + + atomic_store(&context.release, 1); + assert(Seobeo_Thread_Join(p_shutdown) == SEOBEO_WORKER_OK); + assert(Seobeo_Thread_Join(p_wait) == SEOBEO_WORKER_OK); + assert(shutdown.result == SEOBEO_WORKER_OK); + assert(atomic_load(&context.executed) == 1); + assert(atomic_load(&context.cleaned) == 2); + assert(atomic_load(&recursive.cleaned) == 1); + assert(recursive.wait_result == SEOBEO_WORKER_INVALID_ARGUMENT); + assert(recursive.shutdown_result == SEOBEO_WORKER_INVALID_ARGUMENT); + Seobeo_Worker_Pool_Destroy(p_pool); +} + +int main(void) +{ + assert(Seobeo_Thread_Current_Id() != 0); + assert(Seobeo_Thread_Start(NULL, NULL, NULL) == NULL); + assert(Seobeo_Thread_Start_Detached( + NULL, + NULL, + NULL) == SEOBEO_WORKER_INVALID_ARGUMENT); + assert(Seobeo_Worker_Pool_Create(0, 1) == NULL); + assert(Seobeo_Worker_Pool_Create(1, 0) == NULL); + + test_joinable_thread(); + test_detached_thread(); + test_pool_drain(); + test_pool_queue_limit(); + test_pool_cancel_queue(); + return 0; +}