Multithreading with the crossbeam crate

Recipe	Crates	Categories
Spawn a short-lived thread
Create a parallel pipeline
Pass data between two threads

Spawn a short-lived thread

The example uses the crossbeam⮳ crate, which provides data structures and functions for concurrent and parallel programming. crossbeam::thread::Scope::spawn⮳ spawns a new scoped thread that is guaranteed to terminate before returning from the closure that passed into crossbeam::scope⮳ function, meaning that you can reference data from the calling function.

This example 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));
    println!("The maximum is {:?}", max);
}

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);

    crossbeam::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))
    })
    .unwrap()
}

Create a parallel pipeline

This example 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 std::ops::Drop⮳ to prevent the entire program from blocking on the worker for-loops. You can think of the calls to std::ops::Drop⮳ as signaling that no more messages will be sent.

extern crate crossbeam;
extern crate crossbeam_channel;

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

use crossbeam_channel::bounded;

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

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

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

use crossbeam_channel::unbounded;

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);
    }
}