Advanced Topics: Concurrency and Asynchronous Programming

Concurrency (doing multiple things at the same time) and asynchronous programming (doing multiple things in an overlapping manner, without necessarily at the exact same time) are critical for building high-performance and responsive applications. Rust tackles these complex domains with a unique “fearless concurrency” model, leveraging its ownership and type system to prevent common concurrency bugs like data races at compile time.

This chapter introduces you to Rust’s concurrency primitives and then dives into the modern asynchronous programming landscape with the async/await syntax and the popular Tokio runtime.

Threads and Fearless Concurrency

Rust’s standard library provides basic threading capabilities. However, its real power comes from how the compiler ensures safety when threads interact.

Creating Threads

You create new threads using std::thread::spawn. The closure passed to spawn is the code that the new thread will execute.

use std::thread;
use std::time::Duration;

fn main() {
    let handle = thread::spawn(|| { // `spawn` returns a `JoinHandle`
        for i in 1..=5 {
            println!("Hi number {} from the spawned thread!", i);
            thread::sleep(Duration::from_millis(1));
        }
    });

    for i in 1..=3 {
        println!("Hi number {} from the main thread!", i);
        thread::sleep(Duration::from_millis(1));
    }

    // `join()` blocks the current thread until the spawned thread finishes.
    // This ensures the spawned thread completes its work before the main thread exits.
    handle.join().unwrap(); // Wait for the spawned thread to finish
    println!("Spawned thread finished.");
}

`move` Closures with Threads

When using closures with thread::spawn, you might need the move keyword. move transfers ownership of the variables used in the closure’s body to the new thread, preventing potential lifetime issues or unintended shared mutable state.

use std::thread;

fn main() {
    let v = vec![1, 2, 3];

    // The `move` keyword forces the closure to take ownership of `v`.
    // Without `move`, `v` would be borrowed, and the compiler couldn't guarantee
    // that `v` outlives the spawned thread.
    let handle = thread::spawn(move || {
        println!("Vector from spawned thread: {:?}", v);
    });

    // println!("Vector from main thread: {:?}", v); // ERROR: `v` was moved!

    handle.join().unwrap();
}

Message Passing with Channels

A common and safe way to share data between threads is message passing. Rust’s standard library provides Multiple Producer, Single Consumer (MPSC) channels.

mpsc::channel() creates a (Sender, Receiver) pair.
Senders can be cloned, allowing multiple producers.
Receiver blocks until a message is available or the channel is closed.

use std::sync::mpsc;
use std::thread;
use std::time::Duration;

fn main() {
    let (tx, rx) = mpsc::channel(); // Create a new MPSC channel

    thread::spawn(move || {
        let vals = vec![
            String::from("hi"),
            String::from("from"),
            String::from("the"),
            String::from("thread"),
        ];

        for val in vals {
            tx.send(val).unwrap(); // Send a value (ownership is moved)
            thread::sleep(Duration::from_millis(500));
        }
    });

    for received in rx { // `rx` is an iterator
        println!("Got: {}", received);
    }
}

This pattern inherently prevents data races because each piece of data is owned by only one thread at a time—ownership is moved when a message is sent.

Shared-State Concurrency with `Mutex` and `Arc`

Sometimes, threads need to share access to the same mutable data. This is where Mutex (Mutual Exclusion) and Arc (Atomic Reference Count) come in.

Mutex<T>: Allows only one thread to access the inner T data at a time. To access the data, a thread must acquire a lock.
Arc<T>: An atomically reference counted smart pointer. It allows multiple threads to share ownership of data. When the last Arc goes out of scope, the data is cleaned up.

When you need to share mutable state between multiple threads, you typically combine Arc and Mutex.

use std::sync::{Arc, Mutex};
use std::thread;

fn main() {
    // Create an Arc of a Mutex around a Vec.
    // Arc allows multiple ownership across threads.
    // Mutex provides interior mutability and ensures exclusive access.
    let counter = Arc::new(Mutex::new(0));
    let mut handles = vec![];

    for _ in 0..10 {
        let counter_clone = Arc::clone(&counter); // Clone the Arc for each thread
        let handle = thread::spawn(move || {
            let mut num = counter_clone.lock().unwrap(); // Acquire the lock
            *num += 1; // Mutate the protected data
            // Lock is automatically released when `num` goes out of scope
        });
        handles.push(handle);
    }

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

    println!("Result: {}", *counter.lock().unwrap()); // Output: Result: 10
}

Arc::clone(&counter): Increments the reference count.
counter_clone.lock().unwrap(): Attempts to acquire the lock. If another thread holds the lock, this call blocks until the lock is available. unwrap() is used here for simplicity; in real-world code, you’d handle potential PoisonError if a thread holding the lock panicked.

Asynchronous Programming with `async`/`await` and Tokio

Asynchronous programming in Rust uses async/await syntax to make asynchronous code look and feel like synchronous code, while still performing non-blocking operations. The actual execution is handled by an asynchronous runtime, like Tokio.

What is `async`/`await`?

async fn: Declares an asynchronous function that returns a Future. A Future is a value that represents a computation that may not have completed yet.
await: Pauses the execution of the async function until the Future it’s waiting on completes, without blocking the underlying thread. While await is waiting, the runtime can execute other tasks.

The Tokio Runtime

Tokio is the most popular asynchronous runtime for Rust. It provides a multi-threaded scheduler, I/O primitives, and other utilities for building highly concurrent and performant network applications.

To use Tokio, add it to your Cargo.toml:

# Cargo.toml
[dependencies]
tokio = { version = "1", features = ["full"] } # "full" includes many common features

Then, annotate your main function (or test functions) with #[tokio::main].

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

// The `#[tokio::main]` attribute transforms `main` into an async function
// and sets up a Tokio runtime.
#[tokio::main]
async fn main() {
    println!("Hello from async main!");

    // Spawn an asynchronous task
    let task1 = tokio::spawn(async {
        some_async_task(1).await;
    });

    let task2 = tokio::spawn(async {
        some_async_task(2).await;
    });

    // Await the tasks to ensure they complete
    // In a real application, you might not always await them directly in main
    // but rather manage their lifetimes through channels or other means.
    let _ = tokio::join!(task1, task2); // Await multiple futures concurrently
    // task1.await.unwrap(); // Or await individually
    // task2.await.unwrap();

    println!("All async tasks finished.");
}

async fn some_async_task(id: u8) {
    println!("Task {} starting...", id);
    sleep(Duration::from_millis(1000 - (id * 100) as u64)).await; // Simulate work
    println!("Task {} finished!", id);
}

When you run this, you’ll observe the tasks potentially interleaving, demonstrating non-blocking execution.

Async I/O

Tokio provides its own non-blocking versions of standard library I/O. For example, tokio::fs for file operations, tokio::net for networking.

use tokio::{fs, io::{self, AsyncReadExt}};

#[tokio::main]
async fn main() -> io::Result<()> {
    let mut file = fs::File::open("async_file.txt").await?; // Non-blocking file open
    let mut contents = String::new();
    file.read_to_string(&mut contents).await?; // Non-blocking read

    println!("File contents: {}", contents);

    Ok(())
}

To test this, create an async_file.txt in your project root with some content.

Exercises / Mini-Challenges

Exercise 9.1: Multi-threaded Counter with `Arc<Mutex>`

Modify the Arc<Mutex> example to create 5 threads, each incrementing a shared counter 1000 times. Print the final count.

Instructions:

Initialize an Arc<Mutex<i32>> with a starting value of 0.
Create a Vec to hold JoinHandles.
In a loop 5 times:
- Clone the Arc for each thread.
- Spawn a thread that:
  - Loops 1000 times.
  - Acquires the mutex lock.
  - Increments the counter.
  - Releases the lock (automatically).
- Push the JoinHandle to the vector.
Join all threads.
Print the final value of the counter.

// Solution Hint:
/*
use std::sync::{Arc, Mutex};
use std::thread;

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

    for _ in 0..5 { // 5 threads
        let counter_clone = Arc::clone(&counter);
        let handle = thread::spawn(move || {
            for _ in 0..1000 { // Each thread increments 1000 times
                let mut num = counter_clone.lock().unwrap();
                *num += 1;
            }
        });
        handles.push(handle);
    }

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

    println!("Final count: {}", *counter.lock().unwrap()); // Expected: 5000
}
*/

Exercise 9.2: Asynchronous HTTP Request with `reqwest` and `Tokio`

Make an asynchronous HTTP GET request to a public API (e.g., https://httpbin.org/get) and print the response body.

Instructions:

Add reqwest and tokio to your Cargo.toml.

[dependencies]
tokio = { version = "1", features = ["full"] }
reqwest = { version = "0.11", features = ["json"] } # for making HTTP requests

In main.rs, use #[tokio::main] and async fn main().
Make an reqwest::get("URL").await? call.
Convert the response to text using response.text().await?.
Print the text.
Handle any reqwest::Error or io::Error (you can use ? and change main’s return type to Result<(), Box<dyn std::error::Error>>).

// Solution Hint:
/*
use reqwest;
use tokio;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    println!("Fetching data from httpbin.org...");

    let response = reqwest::get("https://httpbin.org/get").await?;
    let body = response.text().await?;

    println!("Response: {}", body);

    Ok(())
}
*/

Exercise 9.3: Asynchronous Producer-Consumer with Channels (`tokio::sync::mpsc`)

Implement an asynchronous producer-consumer pattern using Tokio’s MPSC channels.

Instructions:

Use tokio::sync::mpsc::channel(buffer_size) to create an async channel (e.g., buffer size 10).
Spawn an async task (producer) that sends 5 messages (e.g., strings “message 1”, “message 2”, etc.) to the channel, with a small delay between each.
In the main async function (consumer):
- Receive messages from the channel in a while let loop.
- Print each received message.
Ensure the producer task is spawned correctly and the main function awaits its completion (or the channel closure).

// Solution Hint:
/*
use tokio::sync::mpsc;
use tokio::time::{sleep, Duration};

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

    // Producer task
    let producer_handle = tokio::spawn(async move {
        for i in 1..=5 {
            let msg = format!("Message {}", i);
            println!("Producer sending: {}", msg);
            tx.send(msg).await.unwrap(); // Await on send for backpressure
            sleep(Duration::from_millis(200)).await;
        }
        println!("Producer finished sending.");
    });

    // Consumer in main
    println!("Consumer starting to receive...");
    while let Some(msg) = rx.recv().await { // Await on receive
        println!("Consumer received: {}", msg);
        sleep(Duration::from_millis(300)).await; // Simulate processing
    }
    println!("Consumer finished receiving.");

    producer_handle.await.unwrap(); // Ensure producer task completes
    println!("Program finished.");
}
*/

Advanced Topics: Concurrency and Asynchronous Programming

// table of contents

Advanced Topics: Concurrency and Asynchronous Programming

Threads and Fearless Concurrency

Creating Threads

move Closures with Threads

Message Passing with Channels

Shared-State Concurrency with Mutex and Arc

Asynchronous Programming with async/await and Tokio

What is async/await?

The Tokio Runtime

Async I/O

Exercises / Mini-Challenges

Exercise 9.1: Multi-threaded Counter with Arc<Mutex>

Exercise 9.2: Asynchronous HTTP Request with reqwest and Tokio

Exercise 9.3: Asynchronous Producer-Consumer with Channels (tokio::sync::mpsc)

`move` Closures with Threads

Shared-State Concurrency with `Mutex` and `Arc`

Asynchronous Programming with `async`/`await` and Tokio

What is `async`/`await`?

Exercise 9.1: Multi-threaded Counter with `Arc<Mutex>`

Exercise 9.2: Asynchronous HTTP Request with `reqwest` and `Tokio`

Exercise 9.3: Asynchronous Producer-Consumer with Channels (`tokio::sync::mpsc`)