patterns/
lanes.rs

1use patterns_macros::because;
2
3pub fn parted<T>(items: &[T], lanes: usize) -> Vec<&[T]> {
4    if lanes == 0 || items.is_empty() {
5        return Vec::new();
6    }
7    items.chunks(items.len().div_ceil(lanes)).collect()
8}
9because!(parted, "a sequence cut into at most that many pieces of near equal size, rounding the piece up so the last one is the short one rather than an extra piece beyond the count asked for; a request for no pieces gives none rather than dividing by nothing");
10
11pub fn lanes_here() -> usize {
12    std::thread::available_parallelism()
13        .map(|n| n.get())
14        .unwrap_or(1)
15}
16because!(lanes_here, "how many pieces of work this machine will really run at once, asked in one place so a caller neither hard codes a count nor has to decide what to do when the machine declines to answer, in which case one lane is the sequential run every caller already expects to be correct");
17
18pub fn across<T: Sync, R: Send>(items: &[T], work: fn(&T) -> R) -> Vec<R> {
19    let mut out: Vec<R> = Vec::with_capacity(items.len());
20    std::thread::scope(|lane| {
21        let mut running = Vec::new();
22        for part in parted(items, lanes_here()) {
23            running.push(lane.spawn(move || part.iter().map(work).collect::<Vec<R>>()));
24        }
25        for one in running {
26            match one.join() {
27                Ok(done) => out.extend(done),
28                Err(panicked) => std::panic::resume_unwind(panicked),
29            }
30        }
31    });
32    out
33}
34because!(across, "every item worked on over the whole machine, answering in the order the items came rather than the order the lanes finished, so a caller may swap it for map and read the same answer; a lane that panics is raised again here rather than swallowed, because a silent lane is a test that passed by not running");
patterns/lanes.rs

patterns/
lanes.rs
