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