hydro_lang/sim/
flow.rs

1//! Entrypoint for compiling and running Hydro simulations.
2
3use std::cell::RefCell;
4use std::collections::{HashMap, HashSet};
5use std::panic::RefUnwindSafe;
6use std::rc::Rc;
7
8use dfir_lang::graph::{DfirGraph, FlatGraphBuilder, FlatGraphBuilderOutput};
9use libloading::Library;
10use slotmap::SparseSecondaryMap;
11
12use super::builder::SimBuilder;
13use super::compiled::{CompiledSim, CompiledSimInstance};
14use super::graph::{SimDeploy, SimExternal, SimNode, compile_sim, create_sim_graph_trybuild};
15use crate::compile::ir::HydroRoot;
16use crate::location::LocationKey;
17use crate::location::dynamic::LocationId;
18use crate::prelude::Cluster;
19use crate::sim::graph::SimExternalPortRegistry;
20use crate::staging_util::Invariant;
21
22/// A not-yet-compiled simulator for a Hydro program.
23pub struct SimFlow<'a> {
24    pub(crate) ir: Vec<HydroRoot>,
25
26    /// SimNode for each Process.
27    pub(crate) processes: SparseSecondaryMap<LocationKey, SimNode>,
28    /// SimNode for each Cluster.
29    pub(crate) clusters: SparseSecondaryMap<LocationKey, SimNode>,
30    /// SimExternal for each External.
31    pub(crate) externals: SparseSecondaryMap<LocationKey, SimExternal>,
32
33    /// Max size of each cluster.
34    pub(crate) cluster_max_sizes: SparseSecondaryMap<LocationKey, usize>,
35    /// Handle to state handling `external`s' ports.
36    pub(crate) externals_port_registry: Rc<RefCell<SimExternalPortRegistry>>,
37
38    /// When true, the simulator only tests safety properties (not liveness).
39    pub(crate) test_safety_only: bool,
40
41    /// Number of iterations to use for fuzzing, defaults to 8192
42    pub(crate) unit_test_fuzz_iterations: usize,
43
44    pub(crate) _phantom: Invariant<'a>,
45}
46
47impl<'a> SimFlow<'a> {
48    /// Sets the maximum size of the given cluster in the simulation.
49    pub fn with_cluster_size<C>(mut self, cluster: &Cluster<'a, C>, max_size: usize) -> Self {
50        self.cluster_max_sizes.insert(cluster.key, max_size);
51        self
52    }
53
54    /// Opts in to safety-only testing, which is required when using
55    /// [`lossy_delayed_forever`](crate::networking::NetworkingConfig::lossy_delayed_forever)
56    /// networking.
57    ///
58    /// The simulator models dropped messages as indefinitely delayed, which means
59    /// it only tests safety properties—not liveness—since messages may never arrive.
60    /// Calling this method acknowledges that the simulation will not verify that the
61    /// program eventually makes progress.
62    pub fn test_safety_only(mut self) -> Self {
63        self.test_safety_only = true;
64        self
65    }
66
67    /// Sets the number of fuzz iterations for this test. Overrides the
68    /// the default value of 8192
69    pub fn unit_test_fuzz_iterations(mut self, iterations: usize) -> Self {
70        self.unit_test_fuzz_iterations = iterations;
71        self
72    }
73
74    /// Executes the given closure with a single instance of the compiled simulation.
75    pub fn with_instance<T>(self, thunk: impl FnOnce(CompiledSimInstance) -> T) -> T {
76        self.compiled().with_instance(thunk)
77    }
78
79    /// Uses a fuzzing strategy to explore possible executions of the simulation. The provided
80    /// closure will be repeatedly executed with instances of the Hydro program where the
81    /// batching boundaries, order of messages, and retries are varied.
82    ///
83    /// During development, you should run the test that invokes this function with the `cargo sim`
84    /// command, which will use `libfuzzer` to intelligently explore the execution space. If a
85    /// failure is found, a minimized test case will be produced in a `sim-failures` directory.
86    /// When running the test with `cargo test` (such as in CI), if a reproducer is found it will
87    /// be executed, and if no reproducer is found a small number of random executions will be
88    /// performed.
89    pub fn fuzz(self, thunk: impl AsyncFn() + RefUnwindSafe) {
90        self.compiled().fuzz(thunk)
91    }
92
93    /// Exhaustively searches all possible executions of the simulation. The provided
94    /// closure will be repeatedly executed with instances of the Hydro program where the
95    /// batching boundaries, order of messages, and retries are varied.
96    ///
97    /// Exhaustive searching is feasible when the inputs to the Hydro program are finite and there
98    /// are no dataflow loops that generate infinite messages. Exhaustive searching provides a
99    /// stronger guarantee of correctness than fuzzing, but may take a long time to complete.
100    /// Because no fuzzer is involved, you can run exhaustive tests with `cargo test`.
101    ///
102    /// Returns the number of distinct executions explored.
103    pub fn exhaustive(self, thunk: impl AsyncFnMut() + RefUnwindSafe) -> usize {
104        self.compiled().exhaustive(thunk)
105    }
106
107    /// Compiles the simulation into a dynamically loadable library, and returns a handle to it.
108    pub fn compiled(mut self) -> CompiledSim {
109        use std::collections::BTreeMap;
110
111        use dfir_lang::graph::{eliminate_extra_unions_tees, partition_graph};
112
113        let mut sim_emit = SimBuilder {
114            process_graphs: BTreeMap::new(),
115            cluster_graphs: BTreeMap::new(),
116            process_tick_dfirs: BTreeMap::new(),
117            cluster_tick_dfirs: BTreeMap::new(),
118            extra_stmts_global: vec![],
119            extra_stmts_cluster: BTreeMap::new(),
120            next_hoff_id: 0,
121            test_safety_only: self.test_safety_only,
122        };
123
124        // Ensure the default (0) external is always present.
125        self.externals.insert(
126            LocationKey::FIRST,
127            SimExternal {
128                shared_inner: self.externals_port_registry.clone(),
129            },
130        );
131
132        let mut seen_tees_instantiate: HashMap<_, _> = HashMap::new();
133        let mut seen_cluster_members = HashSet::new();
134        self.ir.iter_mut().for_each(|leaf| {
135            leaf.compile_network::<SimDeploy>(
136                &mut SparseSecondaryMap::new(),
137                &mut seen_tees_instantiate,
138                &mut seen_cluster_members,
139                &self.processes,
140                &self.clusters,
141                &self.externals,
142                &mut (),
143            );
144        });
145
146        let mut seen_tees = HashMap::new();
147        let mut built_tees = HashMap::new();
148        let mut next_stmt_id = 0;
149        let mut fold_hooked_idents = HashSet::new();
150        for leaf in &mut self.ir {
151            leaf.emit(
152                &mut sim_emit,
153                &mut seen_tees,
154                &mut built_tees,
155                &mut next_stmt_id,
156                &mut fold_hooked_idents,
157            );
158        }
159
160        fn build_graphs(
161            graphs: BTreeMap<LocationId, FlatGraphBuilder>,
162        ) -> BTreeMap<LocationId, DfirGraph> {
163            graphs
164                .into_iter()
165                .map(|(l, g)| {
166                    let FlatGraphBuilderOutput { mut flat_graph, .. } =
167                        g.build().expect("Failed to build DFIR flat graph.");
168                    eliminate_extra_unions_tees(&mut flat_graph);
169                    (
170                        l,
171                        partition_graph(flat_graph).expect("Failed to partition (cycle detected)."),
172                    )
173                })
174                .collect()
175        }
176
177        let process_graphs = build_graphs(sim_emit.process_graphs);
178        let cluster_graphs = build_graphs(sim_emit.cluster_graphs);
179        let process_tick_graphs = build_graphs(sim_emit.process_tick_dfirs);
180        let cluster_tick_graphs = build_graphs(sim_emit.cluster_tick_dfirs);
181
182        #[expect(
183            clippy::disallowed_methods,
184            reason = "nondeterministic iteration order, fine for checks"
185        )]
186        for c in self.clusters.keys() {
187            assert!(
188                self.cluster_max_sizes.contains_key(c),
189                "Cluster {:?} missing max size; call with_cluster_size() before compiled()",
190                c
191            );
192        }
193
194        let (bin, trybuild) = create_sim_graph_trybuild(
195            process_graphs,
196            cluster_graphs,
197            self.cluster_max_sizes,
198            process_tick_graphs,
199            cluster_tick_graphs,
200            sim_emit.extra_stmts_global,
201            sim_emit.extra_stmts_cluster,
202        );
203
204        let out = compile_sim(bin, trybuild).unwrap();
205        let lib = unsafe { Library::new(&out).unwrap() };
206
207        CompiledSim {
208            _path: out,
209            lib,
210            externals_port_registry: self.externals_port_registry.take(),
211            unit_test_fuzz_iterations: self.unit_test_fuzz_iterations,
212        }
213    }
214}
hydro_lang/sim/flow.rs

hydro_lang/sim/
flow.rs
