Implementing Cilk's reducers in Rust (concurrency)

help
1

I'm trying to implement Cilk's reducers in Rust to use in this parallel BFS algorithm (slides).

In short, reducers are a mechanism that avoids data races by keeping separate "views" of the object that are private and can be modified separately, and then uses a reduce operation to merge the independently modified copies together to get the final value. (please see the Cilk docs above for more info)

Reducers can support types that have an identity value, and an associative binary reduce operation (i.e. a monoid). For example, i32 can be a reducer type with identity 0 and reduce operation +.

Here's what I have so far:

use std::{
    cell::{RefCell, UnsafeCell},
    sync::{atomic::AtomicBool, Arc},
    thread,
};

/// Identity for `i32`
const fn identity() -> i32 {
    0
}

/// Reduce operation for `i32`
fn reduce(a: i32, b: i32) -> i32 {
    a + b
}

thread_local! {
    static STORAGE: RefCell<i32> = const { RefCell::new(identity()) };
}

struct Reducer {
    data: UnsafeCell<i32>,
    lock: AtomicBool,
}

impl Reducer {
    fn modify(&self, f: impl FnOnce(&mut i32)) {
        // ???
    }
}

unsafe impl Send for Reducer {}
unsafe impl Sync for Reducer {}

fn main() {
    let reducer = Arc::new(Reducer {
        data: UnsafeCell::new(identity()),
        lock: AtomicBool::new(false),
    });

    // Spawn the threads
    let mut handles = Vec::new();
    for i in 1..=1000 {
        let reducer = reducer.clone();
        let handle = thread::spawn(move || reducer.modify(|n| *n += i));
        handles.push(handle);
    }

    // Join the threads
    handles
        .into_iter()
        .try_for_each(|handle| handle.join())
        .unwrap();

    // Check the answer
    let reducer = Arc::into_inner(reducer).unwrap();
    let value = unsafe { *reducer.data.get() };
    assert_eq!(value, (1..=1000).sum());
}

I'm not sure what to put in the modify method. I'd really appreciate if someone could explain how I should implement this, or tell me about any crates that already do. I'm also open to any more "Rust-y" ways of solving this problem, but I hope to stay close to the original paper so I can make better comparisons with it.

Thanks!

EDIT: see my first reply for more clarification: I am trying to implement a generic reducer

2

If you know about AtomicBool, then why would you use unsafe for this instead of just using AtomicI32 directly? If you browse the docs for half a minute, you see that it has a fetch_update method…

1 Like
3

Thanks for your reply. Sorry for not making the requirements clear enough - I want to implement a generic reducer, and chose to implement a concrete one with i32 first to keep things simpler. So I can't just use AtomicI32.

use std::{cell::Cell, sync::Arc, thread};

trait Reducible: Send + Sync {
    fn identity() -> Self;
    fn reduce(a: Self, b: Self) -> Self;
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
struct Number(i32);

impl Reducible for Number {
    fn identity() -> Self {
        Number(0)
    }

    fn reduce(Number(a): Self, Number(b): Self) -> Self {
        Number(a + b)
    }
}

struct Reducer<T: Reducible> {
    data: Cell<T>,
    // ???
}

impl<T: Reducible> Reducer<T> {
    fn modify(&self, f: impl FnOnce(&mut i32)) {
        // ???
    }
}

unsafe impl<T: Reducible> Send for Reducer<T> {}
unsafe impl<T: Reducible> Sync for Reducer<T> {}

fn main() {
    let reducer = Arc::new(Reducer {
        data: Cell::new(Number::identity()),
        // ???
    });

    let mut handles = Vec::new();
    for i in 1..=1000 {
        let reducer = reducer.clone();
        let handle = thread::spawn(move || reducer.modify(|n| *n += i));
        handles.push(handle);
    }

    handles
        .into_iter()
        .try_for_each(|handle| handle.join())
        .unwrap();

    let reducer = Arc::into_inner(reducer).unwrap();
    assert_eq!(reducer.data.into_inner(), Number((1..=1000).sum()));
}

The actual data structure in the parallel BFS algorithm is a bag, which has an identity of an empty bag and a union operation for reducing.

I tried implementing the algorithm with Mutex and RwLock but I found that it easily became a point of contention since the shared data needed to be accessed by multiple threads very frequently, which is why I turned to the paper's approach of using reducers.

Sorry for the confusion, hope this clears things up!

4

With the atomic bool + unsafe approach, you'd essentially re-implement a mutex. I don't see how calling it something else would help. If you are in a concurrent context, and readers and writers frequently access the same resource, then yes, you'll get contention.

5

If you want to reduce contention what you could do is to have each thread generate its own output, indipendent from the others, then reduce them together only at the end, possibly parallelizing by node rather than task.

2 Likes
6

This topic was automatically closed 90 days after the last reply. We invite you to open a new topic if you have further questions or comments.