weavatrix_scan/parallel_multi/
mod.rs

1use crate::parallel::{ParallelWalkReport, ParallelWalker};
2use crate::runtime::ParallelRuntime;
3use crate::walk_types::{WalkError, WalkOptions};
4use std::path::PathBuf;
5
6mod visit;
7
8pub use visit::{ParallelMultiVisitReport, ParallelMultiWalkEvent};
9
10/// Collected raw walk reports for independent roots in insertion order.
11#[derive(Debug)]
12pub struct ParallelMultiWalkReport {
13    pub reports: Vec<ParallelWalkReport>,
14}
15
16impl ParallelMultiWalkReport {
17    #[must_use]
18    pub const fn len(&self) -> usize {
19        self.reports.len()
20    }
21
22    #[must_use]
23    pub const fn is_empty(&self) -> bool {
24        self.reports.is_empty()
25    }
26}
27
28/// Walks independent raw roots concurrently while preserving root order.
29pub struct ParallelMultiWalker {
30    roots: Vec<PathBuf>,
31    options: WalkOptions,
32    root_parallelism: usize,
33    traversal_parallelism: usize,
34    skip_stdout: bool,
35    runtime: ParallelRuntime,
36}
37
38impl ParallelMultiWalker {
39    #[must_use]
40    pub fn new(root: impl Into<PathBuf>) -> Self {
41        Self {
42            roots: vec![root.into()],
43            options: WalkOptions::default(),
44            root_parallelism: 0,
45            traversal_parallelism: 0,
46            skip_stdout: false,
47            runtime: ParallelRuntime::global(),
48        }
49    }
50
51    #[must_use]
52    pub fn add_root(mut self, root: impl Into<PathBuf>) -> Self {
53        self.roots.push(root.into());
54        self
55    }
56
57    #[must_use]
58    pub const fn options(mut self, options: WalkOptions) -> Self {
59        self.options = options;
60        self
61    }
62
63    /// Sets concurrently active roots. Zero uses available parallelism.
64    #[must_use]
65    pub const fn with_root_parallelism(mut self, parallelism: usize) -> Self {
66        self.root_parallelism = parallelism;
67        self
68    }
69
70    /// Sets directory workers requested by each active root.
71    #[must_use]
72    pub const fn with_traversal_parallelism(mut self, parallelism: usize) -> Self {
73        self.traversal_parallelism = parallelism;
74        self
75    }
76
77    /// Selects the executor shared by all active roots.
78    #[must_use]
79    pub fn runtime(mut self, runtime: ParallelRuntime) -> Self {
80        self.runtime = runtime;
81        self
82    }
83
84    /// Skips a regular file that refers to redirected standard output.
85    #[must_use]
86    pub const fn skip_stdout(mut self, enabled: bool) -> Self {
87        self.skip_stdout = enabled;
88        self
89    }
90
91    /// Walks every root and returns reports in insertion order.
92    ///
93    /// # Errors
94    ///
95    /// Returns the first root error in insertion order after all started root
96    /// workers have joined.
97    ///
98    /// # Panics
99    ///
100    /// Panics if an internal root worker panics.
101    pub fn walk(self) -> Result<ParallelMultiWalkReport, WalkError> {
102        let worker_count = if self.runtime.is_worker_thread() {
103            1
104        } else {
105            root_worker_count(self.root_parallelism, self.roots.len())
106        };
107        if worker_count <= 1 {
108            let reports = self
109                .roots
110                .into_iter()
111                .map(|root| {
112                    ParallelWalker::new(root)
113                        .options(self.options)
114                        .with_parallelism(self.traversal_parallelism)
115                        .runtime(self.runtime.clone())
116                        .skip_stdout(self.skip_stdout)
117                        .walk()
118                })
119                .collect::<Result<Vec<_>, _>>()?;
120            return Ok(ParallelMultiWalkReport { reports });
121        }
122
123        let chunk_size = self.roots.len().div_ceil(worker_count);
124        let indexed = self.roots.into_iter().enumerate().collect::<Vec<_>>();
125        let mut walked = std::thread::scope(|scope| {
126            indexed
127                .chunks(chunk_size)
128                .map(|chunk| {
129                    scope.spawn(|| {
130                        chunk
131                            .iter()
132                            .map(|(index, root)| {
133                                (
134                                    *index,
135                                    ParallelWalker::new(root)
136                                        .options(self.options)
137                                        .with_parallelism(self.traversal_parallelism)
138                                        .runtime(self.runtime.clone())
139                                        .skip_stdout(self.skip_stdout)
140                                        .walk(),
141                                )
142                            })
143                            .collect::<Vec<_>>()
144                    })
145                })
146                .collect::<Vec<_>>()
147                .into_iter()
148                .flat_map(|handle| handle.join().expect("multi-root walk worker panicked"))
149                .collect::<Vec<_>>()
150        });
151        walked.sort_unstable_by_key(|(index, _)| *index);
152        let reports = walked
153            .into_iter()
154            .map(|(_, report)| report)
155            .collect::<Result<Vec<_>, _>>()?;
156        Ok(ParallelMultiWalkReport { reports })
157    }
158}
159
160fn root_worker_count(requested: usize, roots: usize) -> usize {
161    if roots == 0 {
162        return 1;
163    }
164    let available = std::thread::available_parallelism().map_or(1, std::num::NonZeroUsize::get);
165    let requested = if requested == 0 {
166        available.min(if cfg!(windows) { 8 } else { 16 })
167    } else {
168        requested.min(available)
169    };
170    requested.min(roots).max(1)
171}
weavatrix_scan/parallel_multi/mod.rs

weavatrix_scan/parallel_multi/
mod.rs
