Rust Async

Rust Async Patterns

async/await Basics

[dependencies]
tokio = { version = "1", features = ["full"] }

#[tokio::main]
async fn main() {
    let result = fetch_data("https://api.example.com/data").await;
    match result {
        Ok(data) => println!("Got: {data}"),
        Err(e) => eprintln!("Error: {e}"),
    }
}

async fn fetch_data(url: &str) -> Result<String, reqwest::Error> {
    let response = reqwest::get(url).await?;
    response.text().await
}

Spawning Tasks

tokio::spawn creates a new async task that runs concurrently.

use tokio::task::JoinHandle;

async fn process_items(items: Vec<String>) -> Vec<String> {
    let handles: Vec<JoinHandle<String>> = items
        .into_iter()
        .map(|item| tokio::spawn(async move { expensive_transform(item).await }))
        .collect();

    let mut results = Vec::new();
    for handle in handles {
        results.push(handle.await.expect("task panicked"));
    }
    results
}

// Background worker with shutdown signal
async fn start_worker(mut shutdown: tokio::sync::broadcast::Receiver<()>) {
    loop {
        tokio::select! {
            _ = shutdown.recv() => break,
            _ = do_work() => {}
        }
    }
}

Joining Multiple Futures

use tokio::try_join;

async fn parallel_fetch() -> Result<(User, Posts), AppError> {
    // Both run concurrently; fails fast if either errors
    let (user, posts) = try_join!(fetch_user(42), fetch_posts(42))?;
    Ok((user, posts))
}

// FuturesUnordered for dynamic sets
use futures::stream::{FuturesUnordered, StreamExt};

async fn process_dynamic(ids: Vec<u64>) {
    let mut futs: FuturesUnordered<_> = ids.into_iter()
        .map(|id| fetch_item(id))
        .collect();

    while let Some(result) = futs.next().await {
        println!("got: {:?}", result);
    }
}

select! for Racing Futures

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

async fn with_timeout<F, T>(fut: F, dur: Duration) -> Option<T>
where
    F: std::future::Future<Output = T>,
{
    tokio::select! {
        result = fut => Some(result),
        _ = sleep(dur) => None,
    }
}

// Cancellation token
use tokio_util::sync::CancellationToken;

async fn cancellable_work(token: CancellationToken) {
    loop {
        tokio::select! {
            biased;
            _ = token.cancelled() => break,
            _ = do_unit_of_work() => {}
        }
    }
}

Channels

// mpsc: multiple producer, single consumer
use tokio::sync::mpsc;

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

    for i in 0..5 {
        let tx = tx.clone();
        tokio::spawn(async move {
            tx.send(format!("message from {i}")).await.unwrap();
        });
    }
    drop(tx); // close when all producers done

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

// broadcast: single producer, multiple consumers
use tokio::sync::broadcast;

async fn broadcast_example() {
    let (tx, _) = broadcast::channel::<String>(16);
    let mut rx1 = tx.subscribe();
    let mut rx2 = tx.subscribe();

    tx.send("hello everyone".into()).unwrap();
    println!("{}", rx1.recv().await.unwrap());
    println!("{}", rx2.recv().await.unwrap());
}

Streams

use futures::stream::{self, StreamExt};

async fn process_stream() {
    let doubled: Vec<u32> = stream::iter(0u32..10)
        .filter(|&x| async move { x % 2 == 0 })
        .map(|x| x * 2)
        .collect()
        .await;

    // Wrap mpsc receiver as a Stream
    use tokio_stream::wrappers::ReceiverStream;
    let (tx, rx) = tokio::sync::mpsc::channel::<i32>(16);
    let mut stream = ReceiverStream::new(rx);
    while let Some(item) = stream.next().await {
        println!("{item}");
    }
}

Shared State in Async Code

use std::sync::Arc;
use tokio::sync::{Mutex, RwLock};

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

impl AppState {
    async fn increment(&self) {
        let mut guard = self.counter.lock().await;
        *guard += 1;
    }

    async fn get_config(&self) -> Config {
        self.config.read().await.clone()
    }
}

Common Anti-Patterns

• Blocking in async context — use tokio::task::spawn_blocking for CPU-heavy or blocking I/O
• Holding MutexGuard across .await — use tokio::sync::Mutex, not std::sync::Mutex
• std::thread::sleep in async code — use tokio::time::sleep instead
• Spawning unlimited tasks — use a semaphore or bounded channel to limit concurrency
• Forgetting cancellation propagation — check CancellationToken or select! on shutdown signals