nx_pool_ring_gate.nx source
↩ module page · 90 lines · 3019 B
1// nx_pool_ring_gate.nx -- regression witness for the RING ARENA fix
2// in nx_thread_pool (2026-07-08).
3//
4// The task arena was a bump allocator (16384 slots); past that,
5// nx_pool_submit returned -1 without queuing, and a caller that then
6// delta-waited spun to the 2-billion timeout. A long-lived shared
7// pool (the LLM forward submits thousands of tasks per token) hit
8// this and wedged 428s/token. Fix: slot index wraps modulo
9// arena_cap; SAFE because channel depth << arena_cap so a slot is
10// consumed long before the ring wraps back to it.
11//
12// This gate submits N >> arena_cap tasks to ONE pool (forcing several
13// full ring wraps), each atomically incrementing a shared counter,
14// and verifies EVERY task ran exactly once -- the exact scenario that
15// silently dropped tasks before.
16//
17// Checks:
18// 1 shared counter == N (no task lost to a dropped submit)
19// 2 pool completed-counter == N (delta-wait saw them all)
20// 3 a SECOND batch on the same pool also completes (reuse across
21// the wrap boundary)
22//
23// lineage_id: pool_ring_gate_v1
24
25import "nx_thread_pool.nx"
26import "nx_atom.nx"
27import "nx_fmt.nx"
28
29// > 16384 arena slots, forces ~3 ring wraps.
30const RG_N: i64 = 50000
31const RG_N2: i64 = 20000
32
33func rg_task(ctx: i64) -> i64 {
34 let addr: *i64 = ctx as *i64
35 nx_atom_faa_i64(addr, 1, NX_MO_SEQ_CST)
36 return 0
37}
38
39func rg_nl() -> i64 { fmt_puts("\n" as *u8); return 0 }
40
41func main() -> i64 {
42 let pool: *NxThreadPool = nx_pool_new(8, 64)
43 let ctr: *i64 = sys_mmap(8) as *i64
44 ctr[0] = 0
45
46 var pass: i64 = 0
47
48 // ---- batch 1: N >> arena_cap ----
49 let done0: i64 = nx_pool_n_completed(pool)
50 var i: i64 = 0
51 while i < RG_N {
52 nx_pool_submit(pool, rg_task, ctr as i64)
53 i = i + 1
54 }
55 if nx_pool_wait(pool, done0 + RG_N) != 0 {
56 fmt_puts("RING 1 WAIT-TIMEOUT FAIL"); rg_nl(); return 11
57 }
58 if ctr[0] != RG_N {
59 fmt_puts("RING 1 COUNTER FAIL ctr="); fmt_putn(ctr[0]); fmt_puts(" want="); fmt_putn(RG_N); rg_nl()
60 return 12
61 }
62 fmt_puts("RING 1 COUNTER OK ctr="); fmt_putn(ctr[0]); rg_nl()
63 pass = pass + 1
64
65 if nx_pool_n_completed(pool) != done0 + RG_N {
66 fmt_puts("RING 2 COMPLETED-COUNT FAIL"); rg_nl(); return 13
67 }
68 fmt_puts("RING 2 COMPLETED-COUNT OK"); rg_nl()
69 pass = pass + 1
70
71 // ---- batch 2: reuse the same pool across further wraps ----
72 let done1: i64 = nx_pool_n_completed(pool)
73 var j: i64 = 0
74 while j < RG_N2 {
75 nx_pool_submit(pool, rg_task, ctr as i64)
76 j = j + 1
77 }
78 if nx_pool_wait(pool, done1 + RG_N2) != 0 {
79 fmt_puts("RING 3 WAIT-TIMEOUT FAIL"); rg_nl(); return 14
80 }
81 if ctr[0] != RG_N + RG_N2 {
82 fmt_puts("RING 3 REUSE-COUNTER FAIL ctr="); fmt_putn(ctr[0]); rg_nl(); return 15
83 }
84 fmt_puts("RING 3 REUSE-COUNTER OK ctr="); fmt_putn(ctr[0]); rg_nl()
85 pass = pass + 1
86
87 nx_pool_shutdown(pool)
88 fmt_puts("POOL_RING_GATE "); fmt_putn(pass); fmt_puts("/3 GREEN"); rg_nl()
89 return 0
90}