hydro_lang/compile/
built.rs

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

hydro_lang/compile/
built.rs
