Adxptived Rust Skills Rust Async

Async Rust Programming

Comprehensive guide to async programming with Tokio, futures, and concurrent patterns.

Quick Navigation

• references/tokio.md - Tokio runtime deep dive
• references/patterns.md - Async patterns and anti-patterns
• references/channels.md - Channel types and message passing

Core Concepts

Async/Await Basics

// Async function returns a Future
async fn fetch_data(url: &str) -> Result<String, Error> {
    let response = reqwest::get(url).await?;
    let body = response.text().await?;
    Ok(body)
}

// Futures are lazy - nothing happens until awaited
let future = fetch_data("https://api.example.com");
// ... future not executed yet ...
let result = future.await; // Now it runs

The Tokio Runtime

// Main entry point
#[tokio::main]
async fn main() {
    println!("Hello, async world!");
}

// Equivalent to:
fn main() {
    tokio::runtime::Runtime::new()
        .unwrap()
        .block_on(async {
            println!("Hello, async world!");
        });
}

// Multi-threaded runtime (default)
#[tokio::main]
async fn main() { ... }

// Single-threaded runtime
#[tokio::main(flavor = "current_thread")]
async fn main() { ... }

// Custom runtime
#[tokio::main(worker_threads = 4)]
async fn main() { ... }

Spawning Tasks

tokio::spawn

Run tasks concurrently:

use tokio::task;

#[tokio::main]
async fn main() {
    // Spawn concurrent tasks
    let handle1 = task::spawn(async {
        // Task 1 work
        "result1"
    });
    
    let handle2 = task::spawn(async {
        // Task 2 work  
        "result2"
    });
    
    // Wait for both
    let (result1, result2) = tokio::join!(handle1, handle2);
    println!("{:?}, {:?}", result1, result2);
}

spawn_blocking

For CPU-bound work:

// Don't block async runtime with CPU work
let result = tokio::task::spawn_blocking(|| {
    // CPU-intensive computation
    expensive_hash_computation()
}).await?;

// Or for blocking I/O (non-async libraries)
let result = tokio::task::spawn_blocking(move || {
    std::fs::read_to_string("large_file.txt")
}).await??;

Task Types

Concurrency Primitives

tokio::join! - Run Concurrently, Wait for All

use tokio::join;

async fn fetch_all() -> Result<(User, Posts, Comments), Error> {
    let (user, posts, comments) = join!(
        fetch_user(),
        fetch_posts(),
        fetch_comments()
    );
    
    Ok((user?, posts?, comments?))
}

tokio::select! - First to Complete Wins

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

async fn with_timeout() -> Result<Data, Error> {
    select! {
        result = fetch_data() => result,
        _ = sleep(Duration::from_secs(5)) => {
            Err(Error::Timeout)
        }
    }
}

// Multiple branches
select! {
    msg = rx.recv() => handle_message(msg),
    _ = shutdown_signal() => break,
    _ = interval.tick() => do_periodic_work(),
}

FuturesUnordered - Dynamic Task Set

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

async fn process_urls(urls: Vec<String>) -> Vec<Result<String, Error>> {
    let mut futures = FuturesUnordered::new();
    
    for url in urls {
        futures.push(fetch(url));
    }
    
    let mut results = Vec::new();
    while let Some(result) = futures.next().await {
        results.push(result);
    }
    results
}

Channels

MPSC (Multi-Producer, Single-Consumer)

use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel::<String>(100);
    
    // Spawn producer
    let tx2 = tx.clone();
    tokio::spawn(async move {
        tx.send("Hello".to_string()).await.unwrap();
    });
    
    tokio::spawn(async move {
        tx2.send("World".to_string()).await.unwrap();
    });
    
    // Receive
    while let Some(msg) = rx.recv().await {
        println!("Got: {}", msg);
    }
}

Oneshot (Single Value)

use tokio::sync::oneshot;

async fn request_response() {
    let (tx, rx) = oneshot::channel();
    
    tokio::spawn(async move {
        let result = compute().await;
        tx.send(result).unwrap();
    });
    
    let result = rx.await.unwrap();
}

Broadcast (Multi-Consumer)

use tokio::sync::broadcast;

let (tx, _rx) = broadcast::channel::<Event>(100);

// Multiple receivers
let mut rx1 = tx.subscribe();
let mut rx2 = tx.subscribe();

tx.send(Event::Shutdown)?;

// Both receive the event

Watch (Single Value, Multiple Observers)

use tokio::sync::watch;

let (tx, rx) = watch::channel(Config::default());

// Update value
tx.send(new_config)?;

// Observers receive latest value
let current = rx.borrow().clone();

// Wait for changes
rx.changed().await?;

Channel Comparison

Synchronization

Mutex

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

let data = Arc::new(Mutex::new(vec![]));

let data2 = data.clone();
tokio::spawn(async move {
    let mut guard = data2.lock().await;
    guard.push(1);
}); // Lock released when guard drops

// Don't hold locks across .await when possible
let item = {
    let guard = data.lock().await;
    guard.first().cloned()
}; // Lock released before await
process(item).await;

RwLock

use tokio::sync::RwLock;

let data = RwLock::new(HashMap::new());

// Multiple readers
let reader1 = data.read().await;
let reader2 = data.read().await;

// Exclusive writer
let mut writer = data.write().await;
writer.insert("key", "value");

Semaphore

use tokio::sync::Semaphore;
use std::sync::Arc;

let semaphore = Arc::new(Semaphore::new(10)); // Max 10 concurrent

async fn limited_task(sem: Arc<Semaphore>) {
    let permit = sem.acquire().await.unwrap();
    // Do work with permit held
    do_work().await;
    // Permit dropped, slot freed
}

Async I/O

File Operations

use tokio::fs;
use tokio::io::{AsyncReadExt, AsyncWriteExt, BufReader, BufWriter};

// Read file
let content = fs::read_to_string("file.txt").await?;

// Write file
fs::write("output.txt", content).await?;

// Buffered I/O
let file = fs::File::open("large.txt").await?;
let mut reader = BufReader::new(file);
let mut line = String::new();
reader.read_line(&mut line).await?;

TCP

use tokio::net::{TcpListener, TcpStream};
use tokio::io::{AsyncReadExt, AsyncWriteExt};

// Server
let listener = TcpListener::bind("127.0.0.1:8080").await?;
loop {
    let (socket, addr) = listener.accept().await?;
    tokio::spawn(async move {
        handle_connection(socket).await;
    });
}

// Client
let mut stream = TcpStream::connect("127.0.0.1:8080").await?;
stream.write_all(b"Hello").await?;
let mut buf = [0u8; 1024];
let n = stream.read(&mut buf).await?;

Common Patterns

Graceful Shutdown

use tokio::signal;
use tokio::sync::watch;

#[tokio::main]
async fn main() {
    let (shutdown_tx, shutdown_rx) = watch::channel(false);
    
    let server = tokio::spawn(run_server(shutdown_rx.clone()));
    
    // Wait for Ctrl+C
    signal::ctrl_c().await.unwrap();
    
    // Signal shutdown
    shutdown_tx.send(true).unwrap();
    
    // Wait for server to finish
    server.await.unwrap();
}

async fn run_server(mut shutdown: watch::Receiver<bool>) {
    loop {
        select! {
            conn = accept_connection() => handle(conn).await,
            _ = shutdown.changed() => {
                if *shutdown.borrow() {
                    break;
                }
            }
        }
    }
}

Rate Limiting

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

async fn rate_limited_work() {
    let mut interval = interval(Duration::from_millis(100));
    
    for item in items {
        interval.tick().await; // 10 per second max
        process(item).await;
    }
}

Timeout

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

let result = timeout(
    Duration::from_secs(5),
    fetch_data()
).await;

match result {
    Ok(data) => println!("Got data: {:?}", data),
    Err(_) => println!("Timed out"),
}

Retry with Backoff

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

async fn retry_with_backoff<F, Fut, T, E>(
    mut f: F,
    max_retries: u32,
) -> Result<T, E>
where
    F: FnMut() -> Fut,
    Fut: Future<Output = Result<T, E>>,
{
    let mut delay = Duration::from_millis(100);
    
    for attempt in 0..max_retries {
        match f().await {
            Ok(result) => return Ok(result),
            Err(e) if attempt + 1 == max_retries => return Err(e),
            Err(_) => {
                sleep(delay).await;
                delay *= 2; // Exponential backoff
            }
        }
    }
    unreachable!()
}

Structured Concurrency: JoinSet

Manage a dynamic set of tasks with automatic cleanup:

use tokio::task::JoinSet;

async fn process_all(items: Vec<Item>) -> Vec<Result<Output, Error>> {
    let mut set = JoinSet::new();

    for item in items {
        set.spawn(async move { process(item).await });
    }

    let mut results = Vec::new();
    while let Some(result) = set.join_next().await {
        results.push(result.unwrap()); // unwrap JoinError (panic propagation)
    }
    results
}

// With concurrency limit
async fn limited_parallel(items: Vec<Item>, limit: usize) {
    let mut set = JoinSet::new();

    for item in items {
        // Keep at most `limit` tasks running
        if set.len() >= limit {
            set.join_next().await;
        }
        set.spawn(async move { process(item).await });
    }

    // Drain remaining
    while set.join_next().await.is_some() {}
}

Cancellation

CancellationToken (tokio-util)

use tokio_util::sync::CancellationToken;
use tokio::time::{sleep, Duration};

#[tokio::main]
async fn main() {
    let token = CancellationToken::new();

    // Worker task respects cancellation
    let worker_token = token.clone();
    let worker = tokio::spawn(async move {
        loop {
            tokio::select! {
                _ = worker_token.cancelled() => {
                    println!("Worker shutting down cleanly");
                    break;
                }
                _ = do_work() => {}
            }
        }
    });

    // Let it run for 2 seconds, then cancel
    sleep(Duration::from_secs(2)).await;
    token.cancel(); // Notifies all clones

    worker.await.unwrap();
}

// Dropping handle vs cancelling:
let handle = tokio::spawn(async { loop { work().await; } });
drop(handle);        // Task CONTINUES in background (detached)!
token.cancel();      // Cooperative: task stops when it checks

Timeout as Cancellation

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

// Task is dropped (cancelled) if it doesn't finish in time
match timeout(Duration::from_secs(5), long_running_task()).await {
    Ok(result) => println!("Completed: {:?}", result),
    Err(_elapsed) => println!("Timed out — task was cancelled"),
}

Anti-Patterns

Holding Locks Across Await

// BAD: Lock held across await
let guard = mutex.lock().await;
some_async_operation().await; // Lock still held!
drop(guard);

// GOOD: Clone and release
let data = {
    let guard = mutex.lock().await;
    guard.clone()
};
some_async_operation_with(data).await;

Blocking in Async Context

// BAD: Blocks the async runtime
async fn bad() {
    std::thread::sleep(Duration::from_secs(1)); // Blocks!
    std::fs::read_to_string("file.txt"); // Blocks!
}

// GOOD: Use async equivalents
async fn good() {
    tokio::time::sleep(Duration::from_secs(1)).await;
    tokio::fs::read_to_string("file.txt").await;
}

// Or spawn_blocking for unavoidable blocking
let result = tokio::task::spawn_blocking(|| {
    blocking_library_call()
}).await?;

Creating Runtime in Async Context

// BAD: Nested runtime
async fn bad() {
    tokio::runtime::Runtime::new().unwrap()
        .block_on(async { ... }); // Panic or deadlock!
}

// GOOD: Just await
async fn good() {
    some_future().await;
}

Best Practices from the Field

1. Always Prefer Bounded Channels

Avoid unbounded channels (like tokio::sync::mpsc::unbounded_channel) in production pipelines. Without backpressure, slow consumers will cause memory leaks. Always specify a bound (capacity):

// GOOD: Capacity is capped; send will block cooperatively or fail when full
let (tx, mut rx) = tokio::sync::mpsc::channel(100);

2. Use JoinSet for Dynamic Task Groups

Instead of collecting handles and iterating or using unstable futures::future::join_all, use tokio::task::JoinSet to manage lifetimes of dynamically spawned concurrent workers.

use tokio::task::JoinSet;

let mut set = JoinSet::new();
for i in 0..10 {
    set.spawn(async move { i * 2 });
}

while let Some(res) = set.join_next().await {
    println!("Task finished: {:?}", res?);
}

Debugging Tips

1. Use #[tokio::test] for async tests
2. Enable tracing with tracing crate
3. Use tokio-console for runtime inspection (cargo install tokio-console)
4. Check for dropped futures (no .await)
5. Look for blocking calls in async code
6. CancellationToken over Arc<AtomicBool> — atomic flags don't wake sleeping tasks

Production Checklist

• Bound channels and queues to preserve backpressure.
• Use JoinSet or structured task ownership for dynamic task groups.
• Add cancellation paths for long-running tasks.
• Avoid blocking calls on runtime worker threads; use async APIs or spawn_blocking.
• Instrument tasks with tracing spans and inspect with tokio-console when needed.
• Test shutdown and cancellation, not only happy-path completion.

References

• Tokio tutorial
• Async Rust book
• tokio-util CancellationToken docs
• tokio-console

Source: adxptived/Rust-Skills — distributed by TomeVault.