Testcase: map-reduce

Rust makes it very easy to parallelise data processing, without many of the headaches traditionally associated with such an attempt.

The standard library provides great threading primitives out of the box.
These, combined with Rust’s concept of Ownership and aliasing rules, automatically prevent
data races.

The aliasing rules (one writable reference XOR many readable references) automatically prevent
you from manipulating state that is visible to other threads. (Where synchronisation is needed,
there are synchronisation
primitives like Mutexes or Channels.)

In this example, we will calculate the sum of all digits in a block of numbers.
We will do this by parcelling out chunks of the block into different threads. Each thread will sum
its tiny block of digits, and subsequently we will sum the intermediate sums produced by each
thread.

Note that, although we’re passing references across thread boundaries, Rust understands that we’re
only passing read-only references, and that thus no unsafety or data races can occur. Because
we’re move-ing the data segments into the thread, Rust will also ensure the data is kept alive
until the threads exit, so no dangling pointers occur.

  1. use std::thread;
  3. // This is the `main` thread
  4. fn main() {
  6.     // This is our data to process.
  7.     // We will calculate the sum of all digits via a threaded  map-reduce algorithm.
  8.     // Each whitespace separated chunk will be handled in a different thread.
  9.     //
  10.     // TODO: see what happens to the output if you insert spaces!
  11.     let data = "86967897737416471853297327050364959
  12. 11861322575564723963297542624962850
  13. 70856234701860851907960690014725639
  14. 38397966707106094172783238747669219
  15. 52380795257888236525459303330302837
  16. 58495327135744041048897885734297812
  17. 69920216438980873548808413720956532
  18. 16278424637452589860345374828574668";
  20.     // Make a vector to hold the child-threads which we will spawn.
  21.     let mut children = vec![];
  23.     /*************************************************************************
  24.      * "Map" phase
  25.      *
  26.      * Divide our data into segments, and apply initial processing
  27.      ************************************************************************/
  29.     // split our data into segments for individual calculation
  30.     // each chunk will be a reference (&str) into the actual data
  31.     let chunked_data = data.split_whitespace();
  33.     // Iterate over the data segments.
  34.     // .enumerate() adds the current loop index to whatever is iterated
  35.     // the resulting tuple "(index, element)" is then immediately
  36.     // "destructured" into two variables, "i" and "data_segment" with a
  37.     // "destructuring assignment"
  38.     for (i, data_segment) in chunked_data.enumerate() {
  39.         println!("data segment {} is \"{}\"", i, data_segment);
  41.         // Process each data segment in a separate thread
  42.         //
  43.         // spawn() returns a handle to the new thread,
  44.         // which we MUST keep to access the returned value
  45.         //
  46.         // 'move || -> u32' is syntax for a closure that:
  47.         // * takes no arguments ('||')
  48.         // * takes ownership of its captured variables ('move') and
  49.         // * returns an unsigned 32-bit integer ('-> u32')
  50.         //
  51.         // Rust is smart enough to infer the '-> u32' from
  52.         // the closure itself so we could have left that out.
  53.         //
  54.         // TODO: try removing the 'move' and see what happens
  55.         children.push(thread::spawn(move || -> u32 {
  56.             // Calculate the intermediate sum of this segment:
  57.             let result = data_segment
  58.                         // iterate over the characters of our segment..
  59.                         .chars()
  60.                         // .. convert text-characters to their number value..
  61.                         .map(|c| c.to_digit(10).expect("should be a digit"))
  62.                         // .. and sum the resulting iterator of numbers
  63.                         .sum();
  65.             // println! locks stdout, so no text-interleaving occurs
  66.             println!("processed segment {}, result={}", i, result);
  68.             // "return" not needed, because Rust is an "expression language", the
  69.             // last evaluated expression in each block is automatically its value.
  70.             result
  72.         }));
  73.     }
  76.     /*************************************************************************
  77.      * "Reduce" phase
  78.      *
  79.      * Collect our intermediate results, and combine them into a final result
  80.      ************************************************************************/
  82.     // collect each thread's intermediate results into a new Vec
  83.     let mut intermediate_sums = vec![];
  84.     for child in children {
  85.         // collect each child thread's return-value
  86.         let intermediate_sum = child.join().unwrap();
  87.         intermediate_sums.push(intermediate_sum);
  88.     }
  90.     // combine all intermediate sums into a single final sum.
  91.     //
  92.     // we use the "turbofish" ::<> to provide sum() with a type hint.
  93.     //
  94.     // TODO: try without the turbofish, by instead explicitly
  95.     // specifying the type of final_result
  96.     let final_result = intermediate_sums.iter().sum::<u32>();
  98.     println!("Final sum result: {}", final_result);
  99. }

Assignments

It is not wise to let our number of threads depend on user inputted data.
What if the user decides to insert a lot of spaces? Do we really want to spawn 2,000 threads?
Modify the program so that the data is always chunked into a limited number of chunks,
defined by a static constant at the beginning of the program.

See also: