Function parallel_reduce

nodedb/src/engine/graph/olap/parallel.rs:153–188

Execute a per-node computation that produces a single aggregate per partition. Each partition computes a partial result, then the `reduce` function merges all partials into a final result.

(
    snapshot: &Arc<CsrSnapshot>,
    config: &ParallelConfig,
    map_fn: impl Fn(NodeRange, &CsrSnapshot) -> P + Send + Sync,
    reduce_fn: impl Fn(Vec<P>) -> R,
)

/// Each partition computes a partial result, then the `reduce` function
/// merges all partials into a final result.
pub fn parallel_reduce<P: Send, R>(
    snapshot: &Arc<CsrSnapshot>,
    config: &ParallelConfig,
    map_fn: impl Fn(NodeRange, &CsrSnapshot) -> P + Send + Sync,
    reduce_fn: impl Fn(Vec<P>) -> R,
) -> R {
    let n = snapshot.node_count();
    let partitions = compute_partitions(n, config);
    let map_fn = &map_fn;

    if partitions.is_empty() {
        return reduce_fn(Vec::new());
    }

    if partitions.len() <= 1 {
        let partial = map_fn(partitions[0], snapshot);
        return reduce_fn(vec![partial]);
    }

    let partials: Vec<P> = std::thread::scope(|scope| {
        let handles: Vec<_> = partitions
            .iter()
            .map(|&range| {
                let snap = Arc::clone(snapshot);
                scope.spawn(move || map_fn(range, &snap))
            })
            .collect();

        handles
            .into_iter()
            .map(|h| h.join().expect("parallel worker panicked"))
            .collect()
    });

    reduce_fn(partials)
}

#[cfg(test)]
mod tests {