The Reactor, the Timer Wheel, and Task Layout

The scheduler from Runtime Builder and the Scheduler runs tasks that are ready. This lesson covers what makes them ready: the I/O driver (the reactor), which waits on the operating system for socket events, and the time driver, which fires timers. It ends with the memory layout of a task, which explains what a Waker and a JoinHandle point to. Constants are quoted from Tokio 1.53.1 and mio 1.2.3; paths are relative to each crate's src/ in ~/.cargo/registry/src/*/.

The I/O driver

Tokio's I/O driver is built on mio, a crate that wraps the OS readiness APIs: epoll on Linux, kqueue on macOS and BSD, and an IOCP-based emulation on Windows. A readiness API tells a program when a non-blocking socket can be read or written without blocking, rather than doing the read itself.

When a TcpStream is created, Tokio sets the socket to non-blocking and registers it with mio. The registration token is the address of a ScheduledIo struct owned by the driver (runtime/io/scheduled_io.rs). Two tokens are reserved in runtime/io/driver.rs: TOKEN_WAKEUP = Token(0) (used to interrupt a sleeping driver) and TOKEN_SIGNAL = Token(1) (Unix signals).

mio registers every source as edge-triggered: mio/src/sys/unix/selector/epoll.rs starts the interest mask with EPOLLET. Edge-triggered means the OS reports "readable" once when the state changes, not on every poll while data remains. Tokio therefore stores readiness itself, in one AtomicUsize per ScheduledIo:

(READINESS, TICK, SHUTDOWN in runtime/io/scheduled_io.rs.)

How a read parks and resumes

TcpStream::poll_read calls Registration::poll_io (runtime/io/registration.rs), which loops:

1. poll_ready: spend one unit of coop budget, then check the cached readiness. If the resource is not readable, store the task's waker in ScheduledIo and return Pending.
2. Readiness is set: call the non-blocking read syscall.
3. Ok(n): return it. Err(WouldBlock): the cached readiness was stale, so clear_readiness removes it and the loop returns to step 1, which now parks the task.

clear_readiness passes the tick at which the readiness was observed. If the driver set new readiness at a later tick in the meantime, the clear is ignored, so an event that arrived between the syscall and the clear is not lost.

On the other side, one worker at a time runs the driver's turn (runtime/io/driver.rs): it calls mio::Poll::poll with room for max_io_events_per_tick events (1024 by default), and for each event it ORs the readiness into the ScheduledIo and wakes the stored wakers. A wake from the driver schedules the task through the scheduler rules of the previous lesson.

Example 1: AsyncFd for your own file descriptor

tokio::io::unix::AsyncFd registers any type that implements AsRawFd with the reactor. It does no I/O itself. readable().await waits for the readiness bit, and guard.try_io runs your non-blocking call and clears readiness when it returns WouldBlock, exactly like steps 2 and 3 above.

use std::io::{Read, Write};
use std::os::unix::net::UnixStream;
use tokio::io::unix::AsyncFd;

#[tokio::main]
async fn main() -> std::io::Result<()> {
    let (a, mut b) = UnixStream::pair()?;
    a.set_nonblocking(true)?;
    let a = AsyncFd::new(a)?;

    tokio::spawn(async move {
        tokio::time::sleep(std::time::Duration::from_millis(10)).await;
        b.write_all(b"hello").unwrap();
    });

    let mut buf = [0u8; 16];
    loop {
        let mut guard = a.readable().await?;
        match guard.try_io(|inner| inner.get_ref().read(&mut buf)) {
            Ok(result) => {
                let n = result?;
                println!("read {n} bytes: {}", String::from_utf8_lossy(&buf[..n]));
                break;
            }
            Err(_would_block) => continue,
        }
    }
    Ok(())
}

read 5 bytes: hello

The file descriptor must be non-blocking; otherwise read would block the worker instead of returning WouldBlock. AsyncFd deregisters the descriptor when dropped, before the inner value closes it. It is Unix-only and needs the net feature plus enable_io.

The driver stack and parking

A runtime builds its drivers as a stack (runtime/driver.rs):

enable_all() turns on both enable_io and enable_time. When a worker has no tasks, it parks through the stack: the time driver finds the nearest timer deadline and parks the I/O driver with that timeout, so the thread sleeps inside epoll_wait until either a socket event or the next timer. To wake a parked driver from another thread (a task was spawned from main, for example), Tokio triggers the waker registered under TOKEN_WAKEUP.

The timer wheel

The time driver stores every active timer in a hierarchical timing wheel: an array of slots where each slot holds the timers that expire in one time range (runtime/time/wheel/).

Level n has 64 slots of 64ⁿ ms each: 1 ms, 64 ms, about 4 s, 4 min, 4.6 h, and 12 days. Each level covers one slot of the level above. A timer is placed in the level chosen by the highest bit in which its deadline differs from the current time (level_for). Each level keeps a 64-bit occupied mask, so finding the next non-empty slot is a bit scan, not a loop. When time reaches a slot above level 0, its timers are moved down to finer levels; at level 0 they fire and wake their tasks. Insert and cancel are O(1).

A Sleep does not touch the wheel when it is created. Its first poll spends one unit of coop budget, creates the timer entry, and inserts it; dropping the Sleep removes it. reset moves an entry to a new deadline.

Example 2: Timer placement

The function below repeats the arithmetic of level_for in runtime/time/wheel/mod.rs for a wheel whose elapsed time is 0. The paused clock (from Time: sleep, timeout, and interval) confirms that each sleep ends exactly at its deadline.

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

fn level_for(elapsed: u64, when: u64) -> usize {
    const SLOT_MASK: u64 = (1 << 6) - 1;
    const MAX_DURATION: u64 = (1 << 36) - 1;
    let mut masked = elapsed ^ when | SLOT_MASK;
    if masked >= MAX_DURATION {
        masked = MAX_DURATION - 1;
    }
    let significant = 63 - masked.leading_zeros() as usize;
    significant / 6
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let durations_ms = [10u64, 100, 5_000, 600_000, 259_200_000, 2_592_000_000];
    for ms in durations_ms {
        println!("{ms:>13} ms -> level {}", level_for(0, ms));
    }

    let start = Instant::now();
    sleep(Duration::from_millis(600_000)).await;
    println!("elapsed: {:?}", start.elapsed());
}

           10 ms -> level 0
          100 ms -> level 1
         5000 ms -> level 2
       600000 ms -> level 3
    259200000 ms -> level 4
   2592000000 ms -> level 5
elapsed: 600s

Ten minutes lands on level 3 (slots of about 4 minutes). As the clock approaches the deadline, the entry cascades through levels 2, 1, and 0 before it fires. Because deadlines round up to the next whole millisecond, a sleep never ends before its deadline and can end up to about 1 ms after it.

Anatomy of a task

tokio::spawn allocates one heap block per task, a Cell (runtime/task/core.rs) with three parts:

The future and its output share the same Stage memory. The cell is aligned to 128 bytes on x86_64 and aarch64 (#[repr(align(128))]) so two tasks never share a cache-line pair.

The state word (runtime/task/state.rs) is one AtomicUsize:

A new task starts with three references (the owned-tasks list, the first Notified sent to the scheduler, and the JoinHandle) and with NOTIFIED | JOIN_INTEREST set (INITIAL_STATE). The memory is freed when the count reaches zero, whichever holder is last.

The vtable (runtime/task/raw.rs) holds functions generated for the concrete future and scheduler types: poll, schedule, dealloc, try_read_output, drop_join_handle_slow, drop_abort_handle, and shutdown, plus field offsets. Every other type (RawTask, JoinHandle, Waker) is a pointer to the Header, so they are all one word and need no generics.

The waker (runtime/task/waker.rs) is a RawWaker whose data pointer is the Header and whose vtable is WAKER_VTABLE (clone_waker, wake_by_val, wake_by_ref, drop_waker). Cloning increments the reference count. Waking runs transition_to_notified_*:

• task idle and not notified: set NOTIFIED, add a reference, call the scheduler's schedule;
• task already notified, or complete: do nothing;
• task running: set NOTIFIED only; the polling thread reschedules it when poll returns Pending.

Example 3: Many wakes, one poll

Because of the NOTIFIED bit, waking a task five times before it runs puts it in the run queue once.

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

struct Watched {
    polls: Arc<AtomicUsize>,
    waker: Arc<Mutex<Option<Waker>>>,
    done: Arc<AtomicBool>,
}

impl Future for Watched {
    type Output = ();
    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
        self.polls.fetch_add(1, Ordering::SeqCst);
        if self.done.load(Ordering::SeqCst) {
            return Poll::Ready(());
        }
        *self.waker.lock().unwrap() = Some(cx.waker().clone());
        Poll::Pending
    }
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let polls = Arc::new(AtomicUsize::new(0));
    let waker = Arc::new(Mutex::new(None::<Waker>));
    let done = Arc::new(AtomicBool::new(false));

    let target = tokio::spawn(Watched { polls: polls.clone(), waker: waker.clone(), done: done.clone() });
    tokio::spawn(async move {
        let w = waker.lock().unwrap().take().unwrap();
        for _ in 0..5 {
            w.wake_by_ref();
        }
        done.store(true, Ordering::SeqCst);
    });

    target.await.unwrap();
    println!("wakes: 5, polls: {}", polls.load(Ordering::SeqCst));
}

wakes: 5, polls: 2

On the current-thread runtime the target runs first (FIFO), stores its waker, and returns Pending. The second task wakes it five times: the first wake sets NOTIFIED and queues it, the other four see NOTIFIED and do nothing. The target is polled a second time and completes.

Future size and boxing

A task's Core stores the future inline, so the task allocation is at least the size of the future. The size of an async block is the size of its largest set of variables that are alive at the same .await, plus the state tag (Futures and async/await). A large array held across an .await makes every task of that type large.

spawn, block_on, and spawn_blocking move the future or closure by value several times before it reaches the heap. To avoid a stack overflow, Tokio boxes it first when it is larger than BOX_FUTURE_THRESHOLD: 2048 bytes in debug builds and 16384 bytes in release builds (runtime/mod.rs; checked in task/spawn.rs, runtime/runtime.rs, task/local.rs, runtime/blocking/pool.rs).

Example 4: Measuring and shrinking a future

use std::mem::size_of_val;

#[tokio::main]
async fn main() {
    let small = async {
        tokio::task::yield_now().await;
        1
    };
    let big = async {
        let buf = [0u8; 4096];
        tokio::task::yield_now().await;
        buf.len()
    };
    println!("small under 64 bytes: {}", size_of_val(&small) < 64);
    println!("big at least 4096 bytes: {}", size_of_val(&big) >= 4096);

    let boxed = Box::pin(big);
    println!("boxed: {} bytes", size_of_val(&boxed));
    println!("result: {}", tokio::spawn(boxed).await.unwrap());
}

small under 64 bytes: true
big at least 4096 bytes: true
boxed: 8 bytes
result: 4096

buf is alive across the .await, so it is part of the state machine. Box::pin moves it to the heap; the future that spawn receives is now one pointer (8 bytes on a 64-bit target). The alternatives are to allocate the buffer inside the block (vec![0u8; 4096], 24 bytes inline) or to end its scope before the .await. Exact sizes depend on the compiler version, so tests compare against bounds, not fixed numbers.

The multi-thread runtime requires Send futures because a stolen task is polled on another thread, and everything alive across an .await lives inside the task cell that moves. Spawning costs one allocation of the cell, an insert into the runtime's owned-tasks list (a set of mutex-protected lists), and a queue push.

Interview points

• The I/O driver waits on mio (epoll/kqueue) and registers sources edge-triggered (EPOLLET). Tokio caches readiness per ScheduledIo and clears it only when a syscall returns WouldBlock.
• A read: check readiness, try the syscall, on WouldBlock clear readiness and park with the waker stored in ScheduledIo. A driver tick counter prevents clearing newer readiness.
• AsyncFd registers any non-blocking Unix fd; readable().await then guard.try_io(...).
• Parking a worker means waiting in epoll_wait with a timeout equal to the next timer deadline.
• Timer wheel: 6 levels × 64 slots, 1 ms resolution, max about 2.2 years; deadlines round up; Sleep registers on first poll.
• A task is one allocation: header (state, vtable), core (future or output), trailer (join waker). Wakers and JoinHandles are pointers to the header plus a reference count.
• NOTIFIED coalesces repeated wakes into one queue entry.
• Futures larger than 2048 bytes (debug) or 16384 bytes (release) are boxed by spawn; Box::pin large futures yourself to keep tasks small.

Summary

• mio provides readiness; Tokio stores it in ScheduledIo as 16 readiness bits, a 15-bit tick, and a shutdown bit.
• WouldBlock clears readiness and parks the task; a driver event sets readiness and wakes it.
• The time driver sits on top of the I/O driver; enable_all enables both.
• Timers live in a 6-level, 64-slot wheel with 1 ms resolution and are inserted on first poll.
• The task state word holds RUNNING, COMPLETE, NOTIFIED, JOIN_INTEREST, JOIN_WAKER, CANCELLED, and the reference count.
• A new task has three references; the last holder frees the cell.
• size_of_val measures a future; Box::pin reduces it to one pointer.