The Tokio Ecosystem

Tokio provides the runtime, timers, sockets, and synchronization. Most services also use crates built on Tokio for HTTP, middleware, buffers, and logging. This lesson maps those crates to their roles, shows how a tower service and layer work, and runs an axum server that a reqwest client calls inside the same program.

The stack

Crate	Role	Sits on
`mio`	Cross-platform readiness API over epoll, kqueue, and IOCP; Tokio's I/O driver uses it	the operating system
`tokio`	Runtime, scheduler, timers, TCP/UDP/Unix sockets, channels, locks	`mio`
`tokio-util`	Codecs and `Framed`, `CancellationToken`, `TaskTracker`, `StreamReader`	`tokio`, `bytes`
`tokio-stream`	`StreamExt`, stream wrappers for channels and timers	`tokio`
`bytes`	`Bytes` and `BytesMut` buffers with cheap clones and slices	none
`tracing`	Structured events and spans; subscribers decide where they go	none
`tower`	`Service` and `Layer` traits, plus middleware such as timeouts and rate limits	`tokio`, `futures`
`tower-http`	HTTP middleware built as `tower` layers: CORS, tracing, compression, and others	`tower`, `http`
`hyper`	HTTP/1 and HTTP/2 protocol implementation	`tokio`, `bytes`, `http`
`axum`	HTTP framework: routing, extractors, handlers	`hyper`, `tower`
`reqwest`	HTTP client with connection pooling, JSON, timeouts	`hyper`, `tokio`
`tonic`	gRPC client and server	`hyper`, `tower`, `prost`
`sqlx`	Async SQL driver with a connection pool	`tokio`

The programs in this lesson use every crate in the table except mio, hyper, tonic, and sqlx. mio and hyper are used through tokio, axum, and reqwest and never imported directly. tonic and sqlx are named because they follow the same conventions: they run on the Tokio runtime and expose tower services or async pools.

graph TD
    OS[Operating system: epoll, kqueue, IOCP] --> mio
    mio --> tokio
    tokio --> tokio-util
    tokio --> tokio-stream
    bytes --> tokio-util
    tokio --> tower
    tower --> tower-http
    tokio --> hyper
    hyper --> axum
    tower --> axum
    hyper --> reqwest
    hyper --> tonic
    tower --> tonic
    tokio --> sqlx
    tracing -.-> axum
    tracing -.-> tower-http

Solid arrows mean "is built on". Dotted arrows mean "emits events through". tracing does not depend on Tokio.

Example 1: A Cargo manifest for a service

A service that serves HTTP and calls other services usually depends on the crates below. The versions match those the course examples are checked against.

TOML

[dependencies]
tokio = { version = "1", features = ["full"] }
tokio-util = { version = "0.7", features = ["full"] }
tokio-stream = "0.1"
bytes = "1"
tracing = "0.1"
tracing-subscriber = "0.3"
tower = { version = "0.5", features = ["util"] }
tower-http = { version = "0.6", features = ["trace", "cors"] }
axum = "0.8"
reqwest = { version = "0.13", features = ["json"] }
serde = { version = "1", features = ["derive"] }

tower and tower-http gate most of their middleware behind features (timeout, limit, retry, and others in tower; trace, cors, compression-gzip, and others in tower-http). Enable only the features in use, because each one adds code to the build.

Example 2: The small crates together

tokio-util supplies the codec, bytes supplies the buffer type, and tokio-stream supplies StreamExt. The program sends three lines through an in-memory pipe with LinesCodec (Codecs with tokio-util) and reads them back as a stream (Streams).

Rust

use bytes::Bytes;
use futures::SinkExt;
use tokio_stream::StreamExt;
use tokio_util::codec::{Framed, LinesCodec};

#[tokio::main]
async fn main() {
    let (client_io, server_io) = tokio::io::duplex(256);
    let mut client = Framed::new(client_io, LinesCodec::new());
    let mut server = Framed::new(server_io, LinesCodec::new());

    for line in ["first", "second", "third"] {
        client.send(line).await.unwrap();
    }
    drop(client);

    while let Some(line) = server.next().await {
        let line = line.unwrap();
        let shared = Bytes::from(line.clone());
        let copy = shared.clone(); // no byte copy: both handles share one buffer
        println!("{line}: {} bytes, same buffer: {}", copy.len(), shared.as_ptr() == copy.as_ptr());
    }
}

Output

first: 5 bytes, same buffer: true
second: 6 bytes, same buffer: true
third: 5 bytes, same buffer: true

Bytes::clone increments a reference count instead of copying the data, so both handles point at the same memory (Bytes and Framing).

Example 3: `tracing` events

tracing records structured events. A subscriber from tracing-subscriber decides how to print them (Tracing and Observability). The library crates in this lesson (axum, tower-http) emit events in the same way, so one subscriber collects everything.

Rust

use tracing::{info, warn};

#[tokio::main]
async fn main() {
    tracing_subscriber::fmt()
        .with_ansi(false)
        .without_time()
        .with_target(false)
        .with_writer(std::io::stdout)
        .init();

    info!(service = "billing", "starting");
    warn!(retries = 3, "upstream is slow");
}

Output

 INFO starting service="billing"
 WARN upstream is slow retries=3

Example 4: A `tower` service

tower::Service is a trait for "an async function from a request to a response" with a readiness check. poll_ready says whether the service can take a request, and call starts one. service_fn turns a closure into a service, and ServiceExt::oneshot waits for readiness and makes one call.

Rust

use std::convert::Infallible;
use tower::{Service, ServiceExt, service_fn};

#[tokio::main]
async fn main() {
    let mut greet = service_fn(|name: String| async move {
        Ok::<_, Infallible>(format!("hello, {name}"))
    });

    let first = greet.ready().await.unwrap().call("tokio".to_string()).await.unwrap();
    println!("{first}");

    let second = greet.oneshot("tower".to_string()).await.unwrap();
    println!("{second}");
}

Output

hello, tokio
hello, tower

ready() waits until poll_ready succeeds. call requires &mut self and a ready service, which is why calls go through ready().await or oneshot.

Example 5: A `tower` layer

A Layer wraps one service in another. The wrapper implements Service too, so wrappers stack. This layer counts calls and forwards each request unchanged.

Rust

use std::convert::Infallible;
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::task::{Context, Poll};
use tower::{Layer, Service, ServiceExt, service_fn};

struct CountLayer {
    calls: Arc<AtomicUsize>,
}

impl<S> Layer<S> for CountLayer {
    type Service = Count<S>;

    fn layer(&self, inner: S) -> Count<S> {
        Count { inner, calls: Arc::clone(&self.calls) }
    }
}

#[derive(Clone)]
struct Count<S> {
    inner: S,
    calls: Arc<AtomicUsize>,
}

impl<S, Request> Service<Request> for Count<S>
where
    S: Service<Request>,
{
    type Response = S::Response;
    type Error = S::Error;
    type Future = S::Future;

    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), S::Error>> {
        self.inner.poll_ready(cx)
    }

    fn call(&mut self, request: Request) -> S::Future {
        self.calls.fetch_add(1, Ordering::SeqCst);
        self.inner.call(request)
    }
}

#[tokio::main]
async fn main() {
    let calls = Arc::new(AtomicUsize::new(0));
    let inner = service_fn(|n: u32| async move { Ok::<_, Infallible>(n * 2) });
    let mut service = CountLayer { calls: Arc::clone(&calls) }.layer(inner);

    for n in [1, 2, 3] {
        let doubled = service.ready().await.unwrap().call(n).await.unwrap();
        println!("{n} -> {doubled}");
    }
    println!("calls counted: {}", calls.load(Ordering::SeqCst));
}

Output

1 -> 2
2 -> 4
3 -> 6
calls counted: 3

The wrapper adds behavior (a counter) without changing the inner service, and poll_ready passes through so that the inner service's readiness rules still apply.

Example 6: Stacking layers with `ServiceBuilder`

ServiceBuilder applies layers in order: the first layer added is the outermost, so it sees the request first and the response last. map_request and map_response are small layers built from closures.

Rust

use std::convert::Infallible;
use tower::{ServiceBuilder, ServiceExt, service_fn};

#[tokio::main]
async fn main() {
    let service = ServiceBuilder::new()
        .map_request(|text: &str| {
            println!("layer 1 sees request {text:?}");
            text.trim().to_string()
        })
        .map_request(|text: String| {
            println!("layer 2 sees request {text:?}");
            text.to_uppercase()
        })
        .map_response(|length: usize| {
            println!("layer 1 sees response {length}");
            length
        })
        .service(service_fn(|text: String| async move {
            println!("inner service gets {text:?}");
            Ok::<_, Infallible>(text.len())
        }));

    let length = service.oneshot("  tokio  ").await.unwrap();
    println!("final response: {length}");
}

Output

layer 1 sees request "  tokio  "
layer 2 sees request "tokio"
inner service gets "TOKIO"
layer 1 sees response 5
final response: 5

Requests travel from the first layer inward. Responses travel back outward. Timeouts, rate limits, retries, and load shedding in tower are layers that stack the same way (Backpressure and Load Shedding explains the ideas behind them).

Example 7: An `axum` router is a `tower` service

axum::Router implements Service, so a test can call it without opening a socket. The request is an http::Request (re-exported by axum::http) and the response body is read with axum::body::to_bytes.

Rust

use axum::body::{Body, to_bytes};
use axum::extract::Path;
use axum::http::Request;
use axum::{Router, routing::get};
use tower::ServiceExt;

#[tokio::main]
async fn main() {
    let app = Router::new().route(
        "/hello/{name}",
        get(|Path(name): Path<String>| async move { format!("hello, {name}") }),
    );

    for uri in ["/hello/tokio", "/missing"] {
        let request = Request::builder().uri(uri).body(Body::empty()).unwrap();
        let response = app.clone().oneshot(request).await.unwrap();
        let status = response.status();
        let body = to_bytes(response.into_body(), 1024).await.unwrap();
        println!("{uri} -> {status}, body {:?}", String::from_utf8_lossy(&body));
    }
}

Output

/hello/tokio -> 200 OK, body "hello, tokio"
/missing -> 404 Not Found, body ""

Route parameters use braces ({name}) in axum 0.8. Router is cheap to clone, which is why app.clone().oneshot(...) can run repeatedly.

Example 8: An `axum` server on port 0, called by `reqwest`

Binding to port 0 lets the operating system choose a free port, and local_addr reports it. The server runs in a spawned task, and the client in main calls it over a real TCP connection (TCP Clients and Servers).

Rust

use axum::{Router, routing::get};
use tokio::net::TcpListener;

#[tokio::main]
async fn main() {
    let app = Router::new().route("/ping", get(|| async { "pong" }));
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    tokio::spawn(async move {
        axum::serve(listener, app).await.unwrap();
    });

    let response = reqwest::get(format!("http://{addr}/ping")).await.unwrap();
    println!("status: {}", response.status());
    println!("body: {}", response.text().await.unwrap());
}

Output

status: 200 OK
body: pong

reqwest::get builds a temporary client for one request. Later examples build one Client and reuse it.

Example 9: JSON, shared state, and status codes

Handlers receive shared state through the State extractor and JSON through Json. The state is cloned per request, so it holds an Arc for data that must be shared. The handler returns a status code and a JSON body.

Rust

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

use axum::extract::State;
use axum::http::StatusCode;
use axum::routing::post;
use axum::{Json, Router};
use serde::{Deserialize, Serialize};
use tokio::net::TcpListener;

#[derive(Clone, Default)]
struct AppState {
    created: Arc<AtomicUsize>,
}

#[derive(Serialize, Deserialize)]
struct NewJob {
    name: String,
}

#[derive(Serialize, Deserialize)]
struct Created {
    id: usize,
    name: String,
}

async fn create_job(
    State(state): State<AppState>,
    Json(job): Json<NewJob>,
) -> (StatusCode, Json<Created>) {
    let id = state.created.fetch_add(1, Ordering::SeqCst) + 1;
    (StatusCode::CREATED, Json(Created { id, name: job.name }))
}

#[tokio::main]
async fn main() {
    let app = Router::new().route("/jobs", post(create_job)).with_state(AppState::default());
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    tokio::spawn(async move {
        axum::serve(listener, app).await.unwrap();
    });

    let client = reqwest::Client::new();
    for name in ["resize", "email"] {
        let response = client
            .post(format!("http://{addr}/jobs"))
            .json(&NewJob { name: name.to_string() })
            .send()
            .await
            .unwrap();
        let status = response.status();
        let created: Created = response.json().await.unwrap();
        println!("{status}: job {} is {}", created.id, created.name);
    }

    let bad = client
        .post(format!("http://{addr}/jobs"))
        .header("content-type", "application/json")
        .body("{\"wrong\": 1}")
        .send()
        .await
        .unwrap();
    println!("malformed body: {}", bad.status());
}

Output

201 Created: job 1 is resize
201 Created: job 2 is email
malformed body: 422 Unprocessable Entity

axum rejects a body that does not match NewJob before the handler runs and answers 422 Unprocessable Entity.

Example 10: `tower-http` middleware

tower-http layers attach to a Router with .layer(...). CorsLayer answers cross-origin rules by adding response headers. The client sends an Origin header and prints the header that comes back.

Rust

use axum::{Router, routing::get};
use tokio::net::TcpListener;
use tower_http::cors::CorsLayer;

#[tokio::main]
async fn main() {
    let app = Router::new()
        .route("/ping", get(|| async { "pong" }))
        .layer(CorsLayer::permissive());
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    tokio::spawn(async move {
        axum::serve(listener, app).await.unwrap();
    });

    let response = reqwest::Client::new()
        .get(format!("http://{addr}/ping"))
        .header("origin", "https://example.com")
        .send()
        .await
        .unwrap();
    let allow = response.headers().get("access-control-allow-origin");
    println!("access-control-allow-origin: {:?}", allow.map(|value| value.to_str().unwrap()));
}

Output

access-control-allow-origin: Some("*")

CorsLayer::permissive() allows any origin, method, and header. A production service lists the origins it accepts.

Example 11: Client timeouts

A reqwest request has no timeout unless one is configured. Client::builder().timeout(...) sets a total limit for each request, and connect_timeout limits only the connection step. This server has a handler that never answers.

Rust

use std::time::Duration;

use axum::{Router, routing::get};
use tokio::net::TcpListener;

#[tokio::main]
async fn main() {
    let app = Router::new()
        .route("/fast", get(|| async { "done" }))
        .route(
            "/stuck",
            get(|| async {
                std::future::pending::<()>().await;
                "unreachable"
            }),
        );
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    tokio::spawn(async move {
        axum::serve(listener, app).await.unwrap();
    });

    let client = reqwest::Client::builder()
        .connect_timeout(Duration::from_secs(1))
        .timeout(Duration::from_millis(200))
        .build()
        .unwrap();

    let fast = client.get(format!("http://{addr}/fast")).send().await.unwrap();
    println!("fast: {}", fast.text().await.unwrap());

    let stuck = client.get(format!("http://{addr}/stuck")).send().await;
    match stuck {
        Ok(_) => println!("stuck: unexpected response"),
        Err(error) => println!("stuck: is_timeout = {}", error.is_timeout()),
    }

    let patient = client
        .get(format!("http://{addr}/stuck"))
        .timeout(Duration::from_millis(50))
        .send()
        .await;
    println!("per-request timeout: is_timeout = {}", patient.unwrap_err().is_timeout());
}

Output

fast: done
stuck: is_timeout = true
per-request timeout: is_timeout = true

A per-request .timeout(...) overrides the client default for that request. The total timeout runs from the start of connecting until the response body has finished, so it also limits slow body reads. The default is no timeout.

Example 12: Connection reuse

A reqwest::Client keeps a pool of idle connections per host. Requests from the same client reuse a connection when one is idle. Each new Client has its own pool and opens its own connection. The server below wraps its listener to count accepted connections. The Listener trait is axum::serve::Listener.

Rust

use std::io;
use std::net::SocketAddr;
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::Duration;

use axum::serve::Listener;
use axum::{Router, routing::get};
use tokio::net::{TcpListener, TcpStream};

struct CountingListener {
    inner: TcpListener,
    accepted: Arc<AtomicUsize>,
}

impl Listener for CountingListener {
    type Io = TcpStream;
    type Addr = SocketAddr;

    async fn accept(&mut self) -> (Self::Io, Self::Addr) {
        let connection = Listener::accept(&mut self.inner).await;
        self.accepted.fetch_add(1, Ordering::SeqCst);
        connection
    }

    fn local_addr(&self) -> io::Result<Self::Addr> {
        self.inner.local_addr()
    }
}

async fn get_text(client: &reqwest::Client, url: &str) -> String {
    let text = client.get(url).send().await.unwrap().text().await.unwrap();
    // The pool takes the connection back shortly after the body ends.
    tokio::time::sleep(Duration::from_millis(20)).await;
    text
}

#[tokio::main]
async fn main() {
    let accepted = Arc::new(AtomicUsize::new(0));
    let listener = CountingListener {
        inner: TcpListener::bind("127.0.0.1:0").await.unwrap(),
        accepted: Arc::clone(&accepted),
    };
    let addr = listener.local_addr().unwrap();
    let app = Router::new().route("/ping", get(|| async { "pong" }));
    tokio::spawn(async move {
        axum::serve(listener, app).await.unwrap();
    });
    let url = format!("http://{addr}/ping");

    let shared = reqwest::Client::new();
    for _ in 0..5 {
        get_text(&shared, &url).await;
    }
    println!("one client, 5 requests: {} connection(s)", accepted.load(Ordering::SeqCst));

    accepted.store(0, Ordering::SeqCst);
    for _ in 0..5 {
        get_text(&reqwest::Client::new(), &url).await;
    }
    println!("new client per request: {} connection(s)", accepted.load(Ordering::SeqCst));
}

Output

one client, 5 requests: 1 connection(s)
new client per request: 5 connection(s)

The shared client makes one TCP connection for five requests. Building a client per request pays a connection setup and a client setup each time, and it discards the pool. A service builds its Client once and clones it. The reqwest documentation states that Client uses an Arc internally, so clones share the pool.

Example 13: Graceful shutdown of the `axum` server

axum::serve(...).with_graceful_shutdown(signal) stops accepting connections when the signal future completes, and it waits for in-flight requests before returning. The pattern matches Graceful Shutdown, with the future usually being ctrl_c in a real service and a oneshot receiver here.

Rust

use std::time::Duration;

use axum::{Router, routing::get};
use tokio::net::TcpListener;
use tokio::sync::oneshot;

#[tokio::main]
async fn main() {
    let app = Router::new().route(
        "/slow",
        get(|| async {
            tokio::time::sleep(Duration::from_millis(100)).await;
            "finished"
        }),
    );
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let (stop_tx, stop_rx) = oneshot::channel::<()>();
    let server = tokio::spawn(async move {
        axum::serve(listener, app)
            .with_graceful_shutdown(async {
                let _ = stop_rx.await;
            })
            .await
            .unwrap();
    });

    let request = tokio::spawn(async move {
        reqwest::get(format!("http://{addr}/slow")).await.unwrap().text().await.unwrap()
    });
    tokio::time::sleep(Duration::from_millis(30)).await; // the request is now in flight
    stop_tx.send(()).unwrap();

    println!("in-flight request: {}", request.await.unwrap());
    server.await.unwrap();
    println!("server stopped");
}

Output

in-flight request: finished
server stopped

The request that started before the stop signal completes normally, and only then does the server task end.

Example 14: Choosing crates for a service

The criteria below map a requirement to the crate that provides it.

Requirement	Crate	Criterion
Serve HTTP routes with typed extractors	`axum`	Handlers are plain async functions; the router is a `tower` service
Call an HTTP API	`reqwest`	Pooling, JSON, timeouts, redirects; build one `Client`
Add timeouts, rate limits, or retries to any async call	`tower`	The behavior is a layer that wraps a `Service`
Add request tracing, CORS, or compression to HTTP	`tower-http`	Layers that plug into `axum` with `.layer(...)`
Speak a custom TCP protocol	`tokio-util` codecs and `bytes`	`Framed` turns a socket into a stream and sink of typed frames
Coordinate shutdown across tasks	`tokio-util`	`CancellationToken` and `TaskTracker`
Consume channels and intervals as streams	`tokio-stream`	Wrappers such as `ReceiverStream` and `IntervalStream`
Log with context across tasks	`tracing`, `tracing-subscriber`	Spans follow the work across `.await` points
Build gRPC services	`tonic`	Generated from protobuf definitions; runs on `hyper` and `tower`
Query SQL databases	`sqlx`	Async pool; queries are futures
Drive the HTTP protocol directly	`hyper`	Needed when `axum` and `reqwest` do not fit; reachable through them otherwise

Two rules cover mixing versions. Crates that share a type in their public API (http, bytes, tower) must resolve to the same major version, or the types do not match. cargo tree -d lists duplicated crates.

Interview points

mio wraps epoll, kqueue, and IOCP. Tokio's I/O driver is built on it.
tower::Service has poll_ready and call. A Layer turns one service into another.
The first layer added to a ServiceBuilder is the outermost.
axum::Router is a tower service, so it can be called without a socket in tests.
axum and reqwest both build on hyper.
tower-http middleware are tower layers.
A reqwest::Client pools connections; build it once and clone it.
reqwest has no default request timeout.
axum::serve(...).with_graceful_shutdown(...) stops accepting and waits for in-flight requests.
Binding to port 0 and reading local_addr avoids port clashes in tests.

Summary

tokio supplies the runtime; tokio-util, tokio-stream, and bytes supply codecs, stream helpers, and buffers.
tracing records events; tracing-subscriber prints them.
tower defines Service and Layer; middleware wraps services and stacks with ServiceBuilder.
axum serves HTTP on hyper and tower; reqwest calls HTTP on hyper.
Extractors (State, Json, Path) give handlers typed input, and mismatched JSON is rejected before the handler runs.
Set client timeouts explicitly and reuse one Client.
tonic and sqlx follow the same runtime and tower conventions.

The stack

Example 1: A Cargo manifest for a service

Example 2: The small crates together

Example 3: tracing events

Example 4: A tower service

Example 5: A tower layer

Example 6: Stacking layers with ServiceBuilder

Example 7: An axum router is a tower service

Example 8: An axum server on port 0, called by reqwest

Example 9: JSON, shared state, and status codes

Example 10: tower-http middleware

Example 11: Client timeouts

Example 12: Connection reuse

Example 13: Graceful shutdown of the axum server

Example 14: Choosing crates for a service

Interview points

Summary

Example 3: `tracing` events

Example 4: A `tower` service

Example 5: A `tower` layer

Example 6: Stacking layers with `ServiceBuilder`

Example 7: An `axum` router is a `tower` service

Example 8: An `axum` server on port 0, called by `reqwest`

Example 10: `tower-http` middleware

Example 13: Graceful shutdown of the `axum` server