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
	Work queue entry for task-based thread pools. More...
struct	thread_pool_entry
	Thread pool entry (internal linked list node) More...
struct	thread_pool
	Thread pool structure. More...

Typedefs
typedef struct thread_pool	thread_pool_t
typedef struct thread_pool_entry	thread_pool_entry_t
typedef struct thread_pool_work_entry	thread_pool_work_entry_t
	Work queue entry for task-based thread pools.

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.

Typedef Documentation

thread_pool_entry_t

typedef struct thread_pool_entry thread_pool_entry_t

Definition at line 80 of file thread_pool.h.

thread_pool_t

typedef struct thread_pool thread_pool_t

Definition at line 79 of file thread_pool.h.

thread_pool_work_entry_t

typedef struct thread_pool_work_entry thread_pool_work_entry_t

Work queue entry for task-based thread pools.

Represents a single unit of work to be executed by a pool worker.

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 89 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, 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_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 127 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 (asciichat_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 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

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

Definition at line 31 of file stubs/threading.c.

                                              {
  (void)pool;
}

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 acds_server_shutdown(), tcp_server_destroy(), tcp_server_remove_client(), 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 490 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 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

pool	Thread pool to check

Returns

true if pool has threads, false if empty or NULL

Definition at line 501 of file thread_pool.c.

                                                        {
  return thread_pool_get_count(pool) > 0;
}

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

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 505 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("thread_pool_interrupt_all: no-op on Windows (socket shutdown is sufficient)");
  return ASCIICHAT_OK;
#endif
}

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

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 279 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, 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

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 342 of file comprehensive.c.

                                                      {
  (void)pool;
  (void)thread_func;
  (void)thread_arg;
  (void)stop_id;
  (void)name;
  return ASCIICHAT_OK; // No-op in WASM (no real threads)
}

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.

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

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 352 of file comprehensive.c.

                                                   {
  (void)pool;
  return ASCIICHAT_OK; // No-op in WASM
}

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 protocol_stop_connection(), tcp_server_stop_client_threads(), and thread_pool_destroy().