code wiki / (root) / nx_thread_pool.nx

nx_thread_pool.nx

buildroot/runtime/nx_thread_pool.nx

36989 B783 linesdepth 6pulls 9 transitivereach 223 importersview sourcekind tooltopic thread

docs dependencies structs consts functions

about

nx_thread_pool.nx -- shared-queue thread pool (Rayon-precursor). Architecture: hw-sized worker fleet + single Vyukov MPMC queue of task records. Workers spin on the channel; each task carries a typed fn-pointer + ctx; sentinel tasks (is_sentinel == 1) signal a worker to exit. MVP shape -- single shared queue, not yet work-stealing. Work- stealing buys you locality when one worker is starving and another is buried; for compute-bound tasks of comparable cost the shared- queue model is within ~5% of work-stealing (per Rayon's own bench). Composes upward to work-stealing in a follow-up by giving each worker a private deque + a steal protocol; the public API does not change. Tasks live in a bump-allocated arena (NxPoolTask). Submit picks the next slot via atomic FAA on `next_task_slot`, fills it, sends the slot pointer through the channel. Worker receives the pointer, calls task.fn(task.ctx), bumps the completed-counter. Slot reuse is a future concern -- bench first. Composes against: [[vyukov_mpmc_channel]] (queue), [[atomic_intrinsics_real_amo]] (FAA counters), [[thread_clone_native_trampoline]] (worker spawn), [[nx_hw_dynamic_probes]] (worker sizing), [[fn_ptr_indirect_call]] (typed call site).

dependencies 6 imports · 46 importers

diagram shows first 10 each side; +0 more imports, +36 more importers in the complete lists below.

imports: nx_syscalls.nxnx_atom.nxnx_thread.nxnx_chan.nxnx_hw.nxnx_lineconf_lib.nx

imported by: nx_arena_smp_gate.nxnx_beir_eval.nxnx_bert_ce_gate.nxnx_bert_ce_lib.nxnx_conv2d.nxnx_conv_speedup.nxnx_dot_simd_demo.nxnx_f32_llama_block_v4.nxnx_f32_llama_v4b.nxnx_f32_llama_v4p.nxnx_f32_matmul_t.nxnx_f32_q4k_matmul.nxnx_intfp_lm_simd_gate.nxnx_natfu_probe.nxnx_nofloat_llm.nxnx_nofloat_q4k.nxnx_nofloat_q4k_gate.nxnx_parallel.nxnx_parallel_simd_reduce.nxnx_parallel_test.nxnx_pipeline.nxnx_pipeline_test.nxnx_pool_ring_gate.nxnx_pool_stale_gate.nxnx_poolheavy.nxnx_ppdemo_parallel.nxnx_pteam.nxnx_q4k_simd2_gate.nxnx_q5_0_threaded_gate.nxnx_q5q8_ab.nxnx_q8_0_simd_gate.nxnx_q8_coldwarm.nxnx_q8_matmul_micro.nxnx_q8_mix_ab.nxnx_q8_scaling.nxnx_q8_st_bw.nxnx_qwen_native_gate.nxnx_qwen_wsl_timing_gate.nxnx_sc_scope_gate.nxnx_sdfrender_mt.nxnx_sum_bench_all.nxnx_ta_parallel_grad_gate.nxnx_task_graph.nxnx_task_graph_test.nxnx_thread_pool_test.nxnx_vit_encoder_layer.nx

call flow from main pre-order; caps 40 nodes / depth 6 declared; ↻ = already shown

structs

44	struct NxPoolTask
57	struct NxThreadPool
459	struct NxScope
474	struct NxScopeKid

consts

40	const NX_MAGIC_2147483647: i64 = 2147483647
41	const NX_MAGIC_65536: i64 = 65536
42	const NX_MAGIC_2000000000: i64 = 2000000000
50	const NX_POOL_TASK_BYTES: i64 = 24
55	const NX_POOL_ARENA_DEFAULT: i64 = 16384
75	const NX_POOL_OFF_NEXT_SLOT: i64 = 24
76	const NX_POOL_OFF_ALIVE: i64 = 40
77	const NX_POOL_OFF_SUBMITTED: i64 = 48
78	const NX_POOL_OFF_COMPLETED: i64 = 56
79	const NX_POOL_OFF_TASK_EPOCH: i64 = 64
80	const NX_POOL_OFF_WAITING: i64 = 72
97	const NX_FUTEX: i64 = 98 // rv64 futex (table -> x86_64 202)
98	const NX_FUTEX_WAIT_PRIV: i64 = 128 // FUTEX_WAIT \| FUTEX_PRIVATE_FLAG
99	const NX_FUTEX_WAKE_PRIV: i64 = 129 // FUTEX_WAKE \| FUTEX_PRIVATE_FLAG
100	const NX_POOL_SPIN: i64 = 64 // yields before sleeping. (TESTED 2026-07-10: busy-spin@50k helped pteam +54% but pool only +3% = noise, at the cost of idle-CPU burn for ALL shared-pool users -> reverted to idle-friendly yield-spin.)
255	const SC_REFUSE_NO_SCOPE: i64 = 0 - 71 // null scope handle
256	const SC_REFUSE_CLOSED: i64 = 0 - 72 // scope already joined, or freed, or double-join
257	const SC_REFUSE_FULL: i64 = 0 - 73 // scope arena at max-children-per-scope
258	const SC_REFUSE_UNSCOPED: i64 = 0 - 74 // strict mode: a bare submit is not a scoped spawn
259	const SC_REFUSE_CONF: i64 = 0 - 75 // knowledge/sc_scope.conf row missing -- REFUSE, never default
260	const SC_JOIN_GUARD_TRIP: i64 = 0 - 76 // join hit the DERIVED iteration guard (hang detector)
482	const SC_MAGIC: i64 = 1937337155
483	const SC_SCOPE_BYTES: i64 = 96 // 12 i64 fields of NxScope
484	const SC_KID_BYTES: i64 = 40 // 5 i64 fields of NxScopeKid
487	const SC_OFF_SPAWNED: i64 = 24
488	const SC_OFF_FINISHED: i64 = 32
489	const SC_OFF_CANCELLED: i64 = 48
490	const SC_OFF_ERR: i64 = 56
491	const SC_OFF_ERR_KID: i64 = 64
492	const SC_OFF_WAITING: i64 = 72
494	const SC_KID_PENDING: i64 = 0
495	const SC_KID_RUNNING: i64 = 1
496	const SC_KID_DONE: i64 = 2
497	const SC_KID_SKIPPED: i64 = 3
509	const SC_JOIN_SPIN: i64 = NX_POOL_SPIN
510	const SC_JOIN_GUARD_ITERS: i64 = NX_MAGIC_2000000000

functions

108	func nx_pool_set_native(v: i64) -> i64 { g_pool_native = v; return 0 } called by 1: main
109	func nx_pool_is_native() -> i64 { return g_pool_native } called by 2: nx_lw_shared_pool _lw_q8_0_matmul
114	func sys_futex_wait(addr: i64, seen: i64) -> i64 called by 3: main _pt_wait _pool_futex_wait
117	func sys_futex_wake(addr: i64) -> i64 called by 3: fu_worker _pt_wake _pool_futex_wake_all
120	func _pool_futex_wait(addr: *i64, seen: i64) -> i64 called by 3: _nx_pool_worker_main nx_pool_wait sc_scope_join calls 1: sys_futex_wait
124	func _pool_futex_wake_all(addr: *i64) -> i64 called by 4: _nx_pool_worker_main _nx_pool_submit_raw nx_pool_shutdown _sc_kid_main calls 1: sys_futex_wake
130	func _nx_pool_worker_main(arg: *u8) -> i64 calls 7: sys_mmap nx_chan_try_recv nx_thread_yield nx_atom_load_i64 _pool_futex_wait nx_atom_faa_i64+1
181	func nx_pool_new(n_workers: i64, queue_cap: i64) -> *NxThreadPool called by 26: main main nx_conv2d_forward_mt main main nx_f32_q4k_matmul_mt+20 calls 5: nx_hw_worker_count sys_mmap nx_chan_new sys_thread_create nx_thread_spawn_fn
246	func nx_pool_set_serial(v: i64) -> i64 { g_pool_serial = v; return 0 } called by 1: main
283	func sc_strict_set(v: i64) -> i64 { g_sc_strict = v; return 0 } called by 1: main
284	func sc_strict_is() -> i64 { return g_sc_strict }
286	func nx_pool_submit(pool: *NxThreadPool, fn: func(i64) -> i64, ctx: i64) -> i64 called by 46: nx_conv2d_forward_pool main _lw_try_fill _lw_cached_matmul _lw_q5_0_matmul _lw_q8_0_matmul_pool_force+40 calls 1: _nx_pool_submit_raw
291	func _nx_pool_submit_raw(pool: *NxThreadPool, fn: func(i64) -> i64, ctx: i64) -> i64 called by 2: nx_pool_submit sc_scope_spawn calls 3: nx_atom_faa_i64 nx_chan_send _pool_futex_wake_all
321	func nx_pool_wait(pool: *NxThreadPool, expected: i64) -> i64 called by 39: nx_conv2d_forward_pool main _lw_try_fill _lw_cached_matmul _lw_q5_0_matmul _lw_q8_0_matmul_pool_force+33 calls 4: nx_atom_load_i64 nx_thread_yield nx_atom_store_i64 _pool_futex_wait
348	func nx_pool_shutdown(pool: *NxThreadPool) -> i64 called by 28: main nx_conv2d_forward_mt main main main nx_f32_q4k_matmul_mt+22 calls 5: nx_atom_faa_i64 nx_chan_send _pool_futex_wake_all nx_atom_load_i64 nx_thread_yield
376	func nx_pool_n_alive(pool: *NxThreadPool) -> i64 called by 3: main main main calls 1: nx_atom_load_i64
381	func nx_pool_n_completed(pool: *NxThreadPool) -> i64 called by 35: nx_conv2d_forward_pool _lw_try_fill _lw_cached_matmul _lw_q5_0_matmul _lw_q8_0_matmul_pool_force nx_blk4_swiglu_pool+29 calls 1: nx_atom_load_i64
511	func sc_join_spin() -> i64 { return SC_JOIN_SPIN } called by 1: main
512	func sc_join_guard_iters() -> i64 { return SC_JOIN_GUARD_ITERS } called by 1: main
514	func sc_join_guard_matches_pool() -> i64 called by 1: main
526	func sc_conf_path() -> u8 { return "knowledge/sc_scope.conf" as u8 } called by 2: sc_conf_kidcap sc_conf_apply
527	func sc_last_refusal() -> i64 { return g_sc_last_refusal } called by 1: main
529	func sc_acct() -> *i64 called by 3: sc_live_bytes sc_scope_open sc_scope_free calls 1: sys_mmap
537	func sc_live_bytes() -> i64 { let a: *i64 = sc_acct(); return a[0] } called by 2: main main calls 1: sc_acct
541	func sc_conf_kidcap() -> i64 called by 2: sc_conf_apply sc_scope_open calls 2: lcf_int_of sc_conf_path
553	func sc_conf_apply() -> i64 calls 3: sc_conf_kidcap lcf_int_of sc_conf_path
562	func _sc_kid_at(sc: NxScope, i: i64) -> NxScopeKid called by 3: sc_scope_spawn sc_scope_kid_state sc_scope_kid_rc
568	func _sc_kid_main(kid_i: i64) -> i64 calls 4: nx_atom_load_i64 nx_atom_store_i64 nx_atom_faa_i64 _pool_futex_wake_all
600	func sc_scope_open(pool: NxThreadPool) -> NxScope called by 2: main main calls 3: sc_conf_kidcap sys_mmap sc_acct
629	func sc_scope_spawn(sc: *NxScope, fn: func(i64) -> i64, ctx: i64) -> i64 called by 2: main main calls 3: nx_atom_faa_i64 _sc_kid_at _nx_pool_submit_raw
653	func sc_scope_cancel(sc: *NxScope) -> i64 called by 1: main calls 1: nx_atom_store_i64
662	func sc_scope_is_joined(sc: *NxScope) -> i64 called by 1: main calls 1: nx_atom_load_i64
676	func sc_scope_join(sc: *NxScope) -> i64 called by 2: main main calls 4: nx_atom_load_i64 nx_thread_yield nx_atom_store_i64 _pool_futex_wait
706	func sc_scope_free(sc: *NxScope) -> i64 called by 2: main main calls 2: sc_acct sys_munmap
719	func sc_scope_err(sc: *NxScope) -> i64 called by 1: main calls 1: nx_atom_load_i64
723	func sc_scope_err_child(sc: *NxScope) -> i64 called by 1: main calls 1: nx_atom_load_i64
727	func sc_scope_spawned(sc: *NxScope) -> i64 called by 1: main calls 1: nx_atom_load_i64
731	func sc_scope_finished(sc: *NxScope) -> i64 called by 1: main calls 1: nx_atom_load_i64
735	func sc_scope_cancelled(sc: *NxScope) -> i64 called by 1: main calls 1: nx_atom_load_i64
739	func sc_scope_is_open(sc: *NxScope) -> i64
743	func sc_scope_bytes(sc: *NxScope) -> i64 called by 1: main
747	func sc_scope_kid_state(sc: *NxScope, i: i64) -> i64 called by 1: main calls 1: _sc_kid_at
754	func sc_scope_kid_rc(sc: *NxScope, i: i64) -> i64 calls 1: _sc_kid_at
765	func _nx_pool_self_test_task(ctx: i64) -> i64
769	func main() -> i64 calls 6: nx_pool_new nx_pool_submit nx_pool_wait nx_pool_n_completed nx_pool_shutdown nx_pool_n_alive