thread_pool.h File Reference

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

Go to the source code of this file.

Data Structures
struct	thread_pool_work_entry
	One queued function call waiting for a thread-pool worker. More...
struct	thread_pool_entry
	Thread handle and shutdown metadata for one pool-owned thread. More...
struct	thread_pool
	Thread pool structure. More...

Macros
#define	THREAD_POOL_NAME_BUFFER_SIZE   64

Typedefs
typedef struct thread_pool_entry	thread_pool_entry_t
typedef struct thread_pool_work_entry	thread_pool_work_entry_t
	One queued function call waiting for a thread-pool worker.

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 abstraction for managing worker threads

Provides a reusable thread pool implementation for managing multiple worker threads with ordered cleanup support. This abstraction is used by:

• tcp_server: Per-client thread pools for receive/send/render workers
• Server: Background threads (stats logger, etc.)
• ACDS: Background threads (rate limit cleanup, etc.)
• Client: Capture and display threads

Features

• Thread lifecycle management: Create, spawn, stop, and join threads
• Ordered cleanup: Threads with lower stop_id values are stopped first
• Thread naming: Assign names to threads for debugging
• Thread-safe: All operations are protected with mutexes
• Flexible: Supports both entity-bound pools and global singleton pools

Usage Pattern

// Create thread pool
thread_pool_t *pool = thread_pool_create("my_pool");
 
// Spawn workers with ordered cleanup
thread_pool_spawn(pool, worker_func1, arg1, 1, "worker1"); // Stop first
thread_pool_spawn(pool, worker_func2, arg2, 2, "worker2"); // Stop second
thread_pool_spawn(pool, worker_func3, arg3, 3, "worker3"); // Stop last
 
// Stop all threads in stop_id order (1 → 2 → 3)
thread_pool_stop_all(pool);
 
// Destroy pool (frees all resources)
thread_pool_destroy(pool);

Ordered Cleanup Example (Server Per-Client Threads)

// Client connection established
thread_pool_spawn(pool, receive_thread, client, 1, "receive_1");
thread_pool_spawn(pool, video_render, client, 2, "video_2");
thread_pool_spawn(pool, audio_render, client, 2, "audio_2");
thread_pool_spawn(pool, send_thread, client, 3, "send_3");
 
// Client disconnect - threads stopped in order:
// 1. Receive thread (stop_id=1) - no new data arrives
// 2. Render threads (stop_id=2) - no new frames generated
// 3. Send thread (stop_id=3) - all queued data sent
thread_pool_stop_all(pool);

Background Thread Example (Stats Logger)

// Server startup
thread_pool_t *server_pool = thread_pool_create("server");
thread_pool_spawn(server_pool, stats_logger, NULL, 0, "stats");
 
// Server shutdown
thread_pool_stop_all(server_pool);
thread_pool_destroy(server_pool);

Author

Zachary Fogg me@zf.nosp@m.o.gg

Date

December 2025

Definition in file thread_pool.h.

Macro Definition Documentation

THREAD_POOL_NAME_BUFFER_SIZE

#define THREAD_POOL_NAME_BUFFER_SIZE   64

Definition at line 74 of file thread_pool.h.

Typedef Documentation

thread_pool_entry_t

typedef struct thread_pool_entry thread_pool_entry_t

Definition at line 90 of file thread_pool.h.

thread_pool_work_entry_t

typedef struct thread_pool_work_entry thread_pool_work_entry_t

One queued function call waiting for a thread-pool worker.

The pool stores the callback, its opaque argument, and the link to the next queued task. Queue ownership transfers to the pool when a task is submitted; workers remove and free entries after invoking the callback.

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_name

Name for the pool (max 63 chars, used for debugging)

Returns

Pointer to new thread pool, or NULL on allocation failure

Definition at line 92 of file thread_pool.c.

                                                         {
  thread_pool_t *pool = SAFE_MALLOC(sizeof(thread_pool_t), thread_pool_t *);
  if (!pool) {
    SET_ERRNO(ERROR_MEMORY, "Failed to allocate thread pool");
    return NULL;
  }
 
  memset(pool, 0, sizeof(*pool));
 
  // Copy pool name (truncate if necessary)
  if (pool_name) {
    SAFE_STRNCPY(pool->name, pool_name, sizeof(pool->name));
  } else {
    SAFE_STRNCPY(pool->name, "unnamed", sizeof(pool->name));
  }
 
  // Initialize linked list
  pool->threads = NULL;
  pool->thread_count = 0;
  pool->is_work_queue_mode = false;
  pool->num_workers = 0;
 
  // Initialize mutex
  if (mutex_init(&pool->threads_mutex, "thread_pool") != 0) {
    SAFE_FREE(pool);
    SET_ERRNO(ERROR_THREAD, "Failed to initialize thread pool mutex");
    return NULL;
  }
 
  NAMED_REGISTER_THREAD_POOL(pool, pool->name, NULL);
 
  // Register sync primitives for debugging
  NAMED_REGISTER_MUTEX(&pool->threads_mutex, "threads_mutex", (uintptr_t)(const void *)(pool));
 
  log_debug("Thread pool '%s' created (long-lived thread mode)", pool->name);
  return pool;
}

References ERROR_MEMORY, ERROR_THREAD, thread_pool::is_work_queue_mode, log_debug, thread_pool::name, NAMED_REGISTER_MUTEX, NAMED_REGISTER_THREAD_POOL, thread_pool::num_workers, SAFE_FREE, SAFE_MALLOC, SAFE_STRNCPY, SET_ERRNO, thread_pool::thread_count, thread_pool::threads, and thread_pool::threads_mutex.

Referenced by acds_server_init(), client_connection_create(), network_server_create(), and server_main().

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_name	Name for the pool (max 63 chars, used for debugging)
num_workers	Number of pre-allocated worker threads

Returns

Pointer to new thread pool, or NULL on allocation failure

Definition at line 130 of file thread_pool.c.

                                                                                          {
  if (num_workers == 0) {
    SET_ERRNO(ERROR_INVALID_PARAM, "num_workers must be > 0");
    return NULL;
  }
 
  thread_pool_t *pool = SAFE_MALLOC(sizeof(thread_pool_t), thread_pool_t *);
  if (!pool) {
    SET_ERRNO(ERROR_MEMORY, "Failed to allocate thread pool");
    return NULL;
  }
 
  memset(pool, 0, sizeof(*pool));
 
  // Copy pool name
  if (pool_name) {
    SAFE_STRNCPY(pool->name, pool_name, sizeof(pool->name));
  } else {
    SAFE_STRNCPY(pool->name, "unnamed", sizeof(pool->name));
  }
 
  // Mark as work queue mode
  pool->is_work_queue_mode = true;
  pool->num_workers = num_workers;
  pool->work_queue = NULL;
  pool->shutdown_requested = false;
 
  // Initialize mutexes
  if (mutex_init(&pool->threads_mutex, "worker_list") != 0) {
    SAFE_FREE(pool);
    SET_ERRNO(ERROR_THREAD, "Failed to initialize threads_mutex");
    return NULL;
  }
 
  if (mutex_init(&pool->work_queue_mutex, "pending_tasks") != 0) {
    mutex_destroy(&pool->threads_mutex);
    SAFE_FREE(pool);
    SET_ERRNO(ERROR_THREAD, "Failed to initialize work_queue_mutex");
    return NULL;
  }
 
  // Initialize condition variable
  if (cond_init(&pool->work_available, "task_available") != 0) {
    mutex_destroy(&pool->work_queue_mutex);
    mutex_destroy(&pool->threads_mutex);
    SAFE_FREE(pool);
    SET_ERRNO(ERROR_THREAD, "Failed to initialize work_available condition");
    return NULL;
  }
 
  // Create worker threads
  for (size_t i = 0; i < num_workers; i++) {
    thread_pool_entry_t *entry = SAFE_MALLOC(sizeof(thread_pool_entry_t), thread_pool_entry_t *);
    if (!entry) {
      log_error("Failed to allocate worker entry %zu", i);
      thread_pool_destroy(pool);
      SET_ERRNO(ERROR_MEMORY, "Failed to allocate worker entry");
      return NULL;
    }
 
    memset(entry, 0, sizeof(*entry));
    entry->stop_id = 0;
    entry->thread_func = thread_pool_worker_thread;
    entry->thread_arg = pool;
    SAFE_SNPRINTF(entry->name, sizeof(entry->name), "%s_worker_%zu", pool->name, i);
 
    // Create worker thread
    if (platform_thread_create(&entry->thread, entry->name, thread_pool_worker_thread, pool) != 0) {
      log_error("Failed to create worker thread %zu", i);
      SAFE_FREE(entry);
      thread_pool_destroy(pool);
      SET_ERRNO(ERROR_THREAD, "Failed to create worker thread %zu", i);
      return NULL;
    }
 
    // Add to thread list
    mutex_lock(&pool->threads_mutex);
    entry->next = pool->threads;
    pool->threads = entry;
    pool->thread_count++;
    mutex_unlock(&pool->threads_mutex);
 
    log_debug("[ThreadPool] Created worker thread %zu for pool '%s'", i, pool->name);
  }
 
  NAMED_REGISTER_THREAD_POOL(pool, pool->name, NULL);
 
  // Register sync primitives for debugging
  NAMED_REGISTER_MUTEX(&pool->threads_mutex, "threads_mutex", (uintptr_t)(const void *)(pool));
  NAMED_REGISTER_MUTEX(&pool->work_queue_mutex, "work_queue_mutex", (uintptr_t)(const void *)(pool));
  NAMED_REGISTER_COND(&pool->work_available, "work_available", (uintptr_t)(const void *)(pool));
 
  log_info("[ThreadPool] Created work queue pool '%s' with %zu worker threads", pool->name, num_workers);
  return pool;
}

References ERROR_INVALID_PARAM, ERROR_MEMORY, ERROR_THREAD, thread_pool::is_work_queue_mode, log_debug, log_error, log_info, thread_pool::name, thread_pool_entry::name, NAMED_REGISTER_COND, NAMED_REGISTER_MUTEX, NAMED_REGISTER_THREAD_POOL, thread_pool_entry::next, thread_pool::num_workers, SAFE_FREE, SAFE_MALLOC, SAFE_SNPRINTF, SAFE_STRNCPY, SET_ERRNO, thread_pool::shutdown_requested, thread_pool_entry::stop_id, thread_pool_entry::thread, thread_pool_entry::thread_arg, thread_pool::thread_count, thread_pool_entry::thread_func, thread_pool_destroy(), thread_pool::threads, thread_pool::threads_mutex, thread_pool::work_available, thread_pool::work_queue, and thread_pool::work_queue_mutex.

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

pool	Thread pool to destroy (NULL is safe, does nothing)

Definition at line 226 of file thread_pool.c.

                                              {
  if (!pool) {
    return;
  }
 
  log_debug("Destroying thread pool '%s' (thread_count=%zu, work_queue_mode=%d)", pool->name, pool->thread_count,
            pool->is_work_queue_mode);
 
  // Stop all threads first (if not already stopped)
  if (pool->thread_count > 0) {
    log_debug("Thread pool '%s' has %zu threads, stopping them first", pool->name, pool->thread_count);
    thread_pool_stop_all(pool);
    log_debug("After thread_pool_stop_all: pool->threads=%p, pool->thread_count=%zu", (void *)pool->threads,
              pool->thread_count);
  }
 
  mutex_lock(&pool->threads_mutex);
  bool threads_still_running = pool->thread_count > 0;
  mutex_unlock(&pool->threads_mutex);
  if (threads_still_running) {
    // A timed-out thread still owns access to this pool. Retain the pool and
    // its synchronization objects so that worker cannot race freed memory.
    log_error("Keeping thread pool '%s' alive because %zu worker thread(s) have not exited", pool->name,
              pool->thread_count);
    return;
  }
 
  // Clean up work queue (in case there's pending work)
  if (pool->is_work_queue_mode) {
    mutex_lock(&pool->work_queue_mutex);
    thread_pool_work_entry_t *entry = pool->work_queue;
    while (entry) {
      thread_pool_work_entry_t *next = entry->next;
      SAFE_FREE(entry);
      entry = next;
    }
    pool->work_queue = NULL;
    mutex_unlock(&pool->work_queue_mutex);
 
    // Destroy condition variable and work queue mutex
    cond_destroy(&pool->work_available);
    mutex_destroy(&pool->work_queue_mutex);
  }
 
  // Destroy threads mutex
  mutex_destroy(&pool->threads_mutex);
 
  NAMED_UNREGISTER(pool);
 
  // Free pool
  log_debug("About to free thread pool '%s' at %p", pool->name, (void *)pool);
  SAFE_FREE(pool);
 
  log_debug("Thread pool destroyed (freed memory)");
}

References thread_pool::is_work_queue_mode, log_debug, log_error, thread_pool::name, NAMED_UNREGISTER, thread_pool_work_entry::next, SAFE_FREE, thread_pool::thread_count, thread_pool_stop_all(), thread_pool::threads, thread_pool::threads_mutex, thread_pool::work_available, thread_pool::work_queue, and thread_pool::work_queue_mutex.

Referenced by acds_server_shutdown(), client_connection_destroy(), network_server_destroy(), thread_pool_create_with_workers(), and websocket_server_destroy().

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

pool	Thread pool to query

Returns

Number of threads in pool, or 0 if pool is NULL

Definition at line 493 of file thread_pool.c.

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

References thread_pool::thread_count.

Referenced by 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

pool	Thread pool to check

Returns

true if pool has threads, false if empty or NULL

Definition at line 504 of file thread_pool.c.

                                                        {
  return thread_pool_get_count(pool) > 0;
}

References 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

pool	Thread pool to interrupt
sig	Signal number to send (e.g., SIGUSR1)

Returns

ASCIICHAT_OK on success, error code on failure

Definition at line 508 of file thread_pool.c.

                                                                          {
  if (!pool) {
    return SET_ERRNO(ERROR_INVALID_PARAM, "pool is NULL");
  }
 
#ifndef _WIN32
  // Only available on POSIX systems (Unix/Linux/macOS)
  mutex_lock(&pool->threads_mutex);
 
  if (pool->thread_count == 0) {
    mutex_unlock(&pool->threads_mutex);
    log_debug("Thread pool '%s' has no threads to interrupt", pool->name);
    return ASCIICHAT_OK;
  }
 
  log_debug("Sending signal %d to %zu threads in pool '%s'", sig, pool->thread_count, pool->name);
 
  // Iterate through all threads and send signal
  thread_pool_entry_t *entry = pool->threads;
  int sent_count = 0;
  int failed_count = 0;
 
  while (entry) {
    // pthread_kill returns 0 on success, non-zero error code on failure
    if (pthread_kill(entry->thread, sig) != 0) {
      log_warn("Failed to send signal %d to thread '%s' in pool '%s'", sig, entry->name, pool->name);
      failed_count++;
    } else {
      sent_count++;
    }
    entry = entry->next;
  }
 
  mutex_unlock(&pool->threads_mutex);
 
  log_debug("Sent signal %d to %d/%zu threads in pool '%s' (%d failures)", sig, sent_count, pool->thread_count,
            pool->name, failed_count);
 
  if (failed_count > 0 && sent_count == 0) {
    return SET_ERRNO(ERROR_THREAD, "Failed to send signal to any threads in pool '%s'", pool->name);
  }
 
  return ASCIICHAT_OK;
#else
  // Windows: no pthread_kill equivalent needed - socket_shutdown() is sufficient
  (void)sig; // Unused on Windows
  log_debug("no-op on Windows (socket shutdown is sufficient)");
  return ASCIICHAT_OK;
#endif
}

References ASCIICHAT_OK, ERROR_INVALID_PARAM, ERROR_THREAD, log_debug, log_warn, thread_pool::name, thread_pool_entry::name, thread_pool_entry::next, SET_ERRNO, thread_pool_entry::thread, thread_pool::thread_count, thread_pool::threads, and thread_pool::threads_mutex.

thread_pool_queue_work()

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.

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

name	Debug name for the work item
pool	Thread pool to queue work to (must be in work queue mode)
work_func	Function to execute
work_arg	Argument passed to work function

Returns

ASCIICHAT_OK on success, error code on failure

Definition at line 282 of file thread_pool.c.

                                                         {
  if (!name) {
    return SET_ERRNO(ERROR_INVALID_PARAM, "work name is required");
  }
 
  if (!pool) {
    return SET_ERRNO(ERROR_INVALID_PARAM, "pool is NULL");
  }
 
  if (!pool->is_work_queue_mode) {
    return SET_ERRNO(ERROR_INVALID_STATE, "Pool not in work queue mode. Use thread_pool_create_with_workers()");
  }
 
  if (!work_func) {
    return SET_ERRNO(ERROR_INVALID_PARAM, "work_func is NULL");
  }
 
  // Allocate work entry
  thread_pool_work_entry_t *entry = SAFE_MALLOC(sizeof(thread_pool_work_entry_t), thread_pool_work_entry_t *);
  if (!entry) {
    return SET_ERRNO(ERROR_MEMORY, "Failed to allocate work entry");
  }
 
  entry->work_func = work_func;
  entry->work_arg = work_arg;
  entry->next = NULL;
 
  // Register work with debug naming system
  NAMED_REGISTER_THREADPOOL_WORK(entry, name, NULL);
 
  // Add to work queue (at the end)
  mutex_lock(&pool->work_queue_mutex);
 
  // Find the last entry in the queue
  if (!pool->work_queue) {
    pool->work_queue = entry;
  } else {
    thread_pool_work_entry_t *last = pool->work_queue;
    while (last->next) {
      last = last->next;
    }
    last->next = entry;
  }
 
  // Signal workers that work is available (WHILE HOLDING MUTEX to prevent lost wakeup race)
  cond_signal(&pool->work_available);
 
  mutex_unlock(&pool->work_queue_mutex);
 
  log_dev("Queued work to pool '%s' (workers=%zu)", pool->name, pool->num_workers);
  return ASCIICHAT_OK;
}

References ASCIICHAT_OK, ERROR_INVALID_PARAM, ERROR_INVALID_STATE, ERROR_MEMORY, thread_pool::is_work_queue_mode, log_dev, thread_pool::name, NAMED_REGISTER_THREADPOOL_WORK, thread_pool_work_entry::next, thread_pool::num_workers, SAFE_MALLOC, SET_ERRNO, thread_pool_work_entry::work_arg, thread_pool::work_available, thread_pool_work_entry::work_func, thread_pool::work_queue, and thread_pool::work_queue_mutex.

thread_pool_spawn()

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.

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

pool	Thread pool to spawn into
thread_func	Thread function to execute
thread_arg	Argument passed to thread function
stop_id	Cleanup order (-1=unordered, 0+=ordered)
thread_name	Thread name for debugging (max 63 chars, NULL=auto-generate)

Returns

ASCIICHAT_OK on success, error code on failure

Definition at line 336 of file thread_pool.c.

                                                             {
  if (!pool) {
    return SET_ERRNO(ERROR_INVALID_PARAM, "pool is NULL");
  }
 
  if (pool->is_work_queue_mode) {
    return SET_ERRNO(ERROR_INVALID_STATE, "Cannot spawn threads in work queue mode. Use thread_pool_queue_work()");
  }
 
  if (!thread_func) {
    return SET_ERRNO(ERROR_INVALID_PARAM, "thread_func is NULL");
  }
 
  // Allocate thread entry
  thread_pool_entry_t *entry = SAFE_MALLOC(sizeof(thread_pool_entry_t), thread_pool_entry_t *);
  if (!entry) {
    return SET_ERRNO(ERROR_MEMORY, "Failed to allocate thread pool entry");
  }
 
  memset(entry, 0, sizeof(*entry));
  entry->stop_id = stop_id;
  entry->thread_func = thread_func;
  entry->thread_arg = thread_arg;
  entry->next = NULL;
 
  // Copy thread name (truncate if necessary)
  if (thread_name) {
    SAFE_STRNCPY(entry->name, thread_name, sizeof(entry->name));
  } else {
    // Auto-generate name
    SAFE_SNPRINTF(entry->name, sizeof(entry->name), "%s_worker_%d", pool->name, stop_id);
  }
 
  // Create thread
  if (platform_thread_create(&entry->thread, entry->name, thread_func, thread_arg) != 0) {
    // Save name before freeing entry
    char thread_name_copy[THREAD_POOL_THREAD_NAME_COPY_BUFFER_SIZE];
    SAFE_STRNCPY(thread_name_copy, entry->name, sizeof(thread_name_copy));
    SAFE_FREE(entry);
    return SET_ERRNO(ERROR_THREAD, "Failed to create thread '%s' in pool '%s'", thread_name_copy, pool->name);
  }
 
  // Add to thread list (sorted by stop_id)
  mutex_lock(&pool->threads_mutex);
 
  if (!pool->threads || pool->threads->stop_id > stop_id) {
    // Insert at head
    entry->next = pool->threads;
    pool->threads = entry;
  } else {
    // Find insertion point (maintain sorted order by stop_id)
    thread_pool_entry_t *prev = pool->threads;
    while (prev->next && prev->next->stop_id <= stop_id) {
      prev = prev->next;
    }
    entry->next = prev->next;
    prev->next = entry;
  }
 
  pool->thread_count++;
  mutex_unlock(&pool->threads_mutex);
 
  log_debug("Spawned thread '%s' (stop_id=%d) in pool '%s' (total_threads=%zu)", entry->name, stop_id, pool->name,
            pool->thread_count);
 
  return ASCIICHAT_OK;
}

References ASCIICHAT_OK, ERROR_INVALID_PARAM, ERROR_INVALID_STATE, ERROR_MEMORY, ERROR_THREAD, thread_pool::is_work_queue_mode, log_debug, thread_pool::name, thread_pool_entry::name, thread_pool_entry::next, 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::thread_count, thread_pool_entry::thread_func, THREAD_POOL_THREAD_NAME_COPY_BUFFER_SIZE, thread_pool::threads, and thread_pool::threads_mutex.

Referenced by acds_server_init(), audio_sender_init(), audio_start_thread(), capture_start_thread(), client_protocol_start(), keepalive_start_thread(), and server_main().

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

pool	Thread pool to stop

Returns

ASCIICHAT_OK on success, error code on failure

Definition at line 405 of file thread_pool.c.

                                                            {
  if (!pool) {
    return SET_ERRNO(ERROR_INVALID_PARAM, "pool is NULL");
  }
 
  mutex_lock(&pool->threads_mutex);
 
  if (pool->thread_count == 0) {
    mutex_unlock(&pool->threads_mutex);
    log_debug("Thread pool '%s' has no threads to stop", pool->name);
    return ASCIICHAT_OK;
  }
 
  log_debug("Stopping %zu threads in pool '%s' in stop_id order", pool->thread_count, pool->name);
 
  // For work queue mode, signal workers to shutdown (BEFORE releasing the mutex to prevent new work)
  if (pool->is_work_queue_mode) {
    mutex_lock(&pool->work_queue_mutex);
    pool->shutdown_requested = true;
    cond_broadcast(&pool->work_available);
    mutex_unlock(&pool->work_queue_mutex);
  }
 
  // Detach the current list so workers can finish without holding the pool
  // mutex. Any thread that misses its join deadline is reinserted below.
  thread_pool_entry_t *threads_to_join = pool->threads;
  pool->threads = NULL;
  pool->thread_count = 0;
 
  // Release mutex BEFORE joining threads to prevent deadlock if workers need to acquire it
  mutex_unlock(&pool->threads_mutex);
 
  // Now join threads (WITHOUT holding the mutex to prevent deadlock)
  thread_pool_entry_t *entry = threads_to_join;
  int joined_count = 0, freed_count = 0, still_running_count = 0;
  thread_pool_entry_t *still_running = NULL;
  const uint64_t THREAD_JOIN_TIMEOUT_NS = 2 * 1000000000ULL; // 2 second timeout per thread
 
  while (entry) {
    log_debug("Joining thread '%s' (stop_id=%d) in pool '%s' with 2s timeout", entry->name, entry->stop_id, pool->name);
    thread_pool_entry_t *next = entry->next;
 
    // A timed-out worker may still access its pool and arguments. Keep its
    // entry tracked so later shutdown can join it before freeing those objects.
    int join_result = thread_join_timeout(&entry->thread, NULL, THREAD_JOIN_TIMEOUT_NS);
    if (join_result != 0) {
      if (join_result == ETIMEDOUT) {
        log_warn("Thread '%s' (stop_id=%d) did not exit within 2s timeout; retaining its pool entry.", entry->name,
                 entry->stop_id);
      } else {
        log_error("Failed to join thread '%s' in pool '%s' (error %d); retaining its pool entry", entry->name,
                  pool->name, join_result);
      }
      entry->next = still_running;
      still_running = entry;
      still_running_count++;
      entry = next;
      continue;
    } else {
      joined_count++;
    }
 
    log_debug("Freeing thread_pool_entry '%s' at %p", entry->name, (void *)entry);
    SAFE_FREE(entry);
    freed_count++;
    entry = next;
  }
 
  if (still_running) {
    mutex_lock(&pool->threads_mutex);
    while (still_running) {
      thread_pool_entry_t *entry = still_running;
      still_running = still_running->next;
      entry->next = NULL;
      thread_pool_insert_entry_locked(pool, entry);
    }
    mutex_unlock(&pool->threads_mutex);
  }
 
  log_debug("joined=%d, freed=%d", joined_count, freed_count);
 
  log_debug("All threads stopped in pool '%s'", pool->name);
  if (still_running_count > 0) {
    return SET_ERRNO(ERROR_THREAD, "%d thread(s) in pool '%s' did not stop", still_running_count, pool->name);
  }
  return ASCIICHAT_OK;
}

References ASCIICHAT_OK, ERROR_INVALID_PARAM, ERROR_THREAD, ETIMEDOUT, thread_pool::is_work_queue_mode, log_debug, log_error, log_warn, thread_pool::name, thread_pool_entry::name, thread_pool_entry::next, SAFE_FREE, SET_ERRNO, thread_pool::shutdown_requested, thread_pool_entry::stop_id, thread_pool_entry::thread, thread_pool::thread_count, thread_pool::threads, thread_pool::threads_mutex, thread_pool::work_available, and thread_pool::work_queue_mutex.

Referenced by network_server_stop(), and thread_pool_destroy().