nx_parexec_gate.nx source
↩ module page · 77 lines · 3414 B
1// nx_parexec_gate.nx -- proves the PARALLEL executor is CORRECT: forking one worker per partition over
2// shared memory yields byte-identical results to the sequential run, for several worker counts, including
3// a worker count that does not divide the row count. Timing is nondeterministic and lives in the `bench`
4// verb; this gate proves only correctness -- that real forked children over shared memory do not lose,
5// double, or corrupt a single row. D001 verdict via nx_gate_verdict. expect_exit: 0
6import "nx_gate_verdict.nx"
7import "nx_parexec.nx"
8
9func pg_eq(a: i64, b: i64) -> i64 { if a == b { return 1 } return 0 }
10func pg_frame1(a: *i64, nm: *u8, n: i64) -> *u8 {
11 let cols: *i64 = sys_mmap(8) as *i64
12 let names: *i64 = sys_mmap(8) as *i64
13 cols[0] = a as i64
14 names[0] = nm as i64
15 let tot: i64 = cf_encoded_bytes(names, 1, n)
16 let f: *u8 = sys_mmap(tot + 64)
17 cf_encode(cols, names, 1, n, f)
18 return f
19}
20
21func main() -> i64 {
22 let ctr: *i64 = gv_ctr()
23 gv_head("nx_parexec_gate -- do REAL forked workers over shared memory agree with the sequential run?" as *u8)
24
25 let N: i64 = 100000
26 let v: *i64 = sys_mmap(8 * N) as *i64
27 var s: i64 = 777
28 var i: i64 = 0
29 while i < N { s = (s * 1103515245 + 12345) & 0x7fffffff; v[i] = (s % 200000) - 100000; i = i + 1 }
30 let f: *u8 = pg_frame1(v, "v" as *u8, N)
31
32 // sequential reference
33 let seq: *i64 = sys_mmap(8 * 4) as *i64
34 de_run_agg(f, 0, 1, seq)
35
36 // parallel with 2 workers
37 let p2: *i64 = sys_mmap(8 * 4) as *i64
38 pe_agg_parallel(f, 0, 2, p2)
39 var t1: i64 = 1
40 if p2[0] != seq[0] { t1 = 0 }
41 if p2[1] != seq[1] { t1 = 0 }
42 if p2[2] != seq[2] { t1 = 0 }
43 if p2[3] != seq[3] { t1 = 0 }
44 gv_check("T1 2 forked workers == sequential (sum/count/min/max) -- no row lost across processes" as *u8, t1, ctr)
45
46 // parallel with 4 workers
47 let p4: *i64 = sys_mmap(8 * 4) as *i64
48 pe_agg_parallel(f, 0, 4, p4)
49 var t2: i64 = 1
50 if p4[0] != seq[0] { t2 = 0 }
51 if p4[1] != seq[1] { t2 = 0 }
52 if p4[2] != seq[2] { t2 = 0 }
53 if p4[3] != seq[3] { t2 = 0 }
54 gv_check("T2 4 forked workers == sequential (uses all cores, shared-memory merge is exact)" as *u8, t2, ctr)
55
56 // parallel with 7 workers (does NOT divide 100000 -> ragged partitions across processes)
57 let p7: *i64 = sys_mmap(8 * 4) as *i64
58 pe_agg_parallel(f, 0, 7, p7)
59 var t3: i64 = 1
60 if p7[0] != seq[0] { t3 = 0 }
61 if p7[1] != seq[1] { t3 = 0 }
62 if p7[2] != seq[2] { t3 = 0 }
63 if p7[3] != seq[3] { t3 = 0 }
64 gv_check("T3 7 workers (ragged) == sequential -- ragged partitions parallelize correctly" as *u8, t3, ctr)
65
66 // the count across all workers is EXACTLY N (no double-count, no dropped rows under real concurrency)
67 gv_check("T4 parallel count == N exactly (every row processed once, by exactly one worker)" as *u8, pg_eq(p4[1], N), ctr)
68 // and the parallel sum equals an independent reference total
69 var ref: i64 = 0
70 i = 0
71 while i < N { ref = ref + v[i]; i = i + 1 }
72 gv_check("T5 parallel sum == the independent reference total (shared-memory partials merged correctly)" as *u8, pg_eq(p4[0], ref), ctr)
73
74 let rc: i64 = gv_verdict("PAREXEC-GATE" as *u8, ctr, "real multi-core execution via fork+shared-mmap+wait4 is byte-identical to sequential -- parallel, correct, from the first byte up" as *u8)
75 sys_exit(rc)
76 return rc
77}