Function parallel_map

nodedb/src/engine/graph/olap/parallel.rs:101–147

Execute a per-node computation in parallel across partitions. `f` is called once per node with `(node_id, snapshot)`. Results are collected into a Vec indexed by node ID. Uses `std::thread::scope` for structured concurrency — all threads are joined before this function returns. Safe for `!Send` callers (the snapshot is shared via `Arc`).

(
    snapshot: &Arc<CsrSnapshot>,
    config: &ParallelConfig,
    f: impl Fn(u32, &CsrSnapshot) -> T + Send + Sync,
)

/// are joined before this function returns. Safe for `!Send` callers
/// (the snapshot is shared via `Arc`).
pub fn parallel_map<T: Send + Default + Clone>(
    snapshot: &Arc<CsrSnapshot>,
    config: &ParallelConfig,
    f: impl Fn(u32, &CsrSnapshot) -> T + Send + Sync,
) -> Vec<T> {
    let n = snapshot.node_count();
    if n == 0 {
        return Vec::new();
    }

    let partitions = compute_partitions(n, config);
    let f = &f;

    if partitions.len() <= 1 {
        // Single-threaded fast path.
        return (0..n as u32).map(|node| f(node, snapshot)).collect();
    }

    let mut results = vec![T::default(); n];

    std::thread::scope(|scope| {
        let mut handles = Vec::with_capacity(partitions.len());

        for range in &partitions {
            let snap = Arc::clone(snapshot);
            let range = *range;

            let handle = scope.spawn(move || {
                let mut partial = Vec::with_capacity(range.len());
                for node in range.start..range.end {
                    partial.push(f(node, &snap));
                }
                (range, partial)
            });
            handles.push(handle);
        }

        for handle in handles {
            let (range, partial) = handle.join().expect("parallel worker panicked");
            for (i, val) in partial.into_iter().enumerate() {
                results[range.start as usize + i] = val;
            }
        }
    });

    results
}

/// Execute a per-node computation that produces a single aggregate per partition.
///