Chapter 4: Async Deep Dive

Asynchronous programming is essential for high-performance I/O-bound applications. This chapter explores how Rust's async/await works under the hood and how to use it effectively.

4.1 Understanding Futures

At the core of async Rust is the Future trait. Let's understand what it really means:

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

// The actual Future trait (simplified)
pub trait SimpleFuture {
    type Output;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output>;
}

// Poll is an enum
pub enum Poll<T> {
    Ready(T),      // Future completed with value
    Pending,       // Future not ready yet
}

Creating a Simple Future

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

struct Delay {
    when: Instant,
}

impl Delay {
    fn new(duration: Duration) -> Self {
        Delay {
            when: Instant::now() + duration,
        }
    }
}

impl Future for Delay {
    type Output = ();

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        if Instant::now() >= self.when {
            Poll::Ready(())
        } else {
            // Tell the executor to poll again
            cx.waker().wake_by_ref();
            Poll::Pending
        }
    }
}

#[tokio::main]
async fn main() {
    let delay = Delay::new(Duration::from_secs(1));
    delay.await;
    println!("Done!");
}

4.2 Async/Await Syntax

Basic Async Functions

// Async function
async fn fetch_data(url: &str) -> Result<String, String> {
    // Simulated async work
    tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
    Ok(format!("Data from {}", url))
}

// Async block
fn create_future() -> impl Future<Output = i32> {
    async {
        42
    }
}

// Using async/await
#[tokio::main]
async fn main() {
    // Sequential
    let data1 = fetch_data("api.example.com/1").await.unwrap();
    let data2 = fetch_data("api.example.com/2").await.unwrap();

    println!("{}\n{}", data1, data2);
}

Concurrent Execution

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

async fn task1() -> String {
    sleep(Duration::from_millis(100)).await;
    "Task 1 done".to_string()
}

async fn task2() -> String {
    sleep(Duration::from_millis(100)).await;
    "Task 2 done".to_string()
}

#[tokio::main]
async fn main() {
    // Run concurrently (not parallel - same thread!)
    let (result1, result2) = join!(task1(), task2());

    println!("{}", result1);
    println!("{}", result2);
}

// For dynamic number of tasks
async fn fetch_all(urls: Vec<String>) -> Vec<Result<String, String>> {
    let tasks: Vec<_> = urls
        .into_iter()
        .map(|url| tokio::spawn(async move {
            fetch_data(&url).await
        }))
        .collect();

    let mut results = Vec::new();
    for task in tasks {
        results.push(task.await.unwrap());
    }
    results
}

4.3 Tokio Runtime Deep Dive

Runtime Configuration

use tokio::runtime::Runtime;

fn main() {
    // Multi-threaded runtime (default)
    let runtime = Runtime::new().unwrap();

    runtime.block_on(async {
        println!("Running on multi-threaded runtime");
    });

    // Current thread runtime (single-threaded)
    let runtime = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();

    runtime.block_on(async {
        println!("Running on current thread");
    });

    // Custom configuration
    let runtime = tokio::runtime::Builder::new_multi_thread()
        .worker_threads(4)
        .thread_name("my-worker")
        .thread_stack_size(3 * 1024 * 1024)
        .enable_all()
        .build()
        .unwrap();
}

Spawning Tasks

use tokio::task;

#[tokio::main]
async fn main() {
    // Spawn a task
    let handle = task::spawn(async {
        println!("Task running on tokio thread pool");
        42
    });

    let result = handle.await.unwrap();
    println!("Result: {}", result);

    // Spawn blocking (for CPU-intensive work)
    let handle = task::spawn_blocking(|| {
        // This runs on a dedicated thread pool
        println!("CPU-intensive work");
        expensive_computation()
    });

    handle.await.unwrap();
}

fn expensive_computation() -> u64 {
    (0..1_000_000).sum()
}

4.4 Channels and Message Passing

MPSC Channels

use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel(32);

    // Spawn sender task
    tokio::spawn(async move {
        for i in 0..10 {
            tx.send(i).await.unwrap();
            tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
        }
    });

    // Receive in main task
    while let Some(value) = rx.recv().await {
        println!("Received: {}", value);
    }
}

// Bounded vs Unbounded
async fn channel_types() {
    // Bounded: sender waits if full
    let (tx, rx) = mpsc::channel::<i32>(100);

    // Unbounded: never blocks (use carefully!)
    let (tx_unbounded, rx_unbounded) = mpsc::unbounded_channel::<i32>();
}

Broadcast and Watch Channels

use tokio::sync::{broadcast, watch};

#[tokio::main]
async fn main() {
    // Broadcast: multiple receivers
    let (tx, _rx) = broadcast::channel(16);

    let mut rx1 = tx.subscribe();
    let mut rx2 = tx.subscribe();

    tokio::spawn(async move {
        tx.send("Hello").unwrap();
    });

    println!("rx1: {}", rx1.recv().await.unwrap());
    println!("rx2: {}", rx2.recv().await.unwrap());

    // Watch: single-value state channel
    let (tx, mut rx) = watch::channel("initial");

    tokio::spawn(async move {
        tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
        tx.send("updated").unwrap();
    });

    rx.changed().await.unwrap();
    println!("Value changed to: {}", *rx.borrow());
}

4.5 Synchronization Primitives

Async Mutex

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

#[tokio::main]
async fn main() {
    let counter = Arc::new(Mutex::new(0));
    let mut handles = vec![];

    for _ in 0..10 {
        let counter = Arc::clone(&counter);
        let handle = tokio::spawn(async move {
            let mut num = counter.lock().await;
            *num += 1;
        });
        handles.push(handle);
    }

    for handle in handles {
        handle.await.unwrap();
    }

    println!("Counter: {}", *counter.lock().await);
}

RwLock and Semaphore

use tokio::sync::{RwLock, Semaphore};

#[tokio::main]
async fn main() {
    // RwLock: multiple readers or single writer
    let lock = Arc::new(RwLock::new(5));

    let read = lock.read().await;
    println!("Value: {}", *read);
    drop(read);

    let mut write = lock.write().await;
    *write += 1;

    // Semaphore: limit concurrent access
    let semaphore = Arc::new(Semaphore::new(3));

    let mut handles = vec![];
    for i in 0..10 {
        let sem = semaphore.clone();
        handles.push(tokio::spawn(async move {
            let _permit = sem.acquire().await.unwrap();
            println!("Task {} acquired permit", i);
            tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
        }));
    }

    for handle in handles {
        handle.await.unwrap();
    }
}

4.6 Select! Macro

Race multiple futures and handle whichever completes first:

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

#[tokio::main]
async fn main() {
    let mut interval = tokio::time::interval(Duration::from_secs(1));

    select! {
        _ = sleep(Duration::from_secs(5)) => {
            println!("5 second timeout");
        }
        _ = interval.tick() => {
            println!("Interval fired");
        }
    }

    // With cancellation
    let operation = async {
        sleep(Duration::from_secs(10)).await;
        "Completed"
    };

    let timeout = sleep(Duration::from_secs(2));

    select! {
        result = operation => {
            println!("Operation: {}", result);
        }
        _ = timeout => {
            println!("Timed out!");
        }
    }
}

4.7 Stream Processing

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

#[tokio::main]
async fn main() {
    // Create a stream
    let mut stream = stream::iter(vec![1, 2, 3, 4, 5]);

    // Process with combinator methods
    while let Some(value) = stream.next().await {
        println!("{}", value);
    }

    // Advanced stream processing
    let stream = stream::iter(1..=10)
        .filter(|x| x % 2 == 0)
        .map(|x| x * 2);

    let results: Vec<_> = stream.collect().await;
    println!("{:?}", results);

    // Merge multiple streams
    let s1 = stream::iter(vec![1, 2, 3]);
    let s2 = stream::iter(vec![4, 5, 6]);
    let merged = stream::iter(vec![s1, s2]).flatten();
}

4.8 Error Handling in Async

use tokio::try_join;

async fn fallible_task1() -> Result<i32, String> {
    Ok(1)
}

async fn fallible_task2() -> Result<i32, String> {
    Err("Task 2 failed".to_string())
}

#[tokio::main]
async fn main() {
    // Early return on first error
    match try_join!(fallible_task1(), fallible_task2()) {
        Ok((r1, r2)) => println!("Both succeeded: {}, {}", r1, r2),
        Err(e) => println!("Error: {}", e),
    }
}

// Timeout pattern
use tokio::time::{timeout, Duration};

async fn with_timeout() -> Result<String, String> {
    match timeout(Duration::from_secs(5), slow_operation()).await {
        Ok(result) => result,
        Err(_) => Err("Timeout!".to_string()),
    }
}

async fn slow_operation() -> Result<String, String> {
    tokio::time::sleep(Duration::from_secs(10)).await;
    Ok("Done".to_string())
}

4.9 Async Best Practices

4.10 Chapter Summary