nx_compute_graph.nx source
↩ module page · 293 lines · 10148 B
1// nx_compute_graph.nx -- typed DAG container for compute nodes.
2//
3// The ComfyUI-power-without-annoyance core. ComfyUI's graph is a
4// JSON dict that gets executed by a Python orchestrator; nodes are
5// untyped, edges are stringly-named, cycle detection is best-effort,
6// caching is opaque.
7//
8// Ours:
9// * **typed DAG** -- nodes carry NodeKind + OpCode/KernelKind sealed
10// enums; edges are (src_node, src_port) -> (dst_node, dst_port)
11// index pairs with bounds-checked validity.
12// * **content-addressed** -- every node has a SHA-256 hash from
13// nx_compute_node.nx; two graphs with the same shape have the
14// same root hash, computed at sort time.
15// * **structurally valid** -- nx_cg_validate refuses cycles + dangling
16// inputs + unwired output dependencies. Bad graphs cannot run.
17// * **topo-sorted** -- nx_cg_toposort produces deterministic order
18// via Kahn's algorithm; cycle = REFUSED with sealed verdict.
19//
20// Runner lives in a separate module once kernels exist. This file is
21// the data structure + invariants + topo sort.
22//
23// genealogy_id: comfyui_workflow_dag + onnx_graph + mlir_module +
24// kahn_1962_toposort
25// lineage_id: substrate_compute_graph_v1
26
27// nx_safety_envelope:
28// intended_use: AUTO_APPLIED -- primitive-specific tuning queued
29// sil_target: SIL1
30// evidence: [bulk_applied_2026-05-16, see-file-comment-for-detail]
31// verdict: NOT_YET_EVALUATED
32
33import "nx_syscalls.nx"
34import "nx_tier.nx"
35import "nx_compute_node.nx"
36
37// ===== Sealed-enum: GraphVerdict ==================================
38//
39// Closed taxonomy of validate/toposort outcomes. Caller switches.
40
41const NX_CG_OK: nx_int = 0
42const NX_CG_ERR_CYCLE: nx_int = 1 // cycle detected
43const NX_CG_ERR_BAD_EDGE: nx_int = 2 // edge references non-existent node
44const NX_CG_ERR_BAD_PORT: nx_int = 3 // port index out of range
45const NX_CG_ERR_DANGLING_INPUT: nx_int = 4 // node input slot not wired
46const NX_CG_ERR_NODE_LIMIT: nx_int = 5 // node cap exceeded
47const NX_CG_ERR_UNKNOWN_KIND: nx_int = 6 // node kind sealed-enum invalid
48const NX_CG_ERR_UNKNOWN_OP: nx_int = 7 // op_code sealed-enum invalid
49const NX_CG_N_VERDICTS: nx_int = 8
50
51func nx_cg_verdict_is_valid(v: nx_int) -> nx_int {
52 if v < 0 { return 0 }
53 if v >= NX_CG_N_VERDICTS { return 0 }
54 return 1
55}
56
57// ===== Graph struct ===============================================
58
59struct ComputeGraph {
60 n_nodes: nx_int,
61 cap_nodes: nx_int,
62 nodes: *i64, // [cap_nodes] of *ComputeNode addresses (as i64)
63 topo_order: *i64, // [cap_nodes] node-id sequence after toposort
64 has_topo: nx_int, // 1 if topo_order is fresh, 0 if stale/uncomputed
65 last_verdict: nx_int // last validate/toposort verdict
66}
67
68const NX_CG_BYTES: nx_int = 48 // 6 fields * 8
69
70// ===== Builder ====================================================
71
72func nx_cg_alloc(cap_nodes: nx_int) -> *ComputeGraph {
73 let g: *ComputeGraph = (sys_mmap(NX_CG_BYTES)) as *ComputeGraph
74 g.n_nodes = 0
75 g.cap_nodes = cap_nodes
76 g.nodes = (sys_mmap(cap_nodes * NX_SIZEOF_NX_INT)) as *i64
77 g.topo_order = (sys_mmap(cap_nodes * NX_SIZEOF_NX_INT)) as *i64
78 g.has_topo = 0
79 g.last_verdict = NX_CG_OK
80 var i: nx_int = 0
81 while i < cap_nodes {
82 g.nodes[i] = 0
83 g.topo_order[i] = 0 - 1
84 i = i + 1
85 }
86 return g
87}
88
89// Add a node. Caller pre-assigns node_id == current count for
90// simplicity (sequential ids). Returns the assigned id or -1 if
91// capacity exhausted.
92
93func nx_cg_add_node(g: *ComputeGraph, n: *ComputeNode) -> nx_int {
94 if g.n_nodes >= g.cap_nodes {
95 g.last_verdict = NX_CG_ERR_NODE_LIMIT
96 return 0 - 1
97 }
98 let id: nx_int = g.n_nodes
99 n.node_id = id
100 g.nodes[id] = n as nx_int
101 g.n_nodes = g.n_nodes + 1
102 g.has_topo = 0
103 return id
104}
105
106// Read accessor: lookup a node by id.
107func nx_cg_get_node(g: *ComputeGraph, id: nx_int) -> *ComputeNode {
108 if id < 0 { return 0 as *ComputeNode }
109 if id >= g.n_nodes { return 0 as *ComputeNode }
110 return g.nodes[id] as *ComputeNode
111}
112
113// ===== Structural validation ======================================
114//
115// Per-node checks (no graph traversal):
116// - kind is a valid NodeKind
117// - op_code is a valid OpCode or KernelKind for that kind
118// - every input edge references a node that exists
119// - every input edge's src_port is within the source's n_outputs
120// - every input slot < n_inputs is wired (no DANGLING_INPUT)
121//
122// Returns sealed-enum verdict.
123
124func nx_cg_validate(g: *ComputeGraph) -> nx_int {
125 var i: nx_int = 0
126 while i < g.n_nodes {
127 let n: *ComputeNode = nx_cg_get_node(g, i)
128
129 if nx_cn_node_kind_is_valid(n.kind) == 0 {
130 g.last_verdict = NX_CG_ERR_UNKNOWN_KIND
131 return NX_CG_ERR_UNKNOWN_KIND
132 }
133 // op_code check depends on kind
134 if n.kind == NX_CN_NODE_TENSOR_OP {
135 if nx_cn_op_is_valid(n.op_code) == 0 {
136 g.last_verdict = NX_CG_ERR_UNKNOWN_OP
137 return NX_CG_ERR_UNKNOWN_OP
138 }
139 }
140 if n.kind == NX_CN_NODE_KERNEL {
141 if nx_cn_kernel_is_valid(n.op_code) == 0 {
142 g.last_verdict = NX_CG_ERR_UNKNOWN_OP
143 return NX_CG_ERR_UNKNOWN_OP
144 }
145 }
146
147 // Validate input edges
148 var s: nx_int = 0
149 while s < n.n_inputs {
150 let src: nx_int = n.input_src_node[s]
151 let port: nx_int = n.input_src_port[s]
152 if src < 0 {
153 g.last_verdict = NX_CG_ERR_DANGLING_INPUT
154 return NX_CG_ERR_DANGLING_INPUT
155 }
156 if src >= g.n_nodes {
157 g.last_verdict = NX_CG_ERR_BAD_EDGE
158 return NX_CG_ERR_BAD_EDGE
159 }
160 let src_n: *ComputeNode = nx_cg_get_node(g, src)
161 if port < 0 {
162 g.last_verdict = NX_CG_ERR_BAD_PORT
163 return NX_CG_ERR_BAD_PORT
164 }
165 if port >= src_n.n_outputs {
166 g.last_verdict = NX_CG_ERR_BAD_PORT
167 return NX_CG_ERR_BAD_PORT
168 }
169 s = s + 1
170 }
171 i = i + 1
172 }
173 g.last_verdict = NX_CG_OK
174 return NX_CG_OK
175}
176
177// ===== Topological sort (Kahn 1962) ==============================
178//
179// Standard in-degree-zero queue algorithm. Refuses cyclic graphs
180// with NX_CG_ERR_CYCLE. Successful run leaves topo_order populated
181// in the same order the runner should execute.
182//
183// Requires nx_cg_validate to have passed. We rely on the caller
184// to have validated; if last_verdict != OK we still attempt sort
185// (Kahn's is robust to it) but the verdict communicates the issue.
186
187func nx_cg_toposort(g: *ComputeGraph) -> nx_int {
188 // Compute in-degree per node
189 let in_deg: *i64 = (sys_mmap(g.cap_nodes * NX_SIZEOF_NX_INT)) as *i64
190 var i: nx_int = 0
191 while i < g.n_nodes {
192 in_deg[i] = 0
193 i = i + 1
194 }
195 var k: nx_int = 0
196 while k < g.n_nodes {
197 let n: *ComputeNode = nx_cg_get_node(g, k)
198 in_deg[k] = n.n_inputs
199 k = k + 1
200 }
201
202 // Queue of nodes with in-degree zero. Use a simple array; we
203 // pop the front via a moving read pointer (cheap when nodes are
204 // sequential).
205 let queue: *i64 = (sys_mmap(g.cap_nodes * NX_SIZEOF_NX_INT)) as *i64
206 var q_head: nx_int = 0
207 var q_tail: nx_int = 0
208
209 var j: nx_int = 0
210 while j < g.n_nodes {
211 if in_deg[j] == 0 {
212 queue[q_tail] = j
213 q_tail = q_tail + 1
214 }
215 j = j + 1
216 }
217
218 var emitted: nx_int = 0
219 while q_head < q_tail {
220 let node_id: nx_int = queue[q_head]
221 q_head = q_head + 1
222 g.topo_order[emitted] = node_id
223 emitted = emitted + 1
224
225 // Decrement in-degree of every node that depends on node_id
226 var nn: nx_int = 0
227 while nn < g.n_nodes {
228 let other: *ComputeNode = nx_cg_get_node(g, nn)
229 var s: nx_int = 0
230 while s < other.n_inputs {
231 if other.input_src_node[s] == node_id {
232 in_deg[nn] = in_deg[nn] - 1
233 if in_deg[nn] == 0 {
234 queue[q_tail] = nn
235 q_tail = q_tail + 1
236 }
237 }
238 s = s + 1
239 }
240 nn = nn + 1
241 }
242 }
243
244 if emitted != g.n_nodes {
245 // Some nodes never reached in-degree zero -> cycle
246 g.has_topo = 0
247 g.last_verdict = NX_CG_ERR_CYCLE
248 return NX_CG_ERR_CYCLE
249 }
250 g.has_topo = 1
251 g.last_verdict = NX_CG_OK
252 return NX_CG_OK
253}
254
255// ===== Topo-order accessor =======================================
256
257func nx_cg_topo_at(g: *ComputeGraph, idx: nx_int) -> nx_int {
258 if g.has_topo == 0 { return 0 - 1 }
259 if idx < 0 { return 0 - 1 }
260 if idx >= g.n_nodes { return 0 - 1 }
261 return g.topo_order[idx]
262}
263
264// ===== Graph root hash =============================================
265//
266// Per Merkle DAG: the graph's hash is a function of every node's
267// hash in toposort order. Two graphs with same nodes in same shape
268// produce identical root hashes -> deterministic caching across
269// runs.
270//
271// Caller must have run nx_cg_toposort first. Caller supplies a
272// 32-byte out buffer.
273
274func nx_cg_root_hash(g: *ComputeGraph, out_hash: *u8) -> nx_int {
275 if g.has_topo == 0 { return 0 - 1 }
276 // Concatenate all node hashes in topo order, then SHA-256 that.
277 let total: nx_int = g.n_nodes * NX_CN_HASH_BYTES
278 let buf: *u8 = sys_mmap(total)
279 var i: nx_int = 0
280 while i < g.n_nodes {
281 let node_id: nx_int = g.topo_order[i]
282 let n: *ComputeNode = nx_cg_get_node(g, node_id)
283 // Each node must have its hash precomputed by nx_cn_compute_hash.
284 var b: nx_int = 0
285 while b < NX_CN_HASH_BYTES {
286 buf[i * NX_CN_HASH_BYTES + b] = n.content_hash[b]
287 b = b + 1
288 }
289 i = i + 1
290 }
291 sha256_digest(buf, total, out_hash)
292 return 0
293}