From de8cb5f26c24f5414151acf84b4b6b6e02c9e594 Mon Sep 17 00:00:00 2001 From: rdiperri-wasabi <83606987+rdiperri-wasabi@users.noreply.github.com> Date: Mon, 3 Aug 2026 14:53:22 -0700 Subject: [PATCH] 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 Co-authored-by: Zhengchao An --- crates/ecstore/src/disk/disk_store.rs | 4 + crates/ecstore/src/disk/local.rs | 84 +++++++++++ crates/ecstore/src/disk/mod.rs | 9 ++ crates/ecstore/src/io_support/bitrot.rs | 18 ++- .../src/set_disk/core/io_primitives.rs | 138 +++++++++++++++++- 5 files changed, 247 insertions(+), 6 deletions(-) diff --git a/crates/ecstore/src/disk/disk_store.rs b/crates/ecstore/src/disk/disk_store.rs index 415362aff..3cafd621b 100644 --- a/crates/ecstore/src/disk/disk_store.rs +++ b/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 { + self.disk.get_object_path(volume, path) + } + pub fn runtime_state(&self) -> RuntimeDriveHealthState { self.health.runtime_state() } diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index 42b37f9e8..7257e12d7 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/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> { + 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 { + 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()); + } } diff --git a/crates/ecstore/src/disk/mod.rs b/crates/ecstore/src/disk/mod.rs index c51b1eadf..312722310 100644 --- a/crates/ecstore/src/disk/mod.rs +++ b/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> { + 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 { diff --git a/crates/ecstore/src/io_support/bitrot.rs b/crates/ecstore/src/io_support/bitrot.rs index aabd8004c..84401c584 100644 --- a/crates/ecstore/src/io_support/bitrot.rs +++ b/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 diff --git a/crates/ecstore/src/set_disk/core/io_primitives.rs b/crates/ecstore/src/set_disk/core/io_primitives.rs index dccc23901..60cff555b 100644 --- a/crates/ecstore/src/set_disk/core/io_primitives.rs +++ b/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], + bucket: &str, + object: &str, + part_number: usize, + read_offset: usize, + read_length: usize, + shard_size: usize, + checksum_algo: HashAlgorithm, + skip_verify_bitrot: bool, +) -> Option { + 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], + _bucket: &str, + _object: &str, + _part_number: usize, + _read_offset: usize, + _read_length: usize, + _shard_size: usize, + _checksum_algo: HashAlgorithm, + _skip_verify_bitrot: bool, +) -> Option { + 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, ) -> 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,