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

• https://doc.rust-lang.org/book/ch16-00-concurrency.html
• https://doc.rust-lang.org/book/ch16-01-threads.html
• https://doc.rust-lang.org/book/ch16-02-message-passing.html
• https://doc.rust-lang.org/book/ch16-03-shared-state.html