From 24ca61eb6ed793f6c0a62faad9860d943339a169 Mon Sep 17 00:00:00 2001 From: houseme Date: Thu, 13 Aug 2026 00:04:15 +0800 Subject: [PATCH] perf(get): include small objects in codec streaming (#6004) Co-authored-by: heihutu --- .../src/get_codec_streaming_compat_test.rs | 9 ++- crates/ecstore/src/set_disk/mod.rs | 6 +- crates/ecstore/src/set_disk/read.rs | 75 ++++++++++--------- 3 files changed, 50 insertions(+), 40 deletions(-) diff --git a/crates/e2e_test/src/get_codec_streaming_compat_test.rs b/crates/e2e_test/src/get_codec_streaming_compat_test.rs index 9d7b4ceee..eb758aa7f 100644 --- a/crates/e2e_test/src/get_codec_streaming_compat_test.rs +++ b/crates/e2e_test/src/get_codec_streaming_compat_test.rs @@ -189,8 +189,6 @@ mod tests { ("RUSTFS_GET_CODEC_STREAMING_ROLLOUT_PCT", "100"), ("RUSTFS_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED", "true"), ("RUSTFS_GET_CODEC_STREAMING_HEADER_COMPAT_CONFIRMED", "true"), - // Lower the min-size floor so every non-inline object below is eligible. - ("RUSTFS_GET_CODEC_STREAMING_MIN_SIZE", "4096"), // Route multipart objects through per-part codec streaming too. ("RUSTFS_GET_CODEC_STREAMING_MULTIPART_ENABLE", "true"), // Lock optimization is on by default, but pin it so the gate's @@ -315,6 +313,13 @@ mod tests { }, payload(64 * 1024, 2), ), + ( + Shape { + key: "small-non-inline-256kib-plus", + expect_large: true, + }, + payload(256 * 1024 + 1, 6), + ), ( Shape { key: "mid-1_5mib", diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 95b729640..a360b9e0c 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -639,9 +639,11 @@ const ENV_RUSTFS_GET_CODEC_STREAMING_ENABLE: &str = "RUSTFS_GET_CODEC_STREAMING_ const DEFAULT_RUSTFS_GET_CODEC_STREAMING_ENABLE: bool = true; const ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE: &str = "RUSTFS_GET_CODEC_STREAMING_MIN_SIZE"; -const DEFAULT_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE: usize = MI_B; +// Meet the direct-memory path at its default ceiling. Codec streaming remains +// rollout-gated and starts where the eager small-object path ends. +const DEFAULT_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE: usize = DEFAULT_RUSTFS_GET_SMALL_OBJECT_DIRECT_MEMORY_THRESHOLD; const ENV_RUSTFS_GET_CODEC_STREAMING_RUSTFS_MIN_SIZE: &str = "RUSTFS_GET_CODEC_STREAMING_RUSTFS_MIN_SIZE"; -const DEFAULT_RUSTFS_GET_CODEC_STREAMING_RUSTFS_MIN_SIZE: usize = MI_B; +const DEFAULT_RUSTFS_GET_CODEC_STREAMING_RUSTFS_MIN_SIZE: usize = DEFAULT_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE; const ENV_RUSTFS_GET_CODEC_STREAMING_ENGINE: &str = "RUSTFS_GET_CODEC_STREAMING_ENGINE"; const DEFAULT_RUSTFS_GET_CODEC_STREAMING_ENGINE: &str = GET_CODEC_STREAMING_ENGINE_LEGACY; diff --git a/crates/ecstore/src/set_disk/read.rs b/crates/ecstore/src/set_disk/read.rs index 9fcc41ba3..39bd74cd2 100644 --- a/crates/ecstore/src/set_disk/read.rs +++ b/crates/ecstore/src/set_disk/read.rs @@ -5479,33 +5479,36 @@ mod tests { } #[test] - fn rustfs_codec_streaming_uses_conservative_default_min_size() { - temp_env::with_vars( - [ - (ENV_RUSTFS_GET_CODEC_STREAMING_ENABLE, Some("true")), - (ENV_RUSTFS_GET_CODEC_STREAMING_ENGINE, Some(GET_CODEC_STREAMING_ENGINE_RUSTFS)), - (ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT, Some("benchmark")), - (ENV_RUSTFS_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED, Some("true")), - (ENV_RUSTFS_GET_CODEC_STREAMING_HEADER_COMPAT_CONFIRMED, Some("true")), - (ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE, None::<&str>), - (ENV_RUSTFS_GET_CODEC_STREAMING_RUSTFS_MIN_SIZE, None::<&str>), - ], - || { - let below_threshold_fi = codec_streaming_test_fileinfo(512 * 1024, 1); - let below_threshold_object_info = codec_streaming_test_object_info(&below_threshold_fi); - assert_eq!( - codec_streaming_reader_gate_for_test(&None, &below_threshold_object_info, &below_threshold_fi, true).decision, - GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::BelowMinSize) - ); + fn codec_streaming_default_min_size_meets_direct_memory_ceiling() { + for engine in [None, Some(GET_CODEC_STREAMING_ENGINE_RUSTFS)] { + temp_env::with_vars( + [ + (ENV_RUSTFS_GET_CODEC_STREAMING_ENABLE, Some("true")), + (ENV_RUSTFS_GET_CODEC_STREAMING_ENGINE, engine), + (ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT, Some("benchmark")), + (ENV_RUSTFS_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED, Some("true")), + (ENV_RUSTFS_GET_CODEC_STREAMING_HEADER_COMPAT_CONFIRMED, Some("true")), + (ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE, None::<&str>), + (ENV_RUSTFS_GET_CODEC_STREAMING_RUSTFS_MIN_SIZE, None::<&str>), + ], + || { + let below_threshold_fi = codec_streaming_test_fileinfo(128 * 1024 - 1, 1); + let below_threshold_object_info = codec_streaming_test_object_info(&below_threshold_fi); + assert_eq!( + codec_streaming_reader_gate_for_test(&None, &below_threshold_object_info, &below_threshold_fi, true) + .decision, + GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::BelowMinSize) + ); - let threshold_fi = codec_streaming_test_fileinfo(1_048_576, 1); - let threshold_object_info = codec_streaming_test_object_info(&threshold_fi); - assert_eq!( - codec_streaming_reader_gate_for_test(&None, &threshold_object_info, &threshold_fi, true).decision, - GetCodecStreamingDecision::Use - ); - }, - ); + let threshold_fi = codec_streaming_test_fileinfo(128 * 1024, 1); + let threshold_object_info = codec_streaming_test_object_info(&threshold_fi); + assert_eq!( + codec_streaming_reader_gate_for_test(&None, &threshold_object_info, &threshold_fi, true).decision, + GetCodecStreamingDecision::Use + ); + }, + ); + } } #[test] @@ -5803,10 +5806,10 @@ mod tests { (ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT, None::<&str>), (ENV_RUSTFS_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED, None::<&str>), (ENV_RUSTFS_GET_CODEC_STREAMING_HEADER_COMPAT_CONFIRMED, None::<&str>), - (ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE, Some("1")), + (ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE, None::<&str>), ], || { - let fi = codec_streaming_test_fileinfo(1024, 1); + let fi = codec_streaming_test_fileinfo(128 * 1024, 1); let object_info = codec_streaming_test_object_info(&fi); assert_eq!( @@ -5827,10 +5830,10 @@ mod tests { (ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT, Some("on")), (ENV_RUSTFS_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED, None::<&str>), (ENV_RUSTFS_GET_CODEC_STREAMING_HEADER_COMPAT_CONFIRMED, None::<&str>), - (ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE, Some("1")), + (ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE, None::<&str>), ], || { - let fi = codec_streaming_test_fileinfo(1024, 1); + let fi = codec_streaming_test_fileinfo(128 * 1024, 1); let object_info = codec_streaming_test_object_info(&fi); assert_eq!( @@ -5848,10 +5851,10 @@ mod tests { [ (ENV_RUSTFS_GET_CODEC_STREAMING_ENABLE, Some("false")), (ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT, Some("on")), - (ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE, Some("1")), + (ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE, None::<&str>), ], || { - let fi = codec_streaming_test_fileinfo(1024, 1); + let fi = codec_streaming_test_fileinfo(128 * 1024, 1); let object_info = codec_streaming_test_object_info(&fi); assert_eq!( @@ -5931,10 +5934,10 @@ mod tests { (ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT_PCT, Some("0")), (ENV_RUSTFS_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED, Some("true")), (ENV_RUSTFS_GET_CODEC_STREAMING_HEADER_COMPAT_CONFIRMED, Some("true")), - (ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE, Some("1")), + (ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE, None::<&str>), ], || { - let fi = codec_streaming_test_fileinfo(1024, 1); + let fi = codec_streaming_test_fileinfo(128 * 1024, 1); let object_info = codec_streaming_test_object_info(&fi); assert_eq!( @@ -5951,10 +5954,10 @@ mod tests { (ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT_PCT, Some("100")), (ENV_RUSTFS_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED, Some("true")), (ENV_RUSTFS_GET_CODEC_STREAMING_HEADER_COMPAT_CONFIRMED, Some("true")), - (ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE, Some("1")), + (ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE, None::<&str>), ], || { - let fi = codec_streaming_test_fileinfo(1024, 1); + let fi = codec_streaming_test_fileinfo(128 * 1024, 1); let object_info = codec_streaming_test_object_info(&fi); assert_eq!(