ck_threadpool.h File Reference

Persistent pthread thread pool for CK-Engine inference. More...

#include <stdint.h>
#include <stdatomic.h>
#include <pthread.h>

Go to the source code of this file.

Macros
#define	CK_CACHE_LINE   64
#define	CK_THREADPOOL_MAX_THREADS   64
#define	CK_THREADPOOL_SPIN_COUNT   1024

Typedefs
typedef void(*	ck_work_fn_t) (int ith, int nth, void *args)

Functions
void	ck_threadpool_barrier (ck_threadpool_t *pool)
ck_threadpool_t *	ck_threadpool_create (int n_threads)
void	ck_threadpool_destroy (ck_threadpool_t *pool)
void	ck_threadpool_dispatch (ck_threadpool_t *pool, ck_work_fn_t fn, void *args)
ck_threadpool_t *	ck_threadpool_global (void)
void	ck_threadpool_global_destroy (void)
int	ck_threadpool_n_threads (const ck_threadpool_t *pool)
void	ck_threadpool_pause (ck_threadpool_t *pool)
void	ck_threadpool_resume (ck_threadpool_t *pool)
int	ck_threadpool_thread_id (const ck_threadpool_t *pool)

Detailed Description

Persistent pthread thread pool for CK-Engine inference.

Design goals:

• Sub-microsecond dispatch latency (spin-wait barriers)
• Zero allocation after init (all memory pre-allocated)
• Cache-line aligned atomics to avoid false sharing
• Hybrid polling: spin N rounds, then fall back to condvar
• Thread 0 = main thread (does serial ops + its share of parallel work)

Usage: ck_threadpool_t *pool = ck_threadpool_create(4); // 4 threads total

// In decode loop: ck_threadpool_dispatch(pool, my_work_fn, args); // my_work_fn called on all threads with (ith, nth, args)

// Between batches: ck_threadpool_pause(pool); // workers sleep (0% CPU) ck_threadpool_resume(pool); // wake workers

ck_threadpool_destroy(pool);

Architecture: STARTUP: Main creates N-1 worker pthreads, all spin on atomic counter DISPATCH: Main writes work desc, bumps counter, all threads execute BARRIER: Atomic counter + spin-wait with _mm_pause() PAUSE: Workers sleep on pthread_cond_t (0% CPU between batches)

Definition in file ck_threadpool.h.

Macro Definition Documentation

CK_CACHE_LINE

#define CK_CACHE_LINE   64

Cache line size for alignment (x86-64)

Definition at line 54 of file ck_threadpool.h.

CK_THREADPOOL_MAX_THREADS

#define CK_THREADPOOL_MAX_THREADS   64

Maximum threads supported (main + workers)

Definition at line 48 of file ck_threadpool.h.

CK_THREADPOOL_SPIN_COUNT

#define CK_THREADPOOL_SPIN_COUNT   1024

Number of spin iterations before falling back to condvar wait

Definition at line 51 of file ck_threadpool.h.

Typedef Documentation

ck_work_fn_t

typedef void(* ck_work_fn_t) (int ith, int nth, void *args)

Work function signature. Called on ALL threads (including main thread 0).

Parameters

ith	Thread index (0 = main thread)
nth	Total number of threads
args	Opaque argument pointer (set via dispatch)

Definition at line 68 of file ck_threadpool.h.

Function Documentation

ck_threadpool_barrier()

void ck_threadpool_barrier

(

ck_threadpool_t *

pool

)

Barrier synchronization within a dispatched work function.

ALL threads must call this at the same point. Threads spin-wait until all have arrived, then proceed.

Must only be called from within a work function (during dispatch).

Parameters

pool	Thread pool

Definition at line 320 of file ck_threadpool.c.

{
    if (!pool || pool->n_threads <= 1) return;
    barrier_wait(&pool->barrier);
}

References barrier_wait().

ck_threadpool_create()

ck_threadpool_t* ck_threadpool_create

(

int

n_threads

)

Create a thread pool with n_threads total threads. Thread 0 is the calling (main) thread; n_threads-1 workers are spawned.

Parameters

n_threads

Total thread count (including main). Must be >= 1. Pass 0 for auto-detect (physical cores).

Returns

Pool handle, or NULL on failure.

Definition at line 183 of file ck_threadpool.c.

{
    if (n_threads <= 0) {
        n_threads = ck_get_physical_cores();
        if (n_threads <= 0) n_threads = 1;
        /* Cap at reasonable default for memory-bound workloads */
        if (n_threads > 8) n_threads = 8;
    }
    if (n_threads > CK_THREADPOOL_MAX_THREADS) {
        n_threads = CK_THREADPOOL_MAX_THREADS;
    }
 
    ck_threadpool_t *pool = aligned_alloc(CK_CACHE_LINE, sizeof(ck_threadpool_t));
    if (!pool) return NULL;
    memset(pool, 0, sizeof(*pool));
 
    pool->n_threads = n_threads;
    atomic_store(&pool->n_dispatch, 0);
    atomic_store(&pool->n_complete, 0);
    atomic_store(&pool->stop, 0);
    atomic_store(&pool->paused, 0);
    pool->work_fn = NULL;
    pool->work_args = NULL;
 
    barrier_init(&pool->barrier, n_threads);
 
    pthread_mutex_init(&pool->mutex, NULL);
    pthread_cond_init(&pool->cond_dispatch, NULL);
    pthread_cond_init(&pool->cond_done, NULL);
 
    /* Thread 0 = main thread (no pthread created) */
    pool->workers[0].id = 0;
    pool->workers[0].pool = pool;
    pool->workers[0].thread = pthread_self();
 
    /* Spawn N-1 worker threads */
    for (int i = 1; i < n_threads; i++) {
        pool->workers[i].id = i;
        pool->workers[i].pool = pool;
 
        int rc = pthread_create(&pool->workers[i].thread, NULL,
                                worker_main, &pool->workers[i]);
        if (rc != 0) {
            fprintf(stderr, "[CK threadpool] Failed to create worker %d: %s\n",
                    i, strerror(rc));
            /* Reduce thread count to what we managed to create */
            pool->n_threads = i;
            barrier_init(&pool->barrier, i);
            break;
        }
    }
 
    if (pool->n_threads > 1) {
        fprintf(stderr, "[CK threadpool] Created %d threads (1 main + %d workers)\n",
                pool->n_threads, pool->n_threads - 1);
    }
 
    return pool;
}

References barrier_init(), CK_CACHE_LINE, ck_get_physical_cores(), CK_THREADPOOL_MAX_THREADS, and worker_main().

Referenced by global_pool_init().

ck_threadpool_destroy()

void ck_threadpool_destroy

(

ck_threadpool_t *

pool

)

Destroy the thread pool. Signals all workers to exit and joins them. Safe to call with NULL.

Definition at line 243 of file ck_threadpool.c.

{
    if (!pool) return;
 
    /* Signal shutdown */
    atomic_store_explicit(&pool->stop, 1, memory_order_release);
 
    /* Wake all sleeping workers */
    pthread_mutex_lock(&pool->mutex);
    pthread_cond_broadcast(&pool->cond_dispatch);
    pthread_mutex_unlock(&pool->mutex);
 
    /* Join all worker threads */
    for (int i = 1; i < pool->n_threads; i++) {
        pthread_join(pool->workers[i].thread, NULL);
    }
 
    pthread_cond_destroy(&pool->cond_dispatch);
    pthread_cond_destroy(&pool->cond_done);
    pthread_mutex_destroy(&pool->mutex);
 
    free(pool);
}

Referenced by ck_threadpool_global_destroy().

ck_threadpool_dispatch()

void ck_threadpool_dispatch	(	ck_threadpool_t *	pool,
		ck_work_fn_t	fn,
		void *	args
	)

Dispatch work to all threads and wait for completion.

1. Sets the work function and args
2. Bumps the dispatch counter (wakes workers)
3. Main thread (ith=0) executes its share
4. Waits for all threads to complete via barrier

This is a blocking call — returns when ALL threads have finished.

Parameters

pool	Thread pool
fn	Work function (called on each thread)
args	Argument passed to fn

Definition at line 271 of file ck_threadpool.c.

{
    if (!pool || !fn) return;
 
    /* Single-thread fast path: just call directly */
    if (pool->n_threads == 1) {
        fn(0, 1, args);
        return;
    }
 
    /* Reset barrier phase for this dispatch */
    barrier_init(&pool->barrier, pool->n_threads);
 
    /* Set work descriptor */
    pool->work_fn = fn;
    pool->work_args = args;
    atomic_store_explicit(&pool->n_complete, 0, memory_order_release);
 
    /* Wake workers by bumping dispatch counter */
    atomic_fetch_add_explicit(&pool->n_dispatch, 1, memory_order_release);
 
    /* Also signal condvar for sleeping workers */
    pthread_mutex_lock(&pool->mutex);
    pthread_cond_broadcast(&pool->cond_dispatch);
    pthread_mutex_unlock(&pool->mutex);
 
    /* Main thread (ith=0) does its share */
    fn(0, pool->n_threads, args);
 
    /* Wait for all workers to complete */
    if (pool->n_threads > 1) {
        int spins = 0;
        while (atomic_load_explicit(&pool->n_complete, memory_order_acquire)
               < pool->n_threads - 1) {
            CK_SPIN_PAUSE();
            spins++;
            if (spins >= CK_THREADPOOL_SPIN_COUNT) {
                pthread_mutex_lock(&pool->mutex);
                if (atomic_load_explicit(&pool->n_complete, memory_order_acquire)
                    < pool->n_threads - 1) {
                    pthread_cond_wait(&pool->cond_done, &pool->mutex);
                }
                pthread_mutex_unlock(&pool->mutex);
                spins = 0;
            }
        }
    }
}

References barrier_init(), CK_SPIN_PAUSE, and CK_THREADPOOL_SPIN_COUNT.

ck_threadpool_global()

ck_threadpool_t* ck_threadpool_global

(

void

)

Get or create the global thread pool. Thread-safe (uses pthread_once internally). Uses ck_get_num_threads() for auto-detection.

Returns

Global pool, never NULL after successful first call.

Definition at line 383 of file ck_threadpool.c.

{
    pthread_once(&g_threadpool_once, global_pool_init);
    return g_threadpool;
}

References g_threadpool, g_threadpool_once, and global_pool_init().

Referenced by ck_get_threadpool(), and ck_threadpool_init().

ck_threadpool_global_destroy()

void ck_threadpool_global_destroy

(

void

)

Destroy the global thread pool. Called during engine shutdown.

Definition at line 389 of file ck_threadpool.c.

{
    if (g_threadpool) {
        ck_threadpool_destroy(g_threadpool);
        g_threadpool = NULL;
        /* Reset once control so pool can be re-created if needed */
        g_threadpool_once = PTHREAD_ONCE_INIT;
    }
}

References ck_threadpool_destroy(), g_threadpool, and g_threadpool_once.

Referenced by ck_threadpool_shutdown().

ck_threadpool_n_threads()

int ck_threadpool_n_threads

(

const ck_threadpool_t *

pool

)

Get total thread count (including main thread)

Definition at line 351 of file ck_threadpool.c.

{
    return pool ? pool->n_threads : 1;
}

ck_threadpool_pause()

void ck_threadpool_pause

(

ck_threadpool_t *

pool

)

Pause workers — they sleep on condvar (0% CPU). Call between batches or during interactive waiting. Workers wake on next dispatch or resume.

Definition at line 330 of file ck_threadpool.c.

{
    if (!pool) return;
    atomic_store_explicit(&pool->paused, 1, memory_order_release);
}

ck_threadpool_resume()

void ck_threadpool_resume

(

ck_threadpool_t *

pool

)

Resume workers — transition from sleep to spin-wait. Call before starting a new batch of work.

Definition at line 336 of file ck_threadpool.c.

{
    if (!pool) return;
    atomic_store_explicit(&pool->paused, 0, memory_order_release);
 
    /* Wake sleeping workers */
    pthread_mutex_lock(&pool->mutex);
    pthread_cond_broadcast(&pool->cond_dispatch);
    pthread_mutex_unlock(&pool->mutex);
}

ck_threadpool_thread_id()

int ck_threadpool_thread_id

(

const ck_threadpool_t *

pool

)

Get thread index for current thread (0 = main, -1 if not in pool)

Definition at line 356 of file ck_threadpool.c.

{
    if (!pool) return -1;
    pthread_t self = pthread_self();
    for (int i = 0; i < pool->n_threads; i++) {
        if (pthread_equal(self, pool->workers[i].thread)) {
            return i;
        }
    }
    return -1;
}