kshana 0.32.0

Open, reproducible PNT-resilience simulator with quantum-sensor performance models
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121

// SPDX-License-Identifier: AGPL-3.0-only
//! A batch runner: apply a closure to many inputs on worker threads.
//!
//! Built on `std::thread::scope` and channels only. At most `workers` inputs are in flight
//! at once (the work queue is a bounded `sync_channel`), each worker owns its own buffers
//! through the closure, and results come back in **input order** whatever order the
//! workers finish in, so a batch run is deterministic given a deterministic closure. A
//! panic inside the closure is caught and reported as that input's error; the other inputs
//! still run.

use super::report::BatchResult;
use std::panic::{catch_unwind, AssertUnwindSafe};
use std::path::{Path, PathBuf};
use std::sync::{mpsc, Mutex};

/// The worker count used for `workers = 0`: the machine's available parallelism.
pub fn default_workers() -> usize {
    std::thread::available_parallelism()
        .map(|n| n.get())
        .unwrap_or(1)
}

fn panic_message(p: Box<dyn std::any::Any + Send>) -> String {
    if let Some(s) = p.downcast_ref::<&str>() {
        format!("panicked: {s}")
    } else if let Some(s) = p.downcast_ref::<String>() {
        format!("panicked: {s}")
    } else {
        "panicked".to_string()
    }
}

/// Apply `f` to every item on up to `workers` threads (0 = [`default_workers`]); the
/// results are returned in the order of `items`.
pub fn run_batch_items<I, T, F>(items: &[I], workers: usize, f: F) -> Vec<Result<T, String>>
where
    I: Sync,
    T: Send,
    F: Fn(&I) -> Result<T, String> + Sync,
{
    let n = items.len();
    if n == 0 {
        return Vec::new();
    }
    let workers = if workers == 0 {
        default_workers()
    } else {
        workers
    }
    .clamp(1, n);
    let (work_tx, work_rx) = mpsc::sync_channel::<usize>(workers);
    let work_rx = Mutex::new(work_rx);
    let (res_tx, res_rx) = mpsc::channel::<(usize, Result<T, String>)>();
    let mut slots: Vec<Option<Result<T, String>>> = (0..n).map(|_| None).collect();
    std::thread::scope(|s| {
        for _ in 0..workers {
            let res_tx = res_tx.clone();
            let (work_rx, f) = (&work_rx, &f);
            s.spawn(move || loop {
                let next = work_rx.lock().map(|rx| rx.recv());
                let Ok(Ok(i)) = next else { break };
                let r = catch_unwind(AssertUnwindSafe(|| f(&items[i])))
                    .unwrap_or_else(|p| Err(panic_message(p)));
                if res_tx.send((i, r)).is_err() {
                    break;
                }
            });
        }
        drop(res_tx);
        for i in 0..n {
            if work_tx.send(i).is_err() {
                break;
            }
        }
        drop(work_tx);
        for (i, r) in res_rx.iter() {
            slots[i] = Some(r);
        }
    });
    slots
        .into_iter()
        .map(|r| r.unwrap_or_else(|| Err("worker exited before processing this input".into())))
        .collect()
}

/// Apply `f` to every file on up to `workers` threads (0 = [`default_workers`]),
/// returning one [`BatchResult`] per path in the order given.
pub fn run_batch<T, F>(paths: &[PathBuf], workers: usize, f: F) -> Vec<BatchResult<T>>
where
    T: Send,
    F: Fn(&Path) -> Result<T, String> + Sync,
{
    run_batch_items(paths, workers, |p| f(p))
        .into_iter()
        .enumerate()
        .map(|(index, outcome)| BatchResult {
            index,
            input: paths[index].display().to_string(),
            outcome,
        })
        .collect()
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn a_panicking_input_is_an_error_and_the_rest_still_run() {
        let items: Vec<u32> = (0..10).collect();
        let out = run_batch_items(&items, 3, |&x| {
            if x == 4 {
                panic!("four");
            }
            Ok(x * 2)
        });
        assert_eq!(out.len(), 10);
        assert!(out[4].as_ref().unwrap_err().contains("four"));
        assert_eq!(out[9], Ok(18));
    }
}