Run Code Simultaneously

Rust provides built-in support for concurrent programming through its standard library module std::thread.
Threads in Rust allow multiple parts of a program to execute simultaneously, leveraging multi-core processors.

Threads

You can spawn threads using std::thread::spawn, which takes a closure, runs it in a new thread, and return JoinHandle.
We can wait the spawned thread to finish using join method from the returned JoinHandle.
pub fn join(self) -> Result<T>: Waits for the associated thread to finish. This function will return immediately if the associated thread has already finished.
We often use the move keyword with closures passed to thread::spawn because the closure will then take ownership of the values it uses from the environment, thus transferring ownership of those values from one thread to another

Example

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

fn main() {
    let handle = thread::spawn(|| {
        for i in 1..10 {
            println!("hi number {i} from the spawned thread!");
            thread::sleep(Duration::from_millis(1));
        }
    });

    for i in 1..5 {
        println!("hi number {i} from the main thread!");
        thread::sleep(Duration::from_millis(1));
    }

    handle.join().unwrap();
}

Using Smart Pointers, Mutex, and Channels in Threads

Shared Mutable State with Arc and Mutex

In a multi-threaded program, if we need shared mutable state, we can use:

Arc<T>:
- An atomic reference-counted smart pointer for shared ownership across threads.
Mutex<T>:
- A synchronization primitive for mutual exclusion to safely access shared data.
- lock(): Acquires a lock. Blocks the thread if a lock is held.
- try_lock(): Attempts to acquire a lock immediately. If lock cant be acquired, return an Err.

Example

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

struct JobStatus {
    jobs_done: u32,
}

fn main() {
    let status = Arc::new(Mutex::new(JobStatus { jobs_done: 0 }));

    let mut handles = Vec::new();
    for _ in 0..10 {
        let status_shared = Arc::clone(&status);
        let handle = thread::spawn(move || {
            thread::sleep(Duration::from_millis(250));
            // lock and update jobs_done
            status_shared.lock().unwrap().jobs_done += 1
        });
        handles.push(handle);
    }

    // Waiting for all jobs to complete.
    for handle in handles {
        handle.join().unwrap();
    }

    println!("Jobs done: {}", status.lock().unwrap().jobs_done);
}

Arc<T>: Enables shared ownership of the Mutex across threads.
Mutex<T>: Ensures only one thread accesses the counter at a time.

Multiple Readers Single Writer

`RwLock (short for Read-Write Lock) is a synchronization primitive that allows multiple readers or a single writer to access shared data. It ensures safe concurrent access in a multi-threaded environment.

Multiple Readers: Multiple threads can hold a read lock simultaneously if no thread is holding the write lock.
Single Writer: Only one thread can hold the write lock, and it has exclusive access to the data.

RwLock is part of the std::sync module and is frequently used when you want to share data among threads with both read and write operations while minimizing contention.

Read Lock
- read(): Acquires a read lock. Blocks the thread if a write lock is held. Returns an immutable reference to the data.
- try_read(): Attempts to acquire a read lock immediately and return Err if cannot acquire read lock.
Write Lock
- write(): Acquires a write lock. Blocks the thread if any read or write lock is held. Returns a mutable reference to the data.
- try_write(): Attempts to acquire a write lock immediately and return Err if cannot acquire write lock.

Example


use std::{
    sync::{Arc, RwLock},
    thread,
    time::Duration,
};

struct JobStatus {
    jobs_done: u32,
}

fn main() {
    let status = Arc::new(RwLock::new(JobStatus { jobs_done: 0 }));
    let mut handles = Vec::new();

    // Write jobs
    for _ in 0..10 {
        let status_shared = Arc::clone(&status);
        let handle = thread::spawn(move || {
            thread::sleep(Duration::from_millis(250));
            // write lock and update jobs_done
            status_shared.write().unwrap().jobs_done += 1
        });
        handles.push(handle);
    }

    // Read jobs
    for _ in 0..10 {
        let status_shared = Arc::clone(&status);
        let handle = thread::spawn(move || {
            thread::sleep(Duration::from_millis(250));
            // read lock and get jobs_done
            println!("Jobs done: {}", status_shared.read().unwrap().jobs_done);
        });
        handles.push(handle);
    }

    // Waiting for all jobs to complete.
    for handle in handles {
        handle.join().unwrap();
    }
}

The counter is wrapped in Arc<RwLock<T>> so it can be safely shared between threads.
Arc provides reference counting for thread-safe ownership.
write(): Acquires a write lock, granting exclusive mutable access to the shared counter.
read(): Acquires a read lock, allowing multiple threads to access the counter immutably.
The locks ensure that data races are prevented when accessing the shared counter.

Communication Between Threads with Channels

Rust provides channels for thread communication in the std::sync::mpsc module:

mpsc: Stands for multiple producers, single consumer.
That means we can create multiple Sender using clone method.

Example

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

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

    // clone Sender
    let txc = tx.clone();
    // Spawn a thread to send messages
    thread::spawn(move || {
        let messages = vec![1, 2, 3, 4, 5];
        for msg in messages {
            txc.send(msg).unwrap(); // Send a message
            thread::sleep(Duration::from_millis(100));
        }
    });
    
    // Spawn a thread to send messages
    thread::spawn(move || {
        let messages = vec![6, 7, 8, 9, 10];
        for msg in messages {
            tx.send(msg).unwrap(); // Send a message
            thread::sleep(Duration::from_millis(100));
        }
    });

    // Receive messages in the main thread
    for received in rx {
        println!("Received: {}", received);
    }
}

mpsc::channel creates a transmitter (tx) and a receiver (rx).
Messages are sent via the Sender tx and received via the Receiver rx.

Best Practices for Multithreading

Minimize Lock Contention:
- Keep the critical section (where the mutex is locked) as short as possible to avoid slowing down threads.
Use Arc and Mutex Wisely:
- Avoid using shared state unless necessary.
- Prefer message passing with channels for better isolation.

References

PreviousIterators NextString vs &str

Last updated 2 months ago