nx_chan_mpmc_test.nx source
↩ module page · 151 lines · 5661 B
1// nx_chan_mpmc_test.nx -- prove Vyukov MPMC channel under real
2// multi-producer + multi-consumer contention.
3//
4// Test design:
5// - hw_workers producer threads, each sends ITERATIONS_PER_PRODUCER
6// messages (just their thread index repeated -- value content
7// irrelevant; we count receipts).
8// - hw_workers consumer threads, each drains until total received
9// atomic counter hits expected total.
10// - Atomic done-counter signals when all producers have finished
11// pushing, so consumers know when to stop spinning on empty.
12//
13// PASS criteria:
14// - total_received == producers * ITERATIONS_PER_PRODUCER (no lost
15// messages, no duplicates)
16// - chan_len returns 0 at end (queue drained)
17//
18// If Vyukov MPMC is broken (CAS race exposes wrong cell, ABA-style
19// re-use), we'll see lost or duplicated messages -> total mismatch.
20
21import "nx_kernel_v2.nx"
22import "nx_log.nx"
23import "nx_atom.nx"
24import "nx_thread.nx"
25import "nx_chan.nx"
26import "nx_hw.nx"
27
28const ITERATIONS_PER_PRODUCER: i64 = 2000
29const STRESS_FACTOR: i64 = 4
30
31// Shared state (all in one mmap'd region for cross-thread visibility).
32struct McShared {
33 chan_ptr: i64, // *NxChan
34 producers_done: i64, // atomic; bumped by each producer when finished
35 n_producers: i64, // total producers (constant after init)
36 total_received: i64, // atomic; bumped by each consumer per recv
37 sum_received: i64, // atomic; sum-of-values for sanity check
38}
39
40func producer_main(arg: *u8) -> i64 {
41 let s: *McShared = arg as *McShared
42 let c: *NxChan = s.chan_ptr as *NxChan
43 var i: i64 = 0
44 while i < ITERATIONS_PER_PRODUCER {
45 // Send a constant value per producer = 1 so sum == count.
46 nx_chan_send(c, 1)
47 i = i + 1
48 }
49 let done_addr: *i64 = ((arg as i64) + 8) as *i64
50 nx_atom_faa_i64(done_addr, 1, NX_MO_SEQ_CST)
51 return 0
52}
53
54func consumer_main(arg: *u8) -> i64 {
55 let s: *McShared = arg as *McShared
56 let c: *NxChan = s.chan_ptr as *NxChan
57 let done_addr: *i64 = ((arg as i64) + 8) as *i64
58 let n_prod_addr: *i64 = ((arg as i64) + 16) as *i64
59 let recv_addr: *i64 = ((arg as i64) + 24) as *i64
60 let sum_addr: *i64 = ((arg as i64) + 32) as *i64
61
62 let out_raw: *u8 = sys_mmap(16)
63 let out: *i64 = out_raw as *i64
64
65 var keep_going: i64 = 1
66 while keep_going == 1 {
67 let got: i64 = nx_chan_try_recv(c, out)
68 if got == 1 {
69 nx_atom_faa_i64(recv_addr, 1, NX_MO_SEQ_CST)
70 nx_atom_faa_i64(sum_addr, *out, NX_MO_SEQ_CST)
71 } else {
72 // Empty. If all producers are done AND queue is empty,
73 // we're free to exit. Re-check after each emptiness so
74 // we don't exit before the last producer pushed.
75 let prod_done: i64 = nx_atom_load_i64(done_addr, NX_MO_SEQ_CST)
76 let n_prod: i64 = nx_atom_load_i64(n_prod_addr, NX_MO_RELAXED)
77 if prod_done == n_prod {
78 if nx_chan_len(c) <= 0 { keep_going = 0 }
79 }
80 }
81 }
82 return 0
83}
84
85func main() -> nx_exit {
86 let n_workers: i64 = nx_hw_worker_count() * STRESS_FACTOR
87 println("=== nx_chan MPMC contention smoke ===" as *u8)
88 println("Producers + consumers (cpu_count * STRESS_FACTOR each):" as *u8)
89 // NOTE: no ""-literal here -- the empty-string literal aliases
90 // the string pool (known nx_cc trap) and was phantom-printing the
91 // NEXT pooled literal ("FAIL: producer spawn") into the banner.
92 print_i64(n_workers); println(" workers each" as *u8)
93
94 let raw: *u8 = sys_mmap(128)
95 let s: *McShared = raw as *McShared
96 let c: *NxChan = nx_chan_new(64)
97 s.chan_ptr = c as i64
98 s.producers_done = 0
99 s.n_producers = n_workers
100 s.total_received = 0
101 s.sum_received = 0
102
103 // Spawn producers.
104 var p: i64 = 0
105 while p < n_workers {
106 let tid: i64 = nx_thread_spawn_fn(producer_main, raw, 65536)
107 if tid <= 0 { println("FAIL: producer spawn" as *u8); return 1 }
108 p = p + 1
109 }
110 // Spawn consumers.
111 var q: i64 = 0
112 while q < n_workers {
113 let tid_c: i64 = nx_thread_spawn_fn(consumer_main, raw, 65536)
114 if tid_c <= 0 { println("FAIL: consumer spawn" as *u8); return 2 }
115 q = q + 1
116 }
117
118 let expected: i64 = ITERATIONS_PER_PRODUCER * n_workers
119 let recv_addr: *i64 = ((raw as i64) + 24) as *i64
120 let sum_addr: *i64 = ((raw as i64) + 32) as *i64
121
122 // Spin-wait until total_received hits expected. If MPMC is
123 // broken, this will never happen (timeout fires).
124 var spins: i64 = 0
125 while nx_atom_load_i64(recv_addr, NX_MO_SEQ_CST) < expected {
126 spins = spins + 1
127 if spins > 2000000000 {
128 println("FAIL: timeout, recv != expected" as *u8)
129 print_i64(nx_atom_load_i64(recv_addr, NX_MO_SEQ_CST)); println("" as *u8)
130 return 3
131 }
132 }
133
134 let final_recv: i64 = nx_atom_load_i64(recv_addr, NX_MO_SEQ_CST)
135 let final_sum: i64 = nx_atom_load_i64(sum_addr, NX_MO_SEQ_CST)
136 if final_recv != expected {
137 println("FAIL: recv overshot expected" as *u8); return 4
138 }
139 if final_sum != expected {
140 println("FAIL: sum != count (lost or duplicated values)" as *u8); return 5
141 }
142 if nx_chan_len(c) > 0 {
143 println("FAIL: queue not drained" as *u8); return 6
144 }
145
146 println("PASS: total_received == producers * iterations" as *u8)
147 println("PASS: sum_received == count (no lost / duplicated values)" as *u8)
148 println("PASS: queue drained to empty" as *u8)
149 println("Vyukov MPMC channel proven under hw-sized concurrent contention." as *u8)
150 return 0
151}