nx_thread_pool.nx source
↩ module page · 783 lines · 36989 B
1// nx_thread_pool.nx -- shared-queue thread pool (Rayon-precursor).
2//
3// Architecture: hw-sized worker fleet + single Vyukov MPMC queue
4// of task records. Workers spin on the channel; each task carries
5// a typed fn-pointer + ctx; sentinel tasks (is_sentinel == 1)
6// signal a worker to exit.
7//
8// MVP shape -- single shared queue, not yet work-stealing. Work-
9// stealing buys you locality when one worker is starving and another
10// is buried; for compute-bound tasks of comparable cost the shared-
11// queue model is within ~5% of work-stealing (per Rayon's own bench).
12// Composes upward to work-stealing in a follow-up by giving each
13// worker a private deque + a steal protocol; the public API does
14// not change.
15//
16// Tasks live in a bump-allocated arena (NxPoolTask). Submit picks
17// the next slot via atomic FAA on `next_task_slot`, fills it, sends
18// the slot pointer through the channel. Worker receives the pointer,
19// calls task.fn(task.ctx), bumps the completed-counter. Slot reuse
20// is a future concern -- bench first.
21//
22// Composes against: [[vyukov_mpmc_channel]] (queue),
23// [[atomic_intrinsics_real_amo]] (FAA counters),
24// [[thread_clone_native_trampoline]] (worker spawn),
25// [[nx_hw_dynamic_probes]] (worker sizing),
26// [[fn_ptr_indirect_call]] (typed call site).
27
28// nx_safety_envelope:
29// intended_use: AUTO_APPLIED -- primitive-specific tuning queued
30// sil_target: SIL1
31// evidence: [bulk_applied_2026-05-16, see-file-comment-for-detail]
32// verdict: NOT_YET_EVALUATED
33
34import "nx_syscalls.nx"
35import "nx_atom.nx"
36import "nx_thread.nx"
37import "nx_chan.nx"
38import "nx_hw.nx"
39import "nx_lineconf_lib.nx"
40const NX_MAGIC_2147483647: i64 = 2147483647
41const NX_MAGIC_65536: i64 = 65536
42const NX_MAGIC_2000000000: i64 = 2000000000
43
44struct NxPoolTask {
45 fn_ptr: func(i64) -> i64, // typed fn-pointer; called by worker
46 ctx: i64, // argument passed to fn_ptr
47 is_sentinel: i64, // 1 => worker exits on receipt
48}
49
50const NX_POOL_TASK_BYTES: i64 = 24
51
52// Default arena size in task records. 16k records = 384 KiB. Bump
53// when bench shows the slot recycling matters; until then linear
54// growth is fine and predictable.
55const NX_POOL_ARENA_DEFAULT: i64 = 16384
56
57struct NxThreadPool {
58 chan_ptr: i64, // *NxChan -- task slot pointers
59 arena_base: i64, // *NxPoolTask -- pre-allocated arena
60 arena_cap: i64, // capacity in task records
61 next_task_slot: i64, // atomic; bump for fresh slot
62 n_workers: i64,
63 workers_alive: i64, // atomic; decremented when worker exits
64 tasks_submitted: i64, // atomic; bumped per submit
65 tasks_completed: i64, // atomic; bumped per task run
66 task_epoch: i64, // atomic; bumped per submit -- the futex word
67 // idle workers sleep on (low 32 bits)
68 submitter_waiting: i64, // atomic; 1 while nx_pool_wait is FUTEX-blocked,
69 // 0 while it spins or runs. Workers wake the
70 // submitter ONLY when this is 1 -> skips the
71 // futile per-task completion wake during the
72 // submitter's spin phase (2026-07-10).
73}
74
75const NX_POOL_OFF_NEXT_SLOT: i64 = 24
76const NX_POOL_OFF_ALIVE: i64 = 40
77const NX_POOL_OFF_SUBMITTED: i64 = 48
78const NX_POOL_OFF_COMPLETED: i64 = 56
79const NX_POOL_OFF_TASK_EPOCH: i64 = 64
80const NX_POOL_OFF_WAITING: i64 = 72
81
82// ===== spin-then-BLOCK (futex eventcount, 2026-07-10) ==============
83// Pre-fix, idle workers yield-spun FOREVER in nx_chan_recv and the
84// submitter yield-spun in nx_pool_wait: a 14-worker pool burned 14
85// cores while idle (two LLM seats = 28 spinning threads) and every
86// dispatch paid ~1.5ms of scheduler round-trips through the yield
87// storm (MEASURED nx_q8_matmul_micro: a 128-wide matmul took LONGER
88// than a 4864-wide one). Now: bounded spin (catches back-to-back
89// dispatch), then futex-sleep. Idle pool = ZERO CPU. Protocol is
90// the standard eventcount: sleeper loads the epoch, re-checks the
91// condition, then FUTEX_WAITs on the epoch's low 32 bits; the waker
92// makes work visible FIRST, then bumps the epoch and FUTEX_WAKEs.
93// A bump between load and wait makes the wait return immediately
94// (EAGAIN) -- no lost wakeups. rv64 futex=98 -> x86 202 (table row
95// blessed 2026-07-10).
96
97const NX_FUTEX: i64 = 98 // rv64 futex (table -> x86_64 202)
98const NX_FUTEX_WAIT_PRIV: i64 = 128 // FUTEX_WAIT | FUTEX_PRIVATE_FLAG
99const NX_FUTEX_WAKE_PRIV: i64 = 129 // FUTEX_WAKE | FUTEX_PRIVATE_FLAG
100const 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.)
101
102// Sleep until *addr's low 32 bits differ from seen's low 32 bits.
103// NATIVE-THREADED mode: workers spawn via sys_thread_create (->CreateThread on a
104// native PE) and futex is UNAVAILABLE (no WaitOnAddress bridge) -> the idle-wait
105// becomes a BUSY-SPIN (futex_wait/wake become no-ops; the channel is already
106// lock-free-spin). Additive, default 0 = normal futex pool (WSL2/Linux).
107static g_pool_native: i64
108func nx_pool_set_native(v: i64) -> i64 { g_pool_native = v; return 0 }
109func nx_pool_is_native() -> i64 { return g_pool_native }
110
111// NAMED futex shims (patchable): Linux = the futex syscall; on a native PE the
112// emitter OVERWRITES these with jmp -> WaitOnAddress / WakeByAddressAll thunks,
113// so the pool's idle-wait SLEEPS (not busy-spin) on native too.
114func sys_futex_wait(addr: i64, seen: i64) -> i64 {
115 return __syscall(NX_FUTEX, addr, NX_FUTEX_WAIT_PRIV, seen & 0xFFFFFFFF, 0, 0, 0)
116}
117func sys_futex_wake(addr: i64) -> i64 {
118 return __syscall(NX_FUTEX, addr, NX_FUTEX_WAKE_PRIV, NX_MAGIC_2147483647, 0, 0, 0)
119}
120func _pool_futex_wait(addr: *i64, seen: i64) -> i64 {
121 return sys_futex_wait(addr as i64, seen)
122}
123// Wake every sleeper on *addr.
124func _pool_futex_wake_all(addr: *i64) -> i64 {
125 return sys_futex_wake(addr as i64)
126}
127
128// Worker thread main. Spins on the channel; runs each task; bumps
129// the completed counter; exits when is_sentinel == 1.
130func _nx_pool_worker_main(arg: *u8) -> i64 {
131 let pool: *NxThreadPool = arg as *NxThreadPool
132 let c: *NxChan = pool.chan_ptr as *NxChan
133 let done_addr: *i64 = ((arg as i64) + NX_POOL_OFF_COMPLETED) as *i64
134 let alive_addr: *i64 = ((arg as i64) + NX_POOL_OFF_ALIVE) as *i64
135 let epoch_addr: *i64 = ((arg as i64) + NX_POOL_OFF_TASK_EPOCH) as *i64
136 let box: *i64 = sys_mmap(16) as *i64 // try_recv out box, ONCE per worker
137
138 var keep_going: i64 = 1
139 while keep_going == 1 {
140 // spin-then-block receive (see eventcount comment above).
141 var got: i64 = 0
142 while got == 0 {
143 var spin: i64 = 0
144 while spin < NX_POOL_SPIN {
145 got = nx_chan_try_recv(c, box)
146 if got == 1 { spin = NX_POOL_SPIN } else {
147 nx_thread_yield()
148 spin = spin + 1
149 }
150 }
151 if got == 0 {
152 let e: i64 = nx_atom_load_i64(epoch_addr, NX_MO_SEQ_CST)
153 got = nx_chan_try_recv(c, box) // re-check after epoch load
154 if got == 0 { _pool_futex_wait(epoch_addr, e) }
155 }
156 }
157 let task: *NxPoolTask = box[0] as *NxPoolTask
158 if task.is_sentinel == 1 {
159 keep_going = 0
160 } else {
161 let fp: func(i64) -> i64 = task.fn_ptr
162 fp(task.ctx)
163 nx_atom_faa_i64(done_addr, 1, NX_MO_SEQ_CST)
164 // wake the submitter ONLY if it is actually FUTEX-blocked (flag=1);
165 // during its spin phase the wake is a futile syscall. Safe: the
166 // submitter sets waiting=1 BEFORE re-checking completed + waiting,
167 // and futex_wait's compare-value handles any residual race.
168 let wa_addr: *i64 = ((arg as i64) + NX_POOL_OFF_WAITING) as *i64
169 if nx_atom_load_i64(wa_addr, NX_MO_SEQ_CST) == 1 {
170 _pool_futex_wake_all(done_addr)
171 }
172 }
173 }
174 nx_atom_faa_i64(alive_addr, -1, NX_MO_SEQ_CST)
175 return 0
176}
177
178// Create a new pool with n_workers worker threads. If n_workers <= 0,
179// uses nx_hw_worker_count(). queue_cap is the MPMC channel depth;
180// submit blocks when full.
181func nx_pool_new(n_workers: i64, queue_cap: i64) -> *NxThreadPool {
182 if n_workers < 1 { n_workers = nx_hw_worker_count() }
183 if queue_cap < n_workers { queue_cap = n_workers * 4 }
184
185 let raw: *u8 = sys_mmap(128)
186 let pool: *NxThreadPool = raw as *NxThreadPool
187
188 let chan: *NxChan = nx_chan_new(queue_cap)
189 pool.chan_ptr = chan as i64
190
191 let arena_bytes: i64 = NX_POOL_ARENA_DEFAULT * NX_POOL_TASK_BYTES
192 let arena: *u8 = sys_mmap(arena_bytes)
193 pool.arena_base = arena as i64
194 pool.arena_cap = NX_POOL_ARENA_DEFAULT
195 pool.next_task_slot = 0
196 pool.n_workers = n_workers
197 pool.workers_alive = n_workers
198 pool.tasks_submitted = 0
199 pool.tasks_completed = 0
200 pool.task_epoch = 0
201 pool.submitter_waiting = 0
202
203 // Spawn workers.
204 var i: i64 = 0
205 while i < n_workers {
206 var tid: i64 = 0
207 if g_pool_native != 0 {
208 // native PE: spawn via the patchable sys_thread_create shim (->CreateThread)
209 let argp: *i64 = sys_mmap(16) as *i64
210 argp[0] = _nx_pool_worker_main as i64
211 argp[1] = raw as i64
212 tid = sys_thread_create(argp as i64)
213 } else {
214 tid = nx_thread_spawn_fn(_nx_pool_worker_main, raw, NX_MAGIC_65536)
215 }
216 if tid <= 0 {
217 // Couldn't spawn -- workers_alive will under-count. Caller
218 // can probe via nx_pool_n_alive.
219 return pool
220 }
221 i = i + 1
222 }
223 return pool
224}
225
226// Submit a task. Returns 0 (always succeeds now). Blocks (via
227// nx_chan_send) when the channel is full.
228//
229// RING ARENA (2026-07-08): the slot index wraps modulo arena_cap so a
230// long-lived shared pool (e.g. the LLM forward's process pool doing
231// thousands of submits per token) never "exhausts" the arena. The
232// old bump allocator returned -1 past arena_cap, and callers that
233// then delta-waited on tasks that were never queued spun to the
234// 2-billion timeout (found 2026-07-08: nx_llm_forward_profile wedged
235// 428s/token once F32 block matmuls started submitting 168x/token).
236// SAFE because queue_cap (channel depth) << arena_cap: a slot written
237// at submit N is recv'd + fully read by a worker within queue_cap
238// submits, long before submit N+arena_cap could overwrite it. The
239// completed-counter (tasks_completed) is a separate monotonic FAA,
240// unaffected by slot reuse, so delta-waits stay correct.
241// SERIAL fallback: on a target with no working threads (e.g. native Windows PE,
242// where the inline-clone builtin the pool spawns workers with is a no-op), set
243// this to run every submitted task INLINE so nx_pool_wait passes immediately.
244// Default 0 = normal threaded pool (WSL2/Linux); additive, no behavior change.
245static g_pool_serial: i64
246func nx_pool_set_serial(v: i64) -> i64 { g_pool_serial = v; return 0 }
247
248// ===== STRUCTURED CONCURRENCY (LR2): NAMED REFUSALS + THE SANCTIONED SUBMIT ====
249// Declared HERE, ahead of every use, so each intermediate state of this shared
250// file compiles for the ~20 organs that import it.
251//
252// A refusal is a NAMED negative, never a bare -1: a caller that cannot tell
253// "no such scope" from "scope already joined" from "arena full" will guess, and
254// it will guess the most alarming one available.
255const SC_REFUSE_NO_SCOPE: i64 = 0 - 71 // null scope handle
256const SC_REFUSE_CLOSED: i64 = 0 - 72 // scope already joined, or freed, or double-join
257const SC_REFUSE_FULL: i64 = 0 - 73 // scope arena at max-children-per-scope
258const SC_REFUSE_UNSCOPED: i64 = 0 - 74 // strict mode: a bare submit is not a scoped spawn
259const SC_REFUSE_CONF: i64 = 0 - 75 // knowledge/sc_scope.conf row missing -- REFUSE, never default
260const SC_JOIN_GUARD_TRIP: i64 = 0 - 76 // join hit the DERIVED iteration guard (hang detector)
261
262// ===== PART 1 of 2: THE SANCTIONED SUBMIT PATH =================================
263// The pool's public submit is now a THIN WRAPPER over the unchanged raw path.
264// Two things follow, and nothing else changes:
265//
266// 1. sc_scope_spawn routes through _nx_pool_submit_raw, so a scoped spawn is
267// never subject to the strict check -- the check then means EXACTLY "work
268// reached the pool without a scope", with no heuristic and no cross-thread
269// race to get wrong.
270// 2. sc_strict_set(1) makes the BARE submit refuse by name. In that mode the
271// only way work reaches this pool is through a scope that joins it, which is
272// the LAW form of "no orphan tasks" rather than the library form.
273//
274// DEFAULT IS 0, AND THAT IS DELIBERATE, NOT TIMIDITY: 48 call sites of
275// nx_pool_submit exist in this tree (measured 2026-08-25 over 23257 files,
276// coverage_complete=1 corpus_complete=1), 35 of them in production kernels that
277// submit thousands of tasks per token. Refusing them by default would break the
278// estate in one edit, so the law ships ENFORCEABLE AND OFF; turning it on is an
279// adoption campaign carrying its own evidence, not a flag flipped in this file.
280// With strict=0 this wrapper is one static load and one predicted branch, and
281// nx_pool_submit's contract ("returns 0") is bit-for-bit what it always was.
282static g_sc_strict: i64
283func sc_strict_set(v: i64) -> i64 { g_sc_strict = v; return 0 }
284func sc_strict_is() -> i64 { return g_sc_strict }
285
286func nx_pool_submit(pool: *NxThreadPool, fn: func(i64) -> i64, ctx: i64) -> i64 {
287 if g_sc_strict != 0 { return SC_REFUSE_UNSCOPED }
288 return _nx_pool_submit_raw(pool, fn, ctx)
289}
290
291func _nx_pool_submit_raw(pool: *NxThreadPool, fn: func(i64) -> i64, ctx: i64) -> i64 {
292 if g_pool_serial != 0 {
293 fn(ctx)
294 let dc: *i64 = ((pool as i64) + NX_POOL_OFF_COMPLETED) as *i64
295 nx_atom_faa_i64(dc, 1, NX_MO_SEQ_CST)
296 return 0
297 }
298 let slot_idx_addr: *i64 = ((pool as i64) + NX_POOL_OFF_NEXT_SLOT) as *i64
299 let idx: i64 = nx_atom_faa_i64(slot_idx_addr, 1, NX_MO_SEQ_CST)
300 let slot_i: i64 = idx % pool.arena_cap
301 let slot: *NxPoolTask = ((pool.arena_base) + slot_i * NX_POOL_TASK_BYTES) as *NxPoolTask
302 slot.fn_ptr = fn
303 slot.ctx = ctx
304 slot.is_sentinel = 0
305 let c: *NxChan = pool.chan_ptr as *NxChan
306 nx_chan_send(c, slot as i64)
307 let sub_addr: *i64 = ((pool as i64) + NX_POOL_OFF_SUBMITTED) as *i64
308 nx_atom_faa_i64(sub_addr, 1, NX_MO_SEQ_CST)
309 // work is VISIBLE (sent) -> bump the epoch, wake sleepers (eventcount).
310 let ep_addr: *i64 = ((pool as i64) + NX_POOL_OFF_TASK_EPOCH) as *i64
311 nx_atom_faa_i64(ep_addr, 1, NX_MO_SEQ_CST)
312 _pool_futex_wake_all(ep_addr)
313 return 0
314}
315
316// Wait until tasks_completed >= expected: bounded yield-spin (catches
317// fast completions), then futex-sleep on the completed counter itself
318// (workers wake it after every task). Returns 0 on success, -1 on the
319// iteration safety guard (would need ~2B wakeups -- effectively a hang
320// detector, same guard meaning as the old spin).
321func nx_pool_wait(pool: *NxThreadPool, expected: i64) -> i64 {
322 let done_addr: *i64 = ((pool as i64) + NX_POOL_OFF_COMPLETED) as *i64
323 let wa_addr: *i64 = ((pool as i64) + NX_POOL_OFF_WAITING) as *i64
324 var iters: i64 = 0
325 while nx_atom_load_i64(done_addr, NX_MO_SEQ_CST) < expected {
326 if iters < NX_POOL_SPIN {
327 nx_thread_yield()
328 } else {
329 // announce we are about to block, THEN re-check (eventcount): a
330 // worker completing after this store sees waiting=1 and wakes us;
331 // one completing before is caught by the re-checked `seen`.
332 nx_atom_store_i64(wa_addr, 1, NX_MO_SEQ_CST)
333 let seen: i64 = nx_atom_load_i64(done_addr, NX_MO_SEQ_CST)
334 if seen < expected { _pool_futex_wait(done_addr, seen) }
335 nx_atom_store_i64(wa_addr, 0, NX_MO_SEQ_CST)
336 }
337 iters = iters + 1
338 if iters > NX_MAGIC_2000000000 { nx_atom_store_i64(wa_addr, 0, NX_MO_SEQ_CST); return -1 }
339 }
340 nx_atom_store_i64(wa_addr, 0, NX_MO_SEQ_CST)
341 return 0
342}
343
344// Worker exit signal: send n_workers sentinel tasks then spin until
345// workers_alive reaches 0. Idempotent only in the sense that calling
346// shutdown twice would over-send sentinels (with no live workers to
347// consume them); don't.
348func nx_pool_shutdown(pool: *NxThreadPool) -> i64 {
349 let slot_idx_addr: *i64 = ((pool as i64) + NX_POOL_OFF_NEXT_SLOT) as *i64
350 let c: *NxChan = pool.chan_ptr as *NxChan
351 var i: i64 = 0
352 while i < pool.n_workers {
353 let idx: i64 = nx_atom_faa_i64(slot_idx_addr, 1, NX_MO_SEQ_CST)
354 let slot_i: i64 = idx % pool.arena_cap
355 let slot: *NxPoolTask = ((pool.arena_base) + slot_i * NX_POOL_TASK_BYTES) as *NxPoolTask
356 slot.is_sentinel = 1
357 slot.ctx = 0
358 nx_chan_send(c, slot as i64)
359 i = i + 1
360 }
361 // sentinels are visible -> wake every sleeping worker to consume them
362 // (submit's epoch bump does this for tasks; shutdown must do its own).
363 let ep_addr2: *i64 = ((pool as i64) + NX_POOL_OFF_TASK_EPOCH) as *i64
364 nx_atom_faa_i64(ep_addr2, 1, NX_MO_SEQ_CST)
365 _pool_futex_wake_all(ep_addr2)
366 let alive_addr: *i64 = ((pool as i64) + NX_POOL_OFF_ALIVE) as *i64
367 var spins: i64 = 0
368 while nx_atom_load_i64(alive_addr, NX_MO_SEQ_CST) > 0 {
369 nx_thread_yield()
370 spins = spins + 1
371 if spins > NX_MAGIC_2000000000 { return -1 }
372 }
373 return 0
374}
375
376func nx_pool_n_alive(pool: *NxThreadPool) -> i64 {
377 let alive_addr: *i64 = ((pool as i64) + NX_POOL_OFF_ALIVE) as *i64
378 return nx_atom_load_i64(alive_addr, NX_MO_SEQ_CST)
379}
380
381func nx_pool_n_completed(pool: *NxThreadPool) -> i64 {
382 let done_addr: *i64 = ((pool as i64) + NX_POOL_OFF_COMPLETED) as *i64
383 return nx_atom_load_i64(done_addr, NX_MO_SEQ_CST)
384}
385
386// ===== STRUCTURED CONCURRENCY, PART 2 of 2: THE SCOPE =========================
387// LR2: spawn exists only inside a scope that joins its children; a child error
388// propagates to the scope; cancellation is scoped; the orphan-task shape is
389// unrepresentable.
390//
391// SHAPE. A scope is ONE mapping: a 96-byte header followed by a fixed array of
392// child records. Open takes the mapping; spawn fills the next record and hands
393// its pointer to the pool as an ordinary task ctx; join blocks until every record
394// this scope owns has reported; free returns the mapping. There is no detach
395// verb, no spawn verb that does not take a live scope, and no way to end a scope
396// except by joining it -- sc_scope_free REFUSES an unjoined scope, so a caller
397// cannot even release the memory of a scope whose children may still be running.
398//
399// COLORLESS. A child is a plain func(i64) -> i64: exactly what nx_pool_submit
400// has always taken. No async/sync split is introduced and none is needed -- the
401// scope wraps the task RECORD, not the calling convention. A user function is
402// never rewritten, never turned into a state machine, and can be handed to the
403// bare pool and to a scope interchangeably. The async-split trap the colored-
404// function essay names is still not taken here.
405//
406// HOW THE ERROR TRAVELS. Every child runs through _sc_kid_main, a fixed shim
407// that recovers the scope from the record, checks cancellation, calls the user's
408// function, and records a non-zero return into the scope header. FIRST ERROR
409// WINS, and it also CANCELS the scope, so a failing sibling stops work that has
410// not started yet. errgroup semantics, with the cancellation as a scope field
411// rather than a context object threaded through every signature.
412//
413// WHAT CANCELLATION IS, EXACTLY -- the honest bound, not a silence. Setting the
414// flag makes children that have NOT YET ENTERED their body skip it (state
415// SKIPPED) and finish immediately. A child already inside its body is NOT
416// interrupted. That is deliberate: interrupting means delivering a signal, and a
417// returning signal handler EINTRs whatever blocking syscall the child was in --
418// it would MANUFACTURE the truncated-read defect inside every task it was meant
419// to stop. So cancellation is cooperative at task boundaries, and no signal is
420// installed by this code at all.
421//
422// SCOPED means the flag lives in THIS scope's header. A sibling scope has its
423// own header and its own flag, so cancelling one cannot reach the other BY
424// CONSTRUCTION rather than by discipline.
425//
426// RESOURCE ENVELOPE (a leak-by-design must be BOUNDED AND NAMED; unbounded does
427// not ship):
428// per scope, at open: SC_SCOPE_BYTES + kid_cap * SC_KID_BYTES bytes, ONE mmap,
429// ONE VMA. At the shipped kid_cap that is 41056 B, which is
430// above NXA_SMALL_MAX=256, so it is a real mapping and
431// sys_munmap genuinely returns it (below that threshold
432// nx_syscalls routes to a bump arena and munmap is a no-op).
433// per spawn: SC_KID_BYTES = 40 bytes INSIDE that existing mapping, and
434// ZERO additional syscalls beyond what nx_pool_submit
435// already costs. A scope never allocates while running.
436// per join: zero allocation; at most ONE futex wake per completing
437// child, and only while the joiner is actually blocked --
438// the same eventcount guard the pool's submitter uses, so a
439// fast scope pays no wake syscalls at all.
440// freed: sc_scope_free munmaps the whole mapping. sc_live_bytes()
441// reports currently-held scope bytes, so a leak is
442// MEASURABLE rather than asserted, and nx_sc_scope_gate
443// checks it returns to its baseline.
444// unbounded anywhere: none. Spawn number kid_cap+1 is REFUSED BY NAME; it does
445// NOT wrap the way the pool's task arena does, because a kid
446// record must stay readable until join reads its state.
447//
448// DECLARED IMPRECISIONS, so the next reader does not trust this as exact:
449// - the accounting word is lazily mapped on first scope open; two threads each
450// opening their first scope simultaneously can each map one and lose a word of
451// accounting. It is a diagnostic, never a control.
452// - capacity refusal is exact when the scope's owner is the sole spawner, which
453// is the shape the contract describes. Concurrent spawners racing past the
454// cap each roll their reservation back, so the success path stays correct, but
455// a concurrent reader of sc_scope_spawned can see a transient over-count.
456// - sc_scope_is_joined on a scope with zero children is trivially true. It is a
457// non-blocking probe, not evidence: bind it to spawned > 0 at every use.
458
459struct NxScope {
460 pool: i64, // *NxThreadPool this scope spawns into
461 kids: i64, // *NxScopeKid -- always (self + SC_SCOPE_BYTES)
462 kid_cap: i64, // from knowledge/sc_scope.conf, never a literal
463 spawned: i64, // atomic; reservations that reached the pool
464 finished: i64, // atomic; children that reported (ran OR skipped)
465 open: i64, // 1 until join; spawn and free both consult it
466 cancelled: i64, // atomic; SCOPED -- this header only
467 err: i64, // atomic; first non-zero child return
468 err_kid: i64, // atomic; index of the child that set err
469 joiner_waiting: i64, // atomic; 1 only while join is futex-blocked
470 magic: i64, // SC_MAGIC while live; zeroed at join (double-join guard)
471 map_bytes: i64, // length handed to sys_munmap by sc_scope_free
472}
473
474struct NxScopeKid {
475 scope: i64, // *NxScope back-pointer; the shim's only input
476 fn_ptr: func(i64) -> i64, // the user's COLORLESS function, unmodified
477 ctx: i64, // the user's argument, unmodified
478 state: i64, // SC_KID_*
479 rc: i64, // what the user's function returned
480}
481
482const SC_MAGIC: i64 = 1937337155
483const SC_SCOPE_BYTES: i64 = 96 // 12 i64 fields of NxScope
484const SC_KID_BYTES: i64 = 40 // 5 i64 fields of NxScopeKid
485
486// header byte offsets: an atomic needs the ADDRESS, not the field
487const SC_OFF_SPAWNED: i64 = 24
488const SC_OFF_FINISHED: i64 = 32
489const SC_OFF_CANCELLED: i64 = 48
490const SC_OFF_ERR: i64 = 56
491const SC_OFF_ERR_KID: i64 = 64
492const SC_OFF_WAITING: i64 = 72
493
494const SC_KID_PENDING: i64 = 0
495const SC_KID_RUNNING: i64 = 1
496const SC_KID_DONE: i64 = 2
497const SC_KID_SKIPPED: i64 = 3
498
499// ---- THE JOIN BOUNDS ARE DERIVED, NOT CHOSEN, AND DELIBERATELY UNPINNED ------
500// A scope child IS a pool task. It is therefore impossible for a scope join to
501// legitimately outlive the pool wait the same work would otherwise be joined
502// through: a join looser than that bound is a join that can return while the pool
503// still considers the task in flight -- precisely the defect a scope exists to
504// forbid. So both bounds are taken FROM nx_pool_wait's own constants rather than
505// picked, and knowledge/sc_scope.conf carries NO ROW for either, so the linkage
506// cannot be broken silently by editing a conf. nx_sc_scope_gate asserts the
507// linkage mechanically via sc_join_guard_matches_pool(), and callers can read the
508// provenance at runtime instead of trusting this comment.
509const SC_JOIN_SPIN: i64 = NX_POOL_SPIN
510const SC_JOIN_GUARD_ITERS: i64 = NX_MAGIC_2000000000
511func sc_join_spin() -> i64 { return SC_JOIN_SPIN }
512func sc_join_guard_iters() -> i64 { return SC_JOIN_GUARD_ITERS }
513// 1 iff both bounds still equal the pool's own -- the provenance, made checkable.
514func sc_join_guard_matches_pool() -> i64 {
515 var ok: i64 = 0
516 if SC_JOIN_GUARD_ITERS == NX_MAGIC_2000000000 {
517 if SC_JOIN_SPIN == NX_POOL_SPIN { ok = 1 }
518 }
519 return ok
520}
521
522static g_sc_kid_cap: i64 // cached conf value; 0 = not yet read
523static g_sc_last_refusal: i64 // why the last sc_scope_open returned 0
524static g_sc_acct: *i64 // [0] = scope bytes currently held
525
526func sc_conf_path() -> *u8 { return "knowledge/sc_scope.conf" as *u8 }
527func sc_last_refusal() -> i64 { return g_sc_last_refusal }
528
529func sc_acct() -> *i64 {
530 if (g_sc_acct as i64) == 0 {
531 let p: *i64 = sys_mmap(16) as *i64
532 p[0] = 0
533 g_sc_acct = p
534 }
535 return g_sc_acct
536}
537func sc_live_bytes() -> i64 { let a: *i64 = sc_acct(); return a[0] }
538
539// Children per scope comes from conf. A MISSING row REFUSES; it never defaults,
540// because a silent default is a magic number wearing a config's clothes.
541func sc_conf_kidcap() -> i64 {
542 if g_sc_kid_cap > 0 { return g_sc_kid_cap }
543 let v: i64 = lcf_int_of(sc_conf_path(), "max-children-per-scope" as *u8)
544 if v == LCF_MISS { return SC_REFUSE_CONF }
545 if v < 1 { return SC_REFUSE_CONF }
546 g_sc_kid_cap = v
547 return v
548}
549
550// Apply the conf-declared policy. DELIBERATELY EXPLICIT and never called from
551// the submit path: the pool's hot path must never touch a file, so strict mode is
552// applied by a caller that asks for it, exactly like nx_pool_set_serial.
553func sc_conf_apply() -> i64 {
554 let cap: i64 = sc_conf_kidcap()
555 if cap < 1 { return SC_REFUSE_CONF }
556 let st: i64 = lcf_int_of(sc_conf_path(), "strict-unscoped-submit" as *u8)
557 if st == LCF_MISS { return SC_REFUSE_CONF }
558 g_sc_strict = st
559 return 0
560}
561
562func _sc_kid_at(sc: *NxScope, i: i64) -> *NxScopeKid {
563 return ((sc as i64) + SC_SCOPE_BYTES + i * SC_KID_BYTES) as *NxScopeKid
564}
565
566// THE SHIM every scoped child runs through. Fixed function, so the user's own
567// function stays an ordinary colorless func(i64) -> i64.
568func _sc_kid_main(kid_i: i64) -> i64 {
569 let kid: *NxScopeKid = kid_i as *NxScopeKid
570 let scb: i64 = kid.scope
571 let canc: *i64 = (scb + SC_OFF_CANCELLED) as *i64
572 let fin: *i64 = (scb + SC_OFF_FINISHED) as *i64
573 let wa: *i64 = (scb + SC_OFF_WAITING) as *i64
574 if nx_atom_load_i64(canc, NX_MO_SEQ_CST) == 1 {
575 kid.state = SC_KID_SKIPPED
576 } else {
577 kid.state = SC_KID_RUNNING
578 let fp: func(i64) -> i64 = kid.fn_ptr
579 let rc: i64 = fp(kid.ctx)
580 kid.rc = rc
581 kid.state = SC_KID_DONE
582 if rc != 0 {
583 let ea: *i64 = (scb + SC_OFF_ERR) as *i64
584 if nx_atom_load_i64(ea, NX_MO_SEQ_CST) == 0 {
585 // err_kid is published BEFORE err, so any reader that sees a
586 // non-zero err is guaranteed to read a valid index beside it.
587 let idx: i64 = (kid_i - (scb + SC_SCOPE_BYTES)) / SC_KID_BYTES
588 nx_atom_store_i64((scb + SC_OFF_ERR_KID) as *i64, idx, NX_MO_SEQ_CST)
589 nx_atom_store_i64(ea, rc, NX_MO_SEQ_CST)
590 // the error PROPAGATES: it cancels its own scope and nothing else.
591 nx_atom_store_i64(canc, 1, NX_MO_SEQ_CST)
592 }
593 }
594 }
595 nx_atom_faa_i64(fin, 1, NX_MO_SEQ_CST)
596 if nx_atom_load_i64(wa, NX_MO_SEQ_CST) == 1 { _pool_futex_wake_all(fin) }
597 return 0
598}
599
600func sc_scope_open(pool: *NxThreadPool) -> *NxScope {
601 g_sc_last_refusal = 0
602 if (pool as i64) == 0 { g_sc_last_refusal = SC_REFUSE_NO_SCOPE; return 0 as *NxScope }
603 let cap: i64 = sc_conf_kidcap()
604 if cap < 1 { g_sc_last_refusal = SC_REFUSE_CONF; return 0 as *NxScope }
605 let total: i64 = SC_SCOPE_BYTES + cap * SC_KID_BYTES
606 let raw: *u8 = sys_mmap(total)
607 if (raw as i64) <= 0 { g_sc_last_refusal = SC_REFUSE_FULL; return 0 as *NxScope }
608 let sc: *NxScope = raw as *NxScope
609 sc.pool = pool as i64
610 sc.kids = (raw as i64) + SC_SCOPE_BYTES
611 sc.kid_cap = cap
612 sc.spawned = 0
613 sc.finished = 0
614 sc.open = 1
615 sc.cancelled = 0
616 sc.err = 0
617 sc.err_kid = 0 - 1
618 sc.joiner_waiting = 0
619 sc.magic = SC_MAGIC
620 sc.map_bytes = total
621 let a: *i64 = sc_acct()
622 a[0] = a[0] + total
623 return sc
624}
625
626// THE CONTRACTED SYMBOL. The ONLY spawn verb, and it cannot be called without a
627// live scope: a null handle, a joined handle and a full arena are three DIFFERENT
628// named refusals, so no caller has to guess which negative it got.
629func sc_scope_spawn(sc: *NxScope, fn: func(i64) -> i64, ctx: i64) -> i64 {
630 if (sc as i64) == 0 { return SC_REFUSE_NO_SCOPE }
631 if sc.magic != SC_MAGIC { return SC_REFUSE_CLOSED }
632 if sc.open != 1 { return SC_REFUSE_CLOSED }
633 let sa: *i64 = ((sc as i64) + SC_OFF_SPAWNED) as *i64
634 let idx: i64 = nx_atom_faa_i64(sa, 1, NX_MO_SEQ_CST)
635 if idx >= sc.kid_cap {
636 // roll the reservation back so join never waits on a child that was
637 // never submitted -- a scope that over-counts spawned is a scope that
638 // hangs, which is strictly worse than one that refuses.
639 nx_atom_faa_i64(sa, 0 - 1, NX_MO_SEQ_CST)
640 return SC_REFUSE_FULL
641 }
642 let kid: *NxScopeKid = _sc_kid_at(sc, idx)
643 kid.scope = sc as i64
644 kid.fn_ptr = fn
645 kid.ctx = ctx
646 kid.state = SC_KID_PENDING
647 kid.rc = 0
648 // the SANCTIONED path: a scoped spawn is never subject to the strict check,
649 // so strict mode means exactly "work reached the pool without a scope".
650 return _nx_pool_submit_raw(sc.pool as *NxThreadPool, _sc_kid_main, kid as i64)
651}
652
653func sc_scope_cancel(sc: *NxScope) -> i64 {
654 if (sc as i64) == 0 { return SC_REFUSE_NO_SCOPE }
655 if sc.magic != SC_MAGIC { return SC_REFUSE_CLOSED }
656 nx_atom_store_i64(((sc as i64) + SC_OFF_CANCELLED) as *i64, 1, NX_MO_SEQ_CST)
657 return 0
658}
659
660// Non-blocking probe. NOT evidence on its own: on a scope with zero children it
661// is trivially true, so every consumer must bind it to spawned > 0.
662func sc_scope_is_joined(sc: *NxScope) -> i64 {
663 if (sc as i64) == 0 { return 0 }
664 let f: i64 = nx_atom_load_i64(((sc as i64) + SC_OFF_FINISHED) as *i64, NX_MO_SEQ_CST)
665 let s: i64 = nx_atom_load_i64(((sc as i64) + SC_OFF_SPAWNED) as *i64, NX_MO_SEQ_CST)
666 if f >= s { return 1 }
667 return 0
668}
669
670// THE JOIN. Blocks until every child this scope owns has reported. Returns the
671// first child error (0 when none), or a named refusal. Bounds are DERIVED, see
672// above. Protocol is the pool's own eventcount, reused rather than re-invented:
673// announce, re-check, sleep -- so a completion between the check and the sleep
674// cannot be lost. Closes the scope on the way out, which is what makes a second
675// join, a later spawn, and a later cancel all refuse BY NAME.
676func sc_scope_join(sc: *NxScope) -> i64 {
677 if (sc as i64) == 0 { return SC_REFUSE_NO_SCOPE }
678 if sc.magic != SC_MAGIC { return SC_REFUSE_CLOSED }
679 let sa: *i64 = ((sc as i64) + SC_OFF_SPAWNED) as *i64
680 let fa: *i64 = ((sc as i64) + SC_OFF_FINISHED) as *i64
681 let wa: *i64 = ((sc as i64) + SC_OFF_WAITING) as *i64
682 var iters: i64 = 0
683 var trip: i64 = 0
684 while nx_atom_load_i64(fa, NX_MO_SEQ_CST) < nx_atom_load_i64(sa, NX_MO_SEQ_CST) {
685 if iters < SC_JOIN_SPIN {
686 nx_thread_yield()
687 } else {
688 nx_atom_store_i64(wa, 1, NX_MO_SEQ_CST)
689 let seen: i64 = nx_atom_load_i64(fa, NX_MO_SEQ_CST)
690 if seen < nx_atom_load_i64(sa, NX_MO_SEQ_CST) { _pool_futex_wait(fa, seen) }
691 nx_atom_store_i64(wa, 0, NX_MO_SEQ_CST)
692 }
693 iters = iters + 1
694 if iters > SC_JOIN_GUARD_ITERS { trip = 1; break }
695 }
696 nx_atom_store_i64(wa, 0, NX_MO_SEQ_CST)
697 sc.open = 0
698 sc.magic = 0
699 if trip == 1 { return SC_JOIN_GUARD_TRIP }
700 return nx_atom_load_i64(((sc as i64) + SC_OFF_ERR) as *i64, NX_MO_SEQ_CST)
701}
702
703// REFUSES an unjoined scope. This is half the law: a caller cannot release the
704// memory of a scope whose children may still be running, so there is no sequence
705// of scope verbs that produces a task nobody is waiting for.
706func sc_scope_free(sc: *NxScope) -> i64 {
707 if (sc as i64) == 0 { return SC_REFUSE_NO_SCOPE }
708 if sc.open == 1 { return SC_REFUSE_CLOSED }
709 let n: i64 = sc.map_bytes
710 if n <= 0 { return SC_REFUSE_CLOSED }
711 sc.map_bytes = 0
712 let a: *i64 = sc_acct()
713 a[0] = a[0] - n
714 sys_munmap(sc as *u8, n)
715 return 0
716}
717
718// post-mortem readers: valid after join, deliberately WITHOUT the magic check
719func sc_scope_err(sc: *NxScope) -> i64 {
720 if (sc as i64) == 0 { return SC_REFUSE_NO_SCOPE }
721 return nx_atom_load_i64(((sc as i64) + SC_OFF_ERR) as *i64, NX_MO_SEQ_CST)
722}
723func sc_scope_err_child(sc: *NxScope) -> i64 {
724 if (sc as i64) == 0 { return SC_REFUSE_NO_SCOPE }
725 return nx_atom_load_i64(((sc as i64) + SC_OFF_ERR_KID) as *i64, NX_MO_SEQ_CST)
726}
727func sc_scope_spawned(sc: *NxScope) -> i64 {
728 if (sc as i64) == 0 { return SC_REFUSE_NO_SCOPE }
729 return nx_atom_load_i64(((sc as i64) + SC_OFF_SPAWNED) as *i64, NX_MO_SEQ_CST)
730}
731func sc_scope_finished(sc: *NxScope) -> i64 {
732 if (sc as i64) == 0 { return SC_REFUSE_NO_SCOPE }
733 return nx_atom_load_i64(((sc as i64) + SC_OFF_FINISHED) as *i64, NX_MO_SEQ_CST)
734}
735func sc_scope_cancelled(sc: *NxScope) -> i64 {
736 if (sc as i64) == 0 { return SC_REFUSE_NO_SCOPE }
737 return nx_atom_load_i64(((sc as i64) + SC_OFF_CANCELLED) as *i64, NX_MO_SEQ_CST)
738}
739func sc_scope_is_open(sc: *NxScope) -> i64 {
740 if (sc as i64) == 0 { return 0 }
741 return sc.open
742}
743func sc_scope_bytes(sc: *NxScope) -> i64 {
744 if (sc as i64) == 0 { return SC_REFUSE_NO_SCOPE }
745 return SC_SCOPE_BYTES + sc.kid_cap * SC_KID_BYTES
746}
747func sc_scope_kid_state(sc: *NxScope, i: i64) -> i64 {
748 if (sc as i64) == 0 { return SC_REFUSE_NO_SCOPE }
749 if i < 0 { return SC_REFUSE_NO_SCOPE }
750 if i >= sc.kid_cap { return SC_REFUSE_FULL }
751 let kid: *NxScopeKid = _sc_kid_at(sc, i)
752 return kid.state
753}
754func sc_scope_kid_rc(sc: *NxScope, i: i64) -> i64 {
755 if (sc as i64) == 0 { return SC_REFUSE_NO_SCOPE }
756 if i < 0 { return SC_REFUSE_NO_SCOPE }
757 if i >= sc.kid_cap { return SC_REFUSE_FULL }
758 let kid: *NxScopeKid = _sc_kid_at(sc, i)
759 return kid.rc
760}
761
762// ---- self-test ---------------------------------------------------
763
764// Trivial task that returns 0 (used only by the smoke).
765func _nx_pool_self_test_task(ctx: i64) -> i64 {
766 return ctx + 1
767}
768
769func main() -> i64 {
770 let pool: *NxThreadPool = nx_pool_new(2, 16)
771 if pool.n_workers != 2 { return __syscall(93, 1, 0, 0, 0, 0, 0) }
772 if pool.arena_cap != NX_POOL_ARENA_DEFAULT { return __syscall(93, 2, 0, 0, 0, 0, 0) }
773 if pool.tasks_submitted != 0 { return __syscall(93, 3, 0, 0, 0, 0, 0) }
774
775 // Submit one task; wait; check tasks_completed.
776 nx_pool_submit(pool, _nx_pool_self_test_task, 42)
777 if nx_pool_wait(pool, 1) != 0 { return __syscall(93, 4, 0, 0, 0, 0, 0) }
778 if nx_pool_n_completed(pool) != 1 { return __syscall(93, 5, 0, 0, 0, 0, 0) }
779
780 nx_pool_shutdown(pool)
781 if nx_pool_n_alive(pool) != 0 { return __syscall(93, 6, 0, 0, 0, 0, 0) }
782 return 0
783}