trit_core/
parallel.rs

1//! 轻量并行映射：按块切分，结果顺序与串行一致；不引入第三方线程池。
2//!
3//! 供 ATPG、桥接 SAT、策略断言等独立任务复用。
4
5/// 把请求的线程数收敛到 `[1, tasks]`；`0` 表示自动，wasm32 强制串行。
6pub fn resolve_jobs(requested: usize, tasks: usize) -> usize {
7    if tasks <= 1 {
8        return 1;
9    }
10    #[cfg(target_arch = "wasm32")]
11    {
12        let _ = requested;
13        1
14    }
15    #[cfg(not(target_arch = "wasm32"))]
16    {
17        let auto = std::thread::available_parallelism()
18            .map(std::num::NonZeroUsize::get)
19            .unwrap_or(1);
20        let jobs = if requested == 0 { auto } else { requested };
21        jobs.clamp(1, tasks)
22    }
23}
24
25/// 对 `tasks` 逐项执行 `f`，按原顺序收集结果。
26///
27/// 少于 2 个任务或 `jobs == 1` 时直接串行，避免小输入的线程开销。
28pub fn map<T, R, E, F>(tasks: &[T], requested_jobs: usize, f: F) -> Result<Vec<R>, E>
29where
30    T: Sync,
31    R: Send,
32    E: Send,
33    F: Fn(&T) -> Result<R, E> + Sync,
34{
35    let jobs = resolve_jobs(requested_jobs, tasks.len());
36    if jobs <= 1 {
37        return tasks.iter().map(f).collect();
38    }
39    let chunk = tasks.len().div_ceil(jobs);
40    std::thread::scope(|scope| {
41        let f = &f;
42        let handles: Vec<_> = tasks
43            .chunks(chunk)
44            .map(|chunk| scope.spawn(move || chunk.iter().map(f).collect::<Result<Vec<_>, E>>()))
45            .collect();
46        let mut out = Vec::with_capacity(tasks.len());
47        for handle in handles {
48            match handle.join() {
49                Ok(part) => out.extend(part?),
50                Err(payload) => std::panic::resume_unwind(payload),
51            }
52        }
53        Ok(out)
54    })
55}
56
57#[cfg(test)]
58mod tests {
59    use super::*;
60
61    #[test]
62    fn resolves_jobs_within_bounds() {
63        assert_eq!(resolve_jobs(0, 0), 1);
64        assert_eq!(resolve_jobs(0, 1), 1);
65        assert_eq!(resolve_jobs(1, 100), 1);
66        assert_eq!(resolve_jobs(4, 3), 3);
67        assert!(resolve_jobs(0, 100) >= 1);
68    }
69
70    #[test]
71    fn parallel_matches_serial_and_preserves_order() {
72        let tasks: Vec<usize> = (0..100).collect();
73        let f = |value: &usize| Ok::<usize, ()>(value * 2);
74        let serial = map(&tasks, 1, f).unwrap();
75        let parallel = map(&tasks, 4, f).unwrap();
76        assert_eq!(serial, parallel);
77        assert_eq!(serial[0], 0);
78        assert_eq!(serial[99], 198);
79    }
80
81    #[test]
82    fn parallel_propagates_the_first_error() {
83        let tasks: Vec<usize> = (0..50).collect();
84        let result = map(&tasks, 4, |value| {
85            if *value == 17 { Err("bad") } else { Ok(*value) }
86        });
87        assert_eq!(result.unwrap_err(), "bad");
88    }
89}
trit_core/parallel.rs

trit_core/
parallel.rs
