Polling, Wakers, and a Tiny Executor

A future does nothing until something calls poll on it, and a pending future needs a way to say "poll me again". This lesson builds that machinery by hand: a Waker, a block_on, a timer future, and a small executor with a run queue. Afterwards the reader can explain why a future that returns Pending must arrange a wake, and can spot the bug when it does not.

Idea	Examples
`Context`, `Waker`, the `Wake` trait	1
`Pin`, `Box::pin`, `pin!`, polling by hand	2
`block_on` with thread park and unpark	3
A timer future woken by a helper thread	4
A run-queue executor	5
Timers and spawning inside the executor	6
The missed-wakeup bug and its fix	7
Why `tokio::time::sleep` needs a Tokio runtime	8

The previous lesson showed the Future trait and a future that is pending once. Its poll method takes a Context. This lesson explains what that Context carries.

Example 1: a Waker counts wakes

A waker is a handle a future stores so that it can later tell the executor "this task can make progress, poll it again". A Context is a wrapper that gives poll access to the current task's waker through cx.waker().

The Wake trait is the safe way to build a waker: implement wake on a type inside an Arc, then convert the Arc with Waker::from. The lower-level type RawWaker builds a waker from a raw pointer and a table of function pointers. It needs unsafe code, and Wake covers the same need.

Rust

use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::task::{Context, Wake, Waker};

struct CountWake {
    count: AtomicUsize,
}

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

fn main() {
    let counter = Arc::new(CountWake { count: AtomicUsize::new(0) });
    let waker = Waker::from(counter.clone());
    let cx = Context::from_waker(&waker);

    cx.waker().wake_by_ref();
    cx.waker().wake_by_ref();
    waker.clone().wake();
    println!("wakes: {}", counter.count.load(Ordering::SeqCst));
}

Output

wakes: 3

wake_by_ref signals without consuming the waker. wake consumes it, so the example clones first. A waker is cheap to clone: cloning it clones the Arc.

Example 2: Pin and polling by hand

poll takes self: Pin<&mut Self>. A Pin<&mut T> is a mutable reference that also promises the value will not be moved in memory until it is dropped. Async state machines can hold references into their own fields, and a move would leave those references dangling. Types that hold no such references implement Unpin, which lifts the restriction, so Pin does nothing to them.

Two ways to pin a future without unsafe:

Tool	Where the future lives	Use when
`Box::pin(fut)`	heap	the pinned future must be stored or returned
`std::pin::pin!(fut)`	current stack frame	the future is used only in this function

Rust

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

struct CountWake(AtomicUsize);

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

struct YieldOnce {
    yielded: bool,
}

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

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

fn main() {
    let counter = Arc::new(CountWake(AtomicUsize::new(0)));
    let waker = Waker::from(counter.clone());
    let mut cx = Context::from_waker(&waker);

    let mut boxed = Box::pin(YieldOnce { yielded: false });
    println!("boxed poll 1: {:?}", boxed.as_mut().poll(&mut cx));
    println!("boxed poll 2: {:?}", boxed.as_mut().poll(&mut cx));

    let mut on_stack = pin!(YieldOnce { yielded: false });
    println!("stack poll 1: {:?}", on_stack.as_mut().poll(&mut cx));
    println!("stack poll 2: {:?}", on_stack.as_mut().poll(&mut cx));

    println!("wakes requested: {}", counter.0.load(Ordering::SeqCst));
}

Output

boxed poll 1: Pending
boxed poll 2: Ready("finished")
stack poll 1: Pending
stack poll 2: Ready("finished")
wakes requested: 2

as_mut() reborrows the pin so the same future can be polled again. The future asks for a wake before it returns Pending, and the wake count shows one request per pending poll. Returning Pending without arranging a wake is the bug of Example 7.

Example 3: block_on with park and unpark

An executor is the code that polls futures. The smallest executor is block_on: it polls one future on the current thread, and sleeps while the future is pending. The waker's job is to wake that sleeping thread. thread::park blocks the thread until another thread calls unpark on its Thread handle.

Rust

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

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(),
        }
    }
}

struct YieldOnce(bool);

impl Future for YieldOnce {
    type Output = ();

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

fn main() {
    let total = block_on(async {
        let mut sum = 0;
        for n in 1..=3 {
            YieldOnce(false).await;
            sum += n;
            println!("added {n}");
        }
        sum
    });
    println!("total: {total}");
}

Output

added 1
added 2
added 3
total: 6

Each YieldOnce wakes the thread before it returns Pending, so park returns at once (an unpark before a park makes the next park return immediately). thread::park may also return without a matching unpark. This is harmless because the loop polls again, and polling a future that is not ready returns Pending again.

Example 4: a timer woken from a helper thread

A real timer future has nothing to poll in a loop. It waits for an event outside the executor. This one starts a helper thread that sleeps, marks the timer done, and calls the stored waker. The flag and the waker live in one Mutex so poll and the helper thread never see them half updated.

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 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(),
        }
    }
}

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

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

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

impl Future for Timer {
    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
        }
    }
}

fn main() {
    block_on(async {
        println!("waiting for the timer");
        Timer::new(Duration::from_millis(50)).await;
        println!("timer fired");
    });
}

Output

waiting for the timer
timer fired

The main thread parks after the first poll and uses no CPU while it waits. The helper thread wakes it, and the second poll finds done set. Every poll overwrites the stored waker with the latest one, because the task that polls a future can change between polls.

Example 5: an executor with a run queue

block_on runs one future. An executor that runs many tasks needs a run queue: a queue of tasks that are ready to be polled. The design is:

A task is a future plus a handle to the queue.
The task itself is the waker: wake pushes the task onto the queue.
The executor pops a task and polls it once. If the result is Pending, the executor puts the future back in the task. The task's waker pushes the task onto the queue again when the future is ready to make progress.

Rust

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

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

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

impl Wake for Task {
    fn wake(self: Arc<Self>) {
        let queue = self.queue.clone();
        queue.send(self).unwrap();
    }
}

#[derive(Clone)]
struct Spawner {
    queue: Sender<Arc<Task>>,
}

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

struct Executor {
    ready: Receiver<Arc<Task>>,
}

impl Executor {
    fn run(&self) {
        while let Ok(task) = self.ready.recv() {
            let mut slot = task.future.lock().unwrap();
            if let Some(mut future) = slot.take() {
                let waker = Waker::from(task.clone());
                let mut cx = Context::from_waker(&waker);
                if future.as_mut().poll(&mut cx).is_pending() {
                    *slot = Some(future);
                }
            }
        }
    }
}

fn new_executor() -> (Executor, Spawner) {
    let (queue, ready) = channel();
    (Executor { ready }, Spawner { queue })
}

struct YieldOnce(bool);

impl Future for YieldOnce {
    type Output = ();

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

fn main() {
    let (executor, spawner) = new_executor();
    for id in 0..3 {
        spawner.spawn(async move {
            for step in 0..2 {
                println!("task {id} step {step}");
                YieldOnce(false).await;
            }
            println!("task {id} done");
        });
    }
    drop(spawner);
    executor.run();
}

Output

task 0 step 0
task 1 step 0
task 2 step 0
task 0 step 1
task 1 step 1
task 2 step 1
task 0 done
task 1 done
task 2 done

Each yield sends the task to the back of the queue, so the three tasks take turns on one thread. This is cooperative scheduling: a task keeps the thread until it returns Pending. run ends when the channel closes, which happens once the spawner and all finished tasks (each holds a Sender) are dropped. The drop(spawner) line is required, or run would wait forever.

Example 6: timers and spawning inside the executor

The executor blocks in recv when the queue is empty, so a task waiting on a timer costs no CPU. A task can also spawn new tasks if it holds a Spawner clone. The timer here counts its polls in an atomic counter so the example can show that the task was polled exactly twice: once before the timer fired, once after the wake.

Rust

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

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

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

impl Wake for Task {
    fn wake(self: Arc<Self>) {
        let queue = self.queue.clone();
        queue.send(self).unwrap();
    }
}

#[derive(Clone)]
struct Spawner {
    queue: Sender<Arc<Task>>,
}

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

struct Executor {
    ready: Receiver<Arc<Task>>,
}

impl Executor {
    fn run(&self) {
        while let Ok(task) = self.ready.recv() {
            let mut slot = task.future.lock().unwrap();
            if let Some(mut future) = slot.take() {
                let waker = Waker::from(task.clone());
                let mut cx = Context::from_waker(&waker);
                if future.as_mut().poll(&mut cx).is_pending() {
                    *slot = Some(future);
                }
            }
        }
    }
}

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

struct Timer {
    shared: Arc<Mutex<Shared>>,
    polls: Arc<AtomicUsize>,
}

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

impl Future for Timer {
    type Output = ();

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

fn main() {
    let (queue, ready) = channel();
    let executor = Executor { ready };
    let spawner = Spawner { queue };

    let inner = spawner.clone();
    spawner.spawn(async move {
        let timer = Timer::new(Duration::from_millis(100));
        let polls = timer.polls.clone();
        println!("parent: waiting");
        timer.await;
        println!("parent: timer fired after {} polls", polls.load(Ordering::SeqCst));
        inner.spawn(async {
            println!("child: spawned by the parent");
        });
    });
    drop(spawner);
    executor.run();
    println!("executor finished");
}

Output

parent: waiting
parent: timer fired after 2 polls
child: spawned by the parent
executor finished

The child task runs after the parent prints, because the parent can spawn it only after the timer fires. run returns only after the child finished and dropped its Sender, so the last line prints last.

Example 7: the missed-wakeup bug

A future that returns Pending without storing its waker is never polled again. Nothing else tells the executor that the future is ready, so the task hangs. The next program shows this without hanging: it polls two timers by hand, joins the helper thread, and reports how many wakes arrived.

Rust

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

struct CountWake(AtomicUsize);

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

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

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

struct BadTimer(Arc<Mutex<Shared>>);

impl Future for BadTimer {
    type Output = ();

    fn poll(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<()> {
        if self.0.lock().unwrap().done {
            Poll::Ready(())
        } else {
            Poll::Pending // bug: the waker is never stored
        }
    }
}

struct GoodTimer(Arc<Mutex<Shared>>);

impl Future for GoodTimer {
    type Output = ();

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

fn wakes_received<F: Future<Output = ()>>(make: impl Fn(Arc<Mutex<Shared>>) -> F) -> usize {
    let counter = Arc::new(CountWake(AtomicUsize::new(0)));
    let waker = Waker::from(counter.clone());
    let mut cx = Context::from_waker(&waker);

    let (shared, helper) = start(Duration::from_millis(20));
    let mut timer = pin!(make(shared));
    assert!(timer.as_mut().poll(&mut cx).is_pending());
    helper.join().unwrap();
    counter.0.load(Ordering::SeqCst)
}

fn main() {
    println!("BadTimer wakes:  {}", wakes_received(BadTimer));
    println!("GoodTimer wakes: {}", wakes_received(GoodTimer));
}

Output

BadTimer wakes:  0
GoodTimer wakes: 1

The helper thread finished in both runs, so both timers are ready. Only GoodTimer told the executor. With BadTimer under block_on, park would never return and the program would hang.

A second form of the bug is a race. If poll reads done under one lock, releases it, and stores the waker under a second lock, the helper thread can run between the two steps. It sets done, finds no waker to call, and the waker stored afterwards is never used. The fix is the one GoodTimer uses: check the flag and store the waker under the same lock. The same rule applies to any signal: register interest first (or atomically with the check), then look at the state.

Example 8: futures need their runtime

The timer futures in this lesson depend only on Waker, so any executor can drive them. tokio::time::sleep differs: it registers with a timer driver that lives inside a Tokio runtime. Driven by the block_on from Example 3, it panics because no runtime is present.

Rust

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

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(),
        }
    }
}

fn main() {
    let result = catch_unwind(|| {
        block_on(async {
            tokio::time::sleep(Duration::from_millis(10)).await;
        });
    });
    let payload = result.unwrap_err();
    let message = payload.downcast_ref::<String>().cloned()
        .or_else(|| payload.downcast_ref::<&str>().map(|s| s.to_string()))
        .unwrap();
    println!("{message}");
}

Output

there is no reactor running, must be called from the context of a Tokio 1.x runtime

A Tokio runtime supplies the executor, the timer driver, and the I/O driver together. The multi-thread runtime runs many tasks with the same wake-and-requeue idea as Example 5, using several worker threads.

Interview points

poll may be called any number of times, and must not block. A Pending result is a promise that a wake will follow.
A future that returns Pending without registering its waker is never polled again. The task hangs with no error.
Store the latest waker on every poll. The polling task can change between calls. Waker::will_wake can skip a clone when the waker is unchanged.
Check the state and register the waker atomically (one lock), or register first and check second, to avoid a lost wakeup.
Pin<&mut T> promises the value is not moved until dropped. It matters for self-referential state machines. Unpin types ignore it.
Box::pin puts a future on the heap. pin! pins it on the stack.
An executor is a loop of "pop ready task, poll once". The waker's job is to push the task back on the run queue.
Cooperative scheduling: a task that never returns Pending blocks its thread and every other task on it.

Summary

A Waker signals the executor that a task should be polled again. A Context passes it into poll.
The Wake trait builds a waker from an Arc<T>. RawWaker is the unsafe, lower-level form.
block_on polls one future and parks the thread while it is pending. The waker unparks the thread.
A timer future stores the waker, and a helper thread sets a flag and calls the waker.
A run-queue executor holds tasks as Arc<Task> values, and Task::wake sends the task back to the queue.
A Pending result without a stored waker is a missed wakeup. The flag check and the waker store share one lock.
Runtime-specific futures such as tokio::time::sleep need their runtime's drivers and panic outside it.