perf(ecstore): batch local EC shard preads on GET (#5679)

Collapse per-shard blocking-pool round-trips into one spawn_blocking
pread batch when all online shards are local and mmap-read is enabled.

Co-authored-by: ba <ba@ubuntu-server.alpha30.bos16>
Co-authored-by: Zhengchao An <anzhengchao@gmail.com>

crates/ecstore/src/disk/disk_store.rs

@@ -1088,6 +1088,10 @@ impl LocalDiskWrapper {
         self.disk.clone()
     }
 
+    pub fn get_object_path_if_local(&self, volume: &str, path: &str) -> crate::disk::error::Result<std::path::PathBuf> {
+        self.disk.get_object_path(volume, path)
+    }
+
     pub fn runtime_state(&self) -> RuntimeDriveHealthState {
         self.health.runtime_state()
     }

crates/ecstore/src/disk/local.rs

@@ -6686,6 +6686,50 @@ impl LocalDisk {
     }
 }
 
+/// Batch positioned reads for local EC shard files in a single `spawn_blocking`.
+///
+/// Collapses per-shard blocking-pool round-trips that dominate warm GET
+/// fan-out on single-node multi-disk topologies.
+#[cfg(unix)]
+pub(crate) async fn batch_shard_pread(requests: Vec<(std::path::PathBuf, usize, usize)>) -> Vec<Result<Bytes>> {
+    let n = requests.len();
+    tokio::task::spawn_blocking(move || {
+        use std::os::unix::fs::FileExt;
+
+        let mut results = Vec::with_capacity(n);
+        for (file_path, offset, length) in requests {
+            let r = (|| -> Result<Bytes> {
+                let meta = std::fs::metadata(&file_path).map_err(DiskError::from)?;
+                let end = offset.checked_add(length).ok_or(DiskError::FileCorrupt)?;
+                if meta.len() < end as u64 {
+                    return Err(DiskError::FileCorrupt);
+                }
+
+                let file = std::fs::File::open(&file_path).map_err(DiskError::from)?;
+                let mut buf = vec![0u8; length];
+                let mut total = 0usize;
+                while total < length {
+                    let nbytes = file
+                        .read_at(&mut buf[total..], (offset + total) as u64)
+                        .map_err(DiskError::from)?;
+                    if nbytes == 0 {
+                        return Err(DiskError::FileCorrupt);
+                    }
+                    total += nbytes;
+                }
+                Ok(Bytes::from(buf))
+            })();
+            results.push(r);
+        }
+        results
+    })
+    .await
+    .unwrap_or_else(|e| {
+        let msg = format!("spawn_blocking join: {e}");
+        (0..n).map(|_| Err(DiskError::other(msg.clone()))).collect()
+    })
+}
+
 #[async_trait::async_trait]
 impl DiskAPI for LocalDisk {
     fn to_string(&self) -> String {
@@ -17903,4 +17947,44 @@ mod test {
             (reads * shard_mib) as f64 / wall,
         );
     }
+
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn test_batch_shard_pread_basic() {
+        use tempfile::tempdir;
+
+        let dir = tempdir().unwrap();
+        let payloads: &[&[u8]] = &[b"aaaaaa", b"bbbbbb", b"cccccc"];
+        let mut requests = Vec::new();
+        for (i, payload) in payloads.iter().enumerate() {
+            let p = dir.path().join(format!("shard-{i}.bin"));
+            std::fs::write(&p, payload).unwrap();
+            requests.push((p, 0usize, payload.len()));
+        }
+
+        let results = batch_shard_pread(requests).await;
+        assert_eq!(results.len(), payloads.len());
+        for (result, expected) in results.iter().zip(payloads.iter()) {
+            assert_eq!(result.as_ref().unwrap().as_ref(), *expected);
+        }
+    }
+
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn test_batch_shard_pread_partial_errors() {
+        use tempfile::tempdir;
+
+        let dir = tempdir().unwrap();
+        let good_path = dir.path().join("good.bin");
+        std::fs::write(&good_path, b"good data").unwrap();
+        let missing_path = dir.path().join("does-not-exist.bin");
+
+        let requests = vec![(good_path, 0usize, 9usize), (missing_path, 0usize, 4usize)];
+
+        let results = batch_shard_pread(requests).await;
+        assert_eq!(results.len(), 2);
+        assert!(results[0].is_ok());
+        assert_eq!(results[0].as_ref().unwrap().as_ref(), b"good data");
+        assert!(results[1].is_err());
+    }
 }

crates/ecstore/src/disk/mod.rs

@@ -682,6 +682,15 @@ impl Disk {
             Disk::Remote(remote_disk) => remote_disk.enable_health_check(),
         }
     }
+
+    /// Returns the absolute filesystem path for a (volume, path) pair if this
+    /// disk is local, or `None` if it is a remote disk.
+    pub fn get_object_path_if_local(&self, volume: &str, path: &str) -> Option<crate::disk::error::Result<std::path::PathBuf>> {
+        match self {
+            Disk::Local(w) => Some(w.get_object_path_if_local(volume, path)),
+            Disk::Remote(_) => None,
+        }
+    }
 }
 
 pub async fn new_disk(ep: &Endpoint, opt: &DiskOption) -> Result<DiskStore> {

crates/ecstore/src/io_support/bitrot.rs

@@ -483,10 +483,20 @@ fn instrument_raw_shard_writer(writer: FileWriter, is_local: bool) -> FileWriter
 }
 
 fn bitrot_encoded_range(offset: usize, length: usize, shard_size: usize, checksum_algo: HashAlgorithm) -> (usize, usize) {
-    (
-        offset.div_ceil(shard_size) * checksum_algo.size() + offset,
-        length.div_ceil(shard_size) * checksum_algo.size() + length,
-    )
+    adjust_shard_read_params(offset, length, shard_size, &checksum_algo)
+}
+
+/// Adjusts a raw (offset, length) pair to account for per-shard checksum overhead.
+/// Returns (adjusted_offset, adjusted_length).
+pub(crate) fn adjust_shard_read_params(
+    offset: usize,
+    length: usize,
+    shard_size: usize,
+    checksum_algo: &HashAlgorithm,
+) -> (usize, usize) {
+    let adj_len = length.div_ceil(shard_size) * checksum_algo.size() + length;
+    let adj_off = offset.div_ceil(shard_size) * checksum_algo.size() + offset;
+    (adj_off, adj_len)
 }
 
 /// Create a BitrotReader from either inline data or disk file stream

crates/ecstore/src/set_disk/core/io_primitives.rs

@@ -54,8 +54,8 @@ use crate::disk::{
 use crate::erasure::coding::BitrotReader;
 use crate::io_support::bitrot::ShardReader;
 use crate::io_support::bitrot::{
-    BitrotReaderStageMetrics, DeferredReaderStripeHandle, create_bitrot_reader_with_stage_metrics,
-    create_deferred_bitrot_reader_with_stripe_handle, object_mmap_read_enabled,
+    BitrotReaderStageMetrics, DeferredReaderStripeHandle, adjust_shard_read_params, create_bitrot_reader_with_stage_metrics,
+    create_deferred_bitrot_reader_with_stripe_handle, object_mmap_read_enabled, object_mmap_read_max_length,
 };
 use crate::set_disk::shard_source::ShardReadCost;
 use futures::stream::{FuturesUnordered, StreamExt};
@@ -63,6 +63,7 @@ use metrics::counter;
 use std::{
     collections::{HashMap, HashSet, VecDeque},
     future::Future,
+    io::Cursor,
     pin::Pin,
     sync::OnceLock,
     task::{Context, Poll},
@@ -1406,6 +1407,101 @@ pub(in crate::set_disk) fn record_bitrot_reader_setup_strategy(
     }
 }
 
+/// When all online shards are local and mmap-read is enabled, materialize
+/// shard bytes with one `batch_shard_pread` instead of per-shard
+/// `spawn_blocking` via `open_disk_reader`.
+#[cfg(unix)]
+#[allow(clippy::too_many_arguments)]
+async fn try_create_bitrot_readers_via_batch_pread(
+    files: &[FileInfo],
+    disks: &[Option<DiskStore>],
+    bucket: &str,
+    object: &str,
+    part_number: usize,
+    read_offset: usize,
+    read_length: usize,
+    shard_size: usize,
+    checksum_algo: HashAlgorithm,
+    skip_verify_bitrot: bool,
+) -> Option<BitrotReaderSetup> {
+    use crate::disk::local::batch_shard_pread;
+
+    let (adj_off, adj_len) = adjust_shard_read_params(read_offset, read_length, shard_size, &checksum_algo);
+    if adj_len > object_mmap_read_max_length() {
+        return None;
+    }
+
+    let mut batch_items: Vec<(usize, std::path::PathBuf, usize, usize)> = Vec::new();
+    for (idx, disk_op) in disks.iter().enumerate() {
+        if files.get(idx).is_some_and(|fi| fi.data.is_some()) {
+            return None;
+        }
+        if let Some(disk) = disk_op.as_ref() {
+            let data_dir = files[idx].data_dir.unwrap_or_default();
+            let path_str = format!("{object}/{data_dir}/part.{part_number}");
+            match disk.get_object_path_if_local(bucket, &path_str) {
+                Some(Ok(p)) => batch_items.push((idx, p, adj_off, adj_len)),
+                _ => return None,
+            }
+        }
+    }
+
+    if batch_items.is_empty() {
+        return None;
+    }
+
+    let requests: Vec<_> = batch_items.iter().map(|(_, p, off, len)| (p.clone(), *off, *len)).collect();
+    let batch_results = batch_shard_pread(requests).await;
+
+    let mut setup = BitrotReaderSetup::new(disks.len());
+    for (i, (idx, _, _, _)) in batch_items.iter().enumerate() {
+        setup.mark_scheduled(*idx);
+        match &batch_results[i] {
+            Ok(bytes) => {
+                let reader = BitrotReader::new(
+                    ShardReader::InMemory(Cursor::new(bytes.clone())),
+                    shard_size,
+                    checksum_algo.clone(),
+                    skip_verify_bitrot,
+                );
+                setup.apply_reader_result(*idx, Ok(Some(reader)));
+            }
+            Err(e) => {
+                setup.apply_reader_result(*idx, Err(e.clone()));
+            }
+        }
+    }
+
+    for (idx, disk_op) in disks.iter().enumerate() {
+        if setup.scheduled[idx] {
+            continue;
+        }
+        setup.mark_scheduled(idx);
+        if disk_op.is_none() {
+            setup.apply_reader_result(idx, Ok(None));
+        }
+    }
+
+    Some(setup)
+}
+
+#[cfg(not(unix))]
+#[allow(clippy::too_many_arguments)]
+async fn try_create_bitrot_readers_via_batch_pread(
+    _files: &[FileInfo],
+    _disks: &[Option<DiskStore>],
+    _bucket: &str,
+    _object: &str,
+    _part_number: usize,
+    _read_offset: usize,
+    _read_length: usize,
+    _shard_size: usize,
+    _checksum_algo: HashAlgorithm,
+    _skip_verify_bitrot: bool,
+) -> Option<BitrotReaderSetup> {
+    None
+}
+
 #[allow(clippy::too_many_arguments)]
 pub(in crate::set_disk) async fn create_bitrot_readers_until_quorum_all_shards(
     files: &[FileInfo],
@@ -1564,6 +1660,44 @@ pub(in crate::set_disk) async fn create_bitrot_readers_until_quorum_with_prefere
     attribution: Option<BitrotReaderSetupAttribution>,
 ) -> BitrotReaderSetup {
     let strategy = get_bitrot_reader_setup_strategy(mode, prefer_data_blocks_first);
+
+    if use_mmap_read
+        && let Some(mut setup) = try_create_bitrot_readers_via_batch_pread(
+            files,
+            disks,
+            bucket,
+            object,
+            part_number,
+            read_offset,
+            read_length,
+            shard_size,
+            checksum_algo.clone(),
+            skip_verify_bitrot,
+        )
+        .await
+    {
+        record_bitrot_reader_setup_strategy(strategy, mode, attribution);
+        fill_deferred_bitrot_readers(
+            &mut setup,
+            files,
+            disks,
+            bucket,
+            object,
+            part_number,
+            read_offset,
+            read_length,
+            shard_size,
+            checksum_algo,
+            skip_verify_bitrot,
+            use_mmap_read,
+            data_shards,
+            parity_shards,
+            mode,
+        );
+        record_bitrot_reader_setup_fanout(strategy, mode, &setup, attribution);
+        return setup;
+    }
+
     if strategy == BitrotReaderSetupStrategy::AllShards {
         return create_bitrot_readers_until_quorum_all_shards(
             files,