code wiki / _hdl_build / nx_ale_parallel.nx
nx_ale_parallel.nx source
↩ module page · 203 lines · 9351 B
1// nx_ale_parallel.nx -- ALE PRACTICE: straightforward PARALLEL task consumption. The single-task
2// consumer is the harness (nx_ale_harness <task> <sb> <out> <score>); THIS consumes a whole
3// manifest CONCURRENTLY -- it forks one harness per task ALL AT ONCE (phase 1), then waits for
4// every child (phase 2), then reads + classifies the scores (phase 3). So N tasks are attempted in
5// parallel (wall-clock ~ the slowest single task, not the sum), which is how the team PRACTICES at
6// scale + how the LLM integration throws many tasks at the agent and collects scores. Parallel by
7// construction; correctness == the sequential suite (proven in the self-gate). Reuses the harness
8// (DRY). Each task gets its own sandbox/outdir/scorefile so concurrent attempts never collide.
9// args mode: nx_ale_parallel <manifest> -> consume in parallel; print per-task + elapsed_ms.
10// no-arg: SELF-GATE -- parallel-consume the seed suite (all SOLVED == sequential) + a known-GAP
11// task (classified GAP) -> PARALLELGATE verdict=GREEN to knowledge/status/ale_parallel.log.
12// Landmines: nested ifs (no &&/||), flat exprs, <=6 args/func, no empty-string literal. license_tier: ORIGINAL
13//
14// module: nishi-core.ale.parallel
15// depends: nishi-core.sys.syscalls
16// capability: ALE_PARALLEL_CONSUME
17import "nx_syscalls.nx"
18const PAR_MAGIC_65536: i64 = 65536
19const PAR_MAGIC_2048: i64 = 2048
20const PAR_MAGIC_2047: i64 = 2047
21
22const PAR_MAX: i64 = 256
23
24func pr_p(s: *u8) -> i64 { var n: i64 = 0; while s[n] != (0 as u8) { n = n + 1 } sys_write(1, s, n); return 0 }
25func pr_fp(fd: i64, s: *u8) -> i64 { var n: i64 = 0; while s[n] != (0 as u8) { n = n + 1 } sys_write(fd, s, n); return 0 }
26func pr_fn(fd: i64, v: i64) -> i64 { let bb: *u8 = sys_mmap(28); var m: i64 = v; if m < 0 { m = 0 - m; sys_write(fd, "-" as *u8, 1) }; let t: *u8 = sys_mmap(28); var k: i64 = 0; if m == 0 { t[0] = 48; k = 1 }; while m > 0 { t[k] = (48 + (m % 10)) as u8; m = m / 10; k = k + 1 }; var i: i64 = 0; while i < k { bb[i] = t[k - 1 - i]; i = i + 1 }; sys_write(fd, bb, k); return 0 }
27func pr_len(s: *u8) -> i64 { var n: i64 = 0; while s[n] != (0 as u8) { n = n + 1 } return n }
28func pr_unlink(path: *u8) -> i64 { return __syscall(263, AT_FDCWD, path, 0, 0, 0, 0) }
29
30func pr_read(path: *u8, buf: *u8, cap: i64) -> i64 {
31 let fd: i64 = sys_openat_rd(path)
32 if fd < 0 { return 0 }
33 var n: i64 = 0
34 var go: i64 = 1
35 while go == 1 {
36 let r: i64 = sys_read(fd, (buf as i64 + n) as *u8, cap - 1 - n)
37 if r <= 0 { go = 0 } else { n = n + r }
38 if n >= cap - 1 { go = 0 }
39 }
40 sys_close(fd)
41 return n
42}
43
44// milli-score from a fresh score file; -1 if absent/empty (a gap: the harness never wrote it).
45func pr_score_of(path: *u8) -> i64 {
46 let sb: *u8 = sys_mmap(64)
47 let n: i64 = pr_read(path, sb, 64)
48 if n <= 0 { return 0 - 1 }
49 var v: i64 = 0
50 var i: i64 = 0
51 while i < n { if sb[i] >= (48 as u8) { if sb[i] <= (57 as u8) { v = v * 10 + (sb[i] - 48) } } i = i + 1 }
52 return v
53}
54
55func pr_write_file(path: *u8, s: *u8) -> i64 {
56 pr_unlink(path)
57 let fd: i64 = sys_openat_wr(path, 0x1a4)
58 if fd < 0 { return 0 }
59 sys_write(fd, s, pr_len(s))
60 sys_close(fd)
61 return 1
62}
63
64func pr_idx_path(dst: *u8, prefix: *u8, idx: i64) -> i64 {
65 var o: i64 = 0
66 var i: i64 = 0
67 while prefix[i] != (0 as u8) { dst[o] = prefix[i]; o = o + 1; i = i + 1 }
68 var m: i64 = idx
69 let t: *u8 = sys_mmap(28)
70 var k: i64 = 0
71 if m == 0 { t[0] = 48 as u8; k = 1 }
72 while m > 0 { t[k] = (48 + (m % 10)) as u8; m = m / 10; k = k + 1 }
73 var p: i64 = 0
74 while p < k { dst[o] = t[k - 1 - p]; o = o + 1; p = p + 1 }
75 dst[o] = 0 as u8
76 return o
77}
78
79// fork+exec _offc/nx_ale_harness.elf <task> <sb> <out> <score>; return child PID (NO wait -> parallel).
80func pr_fork(task: *u8, sb: *u8, out: *u8, score: *u8) -> i64 {
81 let pid: i64 = sys_fork()
82 if pid == 0 {
83 let dn: i64 = sys_openat_wr("/dev/null" as *u8, 0x1a4)
84 if dn >= 0 { sys_dup3(dn, 1, 0); sys_dup3(dn, 2, 0) }
85 let argv: *i64 = sys_mmap(48) as *i64
86 argv[0] = "_offc/nx_ale_harness.elf" as *u8 as i64
87 argv[1] = task as i64
88 argv[2] = sb as i64
89 argv[3] = out as i64
90 argv[4] = score as i64
91 argv[5] = 0
92 let envp: *i64 = sys_mmap(16) as *i64
93 envp[0] = 0
94 sys_execve("_offc/nx_ale_harness.elf" as *u8, argv, envp)
95 sys_exit(127)
96 }
97 return pid
98}
99
100// PARALLEL consume: fork ALL harnesses, then wait ALL, then read scores. res[0]=n res[1]=solved res[2]=gaps.
101func pr_consume(manifestpath: *u8, res: *i64, verbose: i64) -> i64 {
102 let buf: *u8 = sys_mmap(PAR_MAGIC_65536)
103 let n: i64 = pr_read(manifestpath, buf, PAR_MAGIC_65536)
104 let tasks: *i64 = sys_mmap(8 * PAR_MAX) as *i64
105 var cnt: i64 = 0
106 var ls: i64 = 0
107 var i: i64 = 0
108 while i <= n {
109 var eol: i64 = 0
110 if i == n { eol = 1 } else { if buf[i] == (10 as u8) { eol = 1 } }
111 if eol == 1 {
112 if i > ls { if buf[ls] != (35 as u8) {
113 if cnt < PAR_MAX {
114 let tp: *u8 = sys_mmap(PAR_MAGIC_2048)
115 var k: i64 = 0
116 var p: i64 = ls
117 while p < i { if k < PAR_MAGIC_2047 { tp[k] = buf[p]; k = k + 1 } p = p + 1 }
118 tp[k] = 0 as u8
119 if k > 0 { tasks[cnt] = tp as i64; cnt = cnt + 1 }
120 }
121 } }
122 ls = i + 1
123 }
124 i = i + 1
125 }
126 // PHASE 1: fork all (concurrent)
127 let pids: *i64 = sys_mmap(8 * PAR_MAX) as *i64
128 var j: i64 = 0
129 while j < cnt {
130 let sb: *u8 = sys_mmap(256); pr_idx_path(sb, "/tmp/_par_sb_" as *u8, j)
131 let out: *u8 = sys_mmap(256); pr_idx_path(out, "/tmp/_par_out_" as *u8, j)
132 let sc: *u8 = sys_mmap(256); pr_idx_path(sc, "/tmp/_par_sc_" as *u8, j)
133 sys_mkdir(sb, 0x1ed); sys_mkdir(out, 0x1ed)
134 pr_unlink(sc)
135 pids[j] = pr_fork(tasks[j] as *u8, sb, out, sc)
136 j = j + 1
137 }
138 // PHASE 2: wait all (the N attempts ran in parallel)
139 j = 0
140 while j < cnt {
141 let st: *i64 = sys_mmap(16) as *i64
142 sys_wait4(pids[j], st, 0)
143 j = j + 1
144 }
145 // PHASE 3: read + classify
146 var solved: i64 = 0
147 var gaps: i64 = 0
148 j = 0
149 while j < cnt {
150 let sc: *u8 = sys_mmap(256); pr_idx_path(sc, "/tmp/_par_sc_" as *u8, j)
151 let score: i64 = pr_score_of(sc)
152 var sv: i64 = 0
153 if score == 1000 { sv = 1 }
154 if sv == 1 { solved = solved + 1 } else { gaps = gaps + 1 }
155 if verbose == 1 { pr_p(" task=" as *u8); pr_p(tasks[j] as *u8); pr_p(" score=" as *u8); pr_fn(1, score); if sv == 1 { pr_p(" SOLVED\n" as *u8) } else { pr_p(" GAP\n" as *u8) } }
156 j = j + 1
157 }
158 res[0] = cnt; res[1] = solved; res[2] = gaps
159 return cnt
160}
161
162func main(argc: i64, argv: *i64) -> i64 {
163 if argc >= 2 {
164 let res: *i64 = sys_mmap(64) as *i64
165 let t0: i64 = sys_now_ms()
166 pr_p("=== ALE parallel consume ===\n" as *u8)
167 pr_consume(argv[1] as *u8, res, 1)
168 let el: i64 = sys_now_ms() - t0
169 pr_p("PARALLEL n_tasks=" as *u8); pr_fn(1, res[0]); pr_p(" solved=" as *u8); pr_fn(1, res[1]); pr_p(" gaps=" as *u8); pr_fn(1, res[2]); pr_p(" elapsed_ms=" as *u8); pr_fn(1, el); pr_p(" mode=RUN\n" as *u8)
170 sys_exit(0); return 0
171 }
172
173 pr_p("=== ALE-parallel gate (PRACTICE: consume tasks in parallel, correctness == sequential) ===\n" as *u8)
174 // CORRECTNESS: parallel-consume the seed suite -> all SOLVED (same as the sequential suite).
175 let res: *i64 = sys_mmap(64) as *i64
176 pr_consume("knowledge/specs/ale_suite.manifest" as *u8, res, 1)
177 var c_seed: i64 = 0
178 if res[0] >= 4 { if res[1] == res[0] { c_seed = 1 } }
179 // DISCRIMINATION: parallel-consume a known PERMANENT-gap task -> classified GAP (not a false solve).
180 pr_write_file("/tmp/_par_gap.manifest" as *u8, "knowledge/specs/ale_examples/task_gapctl.txt\n" as *u8)
181 let resg: *i64 = sys_mmap(64) as *i64
182 pr_consume("/tmp/_par_gap.manifest" as *u8, resg, 0)
183 var c_gap: i64 = 0
184 if resg[0] == 1 { if resg[2] == 1 { c_gap = 1 } }
185
186 pr_p(" seed: n=" as *u8); pr_fn(1, res[0]); pr_p(" solved=" as *u8); pr_fn(1, res[1])
187 pr_p(" | gapctl: n=" as *u8); pr_fn(1, resg[0]); pr_p(" gaps=" as *u8); pr_fn(1, resg[2])
188 pr_p("\n c_seed_all_solved=" as *u8); pr_fn(1, c_seed); pr_p(" c_gap_classified=" as *u8); pr_fn(1, c_gap); pr_p("\n" as *u8)
189
190 var allok: i64 = 1
191 if c_seed == 0 { allok = 0 }
192 if c_gap == 0 { allok = 0 }
193 let lfd: i64 = sys_openat_append("knowledge/status/ale_parallel.log" as *u8, 0x1a4)
194 if allok == 1 {
195 pr_p("PARALLELGATE verdict=GREEN seed_all_solved=1 gap_classified=1 parallel_by_construction=1 rung=ALE-PRACTICE\n" as *u8)
196 if lfd >= 0 { pr_fp(lfd, "PARALLELGATE verdict=GREEN seed_solved=" as *u8); pr_fn(lfd, res[1]); pr_fp(lfd, " gap_classified=1 epoch=" as *u8); pr_fn(lfd, sys_now_realtime_sec()); pr_fp(lfd, "\n" as *u8); sys_close(lfd) }
197 sys_exit(0); return 0
198 }
199 pr_p("PARALLELGATE verdict=RED\n" as *u8)
200 if lfd >= 0 { pr_fp(lfd, "PARALLELGATE verdict=RED c_seed=" as *u8); pr_fn(lfd, c_seed); pr_fp(lfd, " c_gap=" as *u8); pr_fn(lfd, c_gap); pr_fp(lfd, "\n" as *u8); sys_close(lfd) }
201 sys_exit(1)
202 return 1
203}