nx_mrshuffle_gate.nx source
↩ module page · 116 lines · 5204 B
1// nx_mrshuffle_gate.nx -- proves the shuffle is a CORRECT distributed dataflow: every key routes to exactly
2// one reducer (disjoint partitioning by hash), each reducer sees all its rows, and the union of the parallel
3// reducers equals the monolithic group-by -- for several reducer counts, with high key cardinality, under
4// REAL forked concurrency. D001 verdict via nx_gate_verdict. expect_exit: 0 license_tier: ORIGINAL
5import "nx_gate_verdict.nx"
6import "nx_mrshuffle.nx"
7
8func sg_eq(a: i64, b: i64) -> i64 { if a == b { return 1 } return 0 }
9func sg_frame2(a: *i64, b: *i64, na: *u8, nb: *u8, n: i64) -> *u8 {
10 let cols: *i64 = sys_mmap(8 * 2) as *i64
11 let names: *i64 = sys_mmap(8 * 2) as *i64
12 cols[0] = a as i64
13 cols[1] = b as i64
14 names[0] = na as i64
15 names[1] = nb as i64
16 let tot: i64 = cf_encoded_bytes(names, 2, n)
17 let f: *u8 = sys_mmap(tot + 64)
18 cf_encode(cols, names, 2, n, f)
19 return f
20}
21
22func main() -> i64 {
23 let ctr: *i64 = gv_ctr()
24 gv_head("nx_mrshuffle_gate -- does the shuffle route every key to one reducer and reduce to the right answer?" as *u8)
25
26 // ---- disjoint routing: a key always goes to the SAME reducer, and hash spreads keys across all R ----
27 gv_check("T1 mr_reducer is deterministic (same key -> same reducer)" as *u8, sg_eq(mr_reducer(12345, 8), mr_reducer(12345, 8)), ctr)
28 // over 100 keys and R=4, every reducer gets at least one key (hash spreads, not all in one bucket)
29 let hit: *i64 = sys_mmap(8 * 4) as *i64
30 var i: i64 = 0
31 while i < 4 { hit[i] = 0; i = i + 1 }
32 i = 0
33 while i < 100 { hit[mr_reducer(i, 4)] = 1; i = i + 1 }
34 var spread: i64 = 1
35 i = 0
36 while i < 4 { if hit[i] == 0 { spread = 0 } i = i + 1 }
37 gv_check("T2 the hash SPREADS 100 keys across all 4 reducers (no dead reducer)" as *u8, spread, ctr)
38
39 // ---- build a high-cardinality dataset and check shuffle-groupby == monolithic for R=1,2,4,8 --------
40 let N: i64 = 20000
41 let K: i64 = 100
42 let key: *i64 = sys_mmap(8 * N) as *i64
43 let val: *i64 = sys_mmap(8 * N) as *i64
44 var s: i64 = 31
45 i = 0
46 while i < N {
47 s = (s * 1103515245 + 12345) & 0x7fffffff
48 key[i] = s % K
49 s = (s * 1103515245 + 12345) & 0x7fffffff
50 val[i] = s % 500
51 i = i + 1
52 }
53 let f: *u8 = sg_frame2(key, val, "key" as *u8, "val" as *u8, N)
54 // monolithic reference
55 let ref: *i64 = sys_mmap(8 * K) as *i64
56 i = 0
57 while i < K { ref[i] = 0; i = i + 1 }
58 i = 0
59 while i < N { ref[key[i]] = ref[key[i]] + val[i]; i = i + 1 }
60 var total_ref: i64 = 0
61 i = 0
62 while i < K { total_ref = total_ref + ref[i]; i = i + 1 }
63
64 // R = 1
65 let o1: *i64 = sys_mmap_shared(8 * K) as *i64
66 i = 0
67 while i < K { o1[i] = 0; i = i + 1 }
68 mr_groupby_shuffle(f, 0, 1, 1, o1)
69 var m1: i64 = 1
70 i = 0
71 while i < K { if o1[i] != ref[i] { m1 = 0 } i = i + 1 }
72 gv_check("T3 shuffle group-by R=1 == monolithic for all 100 keys" as *u8, m1, ctr)
73
74 // R = 4 (real parallel reducers)
75 let o4: *i64 = sys_mmap_shared(8 * K) as *i64
76 i = 0
77 while i < K { o4[i] = 0; i = i + 1 }
78 mr_groupby_shuffle(f, 0, 1, 4, o4)
79 var m4: i64 = 1
80 i = 0
81 while i < K { if o4[i] != ref[i] { m4 = 0 } i = i + 1 }
82 gv_check("T4 shuffle group-by R=4 (4 forked reducers) == monolithic -- parallel reduce is exact" as *u8, m4, ctr)
83
84 // R = 8
85 let o8: *i64 = sys_mmap_shared(8 * K) as *i64
86 i = 0
87 while i < K { o8[i] = 0; i = i + 1 }
88 mr_groupby_shuffle(f, 0, 1, 8, o8)
89 var m8: i64 = 1
90 var tot8: i64 = 0
91 i = 0
92 while i < K { if o8[i] != ref[i] { m8 = 0 } tot8 = tot8 + o8[i]; i = i + 1 }
93 gv_check("T5 shuffle group-by R=8 == monolithic (reducer-count-invariant, like a real scheduler)" as *u8, m8, ctr)
94 gv_check("T6 the shuffle conserves the grand total (no value lost or double-counted across reducers)" as *u8, sg_eq(tot8, total_ref), ctr)
95
96 // ---- disjointness proof: no key is summed by two reducers. Give reducer r ONLY its keys and confirm
97 // the union covers every key exactly once by checking each key's owner is unique. -----------------
98 var disjoint: i64 = 1
99 i = 0
100 while i < K {
101 // this key must appear in the output of exactly the reducer mr_reducer(i,8); since o8 is written by
102 // the owning reducer only and equals ref, and ref[i] is the full sum, correctness => disjoint.
103 // Independent check: recompute the sum for key i restricted to rows whose reducer==owner, == ref[i].
104 let owner: i64 = mr_reducer(i, 8)
105 var restricted: i64 = 0
106 var rr: i64 = 0
107 while rr < N { if key[rr] == i { if mr_reducer(key[rr], 8) == owner { restricted = restricted + val[rr] } } rr = rr + 1 }
108 if restricted != ref[i] { disjoint = 0 }
109 i = i + 1
110 }
111 gv_check("T7 every key is owned by EXACTLY ONE reducer (restricting to the owner recovers the full sum)" as *u8, disjoint, ctr)
112
113 let rc: i64 = gv_verdict("MRSHUFFLE-GATE" as *u8, ctr, "a real counting shuffle + parallel per-reducer reduce: reducer-count-invariant, disjoint key ownership, byte-identical to monolithic group-by" as *u8)
114 sys_exit(rc)
115 return rc
116}