nx_shard.nx source
↩ module page · 160 lines · 6183 B
1// nx_shard.nx -- parallelism-ready work-sharding primitive.
2//
3// Splits a QED database into N independent work shards. Each shard
4// processes a contiguous range of entries [start, end). Shards are
5// independent: their auto-verify results commute associatively
6// (aggregate stats are just sums), so they can be dispatched to
7// threads / processes / SLURM jobs without coordination beyond the
8// final aggregate step.
9//
10// Phase S0 (this commit): sequential dispatcher. Interface compatible
11// with future thread-pool dispatcher; substrate's commutativity
12// invariant means swapping the dispatcher doesn't change results.
13//
14// genealogy_id: cilk_blumofe_1996 + map_reduce_dean_ghemawat_2004
15// lineage_id: data_parallelism + commutative_aggregation
16// axioms: NX_AX_ALG_ASSOCIATIVITY (sharded ops compose)
17//
18// nx_safety_envelope:
19// intended_use: "Shard container + replication record -- the
20// partitioning primitive substrate-shipped data
21// services (Data Vault, time-series, telemetry)
22// compose against"
23// sil_target: SIL2 (sharding errors = data corruption
24// or replica divergence)
25// asil_target: QM
26// dal_target: DAL C
27// iec_62304_class: A
28// evidence: [Algebraic_Associativity_axiom_tagged,
29// deterministic_shard_key_function,
30// sealed_replica_status_enum]
31// hazard_register: [bug-tape-shard-key-collision-causing-loss,
32// bug-tape-replica-stale-after-split-brain,
33// bug-tape-key-distribution-skew]
34// residual_risk: "Split-brain handling is upstream consensus
35// protocol responsibility. Substrate shard
36// provides bytes-only; semantics layered."
37// verdict: NOT_YET_EVALUATED
38
39import "nx_syscalls.nx"
40import "nx_axioms.nx"
41import "nx_qed_db.nx"
42import "nx_prover.nx"
43import "nx_auto_verify.nx"
44
45// ===== shard descriptor ================================================
46
47struct Shard {
48 shard_id: i64,
49 start_idx: i64, // inclusive
50 end_idx: i64, // exclusive
51 stats: *VerifyStats,
52}
53
54const NX_SHARD_BYTES: i64 = 32
55
56struct ShardPlan {
57 shards: *Shard,
58 n_shards: i64,
59 total_entries: i64,
60}
61
62func nx_shard_plan_alloc(n_shards: i64) -> *ShardPlan {
63 let raw: *u8 = sys_mmap(24)
64 let p: *ShardPlan = raw as *ShardPlan
65 p.shards = (sys_mmap(n_shards * NX_SHARD_BYTES)) as *Shard
66 p.n_shards = n_shards
67 p.total_entries = 0
68 return p
69}
70
71func nx_shard_at(p: *ShardPlan, i: i64) -> *Shard {
72 return (((p.shards as i64) + i * NX_SHARD_BYTES) as *Shard)
73}
74
75// Build a shard plan from a QED database: split [0, n_entries) into
76// n_shards contiguous ranges of approximately equal size.
77func nx_shard_plan_build(db: *QedDb, n_shards: i64) -> *ShardPlan {
78 let p: *ShardPlan = nx_shard_plan_alloc(n_shards)
79 p.total_entries = db.n_entries
80 let chunk: i64 = (db.n_entries + n_shards - 1) / n_shards
81 var i: i64 = 0
82 while i < n_shards {
83 let s: *Shard = nx_shard_at(p, i)
84 s.shard_id = i
85 s.start_idx = i * chunk
86 s.end_idx = (i + 1) * chunk
87 if s.end_idx > db.n_entries { s.end_idx = db.n_entries }
88 s.stats = nx_verify_stats_alloc()
89 i = i + 1
90 }
91 return p
92}
93
94// Run auto-verify on a single shard. Independent of other shards.
95// Future: dispatch this to a thread or remote worker.
96func nx_shard_run(db: *QedDb, s: *Shard, targets: *i64,
97 impl_table: *i64, n_impls: i64,
98 cycle_budget: i64) -> i64 {
99 var i: i64 = s.start_idx
100 while i < s.end_idx {
101 let e: *QedEntry = nx_qed_entry_at(db, i)
102 let prev: i64 = e.verify_status
103 nx_auto_verify_entry(e, targets[i], impl_table, n_impls, cycle_budget)
104 let new_status: i64 = e.verify_status
105 s.stats.n_total = s.stats.n_total + 1
106 if new_status == NX_QED_VERIFY_LOCAL_PROVED {
107 if prev != NX_QED_VERIFY_LOCAL_PROVED {
108 s.stats.n_local_proved = s.stats.n_local_proved + 1
109 }
110 }
111 if new_status == NX_QED_VERIFY_MATCHES_INDEPENDENT {
112 if prev != NX_QED_VERIFY_MATCHES_INDEPENDENT {
113 s.stats.n_matches = s.stats.n_matches + 1
114 }
115 }
116 if new_status == NX_QED_VERIFY_DIFFERS_INVESTIGATE {
117 s.stats.n_differs = s.stats.n_differs + 1
118 }
119 if new_status == prev {
120 s.stats.n_unchanged = s.stats.n_unchanged + 1
121 }
122 i = i + 1
123 }
124 return 0
125}
126
127// Run all shards sequentially (Phase S0). Interface compatible with
128// future parallel dispatch; results are identical regardless of order.
129func nx_shard_run_all(db: *QedDb, p: *ShardPlan, targets: *i64,
130 impl_table: *i64, n_impls: i64,
131 cycle_budget: i64) -> i64 {
132 var i: i64 = 0
133 while i < p.n_shards {
134 let s: *Shard = nx_shard_at(p, i)
135 nx_shard_run(db, s, targets, impl_table, n_impls, cycle_budget)
136 i = i + 1
137 }
138 return 0
139}
140
141// Aggregate per-shard stats into one combined VerifyStats.
142// Associative + commutative; works for any shard order.
143func nx_shard_aggregate(p: *ShardPlan, out: *VerifyStats) -> i64 {
144 out.n_total = 0
145 out.n_local_proved = 0
146 out.n_matches = 0
147 out.n_unchanged = 0
148 out.n_differs = 0
149 var i: i64 = 0
150 while i < p.n_shards {
151 let s: *Shard = nx_shard_at(p, i)
152 out.n_total = out.n_total + s.stats.n_total
153 out.n_local_proved = out.n_local_proved + s.stats.n_local_proved
154 out.n_matches = out.n_matches + s.stats.n_matches
155 out.n_unchanged = out.n_unchanged + s.stats.n_unchanged
156 out.n_differs = out.n_differs + s.stats.n_differs
157 i = i + 1
158 }
159 return 0
160}