code wiki / _hdl_build / nx_toolatency_observation_lib.nx
nx_toolatency_observation_lib.nx source
↩ module page · 279 lines · 16337 B
1// Read-only projection of the existing tools/timing timing_v=1 actlog records.
2// Format constants only; the caller supplies read and row capacities.
3import "nx_syscalls.nx"
4const TO_MAX: i64 = 9223372036854775807
5const TO_MIN: i64 = 0-9223372036854775807-1
6const TO_WORD_BYTES: i64 = 8
7const TO_ROW_SLOTS: i64 = 17
8const TO_NUM_COUNT: i64 = 9
9const TO_FIELD_COUNT: i64 = 6
10const TO_TOKEN_COUNT: i64 = 15
11const TO_STATUS_OK: i64 = 0
12const TO_STATUS_UNKNOWN: i64 = 1
13const TO_STATUS_WINDOW: i64 = 2
14const TO_STATUS_INPUT: i64 = 3
15const TO_STATUS_OUTPUT: i64 = 4
16const TO_ERR_OFFSET: i64 = 0-10001
17const TO_ERR_SEEK_POSITION: i64 = 0-10002
18const TO_ERR_SHORT_READ: i64 = 0-10003
19// Row: epoch,job,pid,origin,begin,end,elapsed,code,bytes,tool-off/n,phase-off/n,lane-off/n,raw-off/n.
20// Window: status,error,size,bytes,next,examined,matched,stored,rejected,prefix,trailing.
21const TO_STATE_SLOTS: i64 = 11
22static to_write_failed: i64
23func to_len(s: *u8) -> i64 { var n: i64=0; while s[n] != (0 as u8) { n=n+1 }; return n }
24func to_eq(b: *u8, s: i64, e: i64, text: *u8) -> i64 {
25 let n: i64=to_len(text); if e-s != n { return 0 }
26 var i: i64=0; while i < n { if b[s+i] != text[i] { return 0 }; i=i+1 }; return 1
27}
28func to_has(b: *u8, s: i64, e: i64, text: *u8) -> i64 {
29 let n: i64=to_len(text); var i: i64=s
30 while i+n <= e { if to_eq(b,i,i+n,text) == 1 { return 1 }; i=i+1 }; return 0
31}
32// Negative accumulation admits signed minimum without abs(INT64_MIN) or overflow.
33func to_int(b: *u8, s: i64, e: i64, out: *i64) -> i64 {
34 if s >= e { return 0 }; var p: i64=s; var negative: i64=0
35 if b[p] == (45 as u8) { negative=1; p=p+1 }
36 if p >= e { return 0 }
37 var lower: i64=0-TO_MAX; if negative == 1 { lower=TO_MIN }
38 var v: i64=0
39 while p < e {
40 let d: i64=(b[p] as i64)-48
41 if d < 0 || d > 9 { return 0 }
42 if v < lower/10 { return 0 }; v=v*10
43 if v < lower+d { return 0 }; v=v-d; p=p+1
44 }
45 if negative == 0 { v=0-v }; out[0]=v; return 1
46}
47func to_utf8(b: *u8, s: i64, e: i64) -> i64 {
48 var i: i64=s
49 while i < e {
50 let c: i64=b[i] as i64
51 if c < 128 { if c == 0 { return 0 }; i=i+1 } else {
52 var count: i64=0; var code: i64=0; var floor: i64=0
53 if c >= 194 && c <= 223 { count=2; code=c-192; floor=128 }
54 if c >= 224 && c <= 239 { count=3; code=c-224; floor=2048 }
55 if c >= 240 && c <= 244 { count=4; code=c-240; floor=65536 }
56 if count == 0 || i+count > e { return 0 }
57 var k: i64=1
58 while k < count { let d: i64=b[i+k] as i64; if d < 128 || d > 191 { return 0 }; code=code*64+d-128; k=k+1 }
59 if code < floor || code > 1114111 || (code >= 55296 && code <= 57343) { return 0 }; i=i+count
60 }
61 }; return 1
62}
63func to_token_key(index: i64) -> *u8 {
64 if index == 0 { return "tools/timing" as *u8 }
65 if index == 1 { return "timing_v=1" as *u8 }
66 if index == 2 { return "lane=" as *u8 }
67 if index == 3 { return "job=" as *u8 }
68 if index == 4 { return "request_pid=" as *u8 }
69 if index == 5 { return "request_begin_us=" as *u8 }
70 if index == 6 { return "clock=monotonic_us" as *u8 }
71 if index == 7 { return "clock_scope=same_process_tree" as *u8 }
72 if index == 8 { return "begin_us=" as *u8 }
73 if index == 9 { return "end_us=" as *u8 }
74 if index == 10 { return "elapsed_us=" as *u8 }
75 if index == 11 { return "code=" as *u8 }
76 if index == 12 { return "bytes=" as *u8 }
77 if index == 13 { return "client_first_byte=unobserved" as *u8 }
78 return "result_first_byte=unobserved" as *u8
79}
80func to_num_slot(token: i64) -> i64 {
81 if token == 3 { return 1 }; if token == 4 { return 2 }; if token == 5 { return 3 }
82 if token >= 8 && token <= 12 { return token-4 }; return 0-1
83}
84// Exact producer grammar; 0 unrelated,1 valid,-1 malformed/unsupported timing record.
85// Fixed token order also refuses duplicate keys rather than choosing first/last silently.
86func to_parse(b: *u8, s: i64, e: i64, row: *i64) -> i64 {
87 if to_has(b,s,e,"tools/timing" as *u8) == 0 { return 0 }
88 if to_utf8(b,s,e) != 1 { return 0-1 }
89 var p: i64=s; var field: i64=0; var start: i64=s
90 while p <= e {
91 var boundary: i64=0; if p == e { boundary=1 } else { if b[p] == (9 as u8) { boundary=1 } }
92 if boundary == 1 {
93 if field >= TO_FIELD_COUNT || start == p { return 0-1 }
94 if field == 0 { if to_int(b,start,p,row) != 1 || row[0] < 0 { return 0-1 } }
95 if field == 1 { if to_eq(b,start,p,"mcp" as *u8) != 1 { return 0-1 } }
96 if field == 2 { row[9]=start; row[10]=p-start }
97 if field == 3 { if to_eq(b,start,p,"observe" as *u8) != 1 { return 0-1 } }
98 if field == 4 { row[11]=start; row[12]=p-start }
99 if field == 5 {
100 var t: i64=start; var index: i64=0
101 while t < p {
102 if index >= TO_TOKEN_COUNT { return 0-1 }
103 var stop: i64=t; while stop < p && b[stop] != (32 as u8) { stop=stop+1 }
104 let key: *u8=to_token_key(index); let kn: i64=to_len(key)
105 if t+kn > stop || to_eq(b,t,t+kn,key) != 1 { return 0-1 }
106 let num: i64=to_num_slot(index)
107 if num >= 0 {
108 if to_int(b,t+kn,stop,((row as i64)+num*TO_WORD_BYTES) as *i64) != 1 { return 0-1 }
109 if num != 7 && row[num] < (0-1) { return 0-1 }
110 } else {
111 if index == 2 { if t+kn == stop { return 0-1 }; row[13]=t+kn; row[14]=stop-t-kn } else {
112 if t+kn != stop { return 0-1 }
113 }
114 }
115 index=index+1; t=stop+1
116 if stop < p && t == p { return 0-1 }
117 }
118 if index != TO_TOKEN_COUNT { return 0-1 }
119 }
120 field=field+1; start=p+1
121 } else { if b[p] < (32 as u8) { return 0-1 } }
122 p=p+1
123 }
124 if field != TO_FIELD_COUNT { return 0-1 }
125 var elapsed: i64=0-1
126 if row[4] >= 0 && row[5] >= row[4] { elapsed=row[5]-row[4] }
127 if row[6] != elapsed { return 0-1 }
128 row[15]=s; row[16]=e-s; return 1
129}
130func to_init(st: *i64) -> i64 { var i: i64=0; while i < TO_STATE_SLOTS { st[i]=0; i=i+1 }; return 0 }
131// `length` contains exactly the checked bytes of the initial file-size window.
132func to_scan(b: *u8, length: i64, offset: i64, size: i64, boundary: i64, rows: *i64, capacity: i64, scratch: *i64, st: *i64) -> i64 {
133 to_init(st); st[2]=size; st[3]=length; st[4]=offset
134 var p: i64=0
135 if boundary == 0 {
136 while p < length && b[p] != (10 as u8) { p=p+1 }
137 if p < length { p=p+1 }; st[9]=p; st[4]=offset+p
138 st[0]=TO_STATUS_WINDOW
139 }
140 while p < length {
141 let start: i64=p; var e: i64=p
142 while e < length && b[e] != (10 as u8) { e=e+1 }
143 if e == length { st[10]=length-start; st[4]=offset+start; if st[0] == 0 { st[0]=TO_STATUS_WINDOW }; return st[0] }
144 let parsed: i64=to_parse(b,start,e,scratch)
145 if parsed == 1 && st[7] >= capacity {
146 st[4]=offset+start; if st[0] == 0 { st[0]=TO_STATUS_WINDOW }; return st[0]
147 }
148 st[5]=st[5]+1
149 if parsed != 0 { st[6]=st[6]+1 }
150 if parsed < 0 { st[8]=st[8]+1; st[0]=TO_STATUS_UNKNOWN }
151 if parsed == 1 {
152 var k: i64=0; while k < TO_ROW_SLOTS { rows[st[7]*TO_ROW_SLOTS+k]=scratch[k]; k=k+1 }; st[7]=st[7]+1
153 }
154 p=e+1; st[4]=offset+p
155 }
156 if st[4] < size && st[0] == 0 { st[0]=TO_STATUS_WINDOW }; return st[0]
157}
158// Opens read-only, bounds all reads by the initial size, checks seek/read/close.
159func to_read(path: *u8, offset: i64, readcap: i64, b: *u8, rows: *i64, rowcap: i64, scratch: *i64, st: *i64) -> i64 {
160 to_init(st); st[2]=0-1; st[4]=offset
161 let fd: i64=sys_openat_rd(path)
162 if fd < 0 { st[0]=TO_STATUS_UNKNOWN; st[1]=fd; return st[0] }
163 let size: i64=sys_lseek(fd,0,2); st[2]=size
164 var err: i64=0; var boundary: i64=1; var got: i64=0
165 if size < 0 { err=size } else { if offset > size { err=TO_ERR_OFFSET; st[0]=TO_STATUS_INPUT } }
166 if err == 0 && offset > 0 {
167 let pos: i64=sys_lseek(fd,offset-1,0)
168 if pos != offset-1 { if pos < 0 { err=pos } else { err=TO_ERR_SEEK_POSITION } } else {
169 let n: i64=sys_read(fd,b,1)
170 if n != 1 { if n < 0 { err=n } else { err=TO_ERR_SHORT_READ } } else { if b[0] != (10 as u8) { boundary=0 } }
171 }
172 }
173 if err == 0 {
174 let pos: i64=sys_lseek(fd,offset,0)
175 if pos != offset { if pos < 0 { err=pos } else { err=TO_ERR_SEEK_POSITION } }
176 }
177 var wanted: i64=0
178 if err == 0 { wanted=size-offset; if wanted > readcap { wanted=readcap } }
179 while err == 0 && got < wanted {
180 let n: i64=sys_read(fd,((b as i64)+got) as *u8,wanted-got)
181 if n <= 0 { if n < 0 { err=n } else { err=TO_ERR_SHORT_READ } } else { got=got+n }
182 }
183 let closed: i64=sys_close(fd); if closed != 0 && err == 0 { err=closed }
184 if err != 0 { if st[0] != TO_STATUS_INPUT { st[0]=TO_STATUS_UNKNOWN }; st[1]=err; st[3]=got; return st[0] }
185 return to_scan(b,got,offset,size,boundary,rows,rowcap,scratch,st)
186}
187func to_raw(b: *u8, n: i64) -> i64 {
188 if to_write_failed != 0 { return 0-1 }; var p: i64=0
189 while p < n { let w: i64=sys_write(1,((b as i64)+p) as *u8,n-p); if w <= 0 { to_write_failed=1; return 0-1 }; p=p+w }; return 0
190}
191func to_text(b: *u8) -> i64 { return to_raw(b,to_len(b)) }
192func to_number(v: i64, scratch: *u8) -> i64 {
193 if v == TO_MIN { return to_text("-9223372036854775808" as *u8) }
194 var x: i64=v; if x < 0 { x=0-x }; var p: i64=TO_NUM_COUNT*TO_WORD_BYTES
195 if x == 0 { p=p-1; scratch[p]=48 as u8 }
196 while x > 0 { p=p-1; scratch[p]=(48+(x%10)) as u8; x=x/10 }
197 if v < 0 { p=p-1; scratch[p]=45 as u8 }
198 return to_raw(((scratch as i64)+p) as *u8,TO_NUM_COUNT*TO_WORD_BYTES-p)
199}
200func to_nullable(v: i64, scratch: *u8) -> i64 { if v == (0-1) { return to_text("null" as *u8) }; return to_number(v,scratch) }
201func to_quote(b: *u8, s: i64, n: i64) -> i64 {
202 to_text("\"" as *u8); var i: i64=0
203 while i < n {
204 if b[s+i] == (34 as u8) || b[s+i] == (92 as u8) { to_text("\\" as *u8) }
205 to_raw(((b as i64)+s+i) as *u8,1); i=i+1
206 }; return to_text("\"" as *u8)
207}
208func to_status(code: i64) -> *u8 {
209 if code == 0 { return "COMPLETE" as *u8 }; if code == 1 { return "UNKNOWN" as *u8 }
210 if code == 2 { return "WINDOW" as *u8 }; return "BAD_INPUT" as *u8
211}
212func to_num_key(index: i64) -> *u8 {
213 if index == 0 { return "observed_unix" as *u8 }; if index == 1 { return "job" as *u8 }
214 if index == 2 { return "request_pid" as *u8 }; if index == 3 { return "request_begin_us" as *u8 }
215 if index == 4 { return "begin_us" as *u8 }; if index == 5 { return "end_us" as *u8 }
216 if index == 6 { return "elapsed_us" as *u8 }; if index == 7 { return "code" as *u8 }; return "bytes" as *u8
217}
218func to_emit(b: *u8, rows: *i64, st: *i64, offset: i64, readcap: i64, rowcap: i64, numeric: *u8) -> i64 {
219 to_text("{\"schema\":\"nishi.tool-observations\",\"v\":1,\"status\":\"" as *u8); to_text(to_status(st[0]))
220 to_text("\",\"source\":\"knowledge/status/actlog.jrnl\",\"scope\":\"initial_size_bounded_non_atomic_file_window\",\"history_complete\":null,\"source_identity_verified\":false,\"client_first_byte\":null,\"result_first_byte\":null,\"request\":{\"offset\":" as *u8)
221 to_number(offset,numeric); to_text(",\"read_capacity\":" as *u8); to_number(readcap,numeric); to_text(",\"row_capacity\":" as *u8); to_number(rowcap,numeric)
222 to_text("},\"window\":{\"reader_error\":" as *u8); to_number(st[1],numeric)
223 to_text(",\"reader_error_domain\":\"negative_errno_or_local_10001_offset_10002_seek_position_10003_short_read\"" as *u8)
224 to_text(",\"initial_size\":" as *u8); to_nullable(st[2],numeric); to_text(",\"bytes_read\":" as *u8); to_number(st[3],numeric)
225 to_text(",\"next_offset\":" as *u8); to_number(st[4],numeric); to_text(",\"lines_examined\":" as *u8); to_number(st[5],numeric)
226 to_text(",\"timing_lines_examined\":" as *u8); to_number(st[6],numeric); to_text(",\"returned_rows\":" as *u8); to_number(st[7],numeric)
227 to_text(",\"rejected_timing_lines\":" as *u8); to_number(st[8],numeric); to_text(",\"skipped_leading_bytes\":" as *u8); to_number(st[9],numeric)
228 to_text(",\"unconsumed_trailing_bytes\":" as *u8); to_number(st[10],numeric)
229 to_text("},\"rows\":[" as *u8); var i: i64=0
230 let hex: *u8="0123456789abcdef" as *u8
231 while i < st[7] {
232 if i > 0 { to_text("," as *u8) }; let row: *i64=((rows as i64)+i*TO_ROW_SLOTS*TO_WORD_BYTES) as *i64
233 to_text("{\"source_offset\":" as *u8); to_number(offset+row[15],numeric)
234 to_text(",\"source_length_without_newline\":" as *u8); to_number(row[16],numeric)
235 to_text(",\"raw_row_hex\":\"" as *u8); var p: i64=0
236 while p < row[16] { let c: i64=b[row[15]+p] as i64; to_raw(((hex as i64)+c/16) as *u8,1); to_raw(((hex as i64)+c%16) as *u8,1); p=p+1 }
237 to_text("\",\"tool\":" as *u8); to_quote(b,row[9],row[10]); to_text(",\"phase\":" as *u8); to_quote(b,row[11],row[12]); to_text(",\"lane\":" as *u8); to_quote(b,row[13],row[14])
238 var k: i64=0
239 while k < TO_NUM_COUNT {
240 to_text(",\"" as *u8); to_text(to_num_key(k)); to_text("\":" as *u8)
241 if k == 7 { to_number(row[k],numeric) } else { to_nullable(row[k],numeric) }; k=k+1
242 }
243 to_text(",\"clock\":\"monotonic_us\",\"clock_scope\":\"same_process_tree\",\"domain\":null,\"rung\":null,\"function\":null,\"artifact_identity\":null,\"investment_row\":null}" as *u8)
244 i=i+1
245 }
246 to_text("]}\n" as *u8); if to_write_failed != 0 { return TO_STATUS_OUTPUT }; return st[0]
247}
248func to_refuse(status: i64, reason: *u8) -> i64 {
249 to_text("{\"schema\":\"nishi.tool-observations\",\"v\":1,\"status\":\"" as *u8); to_text(to_status(status))
250 to_text("\",\"error\":\"" as *u8); to_text(reason); to_text("\"}\n" as *u8)
251 if to_write_failed != 0 { return TO_STATUS_OUTPUT }; return status
252}
253func to_observations(argc: i64, argv: *i64) -> i64 {
254 to_write_failed=0
255 if argc != 5 { return to_refuse(TO_STATUS_INPUT,"usage: observations offset read_capacity row_capacity" as *u8) }
256 let input: *i64=sys_mmap(3*TO_WORD_BYTES) as *i64
257 if (input as i64) <= 0 { return to_refuse(TO_STATUS_UNKNOWN,"allocation_failed" as *u8) }
258 var k: i64=0; var valid: i64=1
259 while k < 3 { let s: *u8=argv[k+2] as *u8; if to_int(s,0,to_len(s),((input as i64)+k*TO_WORD_BYTES) as *i64) != 1 { valid=0 }; k=k+1 }
260 let offset: i64=input[0]; let readcap: i64=input[1]; let rowcap: i64=input[2]; sys_munmap(input as *u8,3*TO_WORD_BYTES)
261 if valid == 0 || offset < 0 || readcap <= 0 || readcap >= TO_MAX/2 || rowcap < 0 || rowcap > TO_MAX/(TO_ROW_SLOTS*TO_WORD_BYTES) {
262 return to_refuse(TO_STATUS_INPUT,"invalid_or_overflowing_capacity_or_offset" as *u8)
263 }
264 let b: *u8=sys_mmap(readcap); let scratch: *i64=sys_mmap(TO_ROW_SLOTS*TO_WORD_BYTES) as *i64
265 let st: *i64=sys_mmap(TO_STATE_SLOTS*TO_WORD_BYTES) as *i64; let numeric: *u8=sys_mmap(TO_NUM_COUNT*TO_WORD_BYTES)
266 var rows: *i64=0 as *i64; if rowcap > 0 { rows=sys_mmap(rowcap*TO_ROW_SLOTS*TO_WORD_BYTES) as *i64 }
267 if (b as i64) <= 0 || (scratch as i64) <= 0 || (st as i64) <= 0 || (numeric as i64) <= 0 || (rowcap > 0 && (rows as i64) <= 0) {
268 if (b as i64) > 0 { sys_munmap(b,readcap) }
269 if (scratch as i64) > 0 { sys_munmap(scratch as *u8,TO_ROW_SLOTS*TO_WORD_BYTES) }
270 if (st as i64) > 0 { sys_munmap(st as *u8,TO_STATE_SLOTS*TO_WORD_BYTES) }
271 if (numeric as i64) > 0 { sys_munmap(numeric,TO_NUM_COUNT*TO_WORD_BYTES) }
272 if (rows as i64) > 0 { sys_munmap(rows as *u8,rowcap*TO_ROW_SLOTS*TO_WORD_BYTES) }
273 return to_refuse(TO_STATUS_UNKNOWN,"allocation_failed" as *u8)
274 }
275 to_read("knowledge/status/actlog.jrnl" as *u8,offset,readcap,b,rows,rowcap,scratch,st)
276 let rc: i64=to_emit(b,rows,st,offset,readcap,rowcap,numeric)
277 sys_munmap(b,readcap); sys_munmap(scratch as *u8,TO_ROW_SLOTS*TO_WORD_BYTES); sys_munmap(st as *u8,TO_STATE_SLOTS*TO_WORD_BYTES); sys_munmap(numeric,TO_NUM_COUNT*TO_WORD_BYTES)
278 if (rows as i64) > 0 { sys_munmap(rows as *u8,rowcap*TO_ROW_SLOTS*TO_WORD_BYTES) }; return rc
279}