← Back to C-Kernel-Engine Docs Doxygen Source Documentation
 
Loading...
Searching...
No Matches
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.

Data Structures

struct  ck_threadpool_profile_t
 

Macros

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

Typedefs

typedef void(* ck_range_fn_t) (int begin, int end, void *args)
 
typedef void(* ck_work_fn_t) (int ith, int nth, void *args)
 

Enumerations

enum  ck_gemm_schedule_t { CK_GEMM_SCHEDULE_AUTO = 0 , CK_GEMM_SCHEDULE_STATIC = 1 , CK_GEMM_SCHEDULE_DYNAMIC = 2 }
 

Functions

int ck_gemm_dynamic_schedule_enabled (void)
 
int ck_get_gemm_schedule (void)
 
int ck_set_gemm_schedule (int policy)
 
void ck_threadpool_barrier (ck_threadpool_t *pool)
 
int ck_threadpool_bounded_capacity (int default_threads, int logical_threads)
 
int ck_threadpool_capacity (const ck_threadpool_t *pool)
 
ck_threadpool_t * ck_threadpool_create (int n_threads)
 
ck_threadpool_t * ck_threadpool_create_capacity (int default_threads, int capacity_threads)
 
void ck_threadpool_destroy (ck_threadpool_t *pool)
 
void ck_threadpool_dispatch (ck_threadpool_t *pool, ck_work_fn_t fn, void *args)
 
void ck_threadpool_dispatch_n (ck_threadpool_t *pool, int active_threads, 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_parallel_for_n (ck_threadpool_t *pool, int active_threads, int begin, int end, int grain_size, ck_range_fn_t fn, void *args)
 
void ck_threadpool_pause (ck_threadpool_t *pool)
 
void ck_threadpool_profile_reset (ck_threadpool_t *pool)
 
void ck_threadpool_profile_snapshot (const ck_threadpool_t *pool, ck_threadpool_profile_t *profile)
 
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_range_fn_t

typedef void(* ck_range_fn_t) (int begin, int end, void *args)

Process the half-open interval [begin, end).

Definition at line 86 of file ck_threadpool.h.

◆ 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
ithThread index (0 = main thread)
nthTotal number of threads
argsOpaque argument pointer (set via dispatch)

Definition at line 68 of file ck_threadpool.h.

Enumeration Type Documentation

◆ ck_gemm_schedule_t

Scheduling policy for independent GEMM output tiles.

Enumerator
CK_GEMM_SCHEDULE_AUTO 
CK_GEMM_SCHEDULE_STATIC 
CK_GEMM_SCHEDULE_DYNAMIC 

Definition at line 89 of file ck_threadpool.h.

89 {
ck_gemm_schedule_t
@ CK_GEMM_SCHEDULE_STATIC
@ CK_GEMM_SCHEDULE_AUTO
@ CK_GEMM_SCHEDULE_DYNAMIC

Function Documentation

◆ ck_gemm_dynamic_schedule_enabled()

int ck_gemm_dynamic_schedule_enabled ( void  )

Return non-zero when independent GEMM tiles should use dynamic claiming.

Definition at line 55 of file ck_threadpool.c.

56{
57 const int policy = ck_get_gemm_schedule();
58 return policy == CK_GEMM_SCHEDULE_AUTO || policy == CK_GEMM_SCHEDULE_DYNAMIC;
59}
int ck_get_gemm_schedule(void)

References CK_GEMM_SCHEDULE_AUTO, CK_GEMM_SCHEDULE_DYNAMIC, and ck_get_gemm_schedule().

Referenced by gemm_nt_q4_k_packed_vnni_x8_q8_k_split_min_threaded_4m().

◆ ck_get_gemm_schedule()

int ck_get_gemm_schedule ( void  )

Return the configured process-wide GEMM scheduling policy.

Definition at line 50 of file ck_threadpool.c.

51{
52 return atomic_load_explicit(&g_gemm_schedule, memory_order_acquire);
53}
static atomic_int g_gemm_schedule

References g_gemm_schedule.

Referenced by ck_gemm_dynamic_schedule_enabled().

◆ ck_set_gemm_schedule()

int ck_set_gemm_schedule ( int  policy)

Set the process-wide GEMM tile scheduling policy.

AUTO is the production default and currently selects dynamic work claiming for providers whose jobs write independent output tiles. Providers with ordered/shared reductions do not consult this policy.

Returns
0 on success, -1 for an invalid policy.

Definition at line 41 of file ck_threadpool.c.

42{
43 if (policy < CK_GEMM_SCHEDULE_AUTO || policy > CK_GEMM_SCHEDULE_DYNAMIC) {
44 return -1;
45 }
46 atomic_store_explicit(&g_gemm_schedule, policy, memory_order_release);
47 return 0;
48}

References CK_GEMM_SCHEDULE_DYNAMIC, and g_gemm_schedule.

◆ 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
poolThread pool

Definition at line 520 of file ck_threadpool.c.

521{
522 if (!pool || pool->n_threads <= 1) return;
523 barrier_wait(&pool->barrier);
524}
static void barrier_wait(ck_barrier_t *b)

References barrier_wait().

◆ ck_threadpool_bounded_capacity()

int ck_threadpool_bounded_capacity ( int  default_threads,
int  logical_threads 
)

Compute the bounded capacity for an SMT-safe provider. The default width is preserved and at most half of the additional logical CPUs are reserved.

Definition at line 256 of file ck_threadpool.c.

257{
258 if (default_threads < 1) default_threads = 1;
259 if (default_threads > CK_THREADPOOL_MAX_THREADS) {
260 default_threads = CK_THREADPOOL_MAX_THREADS;
261 }
262 if (logical_threads <= default_threads) return default_threads;
263
264 int capacity = default_threads + (logical_threads - default_threads) / 2;
265 if (capacity > CK_THREADPOOL_MAX_THREADS) {
266 capacity = CK_THREADPOOL_MAX_THREADS;
267 }
268 return capacity;
269}
#define CK_THREADPOOL_MAX_THREADS

References CK_THREADPOOL_MAX_THREADS.

Referenced by global_pool_init().

◆ ck_threadpool_capacity()

int ck_threadpool_capacity ( const ck_threadpool_t *  pool)

Get the maximum worker capacity available to explicit dispatch_n calls.

Definition at line 556 of file ck_threadpool.c.

557{
558 return pool ? pool->n_threads : 1;
559}

Referenced by gemm_nt_q4_k_packed_vnni_x16_q8_k_split_min_threaded_16m(), and gemm_nt_q4_k_packed_vnni_x8_q8_k_split_min_threaded_4m().

◆ 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_threadsTotal thread count (including main). Must be >= 1. Pass 0 for auto-detect (physical cores).
Returns
Pool handle, or NULL on failure.

Definition at line 342 of file ck_threadpool.c.

343{
344 return ck_threadpool_create_capacity(n_threads, n_threads);
345}
ck_threadpool_t * ck_threadpool_create_capacity(int default_threads, int capacity_threads)

References ck_threadpool_create_capacity().

◆ ck_threadpool_create_capacity()

ck_threadpool_t * ck_threadpool_create_capacity ( int  default_threads,
int  capacity_threads 
)

Create a pool whose ordinary dispatch width is smaller than its worker capacity. Exact providers may opt into the additional workers with ck_threadpool_dispatch_n(); ordinary dispatch remains at default_threads.

Definition at line 271 of file ck_threadpool.c.

273{
274 if (default_threads <= 0) {
275 default_threads = ck_get_physical_cores();
276 if (default_threads <= 0) default_threads = 1;
277 /* Cap at reasonable default for memory-bound workloads */
278 if (default_threads > 8) default_threads = 8;
279 }
280 if (capacity_threads < default_threads) {
281 capacity_threads = default_threads;
282 }
283 if (capacity_threads > CK_THREADPOOL_MAX_THREADS) {
284 capacity_threads = CK_THREADPOOL_MAX_THREADS;
285 }
286 if (default_threads > capacity_threads) {
287 default_threads = capacity_threads;
288 }
289
290 ck_threadpool_t *pool = aligned_alloc(CK_CACHE_LINE, sizeof(ck_threadpool_t));
291 if (!pool) return NULL;
292 memset(pool, 0, sizeof(*pool));
293
294 pool->n_threads = capacity_threads;
295 pool->default_threads = default_threads;
296 atomic_store(&pool->n_dispatch, 0);
297 atomic_store(&pool->n_complete, 0);
298 atomic_store(&pool->active_threads, default_threads);
299 atomic_store(&pool->stop, 0);
300 atomic_store(&pool->paused, 0);
301 atomic_store(&pool->profile_enabled, 0);
302 pool->work_fn = NULL;
303 pool->work_args = NULL;
304
305 barrier_init(&pool->barrier, default_threads);
306
307 pthread_mutex_init(&pool->mutex, NULL);
308 pthread_cond_init(&pool->cond_dispatch, NULL);
309 pthread_cond_init(&pool->cond_done, NULL);
310
311 /* Thread 0 = main thread (no pthread created) */
312 pool->workers[0].id = 0;
313 pool->workers[0].pool = pool;
314 pool->workers[0].thread = pthread_self();
315
316 /* Spawn N-1 worker threads */
317 for (int i = 1; i < capacity_threads; i++) {
318 pool->workers[i].id = i;
319 pool->workers[i].pool = pool;
320
321 int rc = pthread_create(&pool->workers[i].thread, NULL,
322 worker_main, &pool->workers[i]);
323 if (rc != 0) {
324 fprintf(stderr, "[CK threadpool] Failed to create worker %d: %s\n",
325 i, strerror(rc));
326 /* Reduce thread count to what we managed to create */
327 pool->n_threads = i;
328 barrier_init(&pool->barrier, i);
329 break;
330 }
331 }
332
333 if (pool->n_threads > 1) {
334 fprintf(stderr,
335 "[CK threadpool] Created %d threads (default=%d, 1 main + %d workers)\n",
336 pool->n_threads, pool->default_threads, pool->n_threads - 1);
337 }
338
339 return pool;
340}
static void barrier_init(ck_barrier_t *b, int n_threads)
int ck_get_physical_cores(void)
static void * worker_main(void *arg)
#define CK_CACHE_LINE

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

Referenced by ck_threadpool_create(), and 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 347 of file ck_threadpool.c.

348{
349 if (!pool) return;
350
351 /* Signal shutdown */
352 atomic_store_explicit(&pool->stop, 1, memory_order_release);
353
354 /* Wake all sleeping workers */
355 pthread_mutex_lock(&pool->mutex);
356 pthread_cond_broadcast(&pool->cond_dispatch);
357 pthread_mutex_unlock(&pool->mutex);
358
359 /* Join all worker threads */
360 for (int i = 1; i < pool->n_threads; i++) {
361 pthread_join(pool->workers[i].thread, NULL);
362 }
363
364 pthread_cond_destroy(&pool->cond_dispatch);
365 pthread_cond_destroy(&pool->cond_done);
366 pthread_mutex_destroy(&pool->mutex);
367
368 free(pool);
369}

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
poolThread pool
fnWork function (called on each thread)
argsArgument passed to fn

Definition at line 465 of file ck_threadpool.c.

466{
467 if (!pool) return;
468 ck_threadpool_dispatch_n(pool, pool->default_threads, fn, args);
469}
void ck_threadpool_dispatch_n(ck_threadpool_t *pool, int active_threads, ck_work_fn_t fn, void *args)

References ck_threadpool_dispatch_n().

◆ ck_threadpool_dispatch_n()

void ck_threadpool_dispatch_n ( ck_threadpool_t *  pool,
int  active_threads,
ck_work_fn_t  fn,
void *  args 
)

Dispatch work to a subset of the pool and wait for completion.

Threads with ith >= active_threads remain idle for this dispatch. The work function sees nth == active_threads.

Parameters
poolThread pool
active_threadsNumber of active threads including main thread
fnWork function
argsArgument passed to fn

Definition at line 375 of file ck_threadpool.c.

376{
377 if (!pool || !fn) return;
378 if (active_threads <= 0) {
379 active_threads = 1;
380 }
381 if (active_threads > pool->n_threads) {
382 active_threads = pool->n_threads;
383 }
384
385 const int profile = atomic_load_explicit(
386 &pool->profile_enabled, memory_order_relaxed);
387 const uint64_t dispatch_start = profile ? monotonic_ns() : 0;
388
389 /* Single-thread fast path: just call directly */
390 if (active_threads == 1 || pool->n_threads == 1) {
391 fn(0, 1, args);
392 if (profile) {
393 const uint64_t dispatch_end = monotonic_ns();
394 atomic_fetch_add_explicit(&pool->profile_dispatch_count, 1, memory_order_relaxed);
395 atomic_fetch_add_explicit(
396 &pool->profile_dispatch_total_ns,
397 dispatch_end - dispatch_start,
398 memory_order_relaxed);
399 atomic_fetch_add_explicit(
400 &pool->profile_main_work_ns,
401 dispatch_end - dispatch_start,
402 memory_order_relaxed);
403 }
404 return;
405 }
406
407 /* Reset barrier phase for this dispatch */
408 barrier_init(&pool->barrier, active_threads);
409
410 /* Set work descriptor */
411 pthread_mutex_lock(&pool->mutex);
412 pool->work_fn = fn;
413 pool->work_args = args;
414 atomic_store_explicit(&pool->active_threads, active_threads, memory_order_release);
415 atomic_store_explicit(&pool->n_complete, 0, memory_order_release);
416
417 /* Wake workers by bumping dispatch counter */
418 atomic_fetch_add_explicit(&pool->n_dispatch, 1, memory_order_release);
419
420 /* Also signal condvar for sleeping workers */
421 pthread_cond_broadcast(&pool->cond_dispatch);
422 pthread_mutex_unlock(&pool->mutex);
423
424 /* Main thread (ith=0) does its share */
425 const uint64_t main_start = profile ? monotonic_ns() : 0;
426 fn(0, active_threads, args);
427 const uint64_t main_end = profile ? monotonic_ns() : 0;
428
429 /* Wait for all workers to complete */
430 if (active_threads > 1) {
431 int spins = 0;
432 while (atomic_load_explicit(&pool->n_complete, memory_order_acquire)
433 < active_threads - 1) {
435 spins++;
436 if (spins >= CK_THREADPOOL_SPIN_COUNT) {
437 pthread_mutex_lock(&pool->mutex);
438 if (atomic_load_explicit(&pool->n_complete, memory_order_acquire)
439 < active_threads - 1) {
440 pthread_cond_wait(&pool->cond_done, &pool->mutex);
441 }
442 pthread_mutex_unlock(&pool->mutex);
443 spins = 0;
444 }
445 }
446 }
447 if (profile) {
448 const uint64_t dispatch_end = monotonic_ns();
449 atomic_fetch_add_explicit(&pool->profile_dispatch_count, 1, memory_order_relaxed);
450 atomic_fetch_add_explicit(
451 &pool->profile_dispatch_total_ns,
452 dispatch_end - dispatch_start,
453 memory_order_relaxed);
454 atomic_fetch_add_explicit(
455 &pool->profile_main_work_ns,
456 main_end - main_start,
457 memory_order_relaxed);
458 atomic_fetch_add_explicit(
459 &pool->profile_completion_wait_ns,
460 dispatch_end - main_end,
461 memory_order_relaxed);
462 }
463}
static uint64_t monotonic_ns(void)
#define CK_SPIN_PAUSE()
#define CK_THREADPOOL_SPIN_COUNT

References barrier_init(), CK_SPIN_PAUSE, CK_THREADPOOL_SPIN_COUNT, and monotonic_ns().

Referenced by adamw_clip_update_multi_f32(), adamw_update_f32(), attention_forward_causal_head_major_gqa_flash_strided_f16kv_workspace(), attention_forward_causal_head_major_gqa_flash_strided_sliding(), attention_forward_causal_head_major_gqa_flash_strided_sliding_gemma4_impl(), attention_forward_causal_head_major_gqa_prefill_append_bf16cache_pytorch_contract_workspace(), attention_forward_causal_head_major_gqa_prefill_append_f16cache_gqa_reuse_config(), attention_forward_decode_head_major_gqa_flash_f16cache_split_partitioned(), attention_forward_head_major_gqa_flash_impl(), audio_conv1d_channel_major_f32(), ck_attention_f16_prefill_qtile64_dispatch(), ck_attention_forward_causal_head_major_gqa_prefill_segmented_f16cache_schedule_workspace(), ck_attention_forward_full_head_major_gqa_tiled_f16kv_fp32_strided(), ck_attention_forward_query_key_head_major_f32_run(), ck_attention_full_bf16_pytorch_flash(), ck_attention_full_bf16_sdpa_tiled(), ck_moe_q4k_mixed_parallel_workspace(), ck_moe_q4k_mixed_route_parallel(), ck_moe_q4k_q5k_route_parallel(), ck_moe_shared_gated_parallel_workspace(), ck_moe_shared_q4k_parallel_workspace(), ck_moe_swiglu_expert_forward_q4k_q5k_bucketed_impl(), ck_threadpool_dispatch(), ck_threadpool_parallel_for_n(), deepseek_mla_attention_f32_parallel_dispatch(), gemm_backward_f32_train_parallel_dispatch(), gemm_backward_f32_train_parallel_dispatch_v2(), gemm_blocked_serial_train_parallel_dispatch(), gemm_f16_input_fp16_threadpool(), gemm_nt_bf16_amx_bf16_storage_workspace(), gemm_nt_bf16_native_bf16_storage(), gemm_nt_q4_k_packed_meta_q8_k_threaded(), gemm_nt_q4_k_packed_meta_q8_k_threaded_nsplit(), gemm_nt_q4_k_packed_meta_x16_gateup_swiglu_fused_vnni(), gemm_nt_q4_k_packed_meta_x16_q8_k_threaded_mreuse(), gemm_nt_q4_k_packed_meta_x16_q8_k_threaded_mtile(), gemm_nt_q4_k_packed_meta_x8_q8_k_split_min_threaded_4m(), gemm_nt_q4_k_packed_meta_x8_q8_k_split_min_threaded_8m(), gemm_nt_q4_k_packed_meta_x8_q8_k_split_min_threaded_mreuse(), gemm_nt_q4_k_packed_meta_x8_q8_k_threaded_mreuse(), gemm_nt_q4_k_packed_meta_x8_q8_k_threaded_mtile(), gemm_nt_q4_k_packed_meta_x8_q8_k_threaded_nsplit(), gemm_nt_q4_k_packed_u8_x16_q8_k_threaded_mtile(), gemm_nt_q4_k_packed_vnni_x16_q8_k_split_min_threaded_16m(), gemm_nt_q4_k_packed_vnni_x8_q8_k_split_min_threaded_4m(), gemm_nt_q4_k_q8_k_gateup_swiglu_fused_vnni(), gemm_q4_k_q8_k(), gradient_accumulate_f32(), gradient_accumulate_multi_f32(), gradient_clip_norm_f32(), gradient_global_norm_multi_f32(), gradient_scale_f32(), moe_swiglu_expert_forward_q4k_q5k_parallel_workspace(), and patch_projection_image_bf16_native_storage().

◆ 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 the default width. In automatic mode the pool may reserve a bounded subset of SMT siblings as explicit provider capacity; ordinary dispatch remains at the default width.

Returns
Global pool, never NULL after successful first call.

Definition at line 651 of file ck_threadpool.c.

652{
653 pthread_once(&g_threadpool_once, global_pool_init);
654 return g_threadpool;
655}
static pthread_once_t g_threadpool_once
static void global_pool_init(void)
static ck_threadpool_t * g_threadpool

References g_threadpool, g_threadpool_once, and global_pool_init().

Referenced by adamw_clip_update_multi_f32(), adamw_update_f32(), attention_forward_causal_head_major_gqa_flash_strided_f16kv_workspace(), attention_forward_causal_head_major_gqa_flash_strided_sliding(), attention_forward_causal_head_major_gqa_flash_strided_sliding_gemma4_impl(), attention_forward_causal_head_major_gqa_prefill_append_bf16cache_pytorch_contract_workspace(), attention_forward_causal_head_major_gqa_prefill_append_f16cache_auto_workspace(), attention_forward_causal_head_major_gqa_prefill_append_f16cache_gqa_reuse_config(), attention_forward_decode_head_major_gqa_flash_f16cache_split_partitioned(), attention_forward_head_major_gqa_flash_impl(), audio_conv1d_channel_major_f32(), audio_conv2d_whc_grouped_f32(), audio_glu_split_channel_major_f32(), audio_relative_shift_f32(), ck_attention_f16_prefill_qtile64_dispatch(), ck_attention_forward_causal_head_major_gqa_prefill_segmented_f16cache_schedule_workspace(), ck_attention_forward_full_head_major_gqa_tiled_f16kv_fp32_strided(), ck_attention_forward_query_key_head_major_f32_run(), ck_attention_full_bf16_pytorch_flash(), ck_attention_full_bf16_sdpa_tiled(), ck_gemma4_prepare_parallel(), ck_get_threadpool(), ck_moe_q4k_mixed_parallel_workspace(), ck_moe_shared_gated_parallel_workspace(), ck_moe_shared_q4k_parallel_workspace(), ck_moe_swiglu_expert_forward_q4k_q5k_bucketed_impl(), ck_threadpool_init(), deepseek_mla_attention_f32_parallel_dispatch(), deepseek_mla_kv_decompress_bf16_parallel_dispatch(), gemm_backward_f32_train_parallel_dispatch(), gemm_backward_f32_train_parallel_dispatch_v2(), gemm_blocked_serial_train_parallel_dispatch(), gemm_f16_input_fp16_threadpool(), gemm_nt_bf16_amx_bf16_storage_workspace(), gemm_nt_bf16_bf16_storage_parallel_dispatch(), gemm_nt_bf16_native_bf16_storage(), gemm_nt_bf16_parallel_dispatch(), gemm_nt_fp32_exact_parallel_dispatch(), gemm_nt_q4_k_packed_meta_q8_k_threaded(), gemm_nt_q4_k_packed_meta_q8_k_threaded_nsplit(), gemm_nt_q4_k_packed_meta_x16_gateup_swiglu_fused_vnni(), gemm_nt_q4_k_packed_meta_x16_q8_k_threaded_mreuse(), gemm_nt_q4_k_packed_meta_x16_q8_k_threaded_mtile(), gemm_nt_q4_k_packed_meta_x8_q8_k_split_min_threaded_4m(), gemm_nt_q4_k_packed_meta_x8_q8_k_split_min_threaded_8m(), gemm_nt_q4_k_packed_meta_x8_q8_k_split_min_threaded_mreuse(), gemm_nt_q4_k_packed_meta_x8_q8_k_threaded_mreuse(), gemm_nt_q4_k_packed_meta_x8_q8_k_threaded_mtile(), gemm_nt_q4_k_packed_meta_x8_q8_k_threaded_nsplit(), gemm_nt_q4_k_packed_u8_x16_q8_k_threaded_mtile(), gemm_nt_q4_k_packed_vnni_x16_q8_k_split_min_threaded_16m(), gemm_nt_q4_k_packed_vnni_x8_q8_k_split_min_threaded_4m(), gemm_nt_q4_k_q8_k_gateup_swiglu_fused_vnni(), gemm_q4_k_q8_k(), gemma4_per_layer_embed_forward(), gemv_bf16_bf16_storage_parallel_dispatch(), gemv_bf16_parallel_dispatch(), gemv_nvfp4_q8_0_uniform(), gradient_accumulate_f32(), gradient_accumulate_multi_f32(), gradient_clip_norm_f32(), gradient_global_norm_multi_f32(), gradient_scale_f32(), moe_swiglu_expert_forward_q4k_q5k_parallel_workspace(), patch_projection_image_bf16_native_storage(), ssm_conv1d_forward_llama_fma(), and ssm_conv1d_forward_llama_production().

◆ ck_threadpool_global_destroy()

void ck_threadpool_global_destroy ( void  )

Destroy the global thread pool. Called during engine shutdown.

Definition at line 657 of file ck_threadpool.c.

658{
659 if (g_threadpool) {
661 g_threadpool = NULL;
662 /* Reset once control so pool can be re-created if needed */
663 g_threadpool_once = PTHREAD_ONCE_INIT;
664 }
665}
void ck_threadpool_destroy(ck_threadpool_t *pool)

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 the ordinary/default dispatch width (including the main thread).

Definition at line 551 of file ck_threadpool.c.

552{
553 return pool ? pool->default_threads : 1;
554}

Referenced by adamw_clip_update_multi_f32(), adamw_update_f32(), attention_forward_causal_head_major_gqa_flash_strided_f16kv_workspace(), attention_forward_causal_head_major_gqa_prefill_append_bf16cache_pytorch_contract_workspace(), attention_forward_causal_head_major_gqa_prefill_append_f16cache_auto_workspace(), attention_forward_causal_head_major_gqa_prefill_append_f16cache_gqa_reuse_config(), attention_forward_decode_head_major_gqa_flash_f16cache_split_partitioned(), audio_conv1d_channel_major_f32(), audio_conv2d_whc_grouped_f32(), audio_glu_split_channel_major_f32(), audio_relative_shift_f32(), ck_attention_f16_prefill_qtile64_dispatch(), ck_attention_forward_causal_head_major_gqa_prefill_segmented_f16cache_schedule_workspace(), ck_attention_forward_full_head_major_gqa_tiled_f16kv_fp32_strided(), ck_attention_forward_query_key_head_major_f32_run(), ck_attention_full_bf16_pytorch_flash(), ck_attention_full_bf16_sdpa_tiled(), ck_attention_pick_active_threads(), ck_gemm_f16_pick_active_threads(), ck_gemma4_prepare_parallel(), ck_moe_q4k_mixed_parallel_workspace(), ck_moe_q4k_mixed_route_parallel(), ck_moe_q4k_q5k_route_parallel(), ck_moe_shared_gated_parallel_workspace(), ck_moe_shared_q4k_parallel_workspace(), ck_moe_swiglu_expert_forward_q4k_q5k_bucketed_impl(), ck_sliding_attention_pick_threads(), deepseek_mla_attention_f32_parallel_dispatch(), deepseek_mla_kv_decompress_bf16_parallel_dispatch(), gemm_backward_f32_train_parallel_dispatch(), gemm_backward_f32_train_parallel_dispatch_v2(), gemm_blocked_serial_train_parallel_dispatch(), gemm_nt_bf16_amx_bf16_storage_workspace(), gemm_nt_bf16_bf16_storage_parallel_dispatch(), gemm_nt_bf16_native_bf16_storage(), gemm_nt_bf16_parallel_dispatch(), gemm_nt_fp32_exact_parallel_dispatch(), gemm_nt_q4_k_packed_meta_q8_k_threaded(), gemm_nt_q4_k_packed_meta_q8_k_threaded_nsplit(), gemm_nt_q4_k_packed_meta_x16_gateup_swiglu_fused_vnni(), gemm_nt_q4_k_packed_meta_x16_q8_k_threaded_mreuse(), gemm_nt_q4_k_packed_meta_x16_q8_k_threaded_mtile(), gemm_nt_q4_k_packed_meta_x8_q8_k_split_min_threaded_4m(), gemm_nt_q4_k_packed_meta_x8_q8_k_split_min_threaded_8m(), gemm_nt_q4_k_packed_meta_x8_q8_k_split_min_threaded_mreuse(), gemm_nt_q4_k_packed_meta_x8_q8_k_threaded_mreuse(), gemm_nt_q4_k_packed_meta_x8_q8_k_threaded_mtile(), gemm_nt_q4_k_packed_meta_x8_q8_k_threaded_nsplit(), gemm_nt_q4_k_packed_u8_x16_q8_k_threaded_mtile(), gemm_nt_q4_k_q8_k_gateup_swiglu_fused_vnni(), gemm_q4_k_q8_k(), gemma4_per_layer_embed_forward(), gemv_bf16_bf16_storage_parallel_dispatch(), gemv_bf16_parallel_dispatch(), gemv_nvfp4_q8_0_uniform(), gradient_accumulate_f32(), gradient_accumulate_multi_f32(), gradient_clip_norm_f32(), gradient_global_norm_multi_f32(), gradient_scale_f32(), moe_swiglu_expert_forward_q4k_q5k_parallel_workspace(), patch_projection_image_bf16_native_storage(), ssm_conv1d_forward_llama_fma(), and ssm_conv1d_forward_llama_production().

◆ ck_threadpool_parallel_for_n()

void ck_threadpool_parallel_for_n ( ck_threadpool_t *  pool,
int  active_threads,
int  begin,
int  end,
int  grain_size,
ck_range_fn_t  fn,
void *  args 
)

Dynamically distribute independent ranges through the persistent pool.

Workers claim grain_size consecutive indices until [begin, end) is empty. This changes ownership only; callers remain responsible for ensuring that ranges write disjoint outputs and preserve each output's reduction order. Do not use this helper for unordered shared reductions.

Definition at line 494 of file ck_threadpool.c.

501{
502 if (!fn || begin >= end) return;
503 if (grain_size <= 0) grain_size = 1;
504 if (!pool || active_threads <= 1) {
505 fn(begin, end, args);
506 return;
507 }
508
509 ck_parallel_for_work_t work = {
510 .end = end,
511 .grain_size = grain_size,
512 .fn = fn,
513 .args = args,
514 };
515 atomic_init(&work.next, begin);
517 pool, active_threads, ck_parallel_for_worker, &work);
518}
static void ck_parallel_for_worker(int ith, int nth, void *opaque)
uint32_t end
Definition utf8.c:215

References ck_parallel_for_worker(), ck_threadpool_dispatch_n(), and end.

Referenced by audio_conv2d_whc_grouped_f32(), audio_glu_split_channel_major_f32(), audio_relative_shift_f32(), ck_gemma4_prepare_parallel(), deepseek_mla_kv_decompress_bf16_parallel_dispatch(), gemm_nt_bf16_bf16_storage_parallel_dispatch(), gemm_nt_bf16_parallel_dispatch(), gemm_nt_fp32_exact_parallel_dispatch(), gemm_nt_q4_k_packed_vnni_x8_q8_k_split_min_threaded_4m(), gemma4_per_layer_embed_forward(), gemv_bf16_bf16_storage_parallel_dispatch(), gemv_bf16_parallel_dispatch(), gemv_nvfp4_q8_0_uniform(), ssm_conv1d_forward_llama_fma(), and ssm_conv1d_forward_llama_production().

◆ 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 530 of file ck_threadpool.c.

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

◆ ck_threadpool_profile_reset()

void ck_threadpool_profile_reset ( ck_threadpool_t *  pool)

Enable profiling and reset cumulative dispatch timing counters.

Definition at line 573 of file ck_threadpool.c.

574{
575 if (!pool) return;
576 atomic_store_explicit(&pool->profile_dispatch_count, 0, memory_order_relaxed);
577 atomic_store_explicit(&pool->profile_dispatch_total_ns, 0, memory_order_relaxed);
578 atomic_store_explicit(&pool->profile_main_work_ns, 0, memory_order_relaxed);
579 atomic_store_explicit(&pool->profile_completion_wait_ns, 0, memory_order_relaxed);
580 atomic_store_explicit(&pool->profile_enabled, 1, memory_order_release);
581}

◆ ck_threadpool_profile_snapshot()

void ck_threadpool_profile_snapshot ( const ck_threadpool_t *  pool,
ck_threadpool_profile_t profile 
)

Snapshot cumulative dispatch timing counters without stopping workers.

Definition at line 583 of file ck_threadpool.c.

585{
586 if (!profile) return;
587 memset(profile, 0, sizeof(*profile));
588 if (!pool) return;
589 profile->dispatch_count = atomic_load_explicit(
590 &pool->profile_dispatch_count, memory_order_relaxed);
591 profile->dispatch_total_ns = atomic_load_explicit(
592 &pool->profile_dispatch_total_ns, memory_order_relaxed);
593 profile->main_work_ns = atomic_load_explicit(
594 &pool->profile_main_work_ns, memory_order_relaxed);
595 profile->completion_wait_ns = atomic_load_explicit(
596 &pool->profile_completion_wait_ns, memory_order_relaxed);
597}

References ck_threadpool_profile_t::completion_wait_ns, ck_threadpool_profile_t::dispatch_count, ck_threadpool_profile_t::dispatch_total_ns, and ck_threadpool_profile_t::main_work_ns.

◆ 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 536 of file ck_threadpool.c.

537{
538 if (!pool) return;
539 atomic_store_explicit(&pool->paused, 0, memory_order_release);
540
541 /* Wake sleeping workers */
542 pthread_mutex_lock(&pool->mutex);
543 pthread_cond_broadcast(&pool->cond_dispatch);
544 pthread_mutex_unlock(&pool->mutex);
545}

◆ 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 561 of file ck_threadpool.c.

562{
563 if (!pool) return -1;
564 pthread_t self = pthread_self();
565 for (int i = 0; i < pool->n_threads; i++) {
566 if (pthread_equal(self, pool->workers[i].thread)) {
567 return i;
568 }
569 }
570 return -1;
571}

Referenced by attention_forward_causal_head_major_gqa_prefill_append_bf16cache_pytorch_contract_workspace(), attention_forward_causal_head_major_gqa_prefill_append_f16cache_gqa_reuse_config(), ck_attention_f16_prefill_qtile64_dispatch(), ck_attention_forward_causal_head_major_gqa_prefill_segmented_f16cache_schedule_workspace(), and ck_sliding_attention_pick_threads().