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> {
606 let mod_bound_nodes = self
607 .nodes()
608 .filter(|(_nid, node)| matches!(node, GraphNode::ModuleBoundary { .. }))
609 .map(|(nid, _node)| nid)
610 .collect::<Vec<_>>();
611
612 for mod_bound_node in mod_bound_nodes {
613 self.remove_module_boundary(mod_bound_node)?;
614 }
615
616 Ok(())
617 }
618
619 fn remove_module_boundary(&mut self, mod_bound_node: GraphNodeId) -> Result<(), Diagnostic> {
623 assert!(
624 self.node_subgraph.is_empty() && self.subgraph_nodes.is_empty(),
625 "Should not remove intermediate node after subgraph partitioning"
626 );
627
628 let mut mod_pred_ports = BTreeMap::new();
629 let mut mod_succ_ports = BTreeMap::new();
630
631 for mod_out_edge in self.node_predecessor_edges(mod_bound_node) {
632 let (pred_port, succ_port) = self.edge_ports(mod_out_edge);
633 mod_pred_ports.insert(succ_port.clone(), (mod_out_edge, pred_port.clone()));
634 }
635
636 for mod_inn_edge in self.node_successor_edges(mod_bound_node) {
637 let (pred_port, succ_port) = self.edge_ports(mod_inn_edge);
638 mod_succ_ports.insert(pred_port.clone(), (mod_inn_edge, succ_port.clone()));
639 }
640
641 if mod_pred_ports.keys().collect::<BTreeSet<_>>()
642 != mod_succ_ports.keys().collect::<BTreeSet<_>>()
643 {
644 let GraphNode::ModuleBoundary { input, import_expr } = self.node(mod_bound_node) else {
646 panic!();
647 };
648
649 if *input {
650 return Err(Diagnostic {
651 span: *import_expr,
652 level: Level::Error,
653 message: format!(
654 "The ports into the module did not match. input: {:?}, expected: {:?}",
655 mod_pred_ports.keys().map(|x| x.to_string()).join(", "),
656 mod_succ_ports.keys().map(|x| x.to_string()).join(", ")
657 ),
658 });
659 }
660 return Err(Diagnostic {
661 span: *import_expr,
662 level: Level::Error,
663 message: format!(
664 "The ports out of the module did not match. output: {:?}, expected: {:?}",
665 mod_succ_ports.keys().map(|x| x.to_string()).join(", "),
666 mod_pred_ports.keys().map(|x| x.to_string()).join(", "),
667 ),
668 });
669 }
670
671 for (port, (pred_edge, pred_port)) in mod_pred_ports {
672 let (succ_edge, succ_port) = mod_succ_ports.remove(&port).unwrap();
673
674 let (src, _) = self.edge(pred_edge);
675 let (_, dst) = self.edge(succ_edge);
676 self.remove_edge(pred_edge);
677 self.remove_edge(succ_edge);
678
679 let new_edge_id = self.graph.insert_edge(src, dst);
680 self.ports.insert(new_edge_id, (pred_port, succ_port));
681 }
682
683 self.graph.remove_vertex(mod_bound_node);
684 self.nodes.remove(mod_bound_node);
685
686 Ok(())
687 }
688}
689
690impl DfirGraph {
692 pub fn edge(&self, edge_id: GraphEdgeId) -> (GraphNodeId, GraphNodeId) {
694 let (src, dst) = self.graph.edge(edge_id).expect("Edge not found.");
695 (src, dst)
696 }
697
698 pub fn edge_ports(&self, edge_id: GraphEdgeId) -> (&PortIndexValue, &PortIndexValue) {
700 let (src_port, dst_port) = self.ports.get(edge_id).expect("Edge not found.");
701 (src_port, dst_port)
702 }
703
704 pub fn edge_ids(&self) -> slotmap::basic::Keys<'_, GraphEdgeId, (GraphNodeId, GraphNodeId)> {
706 self.graph.edge_ids()
707 }
708
709 pub fn edges(
711 &self,
712 ) -> impl '_
713 + ExactSizeIterator<Item = (GraphEdgeId, (GraphNodeId, GraphNodeId))>
714 + FusedIterator
715 + Clone
716 + Debug {
717 self.graph.edges()
718 }
719
720 pub fn insert_edge(
722 &mut self,
723 src: GraphNodeId,
724 src_port: PortIndexValue,
725 dst: GraphNodeId,
726 dst_port: PortIndexValue,
727 ) -> GraphEdgeId {
728 let edge_id = self.graph.insert_edge(src, dst);
729 self.ports.insert(edge_id, (src_port, dst_port));
730 edge_id
731 }
732
733 pub fn remove_edge(&mut self, edge: GraphEdgeId) {
735 let (_src, _dst) = self.graph.remove_edge(edge).unwrap();
736 let (_src_port, _dst_port) = self.ports.remove(edge).unwrap();
737 }
738}
739
740impl DfirGraph {
742 pub fn subgraph(&self, subgraph_id: GraphSubgraphId) -> &Vec<GraphNodeId> {
744 self.subgraph_nodes
745 .get(subgraph_id)
746 .expect("Subgraph not found.")
747 }
748
749 pub fn subgraph_ids(&self) -> slotmap::basic::Keys<'_, GraphSubgraphId, Vec<GraphNodeId>> {
751 self.subgraph_nodes.keys()
752 }
753
754 pub fn subgraph_toposort(&self) -> &[GraphSubgraphId] {
756 &self.subgraph_toposort
757 }
758
759 pub fn set_subgraph_toposort(&mut self, order: Vec<GraphSubgraphId>) {
761 self.subgraph_toposort = order;
762 }
763
764 pub fn subgraphs(&self) -> slotmap::basic::Iter<'_, GraphSubgraphId, Vec<GraphNodeId>> {
766 self.subgraph_nodes.iter()
767 }
768
769 pub fn insert_subgraph(
771 &mut self,
772 node_ids: Vec<GraphNodeId>,
773 ) -> Result<GraphSubgraphId, (GraphNodeId, GraphSubgraphId)> {
774 for &node_id in node_ids.iter() {
776 if let Some(&old_sg_id) = self.node_subgraph.get(node_id) {
777 return Err((node_id, old_sg_id));
778 }
779 }
780 let subgraph_id = self.subgraph_nodes.insert_with_key(|sg_id| {
781 for &node_id in node_ids.iter() {
782 self.node_subgraph.insert(node_id, sg_id);
783 }
784 node_ids
785 });
786
787 Ok(subgraph_id)
788 }
789
790 pub fn remove_from_subgraph(&mut self, node_id: GraphNodeId) -> bool {
792 if let Some(old_sg_id) = self.node_subgraph.remove(node_id) {
793 self.subgraph_nodes[old_sg_id].retain(|&other_node_id| other_node_id != node_id);
794 true
795 } else {
796 false
797 }
798 }
799
800 pub fn handoff_delay_type(&self, node_id: GraphNodeId) -> Option<DelayType> {
802 self.handoff_delay_type.get(node_id).copied()
803 }
804
805 pub fn set_handoff_delay_type(&mut self, node_id: GraphNodeId, delay_type: DelayType) {
807 self.handoff_delay_type.insert(node_id, delay_type);
808 }
809
810 fn find_pull_to_push_idx(&self, subgraph_nodes: &[GraphNodeId]) -> usize {
812 subgraph_nodes
813 .iter()
814 .position(|&node_id| {
815 self.node_color(node_id)
816 .is_some_and(|color| Color::Pull != color)
817 })
818 .unwrap_or(subgraph_nodes.len())
819 }
820}
821
822impl DfirGraph {
824 fn node_as_ident(&self, node_id: GraphNodeId, is_pred: bool) -> Ident {
826 let name = match &self.nodes[node_id] {
827 GraphNode::Operator(_) => format!("op_{:?}", node_id.data()),
828 GraphNode::Handoff {
829 kind: HandoffKind::Vec,
830 ..
831 } => format!(
832 "hoff_{:?}_{}",
833 node_id.data(),
834 if is_pred { "recv" } else { "send" }
835 ),
836 GraphNode::Handoff {
837 kind: HandoffKind::Singleton | HandoffKind::Optional,
838 ..
839 } => format!(
840 "singleton_{:?}_{}",
841 node_id.data(),
842 if is_pred { "recv" } else { "send" }
843 ),
844 GraphNode::ModuleBoundary { .. } => panic!(),
845 };
846 let span = match (is_pred, &self.nodes[node_id]) {
847 (_, GraphNode::Operator(operator)) => operator.span(),
848 (true, &GraphNode::Handoff { src_span, .. }) => src_span,
849 (false, &GraphNode::Handoff { dst_span, .. }) => dst_span,
850 (_, GraphNode::ModuleBoundary { .. }) => panic!(),
851 };
852 Ident::new(&name, span)
853 }
854
855 fn hoff_buf_ident(&self, hoff_id: GraphNodeId, span: Span) -> Ident {
857 Ident::new(&format!("hoff_{:?}_buf", hoff_id.data()), span)
858 }
859
860 fn hoff_back_ident(&self, hoff_id: GraphNodeId, span: Span) -> Ident {
862 Ident::new(&format!("hoff_{:?}_back", hoff_id.data()), span)
863 }
864
865 fn helper_resolve_singletons(&self, node_id: GraphNodeId, span: Span) -> Vec<TokenStream> {
874 self.node_handoff_references(node_id)
875 .iter()
876 .map(|resolved_ref| {
877 let ref_node_id = resolved_ref
879 .node_id
880 .expect("Expected singleton to be resolved but was not, this is a bug.");
881 let is_mut = resolved_ref.is_mut;
882 match self.node(ref_node_id) {
883 GraphNode::Handoff {
884 kind: HandoffKind::Singleton,
885 ..
886 } => {
887 let buf_ident = self.hoff_buf_ident(ref_node_id, span);
888 if is_mut {
889 quote_spanned! {span=> #buf_ident.as_mut().unwrap() }
890 } else {
891 quote_spanned! {span=> #buf_ident.as_ref().unwrap() }
892 }
893 }
894 GraphNode::Handoff {
895 kind: HandoffKind::Optional | HandoffKind::Vec,
896 ..
897 } => {
898 let buf_ident = self.hoff_buf_ident(ref_node_id, span);
899 if is_mut {
900 quote_spanned! {span=> &mut #buf_ident }
901 } else {
902 quote_spanned! {span=> &#buf_ident }
903 }
904 }
905 _ => {
906 unreachable!("Only handoff nodes should be reachable as handoff references")
907 }
908 }
909 })
910 .collect::<Vec<_>>()
911 }
912
913 fn helper_collect_subgraph_handoffs(
916 &self,
917 ) -> SecondaryMap<GraphSubgraphId, (Vec<GraphNodeId>, Vec<GraphNodeId>)> {
918 let mut subgraph_handoffs: SecondaryMap<
920 GraphSubgraphId,
921 (Vec<GraphNodeId>, Vec<GraphNodeId>),
922 > = self
923 .subgraph_nodes
924 .keys()
925 .map(|k| (k, Default::default()))
926 .collect();
927
928 for (hoff_id, hoff) in self.nodes() {
930 if !matches!(hoff, GraphNode::Handoff { .. }) {
931 continue;
932 }
933 for (_edge, succ_id) in self.node_successors(hoff_id) {
935 let succ_sg = self
936 .node_subgraph(succ_id)
937 .expect("bug: successor not in subgraph, may be a doubled/adjacent handoff");
938 subgraph_handoffs[succ_sg].0.push(hoff_id);
939 }
940 for (_edge, pred_id) in self.node_predecessors(hoff_id) {
942 let pred_sg = self
943 .node_subgraph(pred_id)
944 .expect("bug: predecessor not in subgraph, may be a doubled/adjacent handoff");
945 subgraph_handoffs[pred_sg].1.push(hoff_id);
946 }
947 }
948
949 subgraph_handoffs
950 }
951
952 fn helper_loop_output_handoffs(&self) -> SecondaryMap<GraphLoopId, Vec<GraphNodeId>> {
955 let mut loop_hoffs_out = SecondaryMap::<GraphLoopId, Vec<GraphNodeId>>::new();
956
957 for (hoff_id, hoff) in self.nodes() {
958 if !matches!(hoff, GraphNode::Handoff { .. }) {
959 continue;
960 }
961
962 let loop_pred = self
963 .node_predecessors(hoff_id)
964 .next()
965 .and_then(|(_, pred)| self.node_loop(pred));
966 let loop_succ = self
967 .node_successors(hoff_id)
968 .next()
969 .and_then(|(_, succ)| self.node_loop(succ));
970
971 if let Some(loop_pred) = loop_pred
972 && loop_succ == self.loop_parent(loop_pred)
973 {
974 loop_hoffs_out
976 .entry(loop_pred)
977 .expect("loop removed")
978 .or_default()
979 .push(hoff_id);
980 }
981 }
982
983 loop_hoffs_out
984 }
985
986 fn is_inside_loop(&self, node_loop: Option<GraphLoopId>, loop_id: GraphLoopId) -> bool {
988 let mut current = node_loop;
989 while let Some(l) = current {
990 if l == loop_id {
991 return true;
992 }
993 current = self.loop_parent(l);
994 }
995 false
996 }
997
998 fn emit_loop_gate(
1007 &self,
1008 loop_id: GraphLoopId,
1009 child_body: TokenStream,
1010 loop_input_handoffs: &SecondaryMap<GraphLoopId, Vec<GraphNodeId>>,
1011 back_edge_hoffs_and_lazyness: &SparseSecondaryMap<GraphNodeId, bool>,
1012 loop_swap_code: &std::collections::HashMap<GraphLoopId, Vec<TokenStream>>,
1013 output: &mut TokenStream,
1014 ) {
1015 let swap_code = loop_swap_code
1017 .get(&loop_id)
1018 .map(|v| v.as_slice())
1019 .unwrap_or(&[]);
1020
1021 let is_root_loop = self.loop_parent(loop_id).is_none();
1023
1024 let entry_handoffs = loop_input_handoffs.get(loop_id).expect("loop missing");
1026 let mut gate_checks: Vec<TokenStream> = entry_handoffs
1027 .iter()
1028 .filter(|&&hoff_id| {
1029 let is_lazy = self
1032 .node_successors(hoff_id)
1033 .next()
1034 .and_then(|(_, succ)| self.node_op_inst(succ))
1035 .is_some_and(|op_inst| {
1036 op_inst.op_constraints.flo_type == Some(FloType::WindowingLazy)
1037 });
1038 !is_lazy
1039 })
1040 .map(|&hoff_id| {
1041 let span = self.node(hoff_id).span();
1042 let buf_ident = self.hoff_buf_ident(hoff_id, span);
1043 if back_edge_hoffs_and_lazyness.contains_key(hoff_id) {
1044 let back_ident = self.hoff_back_ident(hoff_id, span);
1045 quote_spanned! {span=> !#back_ident.is_empty() }
1046 } else {
1047 quote_spanned! {span=> !#buf_ident.is_empty() }
1048 }
1049 })
1050 .collect();
1051
1052 if !is_root_loop {
1054 for (hoff_id, hoff) in self.nodes() {
1055 if !matches!(hoff, GraphNode::Handoff { .. }) {
1056 continue;
1057 }
1058 let Some(delay_type) = self.handoff_delay_type(hoff_id) else {
1059 continue;
1060 };
1061 if delay_type != DelayType::Loop {
1062 continue;
1063 }
1064 let hoff_loop = self
1066 .node_successors(hoff_id)
1067 .next()
1068 .and_then(|(_, succ)| self.node_subgraph(succ))
1069 .and_then(|sg| self.subgraph_loop(sg));
1070 if hoff_loop != Some(loop_id) {
1071 continue;
1072 }
1073 let span = self.node(hoff_id).span();
1074 let back_ident = self.hoff_back_ident(hoff_id, span);
1075 gate_checks.push(quote_spanned! {span=> !#back_ident.is_empty() });
1076 }
1077 }
1078
1079 if is_root_loop {
1082 for (hoff_id, hoff) in self.nodes() {
1083 if !matches!(hoff, GraphNode::Handoff { .. }) {
1084 continue;
1085 }
1086 let Some(delay_type) = self.handoff_delay_type(hoff_id) else {
1087 continue;
1088 };
1089 if delay_type != DelayType::Tick {
1090 continue;
1091 }
1092 let hoff_loop = self
1094 .node_successors(hoff_id)
1095 .next()
1096 .and_then(|(_, succ)| self.node_subgraph(succ))
1097 .and_then(|sg| self.subgraph_loop(sg));
1098 if hoff_loop != Some(loop_id) {
1099 continue;
1100 }
1101 let span = self.node(hoff_id).span();
1102 let back_ident = self.hoff_back_ident(hoff_id, span);
1103 gate_checks.push(quote_spanned! {span=> !#back_ident.is_empty() });
1104 }
1105 }
1106
1107 let has_eager = entry_handoffs.iter().any(|&hoff_id| {
1112 self.node_successors(hoff_id)
1113 .next()
1114 .and_then(|(_, succ)| self.node_op_inst(succ))
1115 .is_some_and(|op_inst| {
1116 op_inst.op_constraints.flo_type == Some(FloType::WindowingEager)
1117 })
1118 });
1119
1120 if has_eager && is_root_loop {
1121 output.extend(child_body);
1123 output.extend(quote! { #( #swap_code )* });
1124 } else if gate_checks.is_empty() {
1125 output.extend(child_body);
1127 output.extend(quote! { #( #swap_code )* });
1128 } else if is_root_loop {
1129 output.extend(quote! {
1131 #[allow(clippy::nonminimal_bool, reason = "codegen")]
1132 if false #( || #gate_checks )* {
1133 #child_body
1134 #( #swap_code )*
1135 }
1136 });
1137 } else {
1138 output.extend(quote! {
1140 #[allow(clippy::nonminimal_bool, reason = "codegen")]
1141 while false #( || #gate_checks )* {
1142 #child_body
1143 #( #swap_code )*
1144 }
1145 });
1146 }
1147 }
1148
1149 fn helper_loop_input_handoffs(&self) -> SecondaryMap<GraphLoopId, Vec<GraphNodeId>> {
1151 let mut loop_hoffs_inn = SecondaryMap::<GraphLoopId, Vec<GraphNodeId>>::new();
1152
1153 for (hoff_id, hoff) in self.nodes() {
1155 if !matches!(hoff, GraphNode::Handoff { .. }) {
1156 continue;
1157 }
1158
1159 let loop_pred = self
1161 .node_predecessors(hoff_id)
1162 .next()
1163 .and_then(|(_, pred)| self.node_loop(pred));
1164 let loop_succ = self
1165 .node_successors(hoff_id)
1166 .next()
1167 .and_then(|(_, succ)| self.node_loop(succ));
1168
1169 if let Some(loop_succ) = loop_succ
1170 && loop_pred == self.loop_parent(loop_succ)
1171 {
1172 loop_hoffs_inn
1174 .entry(loop_succ)
1175 .expect("loop removed")
1176 .or_default()
1177 .push(hoff_id);
1178 }
1179 }
1180
1181 loop_hoffs_inn
1182 }
1183
1184 pub fn as_code(
1203 &self,
1204 root: &TokenStream,
1205 include_type_guards: bool,
1206 prefix: TokenStream,
1207 diagnostics: &mut Diagnostics,
1208 ) -> Result<TokenStream, Diagnostics> {
1209 self.as_code_with_options(
1210 root,
1211 &AsCodeOptions {
1212 exclude_type_guards: !include_type_guards,
1213 ..Default::default()
1214 },
1215 prefix,
1216 diagnostics,
1217 )
1218 }
1219
1220 pub fn as_code_with_options(
1227 &self,
1228 root: &TokenStream,
1229 options: &AsCodeOptions,
1230 prefix: TokenStream,
1231 diagnostics: &mut Diagnostics,
1232 ) -> Result<TokenStream, Diagnostics> {
1233 let df = Ident::new(GRAPH, Span::call_site());
1234 let context = Ident::new(CONTEXT, Span::call_site());
1235 let bump_ident = Ident::new("__dfir_bump", Span::call_site());
1237
1238 let handoff_nodes = self
1240 .nodes
1241 .iter()
1242 .filter_map(|(node_id, node)| match node {
1243 &GraphNode::Handoff {
1244 kind,
1245 src_span,
1246 dst_span,
1247 } => Some((node_id, kind, (src_span, dst_span))),
1248 GraphNode::Operator(_) => None,
1249 GraphNode::ModuleBoundary { .. } => panic!(),
1250 })
1251 .collect::<Vec<_>>();
1252
1253 let back_edge_hoffs_and_lazyness = handoff_nodes
1257 .iter()
1258 .map(|&(node_id, _, _)| node_id)
1259 .filter_map(|node_id| {
1260 let delay_type = self.handoff_delay_type(node_id)?;
1261 Some((
1262 node_id,
1263 matches!(delay_type, DelayType::TickLazy | DelayType::LoopLazy),
1264 ))
1265 })
1266 .collect::<SparseSecondaryMap<_, _>>();
1267
1268 let back_buffer_idents_laziness = handoff_nodes
1270 .iter()
1271 .filter_map(|&(hoff_id, _kind, (src_span, dst_span))| {
1272 back_edge_hoffs_and_lazyness.get(hoff_id).map(|&is_lazy| {
1273 let span = src_span.join(dst_span).unwrap_or(src_span);
1274 let back_ident = self.hoff_back_ident(hoff_id, span);
1275 let buf_ident = self.hoff_buf_ident(hoff_id, span);
1276 (back_ident, buf_ident, is_lazy)
1277 })
1278 })
1279 .collect::<Vec<_>>();
1280
1281 let back_edge_swap_code = handoff_nodes
1288 .iter()
1289 .filter(|&&(node_id, _kind, _)| {
1290 self.handoff_delay_type(node_id)
1291 .is_some_and(|dt| matches!(dt, DelayType::Tick | DelayType::TickLazy))
1292 })
1293 .filter(|&&(hoff_id, _kind, _)| {
1294 let consumer_loop = self
1297 .node_successors(hoff_id)
1298 .next()
1299 .and_then(|(_, succ)| self.node_subgraph(succ))
1300 .and_then(|sg| self.subgraph_loop(sg));
1301 if let Some(loop_id) = consumer_loop {
1302 self.loop_parent(loop_id).is_some()
1304 } else {
1305 true
1307 }
1308 })
1309 .map(|&(hoff_id, _kind, _)| {
1310 let span = self.nodes[hoff_id].span();
1311 let buf_ident = self.hoff_buf_ident(hoff_id, span);
1312 let back_ident = self.hoff_back_ident(hoff_id, span);
1313 quote_spanned! {span=>
1314 ::std::mem::swap(&mut #buf_ident, &mut #back_ident);
1315 }
1316 })
1317 .collect::<Vec<_>>();
1318
1319 let mut loop_swap_code: std::collections::HashMap<GraphLoopId, Vec<TokenStream>> =
1323 std::collections::HashMap::new();
1324 for &(hoff_id, _kind, _) in handoff_nodes.iter() {
1325 let Some(delay_type) = self.handoff_delay_type(hoff_id) else {
1326 continue;
1327 };
1328 let loop_id = self
1330 .node_successors(hoff_id)
1331 .next()
1332 .and_then(|(_, succ)| self.node_subgraph(succ))
1333 .and_then(|sg| self.subgraph_loop(sg));
1334 let Some(loop_id) = loop_id else {
1335 continue;
1336 };
1337 let include = match delay_type {
1338 DelayType::Loop | DelayType::LoopLazy => true,
1339 DelayType::Tick | DelayType::TickLazy => {
1340 self.loop_parent(loop_id).is_none()
1342 }
1343 };
1344 if !include {
1345 continue;
1346 }
1347 let span = self.nodes[hoff_id].span();
1348 let buf_ident = self.hoff_buf_ident(hoff_id, span);
1349 let back_ident = self.hoff_back_ident(hoff_id, span);
1350 loop_swap_code
1351 .entry(loop_id)
1352 .or_default()
1353 .push(quote_spanned! {span=>
1354 ::std::mem::swap(&mut #buf_ident, &mut #back_ident);
1355 });
1356 }
1357
1358 let subgraph_handoffs = self.helper_collect_subgraph_handoffs();
1360
1361 let all_subgraphs: Vec<_> = self
1363 .subgraph_toposort()
1364 .iter()
1365 .map(|&sg_id| (sg_id, self.subgraph(sg_id)))
1366 .collect();
1367
1368 let mut op_prologue_code = Vec::new();
1372 let mut op_tick_end_code = Vec::new();
1373
1374 let mut loop_stack: Vec<(GraphLoopId, TokenStream)> = Vec::new();
1378 let mut current_output = TokenStream::new();
1379
1380 let loop_input_handoffs = self.helper_loop_input_handoffs();
1382 let loop_output_handoffs = self.helper_loop_output_handoffs();
1383
1384 {
1385 for &(subgraph_id, subgraph_nodes) in all_subgraphs.iter() {
1386 let sg_loop = self.subgraph_loop(subgraph_id);
1387
1388 while let Some(&(top_loop, _)) = loop_stack.last() {
1391 if sg_loop == Some(top_loop) || self.is_inside_loop(sg_loop, top_loop) {
1392 break;
1393 }
1394 let (closed_loop, child_body) = loop_stack.pop().unwrap();
1396 let target = if let Some((_, parent_body)) = loop_stack.last_mut() {
1397 parent_body
1398 } else {
1399 &mut current_output
1400 };
1401 self.emit_loop_gate(
1402 closed_loop,
1403 child_body,
1404 &loop_input_handoffs,
1405 &back_edge_hoffs_and_lazyness,
1406 &loop_swap_code,
1407 target,
1408 );
1409 }
1410
1411 if let Some(target_loop) = sg_loop
1413 && loop_stack.last().map(|&(l, _)| l) != Some(target_loop)
1414 {
1415 let mut path = Vec::new();
1417 let mut cur = Some(target_loop);
1418 while let Some(l) = cur {
1419 if loop_stack.last().map(|&(top, _)| top) == Some(l) {
1420 break;
1421 }
1422 path.push(l);
1423 cur = self.loop_parent(l);
1424 }
1425 for &loop_id in path.iter().rev() {
1428 if let Some(exit_hoffs) = loop_output_handoffs.get(loop_id) {
1430 let exit_hoff_decls = exit_hoffs.iter().map(|&hoff_id| {
1431 let span = self.nodes[hoff_id].span();
1432 let buf_ident = self.hoff_buf_ident(hoff_id, span);
1433 let GraphNode::Handoff { kind, .. } = self.node(hoff_id) else {
1434 panic!()
1435 };
1436 match kind {
1437 HandoffKind::Vec => quote_spanned! {span=>
1438 let mut #buf_ident = #root::bumpalo::collections::Vec::new_in(&#bump_ident);
1439 },
1440 HandoffKind::Singleton | HandoffKind::Optional => quote_spanned! {span=>
1441 let mut #buf_ident = ::std::option::Option::None;
1442 },
1443 }
1444 });
1445 let target = if let Some((_, body)) = loop_stack.last_mut() {
1446 body
1447 } else {
1448 &mut current_output
1449 };
1450 target.extend(quote! { #( #exit_hoff_decls )* });
1451 }
1452 loop_stack.push((loop_id, TokenStream::new()));
1453 }
1454 }
1455 let sg_metrics_ffi = subgraph_id.data().as_ffi();
1456 let (recv_hoffs, send_hoffs) = &subgraph_handoffs[subgraph_id];
1457
1458 let recv_port_idents: Vec<Ident> = recv_hoffs
1460 .iter()
1461 .map(|&hoff_id| self.node_as_ident(hoff_id, true))
1462 .collect();
1463 let send_port_idents: Vec<Ident> = send_hoffs
1464 .iter()
1465 .map(|&hoff_id| self.node_as_ident(hoff_id, false))
1466 .collect();
1467
1468 let recv_buf_idents: Vec<Ident> = recv_hoffs
1470 .iter()
1471 .map(|&hoff_id| self.hoff_buf_ident(hoff_id, self.nodes[hoff_id].span()))
1472 .collect();
1473 let send_buf_idents: Vec<Ident> = send_hoffs
1474 .iter()
1475 .map(|&hoff_id| self.hoff_buf_ident(hoff_id, self.nodes[hoff_id].span()))
1476 .collect();
1477
1478 let recv_kinds = recv_hoffs
1480 .iter()
1481 .map(|&hoff_id| {
1482 let GraphNode::Handoff { kind, .. } = self.node(hoff_id) else {
1483 panic!()
1484 };
1485 *kind
1486 })
1487 .collect::<Vec<_>>();
1488 let send_kinds = send_hoffs
1489 .iter()
1490 .map(|&hoff_id| {
1491 let GraphNode::Handoff { kind, .. } = self.node(hoff_id) else {
1492 panic!()
1493 };
1494 *kind
1495 })
1496 .collect::<Vec<_>>();
1497
1498 let recv_port_code: Vec<TokenStream> = recv_port_idents
1502 .iter()
1503 .zip(recv_buf_idents.iter())
1504 .zip(recv_kinds.iter())
1505 .zip(recv_hoffs.iter())
1506 .map(|(((port_ident, buf_ident), &kind), &hoff_id)| {
1507 let hoff_ffi = hoff_id.data().as_ffi();
1508 let work_done = Ident::new("__dfir_work_done", Span::call_site());
1512 let metrics = Ident::new("__dfir_metrics", Span::call_site());
1513
1514 let (len_expr, drain_expr) = match kind {
1516 HandoffKind::Singleton | HandoffKind::Optional => (
1517 quote! { if #buf_ident.is_some() { 1usize } else { 0usize } },
1518 quote! { #root::dfir_pipes::pull::iter(#buf_ident.take().into_iter()) },
1519 ),
1520 HandoffKind::Vec => {
1521 let drain_ident = if back_edge_hoffs_and_lazyness.contains_key(hoff_id) {
1525 &self.hoff_back_ident(hoff_id, buf_ident.span())
1526 } else {
1527 buf_ident
1528 };
1529 (
1530 quote! { #drain_ident.len() },
1531 quote! { #root::dfir_pipes::pull::iter(#drain_ident.drain(..)) },
1532 )
1533 }
1534 };
1535
1536 let track_hoff_metrics = options.include_metrics_tracking.then(|| {
1537 quote_spanned! {port_ident.span()=>
1538 let hoff_metrics = &#metrics.handoffs[
1539 #root::slotmap::KeyData::from_ffi(#hoff_ffi).into()
1540 ];
1541 hoff_metrics.total_items_count.update(|x| x + hoff_len);
1542 hoff_metrics.curr_items_count.set(hoff_len);
1543 }
1544 });
1545
1546 quote_spanned! {port_ident.span()=>
1547 {
1548 let hoff_len = #len_expr;
1549 if hoff_len > 0 {
1550 #work_done = true;
1551 }
1552 #track_hoff_metrics
1553 }
1554 let #port_ident = #drain_expr;
1555 }
1556 })
1557 .collect();
1558
1559 let send_port_code: Vec<TokenStream> = send_port_idents
1561 .iter()
1562 .zip(send_buf_idents.iter())
1563 .zip(send_kinds.iter())
1564 .map(|((port_ident, buf_ident), &kind)| {
1565 match kind {
1566 HandoffKind::Singleton => {
1567 quote_spanned! {port_ident.span()=>
1569 let #port_ident = #root::dfir_pipes::push::for_each(|__item| {
1570 if #buf_ident.replace(__item).is_some() {
1571 panic!("singleton() received more than one item");
1572 }
1573 });
1574 }
1575 }
1576 HandoffKind::Optional => {
1577 quote_spanned! {port_ident.span()=>
1579 let #port_ident = #root::dfir_pipes::push::for_each(|__item| {
1580 if #buf_ident.replace(__item).is_some() {
1581 panic!("optional() received more than one item");
1582 }
1583 });
1584 }
1585 }
1586 HandoffKind::Vec => {
1587 quote_spanned! {port_ident.span()=>
1588 let #port_ident = #root::dfir_pipes::push::for_each(|item| { #buf_ident.push(item); });
1590 }
1591 }
1592 }
1593 })
1594 .collect();
1595
1596 let loop_id = self.node_loop(subgraph_nodes[0]);
1598
1599 let mut subgraph_op_iter_code = Vec::new();
1600 let mut subgraph_op_iter_after_code = Vec::new();
1601 {
1602 let pull_to_push_idx = self.find_pull_to_push_idx(subgraph_nodes);
1603
1604 let (pull_half, push_half) = subgraph_nodes.split_at(pull_to_push_idx);
1605 let nodes_iter = pull_half.iter().chain(push_half.iter().rev());
1606
1607 for (idx, &node_id) in nodes_iter.enumerate() {
1608 let node = &self.nodes[node_id];
1609 assert!(
1610 matches!(node, GraphNode::Operator(_)),
1611 "Handoffs are not part of subgraphs."
1612 );
1613 let op_inst = &self.operator_instances[node_id];
1614
1615 let op_span = node.span();
1616 let op_name = op_inst.op_constraints.name;
1617 let root = change_spans(root.clone(), op_span);
1619 let op_constraints = OPERATORS
1620 .iter()
1621 .find(|op| op_name == op.name)
1622 .unwrap_or_else(|| panic!("Failed to find op: {}", op_name));
1623
1624 let ident = self.node_as_ident(node_id, false);
1625
1626 {
1627 let mut input_edges = self
1630 .graph
1631 .predecessor_edges(node_id)
1632 .map(|edge_id| (self.edge_ports(edge_id).1, edge_id))
1633 .collect::<Vec<_>>();
1634 input_edges.sort();
1636
1637 let inputs = input_edges
1638 .iter()
1639 .map(|&(_port, edge_id)| {
1640 let (pred, _) = self.edge(edge_id);
1641 self.node_as_ident(pred, true)
1642 })
1643 .collect::<Vec<_>>();
1644
1645 let mut output_edges = self
1647 .graph
1648 .successor_edges(node_id)
1649 .map(|edge_id| (&self.ports[edge_id].0, edge_id))
1650 .collect::<Vec<_>>();
1651 output_edges.sort();
1653
1654 let outputs = output_edges
1655 .iter()
1656 .map(|&(_port, edge_id)| {
1657 let (_, succ) = self.edge(edge_id);
1658 self.node_as_ident(succ, false)
1659 })
1660 .collect::<Vec<_>>();
1661
1662 let is_pull = idx < pull_to_push_idx;
1663
1664 let df_local = &Ident::new(GRAPH, op_span.resolved_at(df.span()));
1673 let context = &Ident::new(CONTEXT, op_span.resolved_at(context.span()));
1674
1675 let singletons_resolved =
1676 self.helper_resolve_singletons(node_id, op_span);
1677
1678 let arguments = &process_singletons::postprocess_singletons(
1679 op_inst.arguments_raw.clone(),
1680 singletons_resolved,
1681 );
1682
1683 let source_tag = 'a: {
1684 if let Some(tag) = self.operator_tag.get(node_id).cloned() {
1685 break 'a tag;
1686 }
1687
1688 if proc_macro::is_available() {
1689 let op_span = op_span.unwrap();
1690 break 'a format!(
1691 "loc_{}_{}_{}_{}_{}",
1692 crate::pretty_span::make_source_path_relative(
1693 &op_span.file()
1694 )
1695 .display()
1696 .to_string()
1697 .replace(|x: char| !x.is_ascii_alphanumeric(), "_"),
1698 op_span.start().line(),
1699 op_span.start().column(),
1700 op_span.end().line(),
1701 op_span.end().column(),
1702 );
1703 }
1704
1705 format!(
1706 "loc_nopath_{}_{}_{}_{}",
1707 op_span.start().line,
1708 op_span.start().column,
1709 op_span.end().line,
1710 op_span.end().column
1711 )
1712 };
1713
1714 let work_fn = format_ident!(
1715 "{}__{}__{}",
1716 ident,
1717 op_name,
1718 source_tag,
1719 span = op_span
1720 );
1721 let work_fn_async = format_ident!("{}__async", work_fn, span = op_span);
1722
1723 let context_args = WriteContextArgs {
1724 root: &root,
1725 df_ident: df_local,
1726 context,
1727 subgraph_id,
1728 node_id,
1729 loop_id,
1730 op_span,
1731 op_tag: self.operator_tag.get(node_id).cloned(),
1732 work_fn: &work_fn,
1733 work_fn_async: &work_fn_async,
1734 ident: &ident,
1735 is_pull,
1736 inputs: &inputs,
1737 outputs: &outputs,
1738 op_name,
1739 op_inst,
1740 arguments,
1741 };
1742
1743 let write_result =
1744 (op_constraints.write_fn)(&context_args, diagnostics);
1745 let OperatorWriteOutput {
1746 write_prologue,
1747 write_iterator,
1748 write_iterator_after,
1749 write_tick_end,
1750 } = write_result.unwrap_or_else(|()| {
1751 assert!(
1752 diagnostics.has_error(),
1753 "Operator `{}` returned `Err` but emitted no diagnostics, this is a bug.",
1754 op_name,
1755 );
1756 OperatorWriteOutput {
1757 write_iterator: null_write_iterator_fn(&context_args),
1758 ..Default::default()
1759 }
1760 });
1761
1762 op_prologue_code.push(syn::parse_quote! {
1763 #[allow(dead_code, non_snake_case, reason = "codegen")]
1764 #[inline(always)]
1765 fn #work_fn<T>(thunk: impl ::std::ops::FnOnce() -> T) -> T {
1766 thunk()
1767 }
1768
1769 #[allow(dead_code, non_snake_case, reason = "codegen")]
1770 #[inline(always)]
1771 async fn #work_fn_async<T>(
1772 thunk: impl ::std::future::Future<Output = T>,
1773 ) -> T {
1774 thunk.await
1775 }
1776 });
1777 op_prologue_code.push(write_prologue);
1778 op_tick_end_code.push(write_tick_end);
1779 subgraph_op_iter_code.push(write_iterator);
1780
1781 if !options.exclude_type_guards {
1782 let type_guard = if is_pull {
1783 quote_spanned! {op_span=>
1784 let #ident = {
1785 #[allow(non_snake_case)]
1786 #[inline(always)]
1787 pub fn #work_fn<Item, Input>(input: Input)
1788 -> impl #root::dfir_pipes::pull::Pull<Item = Item, Meta = (), CanPend = Input::CanPend, CanEnd = Input::CanEnd>
1789 where
1790 Input: #root::dfir_pipes::pull::Pull<Item = Item, Meta = ()>,
1791 {
1792 #root::pin_project_lite::pin_project! {
1793 #[repr(transparent)]
1794 struct Pull<Item, Input: #root::dfir_pipes::pull::Pull<Item = Item>> {
1795 #[pin]
1796 inner: Input
1797 }
1798 }
1799
1800 impl<Item, Input> #root::dfir_pipes::pull::Pull for Pull<Item, Input>
1801 where
1802 Input: #root::dfir_pipes::pull::Pull<Item = Item>,
1803 {
1804 type Ctx<'ctx> = Input::Ctx<'ctx>;
1805
1806 type Item = Item;
1807 type Meta = Input::Meta;
1808 type CanPend = Input::CanPend;
1809 type CanEnd = Input::CanEnd;
1810
1811 #[inline(always)]
1812 fn pull(
1813 self: ::std::pin::Pin<&mut Self>,
1814 ctx: &mut Self::Ctx<'_>,
1815 ) -> #root::dfir_pipes::pull::PullStep<Self::Item, Self::Meta, Self::CanPend, Self::CanEnd> {
1816 #root::dfir_pipes::pull::Pull::pull(self.project().inner, ctx)
1817 }
1818
1819 #[inline(always)]
1820 fn size_hint(&self) -> (usize, Option<usize>) {
1821 #root::dfir_pipes::pull::Pull::size_hint(&self.inner)
1822 }
1823 }
1824
1825 Pull {
1826 inner: input
1827 }
1828 }
1829 #work_fn::<_, _>( #ident )
1830 };
1831 }
1832 } else {
1833 quote_spanned! {op_span=>
1834 let #ident = {
1835 #[allow(non_snake_case)]
1836 #[inline(always)]
1837 pub fn #work_fn<Item, Psh>(psh: Psh) -> impl #root::dfir_pipes::push::Push<Item, (), CanPend = Psh::CanPend>
1838 where
1839 Psh: #root::dfir_pipes::push::Push<Item, ()>
1840 {
1841 #root::pin_project_lite::pin_project! {
1842 #[repr(transparent)]
1843 struct PushGuard<Psh> {
1844 #[pin]
1845 inner: Psh,
1846 }
1847 }
1848
1849 impl<Item, Psh> #root::dfir_pipes::push::Push<Item, ()> for PushGuard<Psh>
1850 where
1851 Psh: #root::dfir_pipes::push::Push<Item, ()>,
1852 {
1853 type Ctx<'ctx> = Psh::Ctx<'ctx>;
1854
1855 type CanPend = Psh::CanPend;
1856
1857 #[inline(always)]
1858 fn poll_ready(
1859 self: ::std::pin::Pin<&mut Self>,
1860 ctx: &mut Self::Ctx<'_>,
1861 ) -> #root::dfir_pipes::push::PushStep<Self::CanPend> {
1862 #root::dfir_pipes::push::Push::poll_ready(self.project().inner, ctx)
1863 }
1864
1865 #[inline(always)]
1866 fn start_send(
1867 self: ::std::pin::Pin<&mut Self>,
1868 item: Item,
1869 meta: (),
1870 ) {
1871 #root::dfir_pipes::push::Push::start_send(self.project().inner, item, meta)
1872 }
1873
1874 #[inline(always)]
1875 fn poll_finalize(
1876 self: ::std::pin::Pin<&mut Self>,
1877 ctx: &mut Self::Ctx<'_>,
1878 ) -> #root::dfir_pipes::push::PushStep<Self::CanPend> {
1879 #root::dfir_pipes::push::Push::poll_finalize(self.project().inner, ctx)
1880 }
1881
1882 #[inline(always)]
1883 fn size_hint(
1884 self: ::std::pin::Pin<&mut Self>,
1885 hint: (usize, Option<usize>),
1886 ) {
1887 #root::dfir_pipes::push::Push::size_hint(self.project().inner, hint)
1888 }
1889 }
1890
1891 PushGuard {
1892 inner: psh
1893 }
1894 }
1895 #work_fn( #ident )
1896 };
1897 }
1898 };
1899 subgraph_op_iter_code.push(type_guard);
1900 }
1901 subgraph_op_iter_after_code.push(write_iterator_after);
1902 }
1903 }
1904
1905 {
1906 let pull_ident = if 0 < pull_to_push_idx {
1908 self.node_as_ident(subgraph_nodes[pull_to_push_idx - 1], false)
1909 } else {
1910 recv_port_idents[0].clone()
1912 };
1913
1914 #[rustfmt::skip]
1915 let push_ident = if let Some(&node_id) =
1916 subgraph_nodes.get(pull_to_push_idx)
1917 {
1918 self.node_as_ident(node_id, false)
1919 } else if 1 == send_port_idents.len() {
1920 send_port_idents[0].clone()
1922 } else {
1923 diagnostics.push(Diagnostic::spanned(
1924 pull_ident.span(),
1925 Level::Error,
1926 "Degenerate subgraph detected, is there a disconnected `null()` or other degenerate pipeline somewhere?",
1927 ));
1928 continue;
1929 };
1930
1931 let pivot_span = pull_ident
1933 .span()
1934 .join(push_ident.span())
1935 .unwrap_or_else(|| push_ident.span());
1936 let pivot_fn_ident = Ident::new(
1937 &format!("pivot_run_sg_{:?}", subgraph_id.data()),
1938 pivot_span,
1939 );
1940 let root = change_spans(root.clone(), pivot_span);
1941 subgraph_op_iter_code.push(quote_spanned! {pivot_span=>
1942 #[inline(always)]
1943 fn #pivot_fn_ident<Pul, Psh, Item>(pull: Pul, push: Psh)
1944 -> impl ::std::future::Future<Output = ()>
1945 where
1946 Pul: #root::dfir_pipes::pull::Pull<Item = Item>,
1947 Psh: #root::dfir_pipes::push::Push<Item, Pul::Meta>,
1948 {
1949 #root::dfir_pipes::pull::Pull::send_push(pull, push)
1950 }
1951 (#pivot_fn_ident)(#pull_ident, #push_ident).await;
1952 });
1953 }
1954 };
1955
1956 let sg_fut_ident = subgraph_id.as_ident(Span::call_site());
1960
1961 let send_metrics_code = if options.include_metrics_tracking {
1963 send_hoffs
1964 .iter()
1965 .zip(send_buf_idents.iter())
1966 .zip(send_kinds.iter())
1967 .map(|((&hoff_id, buf_ident), &kind)| {
1968 let hoff_ffi = hoff_id.data().as_ffi();
1969 let len_expr = match kind {
1970 HandoffKind::Singleton | HandoffKind::Optional => {
1971 quote! { if #buf_ident.is_some() { 1 } else { 0 } }
1972 }
1973 HandoffKind::Vec => {
1974 quote! { #buf_ident.len() }
1975 }
1976 };
1977 quote! {
1978 __dfir_metrics.handoffs[
1979 #root::slotmap::KeyData::from_ffi(#hoff_ffi).into()
1980 ].curr_items_count.set(#len_expr);
1981 }
1982 })
1983 .collect::<Vec<_>>()
1984 } else {
1985 Vec::new()
1986 };
1987
1988 let send_hoff_make_code = send_buf_idents.iter()
1992 .zip(send_kinds.iter())
1993 .zip(send_hoffs.iter())
1994 .filter_map(|((buf_ident, &kind), &hoff_id)| {
1995 let span = buf_ident.span();
1996 if back_edge_hoffs_and_lazyness.contains_key(hoff_id) {
1997 Some(quote_spanned! {span=>
2000 #buf_ident.clear();
2001 })
2002 } else {
2003 let receiver_loop = self
2006 .node_successors(hoff_id)
2007 .next()
2008 .and_then(|(_, succ)| self.node_loop(succ));
2009 let is_exit = if let Some(sender_loop) = sg_loop {
2010 receiver_loop == self.loop_parent(sender_loop)
2011 } else {
2012 false
2013 };
2014 if is_exit {
2015 None
2017 } else {
2018 Some(match kind {
2019 HandoffKind::Vec => quote_spanned! {span=>
2020 let mut #buf_ident = #root::bumpalo::collections::Vec::new_in(&#bump_ident);
2021 },
2022 HandoffKind::Singleton | HandoffKind::Optional => quote_spanned! {span=>
2023 let mut #buf_ident = ::std::option::Option::None;
2024 },
2025 })
2026 }
2027 }
2028 })
2029 .collect::<Vec<_>>();
2030 let recv_hoff_drop_code = recv_buf_idents
2034 .iter()
2035 .zip(recv_hoffs.iter())
2036 .filter(|&(_, &hoff_id)| !back_edge_hoffs_and_lazyness.contains_key(hoff_id))
2037 .map(|(buf_ident, _)| {
2038 let span = buf_ident.span();
2039 quote_spanned! {span=>
2040 let _ = #buf_ident;
2041 }
2042 });
2043
2044 let run_sg = if options.include_metrics_tracking {
2045 quote! {
2046 let sg_metrics = &__dfir_metrics.subgraphs[
2048 #root::slotmap::KeyData::from_ffi(#sg_metrics_ffi).into()
2049 ];
2050 #root::scheduled::metrics::InstrumentSubgraph::new(
2051 #sg_fut_ident, sg_metrics
2052 ).await;
2053 sg_metrics.total_run_count.update(|x| x + 1);
2054
2055 #( #send_metrics_code )*
2057 }
2058 } else {
2059 quote! {
2060 #sg_fut_ident.await;
2061 }
2062 };
2063
2064 let sg_block = quote! {
2066 #( #send_hoff_make_code )*
2068
2069 let #sg_fut_ident = async {
2070 let #context = &#df;
2071 #( #recv_port_code )*
2072 #( #send_port_code )*
2073 #( #subgraph_op_iter_code )*
2074 #( #subgraph_op_iter_after_code )*
2075 };
2076 {
2077 #run_sg
2078
2079 #( #recv_hoff_drop_code )*
2081 }
2082 };
2083 if let Some((_, body)) = loop_stack.last_mut() {
2084 body.extend(sg_block);
2085 } else {
2086 current_output.extend(sg_block);
2087 }
2088 }
2089 }
2090
2091 let gated_subgraph_code = {
2093 while let Some((closed_loop, child_body)) = loop_stack.pop() {
2094 let target = if let Some((_, parent_body)) = loop_stack.last_mut() {
2095 parent_body
2096 } else {
2097 &mut current_output
2098 };
2099 self.emit_loop_gate(
2100 closed_loop,
2101 child_body,
2102 &loop_input_handoffs,
2103 &back_edge_hoffs_and_lazyness,
2104 &loop_swap_code,
2105 target,
2106 );
2107 }
2108 current_output
2109 };
2110
2111 if diagnostics.has_error() {
2112 return Err(std::mem::take(diagnostics));
2113 }
2114 let _ = diagnostics; let (meta_graph_arg, diagnostics_arg) = if !options.exclude_meta {
2117 let meta_graph_json = serde_json::to_string(&self).unwrap();
2118 let meta_graph_json = Literal::string(&meta_graph_json);
2119
2120 let serde_diagnostics: Vec<_> = diagnostics.iter().map(Diagnostic::to_serde).collect();
2121 let diagnostics_json = serde_json::to_string(&*serde_diagnostics).unwrap();
2122 let diagnostics_json = Literal::string(&diagnostics_json);
2123
2124 (
2125 quote! { Some(#meta_graph_json) },
2126 quote! { Some(#diagnostics_json) },
2127 )
2128 } else {
2129 (quote! { None }, quote! { None })
2130 };
2131
2132 let metrics_init_code = if options.include_metrics_tracking {
2134 let handoff_inits = handoff_nodes.iter().map(|&(node_id, _, _)| {
2135 let ffi = node_id.data().as_ffi();
2136 quote! {
2137 dfir_metrics.handoffs.insert(
2138 #root::slotmap::KeyData::from_ffi(#ffi).into(),
2139 ::std::default::Default::default(),
2140 );
2141 }
2142 });
2143 let subgraph_inits = all_subgraphs.iter().map(|&(sg_id, _)| {
2144 let ffi = sg_id.data().as_ffi();
2145 quote! {
2146 dfir_metrics.subgraphs.insert(
2147 #root::slotmap::KeyData::from_ffi(#ffi).into(),
2148 ::std::default::Default::default(),
2149 );
2150 }
2151 });
2152 handoff_inits.chain(subgraph_inits).collect::<Vec<_>>()
2153 } else {
2154 Vec::new()
2155 };
2156
2157 let back_buffer_idents = back_buffer_idents_laziness
2159 .iter()
2160 .map(|(back_ident, _, _)| back_ident);
2161 let defer_tick_buf_idents = back_buffer_idents_laziness
2163 .iter()
2164 .map(|(_, buf_ident, _)| buf_ident);
2165 let non_lazy_schedule_idents: Vec<&Ident> = handoff_nodes
2170 .iter()
2171 .filter_map(|&(hoff_id, _, _)| {
2172 let delay_type = self.handoff_delay_type(hoff_id)?;
2173 if matches!(delay_type, DelayType::TickLazy | DelayType::LoopLazy) {
2175 return None;
2176 }
2177 let span = self.nodes[hoff_id].span();
2178 let expected_back_ident = self.hoff_back_ident(hoff_id, span);
2179 let entry = back_buffer_idents_laziness
2180 .iter()
2181 .find(|(back_ident, _, _)| *back_ident == expected_back_ident)?;
2182
2183 if delay_type == DelayType::Tick {
2185 let consumer_loop = self
2186 .node_successors(hoff_id)
2187 .next()
2188 .and_then(|(_, succ)| self.node_subgraph(succ))
2189 .and_then(|sg| self.subgraph_loop(sg));
2190 if consumer_loop.is_some_and(|lid| self.loop_parent(lid).is_none()) {
2191 return Some(&entry.0); }
2193 }
2194 Some(&entry.1) })
2196 .collect();
2197
2198 Ok(quote! {
2201 {
2202 #prefix
2203
2204 use #root::{var_expr, var_args};
2205
2206 let __dfir_wake_state = ::std::sync::Arc::new(
2207 #root::scheduled::context::WakeState::default()
2208 );
2209
2210 let __dfir_metrics = {
2211 let mut dfir_metrics = #root::scheduled::metrics::DfirMetrics::default();
2212 #( #metrics_init_code )*
2213 ::std::rc::Rc::new(dfir_metrics)
2214 };
2215
2216 #[allow(unused_mut)]
2217 let mut #df = #root::scheduled::context::Context::new(
2218 ::std::clone::Clone::clone(&__dfir_wake_state),
2219 __dfir_metrics,
2220 );
2221
2222 #( #op_prologue_code )*
2223
2224 #( let mut #back_buffer_idents = ::std::vec::Vec::new(); )*
2228 #( let mut #defer_tick_buf_idents = ::std::vec::Vec::new(); )*
2229
2230 let mut #bump_ident = #root::bumpalo::Bump::new();
2232
2233 let mut __dfir_work_done = true;
2238 #[allow(unused_qualifications, unused_mut, unused_variables, clippy::await_holding_refcell_ref, clippy::deref_addrof)]
2239 let __dfir_inline_tick = async move |#df: &mut #root::scheduled::context::Context| {
2240 #bump_ident.reset();
2242
2243 {
2244 let __dfir_metrics = #df.metrics();
2245
2246 #gated_subgraph_code
2247
2248 #[allow(clippy::nonminimal_bool, reason = "codegen")]
2251 if false #( || !#non_lazy_schedule_idents.is_empty() )* {
2252 #df.schedule_subgraph(true);
2253 }
2254
2255 #( #back_edge_swap_code )*
2258 }
2259
2260 #( #op_tick_end_code )*
2262
2263 #df.__end_tick();
2264
2265 ::std::mem::take(&mut __dfir_work_done)
2266 };
2267 #root::scheduled::context::Dfir::new(
2268 __dfir_inline_tick,
2269 #df,
2270 #meta_graph_arg,
2271 #diagnostics_arg,
2272 )
2273 }
2274 })
2275 }
2276
2277 pub fn node_color_map(&self) -> SparseSecondaryMap<GraphNodeId, Color> {
2280 let mut node_color_map: SparseSecondaryMap<GraphNodeId, Color> = self
2281 .node_ids()
2282 .filter_map(|node_id| {
2283 let op_color = self.node_color(node_id)?;
2284 Some((node_id, op_color))
2285 })
2286 .collect();
2287
2288 for sg_nodes in self.subgraph_nodes.values() {
2290 let pull_to_push_idx = self.find_pull_to_push_idx(sg_nodes);
2291
2292 for (idx, node_id) in sg_nodes.iter().copied().enumerate() {
2293 let is_pull = idx < pull_to_push_idx;
2294 node_color_map.insert(node_id, if is_pull { Color::Pull } else { Color::Push });
2295 }
2296 }
2297
2298 node_color_map
2299 }
2300
2301 pub fn to_mermaid(&self, write_config: &WriteConfig) -> String {
2303 let mut output = String::new();
2304 self.write_mermaid(&mut output, write_config).unwrap();
2305 output
2306 }
2307
2308 pub fn write_mermaid(
2310 &self,
2311 output: impl std::fmt::Write,
2312 write_config: &WriteConfig,
2313 ) -> std::fmt::Result {
2314 let mut graph_write = Mermaid::new(output);
2315 self.write_graph(&mut graph_write, write_config)
2316 }
2317
2318 pub fn to_dot(&self, write_config: &WriteConfig) -> String {
2320 let mut output = String::new();
2321 let mut graph_write = Dot::new(&mut output);
2322 self.write_graph(&mut graph_write, write_config).unwrap();
2323 output
2324 }
2325
2326 pub fn write_dot(
2328 &self,
2329 output: impl std::fmt::Write,
2330 write_config: &WriteConfig,
2331 ) -> std::fmt::Result {
2332 let mut graph_write = Dot::new(output);
2333 self.write_graph(&mut graph_write, write_config)
2334 }
2335
2336 pub(crate) fn write_graph<W>(
2338 &self,
2339 mut graph_write: W,
2340 write_config: &WriteConfig,
2341 ) -> Result<(), W::Err>
2342 where
2343 W: GraphWrite,
2344 {
2345 fn helper_edge_label(
2346 src_port: &PortIndexValue,
2347 dst_port: &PortIndexValue,
2348 ) -> Option<String> {
2349 let src_label = match src_port {
2350 PortIndexValue::Path(path) => Some(path.to_token_stream().to_string()),
2351 PortIndexValue::Int(index) => Some(index.value.to_string()),
2352 _ => None,
2353 };
2354 let dst_label = match dst_port {
2355 PortIndexValue::Path(path) => Some(path.to_token_stream().to_string()),
2356 PortIndexValue::Int(index) => Some(index.value.to_string()),
2357 _ => None,
2358 };
2359 let label = match (src_label, dst_label) {
2360 (Some(l1), Some(l2)) => Some(format!("{}\n{}", l1, l2)),
2361 (Some(l1), None) => Some(l1),
2362 (None, Some(l2)) => Some(l2),
2363 (None, None) => None,
2364 };
2365 label
2366 }
2367
2368 let node_color_map = self.node_color_map();
2370
2371 graph_write.write_prologue()?;
2373
2374 let mut skipped_handoffs = BTreeSet::new();
2376 for (node_id, node) in self.nodes() {
2377 if matches!(node, GraphNode::Handoff { .. }) && write_config.no_handoffs {
2378 skipped_handoffs.insert(node_id);
2379 continue;
2380 }
2381 graph_write.write_node_definition(
2382 node_id,
2383 &if write_config.op_short_text {
2384 node.to_name_string()
2385 } else if write_config.op_text_no_imports {
2386 let full_text = node.to_pretty_string();
2388 let mut output = String::new();
2389 for sentence in full_text.split('\n') {
2390 if sentence.trim().starts_with("use") {
2391 continue;
2392 }
2393 output.push('\n');
2394 output.push_str(sentence);
2395 }
2396 output.into()
2397 } else {
2398 node.to_pretty_string()
2399 },
2400 if write_config.no_pull_push {
2401 None
2402 } else {
2403 node_color_map.get(node_id).copied()
2404 },
2405 )?;
2406 }
2407
2408 for (edge_id, (src_id, mut dst_id)) in self.edges() {
2410 if skipped_handoffs.contains(&src_id) {
2412 continue;
2413 }
2414
2415 let (src_port, mut dst_port) = self.edge_ports(edge_id);
2416 if skipped_handoffs.contains(&dst_id) {
2417 let mut handoff_succs = self.node_successors(dst_id);
2421 if handoff_succs.len() == 0 {
2422 continue;
2423 }
2424 let (succ_edge, succ_node) = handoff_succs.next().unwrap();
2425 dst_id = succ_node;
2426 dst_port = self.edge_ports(succ_edge).1;
2427 }
2428
2429 let label = helper_edge_label(src_port, dst_port);
2430 let delay_type = self
2431 .node_op_inst(dst_id)
2432 .and_then(|op_inst| (op_inst.op_constraints.input_delaytype_fn)(dst_port));
2433 graph_write.write_edge(src_id, dst_id, delay_type, label.as_deref(), false)?;
2434 }
2435
2436 if !write_config.no_references {
2438 for dst_id in self.node_ids() {
2439 for src_ref_id in self
2440 .node_handoff_references(dst_id)
2441 .iter()
2442 .filter_map(|r| r.node_id)
2443 {
2444 let resolved_src = if skipped_handoffs.contains(&src_ref_id) {
2447 self.node_predecessor_nodes(src_ref_id).next()
2448 } else {
2449 Some(src_ref_id)
2450 };
2451 let Some(resolved_src) = resolved_src else {
2452 continue;
2453 };
2454 let label = None;
2455 graph_write.write_edge(resolved_src, dst_id, None, label, true)?;
2456 }
2457 }
2458 }
2459
2460 let loop_subgraphs = self.subgraph_ids().map(|sg_id| {
2468 let loop_id = if write_config.no_loops {
2469 None
2470 } else {
2471 self.subgraph_loop(sg_id)
2472 };
2473 (loop_id, sg_id)
2474 });
2475 let loop_subgraphs = into_group_map(loop_subgraphs);
2476 for (loop_id, subgraph_ids) in loop_subgraphs {
2477 if let Some(loop_id) = loop_id {
2478 graph_write.write_loop_start(loop_id)?;
2479 }
2480
2481 let subgraph_varnames_nodes = subgraph_ids.into_iter().flat_map(|sg_id| {
2483 self.subgraph(sg_id).iter().copied().map(move |node_id| {
2484 let opt_sg_id = if write_config.no_subgraphs {
2485 None
2486 } else {
2487 Some(sg_id)
2488 };
2489 (opt_sg_id, (self.node_varname(node_id), node_id))
2490 })
2491 });
2492 let subgraph_varnames_nodes = into_group_map(subgraph_varnames_nodes);
2493 for (sg_id, varnames) in subgraph_varnames_nodes {
2494 if let Some(sg_id) = sg_id {
2495 graph_write.write_subgraph_start(sg_id)?;
2496 }
2497
2498 let varname_nodes = varnames.into_iter().map(|(varname, node)| {
2500 let varname = if write_config.no_varnames {
2501 None
2502 } else {
2503 varname
2504 };
2505 (varname, node)
2506 });
2507 let varname_nodes = into_group_map(varname_nodes);
2508 for (varname, node_ids) in varname_nodes {
2509 if let Some(varname) = varname {
2510 graph_write.write_varname_start(&varname.0.to_string(), sg_id)?;
2511 }
2512
2513 for node_id in node_ids {
2515 graph_write.write_node(node_id)?;
2516 }
2517
2518 if varname.is_some() {
2519 graph_write.write_varname_end()?;
2520 }
2521 }
2522
2523 if sg_id.is_some() {
2524 graph_write.write_subgraph_end()?;
2525 }
2526 }
2527
2528 if loop_id.is_some() {
2529 graph_write.write_loop_end()?;
2530 }
2531 }
2532
2533 graph_write.write_epilogue()?;
2535
2536 Ok(())
2537 }
2538
2539 pub fn surface_syntax_string(&self) -> String {
2541 let mut string = String::new();
2542 self.write_surface_syntax(&mut string).unwrap();
2543 string
2544 }
2545
2546 pub fn write_surface_syntax(&self, write: &mut impl std::fmt::Write) -> std::fmt::Result {
2548 for (key, node) in self.nodes.iter() {
2549 match node {
2550 GraphNode::Operator(op) => {
2551 writeln!(write, "_{:?} = {};", key.data(), op.to_token_stream())?;
2552 }
2553 GraphNode::Handoff {
2554 kind: HandoffKind::Vec,
2555 ..
2556 } => {
2557 writeln!(write, "_{:?} = handoff();", key.data())?;
2558 }
2559 GraphNode::Handoff {
2560 kind: HandoffKind::Singleton,
2561 ..
2562 } => {
2563 writeln!(write, "_{:?} = singleton();", key.data())?;
2564 }
2565 GraphNode::Handoff {
2566 kind: HandoffKind::Optional,
2567 ..
2568 } => {
2569 writeln!(write, "_{:?} = optional();", key.data())?;
2570 }
2571 GraphNode::ModuleBoundary { .. } => panic!(),
2572 }
2573 }
2574 writeln!(write)?;
2575 for (e, (src_key, dst_key)) in self.graph.edges() {
2576 let (src_port, dst_port) = self.edge_ports(e);
2577 let src_port_str = if src_port.is_specified() {
2578 format!("[{}]", src_port)
2579 } else {
2580 String::new()
2581 };
2582 let dst_port_str = if dst_port.is_specified() {
2583 format!("[{}]", dst_port)
2584 } else {
2585 String::new()
2586 };
2587 writeln!(
2588 write,
2589 "_{:?}{} -> {}_{:?};",
2590 src_key.data(),
2591 src_port_str,
2592 dst_port_str,
2593 dst_key.data()
2594 )?;
2595 }
2596 Ok(())
2597 }
2598
2599 pub fn mermaid_string_flat(&self) -> String {
2601 let mut string = String::new();
2602 self.write_mermaid_flat(&mut string).unwrap();
2603 string
2604 }
2605
2606 pub fn write_mermaid_flat(&self, write: &mut impl std::fmt::Write) -> std::fmt::Result {
2608 writeln!(write, "flowchart TB")?;
2609 for (key, node) in self.nodes.iter() {
2610 match node {
2611 GraphNode::Operator(operator) => writeln!(
2612 write,
2613 " %% {span}\n {id:?}[\"{row_col} <tt>{code}</tt>\"]",
2614 span = PrettySpan(node.span()),
2615 id = key.data(),
2616 row_col = PrettyRowCol(node.span()),
2617 code = operator
2618 .to_token_stream()
2619 .to_string()
2620 .replace('&', "&")
2621 .replace('<', "<")
2622 .replace('>', ">")
2623 .replace('"', """)
2624 .replace('\n', "<br>"),
2625 ),
2626 GraphNode::Handoff {
2627 kind: HandoffKind::Vec,
2628 ..
2629 } => {
2630 writeln!(write, r#" {:?}{{"{}"}}"#, key.data(), HANDOFF_NODE_STR)
2631 }
2632 GraphNode::Handoff {
2633 kind: HandoffKind::Singleton | HandoffKind::Optional,
2634 ..
2635 } => {
2636 writeln!(
2637 write,
2638 r#" {:?}{{"{}"}}"#,
2639 key.data(),
2640 SINGLETON_SLOT_NODE_STR
2641 )
2642 }
2643 GraphNode::ModuleBoundary { .. } => {
2644 writeln!(
2645 write,
2646 r#" {:?}{{"{}"}}"#,
2647 key.data(),
2648 MODULE_BOUNDARY_NODE_STR
2649 )
2650 }
2651 }?;
2652 }
2653 writeln!(write)?;
2654 for (_e, (src_key, dst_key)) in self.graph.edges() {
2655 writeln!(write, " {:?}-->{:?}", src_key.data(), dst_key.data())?;
2656 }
2657 Ok(())
2658 }
2659}
2660
2661impl DfirGraph {
2663 pub fn loop_ids(&self) -> slotmap::basic::Keys<'_, GraphLoopId, Vec<GraphNodeId>> {
2665 self.loop_nodes.keys()
2666 }
2667
2668 pub fn loops(&self) -> slotmap::basic::Iter<'_, GraphLoopId, Vec<GraphNodeId>> {
2670 self.loop_nodes.iter()
2671 }
2672
2673 pub fn loop_nodes(&self, loop_id: GraphLoopId) -> &[GraphNodeId] {
2675 self.loop_nodes.get(loop_id).unwrap()
2676 }
2677
2678 pub fn insert_loop(&mut self, parent_loop: Option<GraphLoopId>) -> GraphLoopId {
2680 let loop_id = self.loop_nodes.insert(Vec::new());
2681 self.loop_children.insert(loop_id, Vec::new());
2682 if let Some(parent_loop) = parent_loop {
2683 self.loop_parent.insert(loop_id, parent_loop);
2684 self.loop_children
2685 .get_mut(parent_loop)
2686 .unwrap()
2687 .push(loop_id);
2688 } else {
2689 self.root_loops.push(loop_id);
2690 }
2691 loop_id
2692 }
2693
2694 pub fn node_loop(&self, node_id: GraphNodeId) -> Option<GraphLoopId> {
2696 self.node_loops.get(node_id).copied()
2697 }
2698
2699 pub fn subgraph_loop(&self, subgraph_id: GraphSubgraphId) -> Option<GraphLoopId> {
2701 let &node_id = self.subgraph(subgraph_id).first().unwrap();
2702 let out = self.node_loop(node_id);
2703 debug_assert!(
2704 self.subgraph(subgraph_id)
2705 .iter()
2706 .all(|&node_id| self.node_loop(node_id) == out),
2707 "Subgraph nodes should all have the same loop context."
2708 );
2709 out
2710 }
2711
2712 pub fn loop_parent(&self, loop_id: GraphLoopId) -> Option<GraphLoopId> {
2714 self.loop_parent.get(loop_id).copied()
2715 }
2716
2717 pub fn loop_children(&self, loop_id: GraphLoopId) -> &Vec<GraphLoopId> {
2719 self.loop_children.get(loop_id).unwrap()
2720 }
2721
2722 pub fn root_loops(&self) -> &[GraphLoopId] {
2724 &self.root_loops
2725 }
2726}
2727
2728#[derive(Default)]
2730#[non_exhaustive]
2731pub struct AsCodeOptions {
2732 pub exclude_type_guards: bool,
2735 pub exclude_meta: bool,
2738 pub include_metrics_tracking: bool,
2742}
2743
2744#[derive(Clone, Debug, Default)]
2746#[cfg_attr(feature = "clap-derive", derive(clap::Args))]
2747pub struct WriteConfig {
2748 #[cfg_attr(feature = "clap-derive", arg(long))]
2750 pub no_subgraphs: bool,
2751 #[cfg_attr(feature = "clap-derive", arg(long))]
2753 pub no_varnames: bool,
2754 #[cfg_attr(feature = "clap-derive", arg(long))]
2756 pub no_pull_push: bool,
2757 #[cfg_attr(feature = "clap-derive", arg(long))]
2759 pub no_handoffs: bool,
2760 #[cfg_attr(feature = "clap-derive", arg(long))]
2762 pub no_references: bool,
2763 #[cfg_attr(feature = "clap-derive", arg(long))]
2765 pub no_loops: bool,
2766
2767 #[cfg_attr(feature = "clap-derive", arg(long))]
2769 pub op_short_text: bool,
2770 #[cfg_attr(feature = "clap-derive", arg(long))]
2772 pub op_text_no_imports: bool,
2773}
2774
2775#[derive(Copy, Clone, Debug)]
2777#[cfg_attr(feature = "clap-derive", derive(clap::Parser, clap::ValueEnum))]
2778pub enum WriteGraphType {
2779 Mermaid,
2781 Dot,
2783}
2784
2785fn into_group_map<K, V>(iter: impl IntoIterator<Item = (K, V)>) -> BTreeMap<K, Vec<V>>
2787where
2788 K: Ord,
2789{
2790 let mut out: BTreeMap<_, Vec<_>> = BTreeMap::new();
2791 for (k, v) in iter {
2792 out.entry(k).or_default().push(v);
2793 }
2794 out
2795}