cimoxide-validation 0.6.2

//! Splitting a walk over a dataset across threads.
//!
//! Validation is memory-bound — each element costs a cache miss into its own
//! field map — and threads overlap those misses well. Work is cut into
//! contiguous runs and the results concatenated in run order, so the output is
//! exactly what one thread produces, order included.

/// Whether this target can start threads. wasm32-unknown-unknown compiles `std::thread` and panics
/// when it is used, so everything here runs in turn there. As results are concatenated in order,
/// the output is the same either way.
const HAS_THREADS: bool = !cfg!(target_family = "wasm");

/// Below this many elements a walk stays on the calling thread.
const PARALLEL_MIN: usize = 20_000;
/// Past a handful of threads a walk is limited by memory, not by cores.
const MAX_THREADS: usize = 8;

/// How many threads a walk over `total` elements gets.
pub(crate) fn threads_for(total: usize) -> usize {
    if total < PARALLEL_MIN || !HAS_THREADS {
        1
    } else {
        std::thread::available_parallelism().map_or(1, |n| n.get()).min(MAX_THREADS)
    }
}

/// Cut `items` into at most `threads` contiguous runs of about equal `weight`.
pub(crate) fn runs<T>(items: &[T], threads: usize, weight: impl Fn(&T) -> usize) -> Vec<&[T]> {
    let total: usize = items.iter().map(&weight).sum();
    let per_thread = total.div_ceil(threads.max(1)).max(1);
    let mut out = Vec::with_capacity(threads);
    let (mut start, mut size) = (0, 0);
    for (i, item) in items.iter().enumerate() {
        size += weight(item);
        if size >= per_thread {
            out.push(&items[start..=i]);
            (start, size) = (i + 1, 0);
        }
    }
    if start < items.len() {
        out.push(&items[start..]);
    }
    out
}

/// Run `f` over each run on its own thread and return the results in run order.
pub(crate) fn par_map<T: Sync, R: Send>(runs: &[&[T]], f: impl Fn(&[T]) -> R + Sync) -> Vec<R> {
    match runs {
        [one] => vec![f(one)],
        _ if !HAS_THREADS => runs.iter().map(|run| f(run)).collect(),
        _ => std::thread::scope(|s| {
            let handles: Vec<_> = runs.iter().map(|run| s.spawn(|| f(run))).collect();
            handles.into_iter().map(|h| h.join().expect("validation thread panicked")).collect()
        }),
    }
}

/// [`par_map`], concatenating the results.
pub(crate) fn par_concat<T: Sync, R: Send>(runs: &[&[T]], f: impl Fn(&[T]) -> Vec<R> + Sync) -> Vec<R> {
    par_map(runs, f).into_iter().flatten().collect()
}

/// Run `f` on each item on its own thread, and return the results in item order.
pub(crate) fn par_each<T: Sync, R: Send>(items: &[T], f: impl Fn(&T) -> R + Sync) -> Vec<R> {
    if !HAS_THREADS {
        return items.iter().map(f).collect();
    }
    std::thread::scope(|s| {
        let handles: Vec<_> = items.iter().map(|item| s.spawn(|| f(item))).collect();
        handles.into_iter().map(|h| h.join().expect("validation thread panicked")).collect()
    })
}

/// Run `first` on a thread of its own and `second` on this one, and return both.
pub(crate) fn par_two<A: Send, B: Send>(first: impl FnOnce() -> A + Send, second: impl FnOnce() -> B) -> (A, B) {
    if !HAS_THREADS {
        let a = first();
        return (a, second());
    }
    std::thread::scope(|s| {
        let first = s.spawn(first);
        let b = second();
        (first.join().expect("validation thread panicked"), b)
    })
}

/// Run independent rule groups over one dataset on their own threads, and
/// concatenate their results in the order given.
pub(crate) fn par_groups(
    dataset: &cimmodel::CimDataset,
    groups: &[fn(&cimmodel::CimDataset) -> Vec<crate::Violation>],
) -> Vec<crate::Violation> {
    if !HAS_THREADS {
        return groups.iter().flat_map(|g| g(dataset)).collect();
    }
    std::thread::scope(|s| {
        let handles: Vec<_> = groups.iter().map(|g| s.spawn(move || g(dataset))).collect();
        handles
            .into_iter()
            .flat_map(|h| h.join().expect("validation thread panicked"))
            .collect()
    })
}

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

    /// The runs, concatenated, are the input — nothing dropped, repeated or
    /// reordered. That is what makes a parallel walk's output the sequential one.
    #[test]
    fn runs_cover_the_input_in_order() {
        let items: Vec<usize> = (0..1000).collect();
        for threads in [1, 2, 3, 7, 8, 16, 2000] {
            let rs = runs(&items, threads, |_| 1);
            assert!(rs.len() <= threads.max(1), "{threads} threads gave {} runs", rs.len());
            assert!(rs.iter().all(|r| !r.is_empty()));
            let joined: Vec<usize> = rs.concat();
            assert_eq!(joined, items, "{threads} threads");
        }
    }

    #[test]
    fn runs_balance_by_weight() {
        // One heavy item and many light ones: the heavy one gets a run of its own.
        let items = [100, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1];
        let rs = runs(&items, 2, |w| *w);
        assert_eq!(rs.concat(), items);
        assert_eq!(rs[0], &[100]);
    }

    #[test]
    fn empty_input_has_no_runs() {
        assert!(runs::<u8>(&[], 4, |_| 1).is_empty());
    }
}