ck_threadpool.c File Reference

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

#include "ck_threadpool.h"
#include <stdlib.h>
#include <string.h>
#include <stdio.h>
#include <errno.h>

Go to the source code of this file.

Macros
#define	CK_SPIN_PAUSE()   ((void)0)

Functions
static void	barrier_init (ck_barrier_t *b, int n_threads)
static void	barrier_wait (ck_barrier_t *b)
int	ck_get_num_threads (void)
int	ck_get_physical_cores (void)
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)
static void	global_pool_init (void)
static void *	worker_main (void *arg)

Variables
static ck_threadpool_t *	g_threadpool = NULL
static pthread_once_t	g_threadpool_once = PTHREAD_ONCE_INIT

Detailed Description

Persistent pthread thread pool for CK-Engine inference.

Architecture:

• N-1 worker pthreads created at startup, main thread is thread 0
• Workers spin on atomic dispatch counter waiting for work
• Barriers use atomic counter + spin-wait with _mm_pause()
• Hybrid polling: spin CK_THREADPOOL_SPIN_COUNT rounds, then condvar
• All atomics on separate cache lines to avoid false sharing

Based on the ggml_threadpool design from llama.cpp, adapted for CK-Engine's kernel dispatch model.

Definition in file ck_threadpool.c.

Macro Definition Documentation

CK_SPIN_PAUSE

#define CK_SPIN_PAUSE

(

)

((void)0)

Definition at line 27 of file ck_threadpool.c.

Function Documentation

barrier_init()

static void barrier_init ( ck_barrier_t *  b, int  n_threads  )

static

Definition at line 78 of file ck_threadpool.c.

{
    atomic_store(&b->n_arrived, 0);
    atomic_store(&b->n_phase, 0);
    b->n_threads = n_threads;
}

Referenced by ck_threadpool_create(), and ck_threadpool_dispatch().

barrier_wait()

static void barrier_wait ( ck_barrier_t *  b )

static

Spin-wait barrier. All threads must call this. Uses phase counter to allow re-use without reset.

Definition at line 89 of file ck_threadpool.c.

{
    const int n = b->n_threads;
    const int phase = atomic_load_explicit(&b->n_phase, memory_order_relaxed);
 
    if (atomic_fetch_add_explicit(&b->n_arrived, 1, memory_order_acq_rel) == n - 1) {
        /* Last thread to arrive — reset and advance phase */
        atomic_store_explicit(&b->n_arrived, 0, memory_order_relaxed);
        atomic_store_explicit(&b->n_phase, phase + 1, memory_order_release);
    } else {
        /* Spin until phase advances */
        int spins = 0;
        while (atomic_load_explicit(&b->n_phase, memory_order_acquire) == phase) {
            CK_SPIN_PAUSE();
            spins++;
            /* After many spins, yield to avoid wasting CPU on oversubscribed systems */
            if (spins > CK_THREADPOOL_SPIN_COUNT * 16) {
                sched_yield();
                spins = 0;
            }
        }
    }
}

References CK_SPIN_PAUSE, and CK_THREADPOOL_SPIN_COUNT.

Referenced by ck_threadpool_barrier().

ck_get_num_threads()

int ck_get_num_threads

(

void

)

Definition at line 178 of file ckernel_strict.c.

{
    // Auto-initialize if not set
    if (!g_threads_initialized) {
        ck_set_num_threads(0);  // Auto-detect
    }
    return g_num_threads;
}

Referenced by global_pool_init().

ck_get_physical_cores()

int ck_get_physical_cores

(

void

)

Definition at line 62 of file ckernel_strict.c.

{
    int physical_cores = 0;
    int logical_cores = (int)sysconf(_SC_NPROCESSORS_ONLN);
    if (logical_cores <= 0) {
        logical_cores = 1;
    }
 
    // Read from /proc/cpuinfo (Linux) and count unique (physical id, core id) pairs.
    FILE *f = fopen("/proc/cpuinfo", "r");
    if (f) {
        char line[256];
        int physical_id = -1;
        int core_id = -1;
 
        struct {
            int physical_id;
            int core_id;
        } seen[8192];
        int seen_count = 0;
 
        const int seen_cap = (int)(sizeof(seen) / sizeof(seen[0]));
 
        // Helper: add (pid,cid) to set if not present.
        #define CK_ADD_PAIR(pid, cid)                                            \
            do {                                                                 \
                if ((pid) >= 0 && (cid) >= 0) {                                  \
                    int exists = 0;                                              \
                    for (int ii = 0; ii < seen_count; ++ii) {                    \
                        if (seen[ii].physical_id == (pid) &&                     \
                            seen[ii].core_id == (cid)) {                         \
                            exists = 1;                                          \
                            break;                                               \
                        }                                                        \
                    }                                                            \
                    if (!exists && seen_count < seen_cap) {                      \
                        seen[seen_count].physical_id = (pid);                    \
                        seen[seen_count].core_id = (cid);                        \
                        ++seen_count;                                            \
                    }                                                            \
                }                                                                \
            } while (0)
 
        while (fgets(line, sizeof(line), f)) {
            int val;
 
            // Blank line separates processor blocks.
            if (line[0] == '\n' || line[0] == '\0') {
                CK_ADD_PAIR(physical_id, core_id);
                physical_id = -1;
                core_id = -1;
                continue;
            }
 
            if (sscanf(line, "physical id : %d", &val) == 1) {
                physical_id = val;
                continue;
            }
            if (sscanf(line, "core id : %d", &val) == 1) {
                core_id = val;
                continue;
            }
        }
        fclose(f);
 
        // Handle file without trailing blank line.
        CK_ADD_PAIR(physical_id, core_id);
 
        #undef CK_ADD_PAIR
 
        physical_cores = seen_count;
    }
 
    // If we couldn't reliably detect physical cores (common in containers),
    // fall back to logical CPUs instead of incorrectly forcing single-thread execution.
    if (physical_cores <= 1 && logical_cores > 1) {
        return logical_cores;
    }
 
    if (physical_cores > 1) {
        return physical_cores;
    }
 
    return logical_cores;
}

Referenced by ck_set_num_threads(), and ck_threadpool_create().

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;
}

global_pool_init()

static void global_pool_init ( void  )

static

Definition at line 377 of file ck_threadpool.c.

{
    int n = ck_get_num_threads();
    g_threadpool = ck_threadpool_create(n);
}

References ck_get_num_threads(), ck_threadpool_create(), and g_threadpool.

Referenced by ck_threadpool_global().

worker_main()

static void* worker_main ( void *  arg )

static

Definition at line 117 of file ck_threadpool.c.

{
    ck_worker_t *w = (ck_worker_t *)arg;
    ck_threadpool_t *pool = w->pool;
    const int ith = w->id;
    int last_dispatch = 0;
 
    for (;;) {
        /* Spin-wait for new dispatch */
        int spins = 0;
        for (;;) {
            /* Check shutdown */
            if (atomic_load_explicit(&pool->stop, memory_order_acquire)) {
                return NULL;
            }
 
            /* Check for new work */
            int current = atomic_load_explicit(&pool->n_dispatch, memory_order_acquire);
            if (current != last_dispatch) {
                last_dispatch = current;
                break;
            }
 
            CK_SPIN_PAUSE();
            spins++;
 
            /* After spinning, fall back to condvar sleep */
            if (spins >= CK_THREADPOOL_SPIN_COUNT) {
                pthread_mutex_lock(&pool->mutex);
                /* Re-check under lock to avoid missed wakeup */
                current = atomic_load_explicit(&pool->n_dispatch, memory_order_acquire);
                if (current == last_dispatch &&
                    !atomic_load_explicit(&pool->stop, memory_order_acquire)) {
                    pthread_cond_wait(&pool->cond_dispatch, &pool->mutex);
                }
                pthread_mutex_unlock(&pool->mutex);
                spins = 0;
            }
        }
 
        /* Execute work */
        ck_work_fn_t fn = pool->work_fn;
        void *args = pool->work_args;
        if (fn) {
            fn(ith, pool->n_threads, args);
        }
 
        /* Signal completion */
        if (atomic_fetch_add_explicit(&pool->n_complete, 1, memory_order_acq_rel)
            == pool->n_threads - 2) {
            /* Last worker done — wake main thread if it's waiting */
            pthread_mutex_lock(&pool->mutex);
            pthread_cond_signal(&pool->cond_done);
            pthread_mutex_unlock(&pool->mutex);
        }
    }
 
    return NULL;
}

References CK_SPIN_PAUSE, and CK_THREADPOOL_SPIN_COUNT.

Referenced by ck_threadpool_create().

Variable Documentation

g_threadpool

ck_threadpool_t* g_threadpool = NULL

static

Definition at line 372 of file ck_threadpool.c.

Referenced by ck_threadpool_global(), ck_threadpool_global_destroy(), and global_pool_init().

g_threadpool_once

pthread_once_t g_threadpool_once = PTHREAD_ONCE_INIT

static

Definition at line 373 of file ck_threadpool.c.

Referenced by ck_threadpool_global(), and ck_threadpool_global_destroy().