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#[derive(Clone, Debug, Serialize, Deserialize)]
32pub struct ResolvedHandoffRef {
33 pub node_id: Option<GraphNodeId>,
35 pub is_mut: bool,
37 pub access_group: Option<u32>,
39}
40
41#[derive(Default, Debug, Serialize, Deserialize)]
51pub struct DfirGraph {
52 nodes: SlotMap<GraphNodeId, GraphNode>,
54
55 #[serde(skip)]
58 operator_instances: SecondaryMap<GraphNodeId, OperatorInstance>,
59 operator_tag: SecondaryMap<GraphNodeId, String>,
61 graph: DiMulGraph<GraphNodeId, GraphEdgeId>,
63 ports: SecondaryMap<GraphEdgeId, (PortIndexValue, PortIndexValue)>,
65
66 node_loops: SecondaryMap<GraphNodeId, GraphLoopId>,
68 loop_nodes: SlotMap<GraphLoopId, Vec<GraphNodeId>>,
70 loop_parent: SparseSecondaryMap<GraphLoopId, GraphLoopId>,
72 root_loops: Vec<GraphLoopId>,
74 loop_children: SecondaryMap<GraphLoopId, Vec<GraphLoopId>>,
76
77 node_subgraph: SecondaryMap<GraphNodeId, GraphSubgraphId>,
79
80 subgraph_nodes: SlotMap<GraphSubgraphId, Vec<GraphNodeId>>,
82 subgraph_toposort: Vec<GraphSubgraphId>,
84
85 node_handoff_references: SparseSecondaryMap<GraphNodeId, Vec<ResolvedHandoffRef>>,
87 node_varnames: SparseSecondaryMap<GraphNodeId, Varname>,
89
90 handoff_delay_type: SparseSecondaryMap<GraphNodeId, DelayType>,
94}
95
96impl DfirGraph {
98 pub fn new() -> Self {
100 Default::default()
101 }
102}
103
104impl DfirGraph {
106 pub fn node(&self, node_id: GraphNodeId) -> &GraphNode {
108 self.nodes.get(node_id).expect("Node not found.")
109 }
110
111 pub fn node_op_inst(&self, node_id: GraphNodeId) -> Option<&OperatorInstance> {
116 self.operator_instances.get(node_id)
117 }
118
119 pub fn node_varname(&self, node_id: GraphNodeId) -> Option<&Varname> {
121 self.node_varnames.get(node_id)
122 }
123
124 pub fn node_subgraph(&self, node_id: GraphNodeId) -> Option<GraphSubgraphId> {
126 self.node_subgraph.get(node_id).copied()
127 }
128
129 pub fn node_degree_in(&self, node_id: GraphNodeId) -> usize {
131 self.graph.degree_in(node_id)
132 }
133
134 pub fn node_degree_out(&self, node_id: GraphNodeId) -> usize {
136 self.graph.degree_out(node_id)
137 }
138
139 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 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 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 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 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 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 pub fn node_ids(&self) -> slotmap::basic::Keys<'_, GraphNodeId, GraphNode> {
219 self.nodes.keys()
220 }
221
222 pub fn nodes(&self) -> slotmap::basic::Iter<'_, GraphNodeId, GraphNode> {
224 self.nodes.iter()
225 }
226
227 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 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 pub fn insert_node_op_insts_all(&mut self, diagnostics: &mut Diagnostics) {
257 let mut op_insts = Vec::new();
262 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 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 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 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 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 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 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 let generics = get_operator_generics(diagnostics, operator);
341 {
343 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 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 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 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 let node_id = self.nodes.insert(new_node);
452 if let Some(op_inst) = op_inst_opt {
454 self.operator_instances.insert(node_id, op_inst);
455 }
456 let (e0, e1) = self
458 .graph
459 .insert_intermediate_vertex(node_id, edge_id)
460 .unwrap();
461
462 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 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 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 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 let inn_degree = self.node_predecessor_nodes(node_id).len();
521 let out_degree = self.node_successor_nodes(node_id).len();
523
524 match (inn_degree, out_degree) {
525 (0, 0) => None, (0, 1) => Some(Color::Pull),
527 (1, 0) => Some(Color::Push),
528 (1, 1) => None, (_many, 0 | 1) => Some(Color::Pull),
530 (0 | 1, _many) => Some(Color::Push),
531 (_many, _to_many) => Some(Color::Comp),
532 }
533 }
534
535 pub fn set_operator_tag(&mut self, node_id: GraphNodeId, tag: String) {
537 self.operator_tag.insert(node_id, tag);
538 }
539}
540
541impl DfirGraph {
543 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 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 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
587pub type NodeHandoffReferenceGroups<'a> =
590 BTreeMap<GraphNodeId, BTreeMap<Option<u32>, Vec<(GraphNodeId, &'a ResolvedHandoffRef, Span)>>>;
591
592impl DfirGraph {
594 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 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 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
687impl DfirGraph {
689 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 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 pub fn edge_ids(&self) -> slotmap::basic::Keys<'_, GraphEdgeId, (GraphNodeId, GraphNodeId)> {
703 self.graph.edge_ids()
704 }
705
706 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 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 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
737impl DfirGraph {
739 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 pub fn subgraph_ids(&self) -> slotmap::basic::Keys<'_, GraphSubgraphId, Vec<GraphNodeId>> {
748 self.subgraph_nodes.keys()
749 }
750
751 pub fn subgraph_toposort(&self) -> &[GraphSubgraphId] {
753 &self.subgraph_toposort
754 }
755
756 pub fn set_subgraph_toposort(&mut self, order: Vec<GraphSubgraphId>) {
758 self.subgraph_toposort = order;
759 }
760
761 pub fn subgraphs(&self) -> slotmap::basic::Iter<'_, GraphSubgraphId, Vec<GraphNodeId>> {
763 self.subgraph_nodes.iter()
764 }
765
766 pub fn insert_subgraph(
768 &mut self,
769 node_ids: Vec<GraphNodeId>,
770 ) -> Result<GraphSubgraphId, (GraphNodeId, GraphSubgraphId)> {
771 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 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 pub fn handoff_delay_type(&self, node_id: GraphNodeId) -> Option<DelayType> {
799 self.handoff_delay_type.get(node_id).copied()
800 }
801
802 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 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
819impl DfirGraph {
821 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 fn hoff_buf_ident(&self, hoff_id: GraphNodeId, span: Span) -> Ident {
854 Ident::new(&format!("hoff_{:?}_buf", hoff_id.data()), span)
855 }
856
857 fn hoff_back_ident(&self, hoff_id: GraphNodeId, span: Span) -> Ident {
859 Ident::new(&format!("hoff_{:?}_back", hoff_id.data()), span)
860 }
861
862 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 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 fn helper_collect_subgraph_handoffs(
913 &self,
914 ) -> SecondaryMap<GraphSubgraphId, (Vec<GraphNodeId>, Vec<GraphNodeId>)> {
915 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 (hoff_id, hoff) in self.nodes() {
927 if !matches!(hoff, GraphNode::Handoff { .. }) {
928 continue;
929 }
930 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 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 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 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 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 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 let swap_code = loop_swap_code
1014 .get(&loop_id)
1015 .map(|v| v.as_slice())
1016 .unwrap_or(&[]);
1017
1018 let is_root_loop = self.loop_parent(loop_id).is_none();
1020
1021 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 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 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 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 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 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 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 output.extend(child_body);
1120 output.extend(quote! { #( #swap_code )* });
1121 } else if gate_checks.is_empty() {
1122 output.extend(child_body);
1124 output.extend(quote! { #( #swap_code )* });
1125 } else if is_root_loop {
1126 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 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 fn helper_loop_input_handoffs(&self) -> SecondaryMap<GraphLoopId, Vec<GraphNodeId>> {
1148 let mut loop_hoffs_inn = SecondaryMap::<GraphLoopId, Vec<GraphNodeId>>::new();
1149
1150 for (hoff_id, hoff) in self.nodes() {
1152 if !matches!(hoff, GraphNode::Handoff { .. }) {
1153 continue;
1154 }
1155
1156 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 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 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 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 let bump_ident = Ident::new("__dfir_bump", Span::call_site());
1234
1235 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 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 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 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 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 self.loop_parent(loop_id).is_some()
1301 } else {
1302 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 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 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 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 let subgraph_handoffs = self.helper_collect_subgraph_handoffs();
1357
1358 let all_subgraphs: Vec<_> = self
1360 .subgraph_toposort()
1361 .iter()
1362 .map(|&sg_id| (sg_id, self.subgraph(sg_id)))
1363 .collect();
1364
1365 let mut op_prologue_code = Vec::new();
1369 let mut op_tick_end_code = Vec::new();
1370
1371 let mut loop_stack: Vec<(GraphLoopId, TokenStream)> = Vec::new();
1375 let mut current_output = TokenStream::new();
1376
1377 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 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 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 if let Some(target_loop) = sg_loop
1410 && loop_stack.last().map(|&(l, _)| l) != Some(target_loop)
1411 {
1412 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 for &loop_id in path.iter().rev() {
1425 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 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 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 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 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 let work_done = Ident::new("__dfir_work_done", Span::call_site());
1509 let metrics = Ident::new("__dfir_metrics", Span::call_site());
1510
1511 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 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 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 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 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 let #port_ident = #root::dfir_pipes::push::for_each(|item| { #buf_ident.push(item); });
1587 }
1588 }
1589 }
1590 })
1591 .collect();
1592
1593 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 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 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 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 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 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 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 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 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 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 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 let sg_fut_ident = subgraph_id.as_ident(Span::call_site());
1957
1958 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 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 Some(quote_spanned! {span=>
1997 #buf_ident.clear();
1998 })
1999 } else {
2000 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 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 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 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 #( #send_metrics_code )*
2054 }
2055 } else {
2056 quote! {
2057 #sg_fut_ident.await;
2058 }
2059 };
2060
2061 let sg_block = quote! {
2063 #( #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 #( #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 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; 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 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 let back_buffer_idents = back_buffer_idents_laziness
2156 .iter()
2157 .map(|(back_ident, _, _)| back_ident);
2158 let defer_tick_buf_idents = back_buffer_idents_laziness
2160 .iter()
2161 .map(|(_, buf_ident, _)| buf_ident);
2162 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 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 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); }
2190 }
2191 Some(&entry.1) })
2193 .collect();
2194
2195 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 #( let mut #back_buffer_idents = ::std::vec::Vec::new(); )*
2225 #( let mut #defer_tick_buf_idents = ::std::vec::Vec::new(); )*
2226
2227 let mut #bump_ident = #root::bumpalo::Bump::new();
2229
2230 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 #bump_ident.reset();
2239
2240 {
2241 let __dfir_metrics = #df.metrics();
2242
2243 #gated_subgraph_code
2244
2245 #[allow(clippy::nonminimal_bool, reason = "codegen")]
2248 if false #( || !#non_lazy_schedule_idents.is_empty() )* {
2249 #df.schedule_subgraph(true);
2250 }
2251
2252 #( #back_edge_swap_code )*
2255 }
2256
2257 #( #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 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 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 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 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 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 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 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 let node_color_map = self.node_color_map();
2367
2368 graph_write.write_prologue()?;
2370
2371 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 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 for (edge_id, (src_id, mut dst_id)) in self.edges() {
2407 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 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 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 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 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 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 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 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 graph_write.write_epilogue()?;
2532
2533 Ok(())
2534 }
2535
2536 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 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 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 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('&', "&")
2618 .replace('<', "<")
2619 .replace('>', ">")
2620 .replace('"', """)
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
2658impl DfirGraph {
2660 pub fn loop_ids(&self) -> slotmap::basic::Keys<'_, GraphLoopId, Vec<GraphNodeId>> {
2662 self.loop_nodes.keys()
2663 }
2664
2665 pub fn loops(&self) -> slotmap::basic::Iter<'_, GraphLoopId, Vec<GraphNodeId>> {
2667 self.loop_nodes.iter()
2668 }
2669
2670 pub fn loop_nodes(&self, loop_id: GraphLoopId) -> &[GraphNodeId] {
2672 self.loop_nodes.get(loop_id).unwrap()
2673 }
2674
2675 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 pub fn node_loop(&self, node_id: GraphNodeId) -> Option<GraphLoopId> {
2693 self.node_loops.get(node_id).copied()
2694 }
2695
2696 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 pub fn loop_parent(&self, loop_id: GraphLoopId) -> Option<GraphLoopId> {
2711 self.loop_parent.get(loop_id).copied()
2712 }
2713
2714 pub fn loop_children(&self, loop_id: GraphLoopId) -> &Vec<GraphLoopId> {
2716 self.loop_children.get(loop_id).unwrap()
2717 }
2718
2719 pub fn root_loops(&self) -> &[GraphLoopId] {
2721 &self.root_loops
2722 }
2723}
2724
2725#[derive(Default)]
2727#[non_exhaustive]
2728pub struct AsCodeOptions {
2729 pub exclude_type_guards: bool,
2732 pub exclude_meta: bool,
2735 pub include_metrics_tracking: bool,
2739}
2740
2741#[derive(Clone, Debug, Default)]
2743#[cfg_attr(feature = "clap-derive", derive(clap::Args))]
2744pub struct WriteConfig {
2745 #[cfg_attr(feature = "clap-derive", arg(long))]
2747 pub no_subgraphs: bool,
2748 #[cfg_attr(feature = "clap-derive", arg(long))]
2750 pub no_varnames: bool,
2751 #[cfg_attr(feature = "clap-derive", arg(long))]
2753 pub no_pull_push: bool,
2754 #[cfg_attr(feature = "clap-derive", arg(long))]
2756 pub no_handoffs: bool,
2757 #[cfg_attr(feature = "clap-derive", arg(long))]
2759 pub no_references: bool,
2760 #[cfg_attr(feature = "clap-derive", arg(long))]
2762 pub no_loops: bool,
2763
2764 #[cfg_attr(feature = "clap-derive", arg(long))]
2766 pub op_short_text: bool,
2767 #[cfg_attr(feature = "clap-derive", arg(long))]
2769 pub op_text_no_imports: bool,
2770}
2771
2772#[derive(Copy, Clone, Debug)]
2774#[cfg_attr(feature = "clap-derive", derive(clap::Parser, clap::ValueEnum))]
2775pub enum WriteGraphType {
2776 Mermaid,
2778 Dot,
2780}
2781
2782fn 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}