Walking directory and limiting concurrency

[mrzv]

I'm trying to walk a directory using the code below. It works fine when the directory is small. But on large directories, I quickly get "Couldn't read the directory" because of "Too many open files".

The basic problem is clear: I'm traversing the tree in a breadth-first manner, with a task for each directory. If too many of these are spawned at once, it exhausts an OS limit. Is there a clean way to solve this? Perhaps by somehow limiting simultaneous concurrency, so only so many scans could be running at once?

use std::path::PathBuf;
use tokio::{fs,sync::mpsc};

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel(1024);

    scan(PathBuf::from("."), tx).await;

    while let Some(p) = rx.recv().await {
        println!("{:?}", p);
    }
}

async fn scan(path: PathBuf, tx: mpsc::Sender<PathBuf>) {
    let mut dir = fs::read_dir(path).await.expect("Couldn't read the directory");

    let mut children = Vec::new();
    while let Some(child) = dir.next_entry().await.expect("Couldn't read the next directory entry") {
        let path = child.path();
        let meta = fs::symlink_metadata(&path).await.expect("Couldn't get metadata");

        tx.send(path.clone()).await;

        if meta.is_dir() {
            let path = path.clone();
            children.push(path);
        }
    }

    scan_children(children, tx);
}

fn scan_children(children: Vec<PathBuf>, tx: mpsc::Sender<PathBuf>) {
    for path in children {
        let tx = tx.clone();
        tokio::spawn(async move {
            scan(path, tx).await
        });
    }
}

[Darksonn]

Put a Semaphore in an Arc to share it between all the tasks and acquire a new permit on the semaphore every time you open a file (but before opening it). If you make sure the permit is dropped after the file descriptor is closed, this will ensure that no more files are opened than there are permits in the semaphore.

[mrzv]

I just tried that, adding

    let s = Arc::new(Semaphore::new(10));

to main, passing it (or its clones) to scan and scan_children, and put

    let _sp = s.acquire();

at the beginning of scan. But I still get the same error.

[Darksonn]

The acquire method returns a future, which must be awaited. The #[must_use] warning was suppressed because you stored the future in variable.

[mrzv]

Ok, great, adding await fixed it. Thanks. Let me just double-check. I need to store the result of the await, the actual permit, in a variable (_sp), so that it's dropped at the end of the function. Otherwise, it will be released as soon as it's acquired. Is that correct?

[Darksonn]

Yes. I usually prefer to explicitly drop it at the end with drop(sp).

[mrzv]

Got it. Thanks.

[mrzv]

I have one more, only tangentially related question. In my example, if I make all the functions return Result<()> and change the first line of scan from:

let mut dir = fs::read_dir(path).await.expect("Couldn't read the directory");

to

let mut dir = fs::read_dir(path).await?;

I don't get the errors, but instead the code fails silently: only a small fraction of the files get reported. I guess the issue is that the errors don't get propagated to the right place. Is there a standard, idiomatic way to fix this?

[Darksonn]

This is most likely because you are ignoring any errors returned by scan here:

tokio::spawn(async move {
    scan(path, tx).await
});

You should match on the return value of scan and handle the error in some way. What way is appropriate will depend on what you want to happen if a failure is detected.

[mrzv]

That makes sense, but is there a simple way to propagate the error to main? And have a "global crash" as one would normally.

[Darksonn]

You could change the sender to mpsc::Sender<Result<PathBuf, io::Error>> and send the error to the main function. If you also want to kill all the other tasks, you could check out my answer on that on stack overflow.

(Though, in this case, all tasks are killed when you return from main.)

[mrzv]

Got it. Thanks for all your help.