code wiki / _hdl_build / nx_race_search_worker.nx
nx_race_search_worker.nx source
↩ module page · 194 lines · 7255 B
1// nx_race_search_worker.nx -- the SOVEREIGN LANE of the IM6b SEARCH race
2// (beat-sqlite-FTS5 on the team-log corpus). Same logical work as the
3// competitor lane, self-timing ONLY the engine work (parse excluded):
4// write <prefix>: index the unique team-log lines -> chunked committed
5// segments (fsync'd, .keys + .terms/.post) -> SOVW us= count=
6// query <prefix>: open-store handle + the shared 9-query set (6 terms +
7// 3 ANDs) x 200 iterations -> SOVQ us= q0=..q8=
8// Per-query counts are printed for the orchestrator's correctness
9// cross-check against the FTS5 lane (counts must MATCH or the race is
10// INVALID -- correctness before speed). license_tier: ORIGINAL
11
12import "nx_syscalls.nx"
13import "nx_itoa_lib.nx" // shared MSB-first emitter (zero-alloc)
14import "nx_seg_store.nx"
15const RS_MAGIC_1000000000: i64 = 1000000000
16const RS_MAGIC_2097152: i64 = 2097152
17
18const RS_ITERS: i64 = 200
19const RS_MAXDOCS: i64 = 4000
20
21func rs_p(s: *u8) -> i64 {
22 var n: i64 = 0
23 while s[n] != (0 as u8) { n = n + 1 }
24 sys_write(1, s, n)
25 return 0
26}
27
28// MIGRATED to the shared emitter (debt 1785563586). The old body mmapped a scratch buffer
29// per call and never freed it. At PAGE granularity that is 4096B leaked PER CALL -- the
30// defect that took 28.5GB of a 36GB host in nx_ts_lumadiff (2MB input, ~3.66M calls).
31// nxi_* is MSB-first, allocates NOTHING, and emits identical bytes including the sign.
32func rs_pn(v: i64) -> i64 { nxi_out(v); return 0 }
33
34func rs_streq(a: *u8, b: *u8) -> i64 {
35 var i: i64 = 0
36 while 1 == 1 {
37 if a[i] != b[i] { return 0 }
38 if a[i] == (0 as u8) { return 1 }
39 i = i + 1
40 }
41 return 0
42}
43
44// append the file's unique non-empty lines into pool/ptrs/lens; returns new n
45func rs_parse_file(path: *u8, pool: *u8, pooloff: *i64, ptrs: *i64, lens: *i64, n0: i64) -> i64 {
46 let szp: *i64 = sys_mmap(16) as *i64
47 let b: *u8 = ss_readall(path, szp)
48 let sz: i64 = szp[0]
49 if sz <= 0 { return 0 - 1 }
50 var n: i64 = n0
51 var i: i64 = 0
52 while i < sz {
53 var e: i64 = i
54 while e < sz {
55 if b[e] == (10 as u8) { e = e + RS_MAGIC_1000000000 } else { e = e + 1 }
56 }
57 var lend: i64 = e
58 if e >= RS_MAGIC_1000000000 { lend = e - RS_MAGIC_1000000000 }
59 let llen: i64 = lend - i
60 if llen > 0 {
61 // dedup by exact bytes (length filter first)
62 var dup: i64 = 0
63 var s: i64 = 0
64 while s < n {
65 if dup == 0 { if lens[s] == llen {
66 var eq: i64 = 1
67 let sp: *u8 = ptrs[s] as *u8
68 var x: i64 = 0
69 while x < llen {
70 if eq == 1 { if sp[x] != b[i + x] { eq = 0; x = llen } }
71 x = x + 1
72 }
73 if eq == 1 { dup = 1 }
74 } }
75 s = s + 1
76 }
77 if dup == 0 {
78 if n >= RS_MAXDOCS { return 0 - 2 }
79 let dst: *u8 = (pool as i64 + pooloff[0]) as *u8
80 var t: i64 = 0
81 while t < llen { dst[t] = b[i + t]; t = t + 1 }
82 ptrs[n] = dst as i64
83 lens[n] = llen
84 pooloff[0] = pooloff[0] + llen
85 n = n + 1
86 }
87 }
88 i = lend + 1
89 }
90 return n
91}
92
93func main(argc: i64, argv: *i64) -> i64 {
94 if argc < 3 { rs_p("usage: nx_race_search_worker <write|query> <prefix>\n" as *u8); return 2 }
95 let mode: *u8 = argv[1] as *u8
96 let prefix: *u8 = argv[2] as *u8
97
98 // shared parse (OUTSIDE the timer, both lanes exclude it)
99 let pool: *u8 = sys_mmap(RS_MAGIC_2097152)
100 let pooloff: *i64 = sys_mmap(16) as *i64
101 let ptrs: *i64 = sys_mmap(8 * (RS_MAXDOCS + 8)) as *i64
102 let lens: *i64 = sys_mmap(8 * (RS_MAXDOCS + 8)) as *i64
103 var n: i64 = rs_parse_file("knowledge/status/pm_plan_durable.log" as *u8, pool, pooloff, ptrs, lens, 0)
104 if n < 0 { rs_p("parse FAILED pm_plan\n" as *u8); return 1 }
105 n = rs_parse_file("knowledge/status/cap_registry_durable.log" as *u8, pool, pooloff, ptrs, lens, n)
106 if n < 0 { rs_p("parse FAILED capreg\n" as *u8); return 1 }
107
108 if rs_streq(mode, "write" as *u8) == 1 {
109 let kbuf: *u8 = sys_mmap(64)
110 let t0: i64 = sys_now_us()
111 // ONE bulk segment, one commit point (capacities are data-driven now;
112 // same architecture as the competitor's single indexed transaction)
113 let w: *i64 = ss_begin()
114 var d: i64 = 0
115 while d < n {
116 var ko: i64 = 0
117 ko = ss_cat(kbuf, ko, "d:" as *u8)
118 ko = ss_catn(kbuf, ko, d)
119 kbuf[ko] = 0 as u8
120 if ss_add(w, 1, kbuf, ptrs[d] as *u8, lens[d]) != 0 { rs_p("add FAILED\n" as *u8); return 1 }
121 d = d + 1
122 }
123 if ss_commit(prefix, w, 1001) != 0 { rs_p("commit FAILED\n" as *u8); return 1 }
124 let t1: i64 = sys_now_us()
125 rs_p("SOVW us=" as *u8)
126 rs_pn(t1 - t0)
127 rs_p(" count=" as *u8)
128 rs_pn(n)
129 rs_p("\n" as *u8)
130 return 0
131 }
132
133 if rs_streq(mode, "query" as *u8) == 1 {
134 // the SHARED query set (identical order in the FTS5 lane)
135 let qt: *i64 = sys_mmap(8 * 8) as *i64
136 qt[0] = "infomgmt" as *u8 as i64
137 qt[1] = "compaction" as *u8 as i64
138 qt[2] = "postings" as *u8 as i64
139 qt[3] = "sovereign" as *u8 as i64
140 qt[4] = "capreg" as *u8 as i64
141 qt[5] = "gate" as *u8 as i64
142 let aa: *i64 = sys_mmap(8 * 8) as *i64
143 aa[0] = "infomgmt" as *u8 as i64
144 aa[1] = "postings" as *u8 as i64
145 aa[2] = "sovereign" as *u8 as i64
146 aa[3] = "gate" as *u8 as i64
147 aa[4] = "compaction" as *u8 as i64
148 aa[5] = "archive" as *u8 as i64
149 let kp: *i64 = sys_mmap(8 * (RS_MAXDOCS + 8)) as *i64
150 let kl: *i64 = sys_mmap(8 * (RS_MAXDOCS + 8)) as *i64
151 let qc: *i64 = sys_mmap(8 * 16) as *i64
152 let at: *i64 = sys_mmap(8 * 4) as *i64
153 let t0: i64 = sys_now_us()
154 let h: *i64 = ss_open(prefix)
155 var it: i64 = 0
156 while it < RS_ITERS {
157 var q: i64 = 0
158 while q < 6 {
159 let c: i64 = ss_term(h, qt[q] as *u8, kp, kl, RS_MAXDOCS)
160 if c < 0 { rs_p("term FAILED\n" as *u8); return 1 }
161 qc[q] = c
162 q = q + 1
163 }
164 var a: i64 = 0
165 while a < 3 {
166 at[0] = aa[a * 2]
167 at[1] = aa[a * 2 + 1]
168 let c2: i64 = ss_term_and(h, at, 2, kp, kl, RS_MAXDOCS)
169 if c2 < 0 { rs_p("and FAILED\n" as *u8); return 1 }
170 qc[6 + a] = c2
171 a = a + 1
172 }
173 it = it + 1
174 }
175 let t1: i64 = sys_now_us()
176 rs_p("SOVQ us=" as *u8)
177 rs_pn(t1 - t0)
178 var q2: i64 = 0
179 while q2 < 9 {
180 rs_p(" q" as *u8)
181 rs_pn(q2)
182 rs_p("=" as *u8)
183 rs_pn(qc[q2])
184 q2 = q2 + 1
185 }
186 rs_p(" count=" as *u8)
187 rs_pn(n)
188 rs_p("\n" as *u8)
189 return 0
190 }
191
192 rs_p("unknown mode\n" as *u8)
193 return 2
194}