ascii-chat 0.11.33
Video chat in your terminal
Loading...
Searching...
No Matches
thread_pool.c File Reference

🧵 Generic thread pool implementation for managing worker threads More...

Go to the source code of this file.

Functions

thread_pool_t * thread_pool_create (const char *pool_name)
 Create a new thread pool (long-lived threads mode)
 
thread_pool_t * thread_pool_create_with_workers (const char *pool_name, size_t num_workers)
 Create a new thread pool in work queue mode.
 
void thread_pool_destroy (thread_pool_t *pool)
 Destroy a thread pool.
 
asciichat_error_t thread_pool_queue_work (const char *name, thread_pool_t *pool, void *(*work_func)(void *), void *work_arg)
 Queue a named work item to a work queue pool.
 
asciichat_error_t thread_pool_spawn (thread_pool_t *pool, void *(*thread_func)(void *), void *thread_arg, int stop_id, const char *thread_name)
 Spawn a worker thread in the pool.
 
asciichat_error_t thread_pool_stop_all (thread_pool_t *pool)
 Stop all threads in the pool in stop_id order.
 
size_t thread_pool_get_count (const thread_pool_t *pool)
 Get thread count in the pool.
 
bool thread_pool_has_threads (const thread_pool_t *pool)
 Check if pool has any threads.
 
asciichat_error_t thread_pool_interrupt_all (thread_pool_t *pool, int sig)
 Send a signal to all threads in the pool.
 

Detailed Description

🧵 Generic thread pool implementation for managing worker threads

Definition in file thread_pool.c.

Function Documentation

◆ thread_pool_create()

thread_pool_t * thread_pool_create ( const char *  pool_name)

Create a new thread pool (long-lived threads mode)

Allocates and initializes a thread pool structure. The pool starts empty with no threads. Use thread_pool_spawn() to add threads to the pool.

This mode is used for long-lived worker threads that run for the lifetime of the application or connection (e.g., stats logger, cleanup thread).

Parameters
pool_nameName for the pool (max 63 chars, used for debugging)
Returns
Pointer to new thread pool, or NULL on allocation failure

Definition at line 89 of file thread_pool.c.

89 {
91 if (!pool) {
92 SET_ERRNO(ERROR_MEMORY, "Failed to allocate thread pool");
93 return NULL;
94 }
95
96 memset(pool, 0, sizeof(*pool));
97
98 // Copy pool name (truncate if necessary)
99 if (pool_name) {
100 SAFE_STRNCPY(pool->name, pool_name, sizeof(pool->name));
101 } else {
102 SAFE_STRNCPY(pool->name, "unnamed", sizeof(pool->name));
103 }
104
105 // Initialize linked list
106 pool->threads = NULL;
107 pool->thread_count = 0;
108 pool->is_work_queue_mode = false;
109 pool->num_workers = 0;
110
111 // Initialize mutex
112 if (mutex_init(&pool->threads_mutex, "thread_pool") != 0) {
114 SET_ERRNO(ERROR_THREAD, "Failed to initialize thread pool mutex");
115 return NULL;
116 }
117
119
120 // Register sync primitives for debugging
121 NAMED_REGISTER_MUTEX(&pool->threads_mutex, "threads_mutex", (uintptr_t)(const void *)(pool));
122
123 log_debug("Thread pool '%s' created (long-lived thread mode)", pool->name);
124 return pool;
125}
buffer_pool_t * pool
#define SAFE_STRNCPY(dst, src, size)
Definition common.h:414
#define SAFE_FREE(ptr)
Definition common.h:376
#define SAFE_MALLOC(size, cast)
Definition common.h:264
#define NAMED_REGISTER_THREAD_POOL(pool, name, parent_ptr)
Register a thread pool with automatic format specifier.
#define NAMED_REGISTER_MUTEX(mutex, name, parent_ptr)
Register a mutex with automatic format specifier.
#define SET_ERRNO(code, context_msg,...)
Set error code with custom context message and log it, returning the error code.
@ ERROR_MEMORY
Definition error_codes.h:56
@ ERROR_THREAD
#define log_debug(...)
Log a DEBUG message.
Definition log/log.h:548
int mutex_init(mutex_t *mutex, const char *name)
Initialize a mutex with a name.
Definition threading.c:16
void thread_pool_t

References ERROR_MEMORY, ERROR_THREAD, log_debug, mutex_init(), NAMED_REGISTER_MUTEX, NAMED_REGISTER_THREAD_POOL, pool, SAFE_FREE, SAFE_MALLOC, SAFE_STRNCPY, and SET_ERRNO.

◆ thread_pool_create_with_workers()

thread_pool_t * thread_pool_create_with_workers ( const char *  pool_name,
size_t  num_workers 
)

Create a new thread pool in work queue mode.

Creates a thread pool with pre-allocated worker threads that pull tasks from a work queue. Use thread_pool_queue_work() to add work to the pool.

This mode is used for short-lived tasks that need to be executed without creating new threads from callback context (e.g., WebSocket handlers).

Worker threads are created immediately and wait for work to be queued. The pool will not accept spawn() calls - use queue_work() instead.

Parameters
pool_nameName for the pool (max 63 chars, used for debugging)
num_workersNumber of pre-allocated worker threads
Returns
Pointer to new thread pool, or NULL on allocation failure

Definition at line 127 of file thread_pool.c.

127 {
128 if (num_workers == 0) {
129 SET_ERRNO(ERROR_INVALID_PARAM, "num_workers must be > 0");
130 return NULL;
131 }
132
134 if (!pool) {
135 SET_ERRNO(ERROR_MEMORY, "Failed to allocate thread pool");
136 return NULL;
137 }
138
139 memset(pool, 0, sizeof(*pool));
140
141 // Copy pool name
142 if (pool_name) {
143 SAFE_STRNCPY(pool->name, pool_name, sizeof(pool->name));
144 } else {
145 SAFE_STRNCPY(pool->name, "unnamed", sizeof(pool->name));
146 }
147
148 // Mark as work queue mode
149 pool->is_work_queue_mode = true;
150 pool->num_workers = num_workers;
151 pool->work_queue = NULL;
152 pool->shutdown_requested = false;
153
154 // Initialize mutexes
155 if (mutex_init(&pool->threads_mutex, "worker_list") != 0) {
157 SET_ERRNO(ERROR_THREAD, "Failed to initialize threads_mutex");
158 return NULL;
159 }
160
161 if (mutex_init(&pool->work_queue_mutex, "pending_tasks") != 0) {
162 mutex_destroy(&pool->threads_mutex);
164 SET_ERRNO(ERROR_THREAD, "Failed to initialize work_queue_mutex");
165 return NULL;
166 }
167
168 // Initialize condition variable
169 if (cond_init(&pool->work_available, "task_available") != 0) {
170 mutex_destroy(&pool->work_queue_mutex);
171 mutex_destroy(&pool->threads_mutex);
173 SET_ERRNO(ERROR_THREAD, "Failed to initialize work_available condition");
174 return NULL;
175 }
176
177 // Create worker threads
178 for (size_t i = 0; i < num_workers; i++) {
180 if (!entry) {
181 log_error("Failed to allocate worker entry %zu", i);
183 SET_ERRNO(ERROR_MEMORY, "Failed to allocate worker entry");
184 return NULL;
185 }
186
187 memset(entry, 0, sizeof(*entry));
188 entry->stop_id = 0;
189 entry->thread_func = thread_pool_worker_thread;
190 entry->thread_arg = pool;
191 SAFE_SNPRINTF(entry->name, sizeof(entry->name), "%s_worker_%zu", pool->name, i);
192
193 // Create worker thread
194 if (asciichat_thread_create(&entry->thread, entry->name, thread_pool_worker_thread, pool) != 0) {
195 log_error("Failed to create worker thread %zu", i);
196 SAFE_FREE(entry);
198 SET_ERRNO(ERROR_THREAD, "Failed to create worker thread %zu", i);
199 return NULL;
200 }
201
202 // Add to thread list
203 mutex_lock(&pool->threads_mutex);
204 entry->next = pool->threads;
205 pool->threads = entry;
206 pool->thread_count++;
207 mutex_unlock(&pool->threads_mutex);
208
209 log_debug("[ThreadPool] Created worker thread %zu for pool '%s'", i, pool->name);
210 }
211
213
214 // Register sync primitives for debugging
215 NAMED_REGISTER_MUTEX(&pool->threads_mutex, "threads_mutex", (uintptr_t)(const void *)(pool));
216 NAMED_REGISTER_MUTEX(&pool->work_queue_mutex, "work_queue_mutex", (uintptr_t)(const void *)(pool));
217 NAMED_REGISTER_COND(&pool->work_available, "work_available", (uintptr_t)(const void *)(pool));
218
219 log_info("[ThreadPool] Created work queue pool '%s' with %zu worker threads", pool->name, num_workers);
220 return pool;
221}
#define SAFE_SNPRINTF(buffer, buffer_size,...)
Definition common.h:492
#define NAMED_REGISTER_COND(cond, name, parent_ptr)
Register a condition variable with automatic format specifier.
@ ERROR_INVALID_PARAM
#define log_error(...)
Log an ERROR message.
Definition log/log.h:587
#define log_info(...)
Log an INFO message.
Definition log/log.h:561
#define mutex_lock(mutex)
Lock a mutex (with debug tracking in debug builds)
int cond_init(cond_t *cond, const char *name)
Initialize a condition variable with a name.
#define mutex_unlock(mutex)
Unlock a mutex (with debug tracking in debug builds)
int mutex_destroy(mutex_t *mutex)
Destroy a mutex.
Definition threading.c:22
#define asciichat_thread_create(thread_ptr, attr, start_routine, arg)
Thread pool entry (internal linked list node)
Definition thread_pool.h:99
void *(* thread_func)(void *)
Thread function.
void * thread_arg
Thread argument.
asciichat_thread_t thread
Thread handle.
struct thread_pool_entry * next
Linked list next pointer.
int stop_id
Cleanup order (lower = stop first, -1 = unordered)
char name[64]
Thread name for debugging.
void thread_pool_destroy(thread_pool_t *pool)
Destroy a thread pool.

References asciichat_thread_create, cond_init(), ERROR_INVALID_PARAM, ERROR_MEMORY, ERROR_THREAD, log_debug, log_error, log_info, mutex_destroy(), mutex_init(), mutex_lock, mutex_unlock, thread_pool_entry::name, NAMED_REGISTER_COND, NAMED_REGISTER_MUTEX, NAMED_REGISTER_THREAD_POOL, thread_pool_entry::next, pool, SAFE_FREE, SAFE_MALLOC, SAFE_SNPRINTF, SAFE_STRNCPY, SET_ERRNO, thread_pool_entry::stop_id, thread_pool_entry::thread, thread_pool_entry::thread_arg, thread_pool_entry::thread_func, and thread_pool_destroy().

Referenced by websocket_server_init().

◆ thread_pool_destroy()

void thread_pool_destroy ( thread_pool_t *  pool)

Destroy a thread pool.

Stops all threads (if not already stopped), waits for them to exit, and frees all pool resources. It's safe to call this even if threads are still running - they will be stopped first.

Parameters
poolThread pool to destroy (NULL is safe, does nothing)

Definition at line 223 of file thread_pool.c.

223 {
224 if (!pool) {
225 return;
226 }
227
228 log_debug("Destroying thread pool '%s' (thread_count=%zu, work_queue_mode=%d)", pool->name, pool->thread_count,
229 pool->is_work_queue_mode);
230
231 // Stop all threads first (if not already stopped)
232 if (pool->thread_count > 0) {
233 log_debug("Thread pool '%s' has %zu threads, stopping them first", pool->name, pool->thread_count);
235 log_debug("After thread_pool_stop_all: pool->threads=%p, pool->thread_count=%zu", (void *)pool->threads,
236 pool->thread_count);
237 }
238
239 mutex_lock(&pool->threads_mutex);
240 bool threads_still_running = pool->thread_count > 0;
241 mutex_unlock(&pool->threads_mutex);
242 if (threads_still_running) {
243 // A timed-out thread still owns access to this pool. Retain the pool and
244 // its synchronization objects so that worker cannot race freed memory.
245 log_error("Keeping thread pool '%s' alive because %zu worker thread(s) have not exited", pool->name,
246 pool->thread_count);
247 return;
248 }
249
250 // Clean up work queue (in case there's pending work)
251 if (pool->is_work_queue_mode) {
252 mutex_lock(&pool->work_queue_mutex);
253 thread_pool_work_entry_t *entry = pool->work_queue;
254 while (entry) {
255 thread_pool_work_entry_t *next = entry->next;
256 SAFE_FREE(entry);
257 entry = next;
258 }
259 pool->work_queue = NULL;
260 mutex_unlock(&pool->work_queue_mutex);
261
262 // Destroy condition variable and work queue mutex
263 cond_destroy(&pool->work_available);
264 mutex_destroy(&pool->work_queue_mutex);
265 }
266
267 // Destroy threads mutex
268 mutex_destroy(&pool->threads_mutex);
269
271
272 // Free pool
273 log_debug("About to free thread pool '%s' at %p", pool->name, (void *)pool);
275
276 log_debug("Thread pool destroyed (freed memory)");
277}
#define NAMED_UNREGISTER(ptr)
Unregister a pointer.
int cond_destroy(cond_t *cond)
Destroy a condition variable.
Work queue entry for task-based thread pools.
Definition thread_pool.h:87
struct thread_pool_work_entry * next
Linked list next pointer.
Definition thread_pool.h:90
asciichat_error_t thread_pool_stop_all(thread_pool_t *pool)
Stop all threads in the pool in stop_id order.

References cond_destroy(), log_debug, log_error, mutex_destroy(), mutex_lock, mutex_unlock, NAMED_UNREGISTER, thread_pool_work_entry::next, pool, SAFE_FREE, and thread_pool_stop_all().

Referenced by thread_pool_create_with_workers().

◆ thread_pool_get_count()

size_t thread_pool_get_count ( const thread_pool_t *  pool)

Get thread count in the pool.

Thread-safe count of active threads in the pool.

Parameters
poolThread pool to query
Returns
Number of threads in pool, or 0 if pool is NULL

Definition at line 490 of file thread_pool.c.

490 {
491 if (!pool) {
492 return 0;
493 }
494
495 // Note: We're not locking here for performance
496 // thread_count is a simple size_t which should be atomic on most platforms
497 // If strict thread safety is required, use mutex_lock/unlock
498 return pool->thread_count;
499}

References pool.

Referenced by tcp_server_get_thread_count(), tcp_server_spawn_thread(), and thread_pool_has_threads().

◆ thread_pool_has_threads()

bool thread_pool_has_threads ( const thread_pool_t *  pool)

Check if pool has any threads.

Convenience function equivalent to thread_pool_get_count(pool) > 0.

Parameters
poolThread pool to check
Returns
true if pool has threads, false if empty or NULL

Definition at line 501 of file thread_pool.c.

501 {
502 return thread_pool_get_count(pool) > 0;
503}
size_t thread_pool_get_count(const thread_pool_t *pool)
Get thread count in the pool.

References pool, and thread_pool_get_count().

◆ thread_pool_interrupt_all()

asciichat_error_t thread_pool_interrupt_all ( thread_pool_t *  pool,
int  sig 
)

Send a signal to all threads in the pool.

Sends the specified signal to every thread in the pool. This is used to interrupt threads that are blocked in system calls (e.g., recv(), connect()) so they unblock and can check the global exit flag.

Thread-safe: acquires threads_mutex to iterate the thread list. Not async-signal-safe: cannot be called from signal handlers.

On POSIX systems, uses pthread_kill() to send the signal. On Windows, this is a no-op (pthread_kill is not available, but socket_shutdown() is sufficient for Windows socket unblocking).

Typically used with SIGUSR1 (which should have a registered no-op handler) or another signal that the threads are prepared to handle gracefully.

Parameters
poolThread pool to interrupt
sigSignal number to send (e.g., SIGUSR1)
Returns
ASCIICHAT_OK on success, error code on failure

Definition at line 505 of file thread_pool.c.

505 {
506 if (!pool) {
507 return SET_ERRNO(ERROR_INVALID_PARAM, "pool is NULL");
508 }
509
510#ifndef _WIN32
511 // Only available on POSIX systems (Unix/Linux/macOS)
512 mutex_lock(&pool->threads_mutex);
513
514 if (pool->thread_count == 0) {
515 mutex_unlock(&pool->threads_mutex);
516 log_debug("Thread pool '%s' has no threads to interrupt", pool->name);
517 return ASCIICHAT_OK;
518 }
519
520 log_debug("Sending signal %d to %zu threads in pool '%s'", sig, pool->thread_count, pool->name);
521
522 // Iterate through all threads and send signal
523 thread_pool_entry_t *entry = pool->threads;
524 int sent_count = 0;
525 int failed_count = 0;
526
527 while (entry) {
528 // pthread_kill returns 0 on success, non-zero error code on failure
529 if (pthread_kill(entry->thread, sig) != 0) {
530 log_warn("Failed to send signal %d to thread '%s' in pool '%s'", sig, entry->name, pool->name);
531 failed_count++;
532 } else {
533 sent_count++;
534 }
535 entry = entry->next;
536 }
537
538 mutex_unlock(&pool->threads_mutex);
539
540 log_debug("Sent signal %d to %d/%zu threads in pool '%s' (%d failures)", sig, sent_count, pool->thread_count,
541 pool->name, failed_count);
542
543 if (failed_count > 0 && sent_count == 0) {
544 return SET_ERRNO(ERROR_THREAD, "Failed to send signal to any threads in pool '%s'", pool->name);
545 }
546
547 return ASCIICHAT_OK;
548#else
549 // Windows: no pthread_kill equivalent needed - socket_shutdown() is sufficient
550 (void)sig; // Unused on Windows
551 log_debug("thread_pool_interrupt_all: no-op on Windows (socket shutdown is sufficient)");
552 return ASCIICHAT_OK;
553#endif
554}
@ ASCIICHAT_OK
Definition error_codes.h:51
#define log_warn(...)
Log a WARN message.
Definition log/log.h:574

References ASCIICHAT_OK, ERROR_INVALID_PARAM, ERROR_THREAD, log_debug, log_warn, mutex_lock, mutex_unlock, thread_pool_entry::name, thread_pool_entry::next, pool, SET_ERRNO, and thread_pool_entry::thread.

◆ thread_pool_queue_work()

asciichat_error_t thread_pool_queue_work ( const char *  name,
thread_pool_t *  pool,
void *(*)(void *)  work_func,
void *  work_arg 
)

Queue a named work item to a work queue pool.

Adds a named work item to the pool's work queue. One of the pre-created worker threads will pick up the work and execute it. This function is only valid for pools created with thread_pool_create_with_workers(). The work is automatically registered with the debug naming system.

Safe to call from callback context (e.g., LWS callbacks). Work is queued without creating new threads.

Parameters
nameDebug name for the work item
poolThread pool to queue work to (must be in work queue mode)
work_funcFunction to execute
work_argArgument passed to work function
Returns
ASCIICHAT_OK on success, error code on failure

Definition at line 279 of file thread_pool.c.

280 {
281 if (!name) {
282 return SET_ERRNO(ERROR_INVALID_PARAM, "work name is required");
283 }
284
285 if (!pool) {
286 return SET_ERRNO(ERROR_INVALID_PARAM, "pool is NULL");
287 }
288
289 if (!pool->is_work_queue_mode) {
290 return SET_ERRNO(ERROR_INVALID_STATE, "Pool not in work queue mode. Use thread_pool_create_with_workers()");
291 }
292
293 if (!work_func) {
294 return SET_ERRNO(ERROR_INVALID_PARAM, "work_func is NULL");
295 }
296
297 // Allocate work entry
299 if (!entry) {
300 return SET_ERRNO(ERROR_MEMORY, "Failed to allocate work entry");
301 }
302
303 entry->work_func = work_func;
304 entry->work_arg = work_arg;
305 entry->next = NULL;
306
307 // Register work with debug naming system
308 NAMED_REGISTER_THREADPOOL_WORK(entry, name, NULL);
309
310 // Add to work queue (at the end)
311 mutex_lock(&pool->work_queue_mutex);
312
313 // Find the last entry in the queue
314 if (!pool->work_queue) {
315 pool->work_queue = entry;
316 } else {
317 thread_pool_work_entry_t *last = pool->work_queue;
318 while (last->next) {
319 last = last->next;
320 }
321 last->next = entry;
322 }
323
324 // Signal workers that work is available (WHILE HOLDING MUTEX to prevent lost wakeup race)
325 cond_signal(&pool->work_available);
326
327 mutex_unlock(&pool->work_queue_mutex);
328
329 log_dev("Queued work to pool '%s' (workers=%zu)", pool->name, pool->num_workers);
330 return ASCIICHAT_OK;
331}
#define NAMED_REGISTER_THREADPOOL_WORK(work, name, parent_ptr)
Register a thread pool work item with automatic format specifier.
@ ERROR_INVALID_STATE
#define log_dev(...)
Log a DEV message (most verbose, development only)
Definition log/log.h:534
int cond_signal(cond_t *cond)
Signal a condition variable (wake one waiting thread)
void * work_arg
Argument passed to work function.
Definition thread_pool.h:89
void *(* work_func)(void *)
Work function to execute.
Definition thread_pool.h:88

References ASCIICHAT_OK, cond_signal(), ERROR_INVALID_PARAM, ERROR_INVALID_STATE, ERROR_MEMORY, log_dev, mutex_lock, mutex_unlock, NAMED_REGISTER_THREADPOOL_WORK, thread_pool_work_entry::next, pool, SAFE_MALLOC, SET_ERRNO, thread_pool_work_entry::work_arg, and thread_pool_work_entry::work_func.

◆ thread_pool_spawn()

asciichat_error_t thread_pool_spawn ( thread_pool_t *  pool,
void *(*)(void *)  thread_func,
void *  thread_arg,
int  stop_id,
const char *  thread_name 
)

Spawn a worker thread in the pool.

Creates and tracks a new worker thread with optional ordered cleanup. Threads are inserted into the pool's linked list in sorted order by stop_id (ascending). When the pool is stopped, threads with lower stop_id values are stopped first.

Stop ID Ordering Guidelines:

  • stop_id < 0: Unordered (stopped last, order undefined)
  • stop_id = 0: General background workers
  • stop_id = 1: Data source threads (e.g., receive, capture)
  • stop_id = 2: Processing threads (e.g., render, encode)
  • stop_id = 3: Data sink threads (e.g., send, write)

Example (Server Per-Client Threads):

  • stop_id=1: Receive thread (stop first to prevent new data)
  • stop_id=2: Render threads (stop after receive)
  • stop_id=3: Send thread (stop last after all processing)
Parameters
poolThread pool to spawn into
thread_funcThread function to execute
thread_argArgument passed to thread function
stop_idCleanup order (-1=unordered, 0+=ordered)
thread_nameThread name for debugging (max 63 chars, NULL=auto-generate)
Returns
ASCIICHAT_OK on success, error code on failure

Definition at line 333 of file thread_pool.c.

334 {
335 if (!pool) {
336 return SET_ERRNO(ERROR_INVALID_PARAM, "pool is NULL");
337 }
338
339 if (pool->is_work_queue_mode) {
340 return SET_ERRNO(ERROR_INVALID_STATE, "Cannot spawn threads in work queue mode. Use thread_pool_queue_work()");
341 }
342
343 if (!thread_func) {
344 return SET_ERRNO(ERROR_INVALID_PARAM, "thread_func is NULL");
345 }
346
347 // Allocate thread entry
349 if (!entry) {
350 return SET_ERRNO(ERROR_MEMORY, "Failed to allocate thread pool entry");
351 }
352
353 memset(entry, 0, sizeof(*entry));
354 entry->stop_id = stop_id;
355 entry->thread_func = thread_func;
356 entry->thread_arg = thread_arg;
357 entry->next = NULL;
358
359 // Copy thread name (truncate if necessary)
360 if (thread_name) {
361 SAFE_STRNCPY(entry->name, thread_name, sizeof(entry->name));
362 } else {
363 // Auto-generate name
364 SAFE_SNPRINTF(entry->name, sizeof(entry->name), "%s_worker_%d", pool->name, stop_id);
365 }
366
367 // Create thread
368 if (asciichat_thread_create(&entry->thread, entry->name, thread_func, thread_arg) != 0) {
369 // Save name before freeing entry
370 char thread_name_copy[64];
371 SAFE_STRNCPY(thread_name_copy, entry->name, sizeof(thread_name_copy));
372 SAFE_FREE(entry);
373 return SET_ERRNO(ERROR_THREAD, "Failed to create thread '%s' in pool '%s'", thread_name_copy, pool->name);
374 }
375
376 // Add to thread list (sorted by stop_id)
377 mutex_lock(&pool->threads_mutex);
378
379 if (!pool->threads || pool->threads->stop_id > stop_id) {
380 // Insert at head
381 entry->next = pool->threads;
382 pool->threads = entry;
383 } else {
384 // Find insertion point (maintain sorted order by stop_id)
385 thread_pool_entry_t *prev = pool->threads;
386 while (prev->next && prev->next->stop_id <= stop_id) {
387 prev = prev->next;
388 }
389 entry->next = prev->next;
390 prev->next = entry;
391 }
392
393 pool->thread_count++;
394 mutex_unlock(&pool->threads_mutex);
395
396 log_debug("Spawned thread '%s' (stop_id=%d) in pool '%s' (total_threads=%zu)", entry->name, stop_id, pool->name,
397 pool->thread_count);
398
399 return ASCIICHAT_OK;
400}

References ASCIICHAT_OK, asciichat_thread_create, ERROR_INVALID_PARAM, ERROR_INVALID_STATE, ERROR_MEMORY, ERROR_THREAD, log_debug, mutex_lock, mutex_unlock, thread_pool_entry::name, thread_pool_entry::next, pool, SAFE_FREE, SAFE_MALLOC, SAFE_SNPRINTF, SAFE_STRNCPY, SET_ERRNO, thread_pool_entry::stop_id, thread_pool_entry::thread, thread_pool_entry::thread_arg, and thread_pool_entry::thread_func.

◆ thread_pool_stop_all()

asciichat_error_t thread_pool_stop_all ( thread_pool_t *  pool)

Stop all threads in the pool in stop_id order.

Joins all threads in ascending stop_id order (lower values first). Threads with negative stop_id are stopped last in undefined order. This function blocks until all threads have exited.

Shutdown Sequence:

  1. Threads with stop_id >= 0 are joined in ascending order
  2. Threads with stop_id < 0 are joined in any order
  3. All thread entries are freed

After this call, the pool is empty but still valid - you can spawn new threads or destroy the pool.

Parameters
poolThread pool to stop
Returns
ASCIICHAT_OK on success, error code on failure

Definition at line 402 of file thread_pool.c.

402 {
403 if (!pool) {
404 return SET_ERRNO(ERROR_INVALID_PARAM, "pool is NULL");
405 }
406
407 mutex_lock(&pool->threads_mutex);
408
409 if (pool->thread_count == 0) {
410 mutex_unlock(&pool->threads_mutex);
411 log_debug("Thread pool '%s' has no threads to stop", pool->name);
412 return ASCIICHAT_OK;
413 }
414
415 log_debug("Stopping %zu threads in pool '%s' in stop_id order", pool->thread_count, pool->name);
416
417 // For work queue mode, signal workers to shutdown (BEFORE releasing the mutex to prevent new work)
418 if (pool->is_work_queue_mode) {
419 mutex_lock(&pool->work_queue_mutex);
420 pool->shutdown_requested = true;
421 cond_broadcast(&pool->work_available);
422 mutex_unlock(&pool->work_queue_mutex);
423 }
424
425 // Detach the current list so workers can finish without holding the pool
426 // mutex. Any thread that misses its join deadline is reinserted below.
427 thread_pool_entry_t *threads_to_join = pool->threads;
428 pool->threads = NULL;
429 pool->thread_count = 0;
430
431 // Release mutex BEFORE joining threads to prevent deadlock if workers need to acquire it
432 mutex_unlock(&pool->threads_mutex);
433
434 // Now join threads (WITHOUT holding the mutex to prevent deadlock)
435 thread_pool_entry_t *entry = threads_to_join;
436 int joined_count = 0, freed_count = 0, still_running_count = 0;
437 thread_pool_entry_t *still_running = NULL;
438 const uint64_t THREAD_JOIN_TIMEOUT_NS = 2 * 1000000000ULL; // 2 second timeout per thread
439
440 while (entry) {
441 log_debug("Joining thread '%s' (stop_id=%d) in pool '%s' with 2s timeout", entry->name, entry->stop_id, pool->name);
442 thread_pool_entry_t *next = entry->next;
443
444 // A timed-out worker may still access its pool and arguments. Keep its
445 // entry tracked so later shutdown can join it before freeing those objects.
446 int join_result = asciichat_thread_join_timeout(&entry->thread, NULL, THREAD_JOIN_TIMEOUT_NS);
447 if (join_result != 0) {
448 if (join_result == ETIMEDOUT) {
449 log_warn("Thread '%s' (stop_id=%d) did not exit within 2s timeout; retaining its pool entry.", entry->name,
450 entry->stop_id);
451 } else {
452 log_error("Failed to join thread '%s' in pool '%s' (error %d); retaining its pool entry", entry->name,
453 pool->name, join_result);
454 }
455 entry->next = still_running;
456 still_running = entry;
457 still_running_count++;
458 entry = next;
459 continue;
460 } else {
461 joined_count++;
462 }
463
464 log_debug("Freeing thread_pool_entry '%s' at %p", entry->name, (void *)entry);
465 SAFE_FREE(entry);
466 freed_count++;
467 entry = next;
468 }
469
470 if (still_running) {
471 mutex_lock(&pool->threads_mutex);
472 while (still_running) {
473 thread_pool_entry_t *entry = still_running;
474 still_running = still_running->next;
475 entry->next = NULL;
476 thread_pool_insert_entry_locked(pool, entry);
477 }
478 mutex_unlock(&pool->threads_mutex);
479 }
480
481 log_debug("thread_pool_stop_all: joined=%d, freed=%d", joined_count, freed_count);
482
483 log_debug("All threads stopped in pool '%s'", pool->name);
484 if (still_running_count > 0) {
485 return SET_ERRNO(ERROR_THREAD, "%d thread(s) in pool '%s' did not stop", still_running_count, pool->name);
486 }
487 return ASCIICHAT_OK;
488}
unsigned long long uint64_t
Definition common.h:59
int asciichat_thread_join_timeout(asciichat_thread_t *thread, void **retval, uint64_t timeout_ns)
Wait for a thread to complete with timeout.
#define ETIMEDOUT
int cond_broadcast(cond_t *cond)
Broadcast to a condition variable (wake all waiting threads)

References ASCIICHAT_OK, asciichat_thread_join_timeout(), cond_broadcast(), ERROR_INVALID_PARAM, ERROR_THREAD, ETIMEDOUT, log_debug, log_error, log_warn, mutex_lock, mutex_unlock, thread_pool_entry::name, thread_pool_entry::next, pool, SAFE_FREE, SET_ERRNO, thread_pool_entry::stop_id, and thread_pool_entry::thread.

Referenced by thread_pool_destroy().