diff --git a/crates/dua-lib/src/lib.rs b/crates/dua-lib/src/lib.rs index 3b173089..0d9aa6b7 100644 --- a/crates/dua-lib/src/lib.rs +++ b/crates/dua-lib/src/lib.rs @@ -152,8 +152,8 @@ pub enum Order { ParentFirst, } -/// Filesystem metadata requested during traversal. -#[derive(Clone, Copy, Debug, Default)] +/// Filesystem metadata and result buffering requested during traversal. +#[derive(Clone, Copy, Debug)] pub struct Options { /// Collect only entry types, leaving [`Entry::metadata`] as `None`. /// @@ -164,6 +164,23 @@ pub struct Options { /// Collect APFS clone identity and data-fork allocation metadata. #[cfg(target_os = "macos")] pub apfs_clone_metadata: bool, + /// Result queue capacity per worker, defaulting to `2`. + /// + /// Workers share a queue with this capacity multiplied by the worker count and block when + /// it is full. Increase this to keep workers busy while the consumer pauses to process entries, + /// at the cost of buffering more batches in memory. `0` disables buffering. + pub queued_batches_per_worker: usize, +} + +impl Default for Options { + fn default() -> Self { + Self { + skip_metadata: false, + #[cfg(target_os = "macos")] + apfs_clone_metadata: false, + queued_batches_per_worker: 2, + } + } } impl Options { @@ -173,6 +190,13 @@ impl Options { self.skip_metadata = true; self } + + /// Set the result queue capacity per worker; see [`Self::queued_batches_per_worker`]. + #[must_use] + pub fn queued_batches_per_worker(mut self, batches: usize) -> Self { + self.queued_batches_per_worker = batches; + self + } } /// Dense identifier of a directory within one filesystem walk. @@ -784,7 +808,11 @@ impl Entry { fn start_pool(threads: usize, order: Order, options: Options, next_directory_id: usize) -> Pool { let workers: Vec<_> = (0..threads).map(|_| Worker::new_lifo()).collect(); let parkers: Vec<_> = (0..threads).map(|_| Parker::new()).collect(); - let (event_tx, event_rx) = sync_channel(threads * 2); + let (event_tx, event_rx) = sync_channel( + threads + .checked_mul(options.queued_batches_per_worker) + .expect("result queue capacity overflow"), + ); let shared = Arc::new(PoolShared { injector: Injector::new(), stealers: workers.iter().map(Worker::stealer).collect(), @@ -1379,6 +1407,7 @@ fn assign_directory_ids(entry: &mut Entry, parent_directory_id: usize, shared: & .then(|| allocate_directory_id(shared)); } +#[allow(deprecated)] // `try_update` requires Rust 1.95; keep the Rust 1.88-compatible alias. fn allocate_directory_id(shared: &PoolShared) -> DirectoryId { DirectoryId::new( shared @@ -1473,6 +1502,27 @@ mod tests { use super::*; use std::collections::HashMap; + #[test] + fn result_queue_capacity_is_configurable_per_worker() { + for threads in [1, 2] { + for (options, batches) in [ + (Options::default(), 2), + (Options::default().queued_batches_per_worker(0), 0), + (Options::default().queued_batches_per_worker(1), 1), + (Options::default().queued_batches_per_worker(128), 128), + ] { + let pool = start_pool(threads, Order::ParentFirst, options, 0); + for _ in 0..threads * batches { + pool.shared.events.try_send(Event::Finished).unwrap(); + } + assert!(matches!( + pool.shared.events.try_send(Event::Finished), + Err(std::sync::mpsc::TrySendError::Full(_)) + )); + } + } + } + #[test] fn roots_added_during_a_blocked_walk_finish_independently() { use std::time::Duration; diff --git a/crates/dua-lib/src/macos/tests.rs b/crates/dua-lib/src/macos/tests.rs index 6c047b59..e4ab8aef 100644 --- a/crates/dua-lib/src/macos/tests.rs +++ b/crates/dua-lib/src/macos/tests.rs @@ -6,6 +6,7 @@ fn options(apfs_clone_metadata: bool) -> Options { Options { apfs_clone_metadata, skip_metadata: false, + ..Options::default() } } diff --git a/crates/dua-lib/src/windows.rs b/crates/dua-lib/src/windows.rs index fbc68648..8d1b7b7d 100644 --- a/crates/dua-lib/src/windows.rs +++ b/crates/dua-lib/src/windows.rs @@ -622,7 +622,10 @@ mod tests { } for skip_metadata in [false, true] { - let options = crate::Options { skip_metadata }; + let options = crate::Options { + skip_metadata, + ..crate::Options::default() + }; let entry = Entry::from_path(&link, options).unwrap(); assert!(entry.file_type.is_symlink()); assert!(!entry.file_type.is_dir()); @@ -690,7 +693,10 @@ mod tests { let entries = crate::ReadDir::open( Arc::from(path.as_path()), 1, - crate::Options { skip_metadata }, + crate::Options { + skip_metadata, + ..crate::Options::default() + }, ) .unwrap() .map(|entry| entry.unwrap().file_name) @@ -718,7 +724,10 @@ mod tests { let entries = crate::ReadDir::open( Arc::from(ordinary_child.as_path()), 1, - crate::Options { skip_metadata }, + crate::Options { + skip_metadata, + ..crate::Options::default() + }, ) .unwrap() .map(|entry| entry.unwrap().file_name) @@ -737,7 +746,10 @@ mod tests { let entries = crate::ReadDir::open( Arc::from(path.as_path()), 1, - crate::Options { skip_metadata }, + crate::Options { + skip_metadata, + ..crate::Options::default() + }, ) .unwrap() .map(|entry| entry.unwrap().file_name) diff --git a/crates/dua-lib/tests/dua.rs b/crates/dua-lib/tests/dua.rs index 36fe075e..84731e3d 100644 --- a/crates/dua-lib/tests/dua.rs +++ b/crates/dua-lib/tests/dua.rs @@ -14,7 +14,7 @@ fn walkers_are_available_to_consumers() { &roots[0], 2, Order::ParentFirst, - dua_core::Options::default(), + dua_core::Options::default().queued_batches_per_worker(128), |_| true, ) .map(|entry| { @@ -32,7 +32,7 @@ fn walkers_are_available_to_consumers() { roots.iter().cloned().enumerate(), 2, Order::Completion, - dua_core::Options::default(), + dua_core::Options::default().queued_batches_per_worker(0), |_, _| true, ) .collect::>(); diff --git a/src/aggregate.rs b/src/aggregate.rs index 7d3c1773..58ce1a77 100644 --- a/src/aggregate.rs +++ b/src/aggregate.rs @@ -920,6 +920,7 @@ mod tests { metadata_options: crate::TraversalOptions { skip_metadata: false, apfs_clone_metadata: true, + ..crate::TraversalOptions::default() }, base_dir: None, }, @@ -943,6 +944,7 @@ mod tests { dua_core::Options { skip_metadata: false, apfs_clone_metadata: true, + ..dua_core::Options::default() }, ) .unwrap() @@ -962,6 +964,7 @@ mod tests { metadata_options: crate::TraversalOptions { skip_metadata: false, apfs_clone_metadata: true, + ..crate::TraversalOptions::default() }, base_dir: None, }, diff --git a/src/clean.rs b/src/clean.rs index 881c80df..130ea03a 100644 --- a/src/clean.rs +++ b/src/clean.rs @@ -429,7 +429,10 @@ mod tests { let root = tmp.path().canonicalize().unwrap(); let direct = directory(&root, "node_modules"); let nested = directory(&root, "project/__pycache__"); - assert!(scan(vec![root.clone()], options(), Some(0)).0.is_empty()); + assert_eq!( + scan(vec![root.clone()], options(), Some(0)).0, + Vec::::new() + ); assert_eq!( scan(vec![root.clone()], options(), Some(1)).0, vec![direct.clone()] @@ -530,7 +533,7 @@ mod tests { directory(&root, &format!("{marker}/objects/node_modules")); let (paths, entries, errors) = scan(vec![root.join(marker).join("objects")], options(), None); - assert!(paths.is_empty()); + assert_eq!(paths, Vec::::new()); assert_eq!(entries, 1); assert_eq!(errors, 0); } @@ -550,10 +553,9 @@ mod tests { let venv = directory(&root, "python/venv"); symlink(root.join("manifest"), venv.join("pyvenv.cfg")).unwrap(); assert_eq!(scan(vec![root.clone()], options(), None).0, vec![real]); - assert!( - scan(vec![root.join("node_modules")], options(), None) - .0 - .is_empty() + assert_eq!( + scan(vec![root.join("node_modules")], options(), None).0, + Vec::::new() ); } @@ -564,7 +566,7 @@ mod tests { fs::write(root.join("ordinary-file"), "").unwrap(); let (paths, entries, errors) = scan(vec![root.clone(), root.join("missing")], options(), None); - assert!(paths.is_empty()); + assert_eq!(paths, Vec::::new()); assert!(entries > 0); assert_eq!(errors, 1); directory(&root, "node_modules"); diff --git a/src/interactive/app/notification.rs b/src/interactive/app/notification.rs index ae310820..ffdef0b9 100644 --- a/src/interactive/app/notification.rs +++ b/src/interactive/app/notification.rs @@ -107,7 +107,7 @@ mod tests { let mut output = Vec::new(); emit_if_unfocused_to(&mut output, true, true, "focused").expect("no-op succeeds"); emit_if_unfocused_to(&mut output, false, false, "disabled").expect("no-op succeeds"); - assert!(output.is_empty()); + assert_eq!(output, b""); emit_if_unfocused_to(&mut output, true, false, "ready").expect("notification written"); assert_eq!( diff --git a/src/interactive/app/tests/unit.rs b/src/interactive/app/tests/unit.rs index ddc58af4..4018d0b3 100644 --- a/src/interactive/app/tests/unit.rs +++ b/src/interactive/app/tests/unit.rs @@ -98,7 +98,7 @@ fn it_can_do_a_case_sensitive_glob_search() { Language::English, ) .unwrap(); - assert!(result_sensitive.is_empty()); + assert_eq!(result_sensitive, Vec::::new()); } #[test] diff --git a/src/interactive/widgets/i18n.rs b/src/interactive/widgets/i18n.rs index 75fa6467..29e2a17a 100644 --- a/src/interactive/widgets/i18n.rs +++ b/src/interactive/widgets/i18n.rs @@ -1243,8 +1243,8 @@ mod tests { assert!(progress.contains("42")); assert!(progress.contains("123 MB")); } - assert!(!language.ui_text().deletion_running.is_empty()); - assert!(!language.ui_text().cancelling_deletion.is_empty()); + assert_ne!(language.ui_text().deletion_running, ""); + assert_ne!(language.ui_text().cancelling_deletion, ""); } } diff --git a/src/main.rs b/src/main.rs index 3f6f0fab..47d0d85a 100644 --- a/src/main.rs +++ b/src/main.rs @@ -732,6 +732,7 @@ fn walk_options_from(traversal: &options::ScanArgs) -> Result skip_metadata: false, #[cfg(target_os = "macos")] apfs_clone_metadata: traversal.deduplicate_apfs_clones && !traversal.apparent_size, + ..dua::TraversalOptions::default() }, base_dir: None, }; diff --git a/src/snapshot/tests.rs b/src/snapshot/tests.rs index 91c4dffc..8c64b77a 100644 --- a/src/snapshot/tests.rs +++ b/src/snapshot/tests.rs @@ -75,7 +75,7 @@ fn decoder_lends_record_names_and_reuses_depth_buffers() { decoder.next_entry(None).unwrap().unwrap(); assert_eq!(decoder.sibling_names[1].as_ptr(), depth_buffer); assert_eq!(decoder.sibling_names[1].capacity(), depth_capacity); - assert!(decoder.sibling_names[1].is_empty()); + assert_eq!(decoder.sibling_names[1], b""); decoder.next_entry(None).unwrap().unwrap(); assert_eq!(decoder.sibling_names[1].as_ptr(), depth_buffer); diff --git a/src/traverse.rs b/src/traverse.rs index 6058a7e6..07c0fb7c 100644 --- a/src/traverse.rs +++ b/src/traverse.rs @@ -2464,6 +2464,7 @@ mod tests { metadata_options: crate::TraversalOptions { skip_metadata: false, apfs_clone_metadata: deduplicate, + ..crate::TraversalOptions::default() }, base_dir: None, }, diff --git a/tests/snapshot.rs b/tests/snapshot.rs index 34d4a084..b31b4946 100644 --- a/tests/snapshot.rs +++ b/tests/snapshot.rs @@ -106,7 +106,7 @@ fn empty_snapshot_round_trips() { let traversal = Traversal::new(); let bytes = encoded(&traversal, &[]); let snapshot = read(Cursor::new(&bytes)).unwrap(); - assert!(snapshot.roots.is_empty()); + assert_eq!(snapshot.roots, Vec::::new()); assert_eq!(snapshot.traversal.tree.len(), 1); assert_eq!( snapshot diff --git a/tests/stack_commands.rs b/tests/stack_commands.rs index 5d11d018..0de408a0 100644 --- a/tests/stack_commands.rs +++ b/tests/stack_commands.rs @@ -39,7 +39,7 @@ fn stacks_and_flamegraph_commands_use_the_folded_output() { .output() .unwrap(); assert!(flamegraph.status.success()); - assert!(flamegraph.stdout.is_empty()); + assert_eq!(flamegraph.stdout, b""); let svg = fs::read_to_string(fixture.path().join("usage.svg")).unwrap(); assert!(svg.contains("