comnum_sim/
runner.rs

1//! The runner: the `Trainer` of a Monte Carlo campaign.
2//!
3//! [`Runner::run`] takes a [`Scenario`] and a [`Grid`] and, for every condition in
4//! order:
5//!
6//! 1. asks the callbacks whether to run it ([`Callback::on_condition_start`]);
7//! 2. builds the state with [`Scenario::setup`];
8//! 3. runs batches of `batch` trials in parallel, trial `t` drawing from
9//!    `stream(seed, name, condition index, t)`, merges their records in trial order,
10//!    and after every batch evaluates the stopping rule and calls
11//!    [`Callback::on_batch_end`]; after the first batch it checks that the targets of
12//!    the rule are recorded ([`Stop::check_metrics`]), and the last batch ends at the
13//!    bound of the rule on the trials ([`Stop::trial_limit`]);
14//! 4. hands the [`ConditionResult`] to the callbacks and the loggers.
15//!
16//! The results are identical whatever the number of threads (see [`crate`]). The batch
17//! size only sets how often the rule is checked (a larger batch keeps every core busy, a
18//! smaller one stops closer to an error or precision target); a bound on the trials is
19//! met exactly, so that `max_trials(1000)` runs 1000 trials with batches of 32 as with
20//! batches of 1.
21//! Every condition runs at least one trial.
22//!
23//! # Examples
24//! ```
25//! use comnum_sim::prelude::*;
26//! use comnum_mapping::{constellation::{Constellation, ConstellationKind}, source as sources};
27//! use comnum_channel::awgn as channels;
28//! use comnum_metrics as metrics;
29//! use comnum_sim::scenario::from_fn;
30//! let coin = from_fn("coin", |cond, rng, rec| {
31//!     let p = cond.f64("p")?;
32//!     rec.errors("heads", (rng.uniform() < p) as u64, 1);
33//!     Ok(())
34//! });
35//! let grid = Grid::new().axis("p", [0.1, 0.5]);
36//! let one = Runner::new().seed(3).threads(1).batch(50).stop(Stop::max_trials(1000))
37//!     .run(&coin, &grid).unwrap();
38//! let four = Runner::new().seed(3).threads(4).batch(50).stop(Stop::max_trials(1000))
39//!     .run(&coin, &grid).unwrap();
40//! assert_eq!(one[0].metric("heads"), four[0].metric("heads"));   // thread-count independent
41//! assert!((one[1].metric("heads").unwrap().value() - 0.5).abs() < 0.05);
42//! ```
43
44use crate::callbacks::{Action, Callback};
45use crate::grid::{Condition, Grid, Value};
46use crate::loggers::Logger;
47use crate::scenario::{Record, Scenario};
48use crate::stats::Estimate;
49use crate::stopping::{Progress, Stop};
50use comnum_core::rng::stream;
51use comnum_core::{Error, Result};
52use rayon::prelude::*;
53use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
54
55/// How a condition ended.
56#[derive(Clone, Debug, PartialEq, Eq)]
57pub enum Status {
58    /// The stopping rule was met.
59    Completed,
60    /// A callback stopped it before the rule was met.
61    Stopped,
62    /// A callback skipped it: no trial was run.
63    Skipped,
64    /// [`Runner::fast_dev_run`]: one batch only.
65    DevRun,
66}
67
68impl Status {
69    /// `completed`, `stopped`, `skipped` or `dev_run`.
70    pub fn as_str(&self) -> &'static str {
71        match self {
72            Status::Completed => "completed",
73            Status::Stopped => "stopped",
74            Status::Skipped => "skipped",
75            Status::DevRun => "dev_run",
76        }
77    }
78}
79
80/// The result of one condition.
81#[derive(Clone, Debug, PartialEq)]
82pub struct ConditionResult {
83    /// Index of the condition in the grid.
84    pub index: usize,
85    /// The condition.
86    pub condition: Condition,
87    /// How it ended.
88    pub status: Status,
89    /// Number of trials run.
90    pub trials: u64,
91    /// Wall time of the condition, setup included (s).
92    pub wall_s: f64,
93    /// The metrics merged over the trials, with the counts of the values dropped
94    /// ([`Record::dropped`]).
95    pub record: Record,
96}
97
98impl ConditionResult {
99    /// The estimate of a metric.
100    pub fn metric(&self, name: &str) -> Option<&Estimate> {
101        self.record.get(name)
102    }
103
104    /// The result as one flat row of named values: `index`, the entries of the
105    /// condition, `status`, `trials`, `wall_s`, then the columns of every metric
106    /// ([`metric_columns`]) with its interval at `confidence`, in the order of the
107    /// names: the columns of the CSV report ([`CsvLogger`](crate::loggers::CsvLogger))
108    /// and the rows of the Python results.
109    ///
110    /// # Examples
111    /// ```
112    /// use comnum_sim::prelude::*;
113    /// use comnum_sim::scenario::from_fn;
114    /// let s = from_fn("coin", |_, rng, rec| { rec.errors("p", rng.bit() as u64, 1); Ok(()) });
115    /// let r = Runner::new().stop(Stop::max_trials(8)).run(&s, &Grid::new().axis("n", [1usize])).unwrap();
116    /// let names: Vec<String> = r[0].row(0.95).into_iter().map(|(k, _)| k).collect();
117    /// assert_eq!(names, ["index", "n", "status", "trials", "wall_s", "p", "p_lo", "p_hi", "p_count", "p_total"]);
118    /// ```
119    pub fn row(&self, confidence: f64) -> Vec<(String, Value)> {
120        let count = |n: u64| Value::I64(i64::try_from(n).unwrap_or(i64::MAX));
121        let mut row = vec![("index".to_string(), count(self.index as u64))];
122        row.extend(self.condition.entries().iter().cloned());
123        row.push(("status".into(), Value::Str(self.status.as_str().into())));
124        row.push(("trials".into(), count(self.trials)));
125        row.push(("wall_s".into(), Value::F64(self.wall_s)));
126        for (k, e) in self.record.iter() {
127            let (lo, hi) = e.interval(confidence);
128            let values = [Value::F64(e.value()), Value::F64(lo), Value::F64(hi), count(e.count()), count(e.total())];
129            row.extend(metric_columns(k).into_iter().zip(values));
130        }
131        row
132    }
133}
134
135/// The names of the columns of the metric `name` in a flat row
136/// ([`ConditionResult::row`]): `name`, `name_lo` and `name_hi` (the estimate and its
137/// confidence interval), `name_count` and `name_total` (the errors and the events of a
138/// rate, the number of values of a mean twice).
139///
140/// # Examples
141/// ```
142/// assert_eq!(comnum_sim::runner::metric_columns("ber"), ["ber", "ber_lo", "ber_hi", "ber_count", "ber_total"]);
143/// ```
144pub fn metric_columns(name: &str) -> [String; 5] {
145    ["", "_lo", "_hi", "_count", "_total"].map(|suffix| format!("{name}{suffix}"))
146}
147
148/// What the callbacks and loggers know about the run.
149#[derive(Clone, Debug, PartialEq)]
150pub struct RunInfo {
151    /// Name of the scenario.
152    pub scenario: String,
153    /// Base seed.
154    pub seed: u64,
155    /// Trials per batch.
156    pub batch: u64,
157    /// Number of worker threads.
158    pub threads: usize,
159    /// The stopping rule.
160    pub stop: String,
161    /// Confidence level of the reported intervals.
162    pub confidence: f64,
163    /// Number of conditions of the grid.
164    pub n_conditions: usize,
165    /// Start of the run (seconds since the Unix epoch).
166    pub started_unix_s: u64,
167    /// Version of the library ([`crate::VERSION`]).
168    pub version: String,
169    /// Commit of the library ([`crate::COMMIT`]).
170    pub commit: String,
171    /// The configuration that produced the run, as a JSON document (a declarative
172    /// specification sets it), if known.
173    pub config: Option<String>,
174}
175
176/// The summary of a finished run.
177#[derive(Clone, Debug, PartialEq)]
178pub struct RunReport {
179    /// The run information.
180    pub info: RunInfo,
181    /// Every condition, in order.
182    pub results: Vec<ConditionResult>,
183    /// Total wall time (s).
184    pub wall_s: f64,
185}
186
187/// The campaign runner (builder; see the module documentation).
188pub struct Runner {
189    seed: u64,
190    batch: u64,
191    threads: usize,
192    stop: Stop,
193    confidence: f64,
194    fast_dev_run: bool,
195    config: Option<String>,
196    callbacks: Vec<Box<dyn Callback>>,
197    loggers: Vec<Box<dyn Logger>>,
198}
199
200impl Default for Runner {
201    fn default() -> Self {
202        Runner::new()
203    }
204}
205
206impl Runner {
207    /// Seed 0, batches of 32 trials, all cores, 100 trials per condition, 95 % intervals.
208    pub fn new() -> Runner {
209        Runner {
210            seed: 0,
211            batch: 32,
212            threads: 0,
213            stop: Stop::max_trials(100),
214            confidence: 0.95,
215            fast_dev_run: false,
216            config: None,
217            callbacks: Vec::new(),
218            loggers: Vec::new(),
219        }
220    }
221
222    /// The configuration that produced the run, a JSON document recorded in the run
223    /// information (and the JSON report) next to the version, the commit and the seed.
224    pub fn config(mut self, json: impl Into<String>) -> Runner {
225        self.config = Some(json.into());
226        self
227    }
228
229    /// The base seed of every stream.
230    pub fn seed(mut self, seed: u64) -> Runner {
231        self.seed = seed;
232        self
233    }
234
235    /// Trials per batch (at least 1): the stopping rule is checked after every batch.
236    pub fn batch(mut self, n: u64) -> Runner {
237        self.batch = n.max(1);
238        self
239    }
240
241    /// Worker threads (0: one per core).
242    pub fn threads(mut self, n: usize) -> Runner {
243        self.threads = n;
244        self
245    }
246
247    /// The stopping rule of every condition.
248    pub fn stop(mut self, rule: Stop) -> Runner {
249        self.stop = rule;
250        self
251    }
252
253    /// Confidence level of the intervals reported by the loggers (default 0.95), in
254    /// $(0, 1)$ ([`Runner::run`] refuses another value).
255    pub fn confidence(mut self, c: f64) -> Runner {
256        self.confidence = c;
257        self
258    }
259
260    /// Run one batch per condition and ignore the stopping rule (a quick check).
261    pub fn fast_dev_run(mut self, on: bool) -> Runner {
262        self.fast_dev_run = on;
263        self
264    }
265
266    /// Add a callback.
267    pub fn callback(mut self, c: impl Callback + 'static) -> Runner {
268        self.callbacks.push(Box::new(c));
269        self
270    }
271
272    /// Add a logger.
273    pub fn logger(mut self, l: impl Logger + 'static) -> Runner {
274        self.loggers.push(Box::new(l));
275        self
276    }
277
278    /// What the callbacks and the loggers know about a run of this runner on the
279    /// scenario `scenario` over `n_conditions` conditions, started now (all the cores
280    /// counted for `threads: 0`): the header of every report, that of a run, of a search
281    /// and of a verification.
282    pub fn run_info(&self, scenario: &str, n_conditions: usize) -> RunInfo {
283        RunInfo {
284            scenario: scenario.to_string(),
285            seed: self.seed,
286            batch: self.batch,
287            threads: if self.threads == 0 { rayon::current_num_threads() } else { self.threads },
288            stop: if self.fast_dev_run { "fast_dev_run".into() } else { self.stop.to_string() },
289            confidence: self.confidence,
290            n_conditions,
291            started_unix_s: SystemTime::now().duration_since(UNIX_EPOCH).map_or(0, |d| d.as_secs()),
292            version: crate::VERSION.into(),
293            commit: crate::COMMIT.into(),
294            config: self.config.clone(),
295        }
296    }
297
298    /// Run the scenario on every condition of the grid.
299    ///
300    /// # Errors
301    /// [`Error::Config`] for a grid that gives a key twice ([`Grid::check`]); for a
302    /// confidence level outside $(0, 1)$ (of the runner or of a precision target, see
303    /// [`check_confidence`](crate::stopping::check_confidence));
304    /// for a stopping rule that can never stop, or whose target is a
305    /// metric that the first batch of a condition does not record (or `min_errors` on a
306    /// mean, see [`Stop::check_metrics`]); the first error of a `setup` or of a trial, of
307    /// the same kind, with the condition and the trial in the message; the errors of the
308    /// loggers.
309    pub fn run<S: Scenario>(mut self, scenario: &S, grid: &Grid) -> Result<Vec<ConditionResult>> {
310        self.check(grid)?;
311        let pool = rayon::ThreadPoolBuilder::new()
312            .num_threads(self.threads)
313            .build()
314            .map_err(|e| Error::Config(format!("thread pool: {e}")))?;
315        let conditions = grid.conditions();
316        let info = RunInfo { threads: pool.current_num_threads(), ..self.run_info(scenario.name(), conditions.len()) };
317        self.start(&info)?;
318        let t_run = Instant::now();
319        let mut results = Vec::with_capacity(conditions.len());
320        let mut stop_run = false;
321        for (index, cond) in conditions.into_iter().enumerate() {
322            let res = self.condition(scenario, index, cond, &pool, stop_run)?;
323            stop_run |= self.end_condition(&res)?;
324            results.push(res);
325        }
326        self.finish(info, results, t_run)
327    }
328
329    /// The checks of [`Runner::run`] before it starts: the confidence levels, the grid and
330    /// a stopping rule that stops.
331    fn check(&self, grid: &Grid) -> Result<()> {
332        crate::stopping::check_confidence(self.confidence, "the runner")?;
333        self.stop.check_confidences()?;
334        grid.check()?;
335        if !self.fast_dev_run && !self.stop.is_bounded() {
336            return Err(Error::Config(format!(
337                "the stopping rule `{}` can run forever: add Stop::max_trials or Stop::max_time with `or`",
338                self.stop
339            )));
340        }
341        Ok(())
342    }
343
344    /// The start of the run, for every callback, then every logger.
345    fn start(&mut self, info: &RunInfo) -> Result<()> {
346        for c in &mut self.callbacks {
347            c.on_run_start(info);
348        }
349        for l in &mut self.loggers {
350            l.start(info)?;
351        }
352        Ok(())
353    }
354
355    /// One condition: skipped if the run was stopped or a callback skips it, run
356    /// otherwise.
357    fn condition<S: Scenario>(
358        &mut self,
359        scenario: &S,
360        index: usize,
361        cond: Condition,
362        pool: &rayon::ThreadPool,
363        stop_run: bool,
364    ) -> Result<ConditionResult> {
365        // every callback sees the start of the condition (no short circuit), unless
366        // the run was stopped
367        let mut skip = stop_run;
368        if !stop_run {
369            for c in &mut self.callbacks {
370                if c.on_condition_start(index, &cond) == Action::Skip {
371                    skip = true;
372                }
373            }
374        }
375        if skip {
376            Ok(ConditionResult {
377                index,
378                condition: cond,
379                status: Status::Skipped,
380                trials: 0,
381                wall_s: 0.0,
382                record: Record::new(),
383            })
384        } else {
385            self.run_condition(scenario, index, cond, pool)
386        }
387    }
388
389    /// The end of a condition, for every callback, then every logger; whether a callback
390    /// stops the run.
391    fn end_condition(&mut self, res: &ConditionResult) -> Result<bool> {
392        let mut stop_run = false;
393        for c in &mut self.callbacks {
394            if c.on_condition_end(res) == Action::StopRun {
395                stop_run = true;
396            }
397        }
398        for l in &mut self.loggers {
399            l.log(res, self.confidence)?;
400        }
401        Ok(stop_run)
402    }
403
404    /// The end of the run, for every callback, then every logger with the wall time since
405    /// `t_run`; the results.
406    fn finish(&mut self, info: RunInfo, results: Vec<ConditionResult>, t_run: Instant) -> Result<Vec<ConditionResult>> {
407        for c in &mut self.callbacks {
408            c.on_run_end(&results);
409        }
410        let report = RunReport { info, results, wall_s: t_run.elapsed().as_secs_f64() };
411        for l in &mut self.loggers {
412            l.finish(&report)?;
413        }
414        Ok(report.results)
415    }
416
417    fn run_condition<S: Scenario>(
418        &mut self,
419        scenario: &S,
420        index: usize,
421        cond: Condition,
422        pool: &rayon::ThreadPool,
423    ) -> Result<ConditionResult> {
424        let t0 = Instant::now();
425        let state = scenario.setup(&cond).map_err(|e| e.context(format!("setup of ({})", cond.label())))?;
426        let name = scenario.name().to_string();
427        let mut record = Record::new();
428        let mut trials = 0u64;
429        // the last batch ends at the bound on the trials (one batch, whole, for a dev run)
430        let limit = if self.fast_dev_run { None } else { self.stop.trial_limit() };
431        let status = loop {
432            let end = limit.map_or(trials + self.batch, |n| (trials + self.batch).min(n).max(trials + 1));
433            let range = trials..end;
434            let seed = self.seed;
435            let recs: Vec<Result<Record>> = pool.install(|| {
436                range
437                    .into_par_iter()
438                    .map(|t| {
439                        let mut rng = stream(seed, &name, index as u64, t);
440                        let mut rec = Record::for_trial(t);
441                        scenario
442                            .trial(&state, &cond, &mut rng, &mut rec)
443                            .map(|()| rec)
444                            .map_err(|e| e.context(format!("trial {t} of ({})", cond.label())))
445                    })
446                    .collect()
447            });
448            for r in recs {
449                record.merge(&r?);
450            }
451            if trials == 0 {
452                self.stop.check_metrics(&record).map_err(|e| e.context(format!("({})", cond.label())))?;
453            }
454            trials = end;
455            let progress = Progress { condition: index, trials, record: &record, elapsed: t0.elapsed() };
456            let mut stopped = false;
457            for c in &mut self.callbacks {
458                if matches!(c.on_batch_end(&progress), Action::StopCondition | Action::Skip) {
459                    stopped = true;
460                }
461            }
462            if self.fast_dev_run {
463                break Status::DevRun;
464            }
465            if self.stop.is_met(&progress) {
466                break Status::Completed;
467            }
468            if stopped {
469                break Status::Stopped;
470            }
471        };
472        Ok(ConditionResult { index, condition: cond, status, trials, wall_s: t0.elapsed().as_secs_f64(), record })
473    }
474}
475
476/// Wall time as a [`Duration`] from seconds (helper of the callbacks): zero for a
477/// negative or NaN value, [`Duration::MAX`] for one too large (`∞`).
478pub(crate) fn secs(s: f64) -> Duration {
479    if s.is_nan() || s <= 0.0 { Duration::ZERO } else { Duration::try_from_secs_f64(s).unwrap_or(Duration::MAX) }
480}
481
482#[cfg(test)]
483mod tests {
484    use super::*;
485    use crate::scenario::from_fn;
486
487    fn coin() -> impl Scenario {
488        from_fn("coin", |_, rng, rec| {
489            rec.errors("heads", u64::from(rng.uniform() < 0.3), 1);
490            rec.value("u", rng.uniform());
491            Ok(())
492        })
493    }
494
495    #[test]
496    fn max_trials_is_met_exactly_whatever_the_batch_and_the_threads() {
497        let grid = Grid::new().axis("x", [1usize]);
498        let run = |batch: u64, threads: usize| {
499            Runner::new()
500                .seed(5)
501                .batch(batch)
502                .threads(threads)
503                .stop(Stop::max_trials(1000))
504                .run(&coin(), &grid)
505                .unwrap()
506        };
507        let reference = run(1000, 1);
508        for (batch, threads) in [(32, 1), (32, 4), (7, 3), (1, 2), (5000, 2)] {
509            let r = run(batch, threads);
510            assert_eq!(r[0].trials, 1000, "batch {batch}");
511            assert_eq!(r[0].record, reference[0].record, "batch {batch}, {threads} threads");
512        }
513        // an error target stops at the end of a batch, the cap exactly
514        let rule = Stop::min_errors("heads", 1_000_000).or(Stop::max_trials(70));
515        let r = Runner::new().batch(32).stop(rule).run(&coin(), &grid).unwrap();
516        assert_eq!(r[0].trials, 70);
517        // at least one trial
518        let r = Runner::new().batch(32).stop(Stop::max_trials(0)).run(&coin(), &grid).unwrap();
519        assert_eq!(r[0].trials, 1);
520    }
521
522    #[test]
523    fn a_stopping_target_that_is_not_recorded_is_an_error() {
524        let grid = Grid::new().axis("x", [1usize, 2]);
525        let rule = Stop::min_errors("ber", 100).or(Stop::max_trials(10_000));
526        let e = Runner::new().batch(4).stop(rule).run(&coin(), &grid).unwrap_err().to_string();
527        assert!(e.contains("`ber`") && e.contains("recorded: heads, u") && e.contains("x=1"), "{e}");
528        let rule = Stop::min_errors("u", 100).or(Stop::max_trials(10_000));
529        let e = Runner::new().batch(4).stop(rule).run(&coin(), &grid).unwrap_err().to_string();
530        assert!(e.contains("is a mean"), "{e}");
531        // a dev run (`comnum check`) checks it too
532        let rule = Stop::precision("evm", 0.1).or(Stop::max_trials(10));
533        assert!(Runner::new().fast_dev_run(true).stop(rule).run(&coin(), &grid).is_err());
534    }
535
536    #[test]
537    fn a_confidence_outside_zero_one_is_an_error() {
538        let grid = Grid::new();
539        for c in [95.0, 0.0, 1.0, f64::NAN] {
540            let e = Runner::new().confidence(c).stop(Stop::max_trials(1)).run(&coin(), &grid).unwrap_err();
541            assert!(e.to_string().contains("between 0 and 1"), "{c}: {e}");
542        }
543        let rule = Stop::precision_at("u", 0.1, 95.0).or(Stop::max_trials(4));
544        assert!(Runner::new().stop(rule).run(&coin(), &grid).is_err());
545        let rule = Stop::precision_at("u", 1e-9, 0.5).or(Stop::max_trials(4));
546        assert_eq!(Runner::new().batch(2).stop(rule).run(&coin(), &grid).unwrap()[0].trials, 4);
547    }
548
549    #[test]
550    fn a_callback_stops_a_condition_after_a_batch() {
551        struct Once;
552        impl Callback for Once {
553            fn on_batch_end(&mut self, _: &Progress<'_>) -> Action {
554                Action::StopCondition
555            }
556        }
557        let r = Runner::new().batch(8).stop(Stop::max_trials(100)).callback(Once).run(&coin(), &Grid::new()).unwrap();
558        assert_eq!((r[0].status.clone(), r[0].trials), (Status::Stopped, 8));
559        assert_eq!(Stop::max_trials(5).and(Stop::min_trials(2)).to_string(), "(max_trials(5) and min_trials(2))");
560    }
561
562    #[test]
563    fn seconds_never_panic() {
564        assert_eq!(secs(f64::INFINITY), Duration::MAX);
565        assert_eq!(secs(f64::NAN), Duration::ZERO);
566        assert_eq!(secs(-1.0), Duration::ZERO);
567        assert_eq!(secs(1.5), Duration::from_millis(1500));
568    }
569
570    #[test]
571    fn the_dropped_values_reach_the_result() {
572        let s = from_fn("inf", |_, rng, rec| {
573            rec.value("snr_db", if rng.uniform() < 0.5 { f64::INFINITY } else { 10.0 });
574            Ok(())
575        });
576        let r = Runner::new().batch(8).stop(Stop::max_trials(64)).run(&s, &Grid::new()).unwrap();
577        let kept = r[0].metric("snr_db").unwrap().count();
578        assert!(kept > 0 && r[0].record.dropped("snr_db") > 0);
579        assert_eq!(kept + r[0].record.dropped("snr_db"), 64);
580    }
581}
comnum_sim/runner.rs

comnum_sim/
runner.rs
