weavatrix_scan/
parallel.rs

1use crate::pool::ThreadPool;
2use crate::walker::{ErrorPolicy, WalkEntry, WalkError, WalkOptions};
3use std::path::PathBuf;
4use std::sync::{Arc, mpsc};
5
6mod collect;
7mod visit;
8mod visit_worker;
9
10use collect::{DirectoryTask, collect_lane, collect_serial, collect_shallow};
11pub use visit::{ParallelVisitReport, WalkControl, WalkEvent};
12
13/// Collected output from a parallel filesystem walk.
14#[derive(Debug)]
15pub struct ParallelWalkReport {
16    pub entries: Vec<WalkEntry>,
17    pub errors: Vec<WalkError>,
18}
19
20/// A breadth-oriented parallel walker for wide repository trees.
21///
22/// Top-level directories are distributed across bounded serial walkers. This
23/// keeps the serial `Walker` API streaming and gives broad trees an explicit
24/// parallel mode without making scanner output order nondeterministic.
25/// Worker completion order does not affect report order. Callers that need a
26/// cross-filesystem stable manifest must still sort, as `Scanner` does.
27/// Link-following falls back to the serial walker so cycle detection has one
28/// authoritative seen set.
29pub struct ParallelWalker {
30    pub(super) root: PathBuf,
31    pub(super) options: WalkOptions,
32    pub(super) parallelism: usize,
33}
34
35impl ParallelWalker {
36    #[must_use]
37    pub fn new(root: impl Into<PathBuf>) -> Self {
38        Self {
39            root: root.into(),
40            options: WalkOptions::default(),
41            parallelism: 0,
42        }
43    }
44
45    #[must_use]
46    pub const fn options(mut self, options: WalkOptions) -> Self {
47        self.options = options;
48        self
49    }
50
51    /// Sets worker count. Zero selects available parallelism.
52    #[must_use]
53    pub const fn with_parallelism(mut self, parallelism: usize) -> Self {
54        self.parallelism = parallelism;
55        self
56    }
57
58    /// Walks the tree using bounded top-level parallelism.
59    ///
60    /// # Errors
61    ///
62    /// With `ErrorPolicy::Abort`, returns the first local traversal error.
63    /// With `ErrorPolicy::Continue`, local errors are collected in the report.
64    ///
65    /// # Panics
66    ///
67    /// Panics if an internal worker panics or shared traversal state is poisoned.
68    pub fn walk(mut self) -> Result<ParallelWalkReport, WalkError> {
69        self.options = self.options.normalized();
70        if self.options.follow_links {
71            return collect_serial(&self.root, self.options);
72        }
73        let shallow = collect_shallow(&self.root, self.options)?;
74        let mut entries = shallow.entries;
75        let mut errors = shallow.errors;
76        let tasks = shallow.tasks;
77        let parallel_root = shallow.root;
78        if self.options.error_policy == ErrorPolicy::Abort && !errors.is_empty() {
79            return Err(errors.remove(0));
80        }
81        if tasks.is_empty() {
82            return Ok(ParallelWalkReport { entries, errors });
83        }
84
85        let pool = ThreadPool::global();
86        let worker_count =
87            parallel_worker_count(self.parallelism, self.options.max_open, tasks.len());
88        let task_count = tasks.len();
89        let mut lanes = (0..worker_count)
90            .map(|_| Vec::<DirectoryTask>::new())
91            .collect::<Vec<_>>();
92        for (index, task) in tasks.into_iter().enumerate() {
93            lanes[index % worker_count].push(task);
94        }
95        let (sender, receiver) = mpsc::channel();
96        for (index, lane) in lanes.into_iter().enumerate() {
97            let sender = sender.clone();
98            let root = Arc::clone(&parallel_root);
99            let options = self.options;
100            pool.execute(move || {
101                let report = collect_lane(lane, options, &root);
102                let _ = sender.send((index, report));
103            });
104        }
105        drop(sender);
106        let mut completed = (0..worker_count).map(|_| None).collect::<Vec<_>>();
107        for (index, report) in receiver {
108            completed[index] = Some(report);
109        }
110        let (additional_entries, additional_errors) =
111            completed
112                .iter()
113                .flatten()
114                .fold((0, 0), |(entries, errors), lane| {
115                    (
116                        entries + lane.report.entries.len(),
117                        errors + lane.report.errors.len(),
118                    )
119                });
120        entries.reserve(additional_entries);
121        errors.reserve(additional_errors);
122        let mut lanes = completed
123            .into_iter()
124            .map(|lane| {
125                let lane = lane.expect("every parallel lane reports completion");
126                (
127                    lane.report.entries.into_iter(),
128                    lane.report.errors.into_iter(),
129                    lane.segments,
130                )
131            })
132            .collect::<Vec<_>>();
133        for task_index in 0..task_count {
134            let (lane_entries, lane_errors, segments) = &mut lanes[task_index % worker_count];
135            let segment = segments
136                .pop_front()
137                .expect("every directory task has an output segment");
138            entries.extend(lane_entries.by_ref().take(segment.entries));
139            errors.extend(lane_errors.by_ref().take(segment.errors));
140        }
141        if self.options.error_policy == ErrorPolicy::Abort && !errors.is_empty() {
142            return Err(errors.remove(0));
143        }
144        Ok(ParallelWalkReport { entries, errors })
145    }
146}
147
148pub(super) fn parallel_worker_count(parallelism: usize, max_open: usize, tasks: usize) -> usize {
149    let available = ThreadPool::global().workers();
150    let requested = if parallelism == 0 {
151        available
152    } else {
153        parallelism
154    };
155    requested.min(max_open.max(1)).min(tasks).max(1)
156}
weavatrix_scan/parallel.rs

weavatrix_scan/
parallel.rs
