opql/
runner.rs

1// TODO: remove!; tmp implementation
2#![cfg_attr(coverage_nightly, coverage(off))]
3
4use super::*;
5
6pub struct PQLRunner {}
7
8/// Runs `n_trails` successful trials on its own clone of the [`Vm`]
9/// (sharing `cache` with the other clones).
10fn run_trials(
11    mut vm: Vm,
12    n_trails: usize,
13    where_program: Option<&VmProgram>,
14    programs: &[VmProgram],
15    selectors: &[ast::Selector],
16) -> PQLResult<RunnerOutput> {
17    let mut rng = rand::rng();
18    let mut output = RunnerOutput::new(vm.static_data.game, selectors);
19
20    while output.n_succ < n_trails {
21        if output.n_fail == n_trails {
22            // TODO: fix this
23            return Err(((0, 1), VmError::SamplingFailed).into());
24        }
25
26        match vm.sample(&mut rng) {
27            Some(()) => {
28                if let Some(wp) = where_program {
29                    let keep =
30                        matches!(wp.execute(&mut vm.as_context())?, VmStackValue::Bool(true));
31
32                    if !keep {
33                        //TODO: refine this
34                        output.n_fail += 1;
35                        continue;
36                    }
37                }
38
39                for (idx, program) in programs.iter().enumerate() {
40                    output.push_value(idx, program.execute(&mut vm.as_context())?);
41                }
42                output.n_succ += 1;
43            }
44            None => output.n_fail += 1,
45        }
46    }
47
48    Ok(output)
49}
50
51impl PQLRunner {
52    // TODO: check max selectors
53    // TODO: refactor run logic
54    // TODO: remove
55    #[allow(clippy::missing_panics_doc)]
56    pub fn try_run_stmt(
57        stmt: &ast::Stmt<'_>,
58        max_trials: Option<usize>,
59        n_threads: Option<usize>,
60    ) -> PQLResult<RunnerOutput> {
61        let mut vm = Vm::from_stmt(stmt)?;
62
63        if let Some(n) = max_trials {
64            vm.static_data.n_trails = n;
65        }
66
67        let n_trails = vm.static_data.n_trails;
68
69        let where_program = match &stmt.where_clause {
70            Some(expr) => Some(vm::compile_where(&mut vm, expr)?),
71            None => None,
72        };
73
74        let programs = stmt
75            .selectors
76            .iter()
77            .map(|s| vm::compile_selector(&mut vm, s))
78            .collect::<PQLResult<Vec<_>>>()?;
79
80        // wasm has no threads: spawning panics at runtime, so clamp to 1
81        // and take the direct path below
82        let n_threads = if cfg!(target_family = "wasm") {
83            1
84        } else {
85            n_threads
86                .or_else(|| thread::available_parallelism().map(usize::from).ok())
87                .unwrap_or(1)
88                .clamp(1, n_trails.max(1))
89        };
90
91        if n_threads == 1 {
92            return run_trials(
93                vm,
94                n_trails,
95                where_program.as_ref(),
96                &programs,
97                &stmt.selectors,
98            );
99        }
100
101        let outputs = thread::scope(|scope| {
102            let vm = &vm;
103            let programs = &programs;
104            let where_program = where_program.as_ref();
105            let selectors = &stmt.selectors;
106
107            (0..n_threads)
108                .map(|i| {
109                    // distribute the remainder over the first few threads
110                    let quota = n_trails / n_threads + usize::from(i < n_trails % n_threads);
111
112                    scope.spawn(move || {
113                        run_trials(vm.clone(), quota, where_program, programs, selectors)
114                    })
115                })
116                .collect::<Vec<_>>()
117                .into_iter()
118                .map(|handle| handle.join().unwrap())
119                .collect::<PQLResult<Vec<_>>>()
120        })?;
121
122        Ok(outputs
123            .into_iter()
124            .reduce(|mut acc, output| {
125                acc.merge(output);
126                acc
127            })
128            .unwrap())
129    }
130
131    // tmp function
132    pub fn run<S: io::Write, T: io::Write>(
133        src: &str,
134        max_trials: Option<usize>,
135        n_threads: Option<usize>,
136        stream_out: &mut S,
137        stream_err: &mut T,
138    ) -> io::Result<()> {
139        match parse_pql(src) {
140            Ok(stmts) => {
141                for (i, stmt) in stmts.iter().enumerate() {
142                    if i > 0 {
143                        writeln!(stream_out, "{:-<80}", "")?;
144                    }
145
146                    match Self::try_run_stmt(stmt, max_trials, n_threads) {
147                        Ok(output) => {
148                            output.report_to_stream(stmt, stream_out)?;
149                            writeln!(stream_out, "{} trials", output.n_succ)?;
150                        }
151                        Err(err) => {
152                            writeln!(stream_err, "{err:?} {}", &src[err.loc.0..err.loc.1])?;
153                        }
154                    }
155                }
156            }
157            Err(err) => writeln!(stream_err, "{err:?}")?,
158        }
159        Ok(())
160    }
161}
opql/runner.rs

opql/
runner.rs
