gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-31 01:09:23 +00:00
Code Issues Packages Projects Releases Wiki Activity

perf(ecstore): optimize bounded small-object GET paths (#6808)

Browse Source 
* perf(ecstore): bound mid-size GET decode buffering

Use a single in-flight decoded stripe for the gated mid-size GET path and avoid its outer synchronization mutex while preserving the general codec reader behavior. Add full, partial, degraded, error, and cancellation coverage for the bounded reader.

Co-Authored-By: heihutu <heihutu@gmail.com>

* perf(ecstore): unify small GET path validation

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(ecstore): bound mid-size prefetch and preserve gate metrics

Co-Authored-By: heihutu <heihutu@gmail.com>

* perf(ecstore): cache small-object read path plan

Co-Authored-By: heihutu <heihutu@gmail.com>

* perf(ecstore): cache GET path plan and verify wiring

Co-Authored-By: heihutu <heihutu@gmail.com>

* test(ecstore): remove redundant metadata clone

Co-Authored-By: heihutu <heihutu@gmail.com>

* test(ecstore): preserve dual inflight prefetch contract

Co-Authored-By: heihutu <heihutu@gmail.com>

* test(ecstore): make prefetch assertion deterministic

Co-Authored-By: heihutu <heihutu@gmail.com>

---------

Co-authored-by: heihutu <heihutu@gmail.com>
This commit is contained in:
houseme
2026-08-29 01:31:10 +08:00
committed by GitHub
parent 75cd3885f3
commit 0953f7e912
4 changed files with 608 additions and 103 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/erasure/coding/decode_reader.rs
+181 -17
View File
@@ -27,6 +27,8 @@ use std::io;
use std::io::ErrorKind;
use std::pin::Pin;
use std::sync::Mutex;
#[cfg(test)]
use std::sync::atomic::{AtomicU64, Ordering};
use std::task::{Context, Poll, ready};
use std::time::Instant;
use tokio::io::{AsyncRead, ReadBuf};
@@ -38,6 +40,14 @@ const DEFAULT_RUSTFS_GET_CODEC_STREAMING_MAX_INFLIGHT: usize = 2;
const FILL_POLICY_SINGLE_INFLIGHT: &str = "single_inflight";
const FILL_POLICY_DUAL_INFLIGHT: &str = "dual_inflight";
#[cfg(test)]
static SINGLE_INFLIGHT_CONSTRUCTIONS: AtomicU64 = AtomicU64::new(0);
#[cfg(test)]
pub(crate) fn test_single_inflight_construction_count() -> u64 {
SINGLE_INFLIGHT_CONSTRUCTIONS.load(Ordering::Relaxed)
}
type FillTask = oneshot::Receiver<FillResult>;
struct FillWorker {
@@ -155,6 +165,23 @@ where
Self::new_with_fill_policy_inner(source, engine, total_length, metrics_path, FillPolicy::from_env())
}
/// Construct the bounded reader without lookahead.
///
/// Mid-size GETs are latency-sensitive and are already gated to a single
/// plain part. Keeping one stripe in flight avoids retaining a second
/// decoded output buffer while preserving the same source, reconstruction,
/// bitrot and cancellation semantics as the general streaming reader.
pub(crate) fn new_single_inflight_with_metrics_path(
source: S,
engine: E,
total_length: usize,
metrics_path: &'static str,
) -> io::Result<Self> {
#[cfg(test)]
SINGLE_INFLIGHT_CONSTRUCTIONS.fetch_add(1, Ordering::Relaxed);
Self::new_with_fill_policy_inner(source, engine, total_length, metrics_path, FillPolicy::SingleInFlight)
}
fn new_with_fill_policy_inner(
source: S,
engine: E,
@@ -602,7 +629,8 @@ where
loop {
if self.output_pos < self.output_buf.len() {
if self.prefetched_bufs.len() < self.fill_policy.max_inflight()
if self.fill_policy == FillPolicy::DualInFlight
&& self.prefetched_bufs.len() < self.fill_policy.max_inflight()
&& self.prefetch_error.is_none()
&& self.remaining > 0
&& let Poll::Ready(result) = self.poll_prefetch(cx)
@@ -1620,6 +1648,149 @@ mod tests {
assert_eq!(decoded, data);
}
#[tokio::test]
async fn single_inflight_reader_reads_full_body_without_lookahead() {
let erasure = Erasure::new(4, 2, 32);
let data = (0..96u8).collect::<Vec<_>>();
let read_count = Arc::new(AtomicUsize::new(0));
let mut source = source_from_data(&erasure, &data, &[]);
source.read_count = Some(Arc::clone(&read_count));
let engine = LegacyEcDecodeEngine::new(erasure);
let mut reader = ErasureDecodeReader::new_single_inflight_with_metrics_path(
source,
engine,
data.len(),
GET_OBJECT_PATH_CODEC_STREAMING,
)
.expect("single-inflight reader should be constructed");
let mut decoded = Vec::new();
reader
.read_to_end(&mut decoded)
.await
.expect("single-inflight reader should decode the complete body");
assert_eq!(decoded, data);
assert_eq!(
read_count.load(Ordering::SeqCst),
3,
"single-inflight must not read ahead after the final stripe"
);
}
#[tokio::test]
async fn single_inflight_reader_preserves_partial_reads() {
let erasure = Erasure::new(4, 2, 32);
let data = (0..83u8).collect::<Vec<_>>();
let engine = LegacyEcDecodeEngine::new(erasure.clone());
let mut reader = ErasureDecodeReader::new_single_inflight_with_metrics_path(
source_from_data(&erasure, &data, &[]),
engine,
data.len(),
GET_OBJECT_PATH_CODEC_STREAMING,
)
.expect("single-inflight reader should be constructed");
let mut decoded = Vec::with_capacity(data.len());
let mut chunk = [0u8; 3];
loop {
let read = reader.read(&mut chunk).await.expect("partial read should succeed");
if read == 0 {
break;
}
decoded.extend_from_slice(&chunk[..read]);
}
assert_eq!(decoded, data);
}
#[tokio::test]
async fn single_inflight_reader_does_not_prefetch_before_output_is_drained() {
let erasure = Erasure::new(4, 2, 32);
let data = (0..96u8).collect::<Vec<_>>();
let read_count = Arc::new(AtomicUsize::new(0));
let mut source = source_from_data(&erasure, &data, &[]);
source.read_count = Some(Arc::clone(&read_count));
let engine = LegacyEcDecodeEngine::new(erasure);
let mut reader = ErasureDecodeReader::new_single_inflight_with_metrics_path(
source,
engine,
data.len(),
GET_OBJECT_PATH_CODEC_STREAMING,
)
.expect("single-inflight reader should be constructed");
let mut first = [0u8; 3];
reader
.read_exact(&mut first)
.await
.expect("first partial read should succeed");
assert_eq!(
read_count.load(Ordering::SeqCst),
1,
"single-inflight must not prefetch while output remains"
);
assert_eq!(&first, &data[..3]);
}
#[tokio::test]
async fn single_inflight_reader_reconstructs_degraded_body() {
let erasure = Erasure::new(4, 2, 32);
let data = (0..97u16).map(|value| value as u8).collect::<Vec<_>>();
let engine = LegacyEcDecodeEngine::new(erasure.clone());
let mut reader = ErasureDecodeReader::new_single_inflight_with_metrics_path(
source_from_data(&erasure, &data, &[1]),
engine,
data.len(),
GET_OBJECT_PATH_CODEC_STREAMING,
)
.expect("single-inflight reader should be constructed");
let mut decoded = Vec::new();
reader
.read_to_end(&mut decoded)
.await
.expect("a readable degraded stripe should be reconstructed");
assert_eq!(decoded, data);
}
#[tokio::test]
async fn single_inflight_reader_surfaces_error_after_buffered_body() {
let erasure = Erasure::new(4, 2, 32);
let first_stripe = (0..32u8).collect::<Vec<_>>();
let first_state = source_from_data(&erasure, &first_stripe, &[])
.stripes
.pop_front()
.expect("first stripe should exist");
let source = VecStripeSource {
stripes: VecDeque::from([
first_state,
StripeReadState::from_parts(Vec::new(), Vec::new(), erasure.data_shards),
]),
read_quorum: erasure.data_shards,
read_count: None,
};
let engine = LegacyEcDecodeEngine::new(erasure);
let mut reader = ErasureDecodeReader::new_single_inflight_with_metrics_path(
source,
engine,
first_stripe.len() + 1,
GET_OBJECT_PATH_CODEC_STREAMING,
)
.expect("single-inflight reader should be constructed");
let mut decoded = Vec::new();
let error = reader
.read_to_end(&mut decoded)
.await
.expect_err("short source error should be returned after buffered bytes");
assert_eq!(error.kind(), ErrorKind::Other);
assert_eq!(decoded, first_stripe);
}
#[tokio::test]
async fn erasure_decode_reader_stops_at_eof_for_empty_object() {
let erasure = Erasure::new(4, 2, 32);
@@ -1882,14 +2053,9 @@ mod tests {
};
let engine = LegacyEcDecodeEngine::new(Erasure::new(1, 0, 32));
let task = tokio::spawn(async move {
let mut reader = ErasureDecodeReader::new_with_fill_policy(
source,
engine,
1,
GET_OBJECT_PATH_CODEC_STREAMING,
FillPolicy::SingleInFlight,
)
.expect("reader should be constructed");
let mut reader =
ErasureDecodeReader::new_single_inflight_with_metrics_path(source, engine, 1, GET_OBJECT_PATH_CODEC_STREAMING)
.expect("reader should be constructed");
let mut first_read = [0u8; 1];
let _ = reader.read(&mut first_read).await;
});
@@ -2226,7 +2392,7 @@ mod tests {
engine,
data.len(),
GET_OBJECT_PATH_CODEC_STREAMING,
FillPolicy::SingleInFlight,
FillPolicy::DualInFlight,
)
.expect("reader should be constructed");
let mut first_read = [0u8; 1];
@@ -2235,13 +2401,11 @@ mod tests {
assert_eq!(read, first_read.len());
assert_eq!(first_read[0], data[0]);
timeout(Duration::from_secs(1), async {
while read_count.load(Ordering::SeqCst) < 2 {
yield_now().await;
}
})
.await
.expect("reader should start reading the next stripe before the current output buffer is fully consumed");
assert_eq!(
read_count.load(Ordering::SeqCst),
2,
"dual-inflight reader should prefetch the next stripe before returning the first byte"
);
}
#[tokio::test]
crates/ecstore/src/set_disk/mod.rs
+245 -20
View File
@@ -2357,6 +2357,7 @@ enum GetCodecStreamingFallbackReason {
ReadQuorumNotSafe,
MultipartPartLimit,
CopySourceDemandBound,
InvalidMetadataShape,
}
impl GetCodecStreamingFallbackReason {
@@ -2379,6 +2380,7 @@ impl GetCodecStreamingFallbackReason {
Self::ReadQuorumNotSafe => "read_quorum_not_safe",
Self::MultipartPartLimit => "multipart_part_limit",
Self::CopySourceDemandBound => "copy_source_demand_bound",
Self::InvalidMetadataShape => "invalid_metadata_shape",
}
}
}
@@ -2438,6 +2440,7 @@ enum GetDirectMemoryFallbackReason {
Remote,
ObjectInfoMultipart,
FileInfoMultipart,
MetadataShape,
InvalidSize,
SizeMismatch,
AboveThreshold,
@@ -2463,6 +2466,7 @@ impl GetDirectMemoryFallbackReason {
Self::Remote => "remote",
Self::ObjectInfoMultipart => "object_info_multipart",
Self::FileInfoMultipart => "file_info_multipart",
Self::MetadataShape => "metadata_shape",
Self::InvalidSize => "invalid_size",
Self::SizeMismatch => "size_mismatch",
Self::AboveThreshold => "above_threshold",
@@ -2511,10 +2515,42 @@ fn record_get_object_reader_path_observation(
object_class: GetCodecStreamingObjectClass,
size_bucket: &'static str,
) {
#[cfg(test)]
LAST_GET_OBJECT_READER_PATH.store(
match path {
crate::set_disk::read::GET_OBJECT_PATH_MID_SIZE_STREAMING => 1,
GET_OBJECT_PATH_DIRECT_MEMORY => 2,
GET_OBJECT_PATH_INLINE_DIRECT => 3,
GET_OBJECT_PATH_BODY_CACHE => 4,
GET_OBJECT_PATH_CODEC_STREAMING => 5,
GET_OBJECT_PATH_REMOTE_TRANSITION => 6,
GET_OBJECT_PATH_EMPTY => 7,
_ => 255,
},
Ordering::Relaxed,
);
rustfs_io_metrics::record_get_object_reader_path(path);
rustfs_io_metrics::record_get_object_reader_path_by_size(path, object_class.as_str(), size_bucket);
}
#[cfg(test)]
static LAST_GET_OBJECT_READER_PATH: AtomicU64 = AtomicU64::new(0);
#[cfg(test)]
pub(crate) fn reset_test_get_object_reader_path() {
LAST_GET_OBJECT_READER_PATH.store(0, Ordering::Relaxed);
}
#[cfg(test)]
pub(crate) fn test_get_object_reader_selected_mid_size() -> bool {
LAST_GET_OBJECT_READER_PATH.load(Ordering::Relaxed) == 1
}
#[cfg(test)]
pub(crate) fn test_get_object_reader_path_id() -> u64 {
LAST_GET_OBJECT_READER_PATH.load(Ordering::Relaxed)
}
fn classify_get_codec_streaming_object_class(
range: &Option<HTTPRangeSpec>,
object_info: &ObjectInfo,
@@ -2559,6 +2595,26 @@ fn get_small_object_direct_memory_decision_with_threshold(
opts: &ObjectOptions,
enabled: bool,
threshold: usize,
) -> GetDirectMemoryDecision {
get_small_object_direct_memory_decision_with_threshold_and_plan(
range,
object_info,
fi,
opts,
enabled,
threshold,
ReadPathPlan::new(object_info, fi),
)
}
fn get_small_object_direct_memory_decision_with_threshold_and_plan(
range: &Option<HTTPRangeSpec>,
object_info: &ObjectInfo,
fi: &FileInfo,
opts: &ObjectOptions,
enabled: bool,
threshold: usize,
plan: ReadPathPlan,
) -> GetDirectMemoryDecision {
if !enabled {
return GetDirectMemoryDecision::Fallback(GetDirectMemoryFallbackReason::Disabled);
@@ -2617,14 +2673,24 @@ fn get_small_object_direct_memory_decision_with_threshold(
return GetDirectMemoryDecision::Fallback(GetDirectMemoryFallbackReason::FileInfoMultipart);
}
let Ok(object_size) = usize::try_from(fi.size) else {
return GetDirectMemoryDecision::Fallback(GetDirectMemoryFallbackReason::InvalidSize);
if object_info.size != fi.size {
return GetDirectMemoryDecision::Fallback(GetDirectMemoryFallbackReason::SizeMismatch);
}
let Some(shape) = plan.shape() else {
return GetDirectMemoryDecision::Fallback(GetDirectMemoryFallbackReason::MetadataShape);
};
let object_size = shape.object_size;
if object_size == 0 {
return GetDirectMemoryDecision::Fallback(GetDirectMemoryFallbackReason::InvalidSize);
}
if object_info.size != fi.size {
return GetDirectMemoryDecision::Fallback(GetDirectMemoryFallbackReason::SizeMismatch);
if !plan.is_plain() {
if object_info.is_encrypted() || fi.metadata.keys().any(|key| is_object_encryption_marker(key)) {
return GetDirectMemoryDecision::Fallback(GetDirectMemoryFallbackReason::Encrypted);
}
if object_info.is_compressed() || fi.is_compressed() {
return GetDirectMemoryDecision::Fallback(GetDirectMemoryFallbackReason::Compressed);
}
return GetDirectMemoryDecision::Fallback(GetDirectMemoryFallbackReason::Remote);
}
if object_size > threshold {
return GetDirectMemoryDecision::Fallback(GetDirectMemoryFallbackReason::AboveThreshold);
@@ -2633,6 +2699,7 @@ fn get_small_object_direct_memory_decision_with_threshold(
GetDirectMemoryDecision::Use { object_size }
}
#[allow(dead_code, reason = "asserted by this file's gate tests")]
fn get_small_object_direct_memory_decision(
range: &Option<HTTPRangeSpec>,
object_info: &ObjectInfo,
@@ -2667,6 +2734,7 @@ fn should_prefer_codec_streaming_data_blocks_first_reader_setup(
max_size > 0 && object_size <= max_size
}
#[allow(dead_code, reason = "asserted by this file's gate tests")]
fn get_codec_streaming_reader_gate(
bucket: &str,
object: &str,
@@ -2675,6 +2743,29 @@ fn get_codec_streaming_reader_gate(
object_info: &ObjectInfo,
fi: &FileInfo,
lock_optimization_enabled: bool,
) -> GetCodecStreamingGate {
get_codec_streaming_reader_gate_with_plan(
bucket,
object,
part_number,
object_class,
object_info,
fi,
lock_optimization_enabled,
ReadPathPlan::new(object_info, fi),
)
}
#[allow(clippy::too_many_arguments, reason = "keeps the hot-path gate inputs explicit")]
fn get_codec_streaming_reader_gate_with_plan(
bucket: &str,
object: &str,
part_number: Option<usize>,
object_class: GetCodecStreamingObjectClass,
object_info: &ObjectInfo,
fi: &FileInfo,
lock_optimization_enabled: bool,
plan: ReadPathPlan,
) -> GetCodecStreamingGate {
let config = get_codec_streaming_config();
@@ -2790,6 +2881,22 @@ fn get_codec_streaming_reader_gate(
};
}
}
if object_class == GetCodecStreamingObjectClass::PlainSinglePart {
let Some(_shape) = plan.shape() else {
return GetCodecStreamingGate {
object_class,
decision: GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::InvalidMetadataShape),
prefer_data_blocks_first_reader_setup: false,
};
};
if !plan.is_plain() {
return GetCodecStreamingGate {
object_class,
decision: GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::InvalidMetadataShape),
prefer_data_blocks_first_reader_setup: false,
};
}
}
let Ok(min_size) = i64::try_from(config.min_size) else {
return GetCodecStreamingGate {
object_class,
@@ -4074,6 +4181,80 @@ fn object_fits_single_block(object_size: i64, block_size: usize) -> bool {
}
}
/// The common metadata contract consumed by all bounded GET fast paths.
///
/// `ObjectInfo` is assembled from `FileInfo`, but callers may also provide a
/// prepared snapshot or metadata from an older peer. Never let either copy
/// independently decide that a request is safe: a disagreement must fall
/// back to the regular reader. The returned size is the only size used for
/// fast-path allocation and reader setup.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(super) struct ReadPathShape {
pub(super) object_size: usize,
}
impl ReadPathShape {
pub(super) fn is_plain(self, object_info: &ObjectInfo, fi: &FileInfo) -> bool {
!object_info.is_encrypted()
&& !fi.metadata.keys().any(|key| is_object_encryption_marker(key))
&& !object_info.is_compressed()
&& !fi.is_compressed()
&& !object_info.is_remote()
&& !fi.is_remote()
}
}
/// Request-local metadata decision reused by all small-object GET gates.
///
/// The metadata pair is immutable for the lifetime of `get_object_reader`, so
/// validating it once avoids repeating part-array and transform scans on every
/// fast-path predicate while keeping the trust boundary fail-closed.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
struct ReadPathPlan {
shape: Option<ReadPathShape>,
plain: bool,
}
impl ReadPathPlan {
fn new(object_info: &ObjectInfo, fi: &FileInfo) -> Self {
let shape = read_path_shape(object_info, fi);
let plain = shape.is_some_and(|shape| shape.is_plain(object_info, fi));
Self { shape, plain }
}
const fn shape(self) -> Option<ReadPathShape> {
self.shape
}
const fn is_plain(self) -> bool {
self.plain
}
}
/// Validate the single-part geometry shared by inline, direct-memory, and
/// bounded mid-size readers. This is deliberately fail-closed: a stale or
/// mixed-version `ObjectInfo`/`FileInfo` pair must use the legacy path rather
/// than risk allocating or decoding with a mismatched size.
pub(super) fn read_path_shape(object_info: &ObjectInfo, fi: &FileInfo) -> Option<ReadPathShape> {
if object_info.parts.len() != 1 || fi.parts.len() != 1 || object_info.size < 0 || fi.size < 0 || object_info.size != fi.size {
return None;
}
let object_part = object_info.parts.first()?;
let file_part = fi.parts.first()?;
let object_size = usize::try_from(fi.size).ok()?;
if object_part.number != file_part.number
|| object_part.size != file_part.size
|| object_part.actual_size != file_part.actual_size
|| file_part.size != object_size
|| file_part.actual_size != fi.size
{
return None;
}
Some(ReadPathShape { object_size })
}
fn should_use_inline_small_fast_path(is_inline_buffer: bool, object_size: i64, block_size: usize) -> bool {
is_inline_buffer && object_fits_single_block(object_size, block_size)
}
@@ -4082,35 +4263,37 @@ fn should_use_single_block_non_inline_fast_path(is_inline_buffer: bool, object_s
!is_inline_buffer && object_fits_single_block(object_size, block_size)
}
#[allow(dead_code, reason = "asserted by this file's gate tests")]
fn should_use_inline_fast_path(
range: &Option<HTTPRangeSpec>,
object_info: &ObjectInfo,
fi: &FileInfo,
opts: &ObjectOptions,
) -> bool {
should_use_inline_fast_path_with_plan(range, object_info, fi, opts, ReadPathPlan::new(object_info, fi))
}
fn should_use_inline_fast_path_with_plan(
range: &Option<HTTPRangeSpec>,
object_info: &ObjectInfo,
fi: &FileInfo,
opts: &ObjectOptions,
plan: ReadPathPlan,
) -> bool {
if !object_info.is_inline_fast_path_eligible() || fi.data.is_none() || range.is_some() || opts.part_number.is_some() {
return false;
}
let Some(shape) = plan.shape() else {
return false;
};
// The persisted marker is authoritative for the storage decision, but it
// is still untrusted metadata at this boundary. Revalidate its geometry so
// a stale/corrupt marker cannot route an oversized payload into the
// in-memory decoder. The persisted marker is the writer's policy decision;
// reloading the current storage-class config here would add a hot-path
// snapshot load and could make an already durable inline object unreadable
// after an operator changes the admission policy. The independent
// direct-memory reader remains capped at 128 KiB below.
let Some(part) = fi.parts.first() else {
return false;
};
// Plain inline metadata must describe one coherent object. In particular,
// do not trust `fi.size` for the allocation below when a corrupt part has
// a different `actual_size`; compressed/unknown `actual_size` objects are
// excluded earlier by the plain-object checks.
if fi.size < 0
|| fi.size > INLINE_FAST_PATH_MAX_OBJECT_SIZE
|| object_info.size != fi.size
|| part.actual_size != fi.size
// in-memory decoder. The independent direct-memory reader remains capped
// at 128 KiB below.
if !plan.is_plain()
|| shape.object_size > usize::try_from(INLINE_FAST_PATH_MAX_OBJECT_SIZE).expect("inline fast path limit fits usize")
|| fi.erasure.data_blocks == 0
|| fi.erasure.block_size == 0
{
@@ -11022,6 +11205,41 @@ mod tests {
));
}
#[test]
fn read_path_shape_rejects_mismatched_metadata_copies_and_transforms() {
let (object_info, fi, _) = direct_memory_test_metadata(1024);
let shape = read_path_shape(&object_info, &fi).expect("matching metadata should have a valid shape");
assert_eq!(shape.object_size, 1024);
assert!(shape.is_plain(&object_info, &fi));
let mut bad_part_number = object_info.clone();
Arc::make_mut(&mut bad_part_number.parts)[0].number = 2;
assert!(read_path_shape(&bad_part_number, &fi).is_none());
let mut bad_part_size = fi.clone();
bad_part_size.parts[0].size += 1;
assert!(read_path_shape(&object_info, &bad_part_size).is_none());
let mut bad_actual_size = object_info.clone();
Arc::make_mut(&mut bad_actual_size.parts)[0].actual_size += 1;
assert!(read_path_shape(&bad_actual_size, &fi).is_none());
let mut compressed = fi.clone();
insert_str(&mut compressed.metadata, SUFFIX_COMPRESSION, "zstd".to_string());
let compressed_shape = read_path_shape(&object_info, &compressed).expect("geometry remains valid");
assert!(!compressed_shape.is_plain(&object_info, &compressed));
let mut encrypted = fi;
encrypted
.metadata
.insert("x-amz-server-side-encryption".to_string(), "AES256".to_string());
assert!(
!read_path_shape(&object_info, &encrypted)
.expect("geometry remains valid")
.is_plain(&object_info, &encrypted)
);
}
#[test]
fn inline_fast_path_rejects_part_number_requests() {
let (mut object_info, mut fi, opts) = direct_memory_test_metadata(1024);
@@ -11170,6 +11388,13 @@ mod tests {
get_small_object_direct_memory_decision_with_threshold(&None, &object_info, &fi, &opts, true, 128 * 1024),
GetDirectMemoryDecision::Use { object_size: 1024 }
);
let mut corrupt_part = fi;
corrupt_part.parts[0].actual_size += 1;
assert_eq!(
get_small_object_direct_memory_decision_with_threshold(&None, &object_info, &corrupt_part, &opts, true, 128 * 1024),
GetDirectMemoryDecision::Fallback(GetDirectMemoryFallbackReason::MetadataShape)
);
}
#[test]
crates/ecstore/src/set_disk/ops/object.rs
+156 -61
View File
@@ -30,32 +30,34 @@ use super::super::{
GET_OBJECT_PATH_DIRECT_MEMORY, GET_OBJECT_PATH_EMPTY, GET_OBJECT_PATH_INLINE_DIRECT, GET_OBJECT_PATH_INTERNAL_META,
GET_OBJECT_PATH_LEGACY_DUPLEX, GET_OBJECT_PATH_REMOTE_TRANSITION, GET_OBJECT_PATH_SET_DISK, GET_STAGE_DECODE, GET_STAGE_EMIT,
GET_STAGE_INLINE_PREPARE, GET_STAGE_LOCK_ACQUIRE, GET_STAGE_METADATA, GET_STAGE_OBJECT_INFO, GET_STAGE_PATH_DECISION,
GET_STAGE_READER_SETUP, GenericError, GetCodecStreamingDecision, GetDirectMemoryDecision, GetObjectReader, HTTPRangeSpec,
HashAlgorithm, HashMap, HashReader, HashSet, HeaderMap, HealChannelPriority, InstanceContext, Instant, LOG_COMPONENT_ECSTORE,
LOG_SUBSYSTEM_SET_DISK, OBJECT_OP_IGNORED_ERRS, ObjectApiError, ObjectInfo, ObjectKey, ObjectLockConfigSnapshot,
ObjectLockConfigState, ObjectOptions, ObjectReader, ObjectToDelete, OffsetDateTime, Ordering, Pin, PutObjReader,
RUSTFS_META_BUCKET, RUSTFS_META_TMP_BUCKET, ReaderImpl, ReplicateDecision, ReplicationObjectBridge, Result,
SET_DISK_COMMIT_TAIL_WARN_THRESHOLD_MS, SLASH_SEPARATOR, SUFFIX_ACTUAL_SIZE, SUFFIX_COMPRESSION, SUFFIX_COMPRESSION_SIZE,
SUFFIX_RESTORE_OPERATION_ID, SetDisks, SmallWritePath, StorageError, TRANSITION_COMPLETE, UpdateMetadataOpts, Uuid,
WriteLayout, X_AMZ_OBJECT_LOCK_LEGAL_HOLD, X_AMZ_OBJECT_LOCK_MODE, X_AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE, X_AMZ_RESTORE,
adaptive_duplex_buffer_size, build_get_object_info, build_inline_bitrot_readers, build_inline_bitrot_readers_from_refs,
can_try_inline_data_shards_direct, check_object_lock_delete, check_object_lock_for_deletion_with_state,
check_object_lock_retention_update, classify_get_codec_streaming_object_class, classify_put_write_path,
classify_storage_error, collect_inline_data_shard_fileinfos_by_index, contains_key_str, create_bitrot_writer, debug,
delete_file_info_version_id, disk, ensure_delete_commit_locks_held, error, explicit_delete_removed_marker,
finish_set_disk_read_lock, get_codec_streaming_reader_gate, get_object_body_cache_hook, get_raw_etag,
get_small_object_direct_memory_decision, get_stage_timer_if_enabled, get_str,
get_transitioned_object_reader_with_tier_manager, inline_erasure_shard_file_offset, inline_erasure_shard_size, insert_str,
is_deadlock_detection_enabled, is_err_object_not_found, is_err_version_not_found, is_explicit_null_version,
is_get_codec_streaming_base_enabled, is_lock_optimization_enabled, issue3031_diag_enabled, join_all,
known_put_object_storage_size, path_join_buf, put_restore_opts, record_compression_total_memory,
record_get_codec_streaming_gate_decision, record_get_direct_memory_decision, record_get_object_pipeline_failure,
record_get_object_pipeline_failure_for_path, record_get_object_reader_path_observation, record_get_stage_duration_if_enabled,
record_lock_acquire, reduce_write_quorum_errs, release_materialized_read_lock, replication_write_may_pass_worm_gate,
require_restore_operation_id, resolve_delete_version_state, resolve_tiered_decommission_write_quorum_result,
resolve_write_layout, restore_commit_operation_id_from_metadata, restore_operation_id_from_metadata, send_event,
GET_STAGE_READER_SETUP, GenericError, GetCodecStreamingDecision, GetCodecStreamingFallbackReason, GetDirectMemoryDecision,
GetObjectReader, HTTPRangeSpec, HashAlgorithm, HashMap, HashReader, HashSet, HeaderMap, HealChannelPriority, InstanceContext,
Instant, LOG_COMPONENT_ECSTORE, LOG_SUBSYSTEM_SET_DISK, OBJECT_OP_IGNORED_ERRS, ObjectApiError, ObjectInfo, ObjectKey,
ObjectLockConfigSnapshot, ObjectLockConfigState, ObjectOptions, ObjectReader, ObjectToDelete, OffsetDateTime, Ordering, Pin,
PutObjReader, RUSTFS_META_BUCKET, RUSTFS_META_TMP_BUCKET, ReadPathPlan, ReaderImpl, ReplicateDecision,
ReplicationObjectBridge, Result, SET_DISK_COMMIT_TAIL_WARN_THRESHOLD_MS, SLASH_SEPARATOR, SUFFIX_ACTUAL_SIZE,
SUFFIX_COMPRESSION, SUFFIX_COMPRESSION_SIZE, SUFFIX_RESTORE_OPERATION_ID, SetDisks, SmallWritePath, StorageError,
TRANSITION_COMPLETE, UpdateMetadataOpts, Uuid, WriteLayout, X_AMZ_OBJECT_LOCK_LEGAL_HOLD, X_AMZ_OBJECT_LOCK_MODE,
X_AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE, X_AMZ_RESTORE, adaptive_duplex_buffer_size, build_get_object_info,
build_inline_bitrot_readers, build_inline_bitrot_readers_from_refs, can_try_inline_data_shards_direct,
check_object_lock_delete, check_object_lock_for_deletion_with_state, check_object_lock_retention_update,
classify_get_codec_streaming_object_class, classify_put_write_path, classify_storage_error,
collect_inline_data_shard_fileinfos_by_index, contains_key_str, create_bitrot_writer, debug, delete_file_info_version_id,
disk, ensure_delete_commit_locks_held, error, explicit_delete_removed_marker, finish_set_disk_read_lock,
get_codec_streaming_reader_gate_with_plan, get_object_body_cache_hook, get_raw_etag,
get_small_object_direct_memory_decision_with_threshold_and_plan, get_small_object_direct_memory_threshold,
get_stage_timer_if_enabled, get_str, get_transitioned_object_reader_with_tier_manager, inline_erasure_shard_file_offset,
inline_erasure_shard_size, insert_str, is_deadlock_detection_enabled, is_err_object_not_found, is_err_version_not_found,
is_explicit_null_version, is_get_codec_streaming_base_enabled, is_get_small_object_direct_memory_enabled,
is_lock_optimization_enabled, issue3031_diag_enabled, join_all, known_put_object_storage_size, path_join_buf,
put_restore_opts, record_compression_total_memory, record_get_codec_streaming_gate_decision,
record_get_direct_memory_decision, record_get_object_pipeline_failure, record_get_object_pipeline_failure_for_path,
record_get_object_reader_path_observation, record_get_stage_duration_if_enabled, record_lock_acquire,
reduce_write_quorum_errs, release_materialized_read_lock, replication_write_may_pass_worm_gate, require_restore_operation_id,
resolve_delete_version_state, resolve_tiered_decommission_write_quorum_result, resolve_write_layout,
restore_commit_operation_id_from_metadata, restore_operation_id_from_metadata, send_event,
set_disk_delete_creates_delete_marker, should_force_delete_marker_for_missing_version,
should_persist_encryption_original_size, should_preserve_delete_replication_state, should_use_inline_fast_path,
should_persist_encryption_original_size, should_preserve_delete_replication_state, should_use_inline_fast_path_with_plan,
take_prepared_get_object_metadata, to_object_err, try_read_inline_data_shards_direct, warn,
};
use super::bitrot_self_verify::{BitrotSelfVerifyTarget, drop_failed_writer_disks, verify_written_bitrot_shards};
@@ -109,6 +111,7 @@ fn is_get_mid_size_streaming_enabled() -> bool {
/// codec-streaming gate: only a whole, plain, single-part object is eligible.
/// Ranges, transforms, remote objects, multipart reads, copy-source reads and
/// special movement/version requests retain their existing legacy semantics.
#[allow(dead_code, reason = "asserted by this file's gate tests")]
fn get_mid_size_streaming_object_size(
range: &Option<HTTPRangeSpec>,
object_info: &ObjectInfo,
@@ -127,6 +130,7 @@ fn get_mid_size_streaming_object_size(
)
}
#[allow(dead_code, reason = "asserted by this file's gate tests")]
fn get_mid_size_streaming_object_size_with_flags(
range: &Option<HTTPRangeSpec>,
object_info: &ObjectInfo,
@@ -135,6 +139,26 @@ fn get_mid_size_streaming_object_size_with_flags(
lock_optimization_enabled: bool,
mid_size_enabled: bool,
codec_base_enabled: bool,
) -> Option<usize> {
get_mid_size_streaming_object_size_with_flags_and_plan(
range,
object_info,
opts,
lock_optimization_enabled,
mid_size_enabled,
codec_base_enabled,
super::super::ReadPathPlan::new(object_info, fi),
)
}
fn get_mid_size_streaming_object_size_with_flags_and_plan(
range: &Option<HTTPRangeSpec>,
object_info: &ObjectInfo,
opts: &ObjectOptions,
lock_optimization_enabled: bool,
mid_size_enabled: bool,
codec_base_enabled: bool,
plan: super::super::ReadPathPlan,
) -> Option<usize> {
if !mid_size_enabled
|| !codec_base_enabled
@@ -149,26 +173,18 @@ fn get_mid_size_streaming_object_size_with_flags(
|| object_info.delete_marker
|| object_info.metadata_only
|| object_info.version_only
|| object_info.is_encrypted()
|| object_info.is_compressed()
|| object_info.is_remote()
|| crate::set_disk::get_object_read_policy() != super::super::GetObjectReadPolicy::Default
|| object_info.parts.len() != 1
|| fi.parts.len() != 1
|| object_info.size != fi.size
{
return None;
}
let object_size = usize::try_from(fi.size).ok()?;
let object_part = object_info.parts.first()?;
let file_part = fi.parts.first()?;
if object_part.number != file_part.number || file_part.size != object_size || file_part.actual_size != fi.size {
let shape = plan.shape()?;
if !plan.is_plain() {
return None;
}
(GET_MID_SIZE_STREAMING_MIN_SIZE..=GET_MID_SIZE_STREAMING_MAX_SIZE)
.contains(&object_size)
.then_some(object_size)
.contains(&shape.object_size)
.then_some(shape.object_size)
}
#[cfg(all(test, feature = "test-util"))]
@@ -1721,6 +1737,11 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
return Ok(reader);
}
// All remaining local fast paths share this immutable, fail-closed
// metadata decision. Build it once after empty/remote exits so those
// requests do not pay for part and transform scans they cannot use.
let read_path_plan = ReadPathPlan::new(&object_info, fi);
// Inline data fast path: skip duplex pipe for small inline objects.
// Uses the shared predicate from ObjectInfo; additionally checks that
// inline data is actually present and neither range nor partNumber is
@@ -1729,7 +1750,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
&& fi.data.is_some()
&& range.is_none()
&& opts.part_number.is_none()
&& should_use_inline_fast_path(&range, &object_info, fi, opts);
&& should_use_inline_fast_path_with_plan(&range, &object_info, fi, opts, read_path_plan);
if use_inline_fast_path {
let mut inline_prepare_stage_start = get_stage_timer_if_enabled(stage_metrics_enabled);
let data_shards = fi.erasure.data_blocks;
@@ -1892,27 +1913,10 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
}
}
let path_decision_stage_start = get_stage_timer_if_enabled(stage_metrics_enabled);
let codec_streaming_gate = get_codec_streaming_reader_gate(
bucket,
object,
opts.part_number,
object_class,
&object_info,
fi,
lock_optimization_enabled,
);
record_get_stage_duration_if_enabled(GET_OBJECT_PATH_SET_DISK, GET_STAGE_PATH_DECISION, path_decision_stage_start);
if object_info.is_remote() {
if let GetCodecStreamingDecision::Fallback(reason) = codec_streaming_gate.decision {
record_get_codec_streaming_gate_decision(
codec_streaming_gate.object_class,
codec_streaming_gate.decision,
size_bucket,
);
rustfs_io_metrics::record_get_object_codec_streaming_fallback(reason.as_str());
}
let decision = GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Remote);
record_get_codec_streaming_gate_decision(object_class, decision, size_bucket);
rustfs_io_metrics::record_get_object_codec_streaming_fallback(GetCodecStreamingFallbackReason::Remote.as_str());
record_get_object_reader_path_observation(GET_OBJECT_PATH_REMOTE_TRANSITION, object_class, size_bucket);
let mut opts = opts.clone();
if object_info.parts.len() == 1 {
@@ -1932,6 +1936,10 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
return Ok(finish_set_disk_read_lock(gr, read_lock_guard.take(), bucket, object));
}
// Metadata resolution and the remote-tier branch are complete here.
// Keep the rollout/configuration gate deferred until the request
// really needs codec streaming so an opted-out codec path cannot add
// fixed cost to the inline/direct-memory/mid-size hot paths.
// App-layer object data cache probe: metadata (etag/size) is resolved
// but no data shards have been read yet, so a hit skips the erasure
// read, bitrot verify and decode entirely. The hook validates object
@@ -1978,7 +1986,15 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
}
}
let direct_memory_decision = get_small_object_direct_memory_decision(&range, &object_info, fi, opts);
let direct_memory_decision = get_small_object_direct_memory_decision_with_threshold_and_plan(
&range,
&object_info,
fi,
opts,
is_get_small_object_direct_memory_enabled(),
get_small_object_direct_memory_threshold(),
read_path_plan,
);
record_get_direct_memory_decision(object_class, direct_memory_decision, size_bucket);
if let GetDirectMemoryDecision::Use { object_size } = direct_memory_decision {
if let Some(body) = Self::try_get_object_direct_data_shards_with_fileinfo(
@@ -2064,7 +2080,17 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
// rollout remains off by default because its worker overhead is not a
// win for tiny objects. A failed setup degrades to the existing codec
// gate/legacy path before any response bytes are returned.
if get_mid_size_streaming_object_size(&range, &object_info, fi, opts, lock_optimization_enabled).is_some() {
if get_mid_size_streaming_object_size_with_flags_and_plan(
&range,
&object_info,
opts,
lock_optimization_enabled,
is_get_mid_size_streaming_enabled(),
is_get_codec_streaming_base_enabled(),
read_path_plan,
)
.is_some()
{
match Self::get_object_mid_size_reader_with_fileinfo(
bucket,
object,
@@ -2096,6 +2122,19 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
}
}
let path_decision_stage_start = get_stage_timer_if_enabled(stage_metrics_enabled);
let codec_streaming_gate = get_codec_streaming_reader_gate_with_plan(
bucket,
object,
opts.part_number,
object_class,
&object_info,
fi,
lock_optimization_enabled,
read_path_plan,
);
record_get_stage_duration_if_enabled(GET_OBJECT_PATH_SET_DISK, GET_STAGE_PATH_DECISION, path_decision_stage_start);
match codec_streaming_gate.decision {
GetCodecStreamingDecision::Use => {
match Self::get_object_decode_reader_with_fileinfo(
@@ -9639,7 +9678,7 @@ mod replication_lww_tests {
mod inline_put_commit_path_tests {
use super::hermetic_set_disks_support::hermetic_set_disks_isolated as hermetic_set_disks;
use super::*;
use crate::config::storageclass::lookup_config_for_pools_without_env;
use crate::config::storageclass::{INLINE_BLOCK_ENV, lookup_config_for_pools, lookup_config_for_pools_without_env};
use crate::disk::ReadOptions;
use rustfs_config::server_config::KVS;
use serial_test::serial;
@@ -9706,6 +9745,62 @@ mod inline_put_commit_path_tests {
assert_eq!(restored, payload);
}
#[tokio::test]
#[serial]
async fn get_object_reader_wires_mid_size_to_single_inflight() {
let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
let bucket = "mid-size-reader-wiring";
let object = "object.bin";
let payload: Vec<u8> = (0..256 * 1024).map(|index| (index % 251) as u8).collect();
make_bucket(&disk_stores, bucket).await;
let storage_class = temp_env::with_var(INLINE_BLOCK_ENV, Some("1KiB"), || lookup_config_for_pools(&KVS::new(), &[4]))
.expect("test storage class should resolve");
set_disks.set_test_storage_class_config(storage_class);
let mut writer = PutObjReader::from_vec(payload.clone());
temp_env::async_with_vars(
[
(ENV_RUSTFS_GET_MID_SIZE_STREAMING_ENABLE, Some("true")),
(crate::set_disk::ENV_RUSTFS_GET_CODEC_STREAMING_ENABLE, Some("true")),
(crate::set_disk::ENV_RUSTFS_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED, Some("true")),
(crate::set_disk::ENV_RUSTFS_GET_CODEC_STREAMING_HEADER_COMPAT_CONFIRMED, Some("true")),
(crate::set_disk::ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT, Some("on")),
(rustfs_config::ENV_OBJECT_LOCK_OPTIMIZATION_ENABLE, Some("true")),
],
async {
set_disks
.put_object(bucket, object, &mut writer, &ObjectOptions::default())
.await
.expect("mid-size wiring fixture should commit");
crate::set_disk::reset_test_get_object_reader_path();
let single_inflight_before = crate::set_disk::coding::decode_reader::test_single_inflight_construction_count();
let mut reader = set_disks
.get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default())
.await
.expect("mid-size wiring GET should succeed");
let mut restored = Vec::new();
reader
.stream
.read_to_end(&mut restored)
.await
.expect("mid-size wiring reader should stream");
assert_eq!(restored, payload);
assert!(
crate::set_disk::test_get_object_reader_selected_mid_size(),
"full get_object_reader path must select mid-size streaming (path id {})",
crate::set_disk::test_get_object_reader_path_id()
);
assert!(
crate::set_disk::coding::decode_reader::test_single_inflight_construction_count() > single_inflight_before,
"mid-size get_object_reader wiring must construct SingleInFlight"
);
},
)
.await;
}
#[tokio::test]
async fn repeated_gets_reuse_the_set_erasure_shell() {
let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
crates/ecstore/src/set_disk/read.rs
+26 -5
View File
@@ -1370,6 +1370,7 @@ impl SetDisks {
prefer_data_blocks_first_reader_setup,
get_codec_streaming_metrics_path(),
false,
false,
)
.await
}
@@ -1408,6 +1409,7 @@ impl SetDisks {
prefer_data_blocks_first_reader_setup,
GET_OBJECT_PATH_MID_SIZE_STREAMING,
true,
true,
)
.await
}
@@ -1428,6 +1430,7 @@ impl SetDisks {
prefer_data_blocks_first_reader_setup: bool,
metrics_path: &'static str,
allow_inplace_legacy_fallback: bool,
single_inflight: bool,
) -> Result<GetCodecStreamingReaderBuildOutcome> {
let erasure = erasure_cache.get_for_file_info(fi)?;
let (disks, files) = Self::shuffle_disks_and_parts_metadata_by_index(disks, files, fi);
@@ -1451,6 +1454,7 @@ impl SetDisks {
metrics_object_class,
metrics_size_bucket,
prefer_data_blocks_first_reader_setup,
single_inflight,
// Single-part objects keep the whole-request fallback: a degraded
// sole part is detected before any byte streams, so the caller can
// still hand the request to the legacy duplex path unchanged.
@@ -1504,6 +1508,7 @@ impl SetDisks {
metrics_object_class,
metrics_size_bucket,
false,
false,
// The first part stays eager and keeps the whole-request fallback:
// if part 1 is already degraded, the entire GET drops to the legacy
// duplex path before a single byte is streamed (semantics unchanged).
@@ -1551,6 +1556,7 @@ impl SetDisks {
ctx.metrics_object_class,
ctx.metrics_size_bucket,
false,
false,
// backlog#879: later parts have already streamed earlier bytes,
// so a whole-request fallback is impossible here. Degrade this
// part in place to a legacy per-part decode reader instead of
@@ -1584,6 +1590,7 @@ impl SetDisks {
metrics_object_class: &'static str,
metrics_size_bucket: &'static str,
prefer_data_blocks_first_reader_setup: bool,
single_inflight: bool,
allow_inplace_legacy_fallback: bool,
metrics_path: &'static str,
) -> Result<GetCodecStreamingReaderBuildOutcome> {
@@ -1704,11 +1711,23 @@ impl SetDisks {
.with_deferred_parity_handles(deferred_stripe_handles)
.with_deferred_parity_reopeners(deferred_reopeners);
let engine = build_get_codec_streaming_decode_engine(erasure.clone())?;
let reader =
coding::decode_reader::ErasureDecodeReader::new_with_metrics_path(source, engine, part_length, metrics_path)?;
Ok(GetCodecStreamingReaderBuildOutcome::Reader(Box::new(
coding::decode_reader::SyncErasureDecodeReader::new_with_metrics_path(reader, metrics_path),
)))
let reader = if single_inflight {
coding::decode_reader::ErasureDecodeReader::new_single_inflight_with_metrics_path(
source,
engine,
part_length,
metrics_path,
)?
} else {
coding::decode_reader::ErasureDecodeReader::new_with_metrics_path(source, engine, part_length, metrics_path)?
};
if single_inflight {
Ok(GetCodecStreamingReaderBuildOutcome::Reader(Box::new(reader)))
} else {
Ok(GetCodecStreamingReaderBuildOutcome::Reader(Box::new(
coding::decode_reader::SyncErasureDecodeReader::new_with_metrics_path(reader, metrics_path),
)))
}
}
}
@@ -4876,6 +4895,7 @@ mod tests {
"test-size-bucket",
false,
false,
false,
get_codec_streaming_metrics_path(),
)
.await;
@@ -4897,6 +4917,7 @@ mod tests {
"test-size-bucket",
false,
false,
false,
get_codec_streaming_metrics_path(),
)
.await;