Threads, Async, and Where Tokio Fits

This lesson explains the problem that async Rust solves, using only std threads and small programs whose output is the same on every run. Afterwards you can decide when threads are enough, when tasks on a runtime are the better tool, and what a runtime provides.

Example 1: A blocking call and a non-blocking call

A blocking call does not return until it has a result. While it waits, the operating system keeps the calling thread asleep, and that thread can do nothing else. A non-blocking call returns at once, with either a result or a report that nothing is ready.

Receiver::recv blocks. Receiver::try_recv does not.

use std::sync::mpsc;
use std::thread;

fn main() {
    let (tx, rx) = mpsc::channel();

    // Non-blocking: returns immediately, nothing has been sent yet.
    println!("try_recv: {:?}", rx.try_recv());

    let sender = thread::spawn(move || {
        tx.send("data").unwrap();
    });

    // Blocking: the main thread sleeps until a message arrives.
    println!("recv: {:?}", rx.recv());
    sender.join().unwrap();
}

try_recv: Err(Empty)
recv: Ok("data")

try_recv reports Empty and the program continues. recv parks the main thread until the helper thread sends. A thread parked in recv uses no processor time, but it also cannot serve any other work.

Example 2: A non-blocking socket says "not ready"

Sockets are blocking by default. set_nonblocking(true) switches a listener to non-blocking mode. Calling accept with no client waiting then fails with the error kind WouldBlock instead of sleeping.

use std::io::ErrorKind;
use std::net::TcpListener;

fn main() {
    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
    listener.set_nonblocking(true).unwrap();

    match listener.accept() {
        Ok(_) => println!("a client connected"),
        Err(e) if e.kind() == ErrorKind::WouldBlock => {
            println!("nothing to accept yet: {:?}", e.kind());
        }
        Err(e) => println!("real error: {e}"),
    }
}

nothing to accept yet: WouldBlock

WouldBlock is not a failure. It means "this call would have blocked; try again later". Async runtimes are built on this signal, and Example 11 shows the cost of using it naively.

Example 3: I/O-bound work spends its time waiting

Work is I/O-bound when the time it takes is dominated by waiting for something outside the processor: a network reply, a disk read, another thread. Work is CPU-bound when the time is dominated by computation.

Each of the 100 threads below waits for one number from its own channel and adds 1. Almost all of a thread's life is spent inside recv.

use std::sync::mpsc;
use std::thread;

fn main() {
    let mut senders = Vec::new();
    let mut workers = Vec::new();

    for _ in 0..100 {
        let (tx, rx) = mpsc::channel::<u64>();
        senders.push(tx);
        workers.push(thread::spawn(move || rx.recv().unwrap() + 1));
    }

    for (i, tx) in senders.iter().enumerate() {
        tx.send(i as u64).unwrap();
    }

    let total: u64 = workers.into_iter().map(|w| w.join().unwrap()).sum();
    println!("threads: {}, total: {total}", senders.len());
}

threads: 100, total: 5050

The total is the sum of 0 to 99 (4950) plus 100. Every thread is idle until its message arrives, so 100 threads do work that one thread could do if it had a way to wait on many channels at once. That observation drives the rest of the lesson.

Example 4: CPU-bound work benefits from more threads

CPU-bound work keeps a core busy, so the only way to finish sooner is to use more cores. Parallelism means several pieces of work running at the same instant on different cores.

This program adds up squares on one thread, then splits the range into four chunks with thread::scope (threads that may borrow local data and are joined before the scope ends).

use std::thread;

fn sum_squares(range: std::ops::Range<u64>) -> u64 {
    range.map(|n| n * n % 1_000_003).sum()
}

fn main() {
    let n = 4_000_000;
    let single = sum_squares(0..n);

    let chunk = n / 4;
    let parallel: u64 = thread::scope(|s| {
        let handles: Vec<_> = (0..4)
            .map(|i| s.spawn(move || sum_squares(i * chunk..(i + 1) * chunk)))
            .collect();
        handles.into_iter().map(|h| h.join().unwrap()).sum()
    });

    println!("single thread: {single}");
    println!("four threads:  {parallel}");
    println!("equal: {}", single == parallel);
}

single thread: 1999589998102
four threads:  1999589998102
equal: true

Both totals are equal. With four or more free cores, the four-thread version finishes sooner. Waiting on a network reply gains nothing from extra cores, because the reply arrives at the same time regardless of how many cores exist.

Example 5: Concurrency without parallelism

Concurrency means several pieces of work are in progress during the same period, whether or not any two run at the same instant. One thread can be concurrent by switching between jobs.

Each Job below does one step per call. A loop on a single thread gives each job a turn.

struct Job {
    name: &'static str,
    steps_left: u32,
}

impl Job {
    /// Runs one step. Returns true when the job is finished.
    fn step(&mut self) -> bool {
        self.steps_left -= 1;
        println!("{} did a step, {} left", self.name, self.steps_left);
        self.steps_left == 0
    }
}

fn main() {
    let mut jobs = vec![
        Job { name: "download", steps_left: 3 },
        Job { name: "compress", steps_left: 2 },
    ];

    while !jobs.is_empty() {
        jobs.retain_mut(|job| !job.step());
    }
    println!("all jobs finished on one thread");
}

download did a step, 2 left
compress did a step, 1 left
download did a step, 1 left
compress did a step, 0 left
download did a step, 0 left
all jobs finished on one thread

The two jobs are both in progress at once, and the output alternates between them, yet only one thread exists. This is the model behind async runtimes: many jobs, few threads, and a scheduler that switches between jobs at points the jobs choose.

Example 6: A blocked thread holds its slot

A thread that blocks is unavailable until the call returns. A fixed-size pool of threads therefore serves at most as many blocked requests as it has threads.

This pool has 2 worker threads and receives 4 requests. Each request waits for a release signal, as a request waiting for a slow database would.

use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Mutex, mpsc};
use std::thread;

fn main() {
    let queue: Arc<Mutex<Vec<u32>>> = Arc::new(Mutex::new(vec![4, 3, 2, 1]));
    let started = Arc::new(AtomicUsize::new(0));
    let finished = Arc::new(AtomicUsize::new(0));
    let (release_tx, release_rx) = mpsc::channel::<()>();
    let release_rx = Arc::new(Mutex::new(release_rx));

    let mut workers = Vec::new();
    for _ in 0..2 {
        let (queue, started, finished, release_rx) =
            (queue.clone(), started.clone(), finished.clone(), release_rx.clone());
        workers.push(thread::spawn(move || {
            loop {
                let job = queue.lock().unwrap().pop();
                let Some(_request) = job else { break };
                started.fetch_add(1, Ordering::SeqCst);
                // The request blocks here until it is released.
                let released = release_rx.lock().unwrap().recv();
                released.unwrap();
                finished.fetch_add(1, Ordering::SeqCst);
            }
        }));
    }

    while started.load(Ordering::SeqCst) < 2 {
        thread::yield_now();
    }
    println!("in progress: 2, still queued: {}", queue.lock().unwrap().len());

    for _ in 0..4 {
        release_tx.send(()).unwrap();
    }
    for w in workers {
        w.join().unwrap();
    }
    println!("finished: {}", finished.load(Ordering::SeqCst));
}

in progress: 2, still queued: 2
finished: 4

Two requests are in progress and two wait in the queue, because both workers are blocked. Adding threads raises the limit, but each thread has a cost, covered next.

Example 7: One thread per connection

The thread-per-connection design gives every client its own thread that blocks on that client's socket. The code is straightforward, because each thread can wait without affecting the others.

The program below starts 1000 threads and holds them on a Barrier, which releases its waiters only when all of them have arrived. Every thread passing the barrier proves that all 1000 existed at the same time.

use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Barrier};
use std::thread;

fn main() {
    let count = 1000;
    let barrier = Arc::new(Barrier::new(count));
    let arrived = Arc::new(AtomicUsize::new(0));

    let handles: Vec<_> = (0..count)
        .map(|_| {
            let barrier = barrier.clone();
            let arrived = arrived.clone();
            thread::spawn(move || {
                arrived.fetch_add(1, Ordering::SeqCst);
                barrier.wait();
                arrived.load(Ordering::SeqCst)
            })
        })
        .collect();

    let smallest_view = handles.into_iter().map(|h| h.join().unwrap()).min().unwrap();
    println!("every thread saw at least {smallest_view} threads alive");
}

every thread saw at least 1000 threads alive

One thousand simultaneous threads work on this machine. The next example shows why the number cannot grow without limit.

Example 8: What a thread costs

Every thread owns a stack, the memory region that holds its function calls and local variables. The standard library documents the default size: "Currently, it is 2 MiB on all Tier-1 platforms" (std::thread, section "Stack size"). The size can be set per thread with thread::Builder::stack_size. The main thread's stack size is set by the operating system, not by Rust.

The program below computes the reserved stack space of a large thread count, starts a thread with a 64 KiB stack, and then measures 200 parked threads by reading VmSize (reserved address space) and VmRSS (memory actually resident) from /proc/self/status. That file exists on Linux only.

use std::sync::{Arc, Barrier};
use std::thread;

fn depth(n: u32) -> u32 {
    if n == 0 { 0 } else { 1 + depth(n - 1) }
}

fn proc_kib(field: &str) -> u64 {
    let status = std::fs::read_to_string("/proc/self/status").unwrap();
    let line = status.lines().find(|l| l.starts_with(field)).unwrap();
    line.split_whitespace().nth(1).unwrap().parse().unwrap()
}

fn main() {
    const MIB: usize = 1024 * 1024;
    let threads = 10_000;
    println!(
        "{threads} threads x 2 MiB = {} MiB of stack address space",
        threads * 2 * MIB / MIB
    );

    let small = thread::Builder::new()
        .stack_size(64 * 1024)
        .spawn(|| depth(100))
        .unwrap();
    println!("recursion depth reached on a 64 KiB stack: {}", small.join().unwrap());

    let (virt_before, resident_before) = (proc_kib("VmSize:"), proc_kib("VmRSS:"));
    let n = 200;
    let barrier = Arc::new(Barrier::new(n + 1));
    let parked: Vec<_> = (0..n)
        .map(|_| {
            let barrier = barrier.clone();
            thread::spawn(move || {
                barrier.wait(); // all threads exist
                barrier.wait(); // released after measuring
            })
        })
        .collect();
    barrier.wait();
    let virt_per_thread = (proc_kib("VmSize:") - virt_before) / n as u64;
    let resident_per_thread = proc_kib("VmRSS:").saturating_sub(resident_before) / n as u64;
    println!("reserved per thread is at least 2 MiB: {}", virt_per_thread >= 2048);
    println!("resident per thread is under 256 KiB: {}", resident_per_thread < 256);
    barrier.wait();
    for t in parked {
        t.join().unwrap();
    }
}

10000 threads x 2 MiB = 20000 MiB of stack address space
recursion depth reached on a 64 KiB stack: 100
reserved per thread is at least 2 MiB: true
resident per thread is under 256 KiB: true

Ten thousand default stacks reserve 20,000 MiB of address space. The measurement shows the difference between reserved and used memory: each parked thread reserves at least its 2 MiB stack, while the operating system commits only the pages a thread touches, so resident memory per thread is far smaller. The reserved size still counts against limits such as a container's virtual memory limit, and a deep recursion needs the full stack. Beyond stack memory, each thread costs a kernel scheduling entry, and switching from one thread to another (a context switch) saves and restores processor state. Thousands of threads that wake at the same time make the scheduler do that work thousands of times.

Example 9: A task is a small value

A task in async Rust is a unit of work that a runtime schedules. It is an ordinary value in memory, not an operating system thread. An async fn compiles into such a value, and Futures and async/await explains how.

The idea can be shown without any async syntax. This struct is the hand-written state of a job that fetches, parses, and stores something.

use std::mem::size_of;

#[allow(dead_code)]
enum Stage {
    Fetch,
    Parse,
    Store,
    Done,
}

#[allow(dead_code)]
struct Request {
    id: u32,
    stage: Stage,
    bytes_seen: u64,
}

fn main() {
    println!("size of one job: {} bytes", size_of::<Request>());
    println!("jobs that fit in one 2 MiB stack: {}", 2 * 1024 * 1024 / size_of::<Request>());
}

size of one job: 16 bytes
jobs that fit in one 2 MiB stack: 131072

The value that stands for a job is 16 bytes. A thread reserves 2 MiB for the same purpose, which is 131,072 times more. A job that is waiting needs only its state, and only the job that is running needs a stack.

Example 10: Ten thousand jobs on one thread

Combining Examples 5 and 9: keep the jobs as small values in a Vec and let one thread give each a turn. Every job has 3 steps. The program runs 10,000 jobs to completion.

enum Stage {
    Fetch,
    Parse,
    Store,
    Done,
}

struct Request {
    stage: Stage,
}

impl Request {
    /// Advances one stage. Returns true when the request is done.
    fn step(&mut self) -> bool {
        self.stage = match self.stage {
            Stage::Fetch => Stage::Parse,
            Stage::Parse => Stage::Store,
            Stage::Store | Stage::Done => Stage::Done,
        };
        matches!(self.stage, Stage::Done)
    }
}

fn main() {
    let mut queue: Vec<Request> = (0..10_000).map(|_| Request { stage: Stage::Fetch }).collect();
    let mut steps = 0;
    let mut finished = 0;

    while !queue.is_empty() {
        queue.retain_mut(|request| {
            steps += 1;
            let done = request.step();
            if done {
                finished += 1;
            }
            !done
        });
    }
    println!("steps executed: {steps}");
    println!("requests finished: {finished}");
}

steps executed: 30000
requests finished: 10000

One thread finished 10,000 requests in 30,000 steps. A runtime does the same thing with two additions: jobs are written as async code instead of hand-written enums, and jobs that wait for I/O are set aside until the operating system reports that they can continue.

Example 11: Finding ready sockets

A job that waits for a network reply must not spin in a loop asking "is it ready?". Example 2 showed the only tool std offers for one socket: try, get WouldBlock, try again. The program below serves 3 clients from one thread by trying every socket in turn.

use std::io::{ErrorKind, Read, Write};
use std::net::{TcpListener, TcpStream};
use std::thread;

fn main() {
    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
    let addr = listener.local_addr().unwrap();
    listener.set_nonblocking(true).unwrap();

    let clients: Vec<_> = (0..3)
        .map(|i| {
            thread::spawn(move || {
                let mut stream = TcpStream::connect(addr).unwrap();
                stream.write_all(format!("hello {i}").as_bytes()).unwrap();
                stream.shutdown(std::net::Shutdown::Write).unwrap();
                let mut reply = String::new();
                stream.read_to_string(&mut reply).unwrap();
                reply
            })
        })
        .collect();

    let mut connections: Vec<(TcpStream, Vec<u8>)> = Vec::new();
    let mut served = 0;
    while served < 3 {
        // Try to accept a new connection.
        match listener.accept() {
            Ok((stream, _)) => {
                stream.set_nonblocking(true).unwrap();
                connections.push((stream, Vec::new()));
            }
            Err(e) if e.kind() == ErrorKind::WouldBlock => {}
            Err(e) => panic!("{e}"),
        }
        // Try every open connection.
        let mut i = 0;
        while i < connections.len() {
            let (stream, buffer) = &mut connections[i];
            let mut chunk = [0u8; 64];
            match stream.read(&mut chunk) {
                Ok(0) => {
                    let reply = format!("echo: {}", String::from_utf8_lossy(buffer));
                    stream.set_nonblocking(false).unwrap();
                    stream.write_all(reply.as_bytes()).unwrap();
                    connections.remove(i);
                    served += 1;
                    continue;
                }
                Ok(n) => buffer.extend_from_slice(&chunk[..n]),
                Err(e) if e.kind() == ErrorKind::WouldBlock => {}
                Err(e) => panic!("{e}"),
            }
            i += 1;
        }
        thread::yield_now();
    }

    let mut replies: Vec<String> = clients.into_iter().map(|c| c.join().unwrap()).collect();
    replies.sort();
    for reply in replies {
        println!("{reply}");
    }
}

echo: hello 0
echo: hello 1
echo: hello 2

One thread served three clients, and no thread blocked on a socket. The loop has a cost, though: it asks every socket on every pass, even when none is ready, and the number of questions grows with the number of connections. Operating systems provide calls that report which sockets are ready: epoll on Linux, kqueue on macOS and BSD, and IOCP on Windows. A program hands the system a list of sockets and sleeps until at least one is ready. The part of a runtime that does this is the reactor, and The Reactor and the Timer Wheel describes Tokio's.

The C10K problem

In 1999 Dan Kegel published a web page titled "The C10K problem": how a single server can handle ten thousand simultaneous client connections. With one thread per connection, ten thousand connections meant ten thousand stacks and ten thousand entries for the kernel scheduler. Readiness notification (epoll and its relatives) let one thread wait on all ten thousand sockets at once. Async runtimes package that technique behind async/await syntax.

Connection counts in use today are larger, and the same reasoning applies: the cost of an idle connection should be the size of its state, not the size of a thread.

Example 12: The same idea with Tokio

A runtime is a library that runs async code: it schedules tasks, waits for I/O and timers, and wakes tasks when they can continue. Tokio is a runtime for Rust. #[tokio::main] turns main into a normal function that starts a Tokio runtime and runs the async body on it; Your First Tokio Program explains what it expands to.

The program uses one more item not explained yet: tokio::spawn starts a task on the runtime and returns a handle whose .await produces the task's result (Spawning Tasks). It starts 10,000 tasks, each adding to a shared counter.

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

#[tokio::main]
async fn main() {
    let counter = Arc::new(AtomicUsize::new(0));
    let mut handles = Vec::new();

    for _ in 0..10_000 {
        let counter = counter.clone();
        handles.push(tokio::spawn(async move {
            counter.fetch_add(1, Ordering::SeqCst);
        }));
    }

    for handle in handles {
        handle.await.unwrap();
    }
    println!("tasks completed: {}", counter.load(Ordering::SeqCst));
}

tasks completed: 10000

Ten thousand tasks were created and finished. The same program with thread::spawn would create ten thousand operating system threads. The multi-thread runtime that #[tokio::main] starts uses a fixed number of worker threads and runs all tasks on them. Without configuration the count equals the number of CPU cores (std::thread::available_parallelism, read from tokio::runtime::Builder's source), unless the TOKIO_WORKER_THREADS environment variable is set.

What a runtime provides

The Rust language and std supply the pieces for writing async code: the Future trait, async/await syntax, and the types Waker and Context used to signal readiness. They do not supply anything that runs a future to completion, does I/O without blocking, or measures time. A runtime provides those:

Polling, Wakers, and a Tiny Executor builds an executor and a timer from scratch.

The Tokio crate ships all three, plus task-aware versions of sockets, files, channels, locks, and process handling. The Tokio README describes it as an event-driven, non-blocking I/O platform with a multithreaded, work-stealing task scheduler, a reactor backed by the operating system's event queue (epoll, kqueue, IOCP), and asynchronous TCP and UDP sockets.

Tokio compared with other options

Thread-per-core runtimes keep each task on one thread, so tasks can hold non-Send data. A work-stealing runtime such as Tokio moves tasks between threads to balance load, so tasks must be Send; Spawning Tasks explains that requirement. Libraries written against one runtime's I/O types generally need that runtime, which is one reason the size of Tokio's ecosystem (see The Tokio Ecosystem) matters when choosing.

Threads or async: criteria

Async code and CPU-bound code can share a program, but a task that computes for a long time without pausing holds its worker thread and delays every other task on it. Later lessons show how to detect and avoid that.

The Rust Programming course covers threads in Threads and async in Async Programming Fundamentals; both are optional further reading.

Interview points

• Blocking calls put the thread to sleep. Non-blocking calls return WouldBlock (or Pending in async) when nothing is ready.
• I/O-bound work is limited by waiting, and CPU-bound work is limited by cores. Threads help CPU-bound work. Async helps I/O-bound work with many concurrent waits.
• Concurrency is overlapping progress. Parallelism is simultaneous execution. One thread can be concurrent, but not parallel.
• A thread costs a stack (2 MiB by default on Tier-1 platforms), kernel bookkeeping, and context switches. A task costs the size of its state.
• The C10K problem asked how to serve ten thousand connections from one server. Readiness notification (epoll, kqueue, IOCP) is the answer that async runtimes build on.
• A runtime provides an executor, a reactor, and a timer. std provides none of them.
• Tokio's default runtime is multi-threaded and work-stealing, so spawned tasks must be Send.

Summary

• Blocking calls park the calling thread. Non-blocking calls return immediately and report "not ready".
• I/O-bound work spends its time waiting. CPU-bound work spends its time computing.
• Concurrency does not require parallelism. Round-robin over small jobs on one thread is concurrent.
• A blocked thread cannot serve other work, so a fixed pool serves at most as many blocked requests as it has threads.
• The default std thread stack is 2 MiB on Tier-1 platforms and can be changed with thread::Builder::stack_size.
• A task is a small value scheduled by a runtime. A thread is an operating system resource.
• A runtime combines an executor, a reactor, and a timer. Tokio provides all three.
• Thread-per-core runtimes and work-stealing runtimes make different trade-offs about Send and load balancing.