Async Pitfalls in Production

use std::time::{Duration, Instant};

async fn handler_blocking(id: u32) -> u32 {
    std::thread::sleep(Duration::from_millis(100)); // blocks the worker thread
    id
}

async fn handler_async(id: u32) -> u32 {
    tokio::time::sleep(Duration::from_millis(100)).await; // yields the worker
    id
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let start = Instant::now();
    let tasks: Vec<_> = (0..5).map(|i| tokio::spawn(handler_blocking(i))).collect();
    for t in tasks {
        t.await.unwrap();
    }
    let blocking = start.elapsed();

    let start = Instant::now();
    let tasks: Vec<_> = (0..5).map(|i| tokio::spawn(handler_async(i))).collect();
    for t in tasks {
        t.await.unwrap();
    }
    let non_blocking = start.elapsed();

    println!("5 blocking handlers took at least 500 ms: {}", blocking >= Duration::from_millis(500));
    println!("5 async handlers took under 200 ms: {}", non_blocking < Duration::from_millis(200));
}

use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{Duration, Instant};

async fn heartbeat(max_lag_ms: Arc<AtomicU64>) {
    let period = Duration::from_millis(10);
    loop {
        let before = Instant::now();
        tokio::time::sleep(period).await;
        let lag = before.elapsed().saturating_sub(period);
        max_lag_ms.fetch_max(lag.as_millis() as u64, Ordering::Relaxed);
    }
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let max_lag = Arc::new(AtomicU64::new(0));
    let hb = tokio::spawn(heartbeat(Arc::clone(&max_lag)));
    tokio::time::sleep(Duration::from_millis(50)).await;
    println!("lag with a healthy runtime under 20 ms: {}", max_lag.load(Ordering::Relaxed) < 20);

    // A request handler that parses a big payload synchronously:
    tokio::spawn(async {
        std::thread::sleep(Duration::from_millis(150)); // stands in for 150 ms of CPU work
    })
    .await
    .unwrap();
    tokio::time::sleep(Duration::from_millis(30)).await;
    println!("lag after a blocking task at least 100 ms: {}", max_lag.load(Ordering::Relaxed) >= 100);
    hb.abort();
}

use tokio::sync::oneshot;

fn expensive_checksum(data: &[u8]) -> u64 {
    data.iter().fold(0xcbf29ce484222325u64, |h, &b| (h ^ b as u64).wrapping_mul(0x100000001b3))
}

/// Runs `f` on rayon's thread pool and awaits the result without blocking.
async fn run_cpu<T: Send + 'static>(f: impl FnOnce() -> T + Send + 'static) -> T {
    let (tx, rx) = oneshot::channel();
    rayon::spawn(move || {
        let _ = tx.send(f());
    });
    rx.await.expect("the rayon task panicked")
}

#[tokio::main]
async fn main() {
    let payloads: Vec<Vec<u8>> = (0..8).map(|i| vec![i as u8; 2_000_000]).collect();
    let handles: Vec<_> = payloads.into_iter().map(|p| tokio::spawn(run_cpu(move || expensive_checksum(&p)))).collect();
    let mut sums = Vec::new();
    for h in handles {
        sums.push(h.await.unwrap());
    }
    println!("{} checksums, all distinct: {}", sums.len(), sums.iter().collect::<std::collections::HashSet<_>>().len() == sums.len());
}

use std::sync::{Arc, Mutex};
use std::time::Duration;

async fn save(v: u64) -> u64 {
    tokio::time::sleep(Duration::from_millis(1)).await;
    v
}

#[tokio::main]
async fn main() {
    // Fix 1: std Mutex, guard dropped before the .await.
    let counter = Arc::new(Mutex::new(0u64));
    let tasks: Vec<_> = (0..10)
        .map(|_| {
            let c = Arc::clone(&counter);
            tokio::spawn(async move {
                let value = {
                    let mut n = c.lock().unwrap();
                    *n += 1;
                    *n
                }; // guard dropped here
                save(value).await
            })
        })
        .collect();
    for t in tasks {
        t.await.unwrap();
    }
    println!("std mutex: {}", counter.lock().unwrap());

    // Fix 2: tokio Mutex, when the critical section itself must await.
    let conn = Arc::new(tokio::sync::Mutex::new(Vec::<u64>::new()));
    let tasks: Vec<_> = (0..10)
        .map(|i| {
            let conn = Arc::clone(&conn);
            tokio::spawn(async move {
                let mut log = conn.lock().await;
                let saved = save(i).await; // allowed: the guard is Send
                log.push(saved);
            })
        })
        .collect();
    for t in tasks {
        t.await.unwrap();
    }
    println!("tokio mutex: {} writes", conn.lock().await.len());
}

use std::collections::HashMap;
use tokio::sync::{mpsc, oneshot};

enum Command {
    Add { user: String, amount: i64 },
    Balance { user: String, reply: oneshot::Sender<i64> },
}

async fn ledger_actor(mut inbox: mpsc::Receiver<Command>) {
    let mut balances: HashMap<String, i64> = HashMap::new(); // owned by this task only
    while let Some(cmd) = inbox.recv().await {
        match cmd {
            Command::Add { user, amount } => *balances.entry(user).or_default() += amount,
            Command::Balance { user, reply } => {
                let _ = reply.send(balances.get(&user).copied().unwrap_or(0));
            }
        }
    }
}

#[tokio::main]
async fn main() {
    let (tx, rx) = mpsc::channel(64);
    let actor = tokio::spawn(ledger_actor(rx));

    let writers: Vec<_> = (0..4)
        .map(|_| {
            let tx = tx.clone();
            tokio::spawn(async move {
                for _ in 0..25 {
                    tx.send(Command::Add { user: "ana".into(), amount: 10 }).await.unwrap();
                }
            })
        })
        .collect();
    for w in writers {
        w.await.unwrap();
    }

    let (reply, answer) = oneshot::channel();
    tx.send(Command::Balance { user: "ana".into(), reply }).await.unwrap();
    println!("ana: {}", answer.await.unwrap());

    drop(tx); // closing the channel stops the actor
    actor.await.unwrap();
}

use std::collections::VecDeque;
use std::sync::{Arc, Mutex};
use std::time::Duration;

type Queue = Arc<Mutex<VecDeque<u32>>>;

/// NOT cancel-safe: it removes the job first, then awaits.
async fn take_and_process(queue: Queue) -> Option<u32> {
    let job = queue.lock().unwrap().pop_front()?;
    tokio::time::sleep(Duration::from_millis(30)).await; // "process" the job
    Some(job)
}

#[tokio::main]
async fn main() {
    let queue: Queue = Arc::new(Mutex::new((1..=5).collect()));
    let mut ticker = tokio::time::interval(Duration::from_millis(10));
    let mut done = Vec::new();

    loop {
        tokio::select! {
            job = take_and_process(Arc::clone(&queue)) => match job {
                Some(j) => done.push(j),
                None => break,
            },
            _ = ticker.tick() => {} // e.g. report metrics every 10 ms
        }
    }
    println!("completed jobs: {done:?}");
    println!("left in queue: {:?}", queue.lock().unwrap());
}

use std::collections::VecDeque;
use std::sync::{Arc, Mutex};
use std::time::Duration;

type Queue = Arc<Mutex<VecDeque<u32>>>;

async fn take_and_process(queue: Queue) -> Option<u32> {
    let job = queue.lock().unwrap().pop_front()?;
    tokio::time::sleep(Duration::from_millis(30)).await;
    Some(job)
}

#[tokio::main]
async fn main() {
    let queue: Queue = Arc::new(Mutex::new((1..=5).collect()));
    let mut ticker = tokio::time::interval(Duration::from_millis(10));
    let mut done = Vec::new();
    let mut ticks = 0;

    let mut current = Box::pin(take_and_process(Arc::clone(&queue)));
    loop {
        tokio::select! {
            job = &mut current => match job { // poll the same future: it keeps its progress
                Some(j) => {
                    done.push(j);
                    current = Box::pin(take_and_process(Arc::clone(&queue))); // start the next one
                }
                None => break,
            },
            _ = ticker.tick() => ticks += 1,
        }
    }
    println!("completed jobs: {done:?}");
    println!("ticker still ran: {}", ticks > 5);
}

use std::sync::{Arc, Mutex};
use std::time::Duration;
use tokio::time::timeout;

#[derive(Default, Debug)]
struct Accounts {
    alice: i64,
    bob: i64,
}

async fn slow_network_call() {
    tokio::time::sleep(Duration::from_millis(100)).await;
}

async fn transfer(accounts: Arc<Mutex<Accounts>>, amount: i64) {
    accounts.lock().unwrap().alice -= amount;
    slow_network_call().await; // e.g. notify a fraud-check service
    accounts.lock().unwrap().bob += amount; // never runs if the future is dropped above
}

#[tokio::main]
async fn main() {
    let accounts = Arc::new(Mutex::new(Accounts { alice: 100, bob: 0 }));

    // The client gives up after 20 ms, so the handler future is dropped.
    let result = timeout(Duration::from_millis(20), transfer(Arc::clone(&accounts), 30)).await;
    println!("timed out: {}", result.is_err());
    println!("after a cancelled request: {:?}", accounts.lock().unwrap());

    // Fix: run work that must complete as its own task; the request only waits for it.
    let accounts = Arc::new(Mutex::new(Accounts { alice: 100, bob: 0 }));
    let job = tokio::spawn(transfer(Arc::clone(&accounts), 30));
    let result = timeout(Duration::from_millis(20), async { job.await.unwrap() }).await;
    println!("timed out: {}", result.is_err());
    tokio::time::sleep(Duration::from_millis(150)).await; // the spawned task finishes on its own
    println!("with a spawned task:        {:?}", accounts.lock().unwrap());
}

use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::Duration;
use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    // Unbounded: the producer finishes immediately, everything piles up in memory.
    let (tx, mut rx) = mpsc::unbounded_channel::<Vec<u8>>();
    for _ in 0..10_000 {
        tx.send(vec![0u8; 1024]).unwrap();
    }
    println!("unbounded: {} messages ({} KiB) waiting", rx.len(), rx.len());
    while rx.try_recv().is_ok() {}

    // Bounded: at most 100 messages in flight, the producer waits for the consumer.
    let (tx, mut rx) = mpsc::channel::<Vec<u8>>(100);
    let max_seen = Arc::new(AtomicUsize::new(0));
    let seen = Arc::clone(&max_seen);
    let consumer = tokio::spawn(async move {
        let mut n = 0;
        while let Some(_msg) = rx.recv().await {
            seen.fetch_max(rx.len(), Ordering::Relaxed);
            n += 1;
            if n % 1000 == 0 {
                tokio::time::sleep(Duration::from_millis(1)).await; // a slow consumer
            }
        }
        n
    });
    for _ in 0..10_000 {
        tx.send(vec![0u8; 1024]).await.unwrap(); // waits when 100 are queued
    }
    drop(tx);
    println!("bounded: consumed {}, queue never above {}: {}", consumer.await.unwrap(), 100, max_seen.load(Ordering::Relaxed) <= 100);
}

use futures::stream::{self, StreamExt};
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::Duration;

async fn fetch(id: u32, in_flight: Arc<AtomicUsize>, peak: Arc<AtomicUsize>) -> u32 {
    let now = in_flight.fetch_add(1, Ordering::SeqCst) + 1;
    peak.fetch_max(now, Ordering::SeqCst);
    tokio::time::sleep(Duration::from_millis(5)).await; // a remote call
    in_flight.fetch_sub(1, Ordering::SeqCst);
    id * 2
}

#[tokio::main]
async fn main() {
    let in_flight = Arc::new(AtomicUsize::new(0));
    let peak = Arc::new(AtomicUsize::new(0));

    let results: Vec<u32> = stream::iter(0..200)
        .map(|id| fetch(id, Arc::clone(&in_flight), Arc::clone(&peak)))
        .buffer_unordered(16) // at most 16 requests at a time
        .collect()
        .await;

    println!("fetched {} items", results.len());
    println!("peak concurrency: {}", peak.load(Ordering::SeqCst));
}

use std::sync::Arc;
use std::sync::atomic::{AtomicU32, Ordering};
use std::time::Duration;
use tokio::task::JoinSet;

async fn poll_forever(counter: Arc<AtomicU32>) {
    loop {
        counter.fetch_add(1, Ordering::Relaxed);
        tokio::time::sleep(Duration::from_millis(5)).await;
    }
}

#[tokio::main]
async fn main() {
    let detached = Arc::new(AtomicU32::new(0));
    {
        let _handle = tokio::spawn(poll_forever(Arc::clone(&detached)));
    } // handle dropped: the task keeps running

    let owned = Arc::new(AtomicU32::new(0));
    {
        let mut set = JoinSet::new();
        set.spawn(poll_forever(Arc::clone(&owned)));
        tokio::time::sleep(Duration::from_millis(20)).await;
    } // JoinSet dropped: its tasks are aborted

    let (d1, o1) = (detached.load(Ordering::Relaxed), owned.load(Ordering::Relaxed));
    tokio::time::sleep(Duration::from_millis(50)).await;
    let (d2, o2) = (detached.load(Ordering::Relaxed), owned.load(Ordering::Relaxed));

    println!("detached task still running: {}", d2 > d1);
    println!("JoinSet task stopped: {}", o2 == o1);
}

use std::sync::atomic::{AtomicU32, Ordering};

static AUDIT_EVENTS: AtomicU32 = AtomicU32::new(0);

async fn audit(_event: &str) {
    AUDIT_EVENTS.fetch_add(1, Ordering::Relaxed);
}

#[tokio::main]
async fn main() {
    let _ = audit("login"); // BUG: creates the future and drops it
    audit("logout").await; // runs
    tokio::spawn(audit("password change")).await.unwrap(); // runs, as its own task

    println!("audit events recorded: {}", AUDIT_EVENTS.load(Ordering::Relaxed));
}

use std::mem::size_of_val;

async fn tiny() -> u8 {
    tokio::task::yield_now().await;
    1
}

async fn big_buffer() -> usize {
    let buf = [0u8; 64 * 1024]; // lives across the .await below
    tokio::task::yield_now().await;
    buf.len()
}

async fn boxed_buffer() -> usize {
    let buf = vec![0u8; 64 * 1024]; // only a pointer lives in the future
    tokio::task::yield_now().await;
    buf.len()
}

async fn nested() -> usize {
    big_buffer().await + big_buffer().await // the parent embeds its children's state
}

#[tokio::main]
async fn main() {
    println!("tiny:         {} bytes", size_of_val(&tiny()));
    println!("big_buffer:   {} bytes", size_of_val(&big_buffer()));
    println!("boxed_buffer: {} bytes", size_of_val(&boxed_buffer()));
    println!("nested:       {} bytes", size_of_val(&nested()));
    println!("Box::pin(nested()): {} bytes", size_of_val(&Box::pin(nested())));
}

use std::time::Duration;
use tokio::time::{Instant, timeout, timeout_at};

async fn call(service: &str, latency_ms: u64) -> String {
    tokio::time::sleep(Duration::from_millis(latency_ms)).await;
    format!("{service} ok")
}

#[tokio::main]
async fn main() {
    let deadline = Instant::now() + Duration::from_millis(100); // the whole request's budget

    let user = timeout(Duration::from_millis(50), call("users", 10)).await;
    println!("users: {user:?}");

    let prices = timeout(Duration::from_millis(50), call("prices", 5_000)).await; // a hung dependency
    println!("prices: {}", prices.map_err(|_| "timed out after 50 ms").unwrap_err());

    let rest = timeout_at(deadline, call("recommendations", 80)).await; // only ~40 ms left
    println!("recommendations: {}", rest.map(|_| "ok").unwrap_or("skipped: request deadline reached"));
}

use std::time::Duration;
use tokio::task::JoinSet;
use tokio_util::sync::CancellationToken;

async fn worker(id: u32, token: CancellationToken) -> String {
    let mut processed = 0;
    loop {
        tokio::select! {
            _ = token.cancelled() => break, // stop taking new work
            _ = tokio::time::sleep(Duration::from_millis(5)) => processed += 1, // one unit of work
        }
    }
    // Cleanup that must happen: flush buffers, commit offsets, close connections.
    format!("worker {id} stopped cleanly after {} jobs", if processed > 0 { "some" } else { "no" })
}

#[tokio::main]
async fn main() {
    let token = CancellationToken::new();
    let mut workers = JoinSet::new();
    for id in 1..=3 {
        workers.spawn(worker(id, token.child_token()));
    }

    // In a real service: tokio::signal::ctrl_c().await, or a SIGTERM handler.
    tokio::time::sleep(Duration::from_millis(30)).await;
    println!("shutdown requested");
    token.cancel();

    let drained = tokio::time::timeout(Duration::from_secs(5), async {
        let mut reports = Vec::new();
        while let Some(r) = workers.join_next().await {
            reports.push(r.unwrap());
        }
        reports.sort();
        reports
    })
    .await;
    match drained {
        Ok(reports) => reports.iter().for_each(|r| println!("{r}")),
        Err(_) => println!("shutdown deadline hit; remaining tasks are aborted"),
    }
}