Threads

Spawn a short-lived thread

std Concurrency

std::thread::scope spawns a new scoped thread that is guaranteed to terminate before returning from the closure, meaning that you can reference data from the calling function without needing Arc or other ownership tricks.

This recipe splits the array in half and performs the work in separate threads.

fn main() {
    let arr = &[1, 25, -4, 10];
    let max = find_max(arr);
    assert_eq!(max, Some(25));
}

fn find_max(arr: &[i32]) -> Option<i32> {
    const THRESHOLD: usize = 2;

    if arr.len() <= THRESHOLD {
        return arr.iter().cloned().max();
    }

    let mid = arr.len() / 2;
    let (left, right) = arr.split_at(mid);

    std::thread::scope(|s| {
        let thread_l = s.spawn(|| find_max(left));
        let thread_r = s.spawn(|| find_max(right));

        let max_l = thread_l.join().unwrap()?;
        let max_r = thread_r.join().unwrap()?;

        Some(max_l.max(max_r))
    })
}

Create a parallel pipeline

crossbeam Concurrency

This recipe uses the crossbeam and crossbeam-channel crates to create a parallel pipeline, similar to that described in the ZeroMQ guide There is a data source and a data sink, with data being processed by two worker threads in parallel on its way from the source to the sink.

We use bounded channels with a capacity of one using crossbeam_channel::bounded. The producer must be on its own thread because it produces messages faster than the workers can process them (since they sleep for half a second) - this means the producer blocks on the call to [crossbeam_channel::Sender::send] for half a second until one of the workers processes the data in the channel. Also note that the data in the channel is consumed by whichever worker calls receive first, so each message is delivered to a single worker rather than both workers.

Reading from the channels via the iterator crossbeam_channel::Receiver::iter method will block, either waiting for new messages or until the channel is closed. Because the channels were created within the crossbeam::scope, we must manually close them via drop to prevent the entire program from blocking on the worker for-loops. You can think of the calls to drop as signaling that no more messages will be sent.

use crossbeam::channel::bounded;
use std::thread;
use std::time::Duration;

fn main() {
    let (snd1, rcv1) = bounded(1);
    let (snd2, rcv2) = bounded(1);
    let n_msgs = 4;
    let n_workers = 2;

    crossbeam::scope(|s| {
        // Producer thread
        s.spawn(|_| {
            for i in 0..n_msgs {
                snd1.send(i).unwrap();
                println!("Source sent {}", i);
            }
            // Close the channel - this is necessary to exit
            // the for-loop in the worker
            drop(snd1);
        });

        // Parallel processing by 2 threads
        for _ in 0..n_workers {
            // Send to sink, receive from source
            let (sendr, recvr) = (snd2.clone(), rcv1.clone());
            // Spawn workers in separate threads
            s.spawn(move |_| {
                thread::sleep(Duration::from_millis(500));
                // Receive until channel closes
                for msg in recvr.iter() {
                    println!("Worker {:?} received {}.", thread::current().id(), msg);
                    sendr.send(msg * 2).unwrap();
                }
            });
        }
        // Close the channel, otherwise sink will never
        // exit the for-loop
        drop(snd2);

        // Sink
        for msg in rcv2.iter() {
            println!("Sink received {}", msg);
        }
    })
    .unwrap();
}

Pass data between two threads

crossbeam Concurrency

This recipe demonstrates the use of crossbeam-channel in a single producer, single consumer (SPSC) setting. We build off the ex-crossbeam-spawn recipe by using crossbeam::scope and Scope::spawn to manage the producer thread. Data is exchanged between the two threads using a crossbeam_channel::unbounded channel, meaning there is no limit to the number of storeable messages. The producer thread sleeps for half a second in between messages.

use crossbeam::channel::unbounded;
use std::{thread, time};

fn main() {
    let (snd, rcv) = unbounded();
    let n_msgs = 5;
    crossbeam::scope(|s| {
        s.spawn(|_| {
            for i in 0..n_msgs {
                snd.send(i).unwrap();
                thread::sleep(time::Duration::from_millis(100));
            }
        });
    })
    .unwrap();
    for _ in 0..n_msgs {
        let msg = rcv.recv().unwrap();
        println!("Received {}", msg);
    }
}

Maintain global mutable state

std Rust patterns

Declare global state using a static Mutex (also see RwLock). Mutex::new is a const fn, so no lazy initialization is needed for an empty collection. If the initial value cannot be computed at compile time, wrap it in a LazyLock, as in static X: LazyLock<Mutex<T>>. The Mutex wrap ensures the state cannot be simultaneously accessed by multiple threads, preventing race conditions. A MutexGuard must be acquired to read or mutate the value stored in a Mutex.

use anyhow::{Result, anyhow};
use std::sync::Mutex;

static FRUIT: Mutex<Vec<String>> = Mutex::new(Vec::new());

fn insert(fruit: &str) -> Result<()> {
    let mut db = FRUIT
        .lock()
        .map_err(|_| anyhow!("Failed to acquire MutexGuard"))?;
    db.push(fruit.to_string());
    Ok(())
}

fn main() -> Result<()> {
    insert("apple")?;
    insert("orange")?;
    insert("peach")?;
    {
        let db = FRUIT
            .lock()
            .map_err(|_| anyhow!("Failed to acquire MutexGuard"))?;

        db.iter()
            .enumerate()
            .for_each(|(i, item)| println!("{}: {}", i, item));
    }
    insert("grape")?;
    Ok(())
}

Calculate SHA256 sum of iso files concurrently

rayon walkdir ring ConcurrencyFilesystem

This recipe calculates the SHA256 for every file with iso extension in the current directory. Walkdir::new iterates the directory and the matching paths are collected into a Vec. par_iter then spreads the work of reading each file and computing its hash across rayon’s global thread pool, which has one thread per available core by default. The digests are gathered with collect into a single Result, so the first error stops the run.

use rayon::prelude::{IntoParallelRefIterator, ParallelIterator};
use ring::digest::{Context, Digest, SHA256};
use std::fs::File;
use std::io::{BufReader, Error, Read};
use std::path::{Path, PathBuf};
use walkdir::WalkDir;

// Verify the iso extension
fn is_iso(entry: &Path) -> bool {
    match entry.extension() {
        Some(e) if e.to_string_lossy().to_lowercase() == "iso" => true,
        _ => false,
    }
}

fn compute_digest<P: AsRef<Path>>(filepath: P) -> Result<Digest, Error> {
    let mut buf_reader = BufReader::new(File::open(filepath)?);
    let mut context = Context::new(&SHA256);
    let mut buffer = [0; 1024];

    loop {
        let count = buf_reader.read(&mut buffer)?;
        if count == 0 {
            break;
        }
        context.update(&buffer[..count]);
    }

    Ok(context.finish())
}

fn main() -> Result<(), Error> {
    let paths: Vec<PathBuf> = WalkDir::new("/home/user/Downloads")
        .follow_links(true)
        .into_iter()
        .filter_map(|e| e.ok())
        .filter(|e| !e.path().is_dir() && is_iso(e.path()))
        .map(|e| e.into_path())
        .collect();

    let digests = paths
        .par_iter()
        .map(|path| compute_digest(path).map(|digest| (digest, path)))
        .collect::<Result<Vec<_>, Error>>()?;

    for (sha, path) in digests {
        println!("{:?} {:?}", sha, path);
    }
    Ok(())
}

Sum the size of every file in a directory

rayon walkdir ConcurrencyFilesystem

This recipe adds up the size of every file under the book’s src directory. Reading a file blocks while the operating system fetches its bytes, so the work is IO-bound: each thread spends most of its time waiting on the disk rather than computing. A fixed-size pool keeps several reads in flight at once.

WalkDir yields every entry under src. A rayon::ThreadPool is built with ThreadPoolBuilder::num_threads set to std::thread::available_parallelism. ThreadPool::install runs the closure inside that pool, so the par_iter below uses exactly those threads instead of rayon’s global pool. Each file is read with std::fs::read, and the byte count (or the io::Error if the read fails) is collected into a single io::Result. Reading the contents is what keeps the workers busy with real IO; std::fs::metadata would be cheaper when only the length matters.

use anyhow::Result;
use rayon::ThreadPoolBuilder;
use rayon::prelude::{IntoParallelRefIterator, ParallelIterator};
use std::fs;
use std::io;
use std::path::PathBuf;
use std::thread::available_parallelism;
use walkdir::WalkDir;

fn main() -> Result<()> {
    let pool = ThreadPoolBuilder::new()
        .num_threads(available_parallelism()?.get())
        .build()?;

    let paths: Vec<PathBuf> = WalkDir::new("src")
        .into_iter()
        .filter_map(|e| e.ok())
        .filter(|e| e.path().is_file())
        .map(|e| e.into_path())
        .collect();

    let sizes = pool.install(|| {
        paths
            .par_iter()
            .map(|path| fs::read(path).map(|bytes| bytes.len()))
            .collect::<io::Result<Vec<usize>>>()
    })?;

    let total: usize = sizes.iter().sum();
    println!("{} files, {} bytes", paths.len(), total);
    Ok(())
}

Draw fractal dispatching work to a thread pool

rayon num image ConcurrencyScienceRendering

This recipe generates an image by drawing a fractal from the Julia set with rayon’s thread pool for distributed computation.

Allocate memory for output image of given width and height with ImageBuffer::new. wavelength_to_rgb calculates RGB pixel values. par_enumerate_pixels_mut hands rayon every pixel of the image along with its coordinates, and rayon distributes them over its global thread pool, which has one thread per available core by default. Each worker writes its pixels directly into the buffer, so no channel is needed. ImageBuffer::save writes the image to output.png.

use anyhow::Result;
use image::{ImageBuffer, Rgb};
use num::complex::Complex;
use rayon::prelude::ParallelIterator;

fn main() -> Result<()> {
    let (width, height) = (1920, 1080);
    let mut img: ImageBuffer<Rgb<u8>, Vec<u8>> = ImageBuffer::new(width, height);
    let iterations = 300;

    let c = Complex::new(-0.8, 0.156);

    img.par_enumerate_pixels_mut().for_each(|(x, y, pixel)| {
        let i = julia(c, x, y, width, height, iterations);
        *pixel = wavelength_to_rgb(380 + i * 400 / iterations);
    });

    img.save("output.png")?;
    Ok(())
}

// Function converting intensity values to RGB
// Based on http://www.efg2.com/Lab/ScienceAndEngineering/Spectra.htm
fn wavelength_to_rgb(wavelength: u32) -> Rgb<u8> {
    let wave = wavelength as f32;

    let (r, g, b) = match wavelength {
        380..=439 => ((440. - wave) / (440. - 380.), 0.0, 1.0),
        440..=489 => (0.0, (wave - 440.) / (490. - 440.), 1.0),
        490..=509 => (0.0, 1.0, (510. - wave) / (510. - 490.)),
        510..=579 => ((wave - 510.) / (580. - 510.), 1.0, 0.0),
        580..=644 => (1.0, (645. - wave) / (645. - 580.), 0.0),
        645..=780 => (1.0, 0.0, 0.0),
        _ => (0.0, 0.0, 0.0),
    };

    let factor = match wavelength {
        380..=419 => 0.3 + 0.7 * (wave - 380.) / (420. - 380.),
        701..=780 => 0.3 + 0.7 * (780. - wave) / (780. - 700.),
        _ => 1.0,
    };

    let (r, g, b) = (
        normalize(r, factor),
        normalize(g, factor),
        normalize(b, factor),
    );
    Rgb([r, g, b])
}

// Maps Julia set distance estimation to intensity values
fn julia(c: Complex<f32>, x: u32, y: u32, width: u32, height: u32, max_iter: u32) -> u32 {
    let width = width as f32;
    let height = height as f32;

    let mut z = Complex {
        // scale and translate the point to image coordinates
        re: 3.0 * (x as f32 - 0.5 * width) / width,
        im: 2.0 * (y as f32 - 0.5 * height) / height,
    };

    let mut i = 0;
    for t in 0..max_iter {
        if z.norm() >= 2.0 {
            break;
        }
        z = z * z + c;
        i = t;
    }
    i
}

// Normalizes color intensity values within RGB range
fn normalize(color: f32, factor: f32) -> u8 {
    ((color * factor).powf(0.8) * 255.) as u8
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn out_of_visible_range_is_black() {
        assert_eq!(wavelength_to_rgb(100), Rgb([0, 0, 0]));
    }

    #[test]
    fn julia_iteration_is_bounded() {
        let c = Complex::new(-0.8, 0.156);
        assert!(julia(c, 960, 540, 1920, 1080, 300) < 300);
    }
}