weavatrix_scan/
parallel_multi.rs

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

weavatrix_scan/
parallel_multi.rs
