weavatrix_scan/parallel/ordered_pull/
mod.rs

1use super::ParallelWalker;
2use super::pull::{ParallelWalkIter, PullBatch};
3use crate::control::CancellationToken;
4use crate::runtime::ParallelRuntime;
5use crate::walk_types::{DirectoryIdentity, FileSystemId};
6use crate::walker::{
7    ErrorPolicy, WalkEntry, WalkError, WalkOperation, WalkOptions, WalkSkipReason, Walker,
8};
9use std::any::Any;
10use std::collections::{HashMap, HashSet, VecDeque};
11use std::io;
12use std::path::PathBuf;
13use std::sync::Arc;
14use std::sync::mpsc::{self, SyncSender, sync_channel};
15
16mod execution;
17mod scheduler;
18
19use execution::{ordered_parallel, ordered_serial};
20
21struct DirectoryTask {
22    id: u64,
23    path: PathBuf,
24    depth: usize,
25    identity: Option<DirectoryIdentity>,
26    ancestors: Arc<HashSet<DirectoryIdentity>>,
27}
28
29struct WorkerResult {
30    id: u64,
31    outcome: Result<DirectoryBatch, Box<dyn Any + Send>>,
32}
33
34struct DirectoryBatch {
35    entries: Vec<Result<WalkEntry, WalkError>>,
36    ancestors: Arc<HashSet<DirectoryIdentity>>,
37}
38
39struct PreparedItem {
40    item: Result<WalkEntry, WalkError>,
41    child: Option<u64>,
42}
43
44struct DirectoryFrame {
45    items: std::vec::IntoIter<PreparedItem>,
46}
47
48struct OrderedScheduler {
49    root: Arc<PathBuf>,
50    root_file_system: Option<FileSystemId>,
51    options: WalkOptions,
52    cancellation: CancellationToken,
53    runtime: ParallelRuntime,
54    limit: usize,
55    next_id: u64,
56    queued: VecDeque<DirectoryTask>,
57    outstanding: usize,
58    ready: HashMap<u64, DirectoryBatch>,
59    result_sender: mpsc::Sender<WorkerResult>,
60    result_receiver: mpsc::Receiver<WorkerResult>,
61    schedule_error: Option<WalkError>,
62}
63
64impl ParallelWalker {
65    /// Starts bounded parallel traversal and yields entries in strict,
66    /// deterministic depth-first order.
67    ///
68    /// Directory reads are prefetched up to `max_open` and the configured
69    /// parallelism. A capacity of zero is normalized to one.
70    ///
71    /// # Panics
72    ///
73    /// Panics if the coordinator thread cannot be created. Use
74    /// [`Self::try_into_iter_ordered_bounded`] for fallible startup.
75    #[must_use]
76    pub fn into_iter_ordered_bounded(self, capacity: usize) -> ParallelWalkIter {
77        self.try_into_iter_ordered_bounded(capacity)
78            .expect("ordered parallel pull coordinator thread can be created")
79    }
80
81    /// Fallible form of [`Self::into_iter_ordered_bounded`].
82    ///
83    /// # Errors
84    ///
85    /// Returns the coordinator thread spawn error.
86    pub fn try_into_iter_ordered_bounded(self, capacity: usize) -> io::Result<ParallelWalkIter> {
87        let capacity = capacity.max(1);
88        let (sender, receiver) = sync_channel(capacity.saturating_sub(1));
89        let cancellation = CancellationToken::new();
90        let coordinator_cancellation = cancellation.clone();
91        let use_serial = self.runtime.is_worker_thread();
92        let coordinator = std::thread::Builder::new()
93            .name("weavatrix-scan-ordered-pull".to_owned())
94            .spawn(move || {
95                if use_serial {
96                    ordered_serial(&self, &coordinator_cancellation, &sender);
97                } else {
98                    ordered_parallel(&self, &coordinator_cancellation, &sender);
99                }
100            })?;
101        Ok(ParallelWalkIter::from_coordinator(
102            receiver,
103            cancellation,
104            coordinator,
105        ))
106    }
107}
108
109fn read_directory(
110    root: &Arc<PathBuf>,
111    root_file_system: Option<FileSystemId>,
112    options: WalkOptions,
113    cancellation: &CancellationToken,
114    task: DirectoryTask,
115) -> DirectoryBatch {
116    let ancestors = Arc::clone(&task.ancestors);
117    let mut worker_options = options;
118    worker_options.error_policy = ErrorPolicy::Continue;
119    worker_options.min_depth = 0;
120    worker_options.max_open = 1;
121    worker_options.max_depth = Some(
122        options
123            .max_depth
124            .unwrap_or(task.depth.saturating_add(1))
125            .min(task.depth.saturating_add(1)),
126    );
127    let mut walker = Walker::from_known_directory_with_ancestry(
128        root,
129        task.path,
130        task.depth,
131        worker_options,
132        root_file_system,
133        task.identity,
134        task.ancestors.as_ref().clone(),
135    );
136    let mut entries = Vec::new();
137    while !cancellation.is_cancelled() {
138        let Some(item) = walker.next() else {
139            break;
140        };
141        match item {
142            Ok(mut entry) => {
143                if entry.is_dir()
144                    && entry.skip_reason() == Some(WalkSkipReason::MaxDepth)
145                    && options
146                        .max_depth
147                        .is_none_or(|maximum| entry.depth() < maximum)
148                {
149                    entry.clear_depth_skip();
150                }
151                if entry.is_dir() {
152                    walker.skip_current_dir();
153                }
154                entries.push(Ok(entry));
155            }
156            Err(error) => entries.push(Err(error)),
157        }
158    }
159    DirectoryBatch { entries, ancestors }
160}
weavatrix_scan/parallel/ordered_pull/mod.rs

weavatrix_scan/parallel/ordered_pull/
mod.rs
