🎓 A1Learner
🔍

Rust Buffered Streams

A buffered stream is a stream that keeps a fixed number of item futures running at once. The plain stream loop runs one item at a time, so every item waits for the one before it.

The slow default

This is the plain stream loop. Correct — but the awaits queue up one behind another:

while let Some(page) = pages.next().await {
    fetch(page).await;
}

At most n at once

map makes one future per item. buffer_unordered(2) keeps two of them in flight. A third future starts only when one of the two finishes:

let results: Vec<String> = stream::iter(pages)
    .map(|(name, ms)| fetch(name, ms))
    .buffer_unordered(2)
    .collect()
    .await;

Same race, input order

buffered(2) runs the identical race, but hands the results out in the order the items arrived:

let results: Vec<String> = stream::iter(pages)
    .map(|(name, ms)| fetch(name, ms))
    .buffered(2)
    .collect()
    .await;

The queue you push by hand

FuturesOrdered takes futures one by one and returns their results in push order, even if a later one finishes first:

let mut queue = FuturesOrdered::new();
queue.push_back(future::ready(10));
queue.push_back(future::ready(20));
while let Some(n) = queue.next().await {
    println!("{n}");
}

Example

// Cargo.toml: tokio = { version = "1", features = ["full"] }, futures = "0.3"
use futures::stream::{self, StreamExt};
use tokio::time::{sleep, Duration};

async fn fetch(page: &str, ms: u64) -> String {
    sleep(Duration::from_millis(ms)).await;
    format!("{page} ({ms} ms)")
}

#[tokio::main]
async fn main() {
    let pages = vec![("alpha", 40), ("beta", 10), ("gamma", 20)];

    let unordered: Vec<String> = stream::iter(pages.clone())
        .map(|(name, ms)| fetch(name, ms))
        .buffer_unordered(2)
        .collect()
        .await;
    println!("unordered: {unordered:?}");

    let ordered: Vec<String> = stream::iter(pages)
        .map(|(name, ms)| fetch(name, ms))
        .buffered(2)
        .collect()
        .await;
    println!("ordered:   {ordered:?}");
}

Key points

  • One by one is the slow default: with next().await in a loop, item two does not start until item one is fully done
  • Use map to turn each item into a future.
  • Use buffer_unordered(n) to keep at most n of those futures running at once.
  • buffer_unordered hands results out in finish order: a fast item overtakes a slow one started earlier
  • Use buffered(n) instead when results must keep the input order.
  • It runs the same n-at-a-time race.
  • It hands results out in the order the items arrived.
  • FuturesOrdered is the same idea as a queue you push into by hand.
  • The futures race inside.
  • The results come out in push order.
  • The number n is a budget.
  • It caps work in flight.
  • It also caps memory.
  • It caps how hard you lean on the other side, too.
  • A rate limit, a database, and a file server are places where that cap matters.
  • The buffered futures run inside your own task.
  • Nothing is spawned.
  • These futures do not need to be Send.
  • Spawning a task for each item has no limit.

In this example, fetch pretends to download a page: it sleeps, then reports the page name and its wait. Three pages go in — alpha needs 40 ms, beta 10, gamma 20 — and the limit is two at a time. In the unordered run, alpha and beta start together. Beta finishes first, so gamma starts. Gamma finishes next, and slow alpha comes last. That is finish order. The buffered run starts the same race, but collects in input order. Alpha comes first, even though it finished last. Beta and gamma may not overtake it.

Note

FuturesUnordered, from the Joining Futures lesson, is the free-standing bag version of the same race.

Try it Yourself

Edit the code, then press Run.

Running is simulated · Rust Playground

Exercise

You run buffer_unordered(4) over a stream of 100 downloads. How many downloads can be in flight at the same moment?