code wiki / _hdl_build / nx_sig2_bp_gate.nx
nx_sig2_bp_gate.nx source
↩ module page · 266 lines · 12481 B
1// nx_sig2_bp_gate.nx -- ADVERSARIAL gate for nx_sig2_txq (the relay backpressure fix).
2// Reproduces the EXACT field failure shape the old path had: a receiver that stops reading while the
3// sender floods video-sized frames. OLD behavior: blocking write -> the whole relay stalls (head-of-line),
4// or a partial write tears a frame and poisons the stream. REQUIRED behavior, proven here on REAL sockets
5// with a tiny SO_SNDBUF:
6// T1 NO-STALL : 300 x 8KB sends to a wedged receiver return in bounded wall time; some accepted,
7// the rest dropped WHOLE (never blocks, never tears).
8// T2 INTEGRITY : when the receiver resumes, EVERY accepted frame arrives byte-perfect, in order,
9// exactly once -- the stream parses cleanly end-to-end (no torn frames possible).
10// T3 FAIL-STOP : sends after the receiver dies report fatal (conn reap), no crash (SIGPIPE ignored).
11// T4 FAST PATH : with a live reader, frames go direct (queue stays empty) and all arrive intact.
12// license_tier: ORIGINAL
13import "nx_syscalls.nx"
14import "nx_connect.nx" // bounded connect: a raw sys_connect hangs ~127s on a black-holed host
15import "nx_signal.nx"
16import "nx_websocket_frame.nx"
17import "nx_websocket_stream.nx"
18import "nx_poll.nx"
19import "nx_sig2_txq.nx"
20import "nx_gate_verdict.nx"
21
22func gw(s: *u8) -> i64 { var n: i64 = 0; while s[n] != (0 as u8) { n = n + 1 } sys_write(1, s, n); return 0 }
23func gn(v: i64) -> i64 {
24 let b: *u8 = sys_mmap(28); var m: i64 = v; if m < 0 { sys_write(1, "-" as *u8, 1); m = 0 - m }
25 let t: *u8 = sys_mmap(28); var k: i64 = 0; if m == 0 { t[0] = 48 as u8; k = 1 }
26 while m > 0 { t[k] = (48 + (m % 10)) as u8; m = m / 10; k = k + 1 }
27 var i: i64 = 0; while i < k { b[i] = t[k - 1 - i]; i = i + 1 } sys_write(1, b, k); return 0 }
28
29func bp_addr(a: *u8, port: i64) -> i64 {
30 a[0] = 2 as u8; a[1] = 0 as u8
31 a[2] = ((port >> 8) & 0xff) as u8
32 a[3] = (port & 0xff) as u8
33 a[4] = 127 as u8; a[5] = 0 as u8; a[6] = 0 as u8; a[7] = 1 as u8
34 var z: i64 = 8
35 while z < 16 { a[z] = 0 as u8; z = z + 1 }
36 return 0 }
37
38// make one loopback TCP pair through the listener; tiny buffers so backpressure hits fast.
39// out[0]=server-side fd (the relay's conn), out[1]=client fd (the receiver). small=1 -> tiny bufs.
40func bp_pair(lfd: i64, port: i64, small: i64, out: *i64) -> i64 {
41 let cf: i64 = sys_socket(AF_INET, SOCK_STREAM, 0)
42 if cf < 0 { return 0 - 1 }
43 let bs: *u8 = sys_mmap(8)
44 if small == 1 {
45 bs[0] = 0 as u8; bs[1] = 16 as u8; bs[2] = 0 as u8; bs[3] = 0 as u8 // 4096 LE
46 sys_setsockopt(cf, 1, 8, bs, 4) // SO_RCVBUF before connect
47 }
48 let ad: *u8 = sys_mmap(16)
49 bp_addr(ad, port)
50 if nx_connect_bounded(cf, ad, 16, NX_CONN_DEFAULT_MS) < 0 { sys_close(cf); return 0 - 1 }
51 let sf: i64 = sys_accept(lfd)
52 if sf < 0 { sys_close(cf); return 0 - 1 }
53 if small == 1 {
54 sys_setsockopt(sf, 1, 7, bs, 4) // SO_SNDBUF tiny on sender
55 }
56 out[0] = sf
57 out[1] = cf
58 return 0 }
59
60// build the 8KB patterned payload for frame id
61func bp_fill(p: *u8, id: i64, n: i64) -> i64 {
62 p[0] = (id & 255) as u8; p[1] = ((id >> 8) & 255) as u8
63 p[2] = ((id >> 16) & 255) as u8; p[3] = ((id >> 24) & 255) as u8
64 var j: i64 = 4
65 while j < n { p[j] = ((id * 31 + j) & 255) as u8; j = j + 1 }
66 return 0 }
67
68func bp_check(p: *u8, n: i64) -> i64 { // returns frame id, or -1 on pattern violation
69 let id: i64 = (p[0] & 0xff) | ((p[1] & 0xff) << 8) | ((p[2] & 0xff) << 16) | ((p[3] & 0xff) << 24)
70 var j: i64 = 4
71 while j < n {
72 if (p[j] & 0xff) != ((id * 31 + j) & 255) { return 0 - 1 }
73 j = j + 1
74 }
75 return id }
76
77func main() -> i64 {
78 gw("=== nx_sig2_bp_gate: relay backpressure -- no-stall / frame-atomic / fail-stop / fast-path ===\n" as *u8)
79 nx_signal_ignore(NX_SIGPIPE)
80 let ctr: *i64 = gv_ctr()
81
82 let lfd: i64 = sys_socket(AF_INET, SOCK_STREAM, 0)
83 let ra: *u8 = sys_mmap(8)
84 ra[0] = 1 as u8; ra[1] = 0 as u8; ra[2] = 0 as u8; ra[3] = 0 as u8
85 sys_setsockopt(lfd, 1, 2, ra, 4) // SO_REUSEADDR: reruns never trip TIME_WAIT
86 let la: *u8 = sys_mmap(16)
87 bp_addr(la, 8724)
88 if sys_bind(lfd, la, 16) < 0 { gw("BIND FAIL (port 8724 busy?)\nSIG2-BP: verdict=RED\n" as *u8); return 1 }
89 sys_listen(lfd, 4)
90
91 // tx state (one conn, ci=0) -- exactly what the daemon allocates per conn
92 let txq: *u8 = sys_mmap(NX_S2TX_QBYTES)
93 let txh: *i64 = sys_mmap(8) as *i64
94 let txl: *i64 = sys_mmap(8) as *i64
95 let hscr: *u8 = sys_mmap(16)
96 let tx: *i64 = sys_mmap(32) as *i64
97 tx[0] = txq as i64; tx[1] = txh as i64; tx[2] = txl as i64; tx[3] = hscr as i64
98 txh[0] = 0; txl[0] = 0
99
100 let fds: *i64 = sys_mmap(16) as *i64
101 if bp_pair(lfd, 8724, 1, fds) < 0 { gw("PAIR FAIL\nSIG2-BP: verdict=RED\n" as *u8); return 1 }
102 let sfd: i64 = fds[0]
103 let cfd: i64 = fds[1]
104
105 // ---- T1: flood a WEDGED receiver: bounded time, whole-frame drops, zero fatal ----
106 let pay: *u8 = sys_mmap(8192)
107 var sent: i64 = 0
108 var dropped: i64 = 0
109 var fatal: i64 = 0
110 let t0: i64 = sys_now_us()
111 var i: i64 = 0
112 while i < 300 {
113 bp_fill(pay, i, 8192)
114 let rc: i64 = s2tx_send(tx, 0, sfd, 258, pay, 8192)
115 if rc == 1 { sent = sent + 1 }
116 if rc == 0 { dropped = dropped + 1 }
117 if rc < 0 { fatal = fatal + 1 }
118 i = i + 1
119 }
120 let el: i64 = sys_now_us() - t0
121 gw(" T1 wedged-recv flood: sent=" as *u8); gn(sent)
122 gw(" dropped=" as *u8); gn(dropped)
123 gw(" fatal=" as *u8); gn(fatal)
124 gw(" queued=" as *u8); gn(txl[0])
125 gw("B elapsed_us=" as *u8); gn(el); gw("\n" as *u8)
126 var t1ok: i64 = 0
127 if el < 3000000 { if fatal == 0 { if sent > 0 { if dropped > 0 { t1ok = 1 } } } }
128 gv_check("T1 no-stall: bounded flood on wedged receiver, whole-frame drops, zero fatal" as *u8, t1ok, ctr)
129
130 // ---- T2: receiver resumes -> every ACCEPTED frame arrives byte-perfect, in order, exactly once ----
131 // POLL-FREE by construction: interleave a drain (push app queue -> socket, non-blocking) with a
132 // NON-BLOCKING recv (MSG_DONTWAIT) that pulls everything available. Terminate on the DETERMINISTIC
133 // condition (queue empty AND a stable run of empty reads) -- never on a poll timeout. (A sub-second
134 // ppoll timeout wedged the earlier version: every data-ready poll returns instantly and never consults
135 // the timeout, so the FIRST empty-buffer poll -- iter 66, after all bytes arrive -- hung. Bytes were
136 // proven flowing; only the terminator was wrong.)
137 let acc: *u8 = sys_mmap(1048576)
138 var accn: i64 = 0
139 var quiet: i64 = 0
140 var reading: i64 = 1
141 var iters: i64 = 0
142 var drainrc: i64 = 0
143 let t2start: i64 = sys_now_us()
144 while reading == 1 {
145 drainrc = s2tx_drain(txq, txh, txl, 0, sfd)
146 if drainrc < 0 { reading = 0 } // fatal socket: stop, report below
147 var got_this: i64 = 0
148 var rrun: i64 = 1
149 while rrun == 1 {
150 let rn: i64 = sys_recvfrom(cfd, (acc as i64 + accn) as *u8, 1048576 - accn, NX_S2TX_DONTWAIT, 0 as *u8, 0 as *i64)
151 if rn > 0 { accn = accn + rn; got_this = got_this + rn } else { rrun = 0 }
152 }
153 if txl[0] <= 0 { // queue flushed: count stable empty passes
154 if got_this == 0 { quiet = quiet + 1 } else { quiet = 0 }
155 if quiet >= 200 { reading = 0 } // nothing left in flight -> done
156 }
157 iters = iters + 1
158 if iters > 200000 { reading = 0 } // hard iteration bound
159 if sys_now_us() - t2start > 8000000 { reading = 0 } // 8s wall watchdog
160 }
161 gw(" T2 loop: iters=" as *u8); gn(iters)
162 gw(" accn=" as *u8); gn(accn)
163 gw(" txl_left=" as *u8); gn(txl[0]); gw("\n" as *u8)
164 // offline parse: [0x82][126][len2 BE][payload] (unmasked server frames, 8KB -> extended-16 length)
165 var got: i64 = 0
166 var lastid: i64 = 0 - 1
167 var order_ok: i64 = 1
168 var pat_ok: i64 = 1
169 var torn: i64 = 0
170 var po: i64 = 0
171 while po < accn {
172 if po + 4 > accn { torn = 1; po = accn }
173 else {
174 if (acc[po] & 0xff) != 0x82 { torn = 1; po = accn }
175 else { if (acc[po + 1] & 0xff) != 126 { torn = 1; po = accn }
176 else {
177 let fl: i64 = ((acc[po + 2] & 0xff) << 8) | (acc[po + 3] & 0xff)
178 if po + 4 + fl > accn { torn = 1; po = accn }
179 else {
180 let fid: i64 = bp_check((acc as i64 + po + 4) as *u8, fl)
181 if fl != 8192 { pat_ok = 0 }
182 if fid < 0 { pat_ok = 0 }
183 if fid <= lastid { order_ok = 0 }
184 lastid = fid
185 got = got + 1
186 po = po + 4 + fl
187 }
188 } }
189 }
190 }
191 gw(" T2 resume: received=" as *u8); gn(got)
192 gw("/" as *u8); gn(sent)
193 gw(" bytes=" as *u8); gn(accn)
194 gw(" order_ok=" as *u8); gn(order_ok)
195 gw(" pattern_ok=" as *u8); gn(pat_ok)
196 gw(" torn=" as *u8); gn(torn); gw("\n" as *u8)
197 var t2ok: i64 = 0
198 if got == sent { if order_ok == 1 { if pat_ok == 1 { if torn == 0 { t2ok = 1 } } } }
199 gv_check("T2 integrity: every accepted frame arrives byte-perfect in order exactly once" as *u8, t2ok, ctr)
200
201 // ---- T3: receiver DIES -> sends turn fatal (conn reap), no crash ----
202 sys_close(cfd)
203 var sawfatal: i64 = 0
204 var t3: i64 = 0
205 while t3 < 50 {
206 bp_fill(pay, 9000 + t3, 8192)
207 let rc3: i64 = s2tx_send(tx, 0, sfd, 258, pay, 8192)
208 if rc3 < 0 { sawfatal = 1; t3 = 50 }
209 t3 = t3 + 1
210 }
211 gw(" T3 dead-recv: fatal_seen=" as *u8); gn(sawfatal); gw(" (alive after EPIPE)\n" as *u8)
212 gv_check("T3 fail-stop: sends after receiver death report fatal, no crash" as *u8, sawfatal, ctr)
213 sys_close(sfd)
214
215 // ---- T4: LIVE reader -> direct fast path (queue never used), all frames intact ----
216 txh[0] = 0; txl[0] = 0
217 if bp_pair(lfd, 8724, 0, fds) < 0 { gw("PAIR2 FAIL\nSIG2-BP: verdict=RED\n" as *u8); return 1 }
218 let sfd2: i64 = fds[0]
219 let cfd2: i64 = fds[1]
220 var t4ok: i64 = 1
221 var everq: i64 = 0
222 var a4n: i64 = 0
223 var k: i64 = 0
224 while k < 50 {
225 bp_fill(pay, k, 64)
226 let rc4: i64 = s2tx_send(tx, 0, sfd2, 258, pay, 64)
227 if rc4 != 1 { t4ok = 0 }
228 if txl[0] > 0 { everq = 1 }
229 var rr4: i64 = 1 // non-blocking drain of whatever arrived
230 while rr4 == 1 {
231 let r4: i64 = sys_recvfrom(cfd2, (acc as i64 + a4n) as *u8, 1048576 - a4n, NX_S2TX_DONTWAIT, 0 as *u8, 0 as *i64)
232 if r4 > 0 { a4n = a4n + r4 } else { rr4 = 0 }
233 }
234 k = k + 1
235 }
236 // final sweep: catch any bytes still in flight after the last send (bounded, non-blocking)
237 var fsw: i64 = 0
238 while fsw < 100000 {
239 let r4: i64 = sys_recvfrom(cfd2, (acc as i64 + a4n) as *u8, 1048576 - a4n, NX_S2TX_DONTWAIT, 0 as *u8, 0 as *i64)
240 if r4 > 0 { a4n = a4n + r4; fsw = 0 } else { fsw = fsw + 1 }
241 if a4n >= 50 * 66 { fsw = 100000 } // got all 50 frames -> stop
242 }
243 // parse the 50 small frames: [0x82][64][payload64]
244 var g4: i64 = 0
245 var q4: i64 = 0
246 while q4 + 66 <= a4n {
247 if (acc[q4] & 0xff) != 0x82 { t4ok = 0; q4 = a4n }
248 else { if (acc[q4 + 1] & 0xff) != 64 { t4ok = 0; q4 = a4n }
249 else {
250 if bp_check((acc as i64 + q4 + 2) as *u8, 64) != g4 { t4ok = 0 }
251 g4 = g4 + 1
252 q4 = q4 + 66
253 } }
254 }
255 if g4 != 50 { t4ok = 0 }
256 gw(" T4 live reader: frames=" as *u8); gn(g4)
257 gw("/50 all_direct_ok=" as *u8); gn(t4ok)
258 gw(" ever_queued=" as *u8); gn(everq); gw("\n" as *u8)
259 var t4pass: i64 = 0
260 if t4ok == 1 { if everq == 0 { t4pass = 1 } }
261 gv_check("T4 fast path: live reader gets all frames direct, queue never used" as *u8, t4pass, ctr)
262 sys_close(sfd2); sys_close(cfd2); sys_close(lfd)
263
264 gw("SIG2-BP: " as *u8); gn(ctr[0]); gw("/" as *u8); gn(ctr[1])
265 if ctr[0] == ctr[1] { gw(" verdict=GREEN -- non-blocking frame-atomic relay sends proven on real sockets\n" as *u8) } else { gw(" RED\n" as *u8) }
266 return gv_verdict("SIG2-BP-GATE" as *u8, ctr, "non-blocking frame-atomic relay proven on real sockets" as *u8) }