hydro_test_local/local/
compute_pi.rs

1use std::time::Duration;
2
3use hydro_lang::deploy::SingleProcessGraph;
4use hydro_lang::dfir_rs::scheduled::graph::Dfir;
5use hydro_lang::*;
6use stageleft::{Quoted, RuntimeData};
7
8pub fn compute_pi<'a>(flow: &FlowBuilder<'a>, batch_size: RuntimeData<usize>) -> Process<'a, ()> {
9    let process = flow.process();
10    let tick = process.tick();
11
12    let trials = tick
13        .spin_batch(q!(batch_size))
14        .map(q!(|_| rand::random::<(f64, f64)>()))
15        .map(q!(|(x, y)| x * x + y * y < 1.0))
16        .fold(
17            q!(|| (0u64, 0u64)),
18            q!(|(inside, total), sample_inside| {
19                if sample_inside {
20                    *inside += 1;
21                }
22
23                *total += 1;
24            }),
25        )
26        .all_ticks();
27
28    let estimate = trials.reduce(q!(|(inside, total), (inside_batch, total_batch)| {
29        *inside += inside_batch;
30        *total += total_batch;
31    }));
32
33    unsafe {
34        // SAFETY: intentional non-determinism
35        estimate.sample_every(q!(Duration::from_secs(1)))
36    }
37    .for_each(q!(|(inside, total)| {
38        println!(
39            "pi: {} ({} trials)",
40            4.0 * inside as f64 / total as f64,
41            total
42        );
43    }));
44
45    process
46}
47
48#[stageleft::entry]
49pub fn compute_pi_runtime<'a>(
50    flow: FlowBuilder<'a>,
51    batch_size: RuntimeData<usize>,
52) -> impl Quoted<'a, Dfir<'a>> {
53    let _ = compute_pi(&flow, batch_size);
54    flow.compile_no_network::<SingleProcessGraph>()
55}