diff --git a/README.md b/README.md index 1ca193ba..8d094d31 100644 --- a/README.md +++ b/README.md @@ -178,8 +178,7 @@ dua aggregate --help On macOS, the `--deduplicate-apfs-clones` traversal option counts fully shared APFS file clones only once in aggregate and interactive runs. It is opt-in -because collecting the additional metadata reduces traversal performance by -about 6%. +because collecting the additional metadata reduces traversal performance. Files that share only some blocks are not deduplicated, and `--apparent-size` still reports each file's logical length. diff --git a/crates/dua-lib/src/lib.rs b/crates/dua-lib/src/lib.rs index 5cd660c5..28b26d67 100644 --- a/crates/dua-lib/src/lib.rs +++ b/crates/dua-lib/src/lib.rs @@ -1,7 +1,7 @@ //! Parallel filesystem traversal backed by a work-stealing worker pool. //! //! [`walk`] yields the root first, then workers read directories and distribute newly discovered -//! subdirectories among themselves. [`Order::ParentFirst`] publishes each directory's entries +//! subdirectories among themselves. [`Order::ParentFirst`] publishes each entry's batch //! before scheduling its children, while [`Order::Completion`] allows descendant batches to arrive //! first when their reads finish sooner. Sibling order is unspecified in both modes. //! @@ -14,8 +14,9 @@ //! //! The root directory starts in a shared injector queue. On platforms where directory-entry //! metadata may require another syscall, directory reads enqueue small metadata batches, and -//! metadata batches enqueue accepted child directories. Windows and macOS workers instead consume -//! native metadata returned by directory enumeration and enqueue child directories immediately. +//! metadata batches enqueue accepted child directories. Windows and macOS walks consume native +//! metadata returned by directory enumeration instead. Multi-threaded macOS walks probe the +//! initial bulk refills and distribute metadata lookups when those reads spend time waiting. //! Every worker can run available jobs from its local LIFO queue or steal from a peer. Each //! successful thief wakes another idle worker, ramping up only while work remains stealable. A //! worker parks when no queue has work and is unparked when new work arrives or the walk stops. The @@ -131,6 +132,13 @@ type Batch = io::Result>>; /// Number of directory entries grouped into each metadata job or result batch. /// Small chunks expose parallel work and stream wide directories while amortizing queue overhead. const ENTRY_CHUNK_SIZE: usize = 4; +/// Keep enough metadata jobs available for thieves without retaining an entire wide directory. +#[cfg(not(windows))] +const MAX_QUEUED_STAT_JOBS: usize = 64; +#[cfg(target_os = "macos")] +type StatEntry = Entry; +#[cfg(not(any(windows, target_os = "macos")))] +type StatEntry = fs::DirEntry; /// Controls when entries are yielded relative to their descendants. #[derive(Clone, Copy)] @@ -216,15 +224,15 @@ enum Job { entry_depth: usize, }, /// Fetch metadata for a chunk of entries from a completed directory read. - #[cfg(not(any(windows, target_os = "macos")))] - StatCompletion { + #[cfg(not(windows))] + Stat { root_idx: usize, path: Arc, /// Dense identifier of the directory containing these entries. directory_id: usize, /// Depth assigned to every entry in this chunk; always at least `1`, i.e. a file in a directory. entry_depth: usize, - entries: Vec, + entries: Vec, }, } @@ -233,8 +241,8 @@ impl Job { fn root_idx(&self) -> usize { match self { Job::ReadDir { root_idx, .. } => *root_idx, - #[cfg(not(any(windows, target_os = "macos")))] - Job::StatCompletion { root_idx, .. } => *root_idx, + #[cfg(not(windows))] + Job::Stat { root_idx, .. } => *root_idx, } } } @@ -773,8 +781,8 @@ fn start_jobs(pool: &Pool, root_jobs: Vec) { debug_assert!( root_jobs.iter().all(|j| match j { Job::ReadDir { entry_depth, .. } => *entry_depth, - #[cfg(not(any(windows, target_os = "macos")))] - Job::StatCompletion { entry_depth, .. } => *entry_depth, + #[cfg(not(windows))] + Job::Stat { entry_depth, .. } => *entry_depth, } == 1), "the first jobs should be root jobs, so active_root counts match" ); @@ -872,20 +880,23 @@ fn run_job(job: Job, worker: &Worker, shared: &PoolShared) { directory_id, entry_depth, } => { + #[cfg(any(windows, target_os = "macos"))] + read_dir_native(root, path, directory_id, entry_depth, worker, shared); + #[cfg(not(any(windows, target_os = "macos")))] if matches!(shared.order, Order::Completion) { - read_dir_completion(root, path, directory_id, entry_depth, worker, shared); + read_dir_parallel(root, path, directory_id, entry_depth, worker, shared); } else { read_dir_parent_first(root, path, directory_id, entry_depth, worker, shared); } } - #[cfg(not(any(windows, target_os = "macos")))] - Job::StatCompletion { + #[cfg(not(windows))] + Job::Stat { root_idx: root, path, directory_id, entry_depth, entries, - } => stat_entries_completion( + } => stat_entries( root, path, directory_id, @@ -897,14 +908,13 @@ fn run_job(job: Job, worker: &Worker, shared: &PoolShared) { } } -/// Read a directory for completion-order traversal. -/// Successful directory entries are split into stealable metadata jobs, while enumeration errors -/// are emitted directly; the directory-read job completes after all chunks are queued. -/// This adds parallelism within wide directories when metadata calls dominate. Both traversal -/// orders already process separate directories concurrently, so typical trees may see no speedup. +/// Read a directory and distribute its metadata lookups among workers. +/// Successful entries are split into stealable metadata jobs; new chunks are processed inline +/// once the local queue is full. Enumeration errors are emitted directly. +/// This adds parallelism within wide directories when metadata calls dominate. /// Type-only walks convert entries inline instead, avoiding metadata-job overhead. #[cfg(not(any(windows, target_os = "macos")))] -fn read_dir_completion( +fn read_dir_parallel( root_idx: usize, path: Arc, directory_id: usize, @@ -935,42 +945,39 @@ fn read_dir_completion( }; let mut chunk = Vec::with_capacity(ENTRY_CHUNK_SIZE); let mut errors = Vec::new(); - let mut has_jobs = false; for entry in dir_entries { + if shared.stop.load(AtomicOrdering::Relaxed) { + finish_pending(root_idx, shared); + return; + } match entry { Ok(entry) => { chunk.push(entry); if chunk.len() == ENTRY_CHUNK_SIZE { - add_pending(root_idx, 1, shared); - worker.push(Job::StatCompletion { + schedule_stat_entries( root_idx, - path: Arc::clone(&path), + &path, directory_id, entry_depth, - entries: std::mem::replace( - &mut chunk, - Vec::with_capacity(ENTRY_CHUNK_SIZE), - ), - }); - has_jobs = true; + std::mem::replace(&mut chunk, Vec::with_capacity(ENTRY_CHUNK_SIZE)), + worker, + shared, + ); } } Err(err) => errors.push(Err(err)), } } if !chunk.is_empty() { - add_pending(root_idx, 1, shared); - worker.push(Job::StatCompletion { + schedule_stat_entries( root_idx, - path, + &path, directory_id, entry_depth, - entries: chunk, - }); - has_jobs = true; - } - if has_jobs { - shared.wake_worker(); + chunk, + worker, + shared, + ); } if !errors.is_empty() && shared @@ -986,13 +993,13 @@ fn read_dir_completion( finish_pending(root_idx, shared); } -/// Read a directory for completion-order traversal. +/// Read a directory in either traversal order. /// -/// Unlike the generic implementation, native readers collect metadata while enumerating, -/// so complete entries are published directly in chunks instead of being split into stealable -/// metadata jobs. This streams wide directories but keeps their metadata work on one worker. +/// Native metadata is published directly in chunks. An initial macOS bulk probe can choose +/// ordinary directory enumeration with bounded parallel metadata jobs. Each parent-first batch +/// is sent before its child jobs become stealable. #[cfg(any(windows, target_os = "macos"))] -fn read_dir_completion( +fn read_dir_native( root_idx: usize, path: Arc, directory_id: usize, @@ -1000,26 +1007,60 @@ fn read_dir_completion( worker: &Worker, shared: &PoolShared, ) { - let dir_entries = match ReadDir::open(path, depth, shared.options) { + let dir_entries = match ReadDir::open(Arc::clone(&path), depth, shared.options) { Ok(entries) => entries, Err(err) => { - if shared - .events - .send(Event::Batch { - root_idx, - batch: Err(err), - }) - .is_err() - { - shared.stop.store(true, AtomicOrdering::Relaxed); - } + publish_directory(root_idx, Err(err), Vec::new(), worker, shared); finish_pending(root_idx, shared); return; } }; + #[cfg(target_os = "macos")] + let dir_entries = { + let mut dir_entries = dir_entries; + let mut prefix = Vec::new(); + if shared.stealers.len() > 1 + && let ReadDir::Metadata(reader) = &mut dir_entries + { + prefix = reader.probe_metadata(); + if reader.metadata_is_io_bound() + && let Ok(reopened) = + ReadDir::open(Arc::clone(&path), depth, shared.options.skip_metadata()) + { + // No entries or child jobs have been published, so restarting cannot duplicate them. + prefix.clear(); + dir_entries = reopened; + } + } + prefix.into_iter().chain(dir_entries) + }; + #[cfg(target_os = "macos")] + let mut deferred = Vec::with_capacity(ENTRY_CHUNK_SIZE); let mut entries = Vec::with_capacity(ENTRY_CHUNK_SIZE); let mut jobs = Vec::new(); for mut entry in dir_entries { + if shared.stop.load(AtomicOrdering::Relaxed) { + finish_pending(root_idx, shared); + return; + } + #[cfg(target_os = "macos")] + if !shared.options.skip_metadata + && entry.as_ref().is_ok_and(|entry| entry.metadata.is_none()) + { + deferred.push(entry.expect("metadata-free entry was checked")); + if deferred.len() == ENTRY_CHUNK_SIZE { + schedule_stat_entries( + root_idx, + &path, + directory_id, + depth, + std::mem::replace(&mut deferred, Vec::with_capacity(ENTRY_CHUNK_SIZE)), + worker, + shared, + ); + } + continue; + } if let Ok(entry) = &mut entry { assign_directory_ids(entry, directory_id, shared); } @@ -1039,65 +1080,85 @@ fn read_dir_completion( } entries.push(entry); if entries.len() == ENTRY_CHUNK_SIZE - && !publish_completion_batch(root_idx, &mut entries, &mut jobs, worker, shared) + && !publish_directory( + root_idx, + Ok(std::mem::replace( + &mut entries, + Vec::with_capacity(ENTRY_CHUNK_SIZE), + )), + std::mem::take(&mut jobs), + worker, + shared, + ) { finish_pending(root_idx, shared); return; } } if !entries.is_empty() { - publish_completion_batch(root_idx, &mut entries, &mut jobs, worker, shared); + publish_directory(root_idx, Ok(entries), jobs, worker, shared); + } + #[cfg(target_os = "macos")] + if !deferred.is_empty() { + schedule_stat_entries( + root_idx, + &path, + directory_id, + depth, + deferred, + worker, + shared, + ); } finish_pending(root_idx, shared); } -#[cfg(any(windows, target_os = "macos"))] -fn publish_completion_batch( +#[cfg(not(windows))] +fn schedule_stat_entries( root_idx: usize, - entries: &mut Vec>, - jobs: &mut Vec, + path: &Arc, + directory_id: usize, + entry_depth: usize, + entries: Vec, worker: &Worker, shared: &PoolShared, -) -> bool { - add_pending(root_idx, jobs.len(), shared); - schedule_jobs(std::mem::take(jobs), worker, shared); - if shared - .events - .send(Event::Batch { - root_idx, - batch: Ok(std::mem::replace( - entries, - Vec::with_capacity(ENTRY_CHUNK_SIZE), - )), - }) - .is_err() - { - shared.stop.store(true, AtomicOrdering::Relaxed); - false +) { + let job = Job::Stat { + root_idx, + path: Arc::clone(path), + directory_id, + entry_depth, + entries, + }; + add_pending(root_idx, 1, shared); + if worker.len() >= MAX_QUEUED_STAT_JOBS { + run_job(job, worker, shared); } else { - true + worker.push(job); + shared.wake_worker(); } } -#[cfg(any(windows, target_os = "macos"))] -fn read_dir_parent_first( +#[cfg(not(windows))] +fn stat_entries( root_idx: usize, path: Arc, directory_id: usize, depth: usize, + entries: Vec, worker: &Worker, shared: &PoolShared, ) { - let dir_entries = match ReadDir::open(path, depth, shared.options) { - Ok(entries) => entries, - Err(err) => { - finish_directory(root_idx, Err(err), Vec::new(), worker, shared); - return; - } - }; + #[cfg(target_os = "macos")] + let _ = (path, depth); let mut jobs = Vec::new(); - let entries = dir_entries + let entries = entries + .into_iter() .map(|entry| { + #[cfg(target_os = "macos")] + let entry = Ok(entry.read_metadata(shared.options)); + #[cfg(not(target_os = "macos"))] + let entry = Entry::from_dir_entry(depth, Arc::clone(&path), entry, shared.options); entry.map(|mut entry| { assign_directory_ids(&mut entry, directory_id, shared); if entry.file_type.is_dir() && (shared.descend)(root_idx, &entry) { @@ -1108,67 +1169,20 @@ fn read_dir_parent_first( .directory_id .expect("directories receive an identifier") .index(), - entry_depth: depth + 1, + entry_depth: entry.depth + 1, }); } entry }) }) .collect(); - finish_directory(root_idx, Ok(entries), jobs, worker, shared); -} - -#[cfg(not(any(windows, target_os = "macos")))] -fn stat_entries_completion( - root_idx: usize, - path: Arc, - directory_id: usize, - depth: usize, - entries: Vec, - worker: &Worker, - shared: &PoolShared, -) { - let mut jobs = Vec::new(); - let entries = entries - .into_iter() - .map(|entry| { - Entry::from_dir_entry(depth, Arc::clone(&path), entry, shared.options).map( - |mut entry| { - assign_directory_ids(&mut entry, directory_id, shared); - if entry.file_type.is_dir() && (shared.descend)(root_idx, &entry) { - jobs.push(Job::ReadDir { - root_idx, - path: Arc::from(entry.path()), - directory_id: entry - .directory_id - .expect("directories receive an identifier") - .index(), - entry_depth: entry.depth + 1, - }); - } - entry - }, - ) - }) - .collect(); - add_pending(root_idx, jobs.len(), shared); - schedule_jobs(jobs, worker, shared); - if shared - .events - .send(Event::Batch { - root_idx, - batch: Ok(entries), - }) - .is_err() - { - shared.stop.store(true, AtomicOrdering::Relaxed); - } + publish_directory(root_idx, Ok(entries), jobs, worker, shared); finish_pending(root_idx, shared); } /// Read a directory for parent-first traversal. -/// Entries are converted inline rather than scheduled as `StatCompletion` jobs, producing the -/// complete parent batch and its child-directory jobs together. This lets `finish_directory` send +/// Entries are converted inline rather than scheduled as `Stat` jobs, producing the +/// complete parent batch and its child-directory jobs together. This lets `publish_directory` send /// the parent batch before making any child job available, preserving parent-before-descendant /// order. Metadata within one directory is serial, although separate directories still run in /// parallel; this often matches completion-order performance unless wide-directory metadata is the @@ -1200,7 +1214,8 @@ fn read_dir_inline( let dir_entries = match fs::read_dir(&path) { Ok(entries) => entries, Err(err) => { - finish_directory(root_idx, Err(err), Vec::new(), worker, shared); + publish_directory(root_idx, Err(err), Vec::new(), worker, shared); + finish_pending(root_idx, shared); return; } }; @@ -1228,7 +1243,8 @@ fn read_dir_inline( }) }) .collect(); - finish_directory(root_idx, Ok(entries), jobs, worker, shared); + publish_directory(root_idx, Ok(entries), jobs, worker, shared); + finish_pending(root_idx, shared); } fn assign_directory_ids(entry: &mut Entry, parent_directory_id: usize, shared: &PoolShared) { @@ -1245,16 +1261,19 @@ fn assign_directory_ids(entry: &mut Entry, parent_directory_id: usize, shared: & }); } -/// Publish a completed directory read and schedule its accepted child-directory jobs. +/// Publish a directory batch and schedule its accepted child-directory jobs. /// `ParentFirst` sends the batch before exposing child jobs; `Completion` exposes child jobs first. -/// Child jobs are counted before either action, and the current job is marked complete afterward. -fn finish_directory( +/// Child jobs are counted before either action; the caller completes the current job after EOF. +/// +/// Returns `true` if the batch was sent and child jobs were scheduled, or `false` if the event +/// receiver disconnected, in which case traversal is marked to stop. +fn publish_directory( root_idx: usize, batch: Batch, jobs: Vec, worker: &Worker, shared: &PoolShared, -) { +) -> bool { add_pending(root_idx, jobs.len(), shared); match shared.order { @@ -1265,7 +1284,7 @@ fn finish_directory( .is_err() { shared.stop.store(true, AtomicOrdering::Relaxed); - return; + return false; } schedule_jobs(jobs, worker, shared); } @@ -1277,12 +1296,12 @@ fn finish_directory( .is_err() { shared.stop.store(true, AtomicOrdering::Relaxed); - return; + return false; } } } - finish_pending(root_idx, shared); + true } fn add_pending(root: usize, count: usize, shared: &PoolShared) { @@ -1742,54 +1761,128 @@ mod tests { ); } + #[cfg(not(windows))] + #[test] + fn metadata_backlog_is_bounded() { + let dir = tempfile::tempdir().unwrap(); + let count = ENTRY_CHUNK_SIZE * (MAX_QUEUED_STAT_JOBS + 1); + for idx in 0..count { + fs::write(dir.path().join(idx.to_string()), b"metadata").unwrap(); + } + + let pool = start_pool( + 1, + HashMap::from([(0, AtomicUsize::new(0))]), + Order::ParentFirst, + Arc::new(|_, _| true), + Options::default(), + 1, + ); + // Leave this queue unconsumed to model workers stalled on metadata lookups. + let worker = Worker::new_lifo(); + let path = Arc::from(dir.path()); + #[cfg(target_os = "macos")] + let entries = read_dir(dir.path(), Options::default().skip_metadata()).unwrap(); + #[cfg(not(target_os = "macos"))] + let entries = fs::read_dir(dir.path()).unwrap(); + let mut entries = entries.map(Result::unwrap); + for _ in 0..=MAX_QUEUED_STAT_JOBS { + schedule_stat_entries( + 0, + &path, + 0, + 1, + entries.by_ref().take(ENTRY_CHUNK_SIZE).collect(), + &worker, + &pool.shared, + ); + } + + assert_eq!(worker.len(), MAX_QUEUED_STAT_JOBS); + let Event::Batch { batch, .. } = pool.events.try_recv().unwrap() else { + panic!("a full queue must process the next metadata batch inline"); + }; + let entries = batch.unwrap(); + assert_eq!(entries.len(), ENTRY_CHUNK_SIZE); + for entry in entries { + let entry = entry.unwrap(); + assert_eq!(entry.metadata.unwrap().unwrap().len(), 8); + assert_eq!(entry.parent_directory_id, Some(DirectoryId::new(0))); + } + } + #[cfg(any(windows, target_os = "macos"))] #[test] - fn native_completion_streams_metadata_before_enumeration_finishes() { + fn native_walks_stream_entries_and_child_jobs() { let dir = tempfile::tempdir().unwrap(); for idx in 0..=ENTRY_CHUNK_SIZE { - fs::create_dir(dir.path().join(idx.to_string())).unwrap(); + fs::create_dir_all(dir.path().join(format!("{idx}/child"))).unwrap(); } - let (continue_tx, continue_rx) = std::sync::mpsc::sync_channel(0); - let continue_rx = Arc::new(std::sync::Mutex::new(continue_rx)); - let seen = Arc::new(AtomicUsize::new(0)); - let seen_in_worker = Arc::clone(&seen); - let mut entries = - walk( - dir.path(), - 2, - Order::Completion, - Options::default(), - move |entry| { + for order in [Order::Completion, Order::ParentFirst] { + for options in [Options::default(), Options::default().skip_metadata()] { + let (continue_tx, continue_rx) = std::sync::mpsc::channel(); + let continue_rx = std::sync::Mutex::new(continue_rx); + let seen = AtomicUsize::new(0); + let mut entries = walk(dir.path(), 2, order, options, move |entry| { if entry.depth == 1 - && seen_in_worker.fetch_add(1, AtomicOrdering::Relaxed) == ENTRY_CHUNK_SIZE + && seen.fetch_add(1, AtomicOrdering::Relaxed) == ENTRY_CHUNK_SIZE { - continue_rx - .lock() + continue_rx.lock().unwrap().recv().ok(); + } + true + }); + let root = entries.next().unwrap(); + let mut batches = Vec::new(); + let mut child_started = false; + while let Ok(Event::Batch { batch, .. }) = entries + .pool + .as_ref() .unwrap() + .events .recv_timeout(std::time::Duration::from_secs(2)) - .expect("the first metadata batch should arrive before enumeration finishes"); + { + child_started = batch.as_ref().is_ok_and(|entries| { + entries + .iter() + .any(|entry| entry.as_ref().is_ok_and(|entry| entry.depth == 2)) + }); + batches.push(batch); + if child_started { + break; } - true - }, - ); + } + // Release the producer and drain its events before asserting, including on failure. + continue_tx.send(()).unwrap(); + let remaining = entries.collect::>(); + assert!( + child_started, + "entries and child jobs must be published before all entries are processed \ + (parent first: {}, options: {options:?})", + matches!(order, Order::ParentFirst) + ); - assert_eq!( - entries.next().unwrap().unwrap().depth, - 0, - "the root entry should be yielded first" - ); - assert_eq!( - entries.next().unwrap().unwrap().depth, - 1, - "the first metadata batch should be yielded before enumeration resumes" - ); - continue_tx.send(()).unwrap(); - entries.for_each(drop); - assert_eq!( - seen.load(AtomicOrdering::Relaxed), - ENTRY_CHUNK_SIZE + 1, - "all directory entries should be inspected" - ); + let entries = std::iter::once(root) + .chain(batches.into_iter().flat_map(Result::unwrap)) + .chain(remaining) + .collect::, _>>() + .unwrap(); + assert_eq!(entries.len(), 1 + 2 * (ENTRY_CHUNK_SIZE + 1)); + assert_eq!(entries[0].depth, 0, "the root must be yielded first"); + if matches!(order, Order::ParentFirst) { + let positions = entries + .iter() + .enumerate() + .map(|(idx, entry)| (entry.directory_id.unwrap(), idx)) + .collect::>(); + for (idx, entry) in entries.iter().enumerate().skip(1) { + assert!( + positions[&entry.parent_directory_id.unwrap()] < idx, + "a parent entry must precede every descendant" + ); + } + } + } + } } } diff --git a/crates/dua-lib/src/macos/mod.rs b/crates/dua-lib/src/macos/mod.rs index 05ad2018..4f5be7e3 100644 --- a/crates/dua-lib/src/macos/mod.rs +++ b/crates/dua-lib/src/macos/mod.rs @@ -13,7 +13,7 @@ use std::{ }, path::{Path, PathBuf}, sync::Arc, - time::SystemTime, + time::{Duration, Instant, SystemTime}, }; mod attributes; @@ -69,6 +69,22 @@ impl Entry { }) } + pub(crate) fn read_metadata(mut self, options: crate::Options) -> Self { + let path = self.path(); + self.metadata = Some(fs::symlink_metadata(&path).map(|metadata| { + self.file_type = FileType::from_std(metadata.file_type()); + let data_fork = + if options.apfs_clone_metadata && metadata.is_file() && metadata.blocks() != 0 { + clone_attributes_at(&path, &metadata) + } else { + None + }; + // Both stat and Apple FTS round the filesystem's total allocation to 512-byte blocks. + Metadata::from_std(&metadata, data_fork) + })); + self + } + /// Return the full path to this entry. #[must_use] pub fn path(&self) -> PathBuf { @@ -226,6 +242,8 @@ pub(crate) struct ReadDir { offset: usize, remaining: usize, exhausted: bool, + /// Consecutive metadata refills spending more time waiting than executing. + waiting_reads: u8, /// Whether bulk reads request APFS extended attributes; disabled when they cannot be served. /// Note that this makes the call more expensive. extended_attributes: bool, @@ -235,6 +253,75 @@ pub(crate) struct ReadDir { } impl ReadDir { + fn record_metadata_time(&mut self, elapsed: Duration, cpu: Option) { + self.waiting_reads = if cpu.is_some_and(|cpu| !cpu.is_zero() && cpu < elapsed / 2) { + self.waiting_reads.saturating_add(1) + } else { + 0 + }; + } + + /// Buffer at most two initial bulk refills before the caller chooses its metadata strategy. + /// Call before iteration; unparsed records stay in this reader for streaming. + pub(crate) fn probe_metadata(&mut self) -> Vec> { + self.probe_metadata_with_cpu_time(thread_cpu_time) + } + + /// Probe at most two initial bulk refills using an injectable cumulative thread CPU clock. + /// Call before iteration. Returns consumed entries (excluding `.` and `..`) and any encountered + /// errors; unparsed records remain in the reader for streaming. + /// + /// Each refill compares CPU time with wall time: CPU time below half the elapsed time means + /// waiting exceeded execution. This ratio avoids an absolute latency threshold that would + /// classify a slow CPU as slow I/O. Two consecutive waiting samples enable parallel metadata; + /// a non-waiting sample or an unavailable, zero, or backwards CPU measurement stops the probe. + /// The classification remains sensitive to scheduling, since time off-CPU also counts as waiting. + fn probe_metadata_with_cpu_time( + &mut self, + mut cpu_time: impl FnMut() -> Option, + ) -> Vec> { + let mut entries = Vec::new(); + for _ in 0..2 { + if self.exhausted || self.fallback.is_some() { + break; + } + let cpu_start = cpu_time(); + let start = Instant::now(); + match self.refill() { + Ok(true) => {} + Ok(false) => break, + Err(error) => { + entries.push(Err(error)); + break; + } + } + // A filesystem fallback or listing without metadata is not a bulk metadata sample. + if self.remaining == 0 || self.listing_error.is_some() { + break; + } + let elapsed = start.elapsed(); + let cpu = cpu_start.and_then(|start| cpu_time()?.checked_sub(start)); + self.record_metadata_time(elapsed, cpu); + // A CPU-bound refill rules out switching. Leave its path stats to the streaming iterator + // so child jobs can start before the whole buffer has been processed. + if self.waiting_reads == 0 { + break; + } + while self.remaining != 0 && !self.exhausted { + match self.next_record() { + Ok(entry) if entry.file_name == "." || entry.file_name == ".." => {} + entry => entries.push(entry), + } + } + } + entries + } + + pub(crate) fn metadata_is_io_bound(&self) -> bool { + // Waiting must exceed execution on two initial refills, independent of clock speed. + self.waiting_reads >= 2 + } + pub(crate) fn open(path: Arc, depth: usize, options: crate::Options) -> io::Result { let directory: OwnedFd = fs::OpenOptions::new() .read(true) @@ -248,6 +335,7 @@ impl ReadDir { offset: 0, remaining: 0, exhausted: false, + waiting_reads: 0, extended_attributes: options.apfs_clone_metadata, listing_error: None, parent_path: path, @@ -424,6 +512,26 @@ impl ReadDir { } } +/// User and kernel CPU time for the calling worker; clock failures leave bulk reads enabled. +/// +/// Differences between these readings count only time the worker spends executing on a CPU. +/// Differences between [`std::time::Instant::now()`] readings also include time blocked on I/O +/// or waiting to be scheduled. Comparing the two lets the probe distinguish execution from waiting. +fn thread_cpu_time() -> Option { + let mut time = libc::timespec { + tv_sec: 0, + tv_nsec: 0, + }; + // SAFETY: `time` is writable and CLOCK_THREAD_CPUTIME_ID accepts the calling thread. + if unsafe { libc::clock_gettime(libc::CLOCK_THREAD_CPUTIME_ID, &raw mut time) } != 0 { + return None; + } + Some(Duration::new( + time.tv_sec.try_into().ok()?, + time.tv_nsec.try_into().ok()?, + )) +} + impl Iterator for ReadDir { type Item = io::Result; diff --git a/crates/dua-lib/src/macos/tests.rs b/crates/dua-lib/src/macos/tests.rs index ff434c38..6c047b59 100644 --- a/crates/dua-lib/src/macos/tests.rs +++ b/crates/dua-lib/src/macos/tests.rs @@ -9,6 +9,71 @@ fn options(apfs_clone_metadata: bool) -> Options { } } +#[test] +fn parallel_metadata_requires_repeated_waiting_regardless_of_clock_speed() { + let directory = tempfile::tempdir().unwrap(); + for unit in [Duration::from_micros(1), Duration::from_millis(10)] { + let mut reader = ReadDir::open(Arc::from(directory.path()), 1, options(false)).unwrap(); + for (elapsed, cpu, message) in [ + (unit * 2, Some(unit * 2), "Busy, however slow the CPU."), + (unit * 3, Some(unit), "One waiting refill is insufficient."), + ( + unit * 2, + Some(unit), + "Waiting must exceed CPU time; reset the streak.", + ), + (unit * 3, Some(unit), "One waiting refill is insufficient."), + (unit * 3, None, "A missing CPU measurement also resets it."), + (unit * 3, Some(unit), "One waiting refill is insufficient."), + ( + unit * 3, + Some(Duration::ZERO), + "Ignore measurements below clock resolution.", + ), + (unit * 3, Some(unit), "One waiting refill is insufficient."), + ] { + reader.record_metadata_time(elapsed, cpu); + assert!(!reader.metadata_is_io_bound(), "{message}"); + } + reader.record_metadata_time(unit * 3, Some(unit)); + assert!( + reader.metadata_is_io_bound(), + "repeated waiting should expose parallel work" + ); + } +} + +#[test] +fn cpu_bound_probe_leaves_directory_stats_for_streaming() { + let directory = tempfile::tempdir().unwrap(); + let child = directory.path().join("child"); + fs::create_dir(&child).unwrap(); + let mut reader = ReadDir::open(Arc::from(directory.path()), 1, options(false)).unwrap(); + // Force a CPU-bound sample without depending on filesystem latency or scheduling. + let mut cpu_times = [Some(Duration::ZERO), Some(Duration::MAX)].into_iter(); + let prefix = reader.probe_metadata_with_cpu_time(|| { + cpu_times.next().expect("only one refill should be probed") + }); + assert!( + prefix.is_empty(), + "CPU-bound probing must not parse entries" + ); + + fs::remove_dir(child).unwrap(); + let entry = reader.next().unwrap().unwrap(); + assert_eq!(entry.file_name, "child"); + let error = entry + .metadata + .unwrap() + .err() + .expect("the directory stat must happen after the probe"); + assert_eq!(error.kind(), io::ErrorKind::NotFound); + assert!( + reader.next().is_none(), + "this would be Some if the entry was cached during the probe" + ); +} + #[test] fn metadata_and_entries_keep_clone_tracking_compact() { assert_eq!( @@ -47,6 +112,10 @@ fn fallback_reader_enumerates_entries_with_metadata() { let mut reader = ReadDir::open(Arc::from(directory.path()), 1, options(false)).unwrap(); reader.fallback = Some(fs::read_dir(fallback_directory.path()).unwrap()); + assert!( + reader.probe_metadata().is_empty(), + "a bulk probe must not buffer the fallback iterator" + ); let entry = reader.next().unwrap().unwrap(); assert_eq!( @@ -268,6 +337,9 @@ fn bulk_metadata_identifies_clones_and_hard_links() { "modifying one byte must diverge the partially cloned data fork" ); fs::hard_link(&original, directory.path().join("hard-link")).unwrap(); + fs::write(clone.join("..namedfork/rsrc"), [3; 8192]).unwrap(); + std::os::unix::fs::symlink("original", directory.path().join("link")).unwrap(); + std::os::unix::fs::symlink("missing", directory.path().join("broken")).unwrap(); let ordinary = ReadDir::open(Arc::from(directory.path()), 1, options(false)) .unwrap() @@ -296,6 +368,26 @@ fn bulk_metadata_identifies_clones_and_hard_links() { (entry.file_name, entry.metadata.unwrap().unwrap()) }) .collect::>(); + let reader = crate::read_dir(directory.path(), options(true).skip_metadata()).unwrap(); + for entry in reader { + let entry = entry.unwrap(); + assert!(entry.metadata.is_none()); + let expected = &entries[&entry.file_name]; + let metadata = entry + .read_metadata(options(true)) + .metadata + .unwrap() + .unwrap(); + assert_eq!(metadata.ino(), expected.ino()); + assert_eq!(metadata.nlink(), expected.nlink()); + assert_eq!(metadata.len(), expected.len()); + assert_eq!(metadata.allocated_size(), expected.allocated_size()); + assert_eq!( + metadata.data_allocated_size(), + expected.data_allocated_size() + ); + assert_eq!(metadata.clone_id(), expected.clone_id()); + } let original = &entries[std::ffi::OsStr::new("original")]; let cloned = &entries[std::ffi::OsStr::new("clone")]; let partially_cloned = &entries[std::ffi::OsStr::new("partial-clone")]; @@ -367,12 +459,23 @@ fn bulk_directory_reader_refills_its_buffer() { .unwrap(); } - let count = ReadDir::open(Arc::from(directory.path()), 1, options(false)) - .unwrap() - .map(Result::unwrap) - .count(); - assert_eq!( - count, 700, - "all entries must survive multiple bulk-buffer refills" - ); + for include_apfs in [false, true] { + let mut reader = + ReadDir::open(Arc::from(directory.path()), 1, options(include_apfs)).unwrap(); + let prefix = reader.probe_metadata(); + assert!( + prefix.len() < 700, + "the initial probe must buffer only a bounded prefix" + ); + let mut names = std::collections::HashSet::new(); + for entry in prefix.into_iter().chain(reader) { + let entry = entry.unwrap(); + assert!(entry.metadata.is_some(), "retain metadata from the probe"); + assert!( + names.insert(entry.file_name), + "never repeat a buffered entry" + ); + } + assert_eq!(names.len(), 700, "buffering must not skip entries"); + } } diff --git a/src/options.rs b/src/options.rs index d13bf398..2090bc88 100644 --- a/src/options.rs +++ b/src/options.rs @@ -154,7 +154,7 @@ pub struct ScanArgs { )] pub count_hard_links: bool, - /// Count fully shared APFS file clones only once. This costs about 6% performance. + /// Count fully shared APFS file clones only once. Collecting clone metadata makes scans slower. #[cfg(target_os = "macos")] #[clap(long, help_heading = "Traversal Options")] pub deduplicate_apfs_clones: bool,