weavatrix_scan/parallel/
mod.rs

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

weavatrix_scan/parallel/
mod.rs
