Splitting work across threads

The work of a loop over a vector is often independent per element, which makes it a candidate for threads: cut the vector into chunks, give each chunk to a thread, and combine the partial results. The standard library has the threads and the chunks, and no parallel iterator of its own.

This lesson sums a million numbers one way and then four ways, reduces two values at once, hands out work with an atomic cursor, and ends with the compiler error for the parallel method that only a crate provides.

The sequential sum

The baseline is a range and a sum, which walks a million numbers in one thread. Every parallel version below has to produce the same number.

fn main() { let total: u64 = (1..=1_000_000).sum(); println!("{total}"); }

rustc --edition 2024 seq.rs
./seq

500000500000

The sum is the closed form for a million, which is a useful check for the parallel result. The lesson on folds is the same fold with a function.

One thread per chunk

chunks gives borrowed slices, and a thread needs an owned value, so each chunk is copied into a vector before it is moved into the spawn. Four chunks and four threads is the simplest division of the work.

The program builds the vector, splits it, spawns, collects the handles and adds the four partial sums. The result is the same number as the sequential sum.

use std::thread; fn main() { let values: Vec<u64> = (1..=1_000_000).collect(); let handles: Vec<_> = values .chunks(250_000) .map(|chunk| { let chunk = chunk.to_vec(); thread::spawn(move || chunk.iter().sum::<u64>()) }) .collect(); let total: u64 = handles.into_iter().map(|handle| handle.join().unwrap()).sum(); println!("{total}"); }

rustc --edition 2024 chunks.rs
./chunks

500000500000

The copy costs a megabyte per chunk and is the price of a plain spawn; the scoped thread lesson removes it when the threads end with the function. The Vec<_> holds the handles whose result type the compiler infers from the closure.

A map and a reduce

A parallel computation usually produces more than one number per chunk. Each worker returns a tuple of partial results, and the main thread folds the tuples into one, which is the map and reduce shape behind every parallel collection.

The program sums one to a hundred in four chunks and counts the even numbers in the same pass. The fold adds both fields.

use std::thread; fn main() { let values: Vec<u64> = (1..=100).collect(); let handles: Vec<_> = values .chunks(25) .map(|chunk| { let chunk = chunk.to_vec(); thread::spawn(move || { let sum = chunk.iter().sum::<u64>(); let evens = chunk.iter().filter(|value| *value % 2 == 0).count(); (sum, evens) }) }) .collect(); let (sum, evens) = handles .into_iter() .fold((0, 0), |(sum, evens), handle| { let (chunk_sum, chunk_evens) = handle.join().unwrap(); (sum + chunk_sum, evens + chunk_evens) }); println!("{sum} {evens}"); }

rustc --edition 2024 reduce.rs
./reduce

5050 50

The two numbers are the sum of one to a hundred and the fifty even numbers in that range, which is a check a reader can do by hand. The fold with a tuple is the same shape as the fold lesson, one level up.

A shared cursor

Equal chunks are only fair when every element costs the same. For uneven work, the threads take the next block from a shared counter: fetch_add hands out a block to exactly one thread, and a thread that finishes early takes another.

The program sums a thousand numbers in blocks of a hundred, with four threads taking blocks until the vector is exhausted. The total is the same as a sequential sum.

use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; use std::thread; fn main() { let values = Arc::new((1..=1000).collect::<Vec<u64>>()); let cursor = Arc::new(AtomicUsize::new(0)); let mut handles = Vec::new(); for _ in 0..4 { let values = Arc::clone(&values); let cursor = Arc::clone(&cursor); handles.push(thread::spawn(move || { let mut sum = 0; loop { let start = cursor.fetch_add(100, Ordering::Relaxed); if start >= values.len() { break; } let end = (start + 100).min(values.len()); sum += values[start..end].iter().sum::<u64>(); } sum })); } let total: u64 = handles.into_iter().map(|handle| handle.join().unwrap()).sum(); println!("{total}"); }

rustc --edition 2024 stealing.rs
./stealing

500500

Five hundred and five hundred is the sum of one to a thousand. The block size is the tuning knob: small blocks balance the load and make the atomic the bottleneck, large blocks do the opposite, and a block of a hundred is a reasonable middle for this size. The atomics lesson is the counter.

The number of threads is the other knob. available_parallelism reports what the operating system allows, and a program that spawns many more threads than that spends its time switching between them. A thread per chunk is the right shape when the chunks are few and large; a fixed pool with a queue is the right shape when the work arrives over time.

The scoped version of the chunk program is the same code with thread::scope around the loop and scope.spawn in place of thread::spawn, and it passes each chunk by reference instead of copying it. The choice between the two is the lifetime of the threads and not the arithmetic.

The parallel method that is not there

A program copied from an article that used rayon calls par_iter on a vector. The method is not in the standard library, so the compiler reports E0599 and suggests the sequential iter.

The program maps over a vector with the parallel method. The fix is a dependency on the crate that provides the trait, or the chunk-and-spawn code of this lesson.

fn main() { let values = vec![1, 2, 3]; let doubled: Vec<i32> = values.par_iter().map(|value| value * 2).collect(); println!("{doubled:?}"); }

rustc --edition 2024 par_err.rs

error[E0599]: no method named `par_iter` found for struct `Vec<{integer}>` in the current scope
--> par_err.rs:3:36
|
3 | let doubled: Vec<i32> = values.par_iter().map(|value| value * 2).collect();
| ^^^^^^^^
|
help: there is a method `iter` with a similar name
|
3 - let doubled: Vec<i32> = values.par_iter().map(|value| value * 2).collect();
3 + let doubled: Vec<i32> = values.iter().map(|value| value * 2).collect();
|

error: aborting due to 1 previous error

For more information about this error, try `rustc --explain E0599`.

Rayon adds the trait that makes the method exist, and its work-stealing scheduler is the same idea as the shared cursor with a better queue. The next lesson measures whether the parallel version is actually faster.

Exercises

  1. Count the primes below a thousand in four chunks and compare with a sequential count.
  2. Change the block size of the shared cursor to ten and to a thousand and explain the difference in the number of atomic calls.
  3. Return the largest value from each chunk and take the maximum of the partial results.

Questions

A new set of questions in a new order every time. Choose one answer for each, then check.

  1. What does chunks(n) give an iterator over?

    Answer

    Borrowed slices of at most n elements

    The chunks are slices into the original vector, so a thread needs a copy.

  2. Why is a vector copied before it is moved into a thread?

    Answer

    A spawned thread needs an owned value

    A plain spawn requires a value that lives for the whole program or is owned.

  3. What does a shared atomic cursor give the threads?

    Answer

    Each block is handed to exactly one thread

    fetch_add is atomic, so two threads never take the same block.

  4. Where does par_iter come from?

    Answer

    A crate such as rayon, not the standard library

    The standard library has threads and chunks and no parallel iterator.

  5. What does this program print? use std::thread; fn main() { let values = vec![1, 2, 3, 4]; let handles: Vec<_> = values .chunks(2) .map(|chunk| { let chunk = chunk.to_vec(); thread::spawn(move || chunk.iter().sum::<i32>()) }) .collect(); let total: i32 = handles.into_iter().map(|handle| handle.join().unwrap()).sum(); println!("{total}"); }

    Answer

    10

    Two chunks of two sum to three and seven, and the total is ten.

  6. What does this program print? fn main() { let total: u64 = (1..=10).sum(); println!("{total}"); }

    Answer

    55

    The numbers one to ten add up to fifty-five.

  7. What does this program print? use std::sync::atomic::{AtomicUsize, Ordering}; fn main() { let cursor = AtomicUsize::new(0); let first = cursor.fetch_add(10, Ordering::Relaxed); let second = cursor.fetch_add(10, Ordering::Relaxed); println!("{first} {second} {}", cursor.load(Ordering::Relaxed)); }

    Answer

    0 10 20

    Each fetch_add returns the value before it added ten.

  8. What does this program print? use std::thread; fn main() { let values: Vec<i32> = (1..=6).collect(); let handles: Vec<_> = values .chunks(3) .map(|chunk| { let chunk = chunk.to_vec(); thread::spawn(move || chunk.iter().max().unwrap() * 1) }) .collect(); let mut maxima: Vec<i32> = handles.into_iter().map(|handle| handle.join().unwrap()).collect(); maxima.sort(); println!("{maxima:?}"); }

    Answer

    [3, 6]

    The two chunks have maxima three and six, and the sort fixes the order.

  9. What is the first line of the compiler output for this program? fn main() { let values = vec![1, 2, 3]; let doubled: Vec<i32> = values.par_iter().map(|value| value * 2).collect(); println!("{doubled:?}"); }

    Answer

    error[E0599]: no method named par_iter found for struct Vec<{integer}> in the current scope

    The method belongs to a trait from a crate, and nothing in scope provides it.

  10. What is the first line of the compiler output for this program? use std::thread; fn main() { let values = vec![1, 2, 3]; let handle = thread::spawn(|| values.iter().sum::<i32>()); println!("{}", handle.join().unwrap()); }

    Answer

    error[E0373]: closure may outlive the current function, but it borrows values, which is owned by the current function

    The closure borrows a local that a plain spawn may outlive, so it must own or be scoped.

The code on this page was built with rustc 1.96.1 and run on Alpine Linux 3.24.3 (x86_64) in a Docker container.

Article author: Arthur Isaev