Async Rust from the executor up

Chapter 20’s echo server gave every client a thread, and chapter 21 parsed HTTP on top of it. That design works for tens or hundreds of clients. It stops working when a program holds thousands of mostly idle connections.

This chapter builds the runtime that replaces one thread per connection with one thread and many waiting tasks. It starts at the bottom, with the single method a Future has, and works up to a client on Tokio.

22.1 Why a thread per connection does not scale

A spawned thread in Rust is given a stack of 2 MiB by default. The stack stays reserved for as long as the thread lives, including all the time the thread spends parked in read. Ten thousand idle connections held as threads reserve about 20 GB of address space before any work happens.

The alternative is to keep one thread and represent each connection as a future. A future is a value that may not have finished yet. While it waits, it is not running on a thread. It is a small object that records where it stopped and what it was doing.

A task is a future a runtime has taken responsibility for. The runtime will poll the future again when there is a reason to.

A suspended future holds only the values alive across its pause points. An idle connection can be a few hundred bytes instead of two megabytes. Ten thousand of them fit in a few megabytes rather than tens of gigabytes. Figure 22.1 contrasts the two arrangements.

The cost of this saving is that waiting has to be written as a state machine. async and .await are the notation that lets the state machine still look like ordinary code.

You can now work out what a thread-per-connection design costs, and name what a future stores instead.

22.2 The contract

The chapter starts with the one method a Future has. A type that implements Future answers a single question: has the work finished?

fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output>;

poll returns Poll::Ready(value) when the work is done. It returns Poll::Pending when the work is not finished yet.

Pending carries an obligation. Before returning it, the future must arrange for cx.waker() to be called once progress is possible. The waker is a handle the executor gives to the future. Calling it tells the executor that this task should be polled again.

The executor has the matching obligation. It polls the task again after a wake. Until a wake arrives it does not poll the task, and it does not spin. Figure 22.2 draws both sides.

Two rules follow from the contract. Polling a future that is not ready is always safe: it returns Pending again, so a spurious wake costs one wasted poll. Forgetting to arrange a wake is not safe: the task returns Pending, nothing calls its waker, and the task never runs again.

One part of the signature is explained in section 22.6: poll takes Pin<&mut Self>, not &mut self. For now, read it as a promise that the future will not move in memory between polls.

You can now state what each side of the contract promises, and say which broken promise shows up as a hang.

22.3 The smallest executor

.await does not call the operating system. Something has to call poll and decide what a Pending result means. That something is an executor.

The smallest useful executor drives one future on the current thread. When the future is not ready, it parks the thread rather than polling again. Listing 22.1 is the whole of it.

Listing 22.1 block_on and its waker (lines 36 to 61). src/problems/async_mini.rs

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

pub fn block_on<F: Future>(future: F) -> F::Output {
    let mut future = Box::pin(future);
    let waker = Waker::from(Arc::new(ThreadWaker(thread::current())));
    let mut context = Context::from_waker(&waker);

    loop {
        match future.as_mut().poll(&mut context) {
            Poll::Ready(value) => return value,
            // `park` may return spuriously, so the loop re-polls either way.
            Poll::Pending => thread::park(),
        }
    }
}

/// Waker that unparks the thread which is driving the future.
struct ThreadWaker(thread::Thread);

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

    fn wake_by_ref(self: &Arc<Self>) {
        self.0.unpark();
    }
}

block_on does three things, in order:

1. Box::pin(future) moves the future to the heap and pins it there, so its address is the same for every poll.
2. Waker::from(Arc::new(ThreadWaker(thread::current()))) builds a waker. ThreadWaker implements Wake, and its wake method unparks the thread driving the future. Waker::from is the safe way to build a waker without writing a raw vtable.
3. The loop polls. Ready returns the value. Pending calls thread::park(), which sleeps until unpark is called on this thread.

park can return without anyone calling unpark. The loop tolerates that by polling again: a second poll of a future that is not ready returns Pending a second time. Spurious wakes are harmless, which is the other half of the contract.

park and unpark also cover a race the loop depends on. If unpark runs before park, it leaves a token behind, and the next park returns at once.

You can now read block_on line by line and explain why an idle future costs no CPU.

22.4 Two futures written by hand

Almost every future has one of two shapes. The first re-arms itself and asks to be polled again. The second waits on an event outside the executor, and keeps the waker of whoever is waiting for it.

22.4.1 A future that yields to itself

YieldTimes returns Pending a set number of times before it completes. Nothing is blocking it. It is giving the executor a chance to run other tasks before it finishes its own work.

Listing 22.2 YieldTimes, a future that re-arms itself (lines 63 to 97). src/problems/async_mini.rs

/// A future that yields to the executor `times` times before completing.
///
/// Completes with the number of times it yielded, which makes it easy to assert
/// that a runtime actually suspended and resumed a task.
pub struct YieldTimes {
    remaining: usize,
    yielded: usize,
}

impl YieldTimes {
    pub fn new(times: usize) -> Self {
        Self {
            remaining: times,
            yielded: 0,
        }
    }
}

impl Future for YieldTimes {
    type Output = usize;

    fn poll(self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<usize> {
        // `YieldTimes` is `Unpin`, so a mutable reference is safe to take.
        let this = self.get_mut();
        if this.remaining == 0 {
            return Poll::Ready(this.yielded);
        }

        this.remaining -= 1;
        this.yielded += 1;
        // Ask to be polled again; an executor that ignores this hangs.
        context.waker().wake_by_ref();
        Poll::Pending
    }
}

YieldTimes holds two usize values, which makes it Unpin: safe to move even after a poll. self.get_mut() uses that to turn Pin<&mut Self> into &mut Self without unsafe.

One line asks for the next poll: context.waker().wake_by_ref(). It runs, then poll returns Pending. This is what tokio::task::yield_now does. YieldTimes completes with the number of times it yielded. A test can use that count to check that the executor suspended and resumed it.

22.4.2 A future that waits on a timer

Delay completes after a wall-clock duration. It is the second shape, and it is the one that looks like real I/O.

Its state is a ready flag and an optional stored Waker, shared with a background thread through Arc<Mutex<_>>.

Listing 22.3 Delay state and constructor (lines 99 to 137). src/problems/async_mini.rs

struct DelayState {
    ready: bool,
    waker: Option<Waker>,
}

/// A future that completes after a wall-clock duration.
///
/// A background thread sleeps and then wakes the task. Storing and waking the
/// [`Waker`] — rather than spinning in `poll` — is the whole point: `poll` never
/// blocks the executor thread.
pub struct Delay {
    state: Arc<Mutex<DelayState>>,
}

impl Delay {
    pub fn new(duration: Duration) -> Self {
        let state = Arc::new(Mutex::new(DelayState {
            ready: false,
            waker: None,
        }));

        let thread_state = Arc::clone(&state);
        thread::spawn(move || {
            thread::sleep(duration);
            let waker = {
                let mut guard = thread_state
                    .lock()
                    .expect("delay mutex poisoned");
                guard.ready = true;
                guard.waker.take()
            };
            if let Some(waker) = waker {
                waker.wake();
            }
        });

        Self { state }
    }
}

poll checks ready. While it is false, poll stores a clone of the current waker and returns Pending.

Listing 22.4 Delay as a future (lines 139 to 154). src/problems/async_mini.rs

impl Future for Delay {
    type Output = ();

    fn poll(self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<()> {
        let mut guard = self
            .state
            .lock()
            .expect("delay mutex poisoned");
        if guard.ready {
            Poll::Ready(())
        } else {
            guard.waker = Some(context.waker().clone());
            Poll::Pending
        }
    }
}

The background thread sleeps, sets ready, takes the waker out of the state, releases the lock, and calls wake(). Figure 22.3 shows the exchange.

Two details make it correct. poll stores the waker on every Pending, not only the first. A task can be polled with a different waker each time. A runtime may move the task to another thread, and only the latest waker reaches its current owner. The timer thread wakes after dropping the lock, so the woken executor does not immediately block on a mutex the waker still held.

A real runtime replaces one thread per timer with a single timer wheel. It replaces the thread per socket with the epoll loop from chapter 20. That loop’s reactor keeps wakers by file descriptor, and wakes them when epoll_wait reports the descriptor ready.

You can now write both shapes: a future that re-arms itself, and a future that keeps a waker for an outside event.

22.5 Waking is scheduling

block_on drives one future. A general executor drives many, and it shows that waking a task and scheduling it are the same operation.

A Task holds its future, a handle to the executor’s queue, and a completed flag. It also implements Wake. Waking a task pushes an Arc of the task back on the queue.

Listing 22.5 A task, and the Wake impl that reschedules it (lines 156 to 177). src/problems/async_mini.rs

struct Task {
    future: Mutex<Pin<Box<dyn Future<Output = ()> + Send>>>,
    queue: Arc<Mutex<VecDeque<Arc<Task>>>>,
    completed: AtomicBool,
}

impl Wake for Task {
    fn wake(self: Arc<Self>) {
        // Re-enqueue the task. Cloning the `Arc` is a refcount bump, not a deep copy.
        self.queue
            .lock()
            .expect("executor queue poisoned")
            .push_back(Arc::clone(&self));
    }

    fn wake_by_ref(self: &Arc<Self>) {
        self.queue
            .lock()
            .expect("executor queue poisoned")
            .push_back(Arc::clone(self));
    }
}

spawn wraps a future in a task and enqueues it.

Listing 22.6 The executor's queue, and spawn (lines 184 to 208). src/problems/async_mini.rs

#[derive(Default)]
pub struct MiniExecutor {
    queue: Arc<Mutex<VecDeque<Arc<Task>>>>,
}

impl MiniExecutor {
    pub fn new() -> Self {
        Self::default()
    }

    /// Add a task to the executor.
    pub fn spawn<F>(&self, future: F)
    where
        F: Future<Output = ()> + Send + 'static,
    {
        let task = Arc::new(Task {
            future: Mutex::new(Box::pin(future)),
            queue: Arc::clone(&self.queue),
            completed: AtomicBool::new(false),
        });
        self.queue
            .lock()
            .expect("executor queue poisoned")
            .push_back(task);
    }
    // ...
}

run pops tasks one at a time, builds a waker from the task itself, and polls the future. A Ready result marks the task completed, so a late wake is skipped by the completed check.

Listing 22.7 MiniExecutor::run (lines 210 to 245). src/problems/async_mini.rs

impl MiniExecutor {
    // ...
    /// Run until the queue is empty.
    ///
    /// A task that returns `Pending` without waking stays suspended, which is why
    /// this returns rather than blocks: a runtime with timers or I/O would also
    /// park on an event source.
    pub fn run(&self) {
        loop {
            let Some(task) = self
                .queue
                .lock()
                .expect("executor queue poisoned")
                .pop_front()
            else {
                return;
            };

            if task.completed.load(Ordering::Acquire) {
                continue;
            }

            let waker = Waker::from(Arc::clone(&task));
            let mut context = Context::from_waker(&waker);
            let mut future = task
                .future
                .lock()
                .expect("task mutex poisoned");
            if future
                .as_mut()
                .poll(&mut context)
                .is_ready()
            {
                task.completed
                    .store(true, Ordering::Release);
            }
        }
    }
}

run returns when the queue is empty. A task that returned Pending without waking is waiting on something outside this executor, and this executor has nothing to park on. A real runtime would park on its timer or its I/O reactor instead of returning.

The + Send bound on spawned futures comes from the waker. A waker may be sent to another thread, as Delay does, so the task it can reach must be safe to send too. This is the same bound tokio::spawn imposes. It is also why holding a std::sync::MutexGuard or an Rc across an .await in a spawned task is a compile error.

Listings 22.8 and 22.9 are the complete pair: the runtime module, and the program that drives it, with no runtime dependency.

Listing 22.8 The complete runtime, with tests. src/problems/async_mini.rs

//! A minimal async runtime, built from `Future`, `Waker`, and `Pin`.
//!
//! `.await` is not magic. A `Future` is a state machine with one method, `poll`,
//! that returns `Ready(value)` or `Pending`; when it returns `Pending` it must
//! arrange for the `Waker` in its `Context` to be called later. An executor is a
//! loop that polls tasks and parks until a waker fires.
//!
//! This module implements that loop twice, from scratch and without a runtime
//! dependency:
//!
//! - [`block_on`] drives one future on the current thread, parking on `Pending`;
//! - [`MiniExecutor`] queues independent tasks and re-enqueues them when woken.
//!
//! It is deliberately small. It makes three things concrete: why `poll` takes `Pin<&mut Self>`, what `Pending` promises, and where
//! `Send` matters for a multi-threaded runtime.

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

/// Block the current thread until `future` completes.
///
/// The future is pinned on the heap so its address is stable across `await`
/// points, and the thread parks whenever the future reports `Pending`. The waker
/// unparks it, so the loop polls again only when there is a reason to.
pub fn block_on<F: Future>(future: F) -> F::Output {
    let mut future = Box::pin(future);
    let waker = Waker::from(Arc::new(ThreadWaker(thread::current())));
    let mut context = Context::from_waker(&waker);

    loop {
        match future.as_mut().poll(&mut context) {
            Poll::Ready(value) => return value,
            // `park` may return spuriously, so the loop re-polls either way.
            Poll::Pending => thread::park(),
        }
    }
}

/// Waker that unparks the thread which is driving the future.
struct ThreadWaker(thread::Thread);

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

    fn wake_by_ref(self: &Arc<Self>) {
        self.0.unpark();
    }
}

/// A future that yields to the executor `times` times before completing.
///
/// Completes with the number of times it yielded, which makes it easy to assert
/// that a runtime actually suspended and resumed a task.
pub struct YieldTimes {
    remaining: usize,
    yielded: usize,
}

impl YieldTimes {
    pub fn new(times: usize) -> Self {
        Self {
            remaining: times,
            yielded: 0,
        }
    }
}

impl Future for YieldTimes {
    type Output = usize;

    fn poll(self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<usize> {
        // `YieldTimes` is `Unpin`, so a mutable reference is safe to take.
        let this = self.get_mut();
        if this.remaining == 0 {
            return Poll::Ready(this.yielded);
        }

        this.remaining -= 1;
        this.yielded += 1;
        // Ask to be polled again; an executor that ignores this hangs.
        context.waker().wake_by_ref();
        Poll::Pending
    }
}

struct DelayState {
    ready: bool,
    waker: Option<Waker>,
}

/// A future that completes after a wall-clock duration.
///
/// A background thread sleeps and then wakes the task. Storing and waking the
/// [`Waker`] — rather than spinning in `poll` — is the whole point: `poll` never
/// blocks the executor thread.
pub struct Delay {
    state: Arc<Mutex<DelayState>>,
}

impl Delay {
    pub fn new(duration: Duration) -> Self {
        let state = Arc::new(Mutex::new(DelayState {
            ready: false,
            waker: None,
        }));

        let thread_state = Arc::clone(&state);
        thread::spawn(move || {
            thread::sleep(duration);
            let waker = {
                let mut guard = thread_state
                    .lock()
                    .expect("delay mutex poisoned");
                guard.ready = true;
                guard.waker.take()
            };
            if let Some(waker) = waker {
                waker.wake();
            }
        });

        Self { state }
    }
}

impl Future for Delay {
    type Output = ();

    fn poll(self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<()> {
        let mut guard = self
            .state
            .lock()
            .expect("delay mutex poisoned");
        if guard.ready {
            Poll::Ready(())
        } else {
            guard.waker = Some(context.waker().clone());
            Poll::Pending
        }
    }
}

struct Task {
    future: Mutex<Pin<Box<dyn Future<Output = ()> + Send>>>,
    queue: Arc<Mutex<VecDeque<Arc<Task>>>>,
    completed: AtomicBool,
}

impl Wake for Task {
    fn wake(self: Arc<Self>) {
        // Re-enqueue the task. Cloning the `Arc` is a refcount bump, not a deep copy.
        self.queue
            .lock()
            .expect("executor queue poisoned")
            .push_back(Arc::clone(&self));
    }

    fn wake_by_ref(self: &Arc<Self>) {
        self.queue
            .lock()
            .expect("executor queue poisoned")
            .push_back(Arc::clone(self));
    }
}

/// A single-threaded, `Send`-only executor.
///
/// Real runtimes add work stealing, timers, and I/O readiness. This keeps only the
/// part that matters for explaining how `.await` makes progress: a queue of tasks
/// where waking a task puts it back on the queue.
#[derive(Default)]
pub struct MiniExecutor {
    queue: Arc<Mutex<VecDeque<Arc<Task>>>>,
}

impl MiniExecutor {
    pub fn new() -> Self {
        Self::default()
    }

    /// Add a task to the executor.
    pub fn spawn<F>(&self, future: F)
    where
        F: Future<Output = ()> + Send + 'static,
    {
        let task = Arc::new(Task {
            future: Mutex::new(Box::pin(future)),
            queue: Arc::clone(&self.queue),
            completed: AtomicBool::new(false),
        });
        self.queue
            .lock()
            .expect("executor queue poisoned")
            .push_back(task);
    }

    /// Run until the queue is empty.
    ///
    /// A task that returns `Pending` without waking stays suspended, which is why
    /// this returns rather than blocks: a runtime with timers or I/O would also
    /// park on an event source.
    pub fn run(&self) {
        loop {
            let Some(task) = self
                .queue
                .lock()
                .expect("executor queue poisoned")
                .pop_front()
            else {
                return;
            };

            if task.completed.load(Ordering::Acquire) {
                continue;
            }

            let waker = Waker::from(Arc::clone(&task));
            let mut context = Context::from_waker(&waker);
            let mut future = task
                .future
                .lock()
                .expect("task mutex poisoned");
            if future
                .as_mut()
                .poll(&mut context)
                .is_ready()
            {
                task.completed
                    .store(true, Ordering::Release);
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use std::{sync::atomic::AtomicUsize, time::Instant};

    use super::*;

    #[test]
    fn block_on_drives_a_hand_written_future() {
        assert_eq!(block_on(YieldTimes::new(3)), 3);
    }

    #[test]
    fn block_on_polls_an_async_block() {
        let total = block_on(async {
            let a = YieldTimes::new(1).await;
            let b = YieldTimes::new(2).await;
            a + b
        });
        assert_eq!(total, 3);
    }

    #[test]
    fn block_on_waits_for_a_delay() {
        let start = Instant::now();
        block_on(Delay::new(Duration::from_millis(30)));
        assert!(start.elapsed() >= Duration::from_millis(20));
    }

    #[test]
    fn executor_runs_spawned_tasks_to_completion() {
        let executor = MiniExecutor::new();
        let counter = Arc::new(AtomicUsize::new(0));

        for _ in 0..4 {
            let counter = Arc::clone(&counter);
            executor.spawn(async move {
                YieldTimes::new(2).await;
                counter.fetch_add(1, Ordering::SeqCst);
            });
        }

        executor.run();
        assert_eq!(counter.load(Ordering::SeqCst), 4);
    }

    #[test]
    fn executor_polls_an_immediately_ready_task() {
        let executor = MiniExecutor::new();
        let flag = Arc::new(AtomicBool::new(false));

        let task_flag = Arc::clone(&flag);
        executor.spawn(async move {
            task_flag.store(true, Ordering::SeqCst);
        });

        executor.run();
        assert!(flag.load(Ordering::SeqCst));
    }
}

Listing 22.9 The program that drives the runtime. src/bin/async_demo.rs

//! Runs the hand-written async runtime in [`systems_lab::problems::async_mini`].
//!
//! There is no Tokio here on purpose: the point is to show that `Future`, `Waker`,
//! `Poll`, and `Pin` are enough to build an executor, and that `.await` is just
//! sugar over a state machine that is polled.
//!
//! ```bash
//! cargo run --bin async_demo
//! ```

use std::{
    sync::{
        Arc,
        atomic::{AtomicUsize, Ordering},
    },
    time::{Duration, Instant},
};

use systems_lab::problems::async_mini::{Delay, MiniExecutor, YieldTimes, block_on};

fn main() {
    println!("== block_on drives a hand-written future ==");
    let yields = block_on(YieldTimes::new(3));
    println!("YieldTimes suspended and resumed {yields} time(s)");

    println!("\n== block_on drives an async block ==");
    let total = block_on(async {
        let first = YieldTimes::new(1).await;
        let second = YieldTimes::new(2).await;
        first + second
    });
    println!("async block finished with {total}");

    println!("\n== block_on parks the thread until the waker fires ==");
    let start = Instant::now();
    block_on(Delay::new(Duration::from_millis(50)));
    println!("Delay completed after {:?}", start.elapsed());

    println!("\n== MiniExecutor runs independent tasks ==");
    let executor = MiniExecutor::new();
    let completed = Arc::new(AtomicUsize::new(0));

    for id in 1..=3 {
        let completed = Arc::clone(&completed);
        executor.spawn(async move {
            YieldTimes::new(id).await;
            completed.fetch_add(1, Ordering::SeqCst);
            println!("task {id} completed after {id} yield(s)");
        });
    }

    executor.run();
    println!("{} task(s) completed", completed.load(Ordering::SeqCst));
}

$ cargo run --bin async_demo
== block_on drives a hand-written future ==
YieldTimes suspended and resumed 3 time(s)

== block_on drives an async block ==
async block finished with 3

== block_on parks the thread until the waker fires ==
Delay completed after 52.083242ms

== MiniExecutor runs independent tasks ==
task 1 completed after 1 yield(s)
task 2 completed after 2 yield(s)
task 3 completed after 3 yield(s)
3 task(s) completed

The tasks finish in the order 1, 2, 3 because each yield sends its task to the back of the queue. Each task takes a turn, as in the round-robin scheduler of chapter 15, with a yield as the end of a time slice.

You can now follow a task from spawn to Ready, and say what wake does to the queue.

22.6 What async compiles to, and why Pin

An async block or function is notation. The compiler turns it into an anonymous type that implements Future. That type is usually an enum with one variant per suspension point, and each variant holds the values alive across that point.

The block used by the demo is:

async {
    let first = YieldTimes::new(1).await;
    let second = YieldTimes::new(2).await;
    first + second
}

The compiler writes an enum with four states: Start, Awaiting first, Awaiting second, and Done. Figure 22.4 shows the states and the values each one keeps.

Polling the block runs code until the next .await whose inner future returns Pending. It saves the state, stores the inner future and the live locals, and returns Pending too. A later poll resumes at the saved state rather than starting the block again.

first survives the second .await, so it is stored in the Awaiting second state. The size of a future is fixed once its type is known. The compiler sizes each state to hold the largest set of locals that can live there.

The remaining question is why poll takes Pin<&mut Self>. A state can hold a reference to another field of the same state machine. One example is a local buffer that is borrowed and still used across an .await. If the machine moved after the borrow was created, the reference would point at the old location. Figure 22.5 shows the two cases.

Pin<&mut Self> is the promise that the value will not move after the first poll. That is why block_on calls Box::pin and why MiniExecutor stores Pin<Box<...>>. Both keep the future at one address for its whole life.

Pinning is not a property of every type. A type whose values are safe to move even after a poll is Unpin, and for those types pinning adds no restriction. YieldTimes is Unpin because it holds two integers, so its poll can call self.get_mut() and work with &mut Self directly.

You can now read an async block as an enum of states, and explain what Pin stops the compiler from doing.

22.7 Two HTTP clients with caching, retries, and idempotency

The last two programs put a runtime to work. Both send POST /post with a JSON body to httpbin.org, a public echo service, with the idempotency key create-demo-001. Both have the same three layers: a cache, a retry policy, and a transport. Figure 22.6 shows the stacks.

Chapter 19 built retries and idempotency keys on their own. Here they meet a network call.

22.7.1 A blocking client over TcpStream

The blocking client is built from four parts: a request type, a cache, a retry policy, and the client that joins them. A request is an enum, and each variant carries exactly the fields its HTTP method needs:

use std::{
    collections::HashMap,
    io::{Read, Write},
    net::TcpStream,
    sync::Mutex,
    time::Duration,
};

type RuntimeError = Box<dyn std::error::Error>;

/// Request headers, as (name, value) pairs.
type RequestHeader = Vec<(String, String)>;

/// A request, with exactly the fields each method needs.
enum RequestType {
    Get {
        path: String,
        headers: RequestHeader,
    },
    Post {
        path: String,
        headers: RequestHeader,
        body: String,
    },
}

RuntimeError is a boxed error, so ? can return any error type from main. RequestHeader is an alias for a list of (name, value) pairs. A Get has no body and a Post has one, so a GET with a body cannot be built.

The cache keeps each successful response under its idempotency key:

/// Successful responses, stored by idempotency key.
struct IdemCache {
    store: Mutex<HashMap<String, String>>,
}

impl IdemCache {
    fn new() -> Self {
        Self {
            store: Mutex::new(HashMap::new()),
        }
    }

    fn get_or_insert<F>(&self, key: &str, f: F) -> Result<String, RuntimeError>
    where
        F: FnOnce() -> Result<String, RuntimeError>,
    {
        let mut store = self
            .store
            .lock()
            .map_err(|_| "lock poisoned")?;
        if let Some(v) = store.get(key) {
            return Ok(v.clone());
        }
        let v = f()?;
        store.insert(key.to_string(), v.clone());
        Ok(v)
    }
}

IdemCache is chapter 19’s Idempotent<K, V> store with String keys and values. get_or_insert returns the stored response if there is one. Otherwise it runs f, stores the result, and returns it. A failure is not stored, so a later call tries again. The lock is held while f runs, so two calls with one key cannot both send.

The retry policy runs an operation up to max_attempts times:

struct RetryPolicy {
    max_attempts: u32,
    base_delay: Duration,
    max_delay: Duration,
}

impl RetryPolicy {
    fn new() -> Self {
        Self {
            max_attempts: 3,
            base_delay: Duration::from_millis(100),
            max_delay: Duration::from_secs(5),
        }
    }

    fn execute<F, T>(&self, mut f: F) -> Result<T, RuntimeError>
    where
        F: FnMut() -> Result<T, RuntimeError>,
    {
        let mut attempt = 0;
        loop {
            attempt += 1;
            match f() {
                Ok(v) => return Ok(v),
                Err(e) if attempt >= self.max_attempts => return Err(e),
                Err(_) => {
                    let delay = (self.base_delay * 2u32.pow(attempt - 1)).min(self.max_delay);
                    std::thread::sleep(delay);
                }
            }
        }
    }
}

After each failure except the last, execute sleeps for base × 2^(attempt - 1), capped at max_delay: 100 ms after the first failure, 200 ms after the second. F is FnMut, not FnOnce, because the loop calls it more than once.

The client holds the server’s address and one of each layer. execute joins the layers in one line:

/// An HTTP/1.1 client with a response cache and a retry policy.
struct HttpClient {
    host: String,
    port: u16,
    cache: IdemCache,
    retry: RetryPolicy,
}

impl HttpClient {
    fn new(host: &str, port: u16) -> Self {
        Self {
            host: host.to_string(),
            port,
            cache: IdemCache::new(),
            retry: RetryPolicy::new(),
        }
    }

    fn execute(&self, key: &str, req: &RequestType) -> Result<String, RuntimeError> {
        self.cache
            .get_or_insert(key, || self.retry.execute(|| self.send(req)))
    }
    // ...
}

The cache calls the retry policy only on a miss, and the retry policy calls send once per attempt.

send writes the request by hand and reads the response:

impl HttpClient {
    // ...
    fn send(&self, req: &RequestType) -> Result<String, RuntimeError> {
        let (method, path, headers, body) = match req {
            RequestType::Get { path, headers } => ("GET", path, headers, None),
            RequestType::Post {
                path,
                headers,
                body,
            } => ("POST", path, headers, Some(body)),
        };

        let mut raw = format!(
            "{method} {path} HTTP/1.1\r\nHost: {}\r\nConnection: close\r\n",
            self.host
        );
        for (k, v) in headers {
            raw += &format!("{k}: {v}\r\n");
        }
        if let Some(b) = body {
            raw += &format!("Content-Length: {}\r\n", b.len());
        }
        raw += "\r\n";
        if let Some(b) = body {
            raw += b;
        }

        let mut stream = TcpStream::connect((self.host.as_str(), self.port))?;
        stream.write_all(raw.as_bytes())?;
        let mut resp = String::new();
        stream.read_to_string(&mut resp)?;
        Ok(resp
            .split_once("\r\n\r\n")
            .map(|(_, b)| b)
            .unwrap_or("")
            .to_string())
    }
}

The match turns either variant into the same four parts: the method name, the path, the headers, and an optional body. The request text then follows section 21.1. It is the request line, Host, Connection: close, and the caller’s headers. A body adds Content-Length, then the blank line, then the body. Connection: close lets read_to_string read the whole response, because the server closes the connection at the end. split_once("\r\n\r\n") drops the status line and the headers, and keeps the body.

main reads with a GET, then creates an item with a POST. Each request has its own idempotency key:

// --- The actual REST call ---

fn create_item(client: &HttpClient, item_name: &str) -> Result<String, RuntimeError> {
    client.execute(
        "create-demo-001",
        &RequestType::Post {
            path: "/post".into(),
            headers: vec![
                ("Content-Type".into(), "application/json".into()),
                ("Accept".into(), "application/json".into()),
            ],
            body: format!(r#"{{"name":"{item_name}"}}"#),
        },
    )
}

fn read_items(client: &HttpClient) -> Result<String, RuntimeError> {
    client.execute(
        "read-demo-001",
        &RequestType::Get {
            path: "/get".into(),
            headers: vec![("Accept".into(), "application/json".into())],
        },
    )
}

fn main() -> Result<(), RuntimeError> {
    let client = HttpClient::new("httpbin.org", 80);
    println!("read:\n{}", read_items(&client)?);
    println!("created:\n{}", create_item(&client, "demo")?);
    Ok(())
}

Listing 22.10 Cache, retry, and a hand-written HTTP/1.1 request. src/bin/fun_network_call.rs

use std::{
    collections::HashMap,
    io::{Read, Write},
    net::TcpStream,
    sync::Mutex,
    time::Duration,
};

type RuntimeError = Box<dyn std::error::Error>;

/// Request headers, as (name, value) pairs.
type RequestHeader = Vec<(String, String)>;

/// A request, with exactly the fields each method needs.
enum RequestType {
    Get {
        path: String,
        headers: RequestHeader,
    },
    Post {
        path: String,
        headers: RequestHeader,
        body: String,
    },
}

/// Successful responses, stored by idempotency key.
struct IdemCache {
    store: Mutex<HashMap<String, String>>,
}

impl IdemCache {
    fn new() -> Self {
        Self {
            store: Mutex::new(HashMap::new()),
        }
    }

    fn get_or_insert<F>(&self, key: &str, f: F) -> Result<String, RuntimeError>
    where
        F: FnOnce() -> Result<String, RuntimeError>,
    {
        let mut store = self
            .store
            .lock()
            .map_err(|_| "lock poisoned")?;
        if let Some(v) = store.get(key) {
            return Ok(v.clone());
        }
        let v = f()?;
        store.insert(key.to_string(), v.clone());
        Ok(v)
    }
}

struct RetryPolicy {
    max_attempts: u32,
    base_delay: Duration,
    max_delay: Duration,
}

impl RetryPolicy {
    fn new() -> Self {
        Self {
            max_attempts: 3,
            base_delay: Duration::from_millis(100),
            max_delay: Duration::from_secs(5),
        }
    }

    fn execute<F, T>(&self, mut f: F) -> Result<T, RuntimeError>
    where
        F: FnMut() -> Result<T, RuntimeError>,
    {
        let mut attempt = 0;
        loop {
            attempt += 1;
            match f() {
                Ok(v) => return Ok(v),
                Err(e) if attempt >= self.max_attempts => return Err(e),
                Err(_) => {
                    let delay = (self.base_delay * 2u32.pow(attempt - 1)).min(self.max_delay);
                    std::thread::sleep(delay);
                }
            }
        }
    }
}

/// An HTTP/1.1 client with a response cache and a retry policy.
struct HttpClient {
    host: String,
    port: u16,
    cache: IdemCache,
    retry: RetryPolicy,
}

impl HttpClient {
    fn new(host: &str, port: u16) -> Self {
        Self {
            host: host.to_string(),
            port,
            cache: IdemCache::new(),
            retry: RetryPolicy::new(),
        }
    }

    fn execute(&self, key: &str, req: &RequestType) -> Result<String, RuntimeError> {
        self.cache
            .get_or_insert(key, || self.retry.execute(|| self.send(req)))
    }

    fn send(&self, req: &RequestType) -> Result<String, RuntimeError> {
        let (method, path, headers, body) = match req {
            RequestType::Get { path, headers } => ("GET", path, headers, None),
            RequestType::Post {
                path,
                headers,
                body,
            } => ("POST", path, headers, Some(body)),
        };

        let mut raw = format!(
            "{method} {path} HTTP/1.1\r\nHost: {}\r\nConnection: close\r\n",
            self.host
        );
        for (k, v) in headers {
            raw += &format!("{k}: {v}\r\n");
        }
        if let Some(b) = body {
            raw += &format!("Content-Length: {}\r\n", b.len());
        }
        raw += "\r\n";
        if let Some(b) = body {
            raw += b;
        }

        let mut stream = TcpStream::connect((self.host.as_str(), self.port))?;
        stream.write_all(raw.as_bytes())?;
        let mut resp = String::new();
        stream.read_to_string(&mut resp)?;
        Ok(resp
            .split_once("\r\n\r\n")
            .map(|(_, b)| b)
            .unwrap_or("")
            .to_string())
    }
}

// --- The actual REST call ---

fn create_item(client: &HttpClient, item_name: &str) -> Result<String, RuntimeError> {
    client.execute(
        "create-demo-001",
        &RequestType::Post {
            path: "/post".into(),
            headers: vec![
                ("Content-Type".into(), "application/json".into()),
                ("Accept".into(), "application/json".into()),
            ],
            body: format!(r#"{{"name":"{item_name}"}}"#),
        },
    )
}

fn read_items(client: &HttpClient) -> Result<String, RuntimeError> {
    client.execute(
        "read-demo-001",
        &RequestType::Get {
            path: "/get".into(),
            headers: vec![("Accept".into(), "application/json".into())],
        },
    )
}

fn main() -> Result<(), RuntimeError> {
    let client = HttpClient::new("httpbin.org", 80);
    println!("read:\n{}", read_items(&client)?);
    println!("created:\n{}", create_item(&client, "demo")?);
    Ok(())
}

Two gaps separate this from a client you would ship, and both are visible in the code:

• The status code is never read. send returns whatever follows the first blank line. A 500 error page is returned as success, cached under the idempotency key, and never retried. Only transport errors reach the retry loop.
• Everything is retried. A policy should ask the error, as Retryable does in chapter 19. Here a DNS failure for a misspelled host is retried three times.

22.7.2 An async client with reqwest

The async client keeps the same layers, on Tokio and the reqwest crate. It also reads the status code, and decides from it whether to retry. Its settings come first:

use std::{collections::HashMap, sync::Mutex, time::Duration};

use reqwest::{Client, Method, StatusCode};
use serde_json::Value;

type Error = Box<dyn std::error::Error + Send + Sync>;

const RETRY: [StatusCode; 5] = [
    StatusCode::TOO_MANY_REQUESTS,
    StatusCode::INTERNAL_SERVER_ERROR,
    StatusCode::BAD_GATEWAY,
    StatusCode::SERVICE_UNAVAILABLE,
    StatusCode::GATEWAY_TIMEOUT,
];

struct Config {
    base: String,
    max_attempts: u32,
    base_delay: Duration,
    max_delay: Duration,
    timeout: Duration,
}

struct Api {
    http: Client,
    cache: Mutex<HashMap<String, String>>,
    cfg: Config,
}

RETRY lists the statuses to retry, where the server may succeed later: 429, Too Many Requests, and four 5xx codes. Config holds the base URL, the retry settings, and a timeout. Api holds a reqwest::Client, the cache, and the configuration. type Error = Box<dyn std::error::Error + Send + Sync> lets errors cross task boundaries, because a Send + Sync error can move between Tokio’s worker threads.

new builds the HTTP client, and two helpers read and write the cache:

impl Api {
    fn new(cfg: Config) -> Result<Self, Error> {
        Ok(Self {
            http: Client::builder()
                .timeout(cfg.timeout)
                .build()?,
            cache: Mutex::new(HashMap::new()),
            cfg,
        })
    }

    fn cached(&self, key: &str) -> Option<String> {
        self.cache
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .get(key)
            .cloned()
    }

    fn store(&self, key: &str, text: &str) {
        self.cache
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .insert(key.into(), text.into());
    }
    // ...
}

• Client::builder().timeout(cfg.timeout) bounds every request. A client without a timeout can wait forever on a server that accepted the connection and never answered.
• The cache is a std::sync::Mutex. That is allowed in async code because the guard is never held across an .await: cached and store lock, act, and return. The lock is taken with unwrap_or_else(|e| e.into_inner()), chapter 16’s ignore-the-poison choice. A HashMap insert cannot leave the map half-updated, so recovering the guard is safe here.

call checks the cache, then sends and retries:

impl Api {
    // ...
    async fn call(
        &self,
        key: &str,
        method: Method,
        path: &str,
        body: Option<&Value>,
    ) -> Result<String, Error> {
        if let Some(hit) = self.cached(key) {
            return Ok(hit);
        }

        let url = format!("{}{}", self.cfg.base, path);
        let mut delay = self.cfg.base_delay;
        let mut attempt = 0;

        loop {
            attempt += 1;

            let req = self
                .http
                .request(method.clone(), &url)
                .header("Idempotency-Key", key)
                .header("Accept", "application/json");
            let req = match body {
                Some(b) => req.json(b),
                None => req,
            };

            let resp = req.send().await?;
            let status = resp.status();
            let text = resp.text().await?;

            match status {
                s if s.is_success() => {
                    self.store(key, &text);
                    return Ok(text);
                }
                s if RETRY.contains(&s) && attempt < self.cfg.max_attempts => {
                    tokio::time::sleep(delay).await;
                    delay = (delay * 2).min(self.cfg.max_delay);
                }
                s => return Err(format!("HTTP {s} after {attempt} attempts: {text}").into()),
            }
        }
    }
}

Each attempt builds a request with self.http.request(method, url). It adds two headers, and for a body calls req.json(b), which serializes the JSON value and sets Content-Type. The request carries Idempotency-Key: create-demo-001, so a server that honors the header can deduplicate a retried POST. The client-side cache gives the same guarantee within one process.

req.send().await? sends the request and waits for the response headers. resp.text().await? waits for the body. The match on the status then has three arms:

• A success is stored in the cache and returned.
• A status in RETRY, with attempts left, sleeps and doubles the delay.
• Anything else is returned as an error that names the status and the attempts.

tokio::time::sleep does not block the thread. It returns a future that registers with Tokio’s timer and returns Pending, which is the Delay of section 22.4.2 done properly. Other tasks run on the same thread while this one waits.

main builds the client and makes one call:

#[tokio::main]
async fn main() -> Result<(), Error> {
    let api = Api::new(Config {
        base: "https://httpbin.org".into(),
        max_attempts: 3,
        base_delay: Duration::from_millis(100),
        max_delay: Duration::from_secs(5),
        timeout: Duration::from_secs(15),
    })?;

    let body = serde_json::json!({ "name": "demo" });
    let out = api
        .call("create-demo-001", Method::POST, "/post", Some(&body))
        .await?;
    println!("created:\n{out}");
    Ok(())
}

#[tokio::main] starts a Tokio runtime and runs main as its first task, so main can be async. serde_json::json! builds a JSON value from JSON-like syntax.

Listing 22.11 The same layers on Tokio. src/bin/reqwest_and_tokio.rs

use std::{collections::HashMap, sync::Mutex, time::Duration};

use reqwest::{Client, Method, StatusCode};
use serde_json::Value;

type Error = Box<dyn std::error::Error + Send + Sync>;

const RETRY: [StatusCode; 5] = [
    StatusCode::TOO_MANY_REQUESTS,
    StatusCode::INTERNAL_SERVER_ERROR,
    StatusCode::BAD_GATEWAY,
    StatusCode::SERVICE_UNAVAILABLE,
    StatusCode::GATEWAY_TIMEOUT,
];

struct Config {
    base: String,
    max_attempts: u32,
    base_delay: Duration,
    max_delay: Duration,
    timeout: Duration,
}

struct Api {
    http: Client,
    cache: Mutex<HashMap<String, String>>,
    cfg: Config,
}

impl Api {
    fn new(cfg: Config) -> Result<Self, Error> {
        Ok(Self {
            http: Client::builder()
                .timeout(cfg.timeout)
                .build()?,
            cache: Mutex::new(HashMap::new()),
            cfg,
        })
    }

    fn cached(&self, key: &str) -> Option<String> {
        self.cache
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .get(key)
            .cloned()
    }

    fn store(&self, key: &str, text: &str) {
        self.cache
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .insert(key.into(), text.into());
    }

    async fn call(
        &self,
        key: &str,
        method: Method,
        path: &str,
        body: Option<&Value>,
    ) -> Result<String, Error> {
        if let Some(hit) = self.cached(key) {
            return Ok(hit);
        }

        let url = format!("{}{}", self.cfg.base, path);
        let mut delay = self.cfg.base_delay;
        let mut attempt = 0;

        loop {
            attempt += 1;

            let req = self
                .http
                .request(method.clone(), &url)
                .header("Idempotency-Key", key)
                .header("Accept", "application/json");
            let req = match body {
                Some(b) => req.json(b),
                None => req,
            };

            let resp = req.send().await?;
            let status = resp.status();
            let text = resp.text().await?;

            match status {
                s if s.is_success() => {
                    self.store(key, &text);
                    return Ok(text);
                }
                s if RETRY.contains(&s) && attempt < self.cfg.max_attempts => {
                    tokio::time::sleep(delay).await;
                    delay = (delay * 2).min(self.cfg.max_delay);
                }
                s => return Err(format!("HTTP {s} after {attempt} attempts: {text}").into()),
            }
        }
    }
}

#[tokio::main]
async fn main() -> Result<(), Error> {
    let api = Api::new(Config {
        base: "https://httpbin.org".into(),
        max_attempts: 3,
        base_delay: Duration::from_millis(100),
        max_delay: Duration::from_secs(5),
        timeout: Duration::from_secs(15),
    })?;

    let body = serde_json::json!({ "name": "demo" });
    let out = api
        .call("create-demo-001", Method::POST, "/post", Some(&body))
        .await?;
    println!("created:\n{out}");
    Ok(())
}

22.7.3 The two clients side by side

The asynchronous client has the opposite gap from the blocking one. It reads the status code, but a transport error from send().await? is not retried at all. The blocking client retries transport errors and never reads the status code. Merging the two means retrying both, with a policy that asks the error which kind it is.

Neither client adds jitter, and neither honors Retry-After. Chapter 19 shows both in its delay policy.

22.7.4 Where the merged client belongs

The repository has an empty crate named rust-async-http-examples, created for this. Its first program can be the client from section 22.7.2 moved into lib.rs, with the retry and cache layers exposed for testing. Its tests can then run against a local server such as chapter 17’s, instead of a public one.

22.8 Questions that come up

“What happens when a future returns Pending?” It has arranged for its waker to be called when it can make progress. The executor stops polling it and runs other tasks. When the waker fires, the executor polls it again.

“Why does poll take Pin<&mut Self>?” An async state machine can hold a reference into itself across an .await. Pinning guarantees the value will not move after the first poll, so that reference stays valid.

“Why must spawned futures be Send in Tokio?” The multi-threaded runtime may resume a task on a different worker thread after any .await. Everything the task holds across an .await must therefore be safe to move between threads.

“Can you call blocking code inside async?” Only briefly. A blocking call holds the worker thread, and every task scheduled on it waits. Use tokio::task::spawn_blocking for blocking work, and async versions of I/O and sleep.

“Is a std::sync::Mutex allowed in async code?” Yes, if the guard is dropped before the next .await. Holding it across an .await can block the worker or deadlock it. Use tokio::sync::Mutex when the lock must be held across an await.

Chapter 23 goes below the system calls these chapters made. It maps a file into memory and watches the page faults that load it. Then it sends a file to a socket without copying it through the program.

Exercises

1. Add a poll counter to block_on and print it. Check that the count matches the number of wakes for Delay, and that a spurious wake adds one poll without breaking anything.
2. Write a ReadyNow future that returns Ready on its first poll. Confirm that MiniExecutor::run completes it in one poll.
3. Change MiniExecutor::run to return the number of tasks it completed. Test it with tasks that yield different numbers of times.
4. Give Delay a Drop that sets a flag the timer thread checks, so a cancelled delay does not keep a thread asleep.
5. Merge the two HTTP clients. Retry a retryable status and a transport error, add jitter to the delay, and honor Retry-After when the server sends it.
6. Move the async client into rust-async-http-examples as a library, and test it against chapter 21’s local server.