nx_chan.nx source

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 // ⚠Bound-and-discarded for the same reason as nx_atom_store_i64: nx_cc refuses a bare 116 // intrinsic statement, and an atomic store has no result worth using. `done` is set to 1 117 // unconditionally on the next line, so consuming the value here cannot change behaviour. 118 let published: i64 = __atomic_store_i64(seq_addr, pos + 1, NX_MO_RELEASE) 119 if published != 0 { done = 1 } 120 done = 1 121 result = 1 122 } 123 } else { 124 if diff < 0 { 125 done = 1 126 result = 0 127 } else { 128 pos = __atomic_load_i64(enq_addr, NX_MO_RELAXED) 129 } 130 } 131 } 132 return result 133} 134 135// Non-blocking receive. Returns 1 + writes message to *out on 136// success; returns 0 when empty. 137func nx_chan_try_recv(c: *NxChan, out: *i64) -> i64 { 138 let deq_addr: *i64 = ((c as i64) + 24) as *i64 139 var pos: i64 = __atomic_load_i64(deq_addr, NX_MO_RELAXED) 140 var done: i64 = 0 141 var result: i64 = 0 142 while done == 0 { 143 let cell: *NxChanCell = _nx_chan_cell(c, pos) 144 let seq_addr: *i64 = (cell as i64) as *i64 145 let seq: i64 = __atomic_load_i64(seq_addr, NX_MO_ACQUIRE) 146 let diff: i64 = seq - (pos + 1) 147 if diff == 0 { 148 let won: i64 = __atomic_cas_i64(deq_addr, pos, pos + 1, NX_MO_RELAXED) 149 if won == 1 { 150 *out = cell.value 151 // Free the slot for the next producer one full lap later. 152 // Bound-and-discarded exactly as in nx_chan_try_send above; `done` is set unconditionally 153 // on the next line, so consuming the store's result cannot change behaviour. 154 let freed: i64 = __atomic_store_i64(seq_addr, pos + c.mask + 1, NX_MO_RELEASE) 155 if freed != 0 { done = 1 } 156 done = 1 157 result = 1 158 } 159 } else { 160 if diff < 0 { 161 done = 1 162 result = 0 163 } else { 164 pos = __atomic_load_i64(deq_addr, NX_MO_RELAXED) 165 } 166 } 167 } 168 return result 169} 170 171// Blocking send -- spins on full, yields the CPU between tries so a 172// blocked sender doesn't starve drainers on single-core systems. 173func nx_chan_send(c: *NxChan, v: i64) -> i64 { 174 var sent: i64 = 0 175 while sent == 0 { 176 sent = nx_chan_try_send(c, v) 177 if sent == 0 { nx_thread_yield() } 178 } 179 return 0 180} 181 182// Blocking receive -- spins on empty with cooperative yield. 183func nx_chan_recv(c: *NxChan) -> i64 { 184 let out_raw: *u8 = sys_mmap(16) 185 let out: *i64 = out_raw as *i64 186 var got: i64 = 0 187 while got == 0 { 188 got = nx_chan_try_recv(c, out) 189 if got == 0 { nx_thread_yield() } 190 } 191 return *out 192} 193 194// Approximate length: enq_pos - deq_pos. Best-effort under contention. 195func nx_chan_len(c: *NxChan) -> i64 { 196 let enq_addr: *i64 = ((c as i64) + 16) as *i64 197 let deq_addr: *i64 = ((c as i64) + 24) as *i64 198 let e: i64 = __atomic_load_i64(enq_addr, NX_MO_RELAXED) 199 let d: i64 = __atomic_load_i64(deq_addr, NX_MO_RELAXED) 200 return e - d 201} 202 203// ---- self-test --------------------------------------------------- 204 205func main() -> i64 { 206 let c: *NxChan = nx_chan_new(8) 207 if c.mask != 7 { return __syscall(93, 1, 0, 0, 0, 0, 0) } 208 if nx_chan_len(c) != 0 { return __syscall(93, 2, 0, 0, 0, 0, 0) } 209 210 // Send 5 values; should succeed. 211 var i: i64 = 0 212 while i < 5 { 213 if nx_chan_try_send(c, 100 + i) != 1 { 214 return __syscall(93, 10, 0, 0, 0, 0, 0) 215 } 216 i = i + 1 217 } 218 if nx_chan_len(c) != 5 { return __syscall(93, 11, 0, 0, 0, 0, 0) } 219 220 // Recv first 3 values; check FIFO order. 221 let out_raw: *u8 = sys_mmap(16) 222 let out: *i64 = out_raw as *i64 223 var j: i64 = 0 224 while j < 3 { 225 if nx_chan_try_recv(c, out) != 1 { 226 return __syscall(93, 20, 0, 0, 0, 0, 0) 227 } 228 if *out != 100 + j { 229 return __syscall(93, 21, 0, 0, 0, 0, 0) 230 } 231 j = j + 1 232 } 233 if nx_chan_len(c) != 2 { return __syscall(93, 22, 0, 0, 0, 0, 0) } 234 235 // Drain remainder. 236 while nx_chan_try_recv(c, out) == 1 { } 237 if nx_chan_len(c) != 0 { return __syscall(93, 30, 0, 0, 0, 0, 0) } 238 239 // Fill to capacity then verify next try_send fails (full). 240 var k: i64 = 0 241 while k < 8 { 242 if nx_chan_try_send(c, 200 + k) != 1 { 243 return __syscall(93, 40, 0, 0, 0, 0, 0) 244 } 245 k = k + 1 246 } 247 if nx_chan_try_send(c, 999) != 0 { 248 return __syscall(93, 41, 0, 0, 0, 0, 0) 249 } 250 251 // Drain again. 252 var n: i64 = 0 253 while nx_chan_try_recv(c, out) == 1 { n = n + 1 } 254 if n != 8 { return __syscall(93, 42, 0, 0, 0, 0, 0) } 255 256 return 0 257}