hydro_lang/compile/
built.rs

1use std::collections::HashSet;
2use std::marker::PhantomData;
3
4use dfir_lang::graph::{
5    DfirGraph, FlatGraphBuilderOutput, PartitionError, eliminate_extra_unions_tees, partition_graph,
6};
7use slotmap::{SecondaryMap, SlotMap};
8
9use super::compiled::CompiledFlow;
10use super::deploy::{DeployFlow, DeployResult};
11use super::deploy_provider::{ClusterSpec, Deploy, ExternalSpec, IntoProcessSpec};
12use super::ir::{HydroRoot, emit};
13use crate::location::{Cluster, External, LocationKey, LocationType, Process};
14#[cfg(stageleft_runtime)]
15#[cfg(feature = "sim")]
16use crate::sim::{flow::SimFlow, graph::SimNode};
17use crate::staging_util::Invariant;
18#[cfg(stageleft_runtime)]
19#[cfg(feature = "viz")]
20use crate::viz::api::GraphApi;
21
22pub struct BuiltFlow<'a> {
23    pub(super) ir: Vec<HydroRoot>,
24    pub(super) locations: SlotMap<LocationKey, LocationType>,
25    pub(super) location_names: SecondaryMap<LocationKey, String>,
26
27    /// Compile-time sidecar directives extracted from the flow state.
28    pub(super) sidecars: Vec<super::builder::Sidecar>,
29
30    /// Clusters whose code reads their own member ID.
31    pub(super) clusters_reading_self_id: HashSet<LocationKey>,
32
33    /// Application name used in telemetry.
34    pub(super) flow_name: String,
35
36    /// The program version each location belongs to (every location has an entry; 0 unless it is a
37    /// `next_version` successor).
38    #[cfg(feature = "sim")]
39    pub(super) location_version: SecondaryMap<LocationKey, u32>,
40
41    /// Maps from a given location to Version 0 of that location (or itself it is already v0).
42    /// [`Cluster::next_version`](crate::location::Cluster::next_version)).
43    #[cfg(feature = "sim")]
44    pub(super) location_version_group_root: SecondaryMap<LocationKey, LocationKey>,
45
46    pub(super) _phantom: Invariant<'a>,
47}
48
49/// Builds the DFIR graph for each location.
50///
51/// Each location's graph is partitioned into subgraphs. Partitioning can fail (e.g. an
52/// intra-tick cycle), so the result is stored per-location as a [`Result`]:
53/// - `Ok(partitioned_graph)` on success.
54/// - `Err(PartitionError)` on failure, which still carries the (renderable) flat graph
55///   plus the diagnostic. This lets callers such as [`preview_compile`] obtain a meta
56///   graph to *diagnose* the failure instead of panicking with the information discarded.
57///
58/// [`preview_compile`]: super::deploy::DeployFlow::preview_compile
59pub(crate) fn build_inner(
60    ir: &mut Vec<HydroRoot>,
61) -> SecondaryMap<LocationKey, Result<DfirGraph, PartitionError>> {
62    emit(ir)
63        .into_iter()
64        .map(|(k, v)| {
65            let FlatGraphBuilderOutput { mut flat_graph, .. } =
66                v.build().expect("Failed to build DFIR flat graph.");
67            eliminate_extra_unions_tees(&mut flat_graph);
68            (k, partition_graph(flat_graph))
69        })
70        .collect()
71}
72
73impl<'a> BuiltFlow<'a> {
74    /// Returns all [`HydroRoot`]s in the IR.
75    pub fn ir(&self) -> &[HydroRoot] {
76        &self.ir
77    }
78
79    /// Serialize the IR as JSON.
80    #[cfg(feature = "viz")]
81    pub fn ir_json(&self) -> Result<String, serde_json::Error> {
82        super::ir::serialize_dedup_shared(|| serde_json::to_string_pretty(&self.ir))
83    }
84
85    /// Returns all raw location ID -> location name mappings.
86    pub fn location_names(&self) -> &SecondaryMap<LocationKey, String> {
87        &self.location_names
88    }
89
90    /// Get a `GraphApi` instance for this built flow
91    #[cfg(stageleft_runtime)]
92    #[cfg(feature = "viz")]
93    pub fn graph_api(&self) -> GraphApi<'_> {
94        GraphApi::new(&self.ir, self.location_names())
95    }
96
97    /// Render graph to string in the given format.
98    #[cfg(feature = "viz")]
99    pub fn render_graph(
100        &self,
101        format: crate::viz::config::GraphType,
102        use_short_labels: bool,
103        show_metadata: bool,
104    ) -> String {
105        self.graph_api()
106            .render(format, use_short_labels, show_metadata)
107    }
108
109    /// Write graph to file.
110    #[cfg(feature = "viz")]
111    pub fn write_graph_to_file(
112        &self,
113        format: crate::viz::config::GraphType,
114        filename: &str,
115        use_short_labels: bool,
116        show_metadata: bool,
117    ) -> Result<(), Box<dyn std::error::Error>> {
118        self.graph_api()
119            .write_to_file(format, filename, use_short_labels, show_metadata)
120    }
121
122    /// Generate graph based on CLI config. Returns Some(path) if written.
123    #[cfg(feature = "viz")]
124    pub fn generate_graph(
125        &self,
126        config: &crate::viz::config::GraphConfig,
127    ) -> Result<Option<String>, Box<dyn std::error::Error>> {
128        self.graph_api().generate_graph(config)
129    }
130
131    pub fn optimize_with(mut self, f: impl FnOnce(&mut [HydroRoot])) -> Self {
132        f(&mut self.ir);
133        self
134    }
135
136    pub fn with_default_optimize<D: Deploy<'a>>(self) -> DeployFlow<'a, D> {
137        self.into_deploy()
138    }
139
140    #[cfg(feature = "sim")]
141    /// Creates a simulation for this builder, which can be used to run deterministic simulations
142    /// of the Hydro program.
143    pub fn sim(self) -> SimFlow<'a> {
144        use std::cell::RefCell;
145        use std::rc::Rc;
146
147        use slotmap::SparseSecondaryMap;
148
149        use crate::sim::graph::SimNodePort;
150
151        let shared_port_counter = Rc::new(RefCell::new(crate::Counter::<SimNodePort>::default()));
152
153        let mut processes = SparseSecondaryMap::new();
154        let mut clusters = SparseSecondaryMap::new();
155        let externals = SparseSecondaryMap::new();
156
157        for (key, loc) in self.locations.iter() {
158            match loc {
159                LocationType::Process => {
160                    processes.insert(
161                        key,
162                        SimNode {
163                            shared_port_counter: shared_port_counter.clone(),
164                        },
165                    );
166                }
167                LocationType::Cluster => {
168                    clusters.insert(
169                        key,
170                        SimNode {
171                            shared_port_counter: shared_port_counter.clone(),
172                        },
173                    );
174                }
175                LocationType::External => {
176                    panic!("Sim cannot have externals");
177                }
178            }
179        }
180
181        SimFlow {
182            ir: self.ir,
183            processes,
184            clusters,
185            externals,
186            cluster_max_sizes: SparseSecondaryMap::new(),
187            externals_port_registry: Default::default(),
188            location_version: self.location_version,
189            location_version_group_root: self.location_version_group_root,
190            test_safety_only: false,
191            skip_consistency_assertions: false,
192            unit_test_fuzz_iterations: 8192,
193            _phantom: PhantomData,
194        }
195    }
196
197    pub fn into_deploy<D: Deploy<'a>>(self) -> DeployFlow<'a, D> {
198        let (processes, clusters, externals) = Default::default();
199        DeployFlow {
200            ir: self.ir,
201            locations: self.locations,
202            location_names: self.location_names,
203            processes,
204            clusters,
205            externals,
206            sidecars: self.sidecars,
207            clusters_reading_self_id: self.clusters_reading_self_id,
208            as_code_options: Default::default(),
209            flow_name: self.flow_name,
210            _phantom: PhantomData,
211        }
212    }
213
214    pub fn with_process<P, D: Deploy<'a>>(
215        self,
216        process: &Process<'_, P>,
217        spec: impl IntoProcessSpec<'a, D>,
218    ) -> DeployFlow<'a, D> {
219        self.into_deploy().with_process(process, spec)
220    }
221
222    pub fn with_remaining_processes<D: Deploy<'a>, S: IntoProcessSpec<'a, D> + 'a>(
223        self,
224        spec: impl Fn() -> S,
225    ) -> DeployFlow<'a, D> {
226        self.into_deploy().with_remaining_processes(spec)
227    }
228
229    pub fn with_external<P, D: Deploy<'a>>(
230        self,
231        process: &External<'_, P>,
232        spec: impl ExternalSpec<'a, D>,
233    ) -> DeployFlow<'a, D> {
234        self.into_deploy().with_external(process, spec)
235    }
236
237    pub fn with_remaining_externals<D: Deploy<'a>, S: ExternalSpec<'a, D> + 'a>(
238        self,
239        spec: impl Fn() -> S,
240    ) -> DeployFlow<'a, D> {
241        self.into_deploy().with_remaining_externals(spec)
242    }
243
244    pub fn with_cluster<C, D: Deploy<'a>>(
245        self,
246        cluster: &Cluster<'_, C>,
247        spec: impl ClusterSpec<'a, D>,
248    ) -> DeployFlow<'a, D> {
249        self.into_deploy().with_cluster(cluster, spec)
250    }
251
252    pub fn with_remaining_clusters<D: Deploy<'a>, S: ClusterSpec<'a, D> + 'a>(
253        self,
254        spec: impl Fn() -> S,
255    ) -> DeployFlow<'a, D> {
256        self.into_deploy().with_remaining_clusters(spec)
257    }
258
259    pub fn compile<D: Deploy<'a, InstantiateEnv = ()>>(self) -> CompiledFlow<'a> {
260        self.into_deploy::<D>().compile()
261    }
262
263    pub fn deploy<D: Deploy<'a>>(self, env: &mut D::InstantiateEnv) -> DeployResult<'a, D> {
264        self.into_deploy::<D>().deploy(env)
265    }
266}
hydro_lang/compile/built.rs

hydro_lang/compile/
built.rs
