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

[dduugg]

walk's result channel holds threads * 2 batches. When the iterator's consumer pauses, each worker that finishes a batch blocks until the consumer drains the channel, so the walk stops while the consumer is busy. This allows 128 batches per worker instead (QUEUED_BATCHES_PER_WORKER). The channel stays bounded.

I ran into this moving pks, a Rust port of Shopify's packwerk, from jwalk to dua-core, as jwalk's changelog suggests. pks walks a Rails app to find its Ruby files and the package.yml files that own them, and parses each package.yml as it reaches it. Each of those pauses fills the channel and leaves the workers blocked.

Measurements

On an Apple M4 Pro (12 cores, macOS 27), with a warm cache. The machine was busy throughout (load average around 14), so the totals are noisy. The time the consumer spends waiting on the iterator is the steadier signal.

pks on a large Rails app, medians of 7 runs, with the bound read from an environment variable for the experiment:

Batches per worker	pks's walk loop	Waiting on the iterator
2 (now)	580 ms	192 ms
8	477 ms	73 ms
32	451 ms	42 ms
128	436 ms	11 ms

With jwalk 0.9, the same loop took 454 ms and waited 25 ms.

A standalone repro (below) walks a tree of 23,431 directories and 90,000 empty files in completion order with skip_metadata, spending 3 µs on each entry and 150 µs more on every 150th. Four runs each:

	Total	Waiting on the iterator
dua-core 4.1.0	537 to 652 ms	63 to 137 ms
This branch	550 to 573 ms	18 to 24 ms
jwalk 0.9.0	561 to 718 ms	22 to 50 ms

dua itself is no slower. With hyperfine, 15 runs each, dua aggregate took 233 ± 40 ms before and 189 ± 33 ms after on a checkout of ruby/ruby, and 805 ± 105 ms and 801 ± 88 ms on the synthetic tree.

Repro

The tree:

import os

for a in range(30):
    for b in range(30):
        for c in range(25):
            d = f"d{a}/d{b}/d{c}"
            os.makedirs(d, exist_ok=True)
            for f in range(4):
                open(f"{d}/f{f}.rb", "w").close()

Cargo.toml:

[package]
name = "walkbench"
version = "0.1.0"
edition = "2021"

[dependencies]
jwalk = "=0.9.0"
dua-core = "=4.1.0"

[profile.release]
codegen-units = 1

src/main.rs:

// Walks a tree with jwalk or dua-core, spending SPIN_NS nanoseconds of busy
// work on each entry, and PAUSE_NS more on every PAUSE_EVERY-th entry, and
// reports how long the consumer waited on the walk.
use std::time::{Duration, Instant};

fn spin(ns: u64) {
    let start = Instant::now();
    while start.elapsed() < Duration::from_nanos(ns) {}
}

fn main() {
    let args: Vec<String> = std::env::args().collect();
    let (mode, root) = (args[1].as_str(), std::path::PathBuf::from(&args[2]));
    let var = |name: &str| std::env::var(name).map_or(0, |v| v.parse::<u64>().unwrap());
    let (ns, pause_every, pause_ns) = (var("SPIN_NS"), var("PAUSE_EVERY"), var("PAUSE_NS"));
    let threads = std::thread::available_parallelism().map_or(1, |n| n.get());

    let start = Instant::now();
    let mut wait = Duration::ZERO;
    let mut entries = 0usize;
    let mut iter: Box<dyn Iterator<Item = ()>> = match mode {
        "jwalk" => Box::new(jwalk::WalkDir::new(&root).into_iter().map(|e| drop(e.unwrap()))),
        "dua" => Box::new(
            dua_core::walk(&root, threads, dua_core::Order::Completion, dua_core::Options::default().skip_metadata(), |_| true)
                .map(|e| drop(e.unwrap())),
        ),
        _ => panic!("mode is jwalk or dua"),
    };
    loop {
        let waiting = Instant::now();
        let Some(()) = iter.next() else { break };
        wait += waiting.elapsed();
        entries += 1;
        spin(ns);
        if pause_every > 0 && entries as u64 % pause_every == 0 {
            spin(pause_ns);
        }
    }
    println!(
        "{mode}: {entries} entries in {:.0} ms, waited {:.0} ms",
        start.elapsed().as_secs_f64() * 1e3,
        wait.as_secs_f64() * 1e3
    );
}

Run with SPIN_NS=3000 PAUSE_EVERY=150 PAUSE_NS=150000 target/release/walkbench dua <tree>, and with a [patch.crates-io] entry pointing dua-core at this branch for the comparison.

Tests

Three unit tests block a worker on a full channel by filling it with two batches: streaming_input_closes_and_cancels_without_draining_output, dropping_a_walk_disconnects_workers_blocked_on_full_output and cancelling_a_walk_disconnects_workers_blocked_on_full_output. With 128 batches per worker they would still pass but no longer block anything, so under cfg(test) the bound stays at two.

cargo test --workspace (309 tests) and cargo clippy --workspace --all-targets -- -D warnings pass.

[Byron]

Thanks! I see why I'd never notice this, as in my tests nobody ever blocks.
Instead of hard-coded it to something else, let's make it configurable while keeping the current default which is optimized for low memory usage.

[dduugg]

Thanks for merging this, and for making it an option.

I tried main (4179fe2) in pks, passing Options::default().skip_metadata().queued_batches_per_worker(128) in place of the thread pks uses for now to drain the walk. Over 15 runs on the same large Rails app, pks's loop waited a median 64 ms on the iterator, against 301 ms with the default of 2 and 113 ms with the thread. 512 did no better than 128. The machine was busier than for the numbers above (a load average of about 22 on 12 cores), which is why every wait is longer. pks's tests pass against main, and its output doesn't change.

Would you cut a dua-core release with it when it suits you? pks will drop the thread then.

[Byron]

Great!
And sure thing, here it is: https://github.com/Byron/dua-cli/releases/tag/dua-core-v4.2.0