From 07212c4e262a8c32e46c68e950e76827fce448f2 Mon Sep 17 00:00:00 2001 From: houseme Date: Sun, 30 Aug 2026 17:33:35 +0800 Subject: [PATCH] perf(ecstore): gate quorum-aware GET early stop (#6885) * perf(ecstore): add gated two-phase GET metadata reads Co-Authored-By: heihutu * fix(ecstore): require data-shard coverage for read plans Co-Authored-By: heihutu * perf(ecstore): avoid inline overhead in read plan rollout Co-Authored-By: heihutu * perf(ecstore): accept quorum-complete read candidates Co-Authored-By: heihutu --------- Co-authored-by: heihutu --- .../src/set_disk/core/io_primitives.rs | 63 ++++++++- crates/ecstore/src/set_disk/mod.rs | 130 ++++++++++++++++++ crates/ecstore/src/set_disk/read.rs | 17 +++ 3 files changed, 206 insertions(+), 4 deletions(-) diff --git a/crates/ecstore/src/set_disk/core/io_primitives.rs b/crates/ecstore/src/set_disk/core/io_primitives.rs index 37fba0dca..251bbc6eb 100644 --- a/crates/ecstore/src/set_disk/core/io_primitives.rs +++ b/crates/ecstore/src/set_disk/core/io_primitives.rs @@ -49,10 +49,11 @@ use super::super::{ capacity_scope_from_disks, coding, collect_inline_data_shard_fileinfos_by_index_or_reason, current_dirty_generation, debug, disk, file_info_is_valid_for_metadata, get_metadata_slowtail_fault_request, info, inline_erasure_shard_file_offset, inline_erasure_shard_size, is_err_object_not_found, is_err_version_not_found, is_get_metadata_data_read_early_stop_enabled, - is_get_metadata_early_stop_bounded_fanout_enabled, is_get_metadata_early_stop_enabled, is_object_dangling, - is_version_early_stop_enabled, issue3031_diag_enabled, join_all, join_errs, log_multipart_write_quorum_failure, - merge_file_meta_versions, path_join_buf, record_global_dirty_scope, reduce_read_quorum_errs, reduce_write_quorum_errs, - send_heal_request_with_admission, should_prevent_write, to_object_err, try_read_inline_data_shards_direct, warn, + is_get_metadata_early_stop_bounded_fanout_enabled, is_get_metadata_early_stop_enabled, + is_get_metadata_two_phase_read_plan_enabled, is_object_dangling, is_version_early_stop_enabled, issue3031_diag_enabled, + join_all, join_errs, log_multipart_write_quorum_failure, merge_file_meta_versions, path_join_buf, record_global_dirty_scope, + reduce_read_quorum_errs, reduce_write_quorum_errs, send_heal_request_with_admission, should_prevent_write, to_object_err, + try_read_inline_data_shards_direct, warn, }; #[cfg(test)] use crate::bucket::lifecycle::lifecycle::TRANSITION_COMPLETE; @@ -1083,6 +1084,50 @@ fn data_read_early_stop_inline_candidate_miss_reason(candidate: &FileInfo) -> Op None } +fn non_inline_data_read_candidate_is_safe( + candidate: &FileInfo, + parts_metadata: &[FileInfo], + disks: &[Option], +) -> bool { + if candidate.inline_data() + || candidate.is_compressed() + || candidate.is_remote() + || candidate + .metadata + .keys() + .any(|key| rustfs_utils::http::is_object_encryption_marker(key)) + || candidate.parts.len() != 1 + { + return false; + } + let Ok(erasure) = coding::Erasure::try_new_with_options( + candidate.erasure.data_blocks, + candidate.erasure.parity_blocks, + candidate.erasure.block_size, + candidate.uses_legacy_checksum, + ) else { + return false; + }; + // The regular reader setup can reconstruct missing data shards from any + // `data_shards` matching metadata entries. Requiring every data slot here + // would unnecessarily wait for one slow data disk even when parity and + // the remaining data shards already form a read quorum. + let mut available_shards = vec![false; erasure.data_shards + erasure.parity_shards]; + for ((file_info, disk), &erasure_index) in parts_metadata + .iter() + .zip(disks.iter()) + .zip(candidate.erasure.distribution.iter()) + { + if erasure_index == 0 || erasure_index > available_shards.len() || disk.is_none() { + continue; + } + if metadata_early_stop_candidate_matches(file_info, candidate) && file_info.erasure.index == erasure_index { + available_shards[erasure_index - 1] = true; + } + } + available_shards.into_iter().filter(|present| *present).count() >= erasure.data_shards +} + fn data_read_inline_missing_shards_are_pending( candidate: &FileInfo, parts_metadata: &[FileInfo], @@ -2869,6 +2914,7 @@ impl SetDisks { read_data, healing, incl_free_versions, + read_data && is_get_metadata_two_phase_read_plan_enabled(), default_parity_count, allow_coalescing, ) @@ -3008,6 +3054,7 @@ impl SetDisks { read_data: bool, healing: bool, incl_free_versions: bool, + allow_non_inline_data_read_early_stop: bool, default_parity_count: usize, allow_coalescing: bool, ) -> disk::error::Result<(Vec, Vec>, MetadataFanoutDiagnostics)> { @@ -3096,6 +3143,8 @@ impl SetDisks { && read_data && !force_full_wait && let Some(reason) = data_read_early_stop_inline_candidate_miss_reason(&file_info) + && !(allow_non_inline_data_read_early_stop + && reason == GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_NOT_INLINE) { force_full_wait = true; final_miss_reason_override.get_or_insert(reason); @@ -3126,6 +3175,12 @@ impl SetDisks { { let should_return_early = if read_data { match accumulator.candidate.as_ref() { + Some(candidate) + if allow_non_inline_data_read_early_stop + && non_inline_data_read_candidate_is_safe(candidate, &ress, disks) => + { + true + } Some(candidate) => match data_read_early_stop_inline_body_miss_reason( bucket.as_ref(), object.as_ref(), diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 88ebc3ac9..4b67f6c80 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -773,6 +773,13 @@ const DEFAULT_RUSTFS_GET_METADATA_VERSION_EARLY_STOP_ENABLE: bool = false; const ENV_RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE: &str = "RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE"; const DEFAULT_RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE: bool = true; +// Two-phase metadata/read-plan rollout (backlog#1309). The first phase reads +// metadata without inline payloads and only fetches inline data from the +// selected data-shard slots. Keep this opt-in until the Linux multi-node +// slow-tail and small-inline cost gates are complete. +const ENV_RUSTFS_GET_METADATA_TWO_PHASE_READ_PLAN_ENABLE: &str = "RUSTFS_GET_METADATA_TWO_PHASE_READ_PLAN_ENABLE"; +const DEFAULT_RUSTFS_GET_METADATA_TWO_PHASE_READ_PLAN_ENABLE: bool = false; + const ENV_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT: &str = "RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT"; const DEFAULT_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT: bool = true; @@ -1168,6 +1175,109 @@ mod prepared_get_object_metadata_tests { ); } + #[tokio::test] + #[serial_test::serial(body_cache_hook)] + async fn two_phase_read_plan_uses_metadata_only_for_non_inline_get() { + let (_dirs, set_disks) = make_local_set_disks(4, 2).await; + let bucket = "two-phase-read-plan"; + let object = object_with_initial_data_shards(bucket, "non-inline-object"); + let payload = vec![0x5a; 2 * 1024 * 1024]; + let opts = ObjectOptions { + no_lock: true, + ..Default::default() + }; + + set_disks + .make_bucket(bucket, &MakeBucketOptions::default()) + .await + .expect("bucket should be created"); + let mut put_reader = PutObjReader::from_vec(payload.clone()); + set_disks + .put_object(bucket, &object, &mut put_reader, &opts) + .await + .expect("object should be written"); + + temp_env::async_with_vars( + [ + ("RUSTFS_GET_METADATA_TWO_PHASE_READ_PLAN_ENABLE", Some("true")), + ("RUSTFS_GET_METADATA_EARLY_STOP_ENABLE", Some("true")), + ("RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT", Some("true")), + ], + async { + let calls = disk_call_counters::observe(&object); + reset_test_get_object_reader_path(); + let mut reader = set_disks + .get_object_reader(bucket, &object, None, HeaderMap::new(), &opts) + .await + .expect("two-phase GET reader should open"); + let mut restored = Vec::new(); + reader + .stream + .read_to_end(&mut restored) + .await + .expect("two-phase GET body should stream"); + assert_eq!(restored, payload); + assert!( + calls.total(disk_call_counters::KIND_READ_VERSION) < 4, + "non-inline two-phase GET should stop metadata fanout at a quorum" + ); + }, + ) + .await; + } + + #[tokio::test] + #[serial_test::serial(body_cache_hook)] + async fn two_phase_read_plan_preserves_inline_early_stop_path() { + let (_dirs, set_disks) = make_local_set_disks(4, 2).await; + let bucket = "two-phase-read-plan-inline"; + let object = object_with_initial_data_shards(bucket, "inline-object"); + let payload = b"two-phase inline payload".repeat(256); + let opts = ObjectOptions { + no_lock: true, + ..Default::default() + }; + + set_disks + .make_bucket(bucket, &MakeBucketOptions::default()) + .await + .expect("bucket should be created"); + let mut put_reader = PutObjReader::from_vec(payload.clone()); + set_disks + .put_object(bucket, &object, &mut put_reader, &opts) + .await + .expect("inline object should be written"); + + temp_env::async_with_vars( + [ + ("RUSTFS_GET_METADATA_TWO_PHASE_READ_PLAN_ENABLE", Some("true")), + ("RUSTFS_GET_METADATA_EARLY_STOP_ENABLE", Some("true")), + ("RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT", Some("true")), + ], + async { + let calls = disk_call_counters::observe(&object); + let mut reader = set_disks + .get_object_reader(bucket, &object, None, HeaderMap::new(), &opts) + .await + .expect("two-phase inline GET reader should open"); + let mut restored = Vec::new(); + reader + .stream + .read_to_end(&mut restored) + .await + .expect("two-phase inline GET body should stream"); + assert_eq!(restored, payload); + assert_eq!( + test_get_object_reader_path_id(), + 3, + "inline GET should retain the direct inline reader path" + ); + assert!(calls.total(disk_call_counters::KIND_READ_VERSION) <= 4); + }, + ) + .await; + } + #[test] #[serial_test::serial(body_cache_hook)] fn inline_data_read_early_stop_defaults_return_exact_body() { @@ -1914,6 +2024,26 @@ fn is_get_metadata_data_read_early_stop_enabled() -> bool { } } +fn is_get_metadata_two_phase_read_plan_enabled() -> bool { + #[cfg(test)] + { + rustfs_utils::get_env_bool( + ENV_RUSTFS_GET_METADATA_TWO_PHASE_READ_PLAN_ENABLE, + DEFAULT_RUSTFS_GET_METADATA_TWO_PHASE_READ_PLAN_ENABLE, + ) + } + #[cfg(not(test))] + { + static CACHED: OnceLock = OnceLock::new(); + *CACHED.get_or_init(|| { + rustfs_utils::get_env_bool( + ENV_RUSTFS_GET_METADATA_TWO_PHASE_READ_PLAN_ENABLE, + DEFAULT_RUSTFS_GET_METADATA_TWO_PHASE_READ_PLAN_ENABLE, + ) + }) + } +} + fn is_get_metadata_early_stop_bounded_fanout_enabled() -> bool { #[cfg(test)] { diff --git a/crates/ecstore/src/set_disk/read.rs b/crates/ecstore/src/set_disk/read.rs index 6962d2189..1c093abc3 100644 --- a/crates/ecstore/src/set_disk/read.rs +++ b/crates/ecstore/src/set_disk/read.rs @@ -89,6 +89,8 @@ use super::ENV_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT; #[cfg(test)] use super::ENV_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE; #[cfg(test)] +use super::ENV_RUSTFS_GET_METADATA_TWO_PHASE_READ_PLAN_ENABLE; +#[cfg(test)] use super::ENV_RUSTFS_GET_METADATA_VERSION_EARLY_STOP_ENABLE; #[cfg(test)] use super::ENV_RUSTFS_GET_MULTIPART_READER_SETUP_PREFETCH; @@ -126,6 +128,8 @@ use super::is_get_metadata_early_stop_bounded_fanout_enabled; #[cfg(test)] use super::is_get_metadata_early_stop_enabled; #[cfg(test)] +use super::is_get_metadata_two_phase_read_plan_enabled; +#[cfg(test)] use super::is_version_early_stop_enabled; #[cfg(test)] use super::load_get_codec_streaming_config; @@ -4267,6 +4271,19 @@ mod tests { ); } + #[test] + fn two_phase_read_plan_gate_defaults_off_and_honors_override() { + temp_env::with_var(ENV_RUSTFS_GET_METADATA_TWO_PHASE_READ_PLAN_ENABLE, None::<&str>, || { + assert!(!is_get_metadata_two_phase_read_plan_enabled()); + }); + temp_env::with_var(ENV_RUSTFS_GET_METADATA_TWO_PHASE_READ_PLAN_ENABLE, Some("true"), || { + assert!(is_get_metadata_two_phase_read_plan_enabled()); + }); + temp_env::with_var(ENV_RUSTFS_GET_METADATA_TWO_PHASE_READ_PLAN_ENABLE, Some("false"), || { + assert!(!is_get_metadata_two_phase_read_plan_enabled()); + }); + } + #[test] fn metadata_early_stop_rejects_healing_and_free_version_requests() { temp_env::with_vars(