Rust 84 🦀 Streams API for Asynchronous Data Sequences (FuturesUnordered)
Rust’s Stream trait is the asynchronous counterpart of Iterator: where an iterator produces values synchronously on demand, a stream produces values over time, yielding Poll::Pending when no value is ready and waking the task when one arrives. This abstraction covers channels, network data, file chunks, and any source that emits items asynchronously. But streams alone are passive. The runtime needs a structure that can hold many futures, poll each one only when it is woken, and yield results as they complete—in any order. That structure is FuturesUnordered, a set of futures that implements Stream by yielding each future’s output the moment it resolves, regardless of insertion order .
The distinction matters because Rust’s async model lacks a built-in select over dynamic collections. join! requires a fixed number of futures known at compile time. join_all waits for every future before yielding anything. FuturesUnordered fills the gap: it manages an unbounded set, polls only futures that have been woken, and delivers each result as soon as it is ready . This makes it the correct tool for “race these operations and process results as they arrive”—the pattern behind connection pools, fan-out requests, and timeout coordination.
This chapter covers the Stream trait’s poll_next contract, the StreamExt extension methods that make streams ergonomic, FuturesUnordered‘s internal wake-up optimization, the push-then-poll_next lifecycle, and the pitfalls around polling new futures and dropping completed ones. It assumes familiarity with the Future trait, the poll/waker contract, and Tokio’s runtime from the preceding chapters.
Key point: FuturesUnordered is a stream of futures that yields each future’s output as it completes. It polls only woken futures, making it efficient for large sets, but a newly pushed future is not polled until poll_next is called again—forgetting this stalls the future silently.
Why FuturesUnordered exists
join_all waits for the slowest. The join_all combinator collects a Vec of futures and resolves only when every one has completed, returning a Vec of results in the original order . This is correct when you need all results together, but wrong when you want to process each result as it becomes available. A set of HTTP requests where one takes thirty seconds should not delay processing the other nineteen that finish in milliseconds. FuturesUnordered yields each result independently, allowing the caller to act on fast completions while slow ones continue .
Static combinators cannot handle dynamic sets. tokio::select! and futures::select! take a fixed number of futures written literally in the macro. A connection pool that acquires new futures at runtime cannot use select!. FuturesUnordered accepts futures via push at any time, making it the only standard structure for dynamic concurrent sets .
Waker-based polling is the efficiency mechanism. The documentation is explicit: “Futures managed by FuturesUnordered will only be polled when they generate wake-up notifications” . A naive implementation might poll every future on every wake, producing O(n²) work. FuturesUnordered maintains internal per-future wakers that re-enqueue a future only when that specific future is ready, reducing the cost of managing thousands of pending operations .
The stream abstraction unifies sources. FuturesUnordered implements Stream, so it composes with the same StreamExt methods as any other stream: map, filter, take, for_each_concurrent. This means a set of futures can be treated as a single asynchronous sequence, transformed, and consumed by the same code that handles a channel or a file chunk stream .
The trade-off. FuturesUnordered does not preserve order. Results arrive in completion order, not insertion order. It also requires careful lifecycle management: pushing a future does not poll it, and the caller must call poll_next (or use next().await) to drive the set forward. For a small, fixed set, join_all is simpler. For a large or dynamic set where completion order matters, FuturesUnordered is the correct choice.
a. The Stream trait and the poll_next contract
The Stream trait defines one required method, poll_next, which returns Poll::Ready(Some(item)) when a value is available, Poll::Pending when the stream is not ready but has registered a waker, and Poll::Ready(None) when the stream is exhausted . This mirrors Future::poll‘s Poll return type but with the Option wrapper distinguishing “not yet” from “finished” .
use std::pin::Pin;
use std::task::{Context, Poll};
pub trait Stream {
type Item;
fn poll_next(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Option<Self::Item>>;
}The StreamExt extension trait provides next(), which returns a future resolving to Option<Item> . This is what enables while let Some(item) = stream.next().await. The trait must be imported for .next() to be in scope; the compiler error “no method named next” is solved by use futures::StreamExt; or use tokio_stream::StreamExt; depending on the crate .
use futures::stream::{self, StreamExt};
#[tokio::main]
async fn main() {
// Create a stream from an iterator.
let mut stream = stream::iter(vec![1, 2, 3]);
// StreamExt::next() drives the stream.
while let Some(value) = stream.next().await {
println!("received: {value}");
}
}Streams are lazy: creating one does nothing until poll_next is called. A stream constructed solely for its side effects and never polled produces no side effects, and the compiler warns about unused streams . This laziness is the same property that makes futures inert until polled, extended over a sequence.
b. FuturesUnordered: construction, push, and the polling obligation
FuturesUnordered<Fut> is a set of futures. It can be constructed empty with new(), or filled by collecting an iterator of futures: futures.into_iter().collect::<FuturesUnordered<_>>() . Futures are added with push(&self, future: Fut), which notably takes &self rather than &mut self—the set uses interior mutability to allow pushing while it is being polled, a requirement for patterns where a task’s completion triggers spawning a replacement .
The critical lifecycle detail: push does not poll the future. The documentation states: “When new futures are added, poll_next must be called in order to begin receiving wake-ups for new futures” . A future pushed after the last poll_next has no waker registered with the set’s internals, so its completion will not wake the outer task. The caller must ensure poll_next is called after pushing, typically by continuing the next().await loop.
use futures::stream::{FuturesUnordered, StreamExt};
use futures::future;
use std::time::Duration;
#[tokio::main]
async fn main() {
let mut set = FuturesUnordered::new();
// Push futures with different completion times.
set.push(tokio::time::sleep(Duration::from_millis(100)));
set.push(tokio::time::sleep(Duration::from_millis(50)));
// poll_next drives the set. Results arrive in completion order.
while set.next().await.is_some() {
println!("one future completed");
}
}In this example, the 50ms future completes first and is yielded first, regardless of push order. If a third future were pushed after the first completion, the while set.next().await loop would continue polling, registering a waker for the new future and eventually yielding it .
The len() and is_empty() methods report the number of in-flight futures. A FuturesUnordered that has yielded all results and has no pending futures returns Poll::Ready(None) from poll_next, terminating the stream . Futures can be inspected via iter() and iter_mut(), though polling requires Pin for non-Unpin futures via iter_pin_mut() .
c. Combining FuturesUnordered with stream adapters
Because FuturesUnordered implements Stream, it composes with StreamExt adapters. The most common pattern for concurrent processing is for_each_concurrent, which limits the number of in-flight futures while driving the set to completion.
use futures::stream::{FuturesUnordered, StreamExt};
use futures::future;
#[tokio::main]
async fn main() {
let urls = vec!["a.com", "b.com", "c.com"];
// Build a FuturesUnordered of fetch operations.
let futures: FuturesUnordered<_> = urls
.into_iter()
.map(|url| async move {
// Simulate a fetch.
tokio::time::sleep(std::time::Duration::from_millis(10)).await;
url
})
.collect();
// Process results as they complete.
futures
.for_each(|result| async move {
println!("fetched: {result}");
})
.await;
}The collect() method on an iterator of futures produces a FuturesUnordered directly, avoiding manual push calls . This is the idiomatic construction when the full set is known upfront. For dynamic sets, push on a mutable binding remains the pattern.
A subtlety when using FuturesUnordered with tokio::spawn: the futures inside the set must be Send + 'static if the entire set is moved into a spawned task. If the set stays on the current task, the futures need only satisfy the current task’s bounds. The current-thread runtime and spawn_local relax Send requirements, allowing FuturesUnordered to hold non-Send futures .
Complete Example Session
// ============================================
// PART 1: THE STREAM TRAIT CONTRACT
// ============================================
use std::pin::Pin;
use std::task::{Context, Poll};
// poll_next returns:
// Ready(Some(item)) when a value is available
// Pending when not ready but waker registered
// Ready(None) when exhausted
// This mirrors Future::poll but for sequences.
// ============================================
// PART 2: STREAMEXT AND NEXT()
// ============================================
use futures::stream::{self, StreamExt};
async fn basic_stream() {
let mut s = stream::iter(vec![10, 20, 30]);
while let Some(v) = s.next().await {
println!("value: {v}");
}
}
// ============================================
// PART 3: FUTURESUNORDERED CONSTRUCTION
// ============================================
use futures::stream::FuturesUnordered;
use std::time::Duration;
async fn unordered_basic() {
let mut set = FuturesUnordered::new();
// push does not poll. poll_next must be called.
set.push(tokio::time::sleep(Duration::from_millis(100)));
set.push(tokio::time::sleep(Duration::from_millis(30)));
// Yield in completion order: 30ms first, then 100ms.
while set.next().await.is_some() {
println!("completion");
}
}
// ============================================
// PART 4: COLLECTING FUTURES INTO A SET
// ============================================
async fn collect_set() {
let set: FuturesUnordered<_> = (0..5)
.map(|i| async move {
tokio::time::sleep(Duration::from_millis(10 * i)).await;
i
})
.collect();
// Results arrive in completion order, not index order.
let results: Vec<_> = set.collect().await;
println!("order: {:?}", results);
}
// ============================================
// PART 5: THE PUSH-AFTER-POLL HAZARD
// ============================================
async fn push_hazard() {
let mut set = FuturesUnordered::new();
set.push(futures::future::ready(1));
// First next() polls the ready future.
assert_eq!(set.next().await, Some(1));
// Push a new future AFTER poll_next returned Ready.
set.push(futures::future::ready(2));
// The new future is not polled until next() is called again.
// If the loop had exited here, the future would be dropped unpolled.
assert_eq!(set.next().await, Some(2));
}
// ============================================
// PART 6: DYNAMIC PUSHING DURING ITERATION
// ============================================
async fn dynamic_push() {
let mut set = FuturesUnordered::new();
set.push(futures::future::ready(1));
while let Some(value) = set.next().await {
println!("got {value}");
if value < 3 {
// Push a replacement while iterating.
// next() will poll it on the following iteration.
set.push(futures::future::ready(value + 1));
}
}
}
// ============================================
// PART 7: FOR_EACH_CONCURRENT WITH LIMIT
// ============================================
async fn limited_concurrency() {
let urls: Vec<u32> = (0..10).collect();
// Each future simulates a request.
let fetches: FuturesUnordered<_> = urls
.into_iter()
.map(|id| async move {
tokio::time::sleep(Duration::from_millis(10)).await;
id
})
.collect();
// Process results as they complete.
fetches
.for_each(|id| async move {
println!("processed {id}");
})
.await;
}
// ============================================
// PART 8: LEN AND EMPTINESS
// ============================================
async fn inspect_set() {
let mut set = FuturesUnordered::new();
assert!(set.is_empty());
assert_eq!(set.len(), 0);
set.push(futures::future::ready(()));
assert_eq!(set.len(), 1);
set.next().await;
assert!(set.is_empty());
}
// ============================================
// PART 9: STREAM COMPOSITION
// ============================================
async fn composed() {
let set: FuturesUnordered<_> = (0..6)
.map(|i| futures::future::ready(i))
.collect();
// map, filter, take from StreamExt.
let results: Vec<_> = set
.map(|i| i * 2)
.filter(|i| futures::future::ready(i % 3 == 0))
.take(2)
.collect()
.await;
println!("{:?}", results);
}
// ============================================
// PART 10: THE COMPLETE PATTERN
// ============================================
// 1. Build FuturesUnordered (collect or push).
// 2. Drive with next().await or for_each().
// 3. Handle each result as it arrives.
// 4. Push new futures during iteration if needed.
// 5. Continue until next() returns None.
// Remember: push does not poll; next() does.This session moves from the Stream trait’s poll_next contract through StreamExt::next(), FuturesUnordered construction and collection, the push-after-poll hazard, dynamic pushing during iteration, limited-concurrency processing, introspection methods, and stream adapter composition, ending with the canonical pattern: build, drive, handle results as they arrive, push new futures when needed, continue until None.
Quick Reference
Stream Trait and Extension
| Item | Purpose |
|---|---|
Stream::poll_next | Core polling method |
StreamExt::next | Async Option<Item> future |
StreamExt::map/filter/take | Iterator-like adapters |
StreamExt::for_each | Drive stream to completion |
FuturesUnordered Methods
| Method | Behavior |
|---|---|
new() | Empty set |
push(fut) | Add future; does not poll |
poll_next | Poll woken futures; yields completions |
len() | In-flight count |
is_empty() | No pending futures |
iter() / iter_mut() | Inspect futures |
Completion Order
| Scenario | Order |
|---|---|
join_all | Original insertion order |
FuturesUnordered | Completion order |
select! | First completed among fixed set |
Common Patterns
| Pattern | Construction |
|---|---|
| Known set | iterator.map(...).collect() |
| Dynamic set | set.push(fut) during loop |
| Limited concurrency | for_each_concurrent(n, f) |
| Process as completed | while let Some(x) = set.next().await |
Best Practices
✅ Do This:
// Collect a known set of futures into FuturesUnordered. // ✅
let set: FuturesUnordered<_> = futures.into_iter().collect();
// Call next() after pushing to poll new futures. // ✅
set.push(f); while let Some(v) = set.next().await { ... }
// Use for_each to process completions. // ✅
set.for_each(|r| async move { handle(r); }).await;
// Push replacements during iteration for dynamic sets. // ✅
if need_more { set.push(make_future()); }❌ Don’t Do This:
// Push and assume the future is polled. // ❌
set.push(f); // f is inert until next().
// Use join_all when completion order matters. // ❌
join_all(futures).await; // Waits for slowest.
// Collect a FuturesUnordered without driving it. // ❌
let set = ...; // Nothing runs until polled.Common Pitfalls
| Pitfall | Why It Happens | Fix |
|---|---|---|
| Future pushed but never polled | push does not poll | Call next() after push |
| Order assumed as insertion order | FuturesUnordered yields by completion | Use join_all for order |
| Set created but never driven | Streams are lazy | next().await or for_each |
Send bound error on spawn | Set moved into multi-thread task | Use current-thread or spawn_local |
len() stale after last poll | Futures removed on completion | Check is_empty() in loop condition |
Real-World Examples
1. Racing HTTP Requests
let set: FuturesUnordered<_> = urls.iter().map(|u| fetch(u)).collect();
while let Some(result) = set.next().await { handle(result); }2. Connection Pool Acquisition
let mut pool = FuturesUnordered::new();
for _ in 0..10 { pool.push(acquire_connection()); }
let first = pool.next().await; // Fastest connection.3. Dynamic Task Spawning
while let Some(job) = job_queue.next().await {
set.push(process(job));
if let Some(result) = set.next().await { handle(result); }
}4. Timeout with Fastest Completion
let mut racers = FuturesUnordered::new();
racers.push(fetch_from_mirror("a"));
racers.push(fetch_from_mirror("b"));
let winner = racers.next().await;5. Bounded Concurrency
let set: FuturesUnordered<_> = items.map(process).collect();
set.for_each_concurrent(5, |r| async move { handle(r); }).await;6. Collect All Results
let results: Vec<_> = set.collect().await;7. Inspect Pending Futures
for fut in set.iter() { /* inspect without polling */ }8. Check Set Emptiness
while !set.is_empty() { set.next().await; }9. Push During Iteration
while let Some(v) = set.next().await {
if v.needs_retry { set.push(retry(v)); }
}10. Filter Completions
set.filter(|r| futures::future::ready(r.is_ok()))
.for_each(|r| async move { handle(r.unwrap()); })
.await;Visual
Stream vs FuturesUnordered
┌──────────────────────────────────────────────┐
│ STREAM FUTURESUNORDERED │
│ │
│ Produces items Holds futures. │
│ over time. Yields results │
│ as they complete. │
│ poll_next → poll_next → │
│ Some(item) Some(result) │
│ Pending Pending │
│ None None │
│ │
│ Source: channel, Source: set of │
│ file, network. futures. │
│ │
│ FuturesUnordered IS a Stream. │
│ It composes with StreamExt adapters. │
│ │
└──────────────────────────────────────────────┘Completion Order
┌──────────────────────────────────────────────┐
│ PUSH ORDER vs COMPLETION ORDER │
│ │
│ Push: [A (100ms), B (30ms), C (60ms)] │
│ │
│ poll_next loop: │
│ 1st yield: B (30ms) │
│ 2nd yield: C (60ms) │
│ 3rd yield: A (100ms) │
│ │
│ The set yields whichever future completes │
│ first. Order is temporal, not positional. │
│ │
│ join_all would yield [A, B, C] after 100ms. │
│ FuturesUnordered yields B immediately. │
│ │
└──────────────────────────────────────────────┘Push-Poll Lifecycle
┌──────────────────────────────────────────────┐
│ PUSH DOES NOT POLL │
│ │
│ set.push(f1) → f1 in set, NOT polled │
│ set.push(f2) → f2 in set, NOT polled │
│ │
│ set.next().await │
│ │ │
│ ▼ │
│ poll_next polls f1 and f2 │
│ Registers wakers for both. │
│ │ │
│ ▼ │
│ f2 completes → waker fires → poll_next │
│ yields f2's output │
│ │ │
│ ▼ │
│ f1 still pending. waker registered. │
│ │ │
│ ▼ │
│ set.next().await again │
│ Waits for f1 (or a newly pushed future). │
│ │
│ Push after last poll without calling next() │
│ → new future never polled, never completes. │
│ │
└──────────────────────────────────────────────┘Waker Optimization
┌──────────────────────────────────────────────┐
│ PER-FUTURE WAKERS │
│ │
│ Naive: every wake → poll all N futures. │
│ O(N) work per wake. │
│ │
│ FuturesUnordered: │
│ Each future has its own waker. │
│ Only the woken future is re-enqueued. │
│ O(1) work per wake. │
│ │
│ Result: scales to thousands of futures │
│ without quadratic polling cost. │
│ │
│ This is why the docs say it is "optimized │
│ to manage a large number of futures." │
│ │
└──────────────────────────────────────────────┘Summary
| Item | Value |
|---|---|
Stream trait | Async iteration via poll_next |
poll_next returns | Ready(Some), Pending, Ready(None) |
StreamExt | Extension methods: next, map, filter, for_each |
FuturesUnordered | Set of futures implementing Stream |
| Yield order | Completion order, not insertion order |
push | Adds future; does not poll it |
| Polling obligation | poll_next must be called after push |
| Waker optimization | Only woken futures are polled |
| Construction | collect() from iterator or new() + push |
join_all difference | Waits for all, preserves order |
Key takeaways:
Streamis asynchronous iteration.poll_nextreturnsSome(item)when a value is ready,Pendingwhen not, andNonewhen exhausted.StreamExt::next()wraps this in a future .FuturesUnorderedyields results as they complete. It does not preserve insertion order. Use it when completion order matters or when the set is dynamic .pushdoes not poll. A newly pushed future is inert untilpoll_nextis called again. Thenext().awaitloop provides this polling; apushafter the loop exits leaves the future unpolled .- The waker optimization makes large sets efficient. Each future has its own waker; only woken futures are re-polled, avoiding O(N²) scanning .
collect()constructs a set from a known iterator. For dynamic sets, usenew()andpushduring iteration.StreamExtadapters compose.FuturesUnorderedis a stream, somap,filter,take, andfor_each_concurrentwork on it exactly as on any other stream .join_allandFuturesUnorderedsolve different problems.join_allwaits for every future and preserves order;FuturesUnorderedyields each result the moment it is ready .
Remember: FuturesUnordered is the bridge between a collection of futures and a stream of results. It holds each future, registers a per-future waker, and yields outputs in completion order. The stream trait makes it composable with the same adapters used for channels and files. The one obligation the caller bears is driving the set: push adds a future but does not poll it, and only poll_next—typically through next().await—activates the wakers. For racing requests, pooling connections, or processing a dynamic set of concurrent operations as they finish, FuturesUnordered is the standard tool.
Stop using slow, ad-bloated tool sites! 🤮
🔎 Search “KandZ Tools” on Google to use many professional utilities for free.
KandZ.me is the ultimate minimalist hub for:
✅ Finance (Mortgage, Interest, Inflation)
✅ Tech (Base64, JSON, Dev Suite, IP)
✅ Health (BMI, BMR, TDEE)
✅ Productivity (Timer, Workspace, QR)
⚡️ Fast & Private
🔒 No data leaves your device
💎 100% Free
🔗 Use it now: https://tools.kandz.me
🔖 Bookmark it—you’ll need it later!