Repository: sharkdp/fd (language: rust)

Issue:
excessive memory usage on huge trees
When traversing some huge directory (5TB data/ 200Million Entries) i noticed that 'fd' will use excessive amounts of memory (I killed it at 35GB RSS). Some investigation shown that 'fd | wc -l' or 'fd >/dev/null' does not have this problem. While 'fd | less' again shows the same problem.

So my guess is that fd uses some unbounded queues to send the output which pile up because the terminal emulator is too slow to print 200 Million entries in time.


Discussion / hints:
Which version of fd are you using?
$ fd --version
fd 8.3.0

installed by cargo install
Ah, I thought that we had set a limit on the channel size in #885, but it looks like that part got removed. So, yes, there is an unbounded queue of files and directories to process, and that could cause high memory usage if the terminal can't keep up. Maybe we should revisit adding a bound to that. I believe with the std mpsc crate there is  a performance cost to doing that, I'm not sure about crossbeam.
> but it looks like that part got removed

Right. I removed it because you wrote *"I switch back to using channel instead of sync_channel than it is, at least not any slower"*. And I believe I was able to confirm this in my own benchmarks.

> Maybe we should revisit adding a bound to that. I believe with the std mpsc crate there is a performance cost to doing that, I'm not sure about crossbeam.

Yes, let's do this in a dedicated PR with new benchmark results (and memory profiles). A quick fix could be to use a much higher limit than the one suggested in #885.
hint: in my own directory traversal code i found that about 4k per thread (for memory calculation, not one channel per thread) is a somewhat sweet spot. Actually the limit isn't terribly important as long it doesn't explode, even with 100MB limit it should be okish.
Is this the cause of fd outputting "Killed" and then quitting when I search for all files with a specific extension from the root (Arch Linux, ~9/~14TB used space to search through across 5 mounted partitions)?
@mwaitzman It might be, do you see anything about the OOM killer in `dmesg`?
Yes, I see when running a similar command
```console
[ 4279.148487] Out of memory: Killed process 7330 (fd) total-vm:61310736kB, anon-rss:24569660kB, file-rss:0kB, shmem-rss:0kB, UID:0 pgtables:111272kB oom_score_adj:0
[ 4285.522887] oom_reaper: reaped process 7330 (fd), now anon-rss:0kB, file-rss:0kB, shmem-rss:0kB
```
Okay yeah, I think this is probably due to the lack of backpressure with the unbounded channel.  It's unfortunate that `std::sync::mpsc::sync_channel()` (the bounded queue) performs so much worse than `channel()` (unbounded).

The SPSC case (`-j1`) is 10-40% worse with bounded queues, and the performance is *worse* for higher capacities, so tuning the capacity is crucial.  With a capacity of `4096 * threads`, bounded queues match unbounded queues at `-j8` but not with fewer threads.

I'd like to figure out https://github.com/sharkdp/fd/issues/933.
On 2022-06-03 12:37, Tavian Barnes wrote:

>Okay yeah, I think this is probably due to the lack of backpressure
>with the unbounded channel.  It's unfortunate that
>`std::sync::mpsc::sync_channel()` (the bounded queue) performs so much
>worse than `channel()` (unbounded).
>
>The SPSC case (`-j1`) is 10-40% worse with bounded queues, and the
>performance is *worse* for higher capacities, so tuning the capacity
>is crucial.  With a capacity of `4096 * threads`, bounded queues match
>unbounded queues at `-j8` but not with fewer threads.

Just an observation I did with other code:

Walking directories can be very fast when they are in RAM (obliviously)
but otherwise it is pretty much I/O-bound. And here lies the trick:
since you are not CPU-Bound but IO-bound you can have *a lot* more
threads than you have cores. Which in turn gives the Kernel the
opportunity to schedule requests in a much better way. I've tested up
to 128 threads on a 12/24 core machine and still seen improvements (in
my code, not the 'fd' tree walking code). This can by far outweigh the
performance degrade you get from bounded channels.

>
>I'd like to figure out https://github.com/sharkdp/fd/issues/933.
>
>-- 
>Reply to this email directly or view it on GitHub:
>https://github.com/sharkdp/fd/issues/918#issuecomment-1146299136
>You are receiving this because you authored the thread.
>
>Message ID: ***@***.***>


Experiencing the exact same issue, but with `--exec`. I'm trying to run `fd "png$" / --exec optipng -o 7` (I know that this usage of optipng may break stuff, I know what I'm doing).

I've tried leaving different variations running on my laptop for the night, finding it absolutely frozen every time i wake up, arriving at `fd "png$" --threads=1 --exec sh -c 'optipng -o 7 "{}"; sync; sleep 1m'` to try to minimize impact. At this point I've checked with a task manager, and it's indeed `fd` ballooning.

Currently on `fd 8.4.0` from Arch Linux repositories. Is there anything I can do on my end to do what I want, besides resorting to `find`?
@Architector4  you might have some luck filtering out some directories. It looks like you are running on /, so excluding /proc, /sys, /tmp, etc. Would likely make a difference (if you are on mac those paths might be different). If the files you care about are all on the same filesystem you could try the --one-file-system option.
@tmccombs I don't think it would make any meaningful difference in my case -- I'm looking for all files that end with "png", of which there would not be any in such special directories. I imagine (I hope?) that `fd` does not put files not matching the search terms into that ballooning cache.

I'd also prefer to go through all filesystems (my SSD mounted on `/` and my HDD mounted in `/media`), so I don't want to use that option either. But yeah, those are good ways to help alleviate the problem for other cases where applicable.

In my case I decided to resort to `find` and have had good success with it. This is what I ended up running to achieve the same task without running out of memory, with GNU `find` and `parallel`:

    find / -iname "*.png" -print0 | parallel -0 optipng -o 7 {}
Oh, I think you're right. In your case probably what is happening is the queues are filling up because running `optipng` is blocking processing additional entries, and there isn't any backpressure to stop new entries added to the queue. If you were to use fd piped to parellel, my guess is it would probably work better. What we really need here is a bounded queue. The question is how to do that without hurting performance in cases where backpressure isn't needed. 
I feel like the searching thread's performance would be I/O bound most of the time (input being filesystem traversal and output being to `--exec` or stdout or whatnot), and hence some simple "if over 10000 items in queue then block adding to it until the next item is consumed" kind of a statement slapped right before the "add to queue" part would probably not introduce too much of performance drawback.

In any case, this feels like a necessary feature for wellbeing of the host machine, because it seems like this excessive memory usage applies to any general case where such output is slower than such input. To demonstrate, I have a silly script, named `delaystream.sh`:

    while read -r line; do
    	echo "$line"
    	sleep 1
    done

This script prints each line it receives on stdin to stdout, but waits 1 second between them, causing artificially slow I/O. So I piped `fd` to it:

    fd / | delaystream.sh

This *reliably* makes `fd` balloon in memory usage.

I argue some kind of a limit is absolutely necessary here, and is more important than the performance from not having to account for it.

Maybe it would be possible to introduce a command line switch that disables the limit for those who wish to have the performance benefits and know ballooning wouldn't happen?
> I feel like the searching thread's performance would be I/O bound most of the time (input being filesystem traversal and output being to `--exec` or stdout or whatnot), and hence some simple "if over 10000 items in queue then block adding to it until the next item is consumed" kind of a statement slapped right before the "add to queue" part would probably not introduce too much of performance drawback.

Ideally yes, but: https://github.com/sharkdp/fd/issues/918#issuecomment-1146299136

> In any case, this feels like a necessary feature for wellbeing of the host machine

I agree, we need to do something.  I can try switching to crossbeam-channels again.

Fix patch:
diff --git a/Cargo.lock b/Cargo.lock
index e856d0da3..0b21f8ef6 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -162,6 +162,16 @@ version = "0.8.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "5827cebf4670468b8772dd191856768aedcb1b0278a04f989f7766351917b9dc"
 
+[[package]]
+name = "crossbeam-channel"
+version = "0.5.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c2dd04ddaf88237dc3b8d8f9a3c1004b506b54b3313403944054d23c0870c521"
+dependencies = [
+ "cfg-if",
+ "crossbeam-utils",
+]
+
 [[package]]
 name = "crossbeam-utils"
 version = "0.8.12"
@@ -295,6 +305,7 @@ dependencies = [
  "chrono",
  "clap",
  "clap_complete",
+ "crossbeam-channel",
  "ctrlc",
  "diff",
  "dirs-next",
diff --git a/Cargo.toml b/Cargo.toml
index 5f627fd62..fad68d532 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -51,6 +51,7 @@ dirs-next = "2.0"
 normpath = "0.3.2"
 chrono = "0.4"
 once_cell = "1.15.0"
+crossbeam-channel = "0.5.6"
 
 [dependencies.clap]
 version = "3.1"
diff --git a/src/exec/job.rs b/src/exec/job.rs
index b803f79c8..1617ea752 100644
--- a/src/exec/job.rs
+++ b/src/exec/job.rs
@@ -1,6 +1,7 @@
-use std::sync::mpsc::Receiver;
 use std::sync::{Arc, Mutex};
 
+use crossbeam_channel::Receiver;
+
 use crate::config::Config;
 use crate::dir_entry::DirEntry;
 use crate::error::print_error;
@@ -13,7 +14,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: Arc<Mutex<Receiver<WorkerResult>>>,
+    rx: Receiver<WorkerResult>,
     cmd: Arc<CommandSet>,
     out_perm: Arc<Mutex<()>>,
     config: &Config,
@@ -23,12 +24,9 @@ pub fn job(
 
     let mut results: Vec<ExitCode> = Vec::new();
     loop {
-        // Create a lock on the shared receiver for this thread.
-        let lock = rx.lock().unwrap();
-
         // Obtain the next result from the receiver, else if the channel
         // has closed, exit from the loop
-        let dir_entry: DirEntry = match lock.recv() {
+        let dir_entry: DirEntry = match rx.recv() {
             Ok(WorkerResult::Entry(dir_entry)) => dir_entry,
             Ok(WorkerResult::Error(err)) => {
                 if config.show_filesystem_errors {
@@ -39,8 +37,6 @@ pub fn job(
             Err(_) => break,
         };
 
-        // Drop the lock so that other threads can read from the receiver.
-        drop(lock);
         // Generate a command, execute it and store its exit code.
         results.push(cmd.execute(
             dir_entry.stripped_path(config),
diff --git a/src/walk.rs b/src/walk.rs
index 799828224..b5b665070 100644
--- a/src/walk.rs
+++ b/src/walk.rs
@@ -3,13 +3,13 @@ use std::io;
 use std::mem;
 use std::path::PathBuf;
 use std::sync::atomic::{AtomicBool, Ordering};
-use std::sync::mpsc::{channel, Receiver, RecvTimeoutError, Sender};
 use std::sync::{Arc, Mutex};
 use std::thread;
 use std::time::{Duration, Instant};
 use std::{borrow::Cow, io::Write};
 
 use anyhow::{anyhow, Result};
+use crossbeam_channel::{bounded, Receiver, RecvTimeoutError, Sender};
 use ignore::overrides::OverrideBuilder;
 use ignore::{self, WalkBuilder};
 use regex::bytes::Regex;
@@ -51,7 +51,9 @@ pub const DEFAULT_MAX_BUFFER_TIME: Duration = Duration::from_millis(100);
 /// path will simply be written to standard output.
 pub fn scan(paths: &[PathBuf], pattern: Arc<Regex>, config: Arc<Config>) -> Result<ExitCode> {
     let first_path = &paths[0];
-    let (tx, rx) = channel();
+
+    // Channel capacity was chosen empircally to perform similarly to an unbounded channel
+    let (tx, rx) = bounded(0x4000 * config.threads);
 
     let mut override_builder = OverrideBuilder::new(first_path);
 
@@ -222,11 +224,7 @@ impl<W: Write> ReceiverBuffer<W> {
         match self.mode {
             ReceiverMode::Buffering => {
                 // Wait at most until we should switch to streaming
-                let now = Instant::now();
-                self.deadline
-                    .checked_duration_since(now)
-                    .ok_or(RecvTimeoutError::Timeout)
-                    .and_then(|t| self.rx.recv_timeout(t))
+                self.rx.recv_deadline(self.deadline)
             }
             ReceiverMode::Streaming => {
                 // Wait however long it takes for a result
@@ -345,15 +343,13 @@ fn spawn_receiver(
             if cmd.in_batch_mode() {
                 exec::batch(rx, cmd, &config)
             } else {
-                let shared_rx = Arc::new(Mutex::new(rx));
-
                 let out_perm = Arc::new(Mutex::new(()));
 
                 // Each spawned job will store it's thread handle in here.
                 let mut handles = Vec::with_capacity(threads);
                 for _ in 0..threads {
                     let config = Arc::clone(&config);
-                    let rx = Arc::clone(&shared_rx);
+                    let rx = rx.clone();
                     let cmd = Arc::clone(cmd);
                     let out_perm = Arc::clone(&out_perm);
 


Test patch: