Skip to main content

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​