code wiki / _hdl_build / nx_team_out_lock_gate.nx
nx_team_out_lock_gate.nx source
↩ module page · 280 lines · 12171 B
1// nx_team_out_lock_gate.nx -- REFEREE for the team_out.tsv concurrent-writer lock (WMS torn-write
2// follow-on). Proves an rt_lock-serialized WRITE SESSION keeps a MULTI-LINE record block contiguous
3// under concurrency, AND that the detector sees failure without the lock (mandatory neg-control).
4//
5// WHY a lock, not fa_appendz: express_lane writes a MULTI-LINE block (ACK lines + the SITREP block);
6// per-line atomicity (fa_appendz) keeps each LINE intact but does NOT stop a second writer's lines
7// from interleaving BETWEEN this writer's lines -> the block is shredded. The fix (this gate proves)
8// is to hold rt_lock across the whole multi-line session so no other writer can slip a line in.
9//
10// GOOD lane: NWORKERS fork; each acquires rt_lock(GOODLOCK), writes NRECS records (each = L lines,
11// each line emitted as a SEQUENCE of sys_write so it CAN tear), releases. Expect: ZERO
12// torn lines AND ZERO block-contiguity breaks AND all N*NRECS*L lines present.
13// BAD lane (NEG-CONTROL): identical, but NO lock -> concurrent workers interleave -> torn lines
14// AND/OR block breaks MUST appear. If bad has none, the detector is worthless -> RED.
15//
16// A line is CLEAN iff it begins "BLK w=" AND the 4 bytes before '\n' are " END" AND exactly one
17// "BLK w=" head AND its w field == its z field (the two wid copies bracket the line; an interleave
18// splits them). A BLOCK BREAK = a clean line with s>0 whose immediately-preceding line is not the
19// same (w,r) at s-1 (the block's L lines were not contiguous). Per-run unique /tmp paths.
20// Self-log via the assembled single sys_write (fd is uncontended). GREEN(0) iff all hold.
21// Sovereign: nx_registry_lock (rt_lock/rt_unlock; nx_syscalls transitive). license_tier: ORIGINAL
22import "nx_registry_lock.nx"
23
24const NWORKERS: i64 = 16
25const NRECS: i64 = 60
26const LPB: i64 = 4 // lines per record block
27const MAXLINES: i64 = 8192 // >= NWORKERS*NRECS*LPB (3840) with headroom
28
29func so(s: *u8) -> i64 { var n: i64=0; while s[n]!=(0 as u8){n=n+1} sys_write(1,s,n); return 0 }
30func son(v: i64) -> i64 {
31 let bb: *u8=sys_mmap(28); var m: i64=v
32 if m<0 { sys_write(1,"-" as *u8,1); m=0-m }
33 let t: *u8=sys_mmap(28); var k: i64=0
34 if m==0 { t[0]=48 as u8; k=1 }
35 while m>0 { t[k]=(48+(m%10)) as u8; m=m/10; k=k+1 }
36 var i: i64=0; while i<k { bb[i]=t[k-1-i]; i=i+1 }
37 sys_write(1,bb,k); return 0
38}
39func tl_cat(dst: *u8, off: i64, s: *u8) -> i64 { var i: i64=0; while s[i]!=(0 as u8){dst[off+i]=s[i];i=i+1} return off+i }
40func tl_catn(dst: *u8, off: i64, v: i64) -> i64 {
41 var m: i64=v; var o: i64=off
42 if m==0 { dst[o]=48 as u8; return o+1 }
43 let t: *u8=sys_mmap(28); var k: i64=0
44 while m>0 { t[k]=(48+(m%10)) as u8; m=m/10; k=k+1 }
45 var i: i64=0; while i<k { dst[o+i]=t[k-1-i]; i=i+1 }
46 return o+k
47}
48func tl_path(out: *u8, tag: *u8, epoch: i64, suffix: *u8) -> i64 {
49 var o: i64=0
50 o=tl_cat(out,o,"/tmp/" as *u8); o=tl_cat(out,o,tag); o=tl_catn(out,o,epoch); o=tl_cat(out,o,suffix)
51 out[o]=0 as u8; return o
52}
53
54// emit one line as a SEQUENCE of writes (so it tears without protection): "BLK w=<wid> r=<r> s=<s> z=<wid> END\n"
55func emit_line(fd: i64, wid: i64, r: i64, s: i64) -> i64 {
56 let nb: *u8 = sys_mmap(28)
57 sys_write(fd, "BLK w=" as *u8, 6)
58 var l: i64=0; var m: i64=wid
59 if m==0 { nb[0]=48 as u8; l=1 }
60 while m>0 { nb[l]=(48+(m%10)) as u8; m=m/10; l=l+1 }
61 let d1: *u8=sys_mmap(28); var x: i64=0
62 while x<l { d1[x]=nb[l-1-x]; x=x+1 }
63 sys_write(fd, d1, l)
64 sys_write(fd, " r=" as *u8, 3)
65 l=0; m=r
66 if m==0 { nb[0]=48 as u8; l=1 }
67 while m>0 { nb[l]=(48+(m%10)) as u8; m=m/10; l=l+1 }
68 let d2: *u8=sys_mmap(28); x=0
69 while x<l { d2[x]=nb[l-1-x]; x=x+1 }
70 sys_write(fd, d2, l)
71 sys_write(fd, " s=" as *u8, 3)
72 l=0; m=s
73 if m==0 { nb[0]=48 as u8; l=1 }
74 while m>0 { nb[l]=(48+(m%10)) as u8; m=m/10; l=l+1 }
75 let d3: *u8=sys_mmap(28); x=0
76 while x<l { d3[x]=nb[l-1-x]; x=x+1 }
77 sys_write(fd, d3, l)
78 sys_write(fd, " z=" as *u8, 3)
79 l=0; m=wid
80 if m==0 { nb[0]=48 as u8; l=1 }
81 while m>0 { nb[l]=(48+(m%10)) as u8; m=m/10; l=l+1 }
82 let d4: *u8=sys_mmap(28); x=0
83 while x<l { d4[x]=nb[l-1-x]; x=x+1 }
84 sys_write(fd, d4, l)
85 sys_write(fd, " END\n" as *u8, 5)
86 return 0
87}
88
89// one worker: NRECS blocks of LPB lines. locked=1 -> rt_lock the whole session.
90func worker(path: *u8, lockpath: *u8, wid: i64, locked: i64) -> i64 {
91 var lkfd: i64 = 0 - 1
92 if locked == 1 { lkfd = rt_lock(lockpath) }
93 let fd: i64 = sys_openat_append(path, 0x1a4)
94 if fd < 0 { if lkfd >= 0 { rt_unlock(lockpath, lkfd) } return 0 - 1 }
95 var r: i64 = 0
96 while r < NRECS {
97 var s: i64 = 0
98 while s < LPB { emit_line(fd, wid, r, s); s = s + 1 }
99 r = r + 1
100 }
101 sys_close(fd)
102 if lkfd >= 0 { rt_unlock(lockpath, lkfd) }
103 return 0
104}
105
106func spawn_all(path: *u8, lockpath: *u8, locked: i64) -> i64 {
107 let pids: *i64 = sys_mmap(8 * (NWORKERS + 4)) as *i64
108 var w: i64 = 0
109 while w < NWORKERS {
110 let pid: i64 = sys_fork()
111 if pid == 0 { worker(path, lockpath, w, locked); sys_exit(0) }
112 pids[w] = pid
113 w = w + 1
114 }
115 let st: *i64 = sys_mmap(16) as *i64
116 w = 0
117 while w < NWORKERS { sys_wait4(pids[w], st, 0); w = w + 1 }
118 return 0
119}
120
121// parse decimal after token `tok`(tlen) within [ls,le); -1 if absent.
122func int_after(b: *u8, ls: i64, le: i64, tok: *u8, tlen: i64) -> i64 {
123 var p: i64 = ls
124 while p + tlen <= le {
125 var k: i64 = 0
126 var hit: i64 = 1
127 while k < tlen { if b[p+k] != tok[k] { hit = 0; k = tlen } else { k = k + 1 } }
128 if hit == 1 {
129 var v: i64 = 0
130 var q: i64 = p + tlen
131 var any: i64 = 0
132 while q < le { let c: i64 = b[q] as i64; if c < 48 { q = le } else { if c > 57 { q = le } else { v = v*10+(c-48); any = 1; q = q + 1 } } }
133 if any == 1 { return v }
134 return 0 - 1
135 }
136 p = p + 1
137 }
138 return 0 - 1
139}
140func head_count(b: *u8, ls: i64, le: i64) -> i64 {
141 let h: *u8 = "BLK w=" as *u8
142 var c: i64 = 0
143 var p: i64 = ls
144 while p + 6 <= le {
145 var k: i64 = 0; var hit: i64 = 1
146 while k < 6 { if b[p+k] != h[k] { hit = 0; k = 6 } else { k = k + 1 } }
147 if hit == 1 { c = c + 1 }
148 p = p + 1
149 }
150 return c
151}
152
153// scan file: fill w[],r[],s[],clean[] per line; outs[0]=lines outs[1]=torn outs[2]=block_breaks
154func scan(path: *u8, wA: *i64, rA: *i64, sA: *i64, clA: *i64, outs: *i64) -> i64 {
155 let szp: *i64 = sys_mmap(16) as *i64
156 let b: *u8 = sys_read_file(path, szp)
157 let sz: i64 = szp[0]
158 var lines: i64 = 0
159 var torn: i64 = 0
160 var ls: i64 = 0
161 var i: i64 = 0
162 while i < sz {
163 if b[i] == (10 as u8) {
164 if lines < MAXLINES {
165 let le: i64 = i
166 var clean: i64 = 1
167 if le - ls < 10 { clean = 0 }
168 if clean == 1 {
169 if b[ls]!=(66 as u8) { clean=0 } // B
170 if b[ls+1]!=(76 as u8) { clean=0 } // L
171 if b[ls+2]!=(75 as u8) { clean=0 } // K
172 if b[ls+3]!=(32 as u8) { clean=0 } // space
173 if b[ls+4]!=(119 as u8) { clean=0 } // w
174 if b[ls+5]!=(61 as u8) { clean=0 } // =
175 }
176 if clean == 1 {
177 let e: i64 = le - 1
178 if b[e-3]!=(32 as u8) { clean=0 }
179 if b[e-2]!=(69 as u8) { clean=0 } // E
180 if b[e-1]!=(78 as u8) { clean=0 } // N
181 if b[e]!=(68 as u8) { clean=0 } // D
182 }
183 var wv: i64 = 0 - 1
184 var rv: i64 = 0 - 1
185 var sv: i64 = 0 - 1
186 if clean == 1 {
187 if head_count(b, ls, le) != 1 { clean = 0 }
188 }
189 if clean == 1 {
190 wv = int_after(b, ls, le, "w=" as *u8, 2)
191 rv = int_after(b, ls, le, " r=" as *u8, 3)
192 sv = int_after(b, ls, le, " s=" as *u8, 3)
193 let zv: i64 = int_after(b, ls, le, " z=" as *u8, 3)
194 if wv < 0 { clean = 0 }
195 if wv != zv { clean = 0 } // split mid-line -> w and z differ
196 }
197 wA[lines] = wv; rA[lines] = rv; sA[lines] = sv; clA[lines] = clean
198 if clean == 0 { torn = torn + 1 }
199 lines = lines + 1
200 }
201 ls = i + 1
202 }
203 i = i + 1
204 }
205 // block contiguity: a clean line with s>0 must follow the same (w,r) at s-1
206 var breaks: i64 = 0
207 var j: i64 = 0
208 while j < lines {
209 if clA[j] == 1 { if sA[j] > 0 {
210 var okp: i64 = 0
211 if j > 0 { if clA[j-1] == 1 { if wA[j-1] == wA[j] { if rA[j-1] == rA[j] { if sA[j-1] == sA[j]-1 { okp = 1 } } } } }
212 if okp == 0 { breaks = breaks + 1 }
213 } }
214 j = j + 1
215 }
216 outs[0] = lines
217 outs[1] = torn
218 outs[2] = breaks
219 return lines
220}
221
222func main() -> i64 {
223 let epoch: i64 = sys_now_realtime_sec()
224 let gp: *u8 = sys_mmap(128)
225 let gl: *u8 = sys_mmap(128)
226 let bp: *u8 = sys_mmap(128)
227 let bl: *u8 = sys_mmap(128)
228 tl_path(gp, "tol_g" as *u8, epoch, ".log" as *u8)
229 tl_path(gl, "tol_g" as *u8, epoch, ".log.lock" as *u8)
230 tl_path(bp, "tol_b" as *u8, epoch, ".log" as *u8)
231 tl_path(bl, "tol_b" as *u8, epoch, ".log.lock" as *u8)
232
233 spawn_all(gp, gl, 1) // GOOD: locked sessions
234 spawn_all(bp, bl, 0) // BAD : unlocked (neg-control)
235
236 let wA: *i64 = sys_mmap(8 * MAXLINES) as *i64
237 let rA: *i64 = sys_mmap(8 * MAXLINES) as *i64
238 let sA: *i64 = sys_mmap(8 * MAXLINES) as *i64
239 let clA: *i64 = sys_mmap(8 * MAXLINES) as *i64
240 let go: *i64 = sys_mmap(32) as *i64
241 let bo: *i64 = sys_mmap(32) as *i64
242 scan(gp, wA, rA, sA, clA, go)
243 scan(bp, wA, rA, sA, clA, bo)
244 let want_lines: i64 = NWORKERS * NRECS * LPB
245
246 var green: i64 = 1
247 if go[0] != want_lines { green = 0 }
248 if go[1] != 0 { green = 0 } // good torn==0
249 if go[2] != 0 { green = 0 } // good block_breaks==0
250 if bo[1] <= 0 { if bo[2] <= 0 { green = 0 } } // bad MUST tear or break (neg-control)
251
252 so("team_out concurrent-writer LOCK gate (rt_lock-serialized multi-line block session)\n" as *u8)
253 so(" GOOD(locked) lines=" as *u8); son(go[0]); so("/" as *u8); son(want_lines)
254 so(" torn=" as *u8); son(go[1]); so(" block_breaks=" as *u8); son(go[2])
255 if go[0]==want_lines { if go[1]==0 { if go[2]==0 { so(" PASS\n" as *u8) } else { so(" FAIL\n" as *u8) } } else { so(" FAIL\n" as *u8) } } else { so(" FAIL\n" as *u8) }
256 so(" BAD(unlocked) torn=" as *u8); son(bo[1]); so(" block_breaks=" as *u8); son(bo[2]); so(" want=POSITIVE" as *u8)
257 if bo[1] > 0 { so(" PASS(detector-sees-interleave)\n" as *u8) } else { if bo[2] > 0 { so(" PASS(detector-sees-interleave)\n" as *u8) } else { so(" FAIL(no-interleave)\n" as *u8) } }
258 so("verdict=" as *u8)
259 if green==1 { so("GREEN\n" as *u8) } else { so("RED\n" as *u8) }
260
261 // single assembled evidence record -> one sys_write (uncontended log fd)
262 let lf: i64 = sys_openat_append("knowledge/status/team_out_lock_gate.log" as *u8, 0x1a4)
263 if lf >= 0 {
264 let rec: *u8 = sys_mmap(256)
265 var o: i64 = 0
266 o = tl_cat(rec, o, "TEAMOUT-LOCK epoch=" as *u8); o = tl_catn(rec, o, epoch)
267 o = tl_cat(rec, o, " good_torn=" as *u8); o = tl_catn(rec, o, go[1])
268 o = tl_cat(rec, o, " good_breaks=" as *u8); o = tl_catn(rec, o, go[2])
269 o = tl_cat(rec, o, " bad_torn=" as *u8); o = tl_catn(rec, o, bo[1])
270 o = tl_cat(rec, o, " bad_breaks=" as *u8); o = tl_catn(rec, o, bo[2])
271 o = tl_cat(rec, o, " verdict=" as *u8)
272 if green==1 { o = tl_cat(rec, o, "GREEN" as *u8) } else { o = tl_cat(rec, o, "RED" as *u8) }
273 rec[o] = 10 as u8
274 sys_write(lf, rec, o + 1)
275 sys_close(lf)
276 }
277
278 if green==1 { return 0 }
279 return 1
280}