From 88e285c523ff0d3d97e796653816aff27089d6ce Mon Sep 17 00:00:00 2001 From: houseme Date: Mon, 10 Aug 2026 13:20:21 +0800 Subject: [PATCH] perf(ecstore): gate bounded GET metadata fanout (#5917) Co-authored-by: heihutu --- .../src/set_disk/core/io_primitives.rs | 254 ++++++++++++++++-- crates/ecstore/src/set_disk/mod.rs | 52 +++- crates/ecstore/src/set_disk/read.rs | 16 +- 3 files changed, 299 insertions(+), 23 deletions(-) diff --git a/crates/ecstore/src/set_disk/core/io_primitives.rs b/crates/ecstore/src/set_disk/core/io_primitives.rs index 1ddd04ee8..20e0924d3 100644 --- a/crates/ecstore/src/set_disk/core/io_primitives.rs +++ b/crates/ecstore/src/set_disk/core/io_primitives.rs @@ -461,6 +461,34 @@ impl MetadataQuorumAccumulator { None } + pub(in crate::set_disk) fn can_still_reach_early_stop_with_pending(&self, pending: usize) -> bool { + if !self.allow_early_stop { + return false; + } + if self.delete_marker_votes.saturating_add(pending) >= self.default_write_quorum() { + return true; + } + if self.conflicting_metadata + || self.delete_marker_seen + || self.not_found_responses > 0 + || self.version_not_found_responses > 0 + || self.hard_errors > 0 + { + return false; + } + if !self.requested_version_id.is_empty() + && self.matching_version_votes.saturating_add(pending) >= self.read_quorum_for_version() + { + return true; + } + match &self.candidate { + Some(candidate) => self + .candidate_latest_quorum(candidate) + .is_some_and(|latest_quorum| self.candidate_votes.saturating_add(pending) >= latest_quorum), + None => pending >= self.default_write_quorum(), + } + } + /// Compute the read quorum threshold for version-aware early-stop. /// Uses `total_disks / 2` (like `missing_response_quorum`) when /// `default_parity_count` is set, otherwise requires all disks. @@ -1982,7 +2010,7 @@ pub(in crate::set_disk) fn should_allow_metadata_early_stop( healing: bool, incl_free_versions: bool, ) -> bool { - if read_data { + if read_data && !is_get_metadata_data_read_early_stop_enabled() { return false; } @@ -2289,23 +2317,39 @@ impl SetDisks { let object = Arc::new(object.to_string()); let version_id = Arc::new(version_id.to_string()); let mut join_set = JoinSet::new(); + let bounded_fanout = is_get_metadata_early_stop_bounded_fanout_enabled(); + let mut next_disk_index = 0usize; + let spawn_read_version = + |join_set: &mut JoinSet<(usize, disk::error::Result, Duration)>, index: usize, disk: Option| { + let opts = opts.clone(); + let org_bucket = org_bucket.clone(); + let bucket = bucket.clone(); + let object = object.clone(); + let version_id = version_id.clone(); + join_set.spawn(async move { + let response_start = Instant::now(); + let result = if let Some(disk) = disk { + Self::record_read_version_call(&object, index); + disk.read_version(&org_bucket, &bucket, &object, &version_id, &opts).await + } else { + Err(DiskError::DiskNotFound) + }; + (index, result, response_start.elapsed()) + }); + }; - for (index, disk) in disks.iter().cloned().enumerate() { - let opts = opts.clone(); - let org_bucket = org_bucket.clone(); - let bucket = bucket.clone(); - let object = object.clone(); - let version_id = version_id.clone(); - join_set.spawn(async move { - let response_start = Instant::now(); - let result = if let Some(disk) = disk { - Self::record_read_version_call(&object, index); - disk.read_version(&org_bucket, &bucket, &object, &version_id, &opts).await - } else { - Err(DiskError::DiskNotFound) - }; - (index, result, response_start.elapsed()) - }); + if bounded_fanout { + let initial_target = accumulator.default_write_quorum().min(disks.len()); + while next_disk_index < initial_target { + if let Some(disk) = disks.get(next_disk_index).cloned() { + spawn_read_version(&mut join_set, next_disk_index, disk); + } + next_disk_index = next_disk_index.saturating_add(1); + } + } else { + for (index, disk) in disks.iter().cloned().enumerate() { + spawn_read_version(&mut join_set, index, disk); + } } while let Some(result) = join_set.join_next().await { @@ -2337,7 +2381,11 @@ impl SetDisks { .early_stop_decision() .or_else(|| accumulator.version_early_stop_decision()) { - let saved_responses = join_set.len(); + let saved_responses = if bounded_fanout { + disks.len().saturating_sub(observations.len()) + } else { + join_set.len() + }; join_set.abort_all(); rustfs_io_metrics::record_get_object_metadata_early_stop_hit(GET_OBJECT_PATH_LEGACY_DUPLEX, decision.reason); rustfs_io_metrics::record_get_object_metadata_early_stop_saved_responses( @@ -2348,6 +2396,16 @@ impl SetDisks { let diagnostics = MetadataFanoutDiagnostics::new(fanout_start.elapsed(), observations); return Ok((ress, errors, diagnostics)); } + + if bounded_fanout + && next_disk_index < disks.len() + && !accumulator.can_still_reach_early_stop_with_pending(join_set.len()) + { + if let Some(disk) = disks.get(next_disk_index).cloned() { + spawn_read_version(&mut join_set, next_disk_index, disk); + } + next_disk_index = next_disk_index.saturating_add(1); + } } rustfs_io_metrics::record_get_object_metadata_early_stop_miss( @@ -5254,6 +5312,166 @@ mod tests { drop(dirs); } + fn valid_metadata_fanout_fileinfo( + bucket: &str, + object: &str, + version_id: Uuid, + data_dir: Uuid, + mod_time: OffsetDateTime, + ) -> FileInfo { + let mut fi = FileInfo::new(object, 2, 2); + fi.volume = bucket.to_string(); + fi.name = object.to_string(); + fi.size = 1; + fi.erasure.index = 1; + fi.version_id = Some(version_id); + fi.is_latest = true; + fi.data_dir = Some(data_dir); + fi.mod_time = Some(mod_time); + fi.metadata.insert("etag".to_string(), "etag-1".to_string()); + fi.add_object_part(1, "part-etag".to_string(), 1, fi.mod_time, 1, None, None); + fi + } + + async fn install_metadata_fanout_fileinfo( + disks: &[Option], + bucket: &str, + object: &str, + missing_part_disk: Option, + ) { + let version_id = Uuid::new_v4(); + let data_dir = Uuid::new_v4(); + let mod_time = OffsetDateTime::now_utc(); + for (index, disk) in disks + .iter() + .enumerate() + .filter_map(|(index, disk)| disk.as_ref().map(|disk| (index, disk))) + { + if missing_part_disk != Some(index) { + disk.write_all(bucket, &format!("{object}/{data_dir}/part.1"), Bytes::from_static(b"x")) + .await + .expect("part data should be installed on every disk"); + } + disk.write_metadata( + bucket, + bucket, + object, + valid_metadata_fanout_fileinfo(bucket, object, version_id, data_dir, mod_time), + ) + .await + .expect("metadata should be installed on every disk"); + } + } + + #[tokio::test] + async fn bounded_metadata_early_stop_ab_limits_data_get_read_version_fanout() { + const DISKS: usize = 4; + let bucket = "bounded-data-get-fanout-bucket"; + let control_object = "bounded-data-get-control-object"; + let treatment_object = "bounded-data-get-treatment-object"; + let (dirs, disks) = call_counter_local_disks(bucket, DISKS).await; + install_metadata_fanout_fileinfo(&disks, bucket, control_object, None).await; + install_metadata_fanout_fileinfo(&disks, bucket, treatment_object, None).await; + + temp_env::async_with_vars( + [ + ("RUSTFS_GET_METADATA_EARLY_STOP_ENABLE", Some("true")), + ("RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE", None), + ("RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT", Some("true")), + ], + async { + let calls = disk_call_counters::observe(control_object); + let (_, _, diagnostics) = + SetDisks::read_all_fileinfo_observed(&disks, bucket, bucket, control_object, "", true, false, false, true, 2) + .await + .expect("control metadata should resolve"); + + assert_eq!( + calls.total(disk_call_counters::KIND_READ_VERSION), + DISKS as u64, + "control path should keep the default data-read full fanout" + ); + assert_eq!(diagnostics.total_responses(), DISKS); + }, + ) + .await; + + temp_env::async_with_vars( + [ + ("RUSTFS_GET_METADATA_EARLY_STOP_ENABLE", Some("true")), + ("RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE", Some("true")), + ("RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT", Some("true")), + ], + async { + let calls = disk_call_counters::observe(treatment_object); + let (parts_metadata, errs, diagnostics) = SetDisks::read_all_fileinfo_observed( + &disks, + bucket, + bucket, + treatment_object, + "", + true, + false, + false, + true, + 2, + ) + .await + .expect("healthy object metadata should reach early-stop quorum"); + + assert_eq!( + calls.total(disk_call_counters::KIND_READ_VERSION), + 3, + "treatment path should stop after the 2+2 read/write quorum instead of issuing every disk read" + ); + assert_eq!(diagnostics.total_responses(), 3); + assert_eq!(parts_metadata.iter().filter(|fi| fi.name == treatment_object).count(), 3); + assert!(errs.iter().all(Option::is_none)); + }, + ) + .await; + + drop(dirs); + } + + #[tokio::test] + async fn bounded_metadata_early_stop_falls_back_to_full_fanout_on_data_read_error() { + const DISKS: usize = 4; + let bucket = "bounded-data-get-error-bucket"; + let object = "bounded-data-get-error-object"; + let (dirs, disks) = call_counter_local_disks(bucket, DISKS).await; + install_metadata_fanout_fileinfo(&disks, bucket, object, Some(0)).await; + + temp_env::async_with_vars( + [ + ("RUSTFS_GET_METADATA_EARLY_STOP_ENABLE", Some("true")), + ("RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE", Some("true")), + ("RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT", Some("true")), + ], + async { + let calls = disk_call_counters::observe(object); + let (_, errs, diagnostics) = + SetDisks::read_all_fileinfo_observed(&disks, bucket, bucket, object, "", true, false, false, true, 2) + .await + .expect("metadata fanout should complete after falling back to all disks"); + + assert_eq!( + calls.total(disk_call_counters::KIND_READ_VERSION), + DISKS as u64, + "a data-read error must force bounded fanout to schedule every disk before returning" + ); + assert_eq!(diagnostics.total_responses(), DISKS); + assert!( + errs.iter() + .any(|err| err.as_ref().is_some_and(|err| matches!(err, DiskError::FileNotFound))) + ); + }, + ) + .await; + + drop(dirs); + } + /// Bound for the pause handshake. This is a hang-guard, not a timing /// dependency: under a working barrier `wait_until_paused` returns via the /// `Notify` handshake far below this bound regardless of IO pressure, so the diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 0a07cc1bf..0d1cea6e8 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -673,9 +673,9 @@ const DEFAULT_RUSTFS_GET_SMALL_OBJECT_DIRECT_MEMORY_THRESHOLD: usize = 128 * 102 const ENV_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE: &str = "RUSTFS_GET_METADATA_EARLY_STOP_ENABLE"; // Enabled by default (backlog#872): the early-stop path only engages for // requests `should_allow_metadata_early_stop` classifies as safe (metadata-only -// reads without version_id / healing / free-version needs) and still requires -// a full read-quorum agreement before stopping. Set the env var to `false` to -// fall back to full-wait metadata fanout. +// reads by default, without version_id / healing / free-version needs) and +// still requires a full read-quorum agreement before stopping. Set the env var +// to `false` to fall back to full-wait metadata fanout. const DEFAULT_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE: bool = true; const ENV_RUSTFS_GET_METADATA_EARLY_STOP_ROLLOUT_PCT: &str = "RUSTFS_GET_METADATA_EARLY_STOP_ROLLOUT_PCT"; @@ -684,6 +684,12 @@ const DEFAULT_RUSTFS_GET_METADATA_EARLY_STOP_ROLLOUT_PCT: u32 = 100; const ENV_RUSTFS_GET_METADATA_VERSION_EARLY_STOP_ENABLE: &str = "RUSTFS_GET_METADATA_VERSION_EARLY_STOP_ENABLE"; 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 = 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 = false; + // --- Multipart Reader-Setup Prefetch Configuration (backlog#870) --- const ENV_RUSTFS_GET_MULTIPART_READER_SETUP_PREFETCH: &str = "RUSTFS_GET_MULTIPART_READER_SETUP_PREFETCH"; @@ -1194,6 +1200,46 @@ fn is_version_early_stop_enabled() -> bool { } } +fn is_get_metadata_data_read_early_stop_enabled() -> bool { + #[cfg(test)] + { + rustfs_utils::get_env_bool( + ENV_RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE, + DEFAULT_RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE, + ) + } + #[cfg(not(test))] + { + static CACHED: OnceLock = OnceLock::new(); + *CACHED.get_or_init(|| { + rustfs_utils::get_env_bool( + ENV_RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE, + DEFAULT_RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE, + ) + }) + } +} + +fn is_get_metadata_early_stop_bounded_fanout_enabled() -> bool { + #[cfg(test)] + { + rustfs_utils::get_env_bool( + ENV_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT, + DEFAULT_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT, + ) + } + #[cfg(not(test))] + { + static CACHED: OnceLock = OnceLock::new(); + *CACHED.get_or_init(|| { + rustfs_utils::get_env_bool( + ENV_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT, + DEFAULT_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT, + ) + }) + } +} + /// Check if multipart reads prefetch the next part's bitrot reader setup /// while the current part decodes (backlog#870). /// diff --git a/crates/ecstore/src/set_disk/read.rs b/crates/ecstore/src/set_disk/read.rs index 548f6c655..89dfa9640 100644 --- a/crates/ecstore/src/set_disk/read.rs +++ b/crates/ecstore/src/set_disk/read.rs @@ -3834,18 +3834,19 @@ mod tests { assert!(metadata_early_stop_permitted(true, true, false, "", false, false)); // observe=false (non-observed fanout) also disables early-stop. assert!(!metadata_early_stop_permitted(true, false, false, "", false, false)); - // Data reads are never eligible regardless of caller opt-in. + // Data reads require their own explicit rollout gate. assert!(!metadata_early_stop_permitted(true, true, true, "", false, false)); }, ); } #[test] - fn metadata_early_stop_rejects_data_reads() { + fn metadata_early_stop_requires_explicit_data_read_opt_in() { temp_env::with_vars( [ (ENV_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE, Some("true")), (ENV_RUSTFS_GET_METADATA_VERSION_EARLY_STOP_ENABLE, Some("true")), + (ENV_RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE, None), ], || { assert!(!should_allow_metadata_early_stop(true, "", false, false)); @@ -3854,6 +3855,17 @@ mod tests { assert!(should_allow_metadata_early_stop(false, "version-id", false, false)); }, ); + temp_env::with_vars( + [ + (ENV_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE, Some("true")), + (ENV_RUSTFS_GET_METADATA_VERSION_EARLY_STOP_ENABLE, Some("true")), + (ENV_RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE, Some("true")), + ], + || { + assert!(should_allow_metadata_early_stop(true, "", false, false)); + assert!(should_allow_metadata_early_stop(true, "version-id", false, false)); + }, + ); } #[test]