Processing a Stream With Buffer Unordered in Rust Explained

The loop "for each item, make an async call" is either too slow (one call at a time) or too dangerous (spawn them all). buffer_unordered(n) is the middle path in one line: it keeps up to n of the futures from a stream in flight, hands back each result as soon as it is ready, and does not pull a new input until a slot frees up. It needs no semaphore, no spawn and no 'static data. This post shows it on a realistic job (fetching 20 records with a limit of 4), the order differences between buffer_unordered and buffered, how errors and early exit behave, how to pick n, and what its single-task design costs. Everything compiled on Rust 1.99, edition 2024, with tokio 1 and futures 0.3.

The pipeline

Turn the inputs into a stream of futures with map, then collect them through buffer_unordered. A paused clock makes the timing exact: each call takes 100 ms and the limit is 4, so twenty calls should take five rounds, 500 ms:

use futures::stream::{self, StreamExt};
use tokio::time::{sleep, Duration, Instant};

async fn fetch(id: u32) -> (u32, String) {
    sleep(Duration::from_millis(100)).await;
    (id, format!("record-{id}"))
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let t = Instant::now();

    let results: Vec<(u32, String)> = stream::iter(1..=20)
        .map(fetch)
        .buffer_unordered(4)
        .collect()
        .await;

    println!("{} results in {:?}", results.len(), t.elapsed());
    println!("first four: {:?}", &results[..4].iter().map(|r| r.0).collect::<Vec<_>>());
}

20 results in 500ms
first four: [1, 2, 3, 4]

The twenty fetches took 500 ms: five waves of four. With a limit of one it would have taken 2 seconds, and with no limit 100 ms but twenty sockets at once. The stream is lazy, so inputs are pulled only as slots free up, which means a million-item input is processed with at most four futures alive at a time. Nothing is spawned: all four futures live inside the BufferUnordered adapter and are polled by whichever task is consuming it.

Ordered or not

With equal latencies the results look ordered, because they complete in the order they started. Give them unequal latencies and the difference appears:

use futures::stream::{self, StreamExt};
use tokio::time::{sleep, Duration, Instant};

async fn fetch(id: u32) -> u32 {
    sleep(Duration::from_millis(u64::from(50 * (5 - id)))).await;
    id
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let t = Instant::now();
    let unordered: Vec<u32> = stream::iter(1..=4).map(fetch).buffer_unordered(4).collect().await;
    println!("unordered {:?} in {:?}", unordered, t.elapsed());

    let t = Instant::now();
    let ordered: Vec<u32> = stream::iter(1..=4).map(fetch).buffered(4).collect().await;
    println!("buffered  {:?} in {:?}", ordered, t.elapsed());
}

unordered [4, 3, 2, 1] in 200ms
buffered  [1, 2, 3, 4] in 200ms

Task 4 finished first (50 ms) and task 1 last (200 ms). buffer_unordered yields in completion order; buffered yields in input order, so it holds finished results back until their turn. Total time is the same, 200 ms, because the work is the same; what changes is when you see each result and how many finished results are parked. Use buffered when downstream needs input order (writing output rows in the original sequence); use buffer_unordered when each result is self-describing, and carry the identifier in the output as the first example does. Unordered gives lower latency to the first result and uses less memory.

Errors and early exit

Items that are Results flow through like any other. To stop at the first error, use try_collect or a loop with try_next. Dropping the stream cancels everything still in flight, since the futures live inside it:

use futures::stream::{self, StreamExt, TryStreamExt};
use std::sync::atomic::{AtomicU32, Ordering};
use std::sync::Arc;
use tokio::time::{sleep, Duration};

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let started = Arc::new(AtomicU32::new(0));
    let s = started.clone();

    let outcome: Result<Vec<u32>, String> = stream::iter(1..=100u32)
        .map(move |id| {
            let s = s.clone();
            async move {
                s.fetch_add(1, Ordering::SeqCst);
                sleep(Duration::from_millis(10)).await;
                if id == 7 { Err(format!("record {id} is corrupt")) } else { Ok(id) }
            }
        })
        .buffer_unordered(4)
        .try_collect()
        .await;

    println!("{outcome:?}");
    println!("calls started before the error stopped it: {}", started.load(Ordering::SeqCst));
}

Err("record 7 is corrupt")
calls started before the error stopped it: 8

The first error ended the whole run. Only 8 of the hundred inputs were ever started: the first four, then the next four as slots freed up, and the error from record 7 arrived while the last of them were still in flight. The in-flight ones were dropped mid-await (cancelled), so each must be cancel safe, an issue explained in cancellation safety. If instead you want every result, success or failure, collect Vec<Result<_, _>> and sort them out afterwards. That is the right choice for a batch job where one corrupt record must not hide the others.

Be careful with the idea that the error "cancels" the rest in a way that undoes it. Record 8 had started and was dropped halfway through its sleep; if its call had been a POST that the server had already received, the server will complete it and you will never see the result. The same hazard as with timeouts applies: anything with an external effect should be safe to repeat, or should not be put in a pipeline that can be abandoned. The comparison with a pipeline that finishes every item and collects the failures is the choice between all-or-nothing and best-effort, and it is worth deciding explicitly instead of letting try_collect decide for you.

Picking n

The number is a statement about the downstream system, not about your CPU. Take the connection or rate limit of whatever the futures call, subtract headroom for other users of it, and use that. For an HTTP API that allows 50 concurrent requests per key, 20 to 40 is sensible. For a database pool of 10 connections shared with the web app, 3 to 5. Throughput is roughly n divided by the latency of one call, so going from 4 to 8 doubles it until the downstream system saturates, and beyond that latency per call rises instead. The same reasoning for semaphore limits is in limiting concurrency.

The cost: one task, no parallelism

All the futures inside the adapter run on the task that polls it. That is concurrency, which is all you need for waiting on I/O. It is not parallelism: if each future does CPU work between awaits, they take turns on one thread. For compute-heavy futures, spawn them (tokio::spawn inside the map, awaiting the handle) so the runtime can use several cores, at the price of the 'static bound:

use futures::stream::{self, StreamExt};

fn crunch(n: u64) -> u64 {
    (0..2_000_000u64).fold(n, |h, i| h.wrapping_mul(31).wrapping_add(i))
}

#[tokio::main]
async fn main() {
    let results: Vec<u64> = stream::iter(0..8u64)
        .map(|n| tokio::spawn(async move { crunch(n) }))
        .buffer_unordered(4)
        .map(|joined| joined.expect("worker did not panic"))
        .collect()
        .await;

    println!("computed {} values on up to 4 worker threads", results.len());
}

computed 8 values on up to 4 worker threads

Now the adapter bounds how many tasks exist, and each task can run on its own worker. The cost is a spawn per item, about 250 to 300 nanoseconds in the measurement in spawn and join, negligible for a call that takes milliseconds. The blocking-vs-async cut for CPU work is covered in blocking the Tokio runtime.

for_each_concurrent, and when not to bother

When you do not need results, only the effect, for_each_concurrent(n, |item| async { ... }) does the same job with a sink at the end; it also takes None for "no limit", which you almost never want. For a small, fixed number of things that run to completion together, join! or join_all is simpler. For work that must outlive the caller or be spread across threads, a JoinSet window gives it, as in JoinSet. And watch the usual trap: a slow consumer downstream of buffer_unordered slows the whole pipeline, because the adapter will not poll more futures than it can hold results for. That backpressure is a feature, and it is why this adapter belongs in pipelines where the final step is also a stream, discussed in async streams.