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        let n_threads = n_threads
81            .or_else(|| thread::available_parallelism().map(usize::from).ok())
82            .unwrap_or(1)
83            .clamp(1, n_trails.max(1));
84
85        let outputs = thread::scope(|scope| {
86            let vm = &vm;
87            let programs = &programs;
88            let where_program = where_program.as_ref();
89            let selectors = &stmt.selectors;
90
91            (0..n_threads)
92                .map(|i| {
93                    // distribute the remainder over the first few threads
94                    let quota = n_trails / n_threads + usize::from(i < n_trails % n_threads);
95
96                    scope.spawn(move || {
97                        run_trials(vm.clone(), quota, where_program, programs, selectors)
98                    })
99                })
100                .collect::<Vec<_>>()
101                .into_iter()
102                .map(|handle| handle.join().unwrap())
103                .collect::<PQLResult<Vec<_>>>()
104        })?;
105
106        Ok(outputs
107            .into_iter()
108            .reduce(|mut acc, output| {
109                acc.merge(output);
110                acc
111            })
112            .unwrap())
113    }
114
115    // tmp function
116    pub fn run<S: io::Write, T: io::Write>(
117        src: &str,
118        max_trials: Option<usize>,
119        n_threads: Option<usize>,
120        stream_out: &mut S,
121        stream_err: &mut T,
122    ) -> io::Result<()> {
123        match parse_pql(src) {
124            Ok(stmts) => {
125                for (i, stmt) in stmts.iter().enumerate() {
126                    if i > 0 {
127                        writeln!(stream_out, "{:-<80}", "")?;
128                    }
129
130                    match Self::try_run_stmt(stmt, max_trials, n_threads) {
131                        Ok(output) => {
132                            output.report_to_stream(stmt, stream_out)?;
133                            writeln!(stream_out, "{} trials", output.n_succ)?;
134                        }
135                        Err(err) => {
136                            writeln!(stream_err, "{err:?} {}", &src[err.loc.0..err.loc.1])?;
137                        }
138                    }
139                }
140            }
141            Err(err) => writeln!(stream_err, "{err:?}")?,
142        }
143        Ok(())
144    }
145}
opql/runner.rs

opql/
runner.rs
