From 4ca76e103123d4f8f7516d8bad85af91cdb9f1d6 Mon Sep 17 00:00:00 2001 From: Douglas Eichelberger Date: Mon, 12 Oct 2026 19:02:58 -0700 Subject: [PATCH 1/3] perf: let walk workers queue more batches ahead of the iterator The result channel held `threads * 2` batches. When the iterator's consumer pauses, each worker that finishes a batch then blocks until the consumer drains the channel, so the walk stops whenever the consumer is busy. A consumer that does real work per entry, such as parsing some of the files it's handed, spends much of its time waiting on workers that sat idle while it worked. Allow 128 batches per worker instead. The channel stays bounded, and `dua aggregate` is no slower. The unit tests that block a worker on a full channel fill two batches to do it, so they keep that bound. --- crates/dua-lib/src/lib.rs | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/crates/dua-lib/src/lib.rs b/crates/dua-lib/src/lib.rs index 3b173089..5e49f980 100644 --- a/crates/dua-lib/src/lib.rs +++ b/crates/dua-lib/src/lib.rs @@ -138,6 +138,14 @@ 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; +/// Completed batches each worker may have waiting for the iterator. Workers block once this +/// many are queued, so a consumer that pauses, say to parse a file, stalls the walk unless +/// there is room to absorb the pause. +#[cfg(not(test))] +const QUEUED_BATCHES_PER_WORKER: usize = 128; +/// Unit tests fill the output with two batches to block a worker on it. +#[cfg(test)] +const QUEUED_BATCHES_PER_WORKER: usize = 2; #[cfg(target_os = "macos")] type StatEntry = Entry; #[cfg(not(any(windows, target_os = "macos")))] @@ -784,7 +792,7 @@ 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 * QUEUED_BATCHES_PER_WORKER); let shared = Arc::new(PoolShared { injector: Injector::new(), stealers: workers.iter().map(Worker::stealer).collect(), From 2989d30d43c035c9c9ff9eb8eff8584b6244cf12 Mon Sep 17 00:00:00 2001 From: Byron Date: Tue, 13 Oct 2026 04:41:39 +0200 Subject: [PATCH 2/3] feat: make walk result buffering configurable per worker Replace the fixed QUEUED_BATCHES_PER_WORKER constant with Options::queued_batches_per_worker and a chainable builder method. Use two queued batches per worker by default in both production and tests, preserving the original bounded buffering behavior. Consumers that pause to process entries can opt into a larger queue to keep traversal workers busy, trading memory for less idle time. A value of zero selects an unbuffered channel. Check the worker-count multiplication so excessive capacities cannot silently overflow. Update existing Options literals to inherit the new default. Verify the actual queue capacity with one and two workers, and exercise larger and unbuffered queues through the public single-root and multi-root walkers. Assisted-by: GPT 6.1 Sol Co-authored-by: GPT 6.1 Sol --- crates/dua-lib/src/lib.rs | 63 +++++++++++++++++++++++++------ crates/dua-lib/src/macos/tests.rs | 1 + crates/dua-lib/src/windows.rs | 20 ++++++++-- crates/dua-lib/tests/dua.rs | 4 +- src/aggregate.rs | 3 ++ src/main.rs | 1 + src/traverse.rs | 1 + 7 files changed, 76 insertions(+), 17 deletions(-) diff --git a/crates/dua-lib/src/lib.rs b/crates/dua-lib/src/lib.rs index 5e49f980..51a3e611 100644 --- a/crates/dua-lib/src/lib.rs +++ b/crates/dua-lib/src/lib.rs @@ -138,14 +138,6 @@ 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; -/// Completed batches each worker may have waiting for the iterator. Workers block once this -/// many are queued, so a consumer that pauses, say to parse a file, stalls the walk unless -/// there is room to absorb the pause. -#[cfg(not(test))] -const QUEUED_BATCHES_PER_WORKER: usize = 128; -/// Unit tests fill the output with two batches to block a worker on it. -#[cfg(test)] -const QUEUED_BATCHES_PER_WORKER: usize = 2; #[cfg(target_os = "macos")] type StatEntry = Entry; #[cfg(not(any(windows, target_os = "macos")))] @@ -160,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`. /// @@ -172,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 { @@ -181,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. @@ -792,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 * QUEUED_BATCHES_PER_WORKER); + 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(), @@ -1481,6 +1501,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/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/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, }, From afbe487bd0d7b158b6b4ef9e80b4e0113f55eee6 Mon Sep 17 00:00:00 2001 From: Byron Date: Tue, 13 Oct 2026 04:52:29 +0200 Subject: [PATCH 3/3] fix: pass strict workspace Clippy on Rust 1.99 Replace empty-container assertions with equality comparisons and empty string checks with inequality comparisons. This satisfies Clippy's assert_is_empty lint while showing actual values when assertions fail. Locally allow the deprecated fetch_update alias in directory identifier allocation. Its replacement, try_update, requires Rust 1.95, while dua-core supports Rust 1.88. Preserve the existing atomic update and overflow handling without raising the minimum supported Rust version. Validation: cargo clippy --workspace --all-targets -- -D warnings -A unknown-lints --no-deps, workspace tests, and formatting all pass. Assisted-by: GPT 6.1 Sol Co-authored-by: GPT 6.1 Sol --- crates/dua-lib/src/lib.rs | 1 + src/clean.rs | 16 +++++++++------- src/interactive/app/notification.rs | 2 +- src/interactive/app/tests/unit.rs | 2 +- src/interactive/widgets/i18n.rs | 4 ++-- src/snapshot/tests.rs | 2 +- tests/snapshot.rs | 2 +- tests/stack_commands.rs | 2 +- 8 files changed, 17 insertions(+), 14 deletions(-) diff --git a/crates/dua-lib/src/lib.rs b/crates/dua-lib/src/lib.rs index 51a3e611..0d9aa6b7 100644 --- a/crates/dua-lib/src/lib.rs +++ b/crates/dua-lib/src/lib.rs @@ -1407,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 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/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/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("