#2.19. par - Parallel processing API

The rok::core::par module provides a data-parallel API. Its design is inspired by Rust's Rayon, and it runs on the same work-stealing scheduler as task::*.

Because use rok::core::**; applies implicitly, par::for_each, par::map and the rest are usable without any import; the module prefix par:: is still kept. Where the parallel intent should be visibly clear, writing use rok::core::par::*; explicitly is recommended.

#2.19.1. Core API

The core API consists of six functions.

fn for_each<I: IntoIterator<Item = T>, T: Send>(iter: I, f: fn(T) + Send + Sync);
fn map<I: IntoIterator<Item = T>, T: Send, U: Send>(iter: I, f: fn(T) -> U + Send + Sync) -> List<U>;
fn reduce<T: Send + Sync>(iter: List<T>, init: T, f: fn(T, T) -> T + Send + Sync) -> T;
fn filter<T: Send>(iter: List<T>, pred: fn(T) -> bool + Send + Sync) -> List<T>;
fn join<T1: Send, T2: Send>(f1: fn() -> T1 + Send, f2: fn() -> T2 + Send) -> (T1, T2);    // 2 in parallel
fn scope(f: fn(Scope));                                          // dynamic spawn

The f of for_each / map / reduce / filter is called by several workers at once, so it is + Send + Sync.

• for_each / map / filter preserve input order. The result List of map / filter is in the same order as the input.
• for_each / map / filter give each item to a worker. Two items must not reach one collection that is not Sync (2.13.9). Each call first looks for a collection reached from two items, following the items' parts and the elements of their collections to any depth, as a hand-off copy does, but not into collections of plain values (numbers, bool, str): if it finds one, the items run one after another on the calling thread, so nothing races and in-place changes stay as with workers, and a warning names the call site (once per site). Slices only read: items may share a list through slices of it (chunks of one big list), as long as no item holds the list itself; the elements a slice reaches are followed one by one. The check marks each collection the items reach in the object's header (about 15 ns per collection for one-list items and about 50 ns per collection on nested items, mostly the memory reads; items with no collection cost nothing): for many small items with little work each it can exceed the work. The warning says what to write instead: give each item its own values (.clone() a shared one before storing it), or share read-only data through Frozen (2.16.4b) or another Sync type, which the check does not follow. A future among the items counts as its frame: two items holding one future run one at a time too. par does not hand its items over, so a future whose frame captures a collection cannot be an item (a compile error): spawn such futures as tasks instead (2.18.1). The result of filter shares the input's items.
• reduce splits the input into chunks, folds each chunk in parallel, then folds init and the chunk results in order. If the input is empty it returns init.
• Children spawned inside a scope are all joined when the scope ends.
• par runs on the same work-stealing pool as tasks (2.18.3). A worker waiting inside par runs other tasks meanwhile, so calling par inside a task does not deadlock. The Scope passed to scope and its handle are:

struct Scope { /* list of jobs in the scope */ }
impl Scope {
    fn spawn<T: Send>(self, f: fn() -> T + Send) -> ScopeHandle<T>;
}

struct ScopeHandle<T> { /* one spawned job */ }
impl<T> ScopeHandle<T> {
    fn join(self) -> T;          // waits until that job finishes
}

When scope ends it also waits for every job that was not joined.

#Example

use rok::core::par::*;

let xs: List<i32> = (0..1000).collect();

par::for_each(xs, |x| { process(x) });

let squared: List<i32> = par::map(xs, |x| x * x);

let total: i32 = par::reduce(squared, 0, |a, b| a + b);

let (a, b) = par::join(
    || compute_a(),
    || compute_b(),
);

par::scope(|s| {
    for chunk in chunks {
        s.spawn(|| process_chunk(chunk));
    }
});           // ← every spawn is guaranteed complete at the end of the scope

#2.19.2. par::for_each vs Iterator

Sequential and parallel processing are distinguished by the form of the call.

xs.into_iter().for_each(f);     // sequential (single thread)
par::for_each(xs, f);            // parallel (worker pool)

Work is distributed data-parallel in chunks, and the work-stealing scheduler automatically balances unbalanced workloads.

#2.19.3. par::scope - structured parallelism

par::scope is a structured parallel API that can spawn work dynamically.

par::scope(|s| {
    let h1 = s.spawn(|| task1());
    let h2 = s.spawn(|| task2());
    // results of h1, h2 are usable (inside the scope)
    // joined automatically at the end of the scope
});

This follows the structured concurrency form: the scope blocks until every child spawn has completed.

#par::try_scope - cancel on the first error

For children that throw errors, use try_scope.

fn try_scope<E: Send>(f: fn(TryScope<E>) throws (E)) throws (E);

impl<E> TryScope<E> {
    fn spawn<T: Send>(self, f: fn() throws (E) -> T + Send) -> TryScopeHandle<T, E>;
    fn is_cancelled(self) -> bool;
}

impl<T, E: Send> TryScopeHandle<T, E> {
    fn join(self) throws (E) -> T;     // its own error, or the scope's first error if cancelled before starting
}

If one child throws, the scope is cancelled. Children not yet started are not run, and children already running run to completion. When all children have finished, try_scope rethrows that first error. A running child can check is_cancelled() to finish early.

try par::try_scope(|s| {
    let a = s.spawn(|| try load("a"));
    let b = s.spawn(|| try load("b"));
    use_both(try a.join(), try b.join());
});

#2.19.4. par::reduce - associativity required

The signature of par::reduce is as follows; using it involves some mathematical conditions.

fn reduce<T: Send + Sync>(iter: List<T>, init: T, f: fn(T, T) -> T + Send + Sync) -> T;

The combining function f must be associative for the result to be deterministic. init should preferably be the identity: 0 for +, 1 for *. For a non-associative function, such as floating-point summation accumulating large values, the result may be nondeterministic and needs explicit care.

let sum: i64 = par::reduce([1, 2, 3, 4, 5], 0, |a, b| a + b);    //→ 15

// floating-point - the result may differ slightly
let total: f64 = par::reduce(values, 0.0, |a, b| a + b);

#2.19.5. par::join - exactly two

Use par::join to run two jobs concurrently and receive the pair of results.

let (a, b) = par::join(
    || expensive_computation_1(),
    || expensive_computation_2(),
);

Two jobs are the most common fork-join form, so a dedicated function is provided. For N jobs, use par::for_each or par::scope.

#2.19.6. SIMD / GPU integration

The par module centers on CPU data parallelism. SIMD multiversioning is applied automatically inside par::map according to target_feature; GPU acceleration is not part of the standard libraries; libraries reach it through rok::bare::mlir (1.9).

par::map(xs, |x| heavy_op(x));     // CPU work-stealing + automatic SIMD

#2.19.7. Module mapping

The module location and internal strategy of each API:

Internally it is one layer on top of task::spawn, sharing the same scheduler.

Edit this page on GitHub