Function replay_segments_parallel

nodedb-wal/src/mmap_reader.rs:327–372

Parallel segment replay using scoped threads. Each segment is read in its own thread via mmap. Since segments are monotonically ordered by LSN, concatenating per-segment results in segment order produces a globally LSN-ordered result.

(
    segments: &[crate::segment::SegmentMeta],
    from_lsn: u64,
)

/// monotonically ordered by LSN, concatenating per-segment results in
/// segment order produces a globally LSN-ordered result.
fn replay_segments_parallel(
    segments: &[crate::segment::SegmentMeta],
    from_lsn: u64,
) -> Result<Vec<WalRecord>> {
    // Collect per-segment results. Index corresponds to segment order.
    let mut per_segment: Vec<Result<Vec<WalRecord>>> = Vec::with_capacity(segments.len());

    std::thread::scope(|scope| {
        let handles: Vec<_> = segments
            .iter()
            .map(|seg| {
                scope.spawn(move || -> Result<Vec<WalRecord>> {
                    let mut reader = MmapWalReader::open(&seg.path)?;
                    let mut seg_records = Vec::new();
                    while let Some(record) = reader.next_record()? {
                        if record.header.lsn >= from_lsn {
                            seg_records.push(record);
                        }
                    }
                    reader.release_pages();
                    Ok(seg_records)
                })
            })
            .collect();

        for handle in handles {
            per_segment.push(handle.join().unwrap_or_else(|_| {
                Err(WalError::Io(std::io::Error::other(
                    "segment replay thread panicked",
                )))
            }));
        }
    });

    // Merge in segment order (preserves LSN ordering).
    let total_estimate: usize = per_segment
        .iter()
        .map(|r| r.as_ref().map(|v| v.len()).unwrap_or(0))
        .sum();
    let mut records = Vec::with_capacity(total_estimate);
    for seg_result in per_segment {
        records.extend(seg_result?);
    }

    Ok(records)
}

/// Paginated mmap replay: reads at most `max_records` from `from_lsn`.
///