Repository: sharkdp/fd (language: rust)

Issue:
Improve startup time
`fd`s startup time is quite slow. On my 12 core system, it takes ~ 20 ms for "searching" an empty folder. This is fast enough not to be noticeable by humans, but it looks bad in benchmarks when comparing `fd` with other tools on small folders [^1]. And it's also an actual problem for use cases where `fd` is called repeatedly from a script.

Some of that overhead is caused by the spawning of threads, and that problem is already tracked in https://github.com/sharkdp/fd/issues/1203. But I think there is more that can be done. Instead of using my usual go-to performance tool (`perf`), let's look at the [magic-trace](https://github.com/janestreet/magic-trace) output of a `fd` call in an empty folder [^2]. If someone is interested, I've attached the [trace](https://github.com/sharkdp/fd/files/13079377/trace.zip) to this post. Go to https://magic-trace.org/ to load it in their viewer.

The full trace looks like this:

![image](https://github.com/sharkdp/fd/assets/4209276/14619eb4-2e1a-4f5a-9f3a-122aa6438173)

The first 2.2 ms are typical process startup things (before `main`). I don't think there is any room for optimization here (?)

![image](https://github.com/sharkdp/fd/assets/4209276/c13a7b28-d22f-40a1-a8e5-104dd6fb16f5)

The next ~2 ms are more interesting:

![image](https://github.com/sharkdp/fd/assets/4209276/51ede8bd-883e-4031-ad6d-0f2934b66c14)

some notable steps (even if insignificant in time)
- parsing command-line arguments (**724 µs**)
- a "is_existing_directory" check on the search path (28 µs)
- the parsing of the (empty) search pattern regex (32 µs)
- `isatty` check (5 ·s)
- `LsColors::from_env` (**579 µs**)
- num_cpus::linux::get_num_cpus` (**441 µs**)
- RegexBuilder::build (171 µs)

Some things were surprising to me. I didn't expect the `get_num_cpus` call to take this long. There might be some room for improvement here by doing things in parallel (e.g. `LsColors::from_env`)? But only if the thread overhead is not too high.

Then we start the actual scan, which takes the majority of the time:

![image](https://github.com/sharkdp/fd/assets/4209276/5dc31bf4-7771-448e-9bee-b6895ced1c5e)

Here, I'm not so sure how to interpret the trace, as things are actually happening on multiple threads. But we can (presumably) see some of the thread spawning/joining time here (~ **5 ms**):

![image](https://github.com/sharkdp/fd/assets/4209276/578a922e-3656-4966-90b9-c37eec18b64d)

and some gitignore matcher logic going on here (370 µs total):

![image](https://github.com/sharkdp/fd/assets/4209276/400de973-fb8f-4e63-9411-4140e9a4d7c1)

Most of the time is actually unaccounted for in the trace, because I can only see:

![image](https://github.com/sharkdp/fd/assets/4209276/253f364a-0a8d-46d8-9fb9-e82af7ece87d)

We can see a bit more when switching off LTO:

![image](https://github.com/sharkdp/fd/assets/4209276/6741b564-99e3-4565-b2d3-1d49d062663b)

Apparently, **11 ms** are spent in `crossbeam_channel::channel::bounded`'s `from_iter` method? (probably the receive call?) — even though we don't have any work to do. On a `-j1` run, this part only takes 1 ms.

[^1]: Those "small" folders can be pretty large, actually. It takes hundreds of thousands of files before we can make up for the startup "penalty".
[^2]: I recently discovered this and used it successfully to benchmark (and then optimize) the *startup* time of other programs. 
[BUG] performance regression in trivial searches between 8.4 and 8.5 persisting in 8.7
### Checks

- [X] I have read the troubleshooting section and still think this is a bug.

### Describe the bug you encountered:

It takes nearly 10x longer to search an empty directory with fd 8.7 or 8.5 versus 8.4
```
> mkdir empty
> hyperfine -N 'fd-v8.4.0-x86_64-unknown-linux-gnu/fd . empty/' 'fd-v8.5.0-x86_64-unknown-linux-gnu/fd . empty/' 'fd-v8.7.0-x86_64-unknown-linux-gnu/fd . empty/'
Benchmark 1: fd-v8.4.0-x86_64-unknown-linux-gnu/fd . empty/
  Time (mean ± σ):       4.0 ms ±   0.4 ms    [User: 2.2 ms, System: 2.6 ms]
  Range (min … max):     3.3 ms …   5.9 ms    594 runs

Benchmark 2: fd-v8.5.0-x86_64-unknown-linux-gnu/fd . empty/
  Time (mean ± σ):      39.2 ms ±   2.3 ms    [User: 3.9 ms, System: 36.2 ms]
  Range (min … max):    36.8 ms …  56.7 ms    79 runs

  Warning: Statistical outliers were detected. Consider re-running this benchmark on a quiet PC without any interferences from other programs. It might help to use the '--warmup' or '--prepare' options.

Benchmark 3: fd-v8.7.0-x86_64-unknown-linux-gnu/fd . empty/
  Time (mean ± σ):      38.4 ms ±   2.9 ms    [User: 3.5 ms, System: 35.9 ms]
  Range (min … max):    35.5 ms …  55.3 ms    81 runs

  Warning: Statistical outliers were detected. Consider re-running this benchmark on a quiet PC without any interferences from other programs. It might help to use the '--warmup' or '--prepare' options.

Summary
  'fd-v8.4.0-x86_64-unknown-linux-gnu/fd . empty/' ran
    9.68 ± 1.12 times faster than 'fd-v8.7.0-x86_64-unknown-linux-gnu/fd . empty/'
    9.88 ± 1.05 times faster than 'fd-v8.5.0-x86_64-unknown-linux-gnu/fd . empty/'
```

FWIW this is not an issue with startup time per se; running `fd --version` does not show any performance drop. 

Using `-F ""` instead of `.` for the pattern show the same performance regression, if that's helpful to know.

I am happy to run any additional experiments that might be useful, but I'm afraid that my rust-fu is too weak to be of any use for deeper analysis or proposing a fix.

### Describe what you expected to happen:

_No response_

### What version of `fd` are you using?

fd 8.4.0 fd 8.5.0 fd 8.7.0

### Which operating system / distribution are you on?

```shell
> uname -srm
Linux 5.15.0-76-generic x86_64
> lsb_release -a
No LSB modules are available.
Distributor ID:	Ubuntu
Description:	Ubuntu 22.04.2 LTS
Release:	22.04
Codename:	jammy
```



Discussion / hints:
> 11 ms are spent in crossbeam_channel::channel::bounded's from_iter method?

That might be where crossbeam_channel initializes the memory for the channel
[here](https://github.com/crossbeam-rs/crossbeam/blob/9dd4c9594dd7c84abf04e89192f1bdfe790100f5/crossbeam-channel/src/flavors/array.rs#L116).
Is there any slowdown for non-empty directories?  Time to search an empty directory is not really something we optimize for, as long as it's still reasonably fast.

I did do a quick `perf diff` and the difference appears to be caused by the switch to a bounded-capacity channel: https://github.com/sharkdp/fd/commit/527840526396341f027b17f4d51eeb5ecdf8eacb.  Looks like the overhead is just from initializing the memory allocated for the channel.
Right, what I am reporting here is not about empty directories per se, it is about initialization overhead. I ran a simple test of searching for `.` in a directory with `N` empty files in it for `N` increasing by powers of 10. Up through 10000 files I see basically a constant slowdown of 30 ms e.g. for N=1 it takes ~ 3 ms on 8.4 and ~ 33 ms on later versions, and for N=1000 it takes ~15 ms for 8.3 and 45 ms for later versions. 

With N=10,000 what I suppose are some under-the-hood asymptotic improvements start to become noticeable. For N=100,000  the gap closes to 20 ms (100 ms vs. 120 ms) and for N=1,000,000 the asymptotic effect dominates and we see what we would always like to:
```
  'fd-v8.7.0-x86_64-unknown-linux-gnu/fd . /tmp/bench/1000000-files/' ran
    1.10 ± 0.03 times faster than 'fd-v8.5.0-x86_64-unknown-linux-gnu/fd . /tmp/bench/1000000-files/'
    1.20 ± 0.03 times faster than 'fd-v8.4.0-x86_64-unknown-linux-gnu/fd . /tmp/bench/1000000-files/'
 ```
Looking at the linked PR suggests to me that the effect out to be linear in `--threads`, which it basically is. So the 10x slowdown for trivial searches that I reported is an artifact of my machine having nproc=16. But with `--threads=128` (not implausible!) we see 
```
  'fd-v8.4.0-x86_64-unknown-linux-gnu/fd --threads=128 . /tmp/bench/1-files/' ran
   35.42 ± 4.09 times faster than 'fd-v8.7.0-x86_64-unknown-linux-gnu/fd --threads=128 . /tmp/bench/1-files/'
``` 
and the respective wall-clock times are 5.5 ms vs. 193.4 ms which is IMO worth optimizing for if not too hard to fix.

I'm not a rust guy, but I am rather curious why it takes so long to initialize memory; even with nproc=128 we're only talking about 2 megabytes, right? Should be ~20x faster.. Maybe a rabbit hole!

oh whoops I just realized that capacity is 2M *messages* and presumably we are allocation rather more than 1 byte per message, duh...
Note that there is also linear overhead just to start N threads, which is significant at -j128.  See also #1203.

Also I suspect the overhead is actually more from syscall/VM locking overhead for mmap() than initialization, now that I think about it.
Ah good catch, I suspect that this issue is probably a duplicate of #1203 since it looks like that guy is based on timings from 8.4. I certainly won't disagree that starting N threads has linear overhead however this really ought to be negligible for N=128. Please note that in my last comparison of 8.3 (5.5 ms) vs. 8.7 (193.4 ms) I used `--threads=128` for _both_ versions. 

  

Fix patch:
diff --git a/src/dir_entry.rs b/src/dir_entry.rs
index 3a19d5982..f44f2beb8 100644
--- a/src/dir_entry.rs
+++ b/src/dir_entry.rs
@@ -8,11 +8,13 @@ use lscolors::{Colorable, LsColors, Style};
 use crate::config::Config;
 use crate::filesystem::strip_current_dir;
 
+#[derive(Debug)]
 enum DirEntryInner {
     Normal(ignore::DirEntry),
     BrokenSymlink(PathBuf),
 }
 
+#[derive(Debug)]
 pub struct DirEntry {
     inner: DirEntryInner,
     metadata: OnceCell<Option<Metadata>>,
diff --git a/src/exec/job.rs b/src/exec/job.rs
index af603cc63..4864d6dca 100644
--- a/src/exec/job.rs
+++ b/src/exec/job.rs
@@ -1,9 +1,6 @@
 use std::sync::Mutex;
 
-use crossbeam_channel::Receiver;
-
 use crate::config::Config;
-use crate::dir_entry::DirEntry;
 use crate::error::print_error;
 use crate::exit_codes::{merge_exitcodes, ExitCode};
 use crate::walk::WorkerResult;
@@ -14,7 +11,7 @@ use super::CommandSet;
 /// generate a command with the supplied command template. The generated command will then
 /// be executed, and this process will continue until the receiver's sender has closed.
 pub fn job(
-    rx: Receiver<WorkerResult>,
+    results: impl IntoIterator<Item = WorkerResult>,
     cmd: &CommandSet,
     out_perm: &Mutex<()>,
     config: &Config,
@@ -22,35 +19,39 @@ pub fn job(
     // Output should be buffered when only running a single thread
     let buffer_output: bool = config.threads > 1;
 
-    let mut results: Vec<ExitCode> = Vec::new();
-    loop {
+    let mut ret = ExitCode::Success;
+    for result in results {
         // Obtain the next result from the receiver, else if the channel
         // has closed, exit from the loop
-        let dir_entry: DirEntry = match rx.recv() {
-            Ok(WorkerResult::Entry(dir_entry)) => dir_entry,
-            Ok(WorkerResult::Error(err)) => {
+        let dir_entry = match result {
+            WorkerResult::Entry(dir_entry) => dir_entry,
+            WorkerResult::Error(err) => {
                 if config.show_filesystem_errors {
                     print_error(err.to_string());
                 }
                 continue;
             }
-            Err(_) => break,
         };
 
         // Generate a command, execute it and store its exit code.
-        results.push(cmd.execute(
+        let code = cmd.execute(
             dir_entry.stripped_path(config),
             config.path_separator.as_deref(),
             out_perm,
             buffer_output,
-        ))
+        );
+        ret = merge_exitcodes([ret, code]);
     }
     // Returns error in case of any error.
-    merge_exitcodes(results)
+    ret
 }
 
-pub fn batch(rx: Receiver<WorkerResult>, cmd: &CommandSet, config: &Config) -> ExitCode {
-    let paths = rx
+pub fn batch(
+    results: impl IntoIterator<Item = WorkerResult>,
+    cmd: &CommandSet,
+    config: &Config,
+) -> ExitCode {
+    let paths = results
         .into_iter()
         .filter_map(|worker_result| match worker_result {
             WorkerResult::Entry(dir_entry) => Some(dir_entry.into_stripped_path(config)),
diff --git a/src/walk.rs b/src/walk.rs
index 691c5d0f0..c6737edf1 100644
--- a/src/walk.rs
+++ b/src/walk.rs
@@ -4,12 +4,12 @@ use std::io::{self, Write};
 use std::mem;
 use std::path::PathBuf;
 use std::sync::atomic::{AtomicBool, Ordering};
-use std::sync::{Arc, Mutex};
+use std::sync::{Arc, Mutex, MutexGuard};
 use std::thread;
 use std::time::{Duration, Instant};
 
 use anyhow::{anyhow, Result};
-use crossbeam_channel::{bounded, Receiver, RecvTimeoutError, Sender};
+use crossbeam_channel::{bounded, Receiver, RecvTimeoutError, SendError, Sender};
 use etcetera::BaseStrategy;
 use ignore::overrides::{Override, OverrideBuilder};
 use ignore::{self, WalkBuilder, WalkParallel, WalkState};
@@ -36,6 +36,7 @@ enum ReceiverMode {
 
 /// The Worker threads can result in a valid entry having PathBuf or an error.
 #[allow(clippy::large_enum_variant)]
+#[derive(Debug)]
 pub enum WorkerResult {
     // Errors should be rare, so it's probably better to allow large_enum_variant than
     // to box the Entry variant
@@ -43,6 +44,83 @@ pub enum WorkerResult {
     Error(ignore::Error),
 }
 
+/// A batch of WorkerResults to send over a channel.
+#[derive(Clone)]
+struct Batch {
+    items: Arc<Mutex<Option<Vec<WorkerResult>>>>,
+}
+
+impl Batch {
+    fn new() -> Self {
+        Self {
+            items: Arc::new(Mutex::new(Some(vec![]))),
+        }
+    }
+
+    fn lock(&self) -> MutexGuard<'_, Option<Vec<WorkerResult>>> {
+        self.items.lock().unwrap()
+    }
+}
+
+impl IntoIterator for Batch {
+    type Item = WorkerResult;
+    type IntoIter = std::vec::IntoIter<WorkerResult>;
+
+    fn into_iter(self) -> Self::IntoIter {
+        self.lock().take().unwrap().into_iter()
+    }
+}
+
+/// Wrapper that sends batches of items at once over a channel.
+struct BatchSender {
+    batch: Batch,
+    tx: Sender<Batch>,
+    limit: usize,
+}
+
+impl BatchSender {
+    fn new(tx: Sender<Batch>, limit: usize) -> Self {
+        Self {
+            batch: Batch::new(),
+            tx,
+            limit,
+        }
+    }
+
+    /// Check if we need to flush a batch.
+    fn needs_flush(&self, batch: Option<&Vec<WorkerResult>>) -> bool {
+        match batch {
+            // Limit the batch size to provide some backpressure
+            Some(vec) => vec.len() >= self.limit,
+            // Batch was already taken by the receiver, so make a new one
+            None => true,
+        }
+    }
+
+    /// Add an item to a batch.
+    fn send(&mut self, item: WorkerResult) -> Result<(), SendError<()>> {
+        let mut batch = self.batch.lock();
+
+        if self.needs_flush(batch.as_ref()) {
+            drop(batch);
+            self.batch = Batch::new();
+            batch = self.batch.lock();
+        }
+
+        let items = batch.as_mut().unwrap();
+        items.push(item);
+
+        if items.len() == 1 {
+            // New batch, send it over the channel
+            self.tx
+                .send(self.batch.clone())
+                .map_err(|_| SendError(()))?;
+        }
+
+        Ok(())
+    }
+}
+
 /// Maximum size of the output buffer before flushing results to the console
 const MAX_BUFFER_LENGTH: usize = 1000;
 /// Default duration until output buffering switches to streaming.
@@ -57,7 +135,7 @@ struct ReceiverBuffer<'a, W> {
     /// The ^C notifier.
     interrupt_flag: &'a AtomicBool,
     /// Receiver for worker results.
-    rx: Receiver<WorkerResult>,
+    rx: Receiver<Batch>,
     /// Standard output.
     stdout: W,
     /// The current buffer mode.
@@ -72,7 +150,7 @@ struct ReceiverBuffer<'a, W> {
 
 impl<'a, W: Write> ReceiverBuffer<'a, W> {
     /// Create a new receiver buffer.
-    fn new(state: &'a WorkerState, rx: Receiver<WorkerResult>, stdout: W) -> Self {
+    fn new(state: &'a WorkerState, rx: Receiver<Batch>, stdout: W) -> Self {
         let config = &state.config;
         let quit_flag = state.quit_flag.as_ref();
         let interrupt_flag = state.interrupt_flag.as_ref();
@@ -103,7 +181,7 @@ impl<'a, W: Write> ReceiverBuffer<'a, W> {
     }
 
     /// Receive the next worker result.
-    fn recv(&self) -> Result<WorkerResult, RecvTimeoutError> {
+    fn recv(&self) -> Result<Batch, RecvTimeoutError> {
         match self.mode {
             ReceiverMode::Buffering => {
                 // Wait at most until we should switch to streaming
@@ -119,36 +197,42 @@ impl<'a, W: Write> ReceiverBuffer<'a, W> {
     /// Wait for a result or state change.
     fn poll(&mut self) -> Result<(), ExitCode> {
         match self.recv() {
-            Ok(WorkerResult::Entry(dir_entry)) => {
-                if self.config.quiet {
-                    return Err(ExitCode::HasResults(true));
-                }
+            Ok(batch) => {
+                for result in batch {
+                    match result {
+                        WorkerResult::Entry(dir_entry) => {
+                            if self.config.quiet {
+                                return Err(ExitCode::HasResults(true));
+                            }
 
-                match self.mode {
-                    ReceiverMode::Buffering => {
-                        self.buffer.push(dir_entry);
-                        if self.buffer.len() > MAX_BUFFER_LENGTH {
-                            self.stream()?;
-                        }
-                    }
-                    ReceiverMode::Streaming => {
-                        self.print(&dir_entry)?;
-                        self.flush()?;
-                    }
-                }
+                            match self.mode {
+                                ReceiverMode::Buffering => {
+                                    self.buffer.push(dir_entry);
+                                    if self.buffer.len() > MAX_BUFFER_LENGTH {
+                                        self.stream()?;
+                                    }
+                                }
+                                ReceiverMode::Streaming => {
+                                    self.print(&dir_entry)?;
+                                    self.flush()?;
+                                }
+                            }
 
-                self.num_results += 1;
-                if let Some(max_results) = self.config.max_results {
-                    if self.num_results >= max_results {
-                        return self.stop();
+                            self.num_results += 1;
+                            if let Some(max_results) = self.config.max_results {
+                                if self.num_results >= max_results {
+                                    return self.stop();
+                                }
+                            }
+                        }
+                        WorkerResult::Error(err) => {
+                            if self.config.show_filesystem_errors {
+                                print_error(err.to_string());
+                            }
+                        }
                     }
                 }
             }
-            Ok(WorkerResult::Error(err)) => {
-                if self.config.show_filesystem_errors {
-                    print_error(err.to_string());
-                }
-            }
             Err(RecvTimeoutError::Timeout) => {
                 self.stream()?;
             }
@@ -319,13 +403,13 @@ impl WorkerState {
 
     /// Run the receiver work, either on this thread or a pool of background
     /// threads (for --exec).
-    fn receive(&self, rx: Receiver<WorkerResult>) -> ExitCode {
+    fn receive(&self, rx: Receiver<Batch>) -> ExitCode {
         let config = &self.config;
 
         // This will be set to `Some` if the `--exec` argument was supplied.
         if let Some(ref cmd) = config.command {
             if cmd.in_batch_mode() {
-                exec::batch(rx, cmd, &config)
+                exec::batch(rx.into_iter().flatten(), cmd, &config)
             } else {
                 let out_perm = Mutex::new(());
 
@@ -337,7 +421,8 @@ impl WorkerState {
                         let rx = rx.clone();
 
                         // Spawn a job thread that will listen for and execute inputs.
-                        let handle = scope.spawn(|| exec::job(rx, cmd, &out_perm, &config));
+                        let handle = scope
+                            .spawn(|| exec::job(rx.into_iter().flatten(), cmd, &out_perm, &config));
 
                         // Push the handle of the spawned thread into the vector for later joining.
                         handles.push(handle);
@@ -355,12 +440,20 @@ impl WorkerState {
     }
 
     /// Spawn the sender threads.
-    fn spawn_senders(&self, walker: WalkParallel, tx: Sender<WorkerResult>) {
+    fn spawn_senders(&self, walker: WalkParallel, tx: Sender<Batch>) {
         walker.run(|| {
             let patterns = &self.patterns;
             let config = &self.config;
             let quit_flag = self.quit_flag.as_ref();
-            let tx = tx.clone();
+
+            let mut limit = 0x100;
+            if let Some(cmd) = &config.command {
+                if !cmd.in_batch_mode() && config.threads > 1 {
+                    // Evenly distribute work between multiple receivers
+                    limit = 1;
+                }
+            }
+            let mut tx = BatchSender::new(tx.clone(), limit);
 
             Box::new(move |entry| {
                 if quit_flag.load(Ordering::Relaxed) {
@@ -545,8 +638,7 @@ impl WorkerState {
             .unwrap();
         }
 
-        // Channel capacity was chosen empircally to perform similarly to an unbounded channel
-        let (tx, rx) = bounded(0x4000 * config.threads);
+        let (tx, rx) = bounded(2 * config.threads);
 
         let exit_code = thread::scope(|scope| {
             // Spawn the receiver thread(s)


Test patch: