Tokio Runtime

Tokio provides task scheduling, asynchronous I/O, timers, and synchronization primitives. The executor polls Futures, the driver waits for OS events and wakes up tasks; user code must actively yield execution by performing operations suitable for an asynchronous environment.

Tokio tasks are lightweight asynchronous tasks that can be scheduled across multiple threads. They should not be described as "true concurrency, therefore not coroutines." How a task saves its state and how many OS threads it runs on are different dimensions.

Two Main Scheduling Modes

current_thread drives tasks on the thread that called block_on, while still being able to concurrently handle a large number of waits; multi_thread uses multiple workers and includes mechanisms like work stealing. The default number of workers depends on available CPUs and other configurations, can be overridden, and does not guarantee that one physical core is permanently bound to one worker.

Platform implementations of the I/O driver and specific queue algorithms are version-specific details; when understanding concepts, you don't need to force every socket, file, or timer into the same system call. Some file operations rely on a blocking pool rather than directly becoming kernel readiness notifications. Runtime

In a current_thread runtime, associated tasks do not continue executing out of thin air when not driven by block_on. The block_on main Future of a multi-threaded runtime is also not simply equivalent to a regular spawn task on a single worker.

A Complete Bounded Producer/Consumer Program

Cargo.toml:

[package]
name = "tokio-demo"
version = "0.1.0"
edition = "2024"

[dependencies]
tokio = { version = "1", features = ["macros", "rt-multi-thread", "sync", "time"] }

src/main.rs:

use tokio::sync::mpsc;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let (tx, mut rx) = mpsc::channel(2);
    let producer = tokio::spawn(async move {
        for value in 1_u32..=4 {
            if tx.send(value).await.is_err() {
                return; // Consumer has ended, stop producing
            }
        }
    });

    let mut total = 0;
    while let Some(value) = rx.recv().await {
        total += value;
    }
    producer.await?;
    assert_eq!(total, 10);
    println!("{total}");
    Ok(())
}

A capacity of 2 limits queued items; send waits when the queue is full; after the producer ends and releases the last sender, the consumer drains the buffer and then receives None. producer.await explicitly cleans up the task result, avoiding treating panics or other failures as "natural task end."

A bounded channel does not limit infinite tasks created before production; design so that backpressure covers the entry points that actually consume resources.

Queue is bounded, but also look at memory outside queued items

Assume a queue capacity of 100, with a single message being at most 1 MiB; the message payload alone could approach 100 MiB; additionally, there are values being constructed by the producer, values being processed, and task states. Capacity counts items, it does not automatically limit the byte size of each item, nor does it limit all Futures created upstream.

If you spawn infinitely first, then wait for a semaphore inside each task, concurrency in resource-intensive areas may be limited, but waiting tasks can still accumulate infinitely. To control the number of tasks, obtain permits before creating tasks, or wait for existing tasks to complete before receiving/creating the next item of work; results must also be cleaned up promptly. Semaphore

Send and Lifetimes of spawn

tokio::spawn usually requires the Future and its output to satisfy Send + 'static, because tasks can survive independently and may move across workers. Even with the current_thread scheduler, this API's type requirements are not automatically relaxed.

For local tasks that are !Send, use LocalSet and spawn_local according to Tokio's local task mechanism, rather than using unsafe to wrap Rc as Send. Temporary Rc or synchronous lock guards do not survive across suspension points; if they don't, they may not prevent the entire Future from being Send; clear lexical scoping is usually the easiest way to express this.

Using Local Tasks to Handle Rc

The following independent program uses the same Cargo dependencies as this article. LocalSet ensures that tasks spawned with spawn_local run in a local execution environment, but spawn_local still requires the captured variables owned by the task to satisfy 'static; it is not a substitute for ordinary borrowing scope threads.

use std::{cell::RefCell, rc::Rc};
use tokio::task::LocalSet;
#[tokio::main(flavor = "current_thread")]
async fn main() {
    let local = LocalSet::new();
    local.run_until(async {
        let values = Rc::new(RefCell::new(vec![1]));
        let shared = Rc::clone(&values);
        let task = tokio::task::spawn_local(async move {
            tokio::task::yield_now().await;
            shared.borrow_mut().push(2);
        });
        task.await.unwrap();
        assert_eq!(&*values.borrow(), &[1, 2]);
    }).await;
}

Rc survives across await in this task, so it is not suitable for tokio::spawn, but fits the explicit local task model. RefCell borrows should still end within a reasonable scope; LocalSet does not eliminate single-threaded reentrancy conflicts. LocalSet

Dropping JoinHandle Does Not Equal Task Cancellation

Dropping a Tokio JoinHandle usually detaches the task from waiting, but the task can continue executing. The abort request cancels the task, taking effect at appropriate scheduling boundaries; waiting on the handle allows observing the cancellation completion. A long-running poll that never yields will not be immediately preempted just because another task calls abort. JoinHandle

This is different from dropping a Future that has not yet been spawned and is directly owned by the current task. Resource cleanup and background work exit require explicit ownership and cooperative protocols; use shutdown signals and wait for completion when necessary.

Explicitly Request Termination and Cleanup After Timeout Waiting on JoinHandle

use std::{sync::{Arc, atomic::{AtomicBool, Ordering}}, time::Duration};
use tokio::{sync::oneshot, time::timeout};
struct MarkDropped(Arc<AtomicBool>);
impl Drop for MarkDropped {
    fn drop(&mut self) { self.0.store(true, Ordering::SeqCst); }
}
#[tokio::main(flavor = "current_thread")]
async fn main() {
    let dropped = Arc::new(AtomicBool::new(false));
    let flag = Arc::clone(&dropped);
    let (ready_tx, ready_rx) = oneshot::channel();
    let mut task = tokio::spawn(async move {
        let resource = MarkDropped(flag);
        ready_tx.send(()).unwrap();
        std::future::pending::<()>().await;
        drop(resource);
    });
    ready_rx.await.unwrap(); // Confirm resource constructed, task will stay on pending.
    assert!(timeout(Duration::from_millis(1), &mut task).await.is_err());
    assert!(!dropped.load(Ordering::SeqCst));
    task.abort();
    assert!(task.await.unwrap_err().is_cancelled());
    assert!(dropped.load(Ordering::SeqCst));
}

timeout borrows &mut task; after timeout, the JoinHandle is retained, allowing the decision to continue waiting, cooperatively shut down, or abort. If the handle is passed by value to timeout, the timeout will drop the handle in the waiting object, but the background task usually continues to exist. The Pending in the example never completes, so the cancellation result is deterministic; in general tasks, normal completion may occur before abort takes effect, and handling code should allow for this race condition.

1 millisecond here is only used to trigger the timeout branch, not a promise of timing precision. timeout cannot preempt a poll that always returns; expensive computations may still continue running past the deadline. It limits the waiting protocol, not an OS-level forced termination. timeout

Does dropping the waiter stop the task?

Identical drop syntax has different effects depending on ownership. A direct Future owns operation state; JoinHandle is a waiting and control interface to a spawned task. Drop the handle, then reset and request abort, comparing task lifetime to distinguish stopping waiting, requesting cancellation and completed cancellation.

select! Cancellation Boundaries Depend on Operand Ownership

select! polls multiple branches within the current task, it does not automatically spawn each branch to different threads. After a branch is selected, other local Futures are dropped; if the branch only borrowed an external fixed Future, its underlying object may still be held externally.

If a branch owns a JoinHandle, dropping it only detaches the task, it does not automatically stop the task. If what is dropped is a read_exact Future that has consumed part of the input, progress may be lost. Repeated select loops must check the cancellation safety of each operation. select!

The default branch polling order and biased mode have different fairness considerations; biased does not guarantee business fairness. Branches containing blocking code will block the entire select from progressing.

JoinSet Places Task Creation and Result Cleanup in the Same Loop

use tokio::task::{JoinError, JoinSet};
fn accept(result: Result<Result<u32, &'static str>, JoinError>,
          values: &mut Vec<u32>, rejected: &mut usize) {
    match result {
        Ok(Ok(value)) => values.push(value),
        Ok(Err("invalid input")) => *rejected += 1,
        other => panic!("Unexpected task result: {other:?}"),
    }
}
#[tokio::main(flavor = "current_thread")]
async fn main() {
    let mut tasks = JoinSet::new();
    let mut values = Vec::new();
    let mut rejected = 0;
    for value in 0u32..5 {
        if tasks.len() >= 2 {
            accept(tasks.join_next().await.unwrap(), &mut values, &mut rejected);
        }
        tasks.spawn(async move {
            tokio::task::yield_now().await;
            if value == 3 { Err("invalid input") } else { Ok(value * 2) }
        });
        assert!(tasks.len() <= 2);
    }
    while let Some(result) = tasks.join_next().await {
        accept(result, &mut values, &mut rejected);
    }
    values.sort_unstable();
    assert_eq!(values, [0, 2, 4, 8]);
    assert_eq!(rejected, 1);
}

Here, at most two tasks that have not yet been cleaned up are retained. JoinSet yields results in completion order, not input order, so the example sorts when verifying; if the business requires original order, carry sequence numbers and rearrange, while considering the rearrangement buffer limit. The output Vec is safely controllable in this finite example; with infinite input, it should be changed to deliver results item by item.

Return values have two layers of failure: the outer JoinError indicates task cancellation or panic, the inner Err is a business-returned failure. A single .await? only propagates the outer layer, you cannot infer business success from this. JoinSet

Dropping a JoinSet requests abort of its tasks, which is different from dropping a single JoinHandle; if you need to confirm all destructors have completed, use and wait for the corresponding shutdown/cleanup process. Simply issuing an abort request does not mean resource cleanup is complete.

Blocking Work vs CPU Work

std::thread::sleep, blocking library calls, and long calculations should not directly occupy an async worker for a long time. Limited blocking tasks can use spawn_blocking, but its already running closures usually cannot be aborted; runtime shutdown must also consider its termination behavior. spawn_blocking

The blocking pool has its own threads and queue strategies, it does not automatically provide a reasonable concurrency limit for infinite CPU work. Large amounts of CPU computation should be explicitly rate-limited or use appropriate dedicated execution pools, and verified against target loads.

Don't Choose Locks Only Based on "async means async locks"

Short, low-contention critical sections that do not cross await can use a standard Mutex. Holding a lock across await can block other tasks or introduce deadlocks; std MutexGuard may also make the Future not satisfy Send.

Tokio's async Mutex allows asynchronous waiting for the lock and can hold the guard across await when necessary, but this does not exempt business logic from deadlocks and long-term exclusive holding. Shared I/O resources are sometimes better owned by a single task, with requests sent via message passing. Tokio Shared State

When diagnosing latency, distinguish between: queuing wait, scheduling not getting run, CPU execution, I/O wait, and lock wait. Simply seeing slow requests does not determine if there are too few workers or insufficient downstream capacity.

Shutdown Order and Observability

If a bounded consumer waits until all senders disappear before exiting, it must know who holds each clone of the sender. Extra clones left in long-lived structures may cause recv to wait for None forever. After actively stopping new sends via Receiver::close, you should still continue receiving and processing buffered messages; permits previously obtained still need to be handled according to channel contracts, do not understand close as instantly clearing everything. mpsc Receiver::close

Service shutdown can sequentially stop entry points, emit cooperative signals, process received work, clean up tasks, and then end the runtime. Recording the number of pending items, the age of the oldest task, and remaining budget for each step allows distinguishing whether the producer didn't stop, a task didn't yield, or the downstream is still waiting.

When manually constructing a Runtime, you must also enable the required time or I/O drivers; "creation successful" does not mean any Future depending on the driver can run. The entry context of the runtime handle is also different from actually continuously driving the runtime, especially in a current_thread environment where explicit block_on ownership should be maintained.