Replace iter() with par_iter(): cannot borrow data mutably in a captured outer variable in an `Fn` closure

Question

I was hoping to replace an iter() with Rayon's par_iter() in a rather simple case like this, but I am failing to do so.

The previous code:

indexes_to_increment
    .iter()
    .for_each(|x| self.some_data[*x as usize] += 1);`

Here's the Rayon-modified code:

extern crate rayon;
use rayon::prelude::*;

fn main() {
    let mut a = SomeStruct::new(vec![1, 0, 0, 1]);
    a.add_factor_indexes(&vec![1, 2]);
    println!("{:?}", a); // spits out "SomeStruct { some_data: [1, 1, 1, 1] }"
}

#[derive(Debug)]
struct SomeStruct {
    some_data: Vec<u8>,
}

impl SomeStruct {
    fn new(some_data: Vec<u8>) -> SomeStruct {
        SomeStruct { some_data }
    }
    fn add_factor_indexes(&mut self, indexes_to_increment: &[u8]) {
        //indexes_to_increment.iter().for_each(|x| self.some_data[*x as usize] += 1);
        indexes_to_increment
            .par_iter()
            .for_each(|x| self.some_data[*x as usize] += 1);
    }
}

(playground)

While I know that the following error message is telling me what to do, at this point I'm unable to do so.

error[E0387]: cannot borrow data mutably in a captured outer variable in an `Fn` closure
  --> src/main.rs:23:27
   |
23 |             .for_each(|x| self.some_data[*x as usize] += 1);
   |                           ^^^^^^^^^^^^^^
   |
help: consider changing this closure to take self by mutable reference
  --> src/main.rs:23:23
   |
23 |             .for_each(|x| self.some_data[*x as usize] += 1);
   |                       ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

If I knew that the indexes_to_increment vector in add_factor_indexes only contains unique u8s and could be replaced with a set, would that change anything?

Answer 1

This error message is an example of exactly the type of error prevention that Rust is designed to provide to you. Said another way, the compiler is preventing you from mutably accessing the same piece of memory concurrently.

Conceptually, the code you are trying to run should be safe because you are always accessing a completely disjoint piece of the vector, there won't be any overlapping mutable borrows of the same index, but the compiler cannot tell that. All it sees is that self.some_data is borrowed mutably multiple times; it doesn't know what the implementation of Index does or what the body of the closure does.

You could find all the matching slots in the vector and then iterate over all of the results:

fn add_factor_indexes(&mut self, indexes_to_increment: &[u8]) {
    self.some_data
        .par_iter_mut()
        .enumerate()
        .filter(|&(i, _)| indexes_to_increment.contains(&(i as u8)))
        .map(|(_, v)| v)
        .for_each(|x| *x += 1);
}

and could be replaced with a set

I'd recommend it due to the repeated lookups, if it's a larger amount of data.

Answer 2

Rayon works best when threads don't share non-constant data. For example, if each closure passed to par_iter only operated on its own piece of data, and those were reassembled together in a final step, Rayon would not complain. (Google MapReduce for a popular example of this strategy that scales well to cloud systems.)

In addition to the solution presented by Shepmaster, a straightforward way to fix your code is to switch from Vec<u8> to Vec<AtomicUsize>, and use the fetch_add method to increment the indices. Since fetch_add accepted a shared reference, Rayon will accept it and it will do what you want.

extern crate rayon;
use rayon::prelude::*;
use std::sync::atomic::{AtomicUsize, Ordering};

fn main() {
    let mut a = SomeStruct::new([1, 0, 0, 1].iter()
        .map(|n| AtomicUsize::new(*n as usize)).collect());
    a.add_factor_indexes(&vec![1, 2]);
    println!("{:?}", a);
}

#[derive(Debug)]
struct SomeStruct {
    some_data: Vec<AtomicUsize>,
}

impl SomeStruct {
    fn new(some_data: Vec<AtomicUsize>) -> SomeStruct {
        SomeStruct { some_data }
    }
    fn add_factor_indexes(&mut self, indexes_to_increment: &[u8]) {
        indexes_to_increment
            .par_iter()
            .for_each(|x| {
                self.some_data[*x as usize].fetch_add(1, Ordering::SeqCst);
            });
    }
}