Skip to main content

dfir_lang/graph/
meta_graph.rs

1#![warn(missing_docs)]
2
3extern crate proc_macro;
4
5use std::collections::{BTreeMap, BTreeSet};
6use std::fmt::Debug;
7use std::iter::FusedIterator;
8
9use itertools::Itertools;
10use proc_macro2::{Ident, Literal, Span, TokenStream};
11use quote::{ToTokens, format_ident, quote, quote_spanned};
12use serde::{Deserialize, Serialize};
13use slotmap::{Key, SecondaryMap, SlotMap, SparseSecondaryMap};
14use syn::spanned::Spanned;
15
16use super::graph_write::{Dot, GraphWrite, Mermaid};
17use super::ops::{
18    DelayType, FloType, OPERATORS, OperatorWriteOutput, WriteContextArgs, find_op_op_constraints,
19    null_write_iterator_fn,
20};
21use super::{
22    CONTEXT, Color, DiMulGraph, GRAPH, GraphEdgeId, GraphLoopId, GraphNode, GraphNodeId,
23    GraphSubgraphId, HANDOFF_NODE_STR, HandoffKind, MODULE_BOUNDARY_NODE_STR, OperatorInstance,
24    PortIndexValue, SINGLETON_SLOT_NODE_STR, Varname, change_spans, get_operator_generics,
25};
26use crate::diagnostic::{Diagnostic, Diagnostics, Level};
27use crate::pretty_span::{PrettyRowCol, PrettySpan};
28use crate::process_singletons;
29
30/// A resolved handoff reference: the target node ID plus mutability and access group info.
31#[derive(Clone, Debug, Serialize, Deserialize)]
32pub struct ResolvedHandoffRef {
33    /// The resolved target node ID (`None` if unresolved/error).
34    pub node_id: Option<GraphNodeId>,
35    /// Whether this is a mutable reference (`#mut var`).
36    pub is_mut: bool,
37    /// Optional access group for ordering (`#{N} var`).
38    pub access_group: Option<u32>,
39}
40
41/// An abstract "meta graph" representation of a DFIR graph.
42///
43/// Can be with or without subgraph partitioning, stratification, and handoff insertion. This is
44/// the meta graph used for generating Rust source code in macros from DFIR sytnax.
45///
46/// This struct has a lot of methods for manipulating the graph, vaguely grouped together in
47/// separate `impl` blocks. You might notice a few particularly specific arbitray-seeming methods
48/// in here--those are just what was needed for the compilation algorithms. If you need another
49/// method then add it.
50#[derive(Default, Debug, Serialize, Deserialize)]
51pub struct DfirGraph {
52    /// Each node type (operator or handoff).
53    nodes: SlotMap<GraphNodeId, GraphNode>,
54
55    /// Instance data corresponding to each operator node.
56    /// This field will be empty after deserialization.
57    #[serde(skip)]
58    operator_instances: SecondaryMap<GraphNodeId, OperatorInstance>,
59    /// Debugging/tracing tag for each operator node.
60    operator_tag: SecondaryMap<GraphNodeId, String>,
61    /// Graph data structure (two-way adjacency list).
62    graph: DiMulGraph<GraphNodeId, GraphEdgeId>,
63    /// Input and output port for each edge.
64    ports: SecondaryMap<GraphEdgeId, (PortIndexValue, PortIndexValue)>,
65
66    /// Which loop a node belongs to (or none for top-level).
67    node_loops: SecondaryMap<GraphNodeId, GraphLoopId>,
68    /// Which nodes belong to each loop.
69    loop_nodes: SlotMap<GraphLoopId, Vec<GraphNodeId>>,
70    /// For the loop, what is its parent (`None` for top-level).
71    loop_parent: SparseSecondaryMap<GraphLoopId, GraphLoopId>,
72    /// What loops are at the root.
73    root_loops: Vec<GraphLoopId>,
74    /// For the loop, what are its child loops.
75    loop_children: SecondaryMap<GraphLoopId, Vec<GraphLoopId>>,
76
77    /// Which subgraph each node belongs to.
78    node_subgraph: SecondaryMap<GraphNodeId, GraphSubgraphId>,
79
80    /// Which nodes belong to each subgraph.
81    subgraph_nodes: SlotMap<GraphSubgraphId, Vec<GraphNodeId>>,
82    /// Subgraph IDs in topological sort order (set during partitioning).
83    subgraph_toposort: Vec<GraphSubgraphId>,
84
85    /// Resolved handoff varnames references, per node.
86    node_handoff_references: SparseSecondaryMap<GraphNodeId, Vec<ResolvedHandoffRef>>,
87    /// What variable name each graph node belongs to (if any). For debugging (graph writing) purposes only.
88    node_varnames: SparseSecondaryMap<GraphNodeId, Varname>,
89
90    /// Delay type for handoff nodes that represent tick-boundary back-edges.
91    /// Set by `order_subgraphs` for `defer_tick` / `defer_tick_lazy`, either on handoff nodes
92    /// it injects or on existing handoff nodes that it marks as tick-boundary back-edges.
93    handoff_delay_type: SparseSecondaryMap<GraphNodeId, DelayType>,
94}
95
96/// Basic methods.
97impl DfirGraph {
98    /// Create a new empty graph.
99    pub fn new() -> Self {
100        Default::default()
101    }
102}
103
104/// Node methods.
105impl DfirGraph {
106    /// Get a node with its operator instance (if applicable).
107    pub fn node(&self, node_id: GraphNodeId) -> &GraphNode {
108        self.nodes.get(node_id).expect("Node not found.")
109    }
110
111    /// Get the `OperatorInstance` for a given node. Node must be an operator and have an
112    /// `OperatorInstance` present, otherwise will return `None`.
113    ///
114    /// Note that no operator instances will be persent after deserialization.
115    pub fn node_op_inst(&self, node_id: GraphNodeId) -> Option<&OperatorInstance> {
116        self.operator_instances.get(node_id)
117    }
118
119    /// Get the debug variable name attached to a graph node.
120    pub fn node_varname(&self, node_id: GraphNodeId) -> Option<&Varname> {
121        self.node_varnames.get(node_id)
122    }
123
124    /// Get subgraph for node.
125    pub fn node_subgraph(&self, node_id: GraphNodeId) -> Option<GraphSubgraphId> {
126        self.node_subgraph.get(node_id).copied()
127    }
128
129    /// Degree into a node, i.e. the number of predecessors.
130    pub fn node_degree_in(&self, node_id: GraphNodeId) -> usize {
131        self.graph.degree_in(node_id)
132    }
133
134    /// Degree out of a node, i.e. the number of successors.
135    pub fn node_degree_out(&self, node_id: GraphNodeId) -> usize {
136        self.graph.degree_out(node_id)
137    }
138
139    /// Successors, iterator of `(GraphEdgeId, GraphNodeId)` of outgoing edges.
140    pub fn node_successors(
141        &self,
142        src: GraphNodeId,
143    ) -> impl '_
144    + DoubleEndedIterator<Item = (GraphEdgeId, GraphNodeId)>
145    + ExactSizeIterator
146    + FusedIterator
147    + Clone
148    + Debug {
149        self.graph.successors(src)
150    }
151
152    /// Predecessors, iterator of `(GraphEdgeId, GraphNodeId)` of incoming edges.
153    pub fn node_predecessors(
154        &self,
155        dst: GraphNodeId,
156    ) -> impl '_
157    + DoubleEndedIterator<Item = (GraphEdgeId, GraphNodeId)>
158    + ExactSizeIterator
159    + FusedIterator
160    + Clone
161    + Debug {
162        self.graph.predecessors(dst)
163    }
164
165    /// Successor edges, iterator of `GraphEdgeId` of outgoing edges.
166    pub fn node_successor_edges(
167        &self,
168        src: GraphNodeId,
169    ) -> impl '_
170    + DoubleEndedIterator<Item = GraphEdgeId>
171    + ExactSizeIterator
172    + FusedIterator
173    + Clone
174    + Debug {
175        self.graph.successor_edges(src)
176    }
177
178    /// Predecessor edges, iterator of `GraphEdgeId` of incoming edges.
179    pub fn node_predecessor_edges(
180        &self,
181        dst: GraphNodeId,
182    ) -> impl '_
183    + DoubleEndedIterator<Item = GraphEdgeId>
184    + ExactSizeIterator
185    + FusedIterator
186    + Clone
187    + Debug {
188        self.graph.predecessor_edges(dst)
189    }
190
191    /// Successor nodes, iterator of `GraphNodeId`.
192    pub fn node_successor_nodes(
193        &self,
194        src: GraphNodeId,
195    ) -> impl '_
196    + DoubleEndedIterator<Item = GraphNodeId>
197    + ExactSizeIterator
198    + FusedIterator
199    + Clone
200    + Debug {
201        self.graph.successor_vertices(src)
202    }
203
204    /// Predecessor nodes, iterator of `GraphNodeId`.
205    pub fn node_predecessor_nodes(
206        &self,
207        dst: GraphNodeId,
208    ) -> impl '_
209    + DoubleEndedIterator<Item = GraphNodeId>
210    + ExactSizeIterator
211    + FusedIterator
212    + Clone
213    + Debug {
214        self.graph.predecessor_vertices(dst)
215    }
216
217    /// Iterator of node IDs `GraphNodeId`.
218    pub fn node_ids(&self) -> slotmap::basic::Keys<'_, GraphNodeId, GraphNode> {
219        self.nodes.keys()
220    }
221
222    /// Iterator over `(GraphNodeId, &Node)` pairs.
223    pub fn nodes(&self) -> slotmap::basic::Iter<'_, GraphNodeId, GraphNode> {
224        self.nodes.iter()
225    }
226
227    /// Insert a node, assigning the given varname.
228    pub fn insert_node(
229        &mut self,
230        node: GraphNode,
231        varname_opt: Option<Ident>,
232        loop_opt: Option<GraphLoopId>,
233    ) -> GraphNodeId {
234        let node_id = self.nodes.insert(node);
235        if let Some(varname) = varname_opt {
236            self.node_varnames.insert(node_id, Varname(varname));
237        }
238        if let Some(loop_id) = loop_opt {
239            self.node_loops.insert(node_id, loop_id);
240            self.loop_nodes[loop_id].push(node_id);
241        }
242        node_id
243    }
244
245    /// Insert an operator instance for the given node. Panics if already set.
246    pub fn insert_node_op_inst(&mut self, node_id: GraphNodeId, op_inst: OperatorInstance) {
247        assert!(matches!(
248            self.nodes.get(node_id),
249            Some(GraphNode::Operator(_))
250        ));
251        let old_inst = self.operator_instances.insert(node_id, op_inst);
252        assert!(old_inst.is_none());
253    }
254
255    /// Assign all operator instances if not set. Write diagnostic messages/errors into `diagnostics`.
256    pub fn insert_node_op_insts_all(&mut self, diagnostics: &mut Diagnostics) {
257        // Handle all nodes in two phases, since the helper methods take total ownership of `&self`.
258        // Possible to do in one phase, but would require accessing fields directly for partial mutable ownership.
259
260        // Collect operator instances, then assign.
261        let mut op_insts = Vec::new();
262        // Collect nodes that should be lowered to handoffs (the `handoff()`/`singleton()` pseudo-operators).
263        let mut handoff_nodes: Vec<(GraphNodeId, HandoffKind, Span)> = Vec::new();
264
265        for (node_id, node) in self.nodes() {
266            let GraphNode::Operator(operator) = node else {
267                continue;
268            };
269            if self.node_op_inst(node_id).is_some() {
270                continue;
271            };
272
273            // Recognize `handoff()`/`singleton()` pseudo-operators and lower to GraphNode::Handoff.
274            let handoff_kind = match &*operator.name_string() {
275                "handoff" => Some(HandoffKind::Vec),
276                "singleton" => Some(HandoffKind::Singleton),
277                "optional" => Some(HandoffKind::Optional),
278                _ => None,
279            };
280            if let Some(kind) = handoff_kind {
281                if !operator.args.is_empty() {
282                    diagnostics.push(Diagnostic::spanned(
283                        operator.path.span(),
284                        Level::Error,
285                        format!("`{}` takes no arguments.", operator.name_string()),
286                    ));
287                }
288                if operator.type_arguments().is_some() {
289                    diagnostics.push(Diagnostic::spanned(
290                        operator.path.span(),
291                        Level::Error,
292                        format!("`{}` takes no generic arguments.", operator.name_string()),
293                    ));
294                }
295                handoff_nodes.push((node_id, kind, operator.path.span()));
296                continue;
297            }
298
299            // Op constraints.
300            let Some(op_constraints) = find_op_op_constraints(operator) else {
301                diagnostics.push(Diagnostic::spanned(
302                    operator.path.span(),
303                    Level::Error,
304                    format!("Unknown operator `{}`", operator.name_string()),
305                ));
306                continue;
307            };
308
309            // Input and output ports.
310            let (input_ports, output_ports) = {
311                let mut input_edges: Vec<(&PortIndexValue, GraphNodeId)> = self
312                    .node_predecessors(node_id)
313                    .map(|(edge_id, pred_id)| (self.edge_ports(edge_id).1, pred_id))
314                    .collect();
315                // Ensure sorted by port index.
316                input_edges.sort();
317                let input_ports: Vec<PortIndexValue> = input_edges
318                    .into_iter()
319                    .map(|(port, _pred)| port)
320                    .cloned()
321                    .collect();
322
323                // Collect output arguments (successors).
324                let mut output_edges: Vec<(&PortIndexValue, GraphNodeId)> = self
325                    .node_successors(node_id)
326                    .map(|(edge_id, succ)| (self.edge_ports(edge_id).0, succ))
327                    .collect();
328                // Ensure sorted by port index.
329                output_edges.sort();
330                let output_ports: Vec<PortIndexValue> = output_edges
331                    .into_iter()
332                    .map(|(port, _succ)| port)
333                    .cloned()
334                    .collect();
335
336                (input_ports, output_ports)
337            };
338
339            // Generic arguments.
340            let generics = get_operator_generics(diagnostics, operator);
341            // Generic argument errors.
342            {
343                // Span of `generic_args` (if it exists), otherwise span of the operator name.
344                let generics_span = generics
345                    .generic_args
346                    .as_ref()
347                    .map(Spanned::span)
348                    .unwrap_or_else(|| operator.path.span());
349
350                if !op_constraints
351                    .persistence_args
352                    .contains(&generics.persistence_args.len())
353                {
354                    diagnostics.push(Diagnostic::spanned(
355                        generics.persistence_args_span().unwrap_or(generics_span),
356                        Level::Error,
357                        format!(
358                            "`{}` should have {} persistence lifetime arguments, actually has {}.",
359                            op_constraints.name,
360                            op_constraints.persistence_args.human_string(),
361                            generics.persistence_args.len()
362                        ),
363                    ));
364                }
365                if !op_constraints.type_args.contains(&generics.type_args.len()) {
366                    diagnostics.push(Diagnostic::spanned(
367                        generics.type_args_span().unwrap_or(generics_span),
368                        Level::Error,
369                        format!(
370                            "`{}` should have {} generic type arguments, actually has {}.",
371                            op_constraints.name,
372                            op_constraints.type_args.human_string(),
373                            generics.type_args.len()
374                        ),
375                    ));
376                }
377            }
378
379            op_insts.push((
380                node_id,
381                OperatorInstance {
382                    op_constraints,
383                    input_ports,
384                    output_ports,
385                    singletons_referenced: operator.singletons_referenced.clone(),
386                    generics,
387                    arguments_pre: operator.args.clone(),
388                    arguments_raw: operator.args_raw.clone(),
389                },
390            ));
391        }
392
393        for (node_id, op_inst) in op_insts {
394            self.insert_node_op_inst(node_id, op_inst);
395        }
396
397        // Replace pseudo-operator nodes with GraphNode::Handoff.
398        for (node_id, kind, span) in handoff_nodes {
399            self.nodes[node_id] = GraphNode::Handoff {
400                kind,
401                src_span: span,
402                dst_span: span,
403            };
404        }
405    }
406
407    /// Inserts a node between two existing nodes connected by the given `edge_id`.
408    ///
409    /// `edge`: (src, dst, `dst_idx`)
410    ///
411    /// Before: A (src) ------------> B (dst)
412    /// After:  A (src) -> X (new) -> B (dst)
413    ///
414    /// Returns the ID of X & ID of edge OUT of X.
415    ///
416    /// Note that both the edges will be new and `edge_id` will be removed. Both new edges will
417    /// get the edge type of the original edge.
418    pub fn insert_intermediate_node(
419        &mut self,
420        edge_id: GraphEdgeId,
421        new_node: GraphNode,
422    ) -> (GraphNodeId, GraphEdgeId) {
423        let span = Some(new_node.span());
424
425        // Make corresponding operator instance (if `node` is an operator).
426        let op_inst_opt = 'oc: {
427            let GraphNode::Operator(operator) = &new_node else {
428                break 'oc None;
429            };
430            let Some(op_constraints) = find_op_op_constraints(operator) else {
431                break 'oc None;
432            };
433            let (input_port, output_port) = self.ports.get(edge_id).cloned().unwrap();
434
435            let mut dummy_diagnostics = Diagnostics::new();
436            let generics = get_operator_generics(&mut dummy_diagnostics, operator);
437            assert!(dummy_diagnostics.is_empty());
438
439            Some(OperatorInstance {
440                op_constraints,
441                input_ports: vec![input_port],
442                output_ports: vec![output_port],
443                singletons_referenced: operator.singletons_referenced.clone(),
444                generics,
445                arguments_pre: operator.args.clone(),
446                arguments_raw: operator.args_raw.clone(),
447            })
448        };
449
450        // Insert new `node`.
451        let node_id = self.nodes.insert(new_node);
452        // Insert corresponding `OperatorInstance` if applicable.
453        if let Some(op_inst) = op_inst_opt {
454            self.operator_instances.insert(node_id, op_inst);
455        }
456        // Update edges to insert node within `edge_id`.
457        let (e0, e1) = self
458            .graph
459            .insert_intermediate_vertex(node_id, edge_id)
460            .unwrap();
461
462        // Update corresponding ports.
463        let (src_idx, dst_idx) = self.ports.remove(edge_id).unwrap();
464        self.ports
465            .insert(e0, (src_idx, PortIndexValue::Elided(span)));
466        self.ports
467            .insert(e1, (PortIndexValue::Elided(span), dst_idx));
468
469        (node_id, e1)
470    }
471
472    /// Remove the node `node_id` but preserves and connects the single predecessor and single successor.
473    /// Panics if the node does not have exactly one predecessor and one successor, or is not in the graph.
474    pub fn remove_intermediate_node(&mut self, node_id: GraphNodeId) {
475        assert_eq!(
476            1,
477            self.node_degree_in(node_id),
478            "Removed intermediate node must have one predecessor"
479        );
480        assert_eq!(
481            1,
482            self.node_degree_out(node_id),
483            "Removed intermediate node must have one successor"
484        );
485        assert!(
486            self.node_subgraph.is_empty() && self.subgraph_nodes.is_empty(),
487            "Should not remove intermediate node after subgraph partitioning"
488        );
489
490        assert!(self.nodes.remove(node_id).is_some());
491        let (new_edge_id, (pred_edge_id, succ_edge_id)) =
492            self.graph.remove_intermediate_vertex(node_id).unwrap();
493        self.operator_instances.remove(node_id);
494        self.node_varnames.remove(node_id);
495
496        let (src_port, _) = self.ports.remove(pred_edge_id).unwrap();
497        let (_, dst_port) = self.ports.remove(succ_edge_id).unwrap();
498        self.ports.insert(new_edge_id, (src_port, dst_port));
499    }
500
501    /// Helper method: determine the "color" (pull vs push) of a node based on its in and out degree,
502    /// excluding reference edges. If linear (1 in, 1 out), color is `None`, indicating it can be
503    /// either push or pull.
504    ///
505    /// Note that this does NOT consider `DelayType` barriers (which generally implies `Pull`).
506    pub(crate) fn node_color(&self, node_id: GraphNodeId) -> Option<Color> {
507        if matches!(self.node(node_id), GraphNode::Handoff { .. }) {
508            return Some(Color::Hoff);
509        }
510
511        // TODO(shadaj): this is a horrible hack
512        if let GraphNode::Operator(op) = self.node(node_id)
513            && (op.name_string() == "resolve_futures_blocking"
514                || op.name_string() == "resolve_futures_blocking_ordered")
515        {
516            return Some(Color::Push);
517        }
518
519        // In-degree, excluding ref-edges.
520        let inn_degree = self.node_predecessor_nodes(node_id).len();
521        // Out-degree excluding ref-edges.
522        let out_degree = self.node_successor_nodes(node_id).len();
523
524        match (inn_degree, out_degree) {
525            (0, 0) => None, // Generally should not happen, "Degenerate subgraph detected".
526            (0, 1) => Some(Color::Pull),
527            (1, 0) => Some(Color::Push),
528            (1, 1) => None, // Linear, can be either push or pull.
529            (_many, 0 | 1) => Some(Color::Pull),
530            (0 | 1, _many) => Some(Color::Push),
531            (_many, _to_many) => Some(Color::Comp),
532        }
533    }
534
535    /// Set the operator tag (for debugging/tracing).
536    pub fn set_operator_tag(&mut self, node_id: GraphNodeId, tag: String) {
537        self.operator_tag.insert(node_id, tag);
538    }
539}
540
541/// Handoff references.
542impl DfirGraph {
543    /// Set the handoff references for the `node_id` operator. Each reference corresponds to the
544    /// same index in the [`crate::parse::Operator::singletons_referenced`] vec.
545    pub fn set_node_handoff_references(
546        &mut self,
547        node_id: GraphNodeId,
548        singletons_referenced: Vec<ResolvedHandoffRef>,
549    ) -> Option<Vec<ResolvedHandoffRef>> {
550        self.node_handoff_references
551            .insert(node_id, singletons_referenced)
552    }
553
554    /// Gets the handoff references for a node. Returns an empty slice for non-operators and
555    /// operators that do not reference handoffs.
556    pub fn node_handoff_references(&self, node_id: GraphNodeId) -> &[ResolvedHandoffRef] {
557        self.node_handoff_references
558            .get(node_id)
559            .map(std::ops::Deref::deref)
560            .unwrap_or_default()
561    }
562
563    /// Collect all refs, grouped by the handoff they're pointing at, then by the access group idx `Option<u32>`.
564    pub fn node_handoff_reference_groups(&self) -> NodeHandoffReferenceGroups<'_> {
565        let mut handoff_references = NodeHandoffReferenceGroups::new();
566        for node_id in self.node_ids() {
567            if let GraphNode::Operator(operator) = self.node(node_id) {
568                let resolved = self.node_handoff_references(node_id);
569                for (resolved_ref, ref_token) in
570                    resolved.iter().zip(operator.singletons_referenced.iter())
571                {
572                    if let Some(target_nid) = resolved_ref.node_id {
573                        handoff_references
574                            .entry(target_nid)
575                            .or_default()
576                            .entry(resolved_ref.access_group)
577                            .or_default()
578                            .push((node_id, resolved_ref, ref_token.span()));
579                    }
580                }
581            }
582        }
583        handoff_references
584    }
585}
586
587/// Per-node handoff references, in turn grouped by access group.
588/// Map: `handoff_node_id` -> `access_group` -> (source `GraphNodeId`, `ResolvedHandoffRef`, `#ref` span)
589pub type NodeHandoffReferenceGroups<'a> =
590    BTreeMap<GraphNodeId, BTreeMap<Option<u32>, Vec<(GraphNodeId, &'a ResolvedHandoffRef, Span)>>>;
591
592/// Module methods.
593impl DfirGraph {
594    /// When modules are imported into a flat graph, they come with an input and output `ModuleBoundary` node.
595    /// The partitioner doesn't understand these nodes and will panic if it encounters them.
596    /// `merge_modules` removes them from the graph, stitching the input and output sides of the `ModuleBoundary`s based on their ports.
597    /// For example:
598    ///
599    /// ```text
600    /// source_iter([]) -> [myport]ModuleBoundary(input)[my_port] -> map(|x| x) -> ModuleBoundary(output) -> null();
601    /// ```
602    ///
603    /// In the above example, the `[myport]` port will be used to connect the `source_iter` with the `map` that is inside of the module.
604    /// The output module boundary has elided ports, this is also used to match up the input/output across the module boundary.
605    pub fn merge_modules(&mut self) -> Result<(), Diagnostic> {
606        let mod_bound_nodes = self
607            .nodes()
608            .filter(|(_nid, node)| matches!(node, GraphNode::ModuleBoundary { .. }))
609            .map(|(nid, _node)| nid)
610            .collect::<Vec<_>>();
611
612        for mod_bound_node in mod_bound_nodes {
613            self.remove_module_boundary(mod_bound_node)?;
614        }
615
616        Ok(())
617    }
618
619    /// see `merge_modules`
620    /// This function removes a singular module boundary from the graph and performs the necessary stitching to fix the graph afterward.
621    /// `merge_modules` calls this function for each module boundary in the graph.
622    fn remove_module_boundary(&mut self, mod_bound_node: GraphNodeId) -> Result<(), Diagnostic> {
623        assert!(
624            self.node_subgraph.is_empty() && self.subgraph_nodes.is_empty(),
625            "Should not remove intermediate node after subgraph partitioning"
626        );
627
628        let mut mod_pred_ports = BTreeMap::new();
629        let mut mod_succ_ports = BTreeMap::new();
630
631        for mod_out_edge in self.node_predecessor_edges(mod_bound_node) {
632            let (pred_port, succ_port) = self.edge_ports(mod_out_edge);
633            mod_pred_ports.insert(succ_port.clone(), (mod_out_edge, pred_port.clone()));
634        }
635
636        for mod_inn_edge in self.node_successor_edges(mod_bound_node) {
637            let (pred_port, succ_port) = self.edge_ports(mod_inn_edge);
638            mod_succ_ports.insert(pred_port.clone(), (mod_inn_edge, succ_port.clone()));
639        }
640
641        if mod_pred_ports.keys().collect::<BTreeSet<_>>()
642            != mod_succ_ports.keys().collect::<BTreeSet<_>>()
643        {
644            // get module boundary node
645            let GraphNode::ModuleBoundary { input, import_expr } = self.node(mod_bound_node) else {
646                panic!();
647            };
648
649            if *input {
650                return Err(Diagnostic {
651                    span: *import_expr,
652                    level: Level::Error,
653                    message: format!(
654                        "The ports into the module did not match. input: {:?}, expected: {:?}",
655                        mod_pred_ports.keys().map(|x| x.to_string()).join(", "),
656                        mod_succ_ports.keys().map(|x| x.to_string()).join(", ")
657                    ),
658                });
659            }
660            return Err(Diagnostic {
661                span: *import_expr,
662                level: Level::Error,
663                message: format!(
664                    "The ports out of the module did not match. output: {:?}, expected: {:?}",
665                    mod_succ_ports.keys().map(|x| x.to_string()).join(", "),
666                    mod_pred_ports.keys().map(|x| x.to_string()).join(", "),
667                ),
668            });
669        }
670
671        for (port, (pred_edge, pred_port)) in mod_pred_ports {
672            let (succ_edge, succ_port) = mod_succ_ports.remove(&port).unwrap();
673
674            let (src, _) = self.edge(pred_edge);
675            let (_, dst) = self.edge(succ_edge);
676            self.remove_edge(pred_edge);
677            self.remove_edge(succ_edge);
678
679            let new_edge_id = self.graph.insert_edge(src, dst);
680            self.ports.insert(new_edge_id, (pred_port, succ_port));
681        }
682
683        self.graph.remove_vertex(mod_bound_node);
684        self.nodes.remove(mod_bound_node);
685
686        Ok(())
687    }
688}
689
690/// Edge methods.
691impl DfirGraph {
692    /// Get the `src` and `dst` for an edge: `(src GraphNodeId, dst GraphNodeId)`.
693    pub fn edge(&self, edge_id: GraphEdgeId) -> (GraphNodeId, GraphNodeId) {
694        let (src, dst) = self.graph.edge(edge_id).expect("Edge not found.");
695        (src, dst)
696    }
697
698    /// Get the source and destination ports for an edge: `(src &PortIndexValue, dst &PortIndexValue)`.
699    pub fn edge_ports(&self, edge_id: GraphEdgeId) -> (&PortIndexValue, &PortIndexValue) {
700        let (src_port, dst_port) = self.ports.get(edge_id).expect("Edge not found.");
701        (src_port, dst_port)
702    }
703
704    /// Iterator of all edge IDs `GraphEdgeId`.
705    pub fn edge_ids(&self) -> slotmap::basic::Keys<'_, GraphEdgeId, (GraphNodeId, GraphNodeId)> {
706        self.graph.edge_ids()
707    }
708
709    /// Iterator over all edges: `(GraphEdgeId, (src GraphNodeId, dst GraphNodeId))`.
710    pub fn edges(
711        &self,
712    ) -> impl '_
713    + ExactSizeIterator<Item = (GraphEdgeId, (GraphNodeId, GraphNodeId))>
714    + FusedIterator
715    + Clone
716    + Debug {
717        self.graph.edges()
718    }
719
720    /// Insert an edge between nodes thru the given ports.
721    pub fn insert_edge(
722        &mut self,
723        src: GraphNodeId,
724        src_port: PortIndexValue,
725        dst: GraphNodeId,
726        dst_port: PortIndexValue,
727    ) -> GraphEdgeId {
728        let edge_id = self.graph.insert_edge(src, dst);
729        self.ports.insert(edge_id, (src_port, dst_port));
730        edge_id
731    }
732
733    /// Removes an edge and its corresponding ports and edge type info.
734    pub fn remove_edge(&mut self, edge: GraphEdgeId) {
735        let (_src, _dst) = self.graph.remove_edge(edge).unwrap();
736        let (_src_port, _dst_port) = self.ports.remove(edge).unwrap();
737    }
738}
739
740/// Subgraph methods.
741impl DfirGraph {
742    /// Nodes belonging to the given subgraph.
743    pub fn subgraph(&self, subgraph_id: GraphSubgraphId) -> &Vec<GraphNodeId> {
744        self.subgraph_nodes
745            .get(subgraph_id)
746            .expect("Subgraph not found.")
747    }
748
749    /// Iterator over all subgraph IDs.
750    pub fn subgraph_ids(&self) -> slotmap::basic::Keys<'_, GraphSubgraphId, Vec<GraphNodeId>> {
751        self.subgraph_nodes.keys()
752    }
753
754    /// Subgraph IDs in topological sort order.
755    pub fn subgraph_toposort(&self) -> &[GraphSubgraphId] {
756        &self.subgraph_toposort
757    }
758
759    /// Set the topological sort order for subgraphs.
760    pub fn set_subgraph_toposort(&mut self, order: Vec<GraphSubgraphId>) {
761        self.subgraph_toposort = order;
762    }
763
764    /// Iterator over all subgraphs, ID and members: `(GraphSubgraphId, Vec<GraphNodeId>)`.
765    pub fn subgraphs(&self) -> slotmap::basic::Iter<'_, GraphSubgraphId, Vec<GraphNodeId>> {
766        self.subgraph_nodes.iter()
767    }
768
769    /// Create a subgraph consisting of `node_ids`. Returns an error if any of the nodes are already in a subgraph.
770    pub fn insert_subgraph(
771        &mut self,
772        node_ids: Vec<GraphNodeId>,
773    ) -> Result<GraphSubgraphId, (GraphNodeId, GraphSubgraphId)> {
774        // Check none are already in subgraphs
775        for &node_id in node_ids.iter() {
776            if let Some(&old_sg_id) = self.node_subgraph.get(node_id) {
777                return Err((node_id, old_sg_id));
778            }
779        }
780        let subgraph_id = self.subgraph_nodes.insert_with_key(|sg_id| {
781            for &node_id in node_ids.iter() {
782                self.node_subgraph.insert(node_id, sg_id);
783            }
784            node_ids
785        });
786
787        Ok(subgraph_id)
788    }
789
790    /// Removes a node from its subgraph. Returns true if the node was in a subgraph.
791    pub fn remove_from_subgraph(&mut self, node_id: GraphNodeId) -> bool {
792        if let Some(old_sg_id) = self.node_subgraph.remove(node_id) {
793            self.subgraph_nodes[old_sg_id].retain(|&other_node_id| other_node_id != node_id);
794            true
795        } else {
796            false
797        }
798    }
799
800    /// Gets the delay type for a handoff node, if set.
801    pub fn handoff_delay_type(&self, node_id: GraphNodeId) -> Option<DelayType> {
802        self.handoff_delay_type.get(node_id).copied()
803    }
804
805    /// Sets the delay type for a handoff node.
806    pub fn set_handoff_delay_type(&mut self, node_id: GraphNodeId, delay_type: DelayType) {
807        self.handoff_delay_type.insert(node_id, delay_type);
808    }
809
810    /// Helper: finds the first index in `subgraph_nodes` where it transitions from pull to push.
811    fn find_pull_to_push_idx(&self, subgraph_nodes: &[GraphNodeId]) -> usize {
812        subgraph_nodes
813            .iter()
814            .position(|&node_id| {
815                self.node_color(node_id)
816                    .is_some_and(|color| Color::Pull != color)
817            })
818            .unwrap_or(subgraph_nodes.len())
819    }
820}
821
822/// Display/output methods.
823impl DfirGraph {
824    /// Helper to generate a deterministic `Ident` for the given node.
825    fn node_as_ident(&self, node_id: GraphNodeId, is_pred: bool) -> Ident {
826        let name = match &self.nodes[node_id] {
827            GraphNode::Operator(_) => format!("op_{:?}", node_id.data()),
828            GraphNode::Handoff {
829                kind: HandoffKind::Vec,
830                ..
831            } => format!(
832                "hoff_{:?}_{}",
833                node_id.data(),
834                if is_pred { "recv" } else { "send" }
835            ),
836            GraphNode::Handoff {
837                kind: HandoffKind::Singleton | HandoffKind::Optional,
838                ..
839            } => format!(
840                "singleton_{:?}_{}",
841                node_id.data(),
842                if is_pred { "recv" } else { "send" }
843            ),
844            GraphNode::ModuleBoundary { .. } => panic!(),
845        };
846        let span = match (is_pred, &self.nodes[node_id]) {
847            (_, GraphNode::Operator(operator)) => operator.span(),
848            (true, &GraphNode::Handoff { src_span, .. }) => src_span,
849            (false, &GraphNode::Handoff { dst_span, .. }) => dst_span,
850            (_, GraphNode::ModuleBoundary { .. }) => panic!(),
851        };
852        Ident::new(&name, span)
853    }
854
855    /// Helper to generate the main buffer `Ident` for a handoff node.
856    fn hoff_buf_ident(&self, hoff_id: GraphNodeId, span: Span) -> Ident {
857        Ident::new(&format!("hoff_{:?}_buf", hoff_id.data()), span)
858    }
859
860    /// Helper to generate the back (double-buffer) `Ident` for a handoff node.
861    fn hoff_back_ident(&self, hoff_id: GraphNodeId, span: Span) -> Ident {
862        Ident::new(&format!("hoff_{:?}_back", hoff_id.data()), span)
863    }
864
865    /// Resolve the handoff references via [`Self::node_handoff_references`] for the given `node_id`.
866    /// Returns token streams for each reference:
867    /// - For `HandoffKind::Singleton`: `buf.as_ref().unwrap()` (shared, `&T`) or
868    ///   `buf.as_mut().unwrap()` (mutable, `&mut T`)
869    /// - For `HandoffKind::Optional`: `&buf` (shared, `&Option<T>`) or
870    ///   `&mut buf` (mutable, `&mut Option<T>`)
871    /// - For `HandoffKind::Vec`: `&buf` (shared, `&Vec<T>`) or
872    ///   `&mut buf` (mutable, `&mut Vec<T>`)
873    fn helper_resolve_singletons(&self, node_id: GraphNodeId, span: Span) -> Vec<TokenStream> {
874        self.node_handoff_references(node_id)
875            .iter()
876            .map(|resolved_ref| {
877                // TODO(mingwei): this `expect` should be caught in error checking
878                let ref_node_id = resolved_ref
879                    .node_id
880                    .expect("Expected singleton to be resolved but was not, this is a bug.");
881                let is_mut = resolved_ref.is_mut;
882                match self.node(ref_node_id) {
883                    GraphNode::Handoff {
884                        kind: HandoffKind::Singleton,
885                        ..
886                    } => {
887                        let buf_ident = self.hoff_buf_ident(ref_node_id, span);
888                        if is_mut {
889                            quote_spanned! {span=> #buf_ident.as_mut().unwrap() }
890                        } else {
891                            quote_spanned! {span=> #buf_ident.as_ref().unwrap() }
892                        }
893                    }
894                    GraphNode::Handoff {
895                        kind: HandoffKind::Optional | HandoffKind::Vec,
896                        ..
897                    } => {
898                        let buf_ident = self.hoff_buf_ident(ref_node_id, span);
899                        if is_mut {
900                            quote_spanned! {span=> &mut #buf_ident }
901                        } else {
902                            quote_spanned! {span=> &#buf_ident }
903                        }
904                    }
905                    _ => {
906                        unreachable!("Only handoff nodes should be reachable as handoff references")
907                    }
908                }
909            })
910            .collect::<Vec<_>>()
911    }
912
913    /// Returns each subgraph's receive and send handoffs.
914    /// `Map<GraphSubgraphId, (recv handoffs, send handoffs)>`
915    fn helper_collect_subgraph_handoffs(
916        &self,
917    ) -> SecondaryMap<GraphSubgraphId, (Vec<GraphNodeId>, Vec<GraphNodeId>)> {
918        // Get data on handoff src and dst subgraphs.
919        let mut subgraph_handoffs: SecondaryMap<
920            GraphSubgraphId,
921            (Vec<GraphNodeId>, Vec<GraphNodeId>),
922        > = self
923            .subgraph_nodes
924            .keys()
925            .map(|k| (k, Default::default()))
926            .collect();
927
928        // For each handoff/singleton node, add it to the `send`/`recv` lists for the corresponding subgraphs.
929        for (hoff_id, hoff) in self.nodes() {
930            if !matches!(hoff, GraphNode::Handoff { .. }) {
931                continue;
932            }
933            // Receivers from the handoff. (Should really only be one).
934            for (_edge, succ_id) in self.node_successors(hoff_id) {
935                let succ_sg = self
936                    .node_subgraph(succ_id)
937                    .expect("bug: successor not in subgraph, may be a doubled/adjacent handoff");
938                subgraph_handoffs[succ_sg].0.push(hoff_id);
939            }
940            // Senders into the handoff. (Should really only be one).
941            for (_edge, pred_id) in self.node_predecessors(hoff_id) {
942                let pred_sg = self
943                    .node_subgraph(pred_id)
944                    .expect("bug: predecessor not in subgraph, may be a doubled/adjacent handoff");
945                subgraph_handoffs[pred_sg].1.push(hoff_id);
946            }
947        }
948
949        subgraph_handoffs
950    }
951
952    /// Compute the output handoffs exiting each loop (sender inside, receiver outside).
953    /// Returns a map from loop ID to the list of handoff node IDs that exit that loop.
954    fn helper_loop_output_handoffs(&self) -> SecondaryMap<GraphLoopId, Vec<GraphNodeId>> {
955        let mut loop_hoffs_out = SecondaryMap::<GraphLoopId, Vec<GraphNodeId>>::new();
956
957        for (hoff_id, hoff) in self.nodes() {
958            if !matches!(hoff, GraphNode::Handoff { .. }) {
959                continue;
960            }
961
962            let loop_pred = self
963                .node_predecessors(hoff_id)
964                .next()
965                .and_then(|(_, pred)| self.node_loop(pred));
966            let loop_succ = self
967                .node_successors(hoff_id)
968                .next()
969                .and_then(|(_, succ)| self.node_loop(succ));
970
971            if let Some(loop_pred) = loop_pred
972                && loop_succ == self.loop_parent(loop_pred)
973            {
974                // Pred is inside a child loop, succ is in the parent/outer.
975                loop_hoffs_out
976                    .entry(loop_pred)
977                    .expect("loop removed")
978                    .or_default()
979                    .push(hoff_id);
980            }
981        }
982
983        loop_hoffs_out
984    }
985
986    /// Returns true if `node_loop` is `loop_id` or a (transitive) child of `loop_id`.
987    fn is_inside_loop(&self, node_loop: Option<GraphLoopId>, loop_id: GraphLoopId) -> bool {
988        let mut current = node_loop;
989        while let Some(l) = current {
990            if l == loop_id {
991                return true;
992            }
993            current = self.loop_parent(l);
994        }
995        false
996    }
997
998    /// Emit a loop gate: wraps `child_body` in the appropriate control structure
999    /// and appends the result + swap code to `output`.
1000    ///
1001    /// - **Root-level loops** (no parent) are fused with the tick: they emit an `if`
1002    ///   so the body runs at most once per tick when their entry condition is met.
1003    /// - **Nested loops** (have a parent loop) emit a `while` so they can iterate
1004    ///   until fixpoint (driven by `defer_tick` back-edges).
1005    /// - If there are no gate checks, the body is emitted unconditionally.
1006    fn emit_loop_gate(
1007        &self,
1008        loop_id: GraphLoopId,
1009        child_body: TokenStream,
1010        loop_input_handoffs: &SecondaryMap<GraphLoopId, Vec<GraphNodeId>>,
1011        back_edge_hoffs_and_lazyness: &SparseSecondaryMap<GraphNodeId, bool>,
1012        loop_swap_code: &std::collections::HashMap<GraphLoopId, Vec<TokenStream>>,
1013        output: &mut TokenStream,
1014    ) {
1015        // Get swap code for this loop's defer_tick handoffs.
1016        let swap_code = loop_swap_code
1017            .get(&loop_id)
1018            .map(|v| v.as_slice())
1019            .unwrap_or(&[]);
1020
1021        // Root-level loops are fused with the tick: emit `if` instead of `while`.
1022        let is_root_loop = self.loop_parent(loop_id).is_none();
1023
1024        // Build the gate condition from entry handoffs (excluding lazy windowing operators).
1025        let entry_handoffs = loop_input_handoffs.get(loop_id).expect("loop missing");
1026        let mut gate_checks: Vec<TokenStream> = entry_handoffs
1027            .iter()
1028            .filter(|&&hoff_id| {
1029                // Check if the successor (windowing operator) is lazy.
1030                // If so, exclude from the gate — it doesn't trigger the loop.
1031                let is_lazy = self
1032                    .node_successors(hoff_id)
1033                    .next()
1034                    .and_then(|(_, succ)| self.node_op_inst(succ))
1035                    .is_some_and(|op_inst| {
1036                        op_inst.op_constraints.flo_type == Some(FloType::WindowingLazy)
1037                    });
1038                !is_lazy
1039            })
1040            .map(|&hoff_id| {
1041                let span = self.node(hoff_id).span();
1042                let buf_ident = self.hoff_buf_ident(hoff_id, span);
1043                if back_edge_hoffs_and_lazyness.contains_key(hoff_id) {
1044                    let back_ident = self.hoff_back_ident(hoff_id, span);
1045                    quote_spanned! {span=> !#back_ident.is_empty() }
1046                } else {
1047                    quote_spanned! {span=> !#buf_ident.is_empty() }
1048                }
1049            })
1050            .collect();
1051
1052        // Non-lazy defer_tick back-buffers also contribute to the gate (nested loops only).
1053        if !is_root_loop {
1054            for (hoff_id, hoff) in self.nodes() {
1055                if !matches!(hoff, GraphNode::Handoff { .. }) {
1056                    continue;
1057                }
1058                let Some(delay_type) = self.handoff_delay_type(hoff_id) else {
1059                    continue;
1060                };
1061                if delay_type != DelayType::Loop {
1062                    continue;
1063                }
1064                // Check this handoff belongs to loop_id.
1065                let hoff_loop = self
1066                    .node_successors(hoff_id)
1067                    .next()
1068                    .and_then(|(_, succ)| self.node_subgraph(succ))
1069                    .and_then(|sg| self.subgraph_loop(sg));
1070                if hoff_loop != Some(loop_id) {
1071                    continue;
1072                }
1073                let span = self.node(hoff_id).span();
1074                let back_ident = self.hoff_back_ident(hoff_id, span);
1075                gate_checks.push(quote_spanned! {span=> !#back_ident.is_empty() });
1076            }
1077        }
1078
1079        // For root-level loops: non-lazy defer_tick back-buffers also contribute to the gate.
1080        // This ensures the loop fires on the next tick when data was deferred via defer_tick.
1081        if is_root_loop {
1082            for (hoff_id, hoff) in self.nodes() {
1083                if !matches!(hoff, GraphNode::Handoff { .. }) {
1084                    continue;
1085                }
1086                let Some(delay_type) = self.handoff_delay_type(hoff_id) else {
1087                    continue;
1088                };
1089                if delay_type != DelayType::Tick {
1090                    continue;
1091                }
1092                // Check this handoff's consumer is inside this root-level loop.
1093                let hoff_loop = self
1094                    .node_successors(hoff_id)
1095                    .next()
1096                    .and_then(|(_, succ)| self.node_subgraph(succ))
1097                    .and_then(|sg| self.subgraph_loop(sg));
1098                if hoff_loop != Some(loop_id) {
1099                    continue;
1100                }
1101                let span = self.node(hoff_id).span();
1102                let back_ident = self.hoff_back_ident(hoff_id, span);
1103                gate_checks.push(quote_spanned! {span=> !#back_ident.is_empty() });
1104            }
1105        }
1106
1107        // An eager windowing operator (`batch_eager()`) forces the loop to fire unconditionally,
1108        // even when its windowed input is empty. It is only valid at the entry of a root-level
1109        // loop (disallowed in nested loops during validation, since forcing a nested loop to
1110        // always fire would prevent its fixpoint iteration from terminating).
1111        let has_eager = entry_handoffs.iter().any(|&hoff_id| {
1112            self.node_successors(hoff_id)
1113                .next()
1114                .and_then(|(_, succ)| self.node_op_inst(succ))
1115                .is_some_and(|op_inst| {
1116                    op_inst.op_constraints.flo_type == Some(FloType::WindowingEager)
1117                })
1118        });
1119
1120        if has_eager && is_root_loop {
1121            // Eager entry: always run the loop body (gate forced true).
1122            output.extend(child_body);
1123            output.extend(quote! { #( #swap_code )* });
1124        } else if gate_checks.is_empty() {
1125            // No entry handoffs — always run.
1126            output.extend(child_body);
1127            output.extend(quote! { #( #swap_code )* });
1128        } else if is_root_loop {
1129            // Root-level loop: fused with tick, fire at most once.
1130            output.extend(quote! {
1131                #[allow(clippy::nonminimal_bool, reason = "codegen")]
1132                if false #( || #gate_checks )* {
1133                    #child_body
1134                    #( #swap_code )*
1135                }
1136            });
1137        } else {
1138            // Nested loop: iterate until fixpoint.
1139            output.extend(quote! {
1140                #[allow(clippy::nonminimal_bool, reason = "codegen")]
1141                while false #( || #gate_checks )* {
1142                    #child_body
1143                    #( #swap_code )*
1144                }
1145            });
1146        }
1147    }
1148
1149    /// Compute the input handoffs into each loop (predecessor outside, successor inside).
1150    fn helper_loop_input_handoffs(&self) -> SecondaryMap<GraphLoopId, Vec<GraphNodeId>> {
1151        let mut loop_hoffs_inn = SecondaryMap::<GraphLoopId, Vec<GraphNodeId>>::new();
1152
1153        // Check each handoff node.
1154        for (hoff_id, hoff) in self.nodes() {
1155            if !matches!(hoff, GraphNode::Handoff { .. }) {
1156                continue;
1157            }
1158
1159            // Get the loop context of the predecessor and successor.
1160            let loop_pred = self
1161                .node_predecessors(hoff_id)
1162                .next()
1163                .and_then(|(_, pred)| self.node_loop(pred));
1164            let loop_succ = self
1165                .node_successors(hoff_id)
1166                .next()
1167                .and_then(|(_, succ)| self.node_loop(succ));
1168
1169            if let Some(loop_succ) = loop_succ
1170                && loop_pred == self.loop_parent(loop_succ)
1171            {
1172                // Pred is parent/outer loop of succ.
1173                loop_hoffs_inn
1174                    .entry(loop_succ)
1175                    .expect("loop removed")
1176                    .or_default()
1177                    .push(hoff_id);
1178            }
1179        }
1180
1181        loop_hoffs_inn
1182    }
1183
1184    /// Emit this graph as runnable Rust source code tokens that execute inline.
1185    /// Generates a flat `async move |df: &mut Context|` closure where subgraph
1186    /// blocks are inlined in topological order, using local `Vec<T>` buffers
1187    /// instead of runtime handoffs. Each call to the closure runs one tick.
1188    ///
1189    /// The generated code block evaluates to a `Dfir` instance wrapping the
1190    /// closure. Operator prologues run at construction time on the `Context`
1191    /// before it is moved into `Dfir::new`. `Dfir` provides the `Context`
1192    /// to the closure on each tick run.
1193    ///
1194    /// Uses the default [`AsCodeOptions`] (aside from `include_type_guards`), so runtime metrics
1195    /// tracking is *not* included; use [`Self::as_code_with_options`] to opt in via
1196    /// [`AsCodeOptions::include_metrics_tracking`].
1197    ///
1198    /// # Errors
1199    ///
1200    /// Returns all diagnostics as `Err(diagnostics)` if any are errors
1201    /// (leaving `&mut diagnostics` empty).
1202    pub fn as_code(
1203        &self,
1204        root: &TokenStream,
1205        include_type_guards: bool,
1206        prefix: TokenStream,
1207        diagnostics: &mut Diagnostics,
1208    ) -> Result<TokenStream, Diagnostics> {
1209        self.as_code_with_options(
1210            root,
1211            &AsCodeOptions {
1212                exclude_type_guards: !include_type_guards,
1213                ..Default::default()
1214            },
1215            prefix,
1216            diagnostics,
1217        )
1218    }
1219
1220    /// Like [`Self::as_code`] but with the full set of [`AsCodeOptions`].
1221    ///
1222    /// The simulator calls `Dfir::new()` on each iteration, and as a part of that
1223    /// it does parsing of the metagraph and diagnostics blob. One of them causes spans to get allocated,
1224    /// each time a span is allocated, some threadlocal u32 is being incremented, and, on a long simulator run,
1225    /// the u32 overflows and panics.
1226    pub fn as_code_with_options(
1227        &self,
1228        root: &TokenStream,
1229        options: &AsCodeOptions,
1230        prefix: TokenStream,
1231        diagnostics: &mut Diagnostics,
1232    ) -> Result<TokenStream, Diagnostics> {
1233        let df = Ident::new(GRAPH, Span::call_site());
1234        let context = Ident::new(CONTEXT, Span::call_site());
1235        // Tick-local bump-allocated Vec handoff declarations (inside the tick closure).
1236        let bump_ident = Ident::new("__dfir_bump", Span::call_site());
1237
1238        // 1. Collect all handoff nodes.
1239        let handoff_nodes = self
1240            .nodes
1241            .iter()
1242            .filter_map(|(node_id, node)| match node {
1243                &GraphNode::Handoff {
1244                    kind,
1245                    src_span,
1246                    dst_span,
1247                } => Some((node_id, kind, (src_span, dst_span))),
1248                GraphNode::Operator(_) => None,
1249                GraphNode::ModuleBoundary { .. } => panic!(),
1250            })
1251            .collect::<Vec<_>>();
1252
1253        // Determine which handoff nodes are tick-boundary (defer_tick) back-edges.
1254        // These must remain as captured Vec<T> since they persist across ticks.
1255        // All other Vec handoffs will be bump-allocated (tick-local).
1256        let back_edge_hoffs_and_lazyness = handoff_nodes
1257            .iter()
1258            .map(|&(node_id, _, _)| node_id)
1259            .filter_map(|node_id| {
1260                let delay_type = self.handoff_delay_type(node_id)?;
1261                Some((
1262                    node_id,
1263                    matches!(delay_type, DelayType::TickLazy | DelayType::LoopLazy),
1264                ))
1265            })
1266            .collect::<SparseSecondaryMap<_, _>>();
1267
1268        // Back buffer idents, buf idents, and if they are lazy.
1269        let back_buffer_idents_laziness = handoff_nodes
1270            .iter()
1271            .filter_map(|&(hoff_id, _kind, (src_span, dst_span))| {
1272                back_edge_hoffs_and_lazyness.get(hoff_id).map(|&is_lazy| {
1273                    let span = src_span.join(dst_span).unwrap_or(src_span);
1274                    let back_ident = self.hoff_back_ident(hoff_id, span);
1275                    let buf_ident = self.hoff_buf_ident(hoff_id, span);
1276                    (back_ident, buf_ident, is_lazy)
1277                })
1278            })
1279            .collect::<Vec<_>>();
1280
1281        // Generate swap code for tick-boundary (defer_tick / defer_tick_lazy) handoffs.
1282        // At the end of each tick, swap the regular buffer and back buffer so the
1283        // consumer reads last tick's data from the back buffer.
1284        // Only tick-level swaps go here; loop-level swaps are emitted inside the loop gate.
1285        // IMPORTANT: For defer_tick handoffs whose consumer is inside a root-level loop,
1286        // the swap is emitted inside the `if` gate (via loop_swap_code), not at tick level.
1287        let back_edge_swap_code = handoff_nodes
1288            .iter()
1289            .filter(|&&(node_id, _kind, _)| {
1290                self.handoff_delay_type(node_id)
1291                    .is_some_and(|dt| matches!(dt, DelayType::Tick | DelayType::TickLazy))
1292            })
1293            .filter(|&&(hoff_id, _kind, _)| {
1294                // Exclude handoffs whose consumer is inside a root-level loop.
1295                // Those get their swap emitted inside the loop gate.
1296                let consumer_loop = self
1297                    .node_successors(hoff_id)
1298                    .next()
1299                    .and_then(|(_, succ)| self.node_subgraph(succ))
1300                    .and_then(|sg| self.subgraph_loop(sg));
1301                if let Some(loop_id) = consumer_loop {
1302                    // If it's a root-level loop, don't include in tick-level swap.
1303                    self.loop_parent(loop_id).is_some()
1304                } else {
1305                    // No loop context: emit at tick level (original behavior).
1306                    true
1307                }
1308            })
1309            .map(|&(hoff_id, _kind, _)| {
1310                let span = self.nodes[hoff_id].span();
1311                let buf_ident = self.hoff_buf_ident(hoff_id, span);
1312                let back_ident = self.hoff_back_ident(hoff_id, span);
1313                quote_spanned! {span=>
1314                    ::std::mem::swap(&mut #buf_ident, &mut #back_ident);
1315                }
1316            })
1317            .collect::<Vec<_>>();
1318
1319        // Collect per-loop swap code for defer_tick / defer_tick_lazy handoffs.
1320        // AND defer_tick / defer_tick_lazy handoffs inside root-level loops.
1321        // Keyed by the loop ID of the consumer (successor) of the handoff.
1322        let mut loop_swap_code: std::collections::HashMap<GraphLoopId, Vec<TokenStream>> =
1323            std::collections::HashMap::new();
1324        for &(hoff_id, _kind, _) in handoff_nodes.iter() {
1325            let Some(delay_type) = self.handoff_delay_type(hoff_id) else {
1326                continue;
1327            };
1328            // Find the loop this handoff belongs to (from its consumer's loop context).
1329            let loop_id = self
1330                .node_successors(hoff_id)
1331                .next()
1332                .and_then(|(_, succ)| self.node_subgraph(succ))
1333                .and_then(|sg| self.subgraph_loop(sg));
1334            let Some(loop_id) = loop_id else {
1335                continue;
1336            };
1337            let include = match delay_type {
1338                DelayType::Loop | DelayType::LoopLazy => true,
1339                DelayType::Tick | DelayType::TickLazy => {
1340                    // Only include in loop swap if this is a root-level loop.
1341                    self.loop_parent(loop_id).is_none()
1342                }
1343            };
1344            if !include {
1345                continue;
1346            }
1347            let span = self.nodes[hoff_id].span();
1348            let buf_ident = self.hoff_buf_ident(hoff_id, span);
1349            let back_ident = self.hoff_back_ident(hoff_id, span);
1350            loop_swap_code
1351                .entry(loop_id)
1352                .or_default()
1353                .push(quote_spanned! {span=>
1354                    ::std::mem::swap(&mut #buf_ident, &mut #back_ident);
1355                });
1356        }
1357
1358        // 2. Collect per-subgraph recv & send handoffs.
1359        let subgraph_handoffs = self.helper_collect_subgraph_handoffs();
1360
1361        // 3. Use pre-computed subgraph topological order.
1362        let all_subgraphs: Vec<_> = self
1363            .subgraph_toposort()
1364            .iter()
1365            .map(|&sg_id| (sg_id, self.subgraph(sg_id)))
1366            .collect();
1367
1368        // TODO(mingwei): If a handoff has no pipe consumers we should drop it as soon as possible, after all reference
1369        // consumers. Right now we just let these handoffs die at the end of the tick.
1370
1371        let mut op_prologue_code = Vec::new();
1372        let mut op_tick_end_code = Vec::new();
1373
1374        // Stack-based hierarchical code generation.
1375        // Each entry is (loop_id, body_tokens) for an open loop context.
1376        // The "current output" is always the innermost open context (or root).
1377        let mut loop_stack: Vec<(GraphLoopId, TokenStream)> = Vec::new();
1378        let mut current_output = TokenStream::new();
1379
1380        // Pre-compute loop gate data.
1381        let loop_input_handoffs = self.helper_loop_input_handoffs();
1382        let loop_output_handoffs = self.helper_loop_output_handoffs();
1383
1384        {
1385            for &(subgraph_id, subgraph_nodes) in all_subgraphs.iter() {
1386                let sg_loop = self.subgraph_loop(subgraph_id);
1387
1388                // Transition loop contexts: close loops we've exited, open loops we've entered.
1389                // Close loops until we're at the right level.
1390                while let Some(&(top_loop, _)) = loop_stack.last() {
1391                    if sg_loop == Some(top_loop) || self.is_inside_loop(sg_loop, top_loop) {
1392                        break;
1393                    }
1394                    // Pop: wrap the body in a loop gate and append to parent.
1395                    let (closed_loop, child_body) = loop_stack.pop().unwrap();
1396                    let target = if let Some((_, parent_body)) = loop_stack.last_mut() {
1397                        parent_body
1398                    } else {
1399                        &mut current_output
1400                    };
1401                    self.emit_loop_gate(
1402                        closed_loop,
1403                        child_body,
1404                        &loop_input_handoffs,
1405                        &back_edge_hoffs_and_lazyness,
1406                        &loop_swap_code,
1407                        target,
1408                    );
1409                }
1410
1411                // Open new loops if we've descended.
1412                if let Some(target_loop) = sg_loop
1413                    && loop_stack.last().map(|&(l, _)| l) != Some(target_loop)
1414                {
1415                    // Find the path of loops to open (from outermost to target).
1416                    let mut path = Vec::new();
1417                    let mut cur = Some(target_loop);
1418                    while let Some(l) = cur {
1419                        if loop_stack.last().map(|&(top, _)| top) == Some(l) {
1420                            break;
1421                        }
1422                        path.push(l);
1423                        cur = self.loop_parent(l);
1424                    }
1425                    // Push in outermost-first order, emitting exit-handoff declarations
1426                    // to the parent level before the while loop.
1427                    for &loop_id in path.iter().rev() {
1428                        // Declare exit-handoff buffers at the current (parent) level.
1429                        if let Some(exit_hoffs) = loop_output_handoffs.get(loop_id) {
1430                            let exit_hoff_decls = exit_hoffs.iter().map(|&hoff_id| {
1431                                let span = self.nodes[hoff_id].span();
1432                                let buf_ident = self.hoff_buf_ident(hoff_id, span);
1433                                let GraphNode::Handoff { kind, .. } = self.node(hoff_id) else {
1434                                    panic!()
1435                                };
1436                                match kind {
1437                                    HandoffKind::Vec => quote_spanned! {span=>
1438                                        let mut #buf_ident = #root::bumpalo::collections::Vec::new_in(&#bump_ident);
1439                                    },
1440                                    HandoffKind::Singleton | HandoffKind::Optional => quote_spanned! {span=>
1441                                        let mut #buf_ident = ::std::option::Option::None;
1442                                    },
1443                                }
1444                            });
1445                            let target = if let Some((_, body)) = loop_stack.last_mut() {
1446                                body
1447                            } else {
1448                                &mut current_output
1449                            };
1450                            target.extend(quote! { #( #exit_hoff_decls )* });
1451                        }
1452                        loop_stack.push((loop_id, TokenStream::new()));
1453                    }
1454                }
1455                let sg_metrics_ffi = subgraph_id.data().as_ffi();
1456                let (recv_hoffs, send_hoffs) = &subgraph_handoffs[subgraph_id];
1457
1458                // Generate buffer ident helpers for this subgraph's handoffs.
1459                let recv_port_idents: Vec<Ident> = recv_hoffs
1460                    .iter()
1461                    .map(|&hoff_id| self.node_as_ident(hoff_id, true))
1462                    .collect();
1463                let send_port_idents: Vec<Ident> = send_hoffs
1464                    .iter()
1465                    .map(|&hoff_id| self.node_as_ident(hoff_id, false))
1466                    .collect();
1467
1468                // Map handoff node IDs to buffer idents.
1469                let recv_buf_idents: Vec<Ident> = recv_hoffs
1470                    .iter()
1471                    .map(|&hoff_id| self.hoff_buf_ident(hoff_id, self.nodes[hoff_id].span()))
1472                    .collect();
1473                let send_buf_idents: Vec<Ident> = send_hoffs
1474                    .iter()
1475                    .map(|&hoff_id| self.hoff_buf_ident(hoff_id, self.nodes[hoff_id].span()))
1476                    .collect();
1477
1478                // Handoff kinds
1479                let recv_kinds = recv_hoffs
1480                    .iter()
1481                    .map(|&hoff_id| {
1482                        let GraphNode::Handoff { kind, .. } = self.node(hoff_id) else {
1483                            panic!()
1484                        };
1485                        *kind
1486                    })
1487                    .collect::<Vec<_>>();
1488                let send_kinds = send_hoffs
1489                    .iter()
1490                    .map(|&hoff_id| {
1491                        let GraphNode::Handoff { kind, .. } = self.node(hoff_id) else {
1492                            panic!()
1493                        };
1494                        *kind
1495                    })
1496                    .collect::<Vec<_>>();
1497
1498                // Recv port code: drain from buffer into iterator, tracking if non-empty.
1499                // For back-edge (defer_tick) handoffs, drain from the back buffer instead.
1500                // Also update handoff metrics (measured at recv, not send — see graph.rs).
1501                let recv_port_code: Vec<TokenStream> = recv_port_idents
1502                    .iter()
1503                    .zip(recv_buf_idents.iter())
1504                    .zip(recv_kinds.iter())
1505                    .zip(recv_hoffs.iter())
1506                    .map(|(((port_ident, buf_ident), &kind), &hoff_id)| {
1507                        let hoff_ffi = hoff_id.data().as_ffi();
1508                        // Use call_site span for internal identifiers to avoid
1509                        // hygiene issues when invoked through declarative macros
1510                        // (e.g. dfir_expect_warnings!). TODO(#2781): define these once.
1511                        let work_done = Ident::new("__dfir_work_done", Span::call_site());
1512                        let metrics = Ident::new("__dfir_metrics", Span::call_site());
1513
1514                        // Compute len and drain expressions based on handoff kind.
1515                        let (len_expr, drain_expr) = match kind {
1516                            HandoffKind::Singleton | HandoffKind::Optional => (
1517                                quote! { if #buf_ident.is_some() { 1usize } else { 0usize } },
1518                                quote! { #root::dfir_pipes::pull::iter(#buf_ident.take().into_iter()) },
1519                            ),
1520                            HandoffKind::Vec => {
1521                                // Special asymmetric handling for defer tick handoffs, which are double-buffered.
1522                                // The producer writes to the regular buffer; at end-of-tick the buffers are swapped,
1523                                // so the consumer drains from the back buffer (here).
1524                                let drain_ident = if back_edge_hoffs_and_lazyness.contains_key(hoff_id) {
1525                                    &self.hoff_back_ident(hoff_id, buf_ident.span())
1526                                } else {
1527                                    buf_ident
1528                                };
1529                                (
1530                                    quote! { #drain_ident.len() },
1531                                    quote! { #root::dfir_pipes::pull::iter(#drain_ident.drain(..)) },
1532                                )
1533                            }
1534                        };
1535
1536                        let track_hoff_metrics = options.include_metrics_tracking.then(|| {
1537                            quote_spanned! {port_ident.span()=>
1538                                let hoff_metrics = &#metrics.handoffs[
1539                                    #root::slotmap::KeyData::from_ffi(#hoff_ffi).into()
1540                                ];
1541                                hoff_metrics.total_items_count.update(|x| x + hoff_len);
1542                                hoff_metrics.curr_items_count.set(hoff_len);
1543                            }
1544                        });
1545
1546                        quote_spanned! {port_ident.span()=>
1547                            {
1548                                let hoff_len = #len_expr;
1549                                if hoff_len > 0 {
1550                                    #work_done = true;
1551                                }
1552                                #track_hoff_metrics
1553                            }
1554                            let #port_ident = #drain_expr;
1555                        }
1556                    })
1557                    .collect();
1558
1559                // Send port code: push into buffer.
1560                let send_port_code: Vec<TokenStream> = send_port_idents
1561                    .iter()
1562                    .zip(send_buf_idents.iter())
1563                    .zip(send_kinds.iter())
1564                    .map(|((port_ident, buf_ident), &kind)| {
1565                        match kind {
1566                            HandoffKind::Singleton => {
1567                                // Singleton slot: store exactly one item, panic on duplicate.
1568                                quote_spanned! {port_ident.span()=>
1569                                    let #port_ident = #root::dfir_pipes::push::for_each(|__item| {
1570                                        if #buf_ident.replace(__item).is_some() {
1571                                            panic!("singleton() received more than one item");
1572                                        }
1573                                    });
1574                                }
1575                            }
1576                            HandoffKind::Optional => {
1577                                // Optional slot: store at most one item, panic on duplicate.
1578                                quote_spanned! {port_ident.span()=>
1579                                    let #port_ident = #root::dfir_pipes::push::for_each(|__item| {
1580                                        if #buf_ident.replace(__item).is_some() {
1581                                            panic!("optional() received more than one item");
1582                                        }
1583                                    });
1584                                }
1585                            }
1586                            HandoffKind::Vec => {
1587                                quote_spanned! {port_ident.span()=>
1588                                    // TODO(mingwei): use `#root::dfir_pipes::push::vec_push`?
1589                                    let #port_ident = #root::dfir_pipes::push::for_each(|item| { #buf_ident.push(item); });
1590                                }
1591                            }
1592                        }
1593                    })
1594                    .collect();
1595
1596                // All nodes in a subgraph should be in the same loop.
1597                let loop_id = self.node_loop(subgraph_nodes[0]);
1598
1599                let mut subgraph_op_iter_code = Vec::new();
1600                let mut subgraph_op_iter_after_code = Vec::new();
1601                {
1602                    let pull_to_push_idx = self.find_pull_to_push_idx(subgraph_nodes);
1603
1604                    let (pull_half, push_half) = subgraph_nodes.split_at(pull_to_push_idx);
1605                    let nodes_iter = pull_half.iter().chain(push_half.iter().rev());
1606
1607                    for (idx, &node_id) in nodes_iter.enumerate() {
1608                        let node = &self.nodes[node_id];
1609                        assert!(
1610                            matches!(node, GraphNode::Operator(_)),
1611                            "Handoffs are not part of subgraphs."
1612                        );
1613                        let op_inst = &self.operator_instances[node_id];
1614
1615                        let op_span = node.span();
1616                        let op_name = op_inst.op_constraints.name;
1617                        // Use op's span for root. #root is expected to be correct, any errors should span back to the op gen.
1618                        let root = change_spans(root.clone(), op_span);
1619                        let op_constraints = OPERATORS
1620                            .iter()
1621                            .find(|op| op_name == op.name)
1622                            .unwrap_or_else(|| panic!("Failed to find op: {}", op_name));
1623
1624                        let ident = self.node_as_ident(node_id, false);
1625
1626                        {
1627                            // TODO clean this up.
1628                            // Collect input arguments (predecessors).
1629                            let mut input_edges = self
1630                                .graph
1631                                .predecessor_edges(node_id)
1632                                .map(|edge_id| (self.edge_ports(edge_id).1, edge_id))
1633                                .collect::<Vec<_>>();
1634                            // Ensure sorted by port index.
1635                            input_edges.sort();
1636
1637                            let inputs = input_edges
1638                                .iter()
1639                                .map(|&(_port, edge_id)| {
1640                                    let (pred, _) = self.edge(edge_id);
1641                                    self.node_as_ident(pred, true)
1642                                })
1643                                .collect::<Vec<_>>();
1644
1645                            // Collect output arguments (successors).
1646                            let mut output_edges = self
1647                                .graph
1648                                .successor_edges(node_id)
1649                                .map(|edge_id| (&self.ports[edge_id].0, edge_id))
1650                                .collect::<Vec<_>>();
1651                            // Ensure sorted by port index.
1652                            output_edges.sort();
1653
1654                            let outputs = output_edges
1655                                .iter()
1656                                .map(|&(_port, edge_id)| {
1657                                    let (_, succ) = self.edge(edge_id);
1658                                    self.node_as_ident(succ, false)
1659                                })
1660                                .collect::<Vec<_>>();
1661
1662                            let is_pull = idx < pull_to_push_idx;
1663
1664                            // There's a bit of dark magic hidden in `Span`s... you'd think it's just a `file:line:column`,
1665                            // but it has one extra bit of info for _name resolution_, used for `Ident`s. `Span::call_site()`
1666                            // has the (unhygienic) resolution we want, an ident is just solely determined by its string name,
1667                            // which is what you'd expect out of unhygienic proc macros like this. Meanwhile, declarative macros
1668                            // use `Span::mixed_site()` which is weird and I don't understand it. It turns out that if you call
1669                            // the dfir syntax proc macro from _within_ a declarative macro then `op_span` will have the
1670                            // bad `Span::mixed_site()` name resolution and cause "Cannot find value `df/context`" errors. So
1671                            // we call `.resolved_at()` to fix resolution back to `Span::call_site()`. -Mingwei
1672                            let df_local = &Ident::new(GRAPH, op_span.resolved_at(df.span()));
1673                            let context = &Ident::new(CONTEXT, op_span.resolved_at(context.span()));
1674
1675                            let singletons_resolved =
1676                                self.helper_resolve_singletons(node_id, op_span);
1677
1678                            let arguments = &process_singletons::postprocess_singletons(
1679                                op_inst.arguments_raw.clone(),
1680                                singletons_resolved,
1681                            );
1682
1683                            let source_tag = 'a: {
1684                                if let Some(tag) = self.operator_tag.get(node_id).cloned() {
1685                                    break 'a tag;
1686                                }
1687
1688                                if proc_macro::is_available() {
1689                                    let op_span = op_span.unwrap();
1690                                    break 'a format!(
1691                                        "loc_{}_{}_{}_{}_{}",
1692                                        crate::pretty_span::make_source_path_relative(
1693                                            &op_span.file()
1694                                        )
1695                                        .display()
1696                                        .to_string()
1697                                        .replace(|x: char| !x.is_ascii_alphanumeric(), "_"),
1698                                        op_span.start().line(),
1699                                        op_span.start().column(),
1700                                        op_span.end().line(),
1701                                        op_span.end().column(),
1702                                    );
1703                                }
1704
1705                                format!(
1706                                    "loc_nopath_{}_{}_{}_{}",
1707                                    op_span.start().line,
1708                                    op_span.start().column,
1709                                    op_span.end().line,
1710                                    op_span.end().column
1711                                )
1712                            };
1713
1714                            let work_fn = format_ident!(
1715                                "{}__{}__{}",
1716                                ident,
1717                                op_name,
1718                                source_tag,
1719                                span = op_span
1720                            );
1721                            let work_fn_async = format_ident!("{}__async", work_fn, span = op_span);
1722
1723                            let context_args = WriteContextArgs {
1724                                root: &root,
1725                                df_ident: df_local,
1726                                context,
1727                                subgraph_id,
1728                                node_id,
1729                                loop_id,
1730                                op_span,
1731                                op_tag: self.operator_tag.get(node_id).cloned(),
1732                                work_fn: &work_fn,
1733                                work_fn_async: &work_fn_async,
1734                                ident: &ident,
1735                                is_pull,
1736                                inputs: &inputs,
1737                                outputs: &outputs,
1738                                op_name,
1739                                op_inst,
1740                                arguments,
1741                            };
1742
1743                            let write_result =
1744                                (op_constraints.write_fn)(&context_args, diagnostics);
1745                            let OperatorWriteOutput {
1746                                write_prologue,
1747                                write_iterator,
1748                                write_iterator_after,
1749                                write_tick_end,
1750                            } = write_result.unwrap_or_else(|()| {
1751                                assert!(
1752                                    diagnostics.has_error(),
1753                                    "Operator `{}` returned `Err` but emitted no diagnostics, this is a bug.",
1754                                    op_name,
1755                                );
1756                                OperatorWriteOutput {
1757                                    write_iterator: null_write_iterator_fn(&context_args),
1758                                    ..Default::default()
1759                                }
1760                            });
1761
1762                            op_prologue_code.push(syn::parse_quote! {
1763                                #[allow(dead_code, non_snake_case, reason = "codegen")]
1764                                #[inline(always)]
1765                                fn #work_fn<T>(thunk: impl ::std::ops::FnOnce() -> T) -> T {
1766                                    thunk()
1767                                }
1768
1769                                #[allow(dead_code, non_snake_case, reason = "codegen")]
1770                                #[inline(always)]
1771                                async fn #work_fn_async<T>(
1772                                    thunk: impl ::std::future::Future<Output = T>,
1773                                ) -> T {
1774                                    thunk.await
1775                                }
1776                            });
1777                            op_prologue_code.push(write_prologue);
1778                            op_tick_end_code.push(write_tick_end);
1779                            subgraph_op_iter_code.push(write_iterator);
1780
1781                            if !options.exclude_type_guards {
1782                                let type_guard = if is_pull {
1783                                    quote_spanned! {op_span=>
1784                                        let #ident = {
1785                                            #[allow(non_snake_case)]
1786                                            #[inline(always)]
1787                                            pub fn #work_fn<Item, Input>(input: Input)
1788                                                -> impl #root::dfir_pipes::pull::Pull<Item = Item, Meta = (), CanPend = Input::CanPend, CanEnd = Input::CanEnd>
1789                                            where
1790                                                Input: #root::dfir_pipes::pull::Pull<Item = Item, Meta = ()>,
1791                                            {
1792                                                #root::pin_project_lite::pin_project! {
1793                                                    #[repr(transparent)]
1794                                                    struct Pull<Item, Input: #root::dfir_pipes::pull::Pull<Item = Item>> {
1795                                                        #[pin]
1796                                                        inner: Input
1797                                                    }
1798                                                }
1799
1800                                                impl<Item, Input> #root::dfir_pipes::pull::Pull for Pull<Item, Input>
1801                                                where
1802                                                    Input: #root::dfir_pipes::pull::Pull<Item = Item>,
1803                                                {
1804                                                    type Ctx<'ctx> = Input::Ctx<'ctx>;
1805
1806                                                    type Item = Item;
1807                                                    type Meta = Input::Meta;
1808                                                    type CanPend = Input::CanPend;
1809                                                    type CanEnd = Input::CanEnd;
1810
1811                                                    #[inline(always)]
1812                                                    fn pull(
1813                                                        self: ::std::pin::Pin<&mut Self>,
1814                                                        ctx: &mut Self::Ctx<'_>,
1815                                                    ) -> #root::dfir_pipes::pull::PullStep<Self::Item, Self::Meta, Self::CanPend, Self::CanEnd> {
1816                                                        #root::dfir_pipes::pull::Pull::pull(self.project().inner, ctx)
1817                                                    }
1818
1819                                                    #[inline(always)]
1820                                                    fn size_hint(&self) -> (usize, Option<usize>) {
1821                                                        #root::dfir_pipes::pull::Pull::size_hint(&self.inner)
1822                                                    }
1823                                                }
1824
1825                                                Pull {
1826                                                    inner: input
1827                                                }
1828                                            }
1829                                            #work_fn::<_, _>( #ident )
1830                                        };
1831                                    }
1832                                } else {
1833                                    quote_spanned! {op_span=>
1834                                        let #ident = {
1835                                            #[allow(non_snake_case)]
1836                                            #[inline(always)]
1837                                            pub fn #work_fn<Item, Psh>(psh: Psh) -> impl #root::dfir_pipes::push::Push<Item, (), CanPend = Psh::CanPend>
1838                                            where
1839                                                Psh: #root::dfir_pipes::push::Push<Item, ()>
1840                                            {
1841                                                #root::pin_project_lite::pin_project! {
1842                                                    #[repr(transparent)]
1843                                                    struct PushGuard<Psh> {
1844                                                        #[pin]
1845                                                        inner: Psh,
1846                                                    }
1847                                                }
1848
1849                                                impl<Item, Psh> #root::dfir_pipes::push::Push<Item, ()> for PushGuard<Psh>
1850                                                where
1851                                                    Psh: #root::dfir_pipes::push::Push<Item, ()>,
1852                                                {
1853                                                    type Ctx<'ctx> = Psh::Ctx<'ctx>;
1854
1855                                                    type CanPend = Psh::CanPend;
1856
1857                                                    #[inline(always)]
1858                                                    fn poll_ready(
1859                                                        self: ::std::pin::Pin<&mut Self>,
1860                                                        ctx: &mut Self::Ctx<'_>,
1861                                                    ) -> #root::dfir_pipes::push::PushStep<Self::CanPend> {
1862                                                        #root::dfir_pipes::push::Push::poll_ready(self.project().inner, ctx)
1863                                                    }
1864
1865                                                    #[inline(always)]
1866                                                    fn start_send(
1867                                                        self: ::std::pin::Pin<&mut Self>,
1868                                                        item: Item,
1869                                                        meta: (),
1870                                                    ) {
1871                                                        #root::dfir_pipes::push::Push::start_send(self.project().inner, item, meta)
1872                                                    }
1873
1874                                                    #[inline(always)]
1875                                                    fn poll_finalize(
1876                                                        self: ::std::pin::Pin<&mut Self>,
1877                                                        ctx: &mut Self::Ctx<'_>,
1878                                                    ) -> #root::dfir_pipes::push::PushStep<Self::CanPend> {
1879                                                        #root::dfir_pipes::push::Push::poll_finalize(self.project().inner, ctx)
1880                                                    }
1881
1882                                                    #[inline(always)]
1883                                                    fn size_hint(
1884                                                        self: ::std::pin::Pin<&mut Self>,
1885                                                        hint: (usize, Option<usize>),
1886                                                    ) {
1887                                                        #root::dfir_pipes::push::Push::size_hint(self.project().inner, hint)
1888                                                    }
1889                                                }
1890
1891                                                PushGuard {
1892                                                    inner: psh
1893                                                }
1894                                            }
1895                                            #work_fn( #ident )
1896                                        };
1897                                    }
1898                                };
1899                                subgraph_op_iter_code.push(type_guard);
1900                            }
1901                            subgraph_op_iter_after_code.push(write_iterator_after);
1902                        }
1903                    }
1904
1905                    {
1906                        // Determine pull and push halves of the `Pivot`.
1907                        let pull_ident = if 0 < pull_to_push_idx {
1908                            self.node_as_ident(subgraph_nodes[pull_to_push_idx - 1], false)
1909                        } else {
1910                            // Entire subgraph is push (with a single recv/pull handoff input).
1911                            recv_port_idents[0].clone()
1912                        };
1913
1914                        #[rustfmt::skip]
1915                        let push_ident = if let Some(&node_id) =
1916                            subgraph_nodes.get(pull_to_push_idx)
1917                        {
1918                            self.node_as_ident(node_id, false)
1919                        } else if 1 == send_port_idents.len() {
1920                            // Entire subgraph is pull (with a single send/push handoff output).
1921                            send_port_idents[0].clone()
1922                        } else {
1923                            diagnostics.push(Diagnostic::spanned(
1924                                pull_ident.span(),
1925                                Level::Error,
1926                                "Degenerate subgraph detected, is there a disconnected `null()` or other degenerate pipeline somewhere?",
1927                            ));
1928                            continue;
1929                        };
1930
1931                        // Pivot span is combination of pull and push spans (or if not possible, just take the push).
1932                        let pivot_span = pull_ident
1933                            .span()
1934                            .join(push_ident.span())
1935                            .unwrap_or_else(|| push_ident.span());
1936                        let pivot_fn_ident = Ident::new(
1937                            &format!("pivot_run_sg_{:?}", subgraph_id.data()),
1938                            pivot_span,
1939                        );
1940                        let root = change_spans(root.clone(), pivot_span);
1941                        subgraph_op_iter_code.push(quote_spanned! {pivot_span=>
1942                            #[inline(always)]
1943                            fn #pivot_fn_ident<Pul, Psh, Item>(pull: Pul, push: Psh)
1944                                -> impl ::std::future::Future<Output = ()>
1945                            where
1946                                Pul: #root::dfir_pipes::pull::Pull<Item = Item>,
1947                                Psh: #root::dfir_pipes::push::Push<Item, Pul::Meta>,
1948                            {
1949                                #root::dfir_pipes::pull::Pull::send_push(pull, push)
1950                            }
1951                            (#pivot_fn_ident)(#pull_ident, #push_ident).await;
1952                        });
1953                    }
1954                };
1955
1956                // Each subgraph block is an async block so it can be individually instrumented.
1957                // Note: this ident is for the subgraph future, not a runtime SubgraphId binding
1958                // (unlike the scheduled path's `sg_ident`).
1959                let sg_fut_ident = subgraph_id.as_ident(Span::call_site());
1960
1961                // Generate send-side curr_items_count updates (after subgraph runs).
1962                let send_metrics_code = if options.include_metrics_tracking {
1963                    send_hoffs
1964                        .iter()
1965                        .zip(send_buf_idents.iter())
1966                        .zip(send_kinds.iter())
1967                        .map(|((&hoff_id, buf_ident), &kind)| {
1968                            let hoff_ffi = hoff_id.data().as_ffi();
1969                            let len_expr = match kind {
1970                                HandoffKind::Singleton | HandoffKind::Optional => {
1971                                    quote! { if #buf_ident.is_some() { 1 } else { 0 } }
1972                                }
1973                                HandoffKind::Vec => {
1974                                    quote! { #buf_ident.len() }
1975                                }
1976                            };
1977                            quote! {
1978                                __dfir_metrics.handoffs[
1979                                    #root::slotmap::KeyData::from_ffi(#hoff_ffi).into()
1980                                ].curr_items_count.set(#len_expr);
1981                            }
1982                        })
1983                        .collect::<Vec<_>>()
1984                } else {
1985                    Vec::new()
1986                };
1987
1988                // Create the handoffs we are about to push to (send).
1989                // Exit handoffs (sender inside a loop, receiver in parent) are already declared
1990                // before the while loop, so skip them here.
1991                let send_hoff_make_code = send_buf_idents.iter()
1992                    .zip(send_kinds.iter())
1993                    .zip(send_hoffs.iter())
1994                    .filter_map(|((buf_ident, &kind), &hoff_id)| {
1995                        let span = buf_ident.span();
1996                        if back_edge_hoffs_and_lazyness.contains_key(hoff_id) {
1997                            // Defer_tick send buffers are declared outside the tick closure
1998                            // as std::vec::Vec for O(1) swap. Just clear here.
1999                            Some(quote_spanned! {span=>
2000                                #buf_ident.clear();
2001                            })
2002                        } else {
2003                            // Check if this is a loop-exit handoff: sender is in a loop,
2004                            // receiver is in the parent (already declared outside the while loop).
2005                            let receiver_loop = self
2006                                .node_successors(hoff_id)
2007                                .next()
2008                                .and_then(|(_, succ)| self.node_loop(succ));
2009                            let is_exit = if let Some(sender_loop) = sg_loop {
2010                                receiver_loop == self.loop_parent(sender_loop)
2011                            } else {
2012                                false
2013                            };
2014                            if is_exit {
2015                                // Exit handoff: buffer already declared at parent level.
2016                                None
2017                            } else {
2018                                Some(match kind {
2019                                    HandoffKind::Vec => quote_spanned! {span=>
2020                                        let mut #buf_ident = #root::bumpalo::collections::Vec::new_in(&#bump_ident);
2021                                    },
2022                                    HandoffKind::Singleton | HandoffKind::Optional => quote_spanned! {span=>
2023                                        let mut #buf_ident = ::std::option::Option::None;
2024                                    },
2025                                })
2026                            }
2027                        }
2028                    })
2029                    .collect::<Vec<_>>();
2030                // Drop the handoffs we just drained (recv).
2031                // TODO(mingwei): we could use `.into_iter()` instead of `.drain(..)` to consume the handoffs directly.
2032                // This only works for handoffs within the tick, though, not `defer_tick` handoffs.
2033                let recv_hoff_drop_code = recv_buf_idents
2034                    .iter()
2035                    .zip(recv_hoffs.iter())
2036                    .filter(|&(_, &hoff_id)| !back_edge_hoffs_and_lazyness.contains_key(hoff_id))
2037                    .map(|(buf_ident, _)| {
2038                        let span = buf_ident.span();
2039                        quote_spanned! {span=>
2040                            let _ = #buf_ident;
2041                        }
2042                    });
2043
2044                let run_sg = if options.include_metrics_tracking {
2045                    quote! {
2046                        // Instrument w/ the subgraph metrics.
2047                        let sg_metrics = &__dfir_metrics.subgraphs[
2048                            #root::slotmap::KeyData::from_ffi(#sg_metrics_ffi).into()
2049                        ];
2050                        #root::scheduled::metrics::InstrumentSubgraph::new(
2051                            #sg_fut_ident, sg_metrics
2052                        ).await;
2053                        sg_metrics.total_run_count.update(|x| x + 1);
2054
2055                        // Update send (output) handoff metrics.
2056                        #( #send_metrics_code )*
2057                    }
2058                } else {
2059                    quote! {
2060                        #sg_fut_ident.await;
2061                    }
2062                };
2063
2064                // Emit subgraph block to the current loop level (top of stack or root).
2065                let sg_block = quote! {
2066                    // Create the handoffs we are about to push to (send).
2067                    #( #send_hoff_make_code )*
2068
2069                    let #sg_fut_ident = async {
2070                        let #context = &#df;
2071                        #( #recv_port_code )*
2072                        #( #send_port_code )*
2073                        #( #subgraph_op_iter_code )*
2074                        #( #subgraph_op_iter_after_code )*
2075                    };
2076                    {
2077                        #run_sg
2078
2079                        // Drop the handoffs we just drained (recv).
2080                        #( #recv_hoff_drop_code )*
2081                    }
2082                };
2083                if let Some((_, body)) = loop_stack.last_mut() {
2084                    body.extend(sg_block);
2085                } else {
2086                    current_output.extend(sg_block);
2087                }
2088            }
2089        }
2090
2091        // Close any remaining open loops.
2092        let gated_subgraph_code = {
2093            while let Some((closed_loop, child_body)) = loop_stack.pop() {
2094                let target = if let Some((_, parent_body)) = loop_stack.last_mut() {
2095                    parent_body
2096                } else {
2097                    &mut current_output
2098                };
2099                self.emit_loop_gate(
2100                    closed_loop,
2101                    child_body,
2102                    &loop_input_handoffs,
2103                    &back_edge_hoffs_and_lazyness,
2104                    &loop_swap_code,
2105                    target,
2106                );
2107            }
2108            current_output
2109        };
2110
2111        if diagnostics.has_error() {
2112            return Err(std::mem::take(diagnostics));
2113        }
2114        let _ = diagnostics; // Ensure no more diagnostics may be added after checking for errors.
2115
2116        let (meta_graph_arg, diagnostics_arg) = if !options.exclude_meta {
2117            let meta_graph_json = serde_json::to_string(&self).unwrap();
2118            let meta_graph_json = Literal::string(&meta_graph_json);
2119
2120            let serde_diagnostics: Vec<_> = diagnostics.iter().map(Diagnostic::to_serde).collect();
2121            let diagnostics_json = serde_json::to_string(&*serde_diagnostics).unwrap();
2122            let diagnostics_json = Literal::string(&diagnostics_json);
2123
2124            (
2125                quote! { Some(#meta_graph_json) },
2126                quote! { Some(#diagnostics_json) },
2127            )
2128        } else {
2129            (quote! { None }, quote! { None })
2130        };
2131
2132        // Generate metrics initialization: one entry per handoff and per subgraph.
2133        let metrics_init_code = if options.include_metrics_tracking {
2134            let handoff_inits = handoff_nodes.iter().map(|&(node_id, _, _)| {
2135                let ffi = node_id.data().as_ffi();
2136                quote! {
2137                    dfir_metrics.handoffs.insert(
2138                        #root::slotmap::KeyData::from_ffi(#ffi).into(),
2139                        ::std::default::Default::default(),
2140                    );
2141                }
2142            });
2143            let subgraph_inits = all_subgraphs.iter().map(|&(sg_id, _)| {
2144                let ffi = sg_id.data().as_ffi();
2145                quote! {
2146                    dfir_metrics.subgraphs.insert(
2147                        #root::slotmap::KeyData::from_ffi(#ffi).into(),
2148                        ::std::default::Default::default(),
2149                    );
2150                }
2151            });
2152            handoff_inits.chain(subgraph_inits).collect::<Vec<_>>()
2153        } else {
2154            Vec::new()
2155        };
2156
2157        // For creating back-buffer handoff vecs.
2158        let back_buffer_idents = back_buffer_idents_laziness
2159            .iter()
2160            .map(|(back_ident, _, _)| back_ident);
2161        // For creating the send-side buffer for defer_tick handoffs (also outside the closure).
2162        let defer_tick_buf_idents = back_buffer_idents_laziness
2163            .iter()
2164            .map(|(_, buf_ident, _)| buf_ident);
2165        // For checking if we should start the next tick (`schedule_subgraph`):
2166        // Collect the ident to check for each non-lazy back-edge handoff.
2167        // - For defer_tick handoffs in a root-level loop: check `back` (swap happened inside `if`)
2168        // - For all others: check `buf` (original behavior; tick-level swap hasn't happened yet)
2169        let non_lazy_schedule_idents: Vec<&Ident> = handoff_nodes
2170            .iter()
2171            .filter_map(|&(hoff_id, _, _)| {
2172                let delay_type = self.handoff_delay_type(hoff_id)?;
2173                // Only non-lazy.
2174                if matches!(delay_type, DelayType::TickLazy | DelayType::LoopLazy) {
2175                    return None;
2176                }
2177                let span = self.nodes[hoff_id].span();
2178                let expected_back_ident = self.hoff_back_ident(hoff_id, span);
2179                let entry = back_buffer_idents_laziness
2180                    .iter()
2181                    .find(|(back_ident, _, _)| *back_ident == expected_back_ident)?;
2182
2183                // For defer_tick inside a root-level loop, check `back`.
2184                if delay_type == DelayType::Tick {
2185                    let consumer_loop = self
2186                        .node_successors(hoff_id)
2187                        .next()
2188                        .and_then(|(_, succ)| self.node_subgraph(succ))
2189                        .and_then(|sg| self.subgraph_loop(sg));
2190                    if consumer_loop.is_some_and(|lid| self.loop_parent(lid).is_none()) {
2191                        return Some(&entry.0); // back ident
2192                    }
2193                }
2194                Some(&entry.1) // buf ident
2195            })
2196            .collect();
2197
2198        // Prologues and buffer declarations persist across ticks (outside the closure).
2199        // Subgraph blocks run each tick (inside the closure).
2200        Ok(quote! {
2201            {
2202                #prefix
2203
2204                use #root::{var_expr, var_args};
2205
2206                let __dfir_wake_state = ::std::sync::Arc::new(
2207                    #root::scheduled::context::WakeState::default()
2208                );
2209
2210                let __dfir_metrics = {
2211                    let mut dfir_metrics = #root::scheduled::metrics::DfirMetrics::default();
2212                    #( #metrics_init_code )*
2213                    ::std::rc::Rc::new(dfir_metrics)
2214                };
2215
2216                #[allow(unused_mut)]
2217                let mut #df = #root::scheduled::context::Context::new(
2218                    ::std::clone::Clone::clone(&__dfir_wake_state),
2219                    __dfir_metrics,
2220                );
2221
2222                #( #op_prologue_code )*
2223
2224                // For tick-boundary handoffs (`defer_tick` / `defer_tick_lazy`), declare both the
2225                // send buffer and the "back" buffer as std::vec::Vec outside the tick closure.
2226                // This enables O(1) mem::swap at end of tick for double-buffering.
2227                #( let mut #back_buffer_idents = ::std::vec::Vec::new(); )*
2228                #( let mut #defer_tick_buf_idents = ::std::vec::Vec::new(); )*
2229
2230                // Bump allocator for handoffs (except for back-edge handoffs, above).
2231                let mut #bump_ident = #root::bumpalo::Bump::new();
2232
2233                // Pre-set to true so the first tick always returns true
2234                // (matching Dfir pre-scheduling behavior). Subsequent ticks
2235                // start false (from take()) and are set true by recv port code
2236                // if any handoff buffer has data.
2237                let mut __dfir_work_done = true;
2238                #[allow(unused_qualifications, unused_mut, unused_variables, clippy::await_holding_refcell_ref, clippy::deref_addrof)]
2239                let __dfir_inline_tick = async move |#df: &mut #root::scheduled::context::Context| {
2240                    // Reset arena between ticks (start-of-tick)
2241                    #bump_ident.reset();
2242
2243                    {
2244                        let __dfir_metrics = #df.metrics();
2245
2246                        #gated_subgraph_code
2247
2248                        // For non-lazy defer_tick: if any deferred buffer has data,
2249                        // signal that another tick should run.
2250                        #[allow(clippy::nonminimal_bool, reason = "codegen")]
2251                        if false #( || !#non_lazy_schedule_idents.is_empty() )* {
2252                            #df.schedule_subgraph(true);
2253                        }
2254
2255                        // Double-buffer swap for defer_tick handoffs: move last tick's producer output (regular buffer)
2256                        // into the back buffer for the consumer to drain.
2257                        #( #back_edge_swap_code )*
2258                    }
2259
2260                    // End-of-tick per-operator state handling (i.e. 'tick persistence).
2261                    #( #op_tick_end_code )*
2262
2263                    #df.__end_tick();
2264
2265                    ::std::mem::take(&mut __dfir_work_done)
2266                };
2267                #root::scheduled::context::Dfir::new(
2268                    __dfir_inline_tick,
2269                    #df,
2270                    #meta_graph_arg,
2271                    #diagnostics_arg,
2272                )
2273            }
2274        })
2275    }
2276
2277    /// Color mode (pull vs. push, handoff vs. comp) for nodes. Some nodes can be push *OR* pull;
2278    /// those nodes will not be set in the returned map.
2279    pub fn node_color_map(&self) -> SparseSecondaryMap<GraphNodeId, Color> {
2280        let mut node_color_map: SparseSecondaryMap<GraphNodeId, Color> = self
2281            .node_ids()
2282            .filter_map(|node_id| {
2283                let op_color = self.node_color(node_id)?;
2284                Some((node_id, op_color))
2285            })
2286            .collect();
2287
2288        // Fill in rest via subgraphs.
2289        for sg_nodes in self.subgraph_nodes.values() {
2290            let pull_to_push_idx = self.find_pull_to_push_idx(sg_nodes);
2291
2292            for (idx, node_id) in sg_nodes.iter().copied().enumerate() {
2293                let is_pull = idx < pull_to_push_idx;
2294                node_color_map.insert(node_id, if is_pull { Color::Pull } else { Color::Push });
2295            }
2296        }
2297
2298        node_color_map
2299    }
2300
2301    /// Writes this graph as mermaid into a string.
2302    pub fn to_mermaid(&self, write_config: &WriteConfig) -> String {
2303        let mut output = String::new();
2304        self.write_mermaid(&mut output, write_config).unwrap();
2305        output
2306    }
2307
2308    /// Writes this graph as mermaid into the given `Write`.
2309    pub fn write_mermaid(
2310        &self,
2311        output: impl std::fmt::Write,
2312        write_config: &WriteConfig,
2313    ) -> std::fmt::Result {
2314        let mut graph_write = Mermaid::new(output);
2315        self.write_graph(&mut graph_write, write_config)
2316    }
2317
2318    /// Writes this graph as DOT (graphviz) into a string.
2319    pub fn to_dot(&self, write_config: &WriteConfig) -> String {
2320        let mut output = String::new();
2321        let mut graph_write = Dot::new(&mut output);
2322        self.write_graph(&mut graph_write, write_config).unwrap();
2323        output
2324    }
2325
2326    /// Writes this graph as DOT (graphviz) into the given `Write`.
2327    pub fn write_dot(
2328        &self,
2329        output: impl std::fmt::Write,
2330        write_config: &WriteConfig,
2331    ) -> std::fmt::Result {
2332        let mut graph_write = Dot::new(output);
2333        self.write_graph(&mut graph_write, write_config)
2334    }
2335
2336    /// Write out this graph using the given `GraphWrite`. E.g. `Mermaid` or `Dot`.
2337    pub(crate) fn write_graph<W>(
2338        &self,
2339        mut graph_write: W,
2340        write_config: &WriteConfig,
2341    ) -> Result<(), W::Err>
2342    where
2343        W: GraphWrite,
2344    {
2345        fn helper_edge_label(
2346            src_port: &PortIndexValue,
2347            dst_port: &PortIndexValue,
2348        ) -> Option<String> {
2349            let src_label = match src_port {
2350                PortIndexValue::Path(path) => Some(path.to_token_stream().to_string()),
2351                PortIndexValue::Int(index) => Some(index.value.to_string()),
2352                _ => None,
2353            };
2354            let dst_label = match dst_port {
2355                PortIndexValue::Path(path) => Some(path.to_token_stream().to_string()),
2356                PortIndexValue::Int(index) => Some(index.value.to_string()),
2357                _ => None,
2358            };
2359            let label = match (src_label, dst_label) {
2360                (Some(l1), Some(l2)) => Some(format!("{}\n{}", l1, l2)),
2361                (Some(l1), None) => Some(l1),
2362                (None, Some(l2)) => Some(l2),
2363                (None, None) => None,
2364            };
2365            label
2366        }
2367
2368        // Make node color map one time.
2369        let node_color_map = self.node_color_map();
2370
2371        // Write prologue.
2372        graph_write.write_prologue()?;
2373
2374        // Define nodes.
2375        let mut skipped_handoffs = BTreeSet::new();
2376        for (node_id, node) in self.nodes() {
2377            if matches!(node, GraphNode::Handoff { .. }) && write_config.no_handoffs {
2378                skipped_handoffs.insert(node_id);
2379                continue;
2380            }
2381            graph_write.write_node_definition(
2382                node_id,
2383                &if write_config.op_short_text {
2384                    node.to_name_string()
2385                } else if write_config.op_text_no_imports {
2386                    // Remove any lines that start with "use" (imports)
2387                    let full_text = node.to_pretty_string();
2388                    let mut output = String::new();
2389                    for sentence in full_text.split('\n') {
2390                        if sentence.trim().starts_with("use") {
2391                            continue;
2392                        }
2393                        output.push('\n');
2394                        output.push_str(sentence);
2395                    }
2396                    output.into()
2397                } else {
2398                    node.to_pretty_string()
2399                },
2400                if write_config.no_pull_push {
2401                    None
2402                } else {
2403                    node_color_map.get(node_id).copied()
2404                },
2405            )?;
2406        }
2407
2408        // Write edges.
2409        for (edge_id, (src_id, mut dst_id)) in self.edges() {
2410            // Handling for if `write_config.no_handoffs` true.
2411            if skipped_handoffs.contains(&src_id) {
2412                continue;
2413            }
2414
2415            let (src_port, mut dst_port) = self.edge_ports(edge_id);
2416            if skipped_handoffs.contains(&dst_id) {
2417                // The destination is a hidden handoff. If it has a successor, skip through.
2418                // If it has 0 successors (ref-only singleton), drop this edge entirely —
2419                // the data dependency is captured via the reference edge instead.
2420                let mut handoff_succs = self.node_successors(dst_id);
2421                if handoff_succs.len() == 0 {
2422                    continue;
2423                }
2424                let (succ_edge, succ_node) = handoff_succs.next().unwrap();
2425                dst_id = succ_node;
2426                dst_port = self.edge_ports(succ_edge).1;
2427            }
2428
2429            let label = helper_edge_label(src_port, dst_port);
2430            let delay_type = self
2431                .node_op_inst(dst_id)
2432                .and_then(|op_inst| (op_inst.op_constraints.input_delaytype_fn)(dst_port));
2433            graph_write.write_edge(src_id, dst_id, delay_type, label.as_deref(), false)?;
2434        }
2435
2436        // Write reference edges.
2437        if !write_config.no_references {
2438            for dst_id in self.node_ids() {
2439                for src_ref_id in self
2440                    .node_handoff_references(dst_id)
2441                    .iter()
2442                    .filter_map(|r| r.node_id)
2443                {
2444                    // When handoffs are hidden, resolve through to the predecessor of
2445                    // the singleton handoff so the edge points from the actual writer.
2446                    let resolved_src = if skipped_handoffs.contains(&src_ref_id) {
2447                        self.node_predecessor_nodes(src_ref_id).next()
2448                    } else {
2449                        Some(src_ref_id)
2450                    };
2451                    let Some(resolved_src) = resolved_src else {
2452                        continue;
2453                    };
2454                    let label = None;
2455                    graph_write.write_edge(resolved_src, dst_id, None, label, true)?;
2456                }
2457            }
2458        }
2459
2460        // The following code is a little bit tricky. Generally, the graph has the hierarchy:
2461        // `loop -> subgraph -> varname -> node`. However, each of these can be disabled via the `write_config`. To
2462        // handle both the enabled and disabled case, this code is structured as a series of nested loops. If the layer
2463        // is disabled, then the HashMap<Option<KEY>, Vec<VALUE>> will only have a single key (`None`) with a
2464        // corresponding `Vec` value containing everything. This way no special handling is needed for the next layer.
2465
2466        // Loop -> Subgraphs
2467        let loop_subgraphs = self.subgraph_ids().map(|sg_id| {
2468            let loop_id = if write_config.no_loops {
2469                None
2470            } else {
2471                self.subgraph_loop(sg_id)
2472            };
2473            (loop_id, sg_id)
2474        });
2475        let loop_subgraphs = into_group_map(loop_subgraphs);
2476        for (loop_id, subgraph_ids) in loop_subgraphs {
2477            if let Some(loop_id) = loop_id {
2478                graph_write.write_loop_start(loop_id)?;
2479            }
2480
2481            // Subgraph -> Varnames.
2482            let subgraph_varnames_nodes = subgraph_ids.into_iter().flat_map(|sg_id| {
2483                self.subgraph(sg_id).iter().copied().map(move |node_id| {
2484                    let opt_sg_id = if write_config.no_subgraphs {
2485                        None
2486                    } else {
2487                        Some(sg_id)
2488                    };
2489                    (opt_sg_id, (self.node_varname(node_id), node_id))
2490                })
2491            });
2492            let subgraph_varnames_nodes = into_group_map(subgraph_varnames_nodes);
2493            for (sg_id, varnames) in subgraph_varnames_nodes {
2494                if let Some(sg_id) = sg_id {
2495                    graph_write.write_subgraph_start(sg_id)?;
2496                }
2497
2498                // Varnames -> Nodes.
2499                let varname_nodes = varnames.into_iter().map(|(varname, node)| {
2500                    let varname = if write_config.no_varnames {
2501                        None
2502                    } else {
2503                        varname
2504                    };
2505                    (varname, node)
2506                });
2507                let varname_nodes = into_group_map(varname_nodes);
2508                for (varname, node_ids) in varname_nodes {
2509                    if let Some(varname) = varname {
2510                        graph_write.write_varname_start(&varname.0.to_string(), sg_id)?;
2511                    }
2512
2513                    // Write all nodes.
2514                    for node_id in node_ids {
2515                        graph_write.write_node(node_id)?;
2516                    }
2517
2518                    if varname.is_some() {
2519                        graph_write.write_varname_end()?;
2520                    }
2521                }
2522
2523                if sg_id.is_some() {
2524                    graph_write.write_subgraph_end()?;
2525                }
2526            }
2527
2528            if loop_id.is_some() {
2529                graph_write.write_loop_end()?;
2530            }
2531        }
2532
2533        // Write epilogue.
2534        graph_write.write_epilogue()?;
2535
2536        Ok(())
2537    }
2538
2539    /// Convert back into surface syntax.
2540    pub fn surface_syntax_string(&self) -> String {
2541        let mut string = String::new();
2542        self.write_surface_syntax(&mut string).unwrap();
2543        string
2544    }
2545
2546    /// Convert back into surface syntax.
2547    pub fn write_surface_syntax(&self, write: &mut impl std::fmt::Write) -> std::fmt::Result {
2548        for (key, node) in self.nodes.iter() {
2549            match node {
2550                GraphNode::Operator(op) => {
2551                    writeln!(write, "_{:?} = {};", key.data(), op.to_token_stream())?;
2552                }
2553                GraphNode::Handoff {
2554                    kind: HandoffKind::Vec,
2555                    ..
2556                } => {
2557                    writeln!(write, "_{:?} = handoff();", key.data())?;
2558                }
2559                GraphNode::Handoff {
2560                    kind: HandoffKind::Singleton,
2561                    ..
2562                } => {
2563                    writeln!(write, "_{:?} = singleton();", key.data())?;
2564                }
2565                GraphNode::Handoff {
2566                    kind: HandoffKind::Optional,
2567                    ..
2568                } => {
2569                    writeln!(write, "_{:?} = optional();", key.data())?;
2570                }
2571                GraphNode::ModuleBoundary { .. } => panic!(),
2572            }
2573        }
2574        writeln!(write)?;
2575        for (e, (src_key, dst_key)) in self.graph.edges() {
2576            let (src_port, dst_port) = self.edge_ports(e);
2577            let src_port_str = if src_port.is_specified() {
2578                format!("[{}]", src_port)
2579            } else {
2580                String::new()
2581            };
2582            let dst_port_str = if dst_port.is_specified() {
2583                format!("[{}]", dst_port)
2584            } else {
2585                String::new()
2586            };
2587            writeln!(
2588                write,
2589                "_{:?}{} -> {}_{:?};",
2590                src_key.data(),
2591                src_port_str,
2592                dst_port_str,
2593                dst_key.data()
2594            )?;
2595        }
2596        Ok(())
2597    }
2598
2599    /// Convert into a [mermaid](https://mermaid-js.github.io/) graph. Ignores subgraphs.
2600    pub fn mermaid_string_flat(&self) -> String {
2601        let mut string = String::new();
2602        self.write_mermaid_flat(&mut string).unwrap();
2603        string
2604    }
2605
2606    /// Convert into a [mermaid](https://mermaid-js.github.io/) graph. Ignores subgraphs.
2607    pub fn write_mermaid_flat(&self, write: &mut impl std::fmt::Write) -> std::fmt::Result {
2608        writeln!(write, "flowchart TB")?;
2609        for (key, node) in self.nodes.iter() {
2610            match node {
2611                GraphNode::Operator(operator) => writeln!(
2612                    write,
2613                    "    %% {span}\n    {id:?}[\"{row_col} <tt>{code}</tt>\"]",
2614                    span = PrettySpan(node.span()),
2615                    id = key.data(),
2616                    row_col = PrettyRowCol(node.span()),
2617                    code = operator
2618                        .to_token_stream()
2619                        .to_string()
2620                        .replace('&', "&amp;")
2621                        .replace('<', "&lt;")
2622                        .replace('>', "&gt;")
2623                        .replace('"', "&quot;")
2624                        .replace('\n', "<br>"),
2625                ),
2626                GraphNode::Handoff {
2627                    kind: HandoffKind::Vec,
2628                    ..
2629                } => {
2630                    writeln!(write, r#"    {:?}{{"{}"}}"#, key.data(), HANDOFF_NODE_STR)
2631                }
2632                GraphNode::Handoff {
2633                    kind: HandoffKind::Singleton | HandoffKind::Optional,
2634                    ..
2635                } => {
2636                    writeln!(
2637                        write,
2638                        r#"    {:?}{{"{}"}}"#,
2639                        key.data(),
2640                        SINGLETON_SLOT_NODE_STR
2641                    )
2642                }
2643                GraphNode::ModuleBoundary { .. } => {
2644                    writeln!(
2645                        write,
2646                        r#"    {:?}{{"{}"}}"#,
2647                        key.data(),
2648                        MODULE_BOUNDARY_NODE_STR
2649                    )
2650                }
2651            }?;
2652        }
2653        writeln!(write)?;
2654        for (_e, (src_key, dst_key)) in self.graph.edges() {
2655            writeln!(write, "    {:?}-->{:?}", src_key.data(), dst_key.data())?;
2656        }
2657        Ok(())
2658    }
2659}
2660
2661/// Loops
2662impl DfirGraph {
2663    /// Iterator over all loop IDs.
2664    pub fn loop_ids(&self) -> slotmap::basic::Keys<'_, GraphLoopId, Vec<GraphNodeId>> {
2665        self.loop_nodes.keys()
2666    }
2667
2668    /// Iterator over all loops, ID and members: `(GraphLoopId, Vec<GraphNodeId>)`.
2669    pub fn loops(&self) -> slotmap::basic::Iter<'_, GraphLoopId, Vec<GraphNodeId>> {
2670        self.loop_nodes.iter()
2671    }
2672
2673    /// Get a loop's member nodes.
2674    pub fn loop_nodes(&self, loop_id: GraphLoopId) -> &[GraphNodeId] {
2675        self.loop_nodes.get(loop_id).unwrap()
2676    }
2677
2678    /// Create a new loop context, with the given parent loop (or `None`).
2679    pub fn insert_loop(&mut self, parent_loop: Option<GraphLoopId>) -> GraphLoopId {
2680        let loop_id = self.loop_nodes.insert(Vec::new());
2681        self.loop_children.insert(loop_id, Vec::new());
2682        if let Some(parent_loop) = parent_loop {
2683            self.loop_parent.insert(loop_id, parent_loop);
2684            self.loop_children
2685                .get_mut(parent_loop)
2686                .unwrap()
2687                .push(loop_id);
2688        } else {
2689            self.root_loops.push(loop_id);
2690        }
2691        loop_id
2692    }
2693
2694    /// Get a node's loop context (or `None` for root).
2695    pub fn node_loop(&self, node_id: GraphNodeId) -> Option<GraphLoopId> {
2696        self.node_loops.get(node_id).copied()
2697    }
2698
2699    /// Get a subgraph's loop context (or `None` for root).
2700    pub fn subgraph_loop(&self, subgraph_id: GraphSubgraphId) -> Option<GraphLoopId> {
2701        let &node_id = self.subgraph(subgraph_id).first().unwrap();
2702        let out = self.node_loop(node_id);
2703        debug_assert!(
2704            self.subgraph(subgraph_id)
2705                .iter()
2706                .all(|&node_id| self.node_loop(node_id) == out),
2707            "Subgraph nodes should all have the same loop context."
2708        );
2709        out
2710    }
2711
2712    /// Get a loop context's parent loop context (or `None` for root).
2713    pub fn loop_parent(&self, loop_id: GraphLoopId) -> Option<GraphLoopId> {
2714        self.loop_parent.get(loop_id).copied()
2715    }
2716
2717    /// Get a loop context's child loops.
2718    pub fn loop_children(&self, loop_id: GraphLoopId) -> &Vec<GraphLoopId> {
2719        self.loop_children.get(loop_id).unwrap()
2720    }
2721
2722    /// Get root-level loops (those with no parent loop).
2723    pub fn root_loops(&self) -> &[GraphLoopId] {
2724        &self.root_loops
2725    }
2726}
2727
2728/// Options for [`DfirGraph::as_code_with_options`].
2729#[derive(Default)]
2730#[non_exhaustive]
2731pub struct AsCodeOptions {
2732    /// Controls whether type guards are emitted in codegen. Excluding type guards may result in worse
2733    /// error messages, and may have little benefit on `--release` builds.
2734    pub exclude_type_guards: bool,
2735    /// Controls whether the runtime meta graph + diagnostics JSON blobs are baked into the generated
2736    /// `Dfir::new(...)` call.
2737    pub exclude_meta: bool,
2738    /// Controls whether metrics are tracked. Metrics tracking is opt-in: no tracking code is
2739    /// generated unless this is set. Even if metrics are tracked, they still need to be reported
2740    /// via the `Context::metrics` field.
2741    pub include_metrics_tracking: bool,
2742}
2743
2744/// Configuration for writing graphs.
2745#[derive(Clone, Debug, Default)]
2746#[cfg_attr(feature = "clap-derive", derive(clap::Args))]
2747pub struct WriteConfig {
2748    /// Subgraphs will not be rendered if set.
2749    #[cfg_attr(feature = "clap-derive", arg(long))]
2750    pub no_subgraphs: bool,
2751    /// Variable names will not be rendered if set.
2752    #[cfg_attr(feature = "clap-derive", arg(long))]
2753    pub no_varnames: bool,
2754    /// Will not render pull/push shapes if set.
2755    #[cfg_attr(feature = "clap-derive", arg(long))]
2756    pub no_pull_push: bool,
2757    /// Will not render handoffs if set.
2758    #[cfg_attr(feature = "clap-derive", arg(long))]
2759    pub no_handoffs: bool,
2760    /// Will not render singleton references if set.
2761    #[cfg_attr(feature = "clap-derive", arg(long))]
2762    pub no_references: bool,
2763    /// Will not render loops if set.
2764    #[cfg_attr(feature = "clap-derive", arg(long))]
2765    pub no_loops: bool,
2766
2767    /// Op text will only be their name instead of the whole source.
2768    #[cfg_attr(feature = "clap-derive", arg(long))]
2769    pub op_short_text: bool,
2770    /// Op text will exclude any line that starts with "use".
2771    #[cfg_attr(feature = "clap-derive", arg(long))]
2772    pub op_text_no_imports: bool,
2773}
2774
2775/// Enum for choosing between mermaid and dot graph writing.
2776#[derive(Copy, Clone, Debug)]
2777#[cfg_attr(feature = "clap-derive", derive(clap::Parser, clap::ValueEnum))]
2778pub enum WriteGraphType {
2779    /// Mermaid graphs.
2780    Mermaid,
2781    /// Dot (Graphviz) graphs.
2782    Dot,
2783}
2784
2785/// [`itertools::Itertools::into_group_map`], but for `BTreeMap`.
2786fn into_group_map<K, V>(iter: impl IntoIterator<Item = (K, V)>) -> BTreeMap<K, Vec<V>>
2787where
2788    K: Ord,
2789{
2790    let mut out: BTreeMap<_, Vec<_>> = BTreeMap::new();
2791    for (k, v) in iter {
2792        out.entry(k).or_default().push(v);
2793    }
2794    out
2795}