From 848b330825de7291553f54f89ab15d3a21d0e6b4 Mon Sep 17 00:00:00 2001 From: houseme Date: Wed, 12 Aug 2026 19:29:46 +0800 Subject: [PATCH] perf(ecstore): reduce inline GET fixed costs (#5985) --- crates/ecstore/src/erasure/coding/bitrot.rs | 46 ++++++++----- crates/ecstore/src/io_support/bitrot.rs | 10 +-- crates/ecstore/src/set_disk/mod.rs | 76 ++++++++++++++++----- rustfs/src/app/object_usecase.rs | 6 +- scripts/check_logging_guardrails.sh | 1 + 5 files changed, 96 insertions(+), 43 deletions(-) diff --git a/crates/ecstore/src/erasure/coding/bitrot.rs b/crates/ecstore/src/erasure/coding/bitrot.rs index 49b1f611c..a12529588 100644 --- a/crates/ecstore/src/erasure/coding/bitrot.rs +++ b/crates/ecstore/src/erasure/coding/bitrot.rs @@ -18,7 +18,11 @@ use std::io::IoSlice; use std::time::Duration; use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; use tracing::error; -use uuid::Uuid; + +const LOG_COMPONENT_ECSTORE: &str = "ecstore"; +const LOG_SUBSYSTEM_ERASURE: &str = "erasure"; +const EVENT_BITROT_SHORT_SHARD_READ: &str = "bitrot_short_shard_read"; +const EVENT_BITROT_HASH_MISMATCH: &str = "bitrot_hash_mismatch"; /// A shard source that may already hold its bytes in memory. /// @@ -73,7 +77,6 @@ pin_project! { buf: Vec, skip_verify: bool, last_verify_duration: Duration, - id: Uuid, } } @@ -90,7 +93,6 @@ where buf: Vec::new(), skip_verify, last_verify_duration: Duration::ZERO, - id: Uuid::new_v4(), } } @@ -118,7 +120,7 @@ where let need = self.hash_algo.size() + want; self.read_scratch_block(need, want).await?; - let (data, verify) = split_and_verify(&self.hash_algo, self.skip_verify, &self.buf[..need], &self.id)?; + let (data, verify) = split_and_verify(&self.hash_algo, self.skip_verify, &self.buf[..need])?; out.copy_from_slice(data); self.last_verify_duration = verify; Ok(want) @@ -157,7 +159,7 @@ where } let filled = fill(&mut self.inner, &mut self.buf[..need]).await?; if filled < need { - return Err(short_shard_read(&self.id, filled.saturating_sub(self.hash_algo.size()), want)); + return Err(short_shard_read(filled.saturating_sub(self.hash_algo.size()), want)); } Ok(()) } @@ -166,15 +168,23 @@ where /// buffer returns its length, a short read is UnexpectedEof (backlog#799 B2). fn finish_len(&self, data_len: usize, want: usize) -> std::io::Result { if data_len < want { - return Err(short_shard_read(&self.id, data_len, want)); + return Err(short_shard_read(data_len, want)); } Ok(data_len) } } /// A truncated shard is `UnexpectedEof`, not a short success (backlog#799 B2). -fn short_shard_read(id: &Uuid, got: usize, want: usize) -> std::io::Error { - error!("bitrot reader short shard read: id={id} got {got} of {want} bytes"); +fn short_shard_read(got: usize, want: usize) -> std::io::Error { + error!( + event = EVENT_BITROT_SHORT_SHARD_READ, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_ERASURE, + state = "failed", + got, + want, + "short shard read: got {got} of {want} bytes" + ); std::io::Error::new(std::io::ErrorKind::UnexpectedEof, format!("short shard read: got {got} of {want} bytes")) } @@ -184,12 +194,7 @@ fn short_shard_read(id: &Uuid, got: usize, want: usize) -> std::io::Error { /// hash never reaches the caller's buffer. The verify duration is returned /// rather than stored so this stays a free function usable while `self` is /// borrowed for the block. -fn split_and_verify<'a>( - hash_algo: &HashAlgorithm, - skip_verify: bool, - block: &'a [u8], - id: &Uuid, -) -> std::io::Result<(&'a [u8], Duration)> { +fn split_and_verify<'a>(hash_algo: &HashAlgorithm, skip_verify: bool, block: &'a [u8]) -> std::io::Result<(&'a [u8], Duration)> { let (hash, data) = block.split_at(hash_algo.size()); if skip_verify { return Ok((data, Duration::ZERO)); @@ -198,7 +203,14 @@ fn split_and_verify<'a>( let actual_hash = hash_algo.hash_encode(data); let verify = verify_start.elapsed(); if actual_hash.as_ref() != hash { - error!("bitrot reader hash mismatch, id={id} data_len={}", data.len()); + error!( + event = EVENT_BITROT_HASH_MISMATCH, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_ERASURE, + state = "failed", + data_len = data.len(), + "bitrot hash mismatch" + ); return Err(std::io::Error::new(std::io::ErrorKind::InvalidData, "bitrot hash mismatch")); } Ok((data, verify)) @@ -254,7 +266,7 @@ where // `need` bytes returns `None` and falls through to the scratch path, // keeping the short-read contract. if let Some(block) = self.inner.try_take_block(need) { - let (data, verify) = split_and_verify(&self.hash_algo, self.skip_verify, &block, &self.id)?; + let (data, verify) = split_and_verify(&self.hash_algo, self.skip_verify, &block)?; out.extend_from_slice(data); self.last_verify_duration = verify; return Ok(want); @@ -264,7 +276,7 @@ where // the sink differs (`extend_from_slice` into `out` instead of // `copy_from_slice` into a pre-zeroed buffer). self.read_scratch_block(need, want).await?; - let (data, verify) = split_and_verify(&self.hash_algo, self.skip_verify, &self.buf[..need], &self.id)?; + let (data, verify) = split_and_verify(&self.hash_algo, self.skip_verify, &self.buf[..need])?; out.extend_from_slice(data); self.last_verify_duration = verify; Ok(want) diff --git a/crates/ecstore/src/io_support/bitrot.rs b/crates/ecstore/src/io_support/bitrot.rs index 84401c584..c52459a9d 100644 --- a/crates/ecstore/src/io_support/bitrot.rs +++ b/crates/ecstore/src/io_support/bitrot.rs @@ -623,11 +623,12 @@ async fn create_bitrot_reader_from_bytes_with_stage_metrics( let reader_construction_start = stage_metrics_enabled.then(Instant::now); let (offset, length) = bitrot_encoded_range(offset, length, shard_size, checksum_algo.clone()); + let inline_source = inline_data.is_some(); let source = BitrotReaderSource { inline_data, disk: disk.cloned(), - bucket: bucket.to_string(), - path: path.to_string(), + bucket: if inline_source { String::new() } else { bucket.to_string() }, + path: if inline_source { String::new() } else { path.to_string() }, offset, length, use_mmap_read, @@ -698,11 +699,12 @@ pub(crate) fn create_deferred_bitrot_reader_with_stripe_handle( ) -> (BitrotReader, DeferredReaderStripeHandle) { let stripe_stride = shard_size + checksum_algo.size(); let (offset, length) = bitrot_encoded_range(offset, length, shard_size, checksum_algo.clone()); + let inline_source = inline_data.is_some(); let source = BitrotReaderSource { inline_data, disk, - bucket: bucket.to_string(), - path: path.to_string(), + bucket: if inline_source { String::new() } else { bucket.to_string() }, + path: if inline_source { String::new() } else { path.to_string() }, offset, length, use_mmap_read, diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 050297606..95b729640 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -3190,23 +3190,28 @@ async fn try_read_inline_data_shards_direct( return None; } - let mut body = Vec::with_capacity(object_size); - let mut remaining = object_size; - for reader in readers.iter_mut().take(data_shards) { + let shards_needed = object_size.div_ceil(read_length); + if shards_needed > data_shards { + return None; + } + let encoded_capacity = read_length.checked_mul(shards_needed)?; + let mut body = Vec::with_capacity(encoded_capacity); + for reader in readers.iter_mut().take(shards_needed) { let reader = reader.as_mut()?; - let mut shard = vec![0u8; read_length]; - let Ok(read) = reader.read(&mut shard).await else { + let Ok(read) = reader.read_appending(&mut body, read_length).await else { return None; }; if read != read_length { return None; } - let take = remaining.min(shard.len()); - body.extend_from_slice(&shard[..take]); - remaining -= take; - if remaining == 0 { - return Some(Bytes::from(body)); + if body.len() >= object_size { + let body = Bytes::from(body); + return Some(if body.len() == object_size { + body + } else { + body.slice(..object_size) + }); } } @@ -8847,10 +8852,17 @@ mod tests { )); } - async fn inline_bitrot_files_for_payload(payload: &[u8]) -> (coding::Erasure, Vec, usize, HashAlgorithm) { - let erasure = coding::Erasure::new(4, 2, 1024 * 1024); + async fn inline_bitrot_files_for_payload_with_mode( + payload: &[u8], + uses_legacy: bool, + ) -> (coding::Erasure, Vec, usize, HashAlgorithm) { + let erasure = coding::Erasure::new_with_options(4, 2, 1024 * 1024, uses_legacy); let read_length = erasure.shard_file_offset(0, payload.len(), payload.len()); - let checksum_algo = HashAlgorithm::HighwayHash256S; + let checksum_algo = if uses_legacy { + HashAlgorithm::HighwayHash256SLegacy + } else { + HashAlgorithm::HighwayHash256S + }; let shards = erasure.encode_data(payload).expect("payload should encode"); let mut files = Vec::with_capacity(shards.len()); @@ -8872,6 +8884,10 @@ mod tests { (erasure, files, read_length, checksum_algo) } + async fn inline_bitrot_files_for_payload(payload: &[u8]) -> (coding::Erasure, Vec, usize, HashAlgorithm) { + inline_bitrot_files_for_payload_with_mode(payload, false).await + } + fn inline_data_shard_fileinfo( name: &str, data_blocks: usize, @@ -8951,15 +8967,41 @@ mod tests { assert_eq!(body.as_ref(), payload); } + #[tokio::test] + async fn inline_data_shards_direct_read_reassembles_legacy_payload_with_padding() { + let payload = b"legacy inline payload whose size is not divisible by the data shard count"; + let (erasure, files, read_length, checksum_algo) = inline_bitrot_files_for_payload_with_mode(payload, true).await; + assert_ne!(payload.len() % erasure.data_shards, 0, "test payload must exercise EC padding"); + let mut readers = build_inline_bitrot_readers( + &files, + erasure.data_shards, + "bucket", + "object", + read_length, + erasure.shard_size(), + &checksum_algo, + false, + ) + .await + .expect("legacy inline bitrot readers should build"); + + let body = try_read_inline_data_shards_direct(&mut readers, erasure.data_shards, read_length, payload.len()) + .await + .expect("legacy data shard direct read should succeed"); + + assert_eq!(body.len(), payload.len()); + assert_eq!(body.as_ref(), payload); + } + #[tokio::test] async fn inline_data_shards_direct_read_rejects_corrupt_shard() { let payload = b"small inline object payload that will be corrupted"; let (erasure, mut files, read_length, checksum_algo) = inline_bitrot_files_for_payload(payload).await; - let first = files[0].data.as_mut().expect("first shard should exist"); - let mut corrupted = first.to_vec(); + let second = files[1].data.as_mut().expect("second shard should exist"); + let mut corrupted = second.to_vec(); let last = corrupted.last_mut().expect("encoded shard should not be empty"); *last ^= 0xff; - *first = Bytes::from(corrupted); + *second = Bytes::from(corrupted); let mut readers = build_inline_bitrot_readers( &files, @@ -8976,7 +9018,7 @@ mod tests { let body = try_read_inline_data_shards_direct(&mut readers, 4, read_length, payload.len()).await; - assert!(body.is_none()); + assert!(body.is_none(), "a later corrupt shard must discard the already-appended body prefix"); } #[test] diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index 48cba8ccb..fde4e9924 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -6432,11 +6432,7 @@ impl DefaultObjectUsecase { }) } - #[instrument( - level = "info", - skip(self, req), - fields(start_time=?time::OffsetDateTime::now_utc()) - )] + #[instrument(level = "trace", skip(self, req))] #[hotpath::measure(impl_type = "DefaultObjectUsecase")] pub async fn execute_get_object(&self, req: S3Request) -> S3Result> { if let Some(context) = &self.context { diff --git a/scripts/check_logging_guardrails.sh b/scripts/check_logging_guardrails.sh index d109262c4..a5095d8b3 100755 --- a/scripts/check_logging_guardrails.sh +++ b/scripts/check_logging_guardrails.sh @@ -988,6 +988,7 @@ trace_hot_spans=( "crates/ecstore/src/core/sets.rs:list_objects_v2" "crates/ecstore/src/set_disk/ops/list.rs:list_objects_v2" "rustfs/src/app/bucket_usecase.rs:execute_list_objects_v2" + "rustfs/src/app/object_usecase.rs:execute_get_object" ) for hot_span in "${trace_hot_spans[@]}"; do