SOP: Async Programming

Updated Mar 2026

Overview

Rust's async/await enables non-blocking I/O with zero-cost abstractions. Futures are lazy — they do nothing until polled. This SOP covers the tokio runtime, async patterns, streams, and common pitfalls.

Async Mental Model

Setting Up Tokio

// Cargo.toml
// [dependencies]
// tokio = { version = "1", features = ["full"] }

// Full runtime (multi-threaded)
#[tokio::main]
async fn main() {
    println!("Running in tokio runtime");
    let result = fetch_data().await;
    println!("Got: {result}");
}

// Single-threaded runtime (for testing / simple apps)
#[tokio::main(flavor = "current_thread")]
async fn main() {
    // Runs on one thread
}

// Manual runtime creation
fn main() {
    let rt = tokio::runtime::Runtime::new().unwrap();
    rt.block_on(async {
        println!("Inside manual runtime");
    });
}

Basic Async/Await

use tokio::time::{sleep, Duration};

async fn fetch_data() -> String {
    // Simulates async I/O
    sleep(Duration::from_millis(100)).await;
    "data from server".to_string()
}

async fn process() {
    let data = fetch_data().await;  // Suspends here, other tasks can run
    println!("Processed: {data}");
}

Concurrent Execution

use tokio::time::{sleep, Duration};

async fn task_a() -> String {
    sleep(Duration::from_secs(2)).await;
    "result A".to_string()
}

async fn task_b() -> String {
    sleep(Duration::from_secs(1)).await;
    "result B".to_string()
}

// join! — run concurrently, wait for all
async fn run_both() {
    let (a, b) = tokio::join!(task_a(), task_b());
    println!("{a}, {b}");  // Takes 2 seconds total, not 3
}

// select! — first one wins
async fn race() {
    tokio::select! {
        a = task_a() => println!("A finished first: {a}"),
        b = task_b() => println!("B finished first: {b}"),
    }
    // B wins (1 second). A is cancelled.
}

// spawn — fire and forget (or await later)
async fn background() {
    let handle = tokio::spawn(async {
        sleep(Duration::from_secs(1)).await;
        42
    });

    // Do other work...
    let result = handle.await.unwrap();  // Collect result later
    println!("Background result: {result}");
}

Async Channels

use tokio::sync::mpsc;

async fn producer_consumer() {
    let (tx, mut rx) = mpsc::channel::<String>(100);  // Bounded channel

    // Producer
    let producer = tokio::spawn(async move {
        for i in 0..5 {
            tx.send(format!("message {i}")).await.unwrap();
        }
        // tx dropped here — receiver will get None
    });

    // Consumer
    while let Some(msg) = rx.recv().await {
        println!("Received: {msg}");
    }

    producer.await.unwrap();
}

// oneshot — single value
use tokio::sync::oneshot;

async fn request_response() {
    let (tx, rx) = oneshot::channel();

    tokio::spawn(async move {
        let result = expensive_computation().await;
        tx.send(result).unwrap();
    });

    let value = rx.await.unwrap();
    println!("Got response: {value}");
}

Shared State in Async

use std::sync::Arc;
use tokio::sync::Mutex;

#[derive(Clone)]
struct AppState {
    counter: Arc<Mutex<u64>>,
}

async fn increment(state: AppState) {
    let mut lock = state.counter.lock().await;  // Async mutex!
    *lock += 1;
}

// Use tokio::sync::Mutex for async, NOT std::sync::Mutex
// std::sync::Mutex blocks the thread — bad in async context
// tokio::sync::Mutex yields to the runtime while waiting

Streams (Async Iterators)

use tokio_stream::{self as stream, StreamExt};

async fn process_stream() {
    let mut stream = stream::iter(vec![1, 2, 3, 4, 5])
        .filter(|x| x % 2 == 0)
        .map(|x| x * 10);

    while let Some(value) = stream.next().await {
        println!("Stream value: {value}");
    }
}

// Creating a stream from a channel
use tokio::sync::mpsc;
use tokio_stream::wrappers::ReceiverStream;

async fn channel_stream() {
    let (tx, rx) = mpsc::channel(10);
    let mut stream = ReceiverStream::new(rx);

    tokio::spawn(async move {
        for i in 0..5 {
            tx.send(i).await.unwrap();
        }
    });

    while let Some(val) = stream.next().await {
        println!("{val}");
    }
}

Common Pitfalls

Pitfall	Problem	Fix
Holding std::sync::Mutex across .await	Blocks thread	Use tokio::sync::Mutex
Blocking in async (CPU-heavy work)	Starves other tasks	Use tokio::task::spawn_blocking
Not using Send types with tokio::spawn	Compilation error	Ensure all captured types are Send
Forgetting .await	Future does nothing	Always await or spawn futures
Unbounded channel in hot loop	Memory leak	Use bounded channels with backpressure

// spawn_blocking for CPU-intensive work
async fn compute() -> u64 {
    tokio::task::spawn_blocking(|| {
        // This runs on a dedicated blocking thread pool
        heavy_computation()
    })
    .await
    .unwrap()
}

Checklist

• [ ] Use #[tokio::main] or manual runtime setup
• [ ] Every async fn call has .await
• [ ] Use tokio::sync::Mutex (not std::sync::Mutex) in async
• [ ] CPU-heavy work goes to spawn_blocking
• [ ] Channels are bounded with appropriate capacity
• [ ] All types captured by tokio::spawn closures are Send
• [ ] Error handling uses ? inside async functions
• [ ] Graceful shutdown with select! or signal handling