weavatrix_scan/
multi_scanner.rs

1use crate::config::ScanOptions;
2use crate::content_visit::{
3    ContentVisitControl, ContentVisitEvent, ContentVisitMode, ContentVisitReport,
4    MultiContentVisitReport,
5};
6use crate::error::Result;
7use crate::report::ScanReport;
8use crate::runtime::ParallelRuntime;
9use crate::scanner::{Scanner, scan_repository_with_runtime};
10use std::path::PathBuf;
11use std::sync::Arc;
12
13/// Deterministic collection of independently rooted scan reports.
14#[derive(Debug, Clone, PartialEq, Eq)]
15pub struct MultiScanReport {
16    /// Reports remain in the same order as roots were added.
17    pub reports: Vec<ScanReport>,
18}
19
20impl MultiScanReport {
21    #[must_use]
22    pub const fn len(&self) -> usize {
23        self.reports.len()
24    }
25
26    #[must_use]
27    pub const fn is_empty(&self) -> bool {
28        self.reports.is_empty()
29    }
30}
31
32/// Scans independent roots concurrently while preserving report order.
33pub struct MultiScanner {
34    roots: Vec<PathBuf>,
35    options: ScanOptions,
36    root_parallelism: usize,
37    runtime: ParallelRuntime,
38}
39
40impl MultiScanner {
41    #[must_use]
42    pub fn new(root: impl Into<PathBuf>) -> Self {
43        Self {
44            roots: vec![root.into()],
45            options: ScanOptions::default(),
46            root_parallelism: 0,
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 fn options(mut self, options: ScanOptions) -> Self {
59        self.options = options;
60        self
61    }
62
63    /// Sets concurrent root scans. Zero selects bounded 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    /// Selects the executor shared by parallel discovery in every root.
71    #[must_use]
72    pub fn runtime(mut self, runtime: ParallelRuntime) -> Self {
73        self.runtime = runtime;
74        self
75    }
76
77    /// Scans all roots concurrently and returns them in insertion order.
78    ///
79    /// # Errors
80    ///
81    /// Returns the first error in insertion order after all already-started
82    /// root scans have joined.
83    ///
84    /// # Panics
85    ///
86    /// Panics if an internal root worker panics.
87    pub fn scan(self) -> Result<MultiScanReport> {
88        let worker_count = if self.runtime.is_worker_thread() {
89            1
90        } else {
91            root_worker_count(self.root_parallelism, self.roots.len())
92        };
93        if worker_count <= 1 {
94            let reports = self
95                .roots
96                .iter()
97                .map(|root| scan_repository_with_runtime(root, &self.options, None, &self.runtime))
98                .collect::<Result<Vec<_>>>()?;
99            return Ok(MultiScanReport { reports });
100        }
101
102        let chunk_size = self.roots.len().div_ceil(worker_count);
103        let indexed = self.roots.into_iter().enumerate().collect::<Vec<_>>();
104        let mut scanned = std::thread::scope(|scope| {
105            indexed
106                .chunks(chunk_size)
107                .map(|chunk| {
108                    let options = &self.options;
109                    let runtime = self.runtime.clone();
110                    scope.spawn(move || {
111                        chunk
112                            .iter()
113                            .map(|(index, root)| {
114                                (
115                                    *index,
116                                    scan_repository_with_runtime(root, options, None, &runtime),
117                                )
118                            })
119                            .collect::<Vec<_>>()
120                    })
121                })
122                .collect::<Vec<_>>()
123                .into_iter()
124                .flat_map(|handle| handle.join().expect("multi-root scanner worker panicked"))
125                .collect::<Vec<_>>()
126        });
127        scanned.sort_unstable_by_key(|(index, _)| *index);
128        let reports = scanned
129            .into_iter()
130            .map(|(_, report)| report)
131            .collect::<Result<Vec<_>>>()?;
132        Ok(MultiScanReport { reports })
133    }
134
135    /// Visits selected content across all roots with globally bounded
136    /// parallelism and returns root reports in insertion order.
137    ///
138    /// The factory receives `(root_index, worker_index)` and creates state for
139    /// one content worker. Every event carries the same `root_index`.
140    /// `ContentVisitControl::Quit` cancels work across every root.
141    ///
142    /// # Errors
143    ///
144    /// Returns the first error in root insertion order after all already
145    /// started root visits have joined.
146    ///
147    /// # Panics
148    ///
149    /// Propagates a panic from a factory or visitor after active root visits
150    /// observe cancellation.
151    pub fn visit_content<Factory, Visitor>(
152        self,
153        factory: Factory,
154    ) -> Result<MultiContentVisitReport>
155    where
156        Factory: Fn(usize, usize) -> Visitor + Send + Sync + 'static,
157        Visitor:
158            for<'event> FnMut(ContentVisitEvent<'event>) -> ContentVisitControl + Send + 'static,
159    {
160        self.visit_content_with_mode(ContentVisitMode::Revision, factory)
161    }
162
163    /// Visits content across all roots without retaining selected-file
164    /// manifests or computing per-root revisions.
165    ///
166    /// # Errors
167    ///
168    /// Returns the same errors as [`Self::visit_content`].
169    pub fn visit_content_streaming<Factory, Visitor>(
170        self,
171        factory: Factory,
172    ) -> Result<MultiContentVisitReport>
173    where
174        Factory: Fn(usize, usize) -> Visitor + Send + Sync + 'static,
175        Visitor:
176            for<'event> FnMut(ContentVisitEvent<'event>) -> ContentVisitControl + Send + 'static,
177    {
178        self.visit_content_with_mode(ContentVisitMode::Streaming, factory)
179    }
180
181    fn visit_content_with_mode<Factory, Visitor>(
182        mut self,
183        mode: ContentVisitMode,
184        factory: Factory,
185    ) -> Result<MultiContentVisitReport>
186    where
187        Factory: Fn(usize, usize) -> Visitor + Send + Sync + 'static,
188        Visitor:
189            for<'event> FnMut(ContentVisitEvent<'event>) -> ContentVisitControl + Send + 'static,
190    {
191        let cancellation = self.options.cancellation.clone().unwrap_or_default();
192        self.options.cancellation = Some(cancellation.clone());
193        let worker_count = if self.runtime.is_worker_thread() {
194            1
195        } else {
196            root_worker_count(self.root_parallelism, self.roots.len())
197        };
198        let factory = Arc::new(factory);
199        if worker_count <= 1 {
200            let reports = self
201                .roots
202                .into_iter()
203                .enumerate()
204                .map(|(root_index, root)| {
205                    visit_root_content(
206                        root,
207                        root_index,
208                        &self.options,
209                        &self.runtime,
210                        mode,
211                        Arc::clone(&factory),
212                    )
213                })
214                .collect::<Result<Vec<_>>>()?;
215            return Ok(MultiContentVisitReport { reports });
216        }
217
218        let chunk_size = self.roots.len().div_ceil(worker_count);
219        let indexed = self.roots.into_iter().enumerate().collect::<Vec<_>>();
220        let mut visited = std::thread::scope(|scope| {
221            indexed
222                .chunks(chunk_size)
223                .map(|chunk| {
224                    let options = &self.options;
225                    let runtime = self.runtime.clone();
226                    let factory = Arc::clone(&factory);
227                    let cancellation = cancellation.clone();
228                    scope.spawn(move || {
229                        chunk
230                            .iter()
231                            .map(|(root_index, root)| {
232                                let result = visit_root_content(
233                                    root.clone(),
234                                    *root_index,
235                                    options,
236                                    &runtime,
237                                    mode,
238                                    Arc::clone(&factory),
239                                );
240                                if result.is_err() {
241                                    cancellation.cancel();
242                                }
243                                (*root_index, result)
244                            })
245                            .collect::<Vec<_>>()
246                    })
247                })
248                .collect::<Vec<_>>()
249                .into_iter()
250                .flat_map(|handle| {
251                    handle
252                        .join()
253                        .expect("multi-root content visitor worker panicked")
254                })
255                .collect::<Vec<_>>()
256        });
257        visited.sort_unstable_by_key(|(root_index, _)| *root_index);
258        let reports = visited
259            .into_iter()
260            .map(|(_, report)| report)
261            .collect::<Result<Vec<ContentVisitReport>>>()?;
262        Ok(MultiContentVisitReport { reports })
263    }
264}
265
266fn visit_root_content<Factory, Visitor>(
267    root: PathBuf,
268    root_index: usize,
269    options: &ScanOptions,
270    runtime: &ParallelRuntime,
271    mode: ContentVisitMode,
272    factory: Arc<Factory>,
273) -> Result<ContentVisitReport>
274where
275    Factory: Fn(usize, usize) -> Visitor + Send + Sync + 'static,
276    Visitor: for<'event> FnMut(ContentVisitEvent<'event>) -> ContentVisitControl + Send + 'static,
277{
278    Scanner::new(root)
279        .options(options.clone())
280        .runtime(runtime.clone())
281        .visit_content_with_root_mode(root_index, mode, move |worker_index| {
282            factory(root_index, worker_index)
283        })
284}
285
286fn root_worker_count(requested: usize, roots: usize) -> usize {
287    if roots == 0 {
288        return 1;
289    }
290    let available = std::thread::available_parallelism().map_or(1, std::num::NonZeroUsize::get);
291    let requested = if requested == 0 {
292        available.min(if cfg!(windows) { 8 } else { 16 })
293    } else {
294        requested.min(available)
295    };
296    requested.min(roots).max(1)
297}
weavatrix_scan/multi_scanner.rs

weavatrix_scan/
multi_scanner.rs
