How Async Works: Futures, Pin & Wakers

The earlier async lessons used async, .await, and Tokio without looking inside. This lesson opens the box. You need it to read the source of tokio, tower, hyper, or any crate that implements Future or Stream by hand, and to understand errors that mention Pin or Unpin.

A Rust future is a plain value that does nothing until something calls its poll method, and the thing that calls poll is the executor (Tokio's runtime, for example).

Rust

// The whole trait, from std::future.
pub trait Future {
    type Output;
    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output>;
}

pub enum Poll<T> {
    Ready(T),
    Pending,
}

Part 1: Futures are polled

1. Nothing runs until the future is polled

To see polling without a runtime, create a Context from Waker::noop() and call poll in a loop yourself. YieldOnce is a tiny future that returns Pending the first time, like a real future waiting for I/O.

Rust

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

struct YieldOnce {
    yielded: bool,
}

impl Future for YieldOnce {
    type Output = ();

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
        if self.yielded {
            Poll::Ready(())
        } else {
            self.yielded = true;
            cx.waker().wake_by_ref();
            Poll::Pending
        }
    }
}

async fn steps() -> &'static str {
    println!("  step 1");
    YieldOnce { yielded: false }.await;
    println!("  step 2");
    YieldOnce { yielded: false }.await;
    println!("  step 3");
    "done"
}

fn main() {
    let future = steps();
    println!("future created, nothing has run yet");

    let mut future = pin!(future);
    let mut cx = Context::from_waker(Waker::noop());
    for poll_number in 1.. {
        println!("poll #{poll_number}");
        match future.as_mut().poll(&mut cx) {
            Poll::Ready(value) => {
                println!("ready: {value}");
                break;
            }
            Poll::Pending => println!("pending"),
        }
    }
}

Output

future created, nothing has run yet
poll #1
  step 1
pending
poll #2
  step 2
pending
poll #3
  step 3
ready: done

Each poll runs the function until the next .await that isn't ready, then returns Pending. The next poll continues from exactly that point.

2. An `async fn` is a state machine

The compiler turns every async fn into an anonymous type that implements Future. It's an enum with one variant per .await point, holding the local variables that are still needed. Here is an async fn next to a hand-written equivalent:

Rust

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

fn fetch_price(cart: u32) -> Ready<u32> {
    ready(cart * 10)
}

fn fetch_tax(price: u32) -> Ready<u32> {
    ready(price / 5)
}

// What you write:
async fn checkout(cart: u32) -> u32 {
    let price = fetch_price(cart).await;
    let tax = fetch_tax(price).await;
    price + tax
}

// Roughly what the compiler generates.
enum Checkout {
    Start { cart: u32 },
    AwaitingPrice { price_future: Ready<u32> },
    AwaitingTax { price: u32, tax_future: Ready<u32> },
    Done,
}

impl Future for Checkout {
    type Output = u32;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<u32> {
        loop {
            match &mut *self {
                Checkout::Start { cart } => {
                    let price_future = fetch_price(*cart);
                    *self = Checkout::AwaitingPrice { price_future };
                }
                Checkout::AwaitingPrice { price_future } => {
                    // `ready!` returns Poll::Pending from this function if the inner future is pending.
                    let price = std::task::ready!(Pin::new(price_future).poll(cx));
                    println!("  state machine: price is {price}");
                    *self = Checkout::AwaitingTax { price, tax_future: fetch_tax(price) };
                }
                Checkout::AwaitingTax { price, tax_future } => {
                    let tax = std::task::ready!(Pin::new(tax_future).poll(cx));
                    let total = *price + tax;
                    *self = Checkout::Done;
                    return Poll::Ready(total);
                }
                Checkout::Done => panic!("polled after completion"),
            }
        }
    }
}

#[tokio::main]
async fn main() {
    println!("async fn: {}", checkout(3).await);
    println!("by hand:  {}", Checkout::Start { cart: 3 }.await);
}

Output

async fn: 36
  state machine: price is 30
by hand:  36

This is why a future's size is known at compile time and why no heap allocation is needed per .await: the whole chain of nested async fn calls becomes one value, sized like its largest state.

3. The `Waker` contract

Returning Pending is a promise: "I have arranged for cx.waker() to be called when I can make progress." The executor doesn't poll a pending task again until its waker is called. A future that returns Pending without doing that is never polled again, so it hangs. Each future below runs as its own task, and tokio::time::timeout waits at most 50 ms for it:

Rust

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

struct Forgetful {
    polled: bool,
}

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

    fn poll(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<&'static str> {
        if self.polled {
            return Poll::Ready("finished");
        }
        self.polled = true;
        Poll::Pending // bug: nobody will wake this task
    }
}

struct Polite {
    polled: bool,
}

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

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<&'static str> {
        if self.polled {
            return Poll::Ready("finished");
        }
        self.polled = true;
        cx.waker().wake_by_ref(); // "poll me again"
        Poll::Pending
    }
}

#[tokio::main]
async fn main() {
    let limit = Duration::from_millis(50);
    let forgetful = tokio::spawn(Forgetful { polled: false });
    let polite = tokio::spawn(Polite { polled: false });
    println!("forgetful: {:?}", tokio::time::timeout(limit, forgetful).await.map(Result::unwrap));
    println!("polite:    {:?}", tokio::time::timeout(limit, polite).await.map(Result::unwrap));
}

Output

forgetful: Err(Elapsed(()))
polite:    Ok("finished")

Wakers belong to tasks. When the 50 ms timer fires, it wakes the main task, not the Forgetful task, which stays pending forever. Calling wake_by_ref() immediately, as Polite does, means "I'm not done, but try again soon". Real futures hand the waker to whatever produces the event: the OS event loop for sockets, a timer wheel for sleep, or another thread.

Part 2: Wakers and executors

4. A timer future woken from another thread

This future completes after a delay. The first poll stores the waker in shared state; a background thread sleeps, marks the state done, and calls wake(). Tokio's sleep works the same way, with one timer driver instead of a thread per timer.

Rust

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

struct Shared {
    done: bool,
    waker: Option<Waker>,
}

struct Delay {
    shared: Arc<Mutex<Shared>>,
}

impl Delay {
    fn new(duration: Duration) -> Delay {
        let shared = Arc::new(Mutex::new(Shared { done: false, waker: None }));
        let background = Arc::clone(&shared);
        thread::spawn(move || {
            thread::sleep(duration);
            let mut state = background.lock().unwrap();
            state.done = true;
            if let Some(waker) = state.waker.take() {
                waker.wake();
            }
        });
        Delay { shared }
    }
}

impl Future for Delay {
    type Output = ();

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
        let mut state = self.shared.lock().unwrap();
        if state.done {
            Poll::Ready(())
        } else {
            // Store the latest waker: a task can move between threads, and each poll may bring a new one.
            state.waker = Some(cx.waker().clone());
            Poll::Pending
        }
    }
}

#[tokio::main]
async fn main() {
    let start = Instant::now();
    let slow = async {
        Delay::new(Duration::from_millis(80)).await;
        println!("slow done");
    };
    let fast = async {
        Delay::new(Duration::from_millis(20)).await;
        println!("fast done");
    };
    tokio::join!(slow, fast);
    println!("ran concurrently: {}", start.elapsed() < Duration::from_millis(100));
}

Output

fast done
slow done
ran concurrently: true

The two delays ran at the same time: the total is about 80 ms, not the 100 ms that running them one after the other would take.

5. Writing `block_on`: the smallest executor

An executor polls a future, and when it's pending, waits until the waker is called. Here the waker unparks the executor's thread. std::task::Wake turns any Arc<T> into a Waker.

Rust

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

struct Shared {
    done: bool,
    waker: Option<Waker>,
}

struct Delay {
    shared: Arc<Mutex<Shared>>,
}

impl Delay {
    fn new(duration: Duration) -> Delay {
        let shared = Arc::new(Mutex::new(Shared { done: false, waker: None }));
        let background = Arc::clone(&shared);
        thread::spawn(move || {
            thread::sleep(duration);
            let mut state = background.lock().unwrap();
            state.done = true;
            if let Some(waker) = state.waker.take() {
                waker.wake();
            }
        });
        Delay { shared }
    }
}

impl Future for Delay {
    type Output = ();

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
        let mut state = self.shared.lock().unwrap();
        if state.done {
            Poll::Ready(())
        } else {
            state.waker = Some(cx.waker().clone());
            Poll::Pending
        }
    }
}

// Waking = unparking the thread that runs `block_on`.
struct ThreadWaker(Thread);

impl Wake for ThreadWaker {
    fn wake(self: Arc<Self>) {
        self.0.unpark();
    }
}

fn block_on<F: Future>(future: F) -> F::Output {
    let mut future = pin!(future);
    let waker = Waker::from(Arc::new(ThreadWaker(thread::current())));
    let mut cx = Context::from_waker(&waker);
    loop {
        match future.as_mut().poll(&mut cx) {
            Poll::Ready(value) => return value,
            Poll::Pending => thread::park(), // sleep until a waker unparks us
        }
    }
}

async fn fetch_order(id: u32) -> String {
    Delay::new(Duration::from_millis(20)).await;
    format!("order {id}")
}

fn main() {
    let result = block_on(async {
        let first = fetch_order(1).await;
        let second = fetch_order(2).await;
        format!("{first}, {second}")
    });
    println!("{result}");
}

Output

order 1, order 2

thread::park may return early without an unpark. That's harmless here: the loop just polls again and gets Pending again. Futures must tolerate being polled more often than necessary.

6. Realistic: a tiny multi-task executor

A runtime runs many tasks. Each task is a boxed future plus a way to put itself back on the run queue. Its waker does exactly that: waking a task sends it to the queue, and the executor loop polls whatever arrives. This is the core of Tokio's scheduler, without the thread pool, I/O driver, and work stealing.

Rust

use std::future::Future;
use std::pin::Pin;
use std::sync::mpsc::{self, Receiver, Sender};
use std::sync::{Arc, Mutex};
use std::task::{Context, Poll, Wake, Waker};
use std::thread;
use std::time::Duration;

struct Shared {
    done: bool,
    waker: Option<Waker>,
}

struct Delay {
    shared: Arc<Mutex<Shared>>,
}

impl Delay {
    fn new(duration: Duration) -> Delay {
        let shared = Arc::new(Mutex::new(Shared { done: false, waker: None }));
        let background = Arc::clone(&shared);
        thread::spawn(move || {
            thread::sleep(duration);
            let mut state = background.lock().unwrap();
            state.done = true;
            if let Some(waker) = state.waker.take() {
                waker.wake();
            }
        });
        Delay { shared }
    }
}

impl Future for Delay {
    type Output = ();

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
        let mut state = self.shared.lock().unwrap();
        if state.done {
            Poll::Ready(())
        } else {
            state.waker = Some(cx.waker().clone());
            Poll::Pending
        }
    }
}

type BoxedFuture = Pin<Box<dyn Future<Output = ()> + Send>>;

struct Task {
    future: Mutex<Option<BoxedFuture>>,
    queue: Sender<Arc<Task>>,
}

impl Wake for Task {
    fn wake(self: Arc<Self>) {
        let queue = self.queue.clone();
        let _ = queue.send(self); // back onto the run queue
    }
}

struct Executor {
    queue: Sender<Arc<Task>>,
    ready: Receiver<Arc<Task>>,
    live_tasks: usize,
}

impl Executor {
    fn new() -> Executor {
        let (queue, ready) = mpsc::channel();
        Executor { queue, ready, live_tasks: 0 }
    }

    fn spawn(&mut self, future: impl Future<Output = ()> + Send + 'static) {
        let task = Arc::new(Task { future: Mutex::new(Some(Box::pin(future))), queue: self.queue.clone() });
        self.live_tasks += 1;
        self.queue.send(task).unwrap();
    }

    fn run(mut self) {
        let mut polls = 0;
        while self.live_tasks > 0 {
            let task = self.ready.recv().unwrap();
            let mut slot = task.future.lock().unwrap();
            let Some(mut future) = slot.take() else { continue }; // already finished
            let waker = Waker::from(Arc::clone(&task));
            polls += 1;
            match future.as_mut().poll(&mut Context::from_waker(&waker)) {
                Poll::Ready(()) => self.live_tasks -= 1,
                Poll::Pending => *slot = Some(future),
            }
        }
        println!("all tasks finished after {polls} polls");
    }
}

async fn job(name: &'static str, steps: u32, step_ms: u64) {
    for step in 1..=steps {
        Delay::new(Duration::from_millis(step_ms)).await;
        println!("{name}: step {step}/{steps}");
    }
}

fn main() {
    let mut executor = Executor::new();
    executor.spawn(job("email", 1, 100));
    executor.spawn(job("resize", 3, 20));
    executor.spawn(job("report", 1, 150));
    executor.run();
}

Output

resize: step 1/3
resize: step 2/3
resize: step 3/3
email: step 1/1
report: step 1/1
all tasks finished after 8 polls

Three tasks ran concurrently on one thread, and the executor polled only when a waker said there was progress to make: 3 initial polls plus one per completed delay.

Part 3: Pin

7. Why futures must not move

A future stores the local variables that live across an .await. If one local borrows another, the future holds a pointer into itself. Moving the future to a new address would leave that pointer dangling.

Rust

async fn checksum() -> u32 {
    let buffer = [1u8, 2, 3, 4];     // stored inside the future
    let view = &buffer[1..];         // points into the future itself
    some_io().await;                 // both are kept across this await
    view.iter().map(|&b| b as u32).sum()
}

That's why poll takes self: Pin<&mut Self> instead of &mut self. A Pin<&mut T> is a mutable reference with one extra promise: the value won't be moved again. Once a future has been polled, nobody can move it, so self-references stay valid.

Most types don't care about being moved, and they implement the auto trait Unpin: numbers, String, Vec, Box<T>, and futures written by hand without self-references. For them, Pin::new(&mut value) is safe and Pin changes nothing. The futures generated by async are not Unpin, because they might contain self-references, so Pin::new refuses them:

Rust✗ does not compile · E0277

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

async fn checksum() -> u32 {
    let buffer = [1u8, 2, 3, 4];
    let view = &buffer[1..];
    view.iter().map(|&b| b as u32).sum()
}

fn main() {
    let mut future = checksum();
    let mut cx = Context::from_waker(Waker::noop());
    let _ = Pin::new(&mut future).poll(&mut cx);
}

Output

error[E0277]: `{async fn body of checksum()}` cannot be unpinned
  --> src/main.rs:14:22
   |
 5 | async fn checksum() -> u32 {
   |                        --- within this `impl Future<Output = u32>`
...
14 |     let _ = Pin::new(&mut future).poll(&mut cx);
   |             -------- ^^^^^^^^^^^ within `impl Future<Output = u32>`, the trait `Unpin` is not implemented for `{async fn body of checksum()}`
   |             |
   |             required by a bound introduced by this call
   |
   = 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 because it appears within the type `impl Future<Output = u32>`
  --> src/main.rs:5:24
   |
 5 | async fn checksum() -> u32 {
   |                        ^^^
note: required by a bound in `Pin::<Ptr>::new`
  --> /rustc/<version>/library/core/src/pin.rs:1183:4

8. Pinning in practice: `pin!` and `Box::pin`

There are two ways to pin a future that isn't Unpin:

	Where the future lives	Use when
`std::pin::pin!(fut)` or `tokio::pin!(fut)`	on the stack, in the current function	polling it by reference in a loop, `select!`
`Box::pin(fut)`	on the heap	returning it, storing it in a struct or a `Vec`, `dyn Future`

The most common place you need this is select! in a loop. &mut future can be polled repeatedly only if the future is pinned; here one deadline covers the whole loop instead of restarting on every message:

Rust

use std::pin::pin;
use std::time::Duration;
use tokio::sync::mpsc;
use tokio::time::sleep;

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel(8);
    tokio::spawn(async move {
        for n in 1.. {
            if tx.send(n).await.is_err() {
                break;
            }
            sleep(Duration::from_millis(40)).await;
        }
    });

    let mut deadline = pin!(sleep(Duration::from_millis(150)));
    let mut received = Vec::new();
    loop {
        tokio::select! {
            _ = &mut deadline => break,
            Some(n) = rx.recv() => received.push(n),
        }
    }
    println!("received {} messages before the deadline", received.len());
}

Output

received 4 messages before the deadline

Without pin!, &mut deadline fails to compile with the same "cannot be unpinned" error as example 7.

9. Pin projection: futures that wrap futures

A future that wraps another future must poll the inner one, and polling needs a Pin<&mut Inner>. Getting that from Pin<&mut Self> is called projection. Doing it by hand requires unsafe code. Almost every async crate uses the pin-project-lite (or pin-project) crate instead, so self.project() is a line you'll see throughout tokio, tower, and hyper:

Terminal

cargo add pin-project-lite

Rust

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

pin_project_lite::pin_project! {
    /// Wraps any future and counts how many times it was polled.
    struct CountPolls<F> {
        #[pin]
        inner: F,     // pinned field: projected to Pin<&mut F>
        polls: u32,   // normal field: projected to &mut 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;
        match this.inner.poll(cx) {
            Poll::Ready(value) => Poll::Ready((value, *this.polls)),
            Poll::Pending => Poll::Pending,
        }
    }
}

async fn three_yields() -> &'static str {
    for _ in 0..3 {
        tokio::task::yield_now().await;
    }
    "done"
}

#[tokio::main]
async fn main() {
    let (value, polls) = CountPolls { inner: three_yields(), polls: 0 }.await;
    println!("{value} (polled {polls} times)");

    let (value, polls) = CountPolls { inner: async { 42 }, polls: 0 }.await;
    println!("{value} (polled {polls} times)");
}

Output

done (polled 4 times)
42 (polled 1 times)

CountPolls itself isn't Unpin when F isn't, which is exactly right: moving it would move the inner future.

10. Realistic: writing a timeout future

tokio::time::timeout is a future with two pinned fields: the inner future and a Sleep. Each poll checks the inner future first, then the timer. Both register their wakers with the same cx, so the task wakes when either can make progress.

Rust

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

#[derive(Debug)]
struct TimedOut;

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

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

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.inner.poll(cx) {
            return Poll::Ready(Ok(value));
        }
        match this.timer.poll(cx) {
            Poll::Ready(()) => Poll::Ready(Err(TimedOut)),
            Poll::Pending => Poll::Pending,
        }
    }
}

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

#[tokio::main]
async fn main() {
    let limit = Duration::from_millis(50);
    println!("{:?}", with_timeout(limit, fetch("cache", 10)).await);
    println!("{:?}", with_timeout(limit, fetch("database", 200)).await);
}

Output

Ok("cache (10 ms)")
Err(TimedOut)

Part 4: Cancellation

11. Cancelling a future means dropping it

There is no cancel() method. A future stops running when its owner drops it instead of polling it again: timeout drops the inner future when time runs out, select! drops the branches that lost, and JoinHandle::abort drops the task. Local variables are dropped as usual, so Drop implementations run as cleanup:

Rust

use std::time::Duration;
use tokio::time::{sleep, timeout};

struct TempFile(&'static str);

impl Drop for TempFile {
    fn drop(&mut self) {
        println!("deleted {}", self.0);
    }
}

async fn upload() -> &'static str {
    let _temp = TempFile("upload.part");
    println!("upload started");
    sleep(Duration::from_secs(10)).await; // the future is dropped while waiting here
    println!("upload finished");
    "uploaded"
}

#[tokio::main]
async fn main() {
    let result = timeout(Duration::from_millis(20), upload()).await;
    println!("{result:?}");
}

Output

upload started
deleted upload.part
Err(Elapsed(()))

The code after the .await never runs. A future is cancellation safe if dropping it at any .await loses no data. Reading into a buffer that lives outside the future is safe; a future that has taken a message off a queue and is holding it across an .await is not. Tokio's documentation states this property for each method that's commonly used in select!.

Summary

A future is a value with a poll method. It does nothing until polled; each poll runs it to the next .await that isn't ready.
The compiler turns an async fn into a state machine enum that stores the locals living across each .await.
Returning Pending promises that cx.waker() will be called later. The executor only polls again after a wake; a future that forgets to arrange one hangs.
An executor is a loop that polls tasks and sleeps until a waker fires. std::task::Wake turns an Arc<T> into a Waker.
Futures from async may point into themselves, so poll takes Pin<&mut Self>. Unpin types don't care; pin others with pin! (stack) or Box::pin (heap).
Futures that wrap futures use pin projection, almost always through pin_project_lite::pin_project! and self.project().
Cancellation is dropping the future; Drop runs for its locals, and code after the pending .await never runs.

This completes the Async section. Next: Memory Without a Garbage Collector, the start of the Memory Management section.

Part 1: Futures are polled

1. Nothing runs until the future is polled

2. An async fn is a state machine

3. The Waker contract

Part 2: Wakers and executors

4. A timer future woken from another thread

5. Writing block_on: the smallest executor

6. Realistic: a tiny multi-task executor

Part 3: Pin

7. Why futures must not move

8. Pinning in practice: pin! and Box::pin

9. Pin projection: futures that wrap futures

10. Realistic: writing a timeout future

Part 4: Cancellation

11. Cancelling a future means dropping it

Summary

2. An `async fn` is a state machine

3. The `Waker` contract

5. Writing `block_on`: the smallest executor

8. Pinning in practice: `pin!` and `Box::pin`