Runtime Builder and the Scheduler

This lesson covers two layers. The first is the public configuration surface: runtime::Builder, Handle, LocalSet, and shutdown. The second is what the multi-thread scheduler does with a task after spawn: which queue it enters, when other workers steal it, and when it is forced to yield. Constants are quoted from Tokio 1.53.1; paths are relative to tokio-1.53.1/src/ in ~/.cargo/registry/src/*/.

Builder defaults

#[tokio::main] builds a runtime with default settings (Your First Tokio Program shows the expansion). runtime::Builder sets each option by hand. Both drivers start off: a builder without enable_time or enable_io produces a runtime that panics on the first sleep or socket.

max_blocking_threads is a limit on the blocking pool, which serves spawn_blocking, tokio::fs, and other blocking work (Blocking and CPU-Bound Work). Idle blocking threads exit after the keep-alive period, so a quiet service does not keep 512 threads alive.

Example 1: Building a runtime by hand

The builder below names its threads and counts thread starts and stops. The worker threads themselves are started on the blocking pool, so the callbacks run for workers and for blocking threads alike.

use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use tokio::runtime::Builder;

fn main() {
    let starts = Arc::new(AtomicUsize::new(0));
    let stops = Arc::new(AtomicUsize::new(0));
    let (s1, s2) = (starts.clone(), stops.clone());

    let rt = Builder::new_multi_thread()
        .worker_threads(2)
        .max_blocking_threads(4)
        .thread_name("my-worker")
        .on_thread_start(move || { s1.fetch_add(1, Ordering::SeqCst); })
        .on_thread_stop(move || { s2.fetch_add(1, Ordering::SeqCst); })
        .enable_all()
        .build()
        .unwrap();

    let m = rt.handle().metrics();
    println!("flavor: {:?}", rt.handle().runtime_flavor());
    println!("workers: {}", m.num_workers());

    let name = rt.block_on(async {
        tokio::spawn(async { std::thread::current().name().unwrap().to_string() })
            .await
            .unwrap()
    });
    println!("task ran on: {name}");

    let name = rt.block_on(async {
        tokio::task::spawn_blocking(|| std::thread::current().name().unwrap().to_string())
            .await
            .unwrap()
    });
    println!("blocking closure ran on: {name}");

    drop(rt);
    println!("starts: {}, stops: {}", starts.load(Ordering::SeqCst), stops.load(Ordering::SeqCst));
}

flavor: MultiThread
workers: 2
task ran on: my-worker
blocking closure ran on: my-worker
starts: 3, stops: 3

Three threads started: two workers and one blocking thread for spawn_blocking. drop(rt) shuts the runtime down and waits for every thread, so all three stop callbacks have run before the last line. runtime_flavor() returns a RuntimeFlavor enum, and metrics().num_workers() is one of the stable metrics.

Example 2: Handle, and the three common panics

A Handle is a cloneable reference to a runtime. Handle::current() returns the runtime of the calling thread and panics outside one. Handle::try_current() returns a Result instead. A Handle can spawn and block_on from any thread, which is how synchronous code hands work to a runtime it does not own.

use std::panic::{AssertUnwindSafe, catch_unwind};
use tokio::runtime::{Builder, Handle};

fn message(e: Box<dyn std::any::Any + Send>) -> String {
    e.downcast_ref::<String>().cloned()
        .or_else(|| e.downcast_ref::<&str>().map(|s| s.to_string()))
        .unwrap()
}

fn main() {
    println!("outside a runtime: {}", Handle::try_current().is_err());
    let e = catch_unwind(|| Handle::current()).unwrap_err();
    println!("1: {}", message(e));

    let rt = Builder::new_current_thread().build().unwrap();
    let e = catch_unwind(AssertUnwindSafe(|| {
        rt.block_on(async { tokio::time::sleep(std::time::Duration::from_millis(1)).await })
    })).unwrap_err();
    println!("2: {}", message(e));

    let rt = Builder::new_current_thread().build().unwrap();
    let e = catch_unwind(AssertUnwindSafe(|| {
        rt.block_on(async {
            let inner = Builder::new_current_thread().build().unwrap();
            inner.block_on(async {});
        })
    })).unwrap_err();
    println!("3: {}", message(e));

    let rt = Builder::new_multi_thread().worker_threads(1).build().unwrap();
    let handle = rt.handle().clone();
    let answer = std::thread::spawn(move || {
        let task = handle.spawn(async { 40 + 2 });
        handle.block_on(task).unwrap()
    }).join().unwrap();
    println!("from a plain thread: {answer}");
}

outside a runtime: true
1: there is no reactor running, must be called from the context of a Tokio 1.x runtime
2: A Tokio 1.x context was found, but timers are disabled. Call `enable_time` on the runtime builder to enable timers.
3: Cannot start a runtime from within a runtime. This happens because a function (like `block_on`) attempted to block the current thread while the thread is being used to drive asynchronous tasks.
from a plain thread: 42

The default panic hook also prints each message to stderr; only stdout is shown. The three panics map to three mistakes:

Several runtimes can exist in one process. Each has its own workers and drivers. A common layout is one runtime for request handling and a second, smaller one for background jobs, joined by channels.

Example 3: LocalSet for !Send futures

tokio::spawn requires Send (a type is Send when it may move to another thread), because a multi-thread worker may steal the task. Rc is not Send. A LocalSet is a set of tasks that always run on the thread that drives it, so task::spawn_local accepts !Send futures.

use std::cell::RefCell;
use std::rc::Rc;
use tokio::task::LocalSet;

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let log = Rc::new(RefCell::new(Vec::new()));
    let local = LocalSet::new();

    local.run_until(async {
        let mut handles = Vec::new();
        for i in 1..=3 {
            let log = log.clone();
            handles.push(tokio::task::spawn_local(async move {
                tokio::task::yield_now().await;
                log.borrow_mut().push(i);
            }));
        }
        for h in handles {
            h.await.unwrap();
        }
    }).await;

    println!("{:?}", log.borrow());
}

[1, 2, 3]

run_until drives the set until the given future finishes. spawn_local panics if called outside a LocalSet context. The trade-off: LocalSet tasks never move to another worker, so they gain no parallelism. Builder::build_local(LocalOptions::default()) creates a LocalRuntime, a current-thread runtime whose spawn_local works without a LocalSet.

Shutdown

Dropping a Runtime shuts it down: every task's future is dropped where it is suspended (its Drop code runs), then the call waits for the blocking pool. A spawn_blocking closure cannot be interrupted, so a stuck closure makes drop wait forever.

Graceful shutdown of application work (stop accepting, drain, deadline) is built on top of this and is covered in Graceful Shutdown.

The multi-thread scheduler

Each worker is an OS thread that owns a core: a local run queue, a LIFO slot, a tick counter, and statistics. The run loop is Context::run in runtime/scheduler/multi_thread/worker.rs. One tick is one pass of the loop, roughly one task poll.

Where a task goes when it becomes runnable (Handle::schedule_task in worker.rs):

• From a worker thread of the same runtime (a spawn or a wake inside a task): into that worker's LIFO slot. A task already in the slot moves to the back of the local queue.
• From a yield_now: to the back of the local queue, never the LIFO slot.
• From any other thread (the main thread, a blocking thread, a different runtime): into the global queue, and one parked worker is woken.

How a worker picks the next task (Core::next_task): every global_queue_interval ticks it checks the global queue first; otherwise it takes the LIFO slot, then the local queue, then a batch from the global queue. When all of those are empty it steals: it picks a random start worker and takes half of the first non-empty queue it finds. If that fails it checks the global queue once more, then parks (sleeps until another thread unparks it). The LIFO slot is never stolen.

The global queue interval self-tunes on the multi-thread runtime unless set by hand. runtime/scheduler/multi_thread/stats.rs keeps a moving average of poll time (TASK_POLL_TIME_EWMA_ALPHA = 0.1) and picks the interval so the global queue is checked about every 200 microseconds (TARGET_GLOBAL_QUEUE_INTERVAL), clamped to between 2 and 127 (MAX_TASKS_POLLED_PER_GLOBAL_QUEUE_INTERVAL).

The event interval (61) decides how often a busy worker polls the I/O and timer drivers without sleeping (Context::maintenance, which parks with a zero timeout). A worker with no tasks polls the drivers while it parks.

Example 4: The LIFO slot and the global queue, observed

With a single worker there are no stealers, so placement decides the order. A task spawns five children; the last one lands in the LIFO slot and runs first. On the current-thread scheduler the same code runs in FIFO order.

use tokio::runtime::Builder;

async fn parent(label: &'static str) {
    let mut handles = Vec::new();
    for i in 1..=5 {
        handles.push(tokio::spawn(async move { print!("{label}{i} "); }));
    }
    for h in handles {
        h.await.unwrap();
    }
    println!();
}

fn main() {
    let mt = Builder::new_multi_thread().worker_threads(1).build().unwrap();
    mt.block_on(async { tokio::spawn(parent("m")).await.unwrap() });

    let ct = Builder::new_current_thread().build().unwrap();
    ct.block_on(parent("c"));

    // Tasks spawned from a non-worker thread go to the global queue.
    let (started_tx, started_rx) = std::sync::mpsc::channel();
    let (release_tx, release_rx) = std::sync::mpsc::channel::<()>();
    let blocker = mt.spawn(async move {
        started_tx.send(()).unwrap();
        release_rx.recv().unwrap(); // blocks the only worker on purpose
    });
    started_rx.recv().unwrap();

    let handles: Vec<_> = (0..5).map(|i| mt.spawn(async move { i })).collect();
    println!("global queue depth: {}", mt.handle().metrics().global_queue_depth());

    release_tx.send(()).unwrap();
    mt.block_on(async {
        blocker.await.unwrap();
        for h in handles { h.await.unwrap(); }
    });
    println!("global queue depth: {}", mt.handle().metrics().global_queue_depth());
}

m5 m1 m2 m3 m4 
c1 c2 c3 c4 c5 
global queue depth: 5
global queue depth: 0

The parent is spawned onto the worker, so its children are scheduled locally: each new child takes the LIFO slot and pushes the previous one to the back of the local queue. After the parent parks on h.await, the worker runs the slot (m5) and then the queue in order. The LIFO slot exists for message passing: a task that sends a message and then waits usually wants the receiver to run next, while the data is still in the CPU cache. The limit of 3 LIFO polls in a row stops two tasks that wake each other from starving the rest of the queue.

In the second half, main is not a worker thread, so every mt.spawn goes to the global queue. The blocked worker cannot take them, and global_queue_depth() (a stable metric) reports 5.

Example 5: The coop budget forces a yield

A loop over a channel that always has data never returns Pending by itself, so it would hold the worker forever. Tokio gives each task poll a budget of 128. Tokio resources (channel receive, socket I/O, sleep, JoinHandle, semaphores) call coop::poll_proceed, which spends one unit; at zero it returns Pending and wakes the task at once, so the scheduler runs something else first.

use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use tokio::sync::mpsc;

async fn drain(mut rx: mpsc::Receiver<u32>, other_ran: Arc<AtomicBool>) -> Option<usize> {
    let mut n = 0;
    while rx.recv().await.is_some() {
        n += 1;
        if other_ran.load(Ordering::SeqCst) {
            return Some(n);
        }
    }
    None
}

async fn run(unconstrained: bool) -> Option<usize> {
    let (tx, rx) = mpsc::channel(1000);
    for i in 0..1000 {
        tx.try_send(i).unwrap();
    }
    drop(tx);
    let flag = Arc::new(AtomicBool::new(false));
    let consumer = if unconstrained {
        tokio::spawn(tokio::task::coop::unconstrained(drain(rx, flag.clone())))
    } else {
        tokio::spawn(drain(rx, flag.clone()))
    };
    tokio::spawn(async move { flag.store(true, Ordering::SeqCst) });
    consumer.await.unwrap()
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    println!("budgeted: {:?}", run(false).await);
    println!("unconstrained: {:?}", run(true).await);
}

budgeted: Some(129)
unconstrained: None

The consumer received 128 messages; the 129th recv found the budget empty and returned Pending. The other task ran, set the flag, and the next recv returned message 129. Wrapped in unconstrained, the consumer never yields and drains all 1000 messages before the other task gets a turn. The budget covers only Tokio resources: a loop of pure computation, or one over a std channel, spends no budget and needs yield_now or spawn_blocking (Blocking and CPU-Bound Work).

The current-thread scheduler

The current-thread scheduler has one VecDeque local queue and a global queue for tasks scheduled from other threads (runtime/scheduler/current_thread/mod.rs). There is no LIFO slot and no stealing. Every 31 ticks it checks the global queue first; otherwise it takes the local queue first. Tasks run only while some thread is inside block_on, because that thread is the only worker.

Interview points

• Builder defaults: one worker per CPU (or TOKIO_WORKER_THREADS), 512 blocking threads, 10 s keep-alive, thread name tokio-rt-worker, event interval 61, drivers off until enable_*.
• Handle::current() panics outside a runtime; try_current() does not. A Handle can spawn and block_on from any thread.
• block_on on a worker thread panics with "Cannot start a runtime from within a runtime".
• Placement: a wake or spawn from a worker goes to its LIFO slot; yield_now goes to the back of the local queue; anything from outside goes to the global queue.
• Local queue: 256 slots. Overflow moves half to the global queue. Stealers take half of a victim's queue. At most half the workers search at once. The LIFO slot is not stealable.
• The coop budget is 128 Tokio-resource operations per poll; when it runs out, the next operation returns Pending and the task is rescheduled.
• LocalSet runs !Send futures on one thread; shutdown_timeout bounds the wait for blocking threads on shutdown.

Summary

• Builder sets flavor, worker count, thread names and hooks, blocking-pool size, drivers, and fairness intervals.
• The time and I/O drivers are off by default in a hand-built runtime.
• Handle connects synchronous code and other threads to a runtime.
• Each worker has a 256-slot local queue and a one-task LIFO slot; the global queue receives remote spawns and overflow.
• Idle workers steal half of another worker's queue, then check the global queue, then park.
• The global queue is checked every global_queue_interval ticks (31 on current-thread, self-tuned on multi-thread); drivers are polled every 61 ticks.
• A task spends one unit of its 128-unit budget per Tokio resource operation; an empty budget forces a yield.