Rust Async Internals

Purpose

Guide agents through Rust async/await internals: the Future trait and poll loop, Pin /Unpin for self-referential types, tokio's task model, diagnosing async stack traces with tokio-console, finding waker leaks, and common select! /join! pitfalls.

Triggers

• "How does async/await actually work in Rust?"

• "What is Pin and Unpin in async Rust?"

• "My async code is slow — how do I profile it?"

• "How do I use tokio-console to debug async tasks?"

• "I have a blocking call in async — what do I do?"

• "How does select! work and what are the pitfalls?"

Workflow

1. The Future trait — poll model

// std::future::Future (simplified) pub trait Future { type Output; fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output>; }

pub enum Poll<T> { Ready(T), // computation done, T is the result Pending, // not ready yet, waker registered, will be polled again }

Execution model:

• Calling .await calls poll() on the future

• If Pending : current task registers its waker and yields to the runtime

• When the waker is triggered (I/O ready, timer fired), the runtime re-polls

• If Ready(val) : the .await expression evaluates to val

2. Implementing a simple Future

use std::{ future::Future, pin::Pin, task::{Context, Poll}, time::{Duration, Instant}, };

struct Delay { deadline: Instant }

impl Delay { fn new(dur: Duration) -> Self { Delay { deadline: Instant::now() + dur } } }

impl Future for Delay { type Output = ();

fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
    if Instant::now() >= self.deadline {
        Poll::Ready(())
    } else {
        // Register the waker — runtime calls waker.wake() to re-poll
        // In production: register with I/O reactor or timer wheel
        let waker = cx.waker().clone();
        let deadline = self.deadline;
        std::thread::spawn(move || {
            let now = Instant::now();
            if deadline > now {
                std::thread::sleep(deadline - now);
            }
            waker.wake();  // notify runtime to re-poll
        });
        Poll::Pending
    }
}

}

// Usage async fn main() { Delay::new(Duration::from_secs(1)).await; println!("Done"); }

3. Pin and Unpin

Pin<P> prevents moving the value behind pointer P . This matters because async state machines contain self-referential pointers (a reference into the same struct where the future lives):

// Why Pin is needed: async fn compiles to a state machine struct // that may have self-references across await points

async fn example() { let data = vec![1, 2, 3]; let ref_to_data = &data; // reference into same stack frame some_async_op().await; // suspension point println!("{:?}", ref_to_data); // reference still used after suspend } // The state machine stores both data and ref_to_data. // If the struct were moved, ref_to_data would dangle. // Pin<&mut State> prevents moving the state machine.

// Unpin: a marker trait for types that are safe to move even when pinned // Most types implement Unpin automatically // Futures generated by async/await do NOT implement Unpin

// Creating a Pin from Box (heap allocation → safe) let boxed: Pin<Box<dyn Future<Output = ()>>> = Box::pin(my_future);

// Pinning to stack (unsafe, use pin! macro) use std::pin::pin; let fut = pin!(my_future); fut.await; // or poll it directly

4. tokio task model

use tokio::task;

// Spawn a task (runs concurrently on the runtime thread pool) let handle = tokio::spawn(async { // ... async work ... 42 }); let result = handle.await.unwrap(); // wait for completion

// spawn_blocking — for CPU-bound or blocking I/O let result = task::spawn_blocking(|| { // runs on a dedicated blocking thread pool std::fs::read_to_string("big_file.txt") }).await.unwrap();

// yield to runtime (cooperative multitasking) tokio::task::yield_now().await;

// LocalSet — for !Send futures (single-threaded) let local = task::LocalSet::new(); local.run_until(async { task::spawn_local(async { /* !Send future */ }).await.unwrap(); }).await;

5. tokio-console — async task inspector

Cargo.toml

[dependencies] console-subscriber = "0.3" tokio = { version = "1", features = ["full", "tracing"] }

// main.rs fn main() { console_subscriber::init(); // must be called before tokio runtime tokio::runtime::Builder::new_multi_thread() .enable_all() .build() .unwrap() .block_on(async_main()); }

Install tokio-console CLI

cargo install --locked tokio-console

Run your app with tracing enabled

RUSTFLAGS="--cfg tokio_unstable" cargo run

In another terminal, connect tokio-console

tokio-console

tokio-console shows:

- Running tasks with their names, poll times, and wakeup counts

- Slow tasks (high poll duration = blocking in async!)

- Tasks that have been pending for a long time (stuck?)

- Resource contention (mutex/semaphore wait times)

6. Blocking in async — common mistake

// WRONG: blocking call in async context blocks entire thread async fn bad() { std::thread::sleep(Duration::from_secs(1)); // blocks runtime thread! std::fs::read_to_string("file.txt").unwrap(); // blocking I/O blocks runtime! }

// CORRECT: use async equivalents async fn good() { tokio::time::sleep(Duration::from_secs(1)).await; // async sleep tokio::fs::read_to_string("file.txt").await.unwrap(); // async I/O }

// CORRECT: if you must block, use spawn_blocking async fn with_blocking() { let content = tokio::task::spawn_blocking(|| { heavy_cpu_computation() // runs on blocking thread pool }).await.unwrap(); }

7. select! and join! pitfalls

use tokio::select;

// select! — complete when FIRST branch completes, cancels others select! { result = fetch_a() => println!("A: {:?}", result), result = fetch_b() => println!("B: {:?}", result), // Pitfall: the LOSING branches are DROPPED immediately // If fetch_a wins, fetch_b's future is dropped (and its state machine cleaned up) // This is correct and safe — but can be surprising }

// join! — wait for ALL to complete let (a, b) = tokio::join!(fetch_a(), fetch_b());

// Biased select (always check first branch first) loop { select! { biased; // prevents fairness, checks in order _ = shutdown_signal.recv() => break, msg = queue.recv() => process(msg), } }

// select! with values from loop (use fuse) let mut fut = some_future().fuse(); // FusedFuture: safe to poll after completion loop { select! { val = &mut fut => { /* ... / break; } _ = interval.tick() => { / periodic work */ } } }

Related skills

• Use skills/rust/rust-debugging for GDB/LLDB debugging of async Rust programs

• Use skills/rust/rust-profiling for cargo-flamegraph with async stack frames

• Use skills/low-level-programming/cpp-coroutines for C++20 coroutine comparison

• Use skills/low-level-programming/memory-model for memory ordering in async contexts