weavatrix_scan/
parallel.rs

1use crate::FileIdentity;
2use crate::runtime::ParallelRuntime;
3use crate::walker::{ErrorPolicy, WalkEntry, WalkError, WalkOperation, WalkOptions};
4use std::path::PathBuf;
5use std::sync::{Arc, Mutex, mpsc};
6
7mod collect;
8pub(crate) mod dynamic;
9mod ordered_pull;
10mod pull;
11mod visit;
12mod visit_worker;
13
14use collect::{DirectoryTask, collect_lane, collect_serial, collect_shallow, expand_frontier};
15pub use pull::ParallelWalkIter;
16pub use visit::{ParallelVisitReport, WalkControl, WalkEvent};
17
18/// Collected output from a parallel filesystem walk.
19#[derive(Debug)]
20pub struct ParallelWalkReport {
21    pub entries: Vec<WalkEntry>,
22    pub errors: Vec<WalkError>,
23}
24
25/// An adaptive parallel filesystem walker.
26///
27/// Broad root frontiers use low-overhead lane traversal. Narrow trees use
28/// dynamic directory scheduling so work below one top-level directory can use
29/// every worker. Link-following carries an immutable ancestry set with each
30/// directory task so aliases remain parallel without losing cycle detection.
31pub struct ParallelWalker {
32    pub(super) root: PathBuf,
33    pub(super) options: WalkOptions,
34    pub(super) parallelism: usize,
35    pub(super) skip_stdout: Option<FileIdentity>,
36    pub(super) runtime: ParallelRuntime,
37}
38
39impl ParallelWalker {
40    #[must_use]
41    pub fn new(root: impl Into<PathBuf>) -> Self {
42        Self {
43            root: root.into(),
44            options: WalkOptions::default(),
45            parallelism: 0,
46            skip_stdout: None,
47            runtime: ParallelRuntime::global(),
48        }
49    }
50
51    #[must_use]
52    pub const fn options(mut self, options: WalkOptions) -> Self {
53        self.options = options;
54        self
55    }
56
57    /// Sets worker count. Zero selects available parallelism.
58    #[must_use]
59    pub const fn with_parallelism(mut self, parallelism: usize) -> Self {
60        self.parallelism = parallelism;
61        self
62    }
63
64    /// Selects the global, dedicated, or application-owned executor.
65    #[must_use]
66    pub fn runtime(mut self, runtime: ParallelRuntime) -> Self {
67        self.runtime = runtime;
68        self
69    }
70
71    /// Skips a regular file that refers to redirected standard output.
72    ///
73    /// This applies consistently to collected, visitor and pull APIs and
74    /// prevents feedback loops when command output is written inside a walked
75    /// tree.
76    #[must_use]
77    pub fn skip_stdout(mut self, enabled: bool) -> Self {
78        self.skip_stdout = enabled.then(crate::stdout::identity).flatten();
79        self
80    }
81
82    /// Walks the tree using bounded adaptive scheduling.
83    ///
84    /// # Errors
85    ///
86    /// With `ErrorPolicy::Abort`, returns the first observed local traversal
87    /// error. With `ErrorPolicy::Continue`, local errors are collected.
88    ///
89    /// # Panics
90    ///
91    /// Panics if an internal worker panics or shared traversal state is
92    /// poisoned.
93    pub fn walk(mut self) -> Result<ParallelWalkReport, WalkError> {
94        self.options = self.options.normalized();
95        let skip_stdout = self.skip_stdout;
96        if self.runtime.is_worker_thread() {
97            return collect_serial(&self.root, self.options)
98                .map(|report| without_stdout(report, skip_stdout));
99        }
100        if self.options.follow_links {
101            return self
102                .walk_dynamic()
103                .map(|report| without_stdout(report, skip_stdout));
104        }
105        let mut shallow = collect_shallow(&self.root, self.options)?;
106        if self.options.error_policy == ErrorPolicy::Abort && !shallow.errors.is_empty() {
107            return Err(shallow.errors.into_iter().next().expect("error exists"));
108        }
109        if !self.options.same_file_system {
110            let target = requested_workers(&self.runtime, self.parallelism, self.options.max_open)
111                .saturating_mul(FRONTIER_TASKS_PER_WORKER);
112            if shallow.tasks.len() < target {
113                shallow = expand_frontier(shallow, self.options, target);
114            }
115        }
116        if shallow.tasks.is_empty() {
117            return Ok(without_stdout(
118                ParallelWalkReport {
119                    entries: shallow.entries,
120                    errors: shallow.errors,
121                },
122                skip_stdout,
123            ));
124        }
125        self.walk_lanes(shallow)
126            .map(|report| without_stdout(report, skip_stdout))
127    }
128
129    fn walk_dynamic(self) -> Result<ParallelWalkReport, WalkError> {
130        let report = Arc::new(Mutex::new(ParallelWalkReport {
131            entries: Vec::new(),
132            errors: Vec::new(),
133        }));
134        let visitor_report = Arc::clone(&report);
135        dynamic::stream_batched(
136            &self.root,
137            self.options,
138            self.parallelism,
139            &self.runtime,
140            &crate::CancellationToken::new(),
141            move |entries, errors| {
142                let mut report = visitor_report
143                    .lock()
144                    .expect("parallel walk report is not poisoned");
145                report.entries.extend(entries);
146                report.errors.extend(errors.iter().map(copy_walk_error));
147                true
148            },
149        )?;
150        let mut report = Arc::try_unwrap(report)
151            .expect("parallel walk visitor released report")
152            .into_inner()
153            .expect("parallel walk report is not poisoned");
154        report
155            .entries
156            .sort_unstable_by(|left, right| left.path().cmp(right.path()));
157        report.errors.sort_unstable_by(|left, right| {
158            left.path()
159                .cmp(right.path())
160                .then_with(|| left.depth().cmp(&right.depth()))
161        });
162        Ok(report)
163    }
164
165    fn walk_lanes(self, shallow: collect::ShallowWalk) -> Result<ParallelWalkReport, WalkError> {
166        let mut entries = shallow.entries;
167        let mut errors = shallow.errors;
168        let task_count = shallow.tasks.len();
169        let worker_count = parallel_worker_count(
170            &self.runtime,
171            self.parallelism,
172            self.options.max_open,
173            task_count,
174        );
175        let mut lanes = (0..worker_count)
176            .map(|_| Vec::<DirectoryTask>::new())
177            .collect::<Vec<_>>();
178        for (index, task) in shallow.tasks.into_iter().enumerate() {
179            lanes[index % worker_count].push(task);
180        }
181        let (sender, receiver) = mpsc::channel();
182        for (index, lane) in lanes.into_iter().enumerate() {
183            let sender = sender.clone();
184            let root = Arc::clone(&shallow.root);
185            let options = self.options;
186            self.runtime
187                .try_execute(move || {
188                    let report = collect_lane(lane, options, &root);
189                    let _ = sender.send((index, report));
190                })
191                .map_err(|source| schedule_error(&self.root, source))?;
192        }
193        drop(sender);
194        let mut completed = (0..worker_count).map(|_| None).collect::<Vec<_>>();
195        for (index, report) in receiver {
196            completed[index] = Some(report);
197        }
198        let mut lanes = completed
199            .into_iter()
200            .map(|lane| {
201                let lane = lane.expect("every parallel lane reports completion");
202                (
203                    lane.report.entries.into_iter(),
204                    lane.report.errors.into_iter(),
205                    lane.segments,
206                )
207            })
208            .collect::<Vec<_>>();
209        for task_index in 0..task_count {
210            let (lane_entries, lane_errors, segments) = &mut lanes[task_index % worker_count];
211            let segment = segments
212                .pop_front()
213                .expect("every directory task has an output segment");
214            entries.extend(lane_entries.by_ref().take(segment.entries));
215            errors.extend(lane_errors.by_ref().take(segment.errors));
216        }
217        if self.options.error_policy == ErrorPolicy::Abort && !errors.is_empty() {
218            return Err(errors.remove(0));
219        }
220        Ok(ParallelWalkReport { entries, errors })
221    }
222}
223
224fn parallel_worker_count(
225    runtime: &ParallelRuntime,
226    parallelism: usize,
227    max_open: usize,
228    tasks: usize,
229) -> usize {
230    requested_workers(runtime, parallelism, max_open)
231        .min(tasks)
232        .max(1)
233}
234
235fn requested_workers(runtime: &ParallelRuntime, parallelism: usize, max_open: usize) -> usize {
236    let available = runtime.parallelism();
237    if parallelism == 0 {
238        available.min(default_traversal_workers())
239    } else {
240        parallelism.min(available)
241    }
242    .min(max_open.max(1))
243    .max(1)
244}
245
246fn schedule_error(root: &std::path::Path, source: std::io::Error) -> WalkError {
247    WalkError::new(root, 0, WalkOperation::ScheduleWorker, source)
248}
249
250const fn default_traversal_workers() -> usize {
251    8
252}
253
254const FRONTIER_TASKS_PER_WORKER: usize = 2;
255
256fn copy_walk_error(error: &WalkError) -> WalkError {
257    WalkError::new(
258        error.path().to_path_buf(),
259        error.depth(),
260        error.operation(),
261        std::io::Error::new(error.io_error().kind(), error.io_error().to_string()),
262    )
263}
264
265pub(super) fn matches_stdout(entry: &WalkEntry, identity: Option<FileIdentity>) -> bool {
266    identity.is_some_and(|identity| {
267        entry.is_file() && crate::stdout::path_matches(entry.path(), identity).unwrap_or(false)
268    })
269}
270
271fn without_stdout(
272    mut report: ParallelWalkReport,
273    identity: Option<FileIdentity>,
274) -> ParallelWalkReport {
275    if identity.is_some() {
276        report
277            .entries
278            .retain(|entry| !matches_stdout(entry, identity));
279    }
280    report
281}
weavatrix_scan/parallel.rs

weavatrix_scan/
parallel.rs
