59. How async compiles: state machines, Pin, and a tiny executor

59.1 What the compiler generates

// fragment: what you write
async fn fetch_and_count(url: String) -> usize {
    let conn = connect(&url).await;
    let body = conn.read_all().await;
    body.len()
}

// fragment: roughly what the compiler generates (simplified)
enum FetchAndCount {
    Start { url: String },
    WaitingConnect { url: String, fut: ConnectFuture },
    WaitingRead { fut: ReadAllFuture },        // `url` is no longer live, so it is not stored
    Done,
}
impl Future for FetchAndCount {
    type Output = usize;
    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<usize> {
        // loop { match state { Start => create the connect future, move to WaitingConnect;
        //                      WaitingConnect => poll it; Pending => return Pending; Ready(conn) => ...
        //                      WaitingRead => poll it; Ready(body) => return Ready(body.len()) } }
    }
}

The size of the future is the size of its largest state. Variables that are alive across an .await are stored in it. Everything else stays on the stack during a poll call. This is why you can have a million tasks: each one is exactly as large as the state it needs.

59.2 Why Pin exists

A state can hold a reference to another variable in the same state:

// fragment
async fn read_header(sock: &mut TcpStream) {
    let mut buf = [0u8; 1024];
    sock.read(&mut buf).await;   // the read future holds &mut buf, and buf lives in the same state machine
}

If the future were moved in memory after that borrow was created, the reference would point at the old location. Pin<&mut Self> is a promise that the value will not move again once it has been polled. That is why poll takes self: Pin<&mut Self>, and why you see Box::pin(fut) or std::pin::pin!(fut) when you store or poll futures by hand. Types that are safe to move anyway, which is almost all of them, implement Unpin, and for those Pin changes nothing.

59.3 A complete executor in 40 lines

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

/// A waker that unparks the thread running block_on.
struct ThreadWaker(thread::Thread);
impl Wake for ThreadWaker {
    fn wake(self: Arc<Self>) { self.0.unpark(); }
}

fn block_on<F: Future>(fut: F) -> F::Output {
    let mut fut = std::pin::pin!(fut);                 // pin on the stack: it will never move again
    let waker = Waker::from(Arc::new(ThreadWaker(thread::current())));
    let mut cx = Context::from_waker(&waker);
    loop {
        match fut.as_mut().poll(&mut cx) {
            Poll::Ready(v) => return v,
            Poll::Pending => thread::park(),           // sleep until someone calls wake()
        }
    }
}

static POLLS: AtomicU32 = AtomicU32::new(0);

/// A timer future: the "reactor" is one helper thread per timer (Tokio uses one timer wheel instead).
struct Delay { deadline: Instant, waker: Option<Arc<Mutex<Waker>>> }

impl Future for Delay {
    type Output = &'static str;
    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<&'static str> {
        POLLS.fetch_add(1, Ordering::Relaxed);
        if Instant::now() >= self.deadline { return Poll::Ready("timer fired"); }
        match &self.waker {
            Some(w) => *w.lock().unwrap() = cx.waker().clone(), // always keep the LATEST waker
            None => {
                let w = Arc::new(Mutex::new(cx.waker().clone()));
                self.waker = Some(Arc::clone(&w));
                let deadline = self.deadline;
                thread::spawn(move || {
                    thread::sleep(deadline.saturating_duration_since(Instant::now()));
                    w.lock().unwrap().wake_by_ref();     // "the event happened: poll me again"
                });
            }
        }
        Poll::Pending
    }
}

fn main() {
    let t = Instant::now();
    let out = block_on(async {
        let a = Delay { deadline: Instant::now() + Duration::from_millis(50), waker: None }.await;
        let b = Delay { deadline: Instant::now() + Duration::from_millis(50), waker: None }.await;
        format!("{a}, {b}")
    });
    println!("{out} after {:?}, {} polls", t.elapsed(), POLLS.load(Ordering::Relaxed));
}

timer fired, timer fired after 103.1ms, 4 polls

Four polls: each Delay is polled once (it returns Pending and arms the timer) and once more after its wake-up. Nothing busy-waits. Tokio has the same structure: a run queue of tasks, wakers that push a task back onto the queue, and a reactor (epoll/kqueue plus a timer wheel) that calls the wakers. It adds multiple worker threads with work stealing, which is the same idea as Go's scheduler (Go §28).