nx_chan.nx source
↩ module page · 250 lines · 8265 B
1// nx_chan.nx -- Vyukov MPMC bounded lock-free queue (i64 messages).
2//
3// Multi-Producer Multi-Consumer queue based on Dmitry Vyukov's
4// 2010 design ("Bounded MPMC queue", 1024cores.net). Each cell
5// carries a sequence number; producer sees `seq == enq_pos` to
6// claim a slot, consumer sees `seq == deq_pos + 1` to drain.
7// CAS bumps the position counter. Wait-free under no contention,
8// lock-free under contention.
9//
10// Properties:
11// * MPMC: any number of producers + any number of consumers
12// * Bounded: capacity rounded up to power of 2 (mask-wrap)
13// * Lock-free: no spin locks, no futex waits in fast path
14// * FIFO order globally
15//
16// Compose-against: [[atomic_intrinsics_real_amo]] for CAS+FAA;
17// [[thread_clone_native_trampoline]] for the worker threads;
18// [[nx_hw_dynamic_probes]] for sizing producer/consumer count.
19//
20// Predecessor nx_channel.nx remains for SPSC code (smaller cell
21// footprint, no per-cell seq); use nx_chan.nx whenever multiple
22// threads can produce or consume.
23//
24// Reference: Vyukov D., "Bounded MPMC queue", 2010
25// (independent rederive from algorithm description; no code copied).
26
27// nx_safety_envelope:
28// intended_use: AUTO_APPLIED -- primitive-specific tuning queued
29// sil_target: SIL1
30// evidence: [bulk_applied_2026-05-16, see-file-comment-for-detail]
31// verdict: NOT_YET_EVALUATED
32
33import "nx_syscalls.nx"
34import "nx_atom.nx"
35import "nx_thread.nx"
36
37// One cell of the ring buffer: a sequence counter + the value.
38// Sequence number tracks "expected enqueue position" for the slot;
39// after a successful enqueue it becomes "expected dequeue position".
40struct NxChanCell {
41 seq: i64,
42 value: i64,
43}
44
45const NX_CHAN_CELL_BYTES: i64 = 16
46
47struct NxChan {
48 buf: *u8, // base of cell array
49 mask: i64, // capacity - 1; cap is power of 2
50 enq_pos: i64, // monotonic enqueue counter
51 deq_pos: i64, // monotonic dequeue counter
52}
53
54const NX_CHAN_HEADER_BYTES: i64 = 64
55
56// Compute address of cell[idx]. idx is wrapped via the mask.
57func _nx_chan_cell(c: *NxChan, idx: i64) -> *NxChanCell {
58 let off: i64 = (idx & c.mask) * NX_CHAN_CELL_BYTES
59 return ((c.buf as i64) + off) as *NxChanCell
60}
61
62// Allocate a new MPMC channel. cap_hint is rounded up to the next
63// power of 2; minimum 2. Each cell starts with seq == its index so
64// the very first producer at position 0 sees seq == 0 and proceeds.
65func nx_chan_new(cap_hint: i64) -> *NxChan {
66 var cap: i64 = 1
67 while cap < cap_hint { cap = cap * 2 }
68 if cap < 2 { cap = 2 }
69
70 let total: i64 = NX_CHAN_HEADER_BYTES + cap * NX_CHAN_CELL_BYTES
71 let raw: *u8 = sys_mmap(total)
72 let c: *NxChan = raw as *NxChan
73 c.buf = ((raw as i64) + NX_CHAN_HEADER_BYTES) as *u8
74 c.mask = cap - 1
75 c.enq_pos = 0
76 c.deq_pos = 0
77
78 // Seed seq numbers: cell[i].seq = i.
79 var i: i64 = 0
80 while i < cap {
81 let cell: *NxChanCell = _nx_chan_cell(c, i)
82 cell.seq = i
83 cell.value = 0
84 i = i + 1
85 }
86 return c
87}
88
89// Non-blocking send. Returns 1 on success, 0 when full.
90//
91// Algorithm:
92// pos = atomic_load(enq_pos)
93// loop:
94// cell = &buf[pos & mask]
95// seq = atomic_load(cell.seq)
96// diff = seq - pos
97// if diff == 0: try CAS enq_pos pos -> pos+1
98// on success: cell.value = v; atomic_store(cell.seq, pos+1)
99// elif diff < 0: queue full -- return 0
100// else: another producer raced ahead -- reload pos and retry
101func nx_chan_try_send(c: *NxChan, v: i64) -> i64 {
102 let enq_addr: *i64 = ((c as i64) + 16) as *i64
103 var pos: i64 = __atomic_load_i64(enq_addr, NX_MO_RELAXED)
104 var done: i64 = 0
105 var result: i64 = 0
106 while done == 0 {
107 let cell: *NxChanCell = _nx_chan_cell(c, pos)
108 let seq_addr: *i64 = (cell as i64) as *i64
109 let seq: i64 = __atomic_load_i64(seq_addr, NX_MO_ACQUIRE)
110 let diff: i64 = seq - pos
111 if diff == 0 {
112 let won: i64 = __atomic_cas_i64(enq_addr, pos, pos + 1, NX_MO_RELAXED)
113 if won == 1 {
114 cell.value = v
115 __atomic_store_i64(seq_addr, pos + 1, NX_MO_RELEASE)
116 done = 1
117 result = 1
118 }
119 } else {
120 if diff < 0 {
121 done = 1
122 result = 0
123 } else {
124 pos = __atomic_load_i64(enq_addr, NX_MO_RELAXED)
125 }
126 }
127 }
128 return result
129}
130
131// Non-blocking receive. Returns 1 + writes message to *out on
132// success; returns 0 when empty.
133func nx_chan_try_recv(c: *NxChan, out: *i64) -> i64 {
134 let deq_addr: *i64 = ((c as i64) + 24) as *i64
135 var pos: i64 = __atomic_load_i64(deq_addr, NX_MO_RELAXED)
136 var done: i64 = 0
137 var result: i64 = 0
138 while done == 0 {
139 let cell: *NxChanCell = _nx_chan_cell(c, pos)
140 let seq_addr: *i64 = (cell as i64) as *i64
141 let seq: i64 = __atomic_load_i64(seq_addr, NX_MO_ACQUIRE)
142 let diff: i64 = seq - (pos + 1)
143 if diff == 0 {
144 let won: i64 = __atomic_cas_i64(deq_addr, pos, pos + 1, NX_MO_RELAXED)
145 if won == 1 {
146 *out = cell.value
147 // Free the slot for the next producer one full lap later.
148 __atomic_store_i64(seq_addr, pos + c.mask + 1, NX_MO_RELEASE)
149 done = 1
150 result = 1
151 }
152 } else {
153 if diff < 0 {
154 done = 1
155 result = 0
156 } else {
157 pos = __atomic_load_i64(deq_addr, NX_MO_RELAXED)
158 }
159 }
160 }
161 return result
162}
163
164// Blocking send -- spins on full, yields the CPU between tries so a
165// blocked sender doesn't starve drainers on single-core systems.
166func nx_chan_send(c: *NxChan, v: i64) -> i64 {
167 var sent: i64 = 0
168 while sent == 0 {
169 sent = nx_chan_try_send(c, v)
170 if sent == 0 { nx_thread_yield() }
171 }
172 return 0
173}
174
175// Blocking receive -- spins on empty with cooperative yield.
176func nx_chan_recv(c: *NxChan) -> i64 {
177 let out_raw: *u8 = sys_mmap(16)
178 let out: *i64 = out_raw as *i64
179 var got: i64 = 0
180 while got == 0 {
181 got = nx_chan_try_recv(c, out)
182 if got == 0 { nx_thread_yield() }
183 }
184 return *out
185}
186
187// Approximate length: enq_pos - deq_pos. Best-effort under contention.
188func nx_chan_len(c: *NxChan) -> i64 {
189 let enq_addr: *i64 = ((c as i64) + 16) as *i64
190 let deq_addr: *i64 = ((c as i64) + 24) as *i64
191 let e: i64 = __atomic_load_i64(enq_addr, NX_MO_RELAXED)
192 let d: i64 = __atomic_load_i64(deq_addr, NX_MO_RELAXED)
193 return e - d
194}
195
196// ---- self-test ---------------------------------------------------
197
198func main() -> i64 {
199 let c: *NxChan = nx_chan_new(8)
200 if c.mask != 7 { return __syscall(93, 1, 0, 0, 0, 0, 0) }
201 if nx_chan_len(c) != 0 { return __syscall(93, 2, 0, 0, 0, 0, 0) }
202
203 // Send 5 values; should succeed.
204 var i: i64 = 0
205 while i < 5 {
206 if nx_chan_try_send(c, 100 + i) != 1 {
207 return __syscall(93, 10, 0, 0, 0, 0, 0)
208 }
209 i = i + 1
210 }
211 if nx_chan_len(c) != 5 { return __syscall(93, 11, 0, 0, 0, 0, 0) }
212
213 // Recv first 3 values; check FIFO order.
214 let out_raw: *u8 = sys_mmap(16)
215 let out: *i64 = out_raw as *i64
216 var j: i64 = 0
217 while j < 3 {
218 if nx_chan_try_recv(c, out) != 1 {
219 return __syscall(93, 20, 0, 0, 0, 0, 0)
220 }
221 if *out != 100 + j {
222 return __syscall(93, 21, 0, 0, 0, 0, 0)
223 }
224 j = j + 1
225 }
226 if nx_chan_len(c) != 2 { return __syscall(93, 22, 0, 0, 0, 0, 0) }
227
228 // Drain remainder.
229 while nx_chan_try_recv(c, out) == 1 { }
230 if nx_chan_len(c) != 0 { return __syscall(93, 30, 0, 0, 0, 0, 0) }
231
232 // Fill to capacity then verify next try_send fails (full).
233 var k: i64 = 0
234 while k < 8 {
235 if nx_chan_try_send(c, 200 + k) != 1 {
236 return __syscall(93, 40, 0, 0, 0, 0, 0)
237 }
238 k = k + 1
239 }
240 if nx_chan_try_send(c, 999) != 0 {
241 return __syscall(93, 41, 0, 0, 0, 0, 0)
242 }
243
244 // Drain again.
245 var n: i64 = 0
246 while nx_chan_try_recv(c, out) == 1 { n = n + 1 }
247 if n != 8 { return __syscall(93, 42, 0, 0, 0, 0, 0) }
248
249 return 0
250}