weavatrix_scan/parallel/
visit.rs

1use super::ParallelWalker;
2use super::dynamic;
3use super::visit_worker::visit_serial;
4use crate::control::CancellationToken;
5use crate::walker::{WalkEntry, WalkError};
6
7/// A streaming event emitted from a parallel traversal worker.
8#[derive(Debug)]
9pub enum WalkEvent<'a> {
10    Entry(&'a WalkEntry),
11    Error(&'a WalkError),
12}
13
14/// Controls traversal after a streaming visitor handles an event.
15#[derive(Debug, Clone, Copy, PartialEq, Eq)]
16pub enum WalkControl {
17    Continue,
18    /// Prevents descent when the current event is a directory entry.
19    Skip,
20    /// Cooperatively stops every traversal worker.
21    Quit,
22}
23
24/// Summary of a streaming parallel traversal.
25#[derive(Debug)]
26pub struct ParallelVisitReport {
27    pub visited: u64,
28    pub errors: Vec<WalkError>,
29    pub quit: bool,
30    pub cancelled: bool,
31}
32
33impl ParallelWalker {
34    /// Visits entries directly on traversal workers without collecting paths.
35    ///
36    /// Visitor calls may run concurrently and their order is intentionally
37    /// unspecified. Use [`Self::walk`] when deterministic collected order is
38    /// required.
39    ///
40    /// # Errors
41    ///
42    /// Returns a root error or the first traversal error under
43    /// `ErrorPolicy::Abort`.
44    ///
45    /// # Panics
46    ///
47    /// Panics if the visitor or an internal traversal worker panics.
48    pub fn visit<F>(self, visitor: F) -> Result<ParallelVisitReport, WalkError>
49    where
50        F: for<'entry> Fn(WalkEvent<'entry>) -> WalkControl + Send + Sync + 'static,
51    {
52        self.visit_with_cancellation(&CancellationToken::new(), visitor)
53    }
54
55    /// Streaming parallel traversal with cooperative cancellation.
56    ///
57    /// # Errors
58    ///
59    /// Returns a root error or the first traversal error under
60    /// `ErrorPolicy::Abort`.
61    ///
62    /// # Panics
63    ///
64    /// Panics if the visitor or an internal traversal worker panics.
65    pub fn visit_with_cancellation<F>(
66        mut self,
67        cancellation: &CancellationToken,
68        visitor: F,
69    ) -> Result<ParallelVisitReport, WalkError>
70    where
71        F: for<'entry> Fn(WalkEvent<'entry>) -> WalkControl + Send + Sync + 'static,
72    {
73        self.options = self.options.normalized();
74        if self.runtime.is_worker_thread() {
75            visit_serial(
76                &self.root,
77                self.options,
78                self.skip_stdout,
79                cancellation,
80                visitor,
81            )
82        } else {
83            dynamic::visit(
84                &self.root,
85                self.options,
86                self.parallelism,
87                &self.runtime,
88                self.skip_stdout,
89                cancellation,
90                visitor,
91            )
92        }
93    }
94}
weavatrix_scan/parallel/visit.rs

weavatrix_scan/parallel/
visit.rs
