gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-13 08:36:54 +00:00
Code Issues Packages Projects Releases Wiki Activity

perf(ecstore): reduce inline GET fixed costs (#5985)

Browse Source
This commit is contained in:
houseme
2026-08-12 19:29:46 +08:00
committed by GitHub
parent 270a003c55
commit 848b330825
5 changed files with 96 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/erasure/coding/bitrot.rs
+29 -17
View File
@@ -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<u8>,
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<usize> {
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)
crates/ecstore/src/io_support/bitrot.rs
+6 -4
View File
@@ -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<ShardReader>, 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,
crates/ecstore/src/set_disk/mod.rs
+59 -17
View File
@@ -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<FileInfo>, 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<FileInfo>, 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<FileInfo>, 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]
rustfs/src/app/object_usecase.rs
+1 -5
View File
@@ -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<GetObjectInput>) -> S3Result<S3Response<GetObjectOutput>> {
if let Some(context) = &self.context {
scripts/check_logging_guardrails.sh
+1
View File
@@ -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