Writing Futures and Streams by Hand

An async fn is compiled into a type that implements Future. This lesson writes such types by hand: plain futures, state machines, combinators that wrap other futures, streams, and an AsyncRead wrapper. After it you can read the source of Tokio and tokio-util, and write the small adapters that async blocks cannot express.

Item	What it is	Example
`impl Future for T`	a type whose `poll` returns `Poll::Ready` or `Poll::Pending`	1, 2
waker contract	`Pending` must be returned only after a wake-up is arranged	2, 3
hand-written state machine	an enum of states advanced inside `poll`	4
shared-state future	a value and a waker behind a lock, filled by another thread	5, 6
`std::future::poll_fn`	a future built from a closure	7
`std::task::ready!`	returns `Poll::Pending` early, unwraps `Poll::Ready`	8
`pin-project-lite`	safe `Pin<&mut Field>` access for fields of a pinned struct	9, 10
combinators	map, timeout, join of two, retry	8 to 13
`impl Stream for T`	`poll_next` returning `Poll<Option<Item>>`	14, 15, 16
`impl AsyncRead for T`	`poll_read` filling a `ReadBuf`	17, 18

The examples use pin-project-lite and tokio-stream next to Tokio:

TOMLCargo.toml

[dependencies]
tokio = { version = "1", features = ["full", "test-util"] }
tokio-stream = "0.1"
pin-project-lite = "0.2"

The test-util feature enables the paused clock (start_paused), introduced in Time: sleep, timeout, and interval. With a paused clock, Tokio moves time forward only when every task is waiting, so printed durations are exact.

The contract of `poll`

Futures and async/await and Polling, Wakers, and a Tiny Executor introduced the trait:

Rust

pub trait Future {
    type Output;
    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output>;
}

Every hand-written poll follows three rules:

Return Poll::Ready(value) exactly once. Polling again after Ready is a caller bug. The future may panic or return anything.
Before returning Poll::Pending, arrange a wake-up. Either an inner future returned Pending (it has stored the waker), or this future stored cx.waker() somewhere that will call wake. Otherwise nothing polls the task again.
Do not block. poll runs on a worker thread; a blocking call stalls every task on that thread, as shown in Blocking and CPU-Bound Work.

Example 1: A future that is ready at once

The smallest useful future holds a value and returns it on the first poll. The value is in an Option so poll can move it out through &mut self.

Rust

use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};

struct Greeting {
    message: Option<String>,
}

impl Future for Greeting {
    type Output = String;

    fn poll(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<String> {
        let message = self.message.take().expect("Greeting polled after completion");
        Poll::Ready(message)
    }
}

#[tokio::main]
async fn main() {
    let greeting = Greeting { message: Some("hello from a hand-written future".to_string()) };
    let text = greeting.await;
    println!("{text}");
}

Output

hello from a hand-written future

self has type Pin<&mut Greeting>. Pin gives &mut access to the fields only when the type is Unpin. Greeting contains only an Option<String>, so the compiler marks it Unpin automatically and self.message.take() compiles. .await calls poll for you; Example 2 shows what happens when poll returns Pending.

Example 2: Returning Pending and waking yourself

A future that is not ready returns Poll::Pending. This one counts down, and before each Pending it calls cx.waker().wake_by_ref(), which tells the runtime to poll the task again.

Rust

use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};

struct CountDown {
    remaining: u32,
    polls: u32,
}

impl Future for CountDown {
    type Output = u32;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<u32> {
        self.polls += 1;
        if self.remaining == 0 {
            return Poll::Ready(self.polls);
        }
        println!("poll {}: {} left, returning Pending", self.polls, self.remaining);
        self.remaining -= 1;
        cx.waker().wake_by_ref();
        Poll::Pending
    }
}

#[tokio::main]
async fn main() {
    let polls = CountDown { remaining: 3, polls: 0 }.await;
    println!("ready after {polls} polls");
}

Output

poll 1: 3 left, returning Pending
poll 2: 2 left, returning Pending
poll 3: 1 left, returning Pending
ready after 4 polls

Waking yourself right away is how a future gives other tasks a turn without waiting for any event. tokio::task::yield_now works the same way. wake_by_ref borrows the waker; wake consumes it; clone keeps a copy for later.

Example 3: Forgetting to wake

This future returns Pending on the first poll without storing or calling the waker. Nothing will poll it again on its own. It is wrapped in tokio::time::timeout so the program ends; the paused clock makes the five seconds pass instantly.

Rust

use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
use std::time::Duration;
use tokio::time::{Instant, timeout};

struct NeverWakes {
    polls: u32,
}

impl Future for NeverWakes {
    type Output = u32;

    fn poll(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<u32> {
        self.polls += 1;
        if self.polls == 1 {
            Poll::Pending // bug: no waker stored, no wake arranged
        } else {
            Poll::Ready(self.polls)
        }
    }
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let start = Instant::now();
    let result = timeout(Duration::from_secs(5), NeverWakes { polls: 0 }).await;
    println!("result: {result:?}");
    println!("elapsed: {:?}", start.elapsed());
}

Output

result: Ok(2)
elapsed: 5s

The inner future was ready on its second poll, but that poll happened only when the timeout's timer woke the task five seconds later. Timeout polls the inner future first, so it returned Ok(2) instead of Elapsed. Without the timeout, the task would wait forever. A missing wake-up is a silent hang, not an error.

Example 4: A state machine by hand

An async block with two .await points compiles into an enum with one variant per suspension point. This example writes that enum by hand. Each state owns the timer it waits on. Sleep is not Unpin, so each timer is stored as Pin<Box<Sleep>>, which is Unpin and keeps the whole enum Unpin.

Rust

use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
use std::time::Duration;
use tokio::time::{Instant, Sleep, sleep};

enum State {
    Boiling(Pin<Box<Sleep>>),
    Steeping(Pin<Box<Sleep>>),
    Done,
}

struct MakeTea {
    start: Instant,
    state: State,
}

impl MakeTea {
    fn new() -> Self {
        MakeTea {
            start: Instant::now(),
            state: State::Boiling(Box::pin(sleep(Duration::from_millis(200)))),
        }
    }
}

impl Future for MakeTea {
    type Output = &'static str;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<&'static str> {
        loop {
            let elapsed = self.start.elapsed();
            match &mut self.state {
                State::Boiling(timer) => {
                    if timer.as_mut().poll(cx).is_pending() {
                        return Poll::Pending;
                    }
                    println!("{elapsed:?}: water boiled");
                    self.state = State::Steeping(Box::pin(sleep(Duration::from_millis(300))));
                }
                State::Steeping(timer) => {
                    if timer.as_mut().poll(cx).is_pending() {
                        return Poll::Pending;
                    }
                    println!("{elapsed:?}: tea steeped");
                    self.state = State::Done;
                    return Poll::Ready("tea is ready");
                }
                State::Done => panic!("MakeTea polled after completion"),
            }
        }
    }
}

// The same logic as an async fn, for comparison.
async fn make_tea() -> &'static str {
    sleep(Duration::from_millis(200)).await;
    sleep(Duration::from_millis(300)).await;
    "tea is ready"
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let start = Instant::now();
    println!("{}", MakeTea::new().await);
    println!("hand-written: {:?}", start.elapsed());

    let start = Instant::now();
    println!("{}", make_tea().await);
    println!("async fn: {:?}", start.elapsed());
}

Output

200ms: water boiled
500ms: tea steeped
tea is ready
hand-written: 500ms
tea is ready
async fn: 500ms

Three details matter in every hand-written state machine:

The loop. After a state finishes, the next state is polled in the same call. Returning Pending after a transition without polling the new timer would break rule 2: no waker would be registered with the new timer.
Pending only comes from an inner Pending. The code returns Pending only when the timer returned Pending, so the timer holds the waker.
A Done state. It replaces the finished state, drops the old timer, and turns a second poll into a clear panic.

Example 5: A value filled by another thread

Many futures wait for an event that happens outside the runtime. The pattern is a shared slot behind a lock: the future stores its waker there, and the producer stores the value and calls wake. This example builds a one-shot channel that way.

Rust

use std::future::Future;
use std::pin::Pin;
use std::sync::{Arc, Mutex};
use std::task::{Context, Poll, Waker};
use std::time::Duration;

struct Shared<T> {
    value: Option<T>,
    waker: Option<Waker>,
}

struct Sender<T> {
    shared: Arc<Mutex<Shared<T>>>,
}

struct Receiver<T> {
    shared: Arc<Mutex<Shared<T>>>,
}

fn channel<T>() -> (Sender<T>, Receiver<T>) {
    let shared = Arc::new(Mutex::new(Shared { value: None, waker: None }));
    (Sender { shared: shared.clone() }, Receiver { shared })
}

impl<T> Sender<T> {
    fn send(self, value: T) {
        let waker = {
            let mut shared = self.shared.lock().unwrap();
            shared.value = Some(value); // 1. change the state
            shared.waker.take()         // 2. take the waker
        }; // 3. release the lock
        if let Some(waker) = waker {
            waker.wake(); // 4. wake after the state change is visible
        }
    }
}

impl<T> Future for Receiver<T> {
    type Output = T;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<T> {
        let mut shared = self.shared.lock().unwrap();
        if let Some(value) = shared.value.take() {
            return Poll::Ready(value);
        }
        // Store the waker of the latest poll, cloning only when it changed.
        match &shared.waker {
            Some(stored) if stored.will_wake(cx.waker()) => {}
            _ => shared.waker = Some(cx.waker().clone()),
        }
        Poll::Pending
    }
}

#[tokio::main]
async fn main() {
    let (tx, rx) = channel();
    std::thread::spawn(move || {
        std::thread::sleep(Duration::from_millis(50));
        tx.send("computed on a std thread");
    });
    println!("{}", rx.await);
}

Output

computed on a std thread

The order in send is the waker discipline:

Rule	Reason
Store the value before calling `wake`	a woken task may be polled on another thread at once; it must find the value
Check the value and store the waker under the same lock	otherwise `send` can run between the check and the store, and the wake-up is lost
Call `wake` after releasing the lock	the woken task locks the same mutex; waking inside the lock makes it wait
Store the waker from the latest poll	a future can move to another task between polls (Example 6)
Compare with `will_wake` before cloning	cloning a Tokio waker increments the task's reference count atomically; most polls come from the same task

Example 6: Why the latest waker matters

A future can be polled by one task and later moved into another, for example when it is first raced in select! by reference and then awaited somewhere else. Each task has a different waker. This example polls a receiver by hand with two counting wakers, built with the Wake trait from Polling, Wakers, and a Tiny Executor, and compares a receiver that keeps only the first waker with one that keeps the latest.

Rust

use std::future::Future;
use std::pin::Pin;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use std::task::{Context, Poll, Wake, Waker};

struct CountingWaker {
    wakes: AtomicUsize,
}

impl Wake for CountingWaker {
    fn wake(self: Arc<Self>) {
        self.wakes.fetch_add(1, Ordering::SeqCst);
    }
}

struct Slot {
    value: Option<u32>,
    waker: Option<Waker>,
}

struct Receiver {
    slot: Arc<Mutex<Slot>>,
    keep_first_waker: bool,
}

impl Future for Receiver {
    type Output = u32;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<u32> {
        let mut slot = self.slot.lock().unwrap();
        if let Some(v) = slot.value.take() {
            return Poll::Ready(v);
        }
        if self.keep_first_waker {
            if slot.waker.is_none() {
                slot.waker = Some(cx.waker().clone()); // bug: later wakers ignored
            }
        } else if !slot.waker.as_ref().is_some_and(|w| w.will_wake(cx.waker())) {
            slot.waker = Some(cx.waker().clone());
        }
        Poll::Pending
    }
}

fn run(keep_first_waker: bool) {
    let slot = Arc::new(Mutex::new(Slot { value: None, waker: None }));
    let mut rx = Receiver { slot: slot.clone(), keep_first_waker };

    let task_a = Arc::new(CountingWaker { wakes: AtomicUsize::new(0) });
    let task_b = Arc::new(CountingWaker { wakes: AtomicUsize::new(0) });
    let waker_a = Waker::from(task_a.clone());
    let waker_b = Waker::from(task_b.clone());

    // Task A polls first, then the future moves to task B.
    let _ = Pin::new(&mut rx).poll(&mut Context::from_waker(&waker_a));
    let _ = Pin::new(&mut rx).poll(&mut Context::from_waker(&waker_b));

    // The producer sends and wakes whatever waker is stored.
    let waker = {
        let mut s = slot.lock().unwrap();
        s.value = Some(7);
        s.waker.take()
    };
    if let Some(w) = waker {
        w.wake();
    }

    println!(
        "keep_first_waker={keep_first_waker}: task A woken {}, task B woken {}",
        task_a.wakes.load(Ordering::SeqCst),
        task_b.wakes.load(Ordering::SeqCst),
    );
}

fn main() {
    run(true);
    run(false);
}

Output

keep_first_waker=true: task A woken 1, task B woken 0
keep_first_waker=false: task A woken 0, task B woken 1

With the bug, task A is woken but no longer owns the future, and task B, which does, sleeps forever. AtomicWaker in the futures crate packages the "keep the latest waker" rule for code that registers from one place and wakes from another.

Example 7: A future from a closure with `poll_fn`

std::future::poll_fn turns a closure FnMut(&mut Context) -> Poll<T> into a future. It avoids a new struct when the state already lives in local variables. Many Tokio types expose poll_* methods for this use; mpsc::Receiver::poll_recv is one. This example receives from two channels and always prefers the high-priority one.

Rust

use std::future::poll_fn;
use std::task::Poll;
use tokio::sync::mpsc;

async fn next_message(
    high: &mut mpsc::Receiver<&'static str>,
    low: &mut mpsc::Receiver<&'static str>,
) -> Option<&'static str> {
    poll_fn(|cx| {
        let high_state = high.poll_recv(cx);
        if let Poll::Ready(Some(msg)) = high_state {
            return Poll::Ready(Some(msg));
        }
        let low_state = low.poll_recv(cx);
        match (high_state, low_state) {
            (_, Poll::Ready(Some(msg))) => Poll::Ready(Some(msg)),
            (Poll::Ready(None), Poll::Ready(None)) => Poll::Ready(None), // both closed
            _ => Poll::Pending, // at least one returned Pending and stored the waker
        }
    })
    .await
}

#[tokio::main]
async fn main() {
    let (high_tx, mut high_rx) = mpsc::channel(8);
    let (low_tx, mut low_rx) = mpsc::channel(8);

    low_tx.send("low 1").await.unwrap();
    low_tx.send("low 2").await.unwrap();
    high_tx.send("high 1").await.unwrap();
    high_tx.send("high 2").await.unwrap();
    drop(high_tx);
    drop(low_tx);

    while let Some(msg) = next_message(&mut high_rx, &mut low_rx).await {
        println!("{msg}");
    }
    println!("both channels closed");
}

Output

high 1
high 2
low 1
low 2
both channels closed

The closure returns Pending only when at least one poll_recv returned Pending, so a waker is registered with that channel. Returning Pending when both channels are closed would hang. select! with biased; from select!: Racing Futures expresses the same priority; poll_fn is the lower-level form that select! is built on.

Example 8: `ready!` and a map combinator that needs `Unpin`

A combinator is a future that wraps another future and changes its behavior. std::task::ready!(expr) evaluates a Poll<T>: on Pending it returns Poll::Pending from the enclosing function, and on Ready(v) it evaluates to v. This first version of Map polls its inner future through Pin::new, which requires the inner future to be Unpin.

Rust

use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll, ready};

struct Map<Fut, F> {
    fut: Fut,
    f: Option<F>,
}

impl<Fut, F, T> Future for Map<Fut, F>
where
    Fut: Future + Unpin,
    F: FnOnce(Fut::Output) -> T + Unpin,
{
    type Output = T;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<T> {
        let output = ready!(Pin::new(&mut self.fut).poll(cx));
        let f = self.f.take().expect("Map polled after completion");
        Poll::Ready(f(output))
    }
}

#[tokio::main]
async fn main() {
    // A boxed future is Unpin, so it satisfies the bound.
    let boxed = Box::pin(async { 21 });
    let doubled = Map { fut: boxed, f: Some(|n: i32| n * 2) }.await;
    println!("{doubled}");
}

Output

Without ready!, the first line of poll would be a match with a Poll::Pending => return Poll::Pending arm. The Unpin bound is the problem this version leaves: an async block is not Unpin, because it may hold references into itself across .await.

Rust✗ does not compile · E0277

use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll, ready};

struct Map<Fut, F> {
    fut: Fut,
    f: Option<F>,
}

impl<Fut, F, T> Future for Map<Fut, F>
where
    Fut: Future + Unpin,
    F: FnOnce(Fut::Output) -> T + Unpin,
{
    type Output = T;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<T> {
        let output = ready!(Pin::new(&mut self.fut).poll(cx));
        let f = self.f.take().expect("Map polled after completion");
        Poll::Ready(f(output))
    }
}

#[tokio::main]
async fn main() {
    let doubled = Map { fut: async { 21 }, f: Some(|n: i32| n * 2) }.await;
    println!("{doubled}");
}

Output

error[E0277]: `{async block@src/main.rs:26:30: 26:35}` cannot be unpinned
  --> src/main.rs:26:70
   |
26 |     let doubled = Map { fut: async { 21 }, f: Some(|n: i32| n * 2) }.await;
   |                                                                      ^^^^^ the trait `Unpin` is not implemented for `{async block@src/main.rs:26:30: 26:35}`
   |
   = note: consider using the `pin!` macro
           consider using `Box::pin` if you need to access the pinned value outside of the current scope
note: required for `Map<..., ...>` to implement `Future`
  --> src/main.rs:10:17
   |
10 | impl<Fut, F, T> Future for Map<Fut, F>
   |                 ^^^^^^     ^^^^^^^^^^^
11 | where
12 |     Fut: Future + Unpin,
   |                   ----- unsatisfied trait bound introduced here
   = note: required for `Map<..., ...>` to implement `IntoFuture`
help: remove the `.await`
   |
26 -     let doubled = Map { fut: async { 21 }, f: Some(|n: i32| n * 2) }.await;
26 +     let doubled = Map { fut: async { 21 }, f: Some(|n: i32| n * 2) };
   |

Boxing every inner future costs an allocation per combinator. Pin projection removes both the box and the bound.

Example 9: Pin projection with `pin-project-lite`

Pin projection turns Pin<&mut Struct> into a Pin<&mut Field> for fields that must stay pinned, and a plain &mut Field for the others. Writing it by hand needs unsafe. The pin_project! macro from pin-project-lite generates a safe project() method: fields marked #[pin] come out as Pin<&mut T>, the rest as &mut T.

Rust

use pin_project_lite::pin_project;
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll, ready};

pin_project! {
    struct Map<Fut, F> {
        #[pin]
        fut: Fut,
        f: Option<F>,
    }
}

impl<Fut, F, T> Future for Map<Fut, F>
where
    Fut: Future,
    F: FnOnce(Fut::Output) -> T,
{
    type Output = T;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<T> {
        let this = self.project(); // this.fut: Pin<&mut Fut>, this.f: &mut Option<F>
        let output = ready!(this.fut.poll(cx));
        let f = this.f.take().expect("Map polled after completion");
        Poll::Ready(f(output))
    }
}

fn map<Fut: Future, F>(fut: Fut, f: F) -> Map<Fut, F> {
    Map { fut, f: Some(f) }
}

#[tokio::main]
async fn main() {
    let doubled = map(async { 21 }, |n: i32| n * 2).await;
    let text = map(async { "tokio" }, |s: &str| s.to_uppercase()).await;
    println!("{doubled} {text}");
}

Output

42 TOKIO

The macro also makes Map<Fut, F> Unpin exactly when the pinned field Fut is Unpin, and forbids a manual Drop that could move pinned fields. Tokio uses pin-project-lite for its own combinators, such as Timeout and Take.

Field kind	After `project()`	Use it for
`#[pin] fut: Fut`	`Pin<&mut Fut>`	inner futures, streams, readers, `Sleep`
`f: Option<F>`	`&mut Option<F>`	counters, closures, buffers, outputs

Example 10: A timeout combinator

A timeout is a combinator with two pinned fields: the inner future and a Sleep. It polls the inner future first, then the timer. Both register the same waker, so whichever becomes ready first wakes the task.

Rust

use pin_project_lite::pin_project;
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
use std::time::Duration;
use tokio::time::{Instant, Sleep, sleep};

#[derive(Debug)]
struct TimedOut;

pin_project! {
    struct Timeout<F> {
        #[pin]
        fut: F,
        #[pin]
        timer: Sleep,
    }
}

impl<F: Future> Future for Timeout<F> {
    type Output = Result<F::Output, TimedOut>;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let this = self.project();
        if let Poll::Ready(value) = this.fut.poll(cx) {
            return Poll::Ready(Ok(value));
        }
        match this.timer.poll(cx) {
            Poll::Ready(()) => Poll::Ready(Err(TimedOut)),
            Poll::Pending => Poll::Pending,
        }
    }
}

fn with_timeout<F: Future>(limit: Duration, fut: F) -> Timeout<F> {
    Timeout { fut, timer: sleep(limit) }
}

async fn work(ms: u64) -> &'static str {
    sleep(Duration::from_millis(ms)).await;
    "done"
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let start = Instant::now();
    let fast = with_timeout(Duration::from_millis(100), work(30)).await;
    println!("fast: {fast:?} after {:?}", start.elapsed());

    let start = Instant::now();
    let slow = with_timeout(Duration::from_millis(100), work(500)).await;
    println!("slow: {slow:?} after {:?}", start.elapsed());
}

Output

fast: Ok("done") after 30ms
slow: Err(TimedOut) after 100ms

Sleep is !Unpin (it contains a timer entry that the timer wheel refers to by address), which is why it needs #[pin] here or a Box::pin in Example 4. When Timeout is dropped, the inner future and the timer are dropped with it; that drop is how the slow work is cancelled.

Example 11: Joining two futures

A join polls two futures until both finish and returns both outputs. It must not poll a future again after that future returned Ready, so each output is kept in an Option and a future is polled only while its output is None.

Rust

use pin_project_lite::pin_project;
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
use std::time::Duration;
use tokio::time::{Instant, sleep};

pin_project! {
    struct Join2<A, B>
    where
        A: Future,
        B: Future,
    {
        #[pin]
        a: A,
        #[pin]
        b: B,
        a_out: Option<A::Output>,
        b_out: Option<B::Output>,
    }
}

impl<A: Future, B: Future> Future for Join2<A, B> {
    type Output = (A::Output, B::Output);

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let this = self.project();
        if this.a_out.is_none() {
            if let Poll::Ready(v) = this.a.poll(cx) {
                *this.a_out = Some(v);
            }
        }
        if this.b_out.is_none() {
            if let Poll::Ready(v) = this.b.poll(cx) {
                *this.b_out = Some(v);
            }
        }
        if this.a_out.is_some() && this.b_out.is_some() {
            Poll::Ready((this.a_out.take().unwrap(), this.b_out.take().unwrap()))
        } else {
            Poll::Pending
        }
    }
}

fn join2<A: Future, B: Future>(a: A, b: B) -> Join2<A, B> {
    Join2 { a, b, a_out: None, b_out: None }
}

async fn step(name: &'static str, ms: u64, start: Instant) -> &'static str {
    sleep(Duration::from_millis(ms)).await;
    println!("{name} finished at {:?}", start.elapsed());
    name
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let start = Instant::now();
    let (a, b) = join2(step("a", 300, start), step("b", 100, start)).await;
    println!("joined ({a}, {b}) after {:?}", start.elapsed());
}

Output

b finished at 100ms
a finished at 300ms
joined (a, b) after 300ms

The total time is the longer of the two, not the sum, because both futures make progress inside one poll. This is what tokio::join! does for any number of futures. The pending branch is correct under rule 2: at least one of the two futures is unfinished, and it returned Pending during this call.

Example 12: A retry combinator

A retry combinator holds a factory closure and the current attempt. When the attempt fails, it builds a new future and replaces the pinned field with Pin::set, which drops the old future in place and pins the new one in the same memory.

Rust

use pin_project_lite::pin_project;
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll, ready};

pin_project! {
    struct Retry<M, Fut> {
        make: M,
        #[pin]
        fut: Fut,
        attempt: u32,
        max_attempts: u32,
    }
}

impl<M, Fut, T, E> Future for Retry<M, Fut>
where
    M: FnMut(u32) -> Fut,
    Fut: Future<Output = Result<T, E>>,
    E: std::fmt::Display,
{
    type Output = Result<T, E>;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let mut this = self.project();
        loop {
            match ready!(this.fut.as_mut().poll(cx)) {
                Ok(value) => return Poll::Ready(Ok(value)),
                Err(e) if *this.attempt < *this.max_attempts => {
                    println!("attempt {} failed: {e}", this.attempt);
                    *this.attempt += 1;
                    let next = (this.make)(*this.attempt);
                    this.fut.set(next); // poll the new attempt in the next loop turn
                }
                Err(e) => return Poll::Ready(Err(e)),
            }
        }
    }
}

fn retry<M, Fut>(max_attempts: u32, mut make: M) -> Retry<M, Fut>
where
    M: FnMut(u32) -> Fut,
{
    let fut = make(1);
    Retry { make, fut, attempt: 1, max_attempts }
}

async fn flaky(attempt: u32, succeed_on: u32) -> Result<String, String> {
    tokio::task::yield_now().await;
    if attempt >= succeed_on {
        Ok(format!("succeeded on attempt {attempt}"))
    } else {
        Err("connection refused".to_string())
    }
}

#[tokio::main]
async fn main() {
    let ok = retry(5, |attempt| flaky(attempt, 3)).await;
    println!("{ok:?}");

    let err = retry(2, |attempt| flaky(attempt, 10)).await;
    println!("{err:?}");
}

Output

attempt 1 failed: connection refused
attempt 2 failed: connection refused
Ok("succeeded on attempt 3")
attempt 1 failed: connection refused
Err("connection refused")

this.fut.as_mut() reborrows the Pin<&mut Fut> so it can be used again after poll. The loop polls the new attempt immediately, so a Pending return always comes from a live inner future. A production version would add a delay between attempts by storing a Sleep in a second pinned field, which is Project 24 in Workshop 6: Runtime Internals.

Example 13: Spawning a hand-written future

A hand-written future is an ordinary future to the runtime. tokio::spawn accepts it when it is Send + 'static, and it can be combined with async code freely.

Rust

use pin_project_lite::pin_project;
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll, ready};

pin_project! {
    struct CountPolls<F> {
        #[pin]
        inner: F,
        polls: u32,
    }
}

impl<F: Future> Future for CountPolls<F> {
    type Output = (F::Output, u32);

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let this = self.project();
        *this.polls += 1;
        let out = ready!(this.inner.poll(cx));
        Poll::Ready((out, *this.polls))
    }
}

#[tokio::main]
async fn main() {
    let work = async {
        for _ in 0..3 {
            tokio::task::yield_now().await;
        }
        "work done"
    };
    let handle = tokio::spawn(CountPolls { inner: work, polls: 0 });
    let (out, polls) = handle.await.unwrap();
    println!("{out} after {polls} polls");
}

Output

work done after 4 polls

Each yield_now returns Pending once, so the task is polled four times: three times to pass the yields and once to finish. Wrapping a future to count or time its polls is how instrumentation layers such as tracing::Instrument work.

Example 14: Implementing `Stream`

A stream yields many values over time. Streams introduced the trait from tokio-stream (a re-export of futures_core::Stream). Its method is poll_next, which returns Poll<Option<Item>>: Ready(Some(item)) for a value, Ready(None) at the end, and Pending when no value is available yet.

Rust

use std::pin::Pin;
use std::task::{Context, Poll};
use tokio_stream::{Stream, StreamExt};

struct Countdown {
    next: u32,
}

impl Stream for Countdown {
    type Item = u32;

    fn poll_next(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Option<u32>> {
        if self.next == 0 {
            return Poll::Ready(None);
        }
        let n = self.next;
        self.next -= 1;
        Poll::Ready(Some(n))
    }

    fn size_hint(&self) -> (usize, Option<usize>) {
        let n = self.next as usize;
        (n, Some(n))
    }
}

#[tokio::main]
async fn main() {
    let mut countdown = Countdown { next: 3 };
    println!("size_hint: {:?}", countdown.size_hint());
    while let Some(n) = countdown.next().await {
        println!("{n}");
    }

    let tens: Vec<u32> = Countdown { next: 4 }.map(|n| n * 10).collect().await;
    println!("{tens:?}");
}

Output

size_hint: (3, Some(3))
3
2
1
[40, 30, 20, 10]

size_hint is optional and returns (0, None) by default. Every StreamExt adapter (map, filter, collect) works on the new type because StreamExt is implemented for every Stream.

Example 15: A stream driven by a timer

A stream that returns Pending follows the same rule as a future: an inner source must hold the waker. tokio::time::Interval has poll_tick(&mut self, cx), so a stream can wrap it without pinning (Interval is Unpin).

Rust

use std::pin::Pin;
use std::task::{Context, Poll, ready};
use std::time::Duration;
use tokio::time::{Instant, Interval, interval};
use tokio_stream::{Stream, StreamExt};

struct Ticks {
    interval: Interval,
    remaining: u32,
}

impl Stream for Ticks {
    type Item = u32;

    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<u32>> {
        if self.remaining == 0 {
            return Poll::Ready(None);
        }
        ready!(self.interval.poll_tick(cx));
        self.remaining -= 1;
        Poll::Ready(Some(self.remaining))
    }
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let start = Instant::now();
    let mut ticks = Ticks { interval: interval(Duration::from_millis(100)), remaining: 4 };
    while let Some(left) = ticks.next().await {
        println!("{:?}: tick, {left} left", start.elapsed());
    }
}

Output

0ns: tick, 3 left
100ms: tick, 2 left
200ms: tick, 1 left
300ms: tick, 0 left

The first tick of an Interval completes immediately, as described in Time: sleep, timeout, and interval, so the items arrive at 0, 100, 200, and 300 ms.

Example 16: A stream combinator with a loop

A filtering stream may need several items from its source before it has one to return. It loops inside poll_next until it finds a match, the source ends, or the source returns Pending. Returning Pending after skipping an item, without polling the source again, would break rule 2.

Rust

use pin_project_lite::pin_project;
use std::pin::Pin;
use std::task::{Context, Poll, ready};
use tokio_stream::{Stream, StreamExt};

pin_project! {
    struct Evens<S> {
        #[pin]
        inner: S,
        skipped: u32,
    }
}

impl<S: Stream<Item = u32>> Stream for Evens<S> {
    type Item = u32;

    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<u32>> {
        let mut this = self.project();
        loop {
            match ready!(this.inner.as_mut().poll_next(cx)) {
                Some(n) if n % 2 == 0 => return Poll::Ready(Some(n)),
                Some(_) => *this.skipped += 1,
                None => return Poll::Ready(None),
            }
        }
    }
}

#[tokio::main]
async fn main() {
    let mut evens = Evens { inner: tokio_stream::iter(1..=9), skipped: 0 };
    let mut found = Vec::new();
    while let Some(n) = evens.next().await {
        found.push(n);
    }
    println!("evens: {found:?}, skipped: {}", evens.skipped);
}

Output

evens: [2, 4, 6, 8], skipped: 5

The loop has no upper bound of its own. While the source keeps returning ready items, the task stays inside one poll_next call. tokio_stream::iter takes part in Tokio's cooperative budget (its poll_next calls tokio::task::coop::poll_proceed), so after a fixed number of items it returns Pending and the loop ends; The Multi-Thread Scheduler explains the budget. A hand-written source such as Countdown in Example 14 has no budget, so a filter over it can add a counter, call cx.waker().wake_by_ref(), and return Pending after a fixed number of skips.

Example 17: A counting `AsyncRead` wrapper

AsyncRead from AsyncRead and AsyncWrite has one required method:

Rust

fn poll_read(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<io::Result<()>>;

ReadBuf tracks how much of the caller's buffer is filled. A read adds bytes to the filled part; returning Ok(()) with nothing added means end of file. This wrapper forwards to an inner reader and counts calls and bytes.

Rust

use pin_project_lite::pin_project;
use std::io;
use std::pin::Pin;
use std::task::{Context, Poll, ready};
use tokio::io::{AsyncBufReadExt, AsyncRead, AsyncReadExt, BufReader, ReadBuf};

pin_project! {
    struct CountingReader<R> {
        #[pin]
        inner: R,
        reads: usize,
        bytes: usize,
    }
}

impl<R> CountingReader<R> {
    fn new(inner: R) -> Self {
        CountingReader { inner, reads: 0, bytes: 0 }
    }
}

impl<R: AsyncRead> AsyncRead for CountingReader<R> {
    fn poll_read(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &mut ReadBuf<'_>,
    ) -> Poll<io::Result<()>> {
        let this = self.project();
        let before = buf.filled().len();
        ready!(this.inner.poll_read(cx, buf))?;
        *this.reads += 1;
        *this.bytes += buf.filled().len() - before;
        Poll::Ready(Ok(()))
    }
}

#[tokio::main]
async fn main() -> io::Result<()> {
    let data: &[u8] = b"alpha\nbeta\ngamma\n";

    let mut reader = CountingReader::new(data);
    let mut text = String::new();
    reader.read_to_string(&mut text).await?;
    println!("read_to_string: {} calls, {} bytes", reader.reads, reader.bytes);

    let mut lines = BufReader::new(CountingReader::new(data)).lines();
    let mut count = 0;
    while let Some(_line) = lines.next_line().await? {
        count += 1;
    }
    let inner = lines.into_inner().into_inner();
    println!("lines: {count} lines, {} calls, {} bytes", inner.reads, inner.bytes);
    Ok(())
}

Output

read_to_string: 2 calls, 17 bytes
lines: 3 lines, 2 calls, 17 bytes

read_to_string keeps calling poll_read until a call adds no bytes, so the last call is the end-of-file check. BufReader reads into its own 8 KiB buffer and splits lines from there, so the inner reader sees few large reads instead of one read per line.

Example 18: A throttled reader

A wrapper can also limit how fast data flows. This reader waits for an Interval tick before each read and hands out at most chunk bytes per read. It reads into a small stack buffer and copies the bytes into the caller's ReadBuf with put_slice, which needs no unsafe.

Rust

use pin_project_lite::pin_project;
use std::io;
use std::pin::Pin;
use std::task::{Context, Poll, ready};
use std::time::Duration;
use tokio::io::{AsyncRead, AsyncReadExt, ReadBuf};
use tokio::time::{Instant, Interval, interval};

pin_project! {
    struct Throttled<R> {
        #[pin]
        inner: R,
        interval: Interval,
        chunk: usize,
    }
}

impl<R: AsyncRead> AsyncRead for Throttled<R> {
    fn poll_read(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &mut ReadBuf<'_>,
    ) -> Poll<io::Result<()>> {
        let this = self.project();
        ready!(this.interval.poll_tick(cx));

        let mut scratch = [0u8; 64];
        let limit = (*this.chunk).min(buf.remaining()).min(scratch.len());
        let mut small = ReadBuf::new(&mut scratch[..limit]);
        ready!(this.inner.poll_read(cx, &mut small))?;
        buf.put_slice(small.filled());
        Poll::Ready(Ok(()))
    }
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() -> io::Result<()> {
    let start = Instant::now();
    let data: &[u8] = b"0123456789";
    let mut reader = Throttled { inner: data, interval: interval(Duration::from_millis(100)), chunk: 4 };

    let mut buf = [0u8; 16];
    loop {
        let n = reader.read(&mut buf).await?;
        println!("{:?}: {n} bytes {:?}", start.elapsed(), String::from_utf8_lossy(&buf[..n]));
        if n == 0 {
            break;
        }
    }
    Ok(())
}

Output

0ns: 4 bytes "0123"
100ms: 4 bytes "4567"
200ms: 2 bytes "89"
300ms: 0 bytes ""

Ten bytes in chunks of four take three reads plus a fourth read that returns 0 (end of file), one tick each. One limit of this design: if the inner reader returns Pending after a tick was consumed, that tick is lost and the next read waits a full period. A version for sockets would remember an unspent tick in a bool field.

When to write a future by hand

Situation	Choice
sequential steps with `.await`	`async fn` or `async` block
a reusable wrapper around any future or stream	struct + `pin_project!` + `impl Future`/`Stream`
a type that other libraries must name (for example, a return type in a trait before `async fn` in traits, or a public combinator)	struct with `impl Future`
an event from a callback or another thread	shared slot with value + waker (Example 5), or a `tokio::sync::oneshot`
a one-off future over existing `poll_*` methods	`std::future::poll_fn`
implementing `AsyncRead`/`AsyncWrite`/`Stream` for a protocol type	hand-written `poll_*` methods

Interview points

The three rules of poll: return Ready once, return Pending only after arranging a wake-up (an inner Pending or a stored waker), and never block.
A missing wake-up is a hang with no error. It shows up as a task that finishes only when some unrelated event (a timer, another message) happens to poll it.
An async fn becomes an enum state machine; a hand-written one needs a loop so a new state is polled in the same call.
Pin<&mut Self> gives field access only for Unpin types. Pin projection gives Pin<&mut Field> for fields that must stay pinned; pin-project-lite generates it without unsafe.
Sleep and async blocks are !Unpin; store them with #[pin] or Box::pin.
Pin::set replaces a pinned value in place and drops the old one; retry combinators use it.
Waker discipline: change state, then take the waker, release the lock, then wake. Store the waker from the latest poll; will_wake avoids needless clones.
ready! is early return for Poll; poll_fn builds a future from a closure over poll_* methods.
Stream::poll_next returns Poll<Option<Item>>; Ready(None) ends the stream.
AsyncRead::poll_read signals EOF by returning Ok(()) without filling any bytes.

Summary

impl Future needs type Output and poll(self: Pin<&mut Self>, cx) -> Poll<Output>.
Pending is correct only when an inner future returned Pending or the future stored cx.waker() for someone to call.
Hand-written state machines loop through states within one poll and panic on poll after completion.
A shared-state future stores the value before waking and keeps the latest waker.
std::future::poll_fn and std::task::ready! are in the standard library.
pin_project! marks pinned fields with #[pin]; project() returns Pin<&mut T> for them and &mut T for the rest.
Map, timeout, join, and retry are each a struct with pinned inner futures and a few plain fields.
Stream and AsyncRead follow the same waker rules as Future.

The contract of poll

Example 1: A future that is ready at once

Example 2: Returning Pending and waking yourself

Example 3: Forgetting to wake

Example 4: A state machine by hand

Example 5: A value filled by another thread

Example 6: Why the latest waker matters

Example 7: A future from a closure with poll_fn

Example 8: ready! and a map combinator that needs Unpin

Example 9: Pin projection with pin-project-lite

Example 10: A timeout combinator

Example 11: Joining two futures

Example 12: A retry combinator

Example 13: Spawning a hand-written future

Example 14: Implementing Stream

Example 15: A stream driven by a timer

Example 16: A stream combinator with a loop

Example 17: A counting AsyncRead wrapper

Example 18: A throttled reader

When to write a future by hand

Interview points

Summary

The contract of `poll`

Example 7: A future from a closure with `poll_fn`

Example 8: `ready!` and a map combinator that needs `Unpin`

Example 9: Pin projection with `pin-project-lite`

Example 14: Implementing `Stream`

Example 17: A counting `AsyncRead` wrapper