perf: let walk workers queue more batches ahead of the iterator

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;

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()
     }
 }
 

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)

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::<Vec<_>>();

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,
             },

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::<PathBuf>::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::<PathBuf>::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::<PathBuf>::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::<PathBuf>::new());
         assert!(entries > 0);
         assert_eq!(errors, 1);
         directory(&root, "node_modules");

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!(

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::<TreeIndex>::new());
 }
 
 #[test]

src/interactive/widgets/i18n.rs

@@ -159,7 +159,7 @@
     pub mark_trash_2: &'static str,
 
     pub app_title: &'static str,
     pub app_suspend: &'static str,
     pub app_repaint: &'static str,
     pub app_quit: &'static str,
 }
@@ -1243,8 +1243,8 @@
                 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, "");
         }
     }
 

src/main.rs

@@ -732,6 +732,7 @@ fn walk_options_from(traversal: &options::ScanArgs) -> Result<dua::WalkOptions>
             skip_metadata: false,
             #[cfg(target_os = "macos")]
             apfs_clone_metadata: traversal.deduplicate_apfs_clones && !traversal.apparent_size,
+            ..dua::TraversalOptions::default()
         },
         base_dir: None,
     };

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);

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,
                 },

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::<TreeIndex>::new());
     assert_eq!(snapshot.traversal.tree.len(), 1);
     assert_eq!(
         snapshot

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("<svg"));
     assert!(svg.contains("width=\"640\""));