From d214c2758318553cdc7717b810a98137a041ba38 Mon Sep 17 00:00:00 2001 From: houseme Date: Mon, 31 Aug 2026 01:15:48 +0800 Subject: [PATCH] perf(ecstore): consolidate non-inline read planning (#6892) Co-authored-by: heihutu --- .../src/set_disk/core/io_primitives.rs | 332 +++++++++++++++--- crates/ecstore/src/set_disk/mod.rs | 160 +++++++-- crates/ecstore/src/set_disk/read.rs | 11 +- 3 files changed, 430 insertions(+), 73 deletions(-) diff --git a/crates/ecstore/src/set_disk/core/io_primitives.rs b/crates/ecstore/src/set_disk/core/io_primitives.rs index 1aade1908..499a47352 100644 --- a/crates/ecstore/src/set_disk/core/io_primitives.rs +++ b/crates/ecstore/src/set_disk/core/io_primitives.rs @@ -37,6 +37,8 @@ use super::super::ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DISKS; #[cfg(test)] use super::super::ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_OBJECT_PREFIX; #[cfg(test)] +use super::super::ENV_RUSTFS_GET_METADATA_TWO_PHASE_READ_PLAN_ENABLE; +#[cfg(test)] use super::super::get_metadata_slowtail_fault_delay; use super::super::{ Bytes, CHECK_PART_DISK_NOT_FOUND, DeleteOptions, DiskError, DiskStore, EVENT_SET_DISK_RENAME_TAIL_DRAIN_FAILED, @@ -50,10 +52,10 @@ use super::super::{ 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_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, + is_get_metadata_non_inline_data_read_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, }; #[cfg(test)] use crate::bucket::lifecycle::lifecycle::TRANSITION_COMPLETE; @@ -688,6 +690,10 @@ pub(in crate::set_disk) struct MetadataQuorumAccumulator { pub(in crate::set_disk) hard_errors: usize, pub(in crate::set_disk) candidate: Option, pub(in crate::set_disk) candidate_votes: usize, + // Bitset of shard indexes whose metadata matches the candidate. Erasure + // layouts are capped at 16 shards, so this stays allocation-free on the + // GET metadata hot path. + candidate_shard_mask: u16, pub(in crate::set_disk) conflicting_metadata: bool, pub(in crate::set_disk) delete_marker_seen: bool, pub(in crate::set_disk) delete_marker_candidates: Vec<(FileInfo, usize)>, @@ -709,6 +715,7 @@ impl MetadataQuorumAccumulator { hard_errors: 0, candidate: None, candidate_votes: 0, + candidate_shard_mask: 0, conflicting_metadata: false, delete_marker_seen: false, delete_marker_candidates: Vec::new(), @@ -724,6 +731,14 @@ impl MetadataQuorumAccumulator { } pub(in crate::set_disk) fn observe_file_info(&mut self, file_info: &FileInfo) { + self.observe_file_info_with_index(None, file_info); + } + + pub(in crate::set_disk) fn observe_file_info_at(&mut self, disk_index: usize, file_info: &FileInfo) { + self.observe_file_info_with_index(Some(disk_index), file_info); + } + + fn observe_file_info_with_index(&mut self, disk_index: Option, file_info: &FileInfo) { if !file_info_is_valid_for_metadata(file_info) { self.hard_errors = self.hard_errors.saturating_add(1); return; @@ -763,6 +778,11 @@ impl MetadataQuorumAccumulator { match &self.candidate { Some(candidate) if metadata_early_stop_candidate_matches(candidate, file_info) => { self.candidate_votes = self.candidate_votes.saturating_add(1); + if let Some(disk_index) = disk_index + && let Some(bit) = Self::candidate_shard_bit(candidate, file_info, disk_index) + { + self.candidate_shard_mask |= bit; + } } Some(_) => { self.conflicting_metadata = true; @@ -770,10 +790,38 @@ impl MetadataQuorumAccumulator { None => { self.candidate = Some(file_info.clone()); self.candidate_votes = 1; + if let Some(disk_index) = disk_index + && let Some(bit) = Self::candidate_shard_bit(file_info, file_info, disk_index) + { + self.candidate_shard_mask |= bit; + } } } } + fn candidate_shard_bit(candidate: &FileInfo, file_info: &FileInfo, disk_index: usize) -> Option { + let &erasure_index = candidate.erasure.distribution.get(disk_index)?; + if erasure_index == 0 || erasure_index > u16::BITS as usize || file_info.erasure.index != erasure_index { + return None; + } + Some(1u16 << (erasure_index - 1)) + } + + pub(in crate::set_disk) fn candidate_has_read_reserve(&self) -> bool { + self.candidate_read_reserve_target() + .is_some_and(|required| self.candidate_shard_mask.count_ones() as usize >= required) + } + + pub(in crate::set_disk) fn candidate_read_reserve_target(&self) -> Option { + let candidate = self.candidate.as_ref()?; + Some( + candidate + .erasure + .data_blocks + .saturating_add(usize::from(candidate.erasure.parity_blocks > 0)), + ) + } + pub(in crate::set_disk) fn observe_error(&mut self, err: &DiskError) { match err { DiskError::FileNotFound | DiskError::VolumeNotFound => { @@ -1084,11 +1132,7 @@ 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 { +fn non_inline_data_read_candidate_is_safe(candidate: &FileInfo) -> bool { if candidate.inline_data() || candidate.is_compressed() || candidate.is_remote() @@ -1100,34 +1144,11 @@ fn non_inline_data_read_candidate_is_safe( { 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 + candidate.has_valid_erasure_geometry() } +const NON_INLINE_SINGLE_PENDING_HEDGE_DELAY: Duration = Duration::from_millis(100); + fn data_read_inline_missing_shards_are_pending( candidate: &FileInfo, parts_metadata: &[FileInfo], @@ -2910,7 +2931,7 @@ impl SetDisks { read_data, healing, incl_free_versions, - read_data && is_get_metadata_two_phase_read_plan_enabled(), + read_data && is_get_metadata_non_inline_data_read_early_stop_enabled(), default_parity_count, allow_coalescing, ) @@ -3081,6 +3102,8 @@ impl SetDisks { let mut scheduled_count = 0usize; let mut force_full_wait = false; let mut final_miss_reason_override = None; + let mut non_inline_candidate_eligible = None; + let mut single_pending_hedge_deadline = None; let slowtail_fault = get_metadata_slowtail_fault_request(bucket.as_ref(), object.as_ref(), read_data); let spawn_read_version = |join_set: &mut JoinSet<(usize, disk::error::Result, Duration)>, index: usize, disk: Option| { @@ -3128,18 +3151,54 @@ impl SetDisks { } } - while let Some(result) = join_set.join_next().await { + loop { let mut defer_pending_inline_data_shard = false; + let result = if let Some(deadline) = single_pending_hedge_deadline.take() { + tokio::select! { + result = join_set.join_next() => result, + _ = tokio::time::sleep_until(deadline) => { + if bounded_fanout + && !force_full_wait + && join_set.len() == 1 + && non_inline_candidate_eligible == Some(true) + && !accumulator.candidate_has_read_reserve() + && next_fanout_index < disks.len() + { + while next_fanout_index < disks.len() { + let disk_index = fanout_order[next_fanout_index]; + next_fanout_index = next_fanout_index.saturating_add(1); + if let Some(disk) = disks.get(disk_index).cloned() { + spawn_read_version(&mut join_set, disk_index, disk); + scheduled_count = scheduled_count.saturating_add(1); + break; + } + } + } + continue; + } + } + } else { + join_set.join_next().await + }; + let Some(result) = result else { break }; match result { Ok((index, res, elapsed)) => match res { Ok(file_info) => { observations.push(MetadataFanoutObservation::from_file_info(&file_info, elapsed)); - accumulator.observe_file_info(&file_info); + if allow_non_inline_data_read_early_stop { + accumulator.observe_file_info_at(index, &file_info); + } else { + accumulator.observe_file_info(&file_info); + } + if allow_non_inline_data_read_early_stop && non_inline_candidate_eligible.is_none() { + non_inline_candidate_eligible = + accumulator.candidate.as_ref().map(non_inline_data_read_candidate_is_safe); + } if bounded_fanout && 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 + && !(non_inline_candidate_eligible == Some(true) && reason == GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_NOT_INLINE) { force_full_wait = true; @@ -3171,11 +3230,8 @@ 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) if non_inline_candidate_eligible == Some(true) => { + accumulator.candidate_has_read_reserve() } Some(candidate) => match data_read_early_stop_inline_body_miss_reason( bucket.as_ref(), @@ -3247,12 +3303,37 @@ impl SetDisks { } let pending_responses = join_set.len(); - let should_hedge_single_pending_data_read = read_data + // Inline verification can still depend on a missing data shard; + // issue one immediate spare when only that shard remains. The + // non-inline path keeps its delayed hedge below to avoid healthy + // reads paying speculative I/O before the candidate is classified. + let should_hedge_single_pending_inline_read = read_data && !force_full_wait && !defer_pending_inline_data_shard && pending_responses == 1 + && non_inline_candidate_eligible != Some(true) && accumulator.can_still_reach_early_stop_with_pending(pending_responses); - if bounded_fanout && force_full_wait { + // A non-inline plan must retain one extra matching shard as a + // reconstruction reserve. Schedule that reserve only after the + // candidate is known to be eligible, so inline GETs do not pay an + // extra fanout and the healthy path remains allocation-free. + let needs_non_inline_read_reserve = non_inline_candidate_eligible == Some(true) + && !accumulator.candidate_has_read_reserve() + && accumulator + .candidate_read_reserve_target() + .is_some_and(|reserve_target| scheduled_count < reserve_target || pending_responses == 0); + if bounded_fanout + && !force_full_wait + && (needs_non_inline_read_reserve || should_hedge_single_pending_inline_read) + && next_fanout_index < disks.len() + { + let disk_index = fanout_order[next_fanout_index]; + if let Some(disk) = disks.get(disk_index).cloned() { + spawn_read_version(&mut join_set, disk_index, disk); + scheduled_count = scheduled_count.saturating_add(1); + } + next_fanout_index = next_fanout_index.saturating_add(1); + } else if bounded_fanout && force_full_wait { while next_fanout_index < disks.len() { let disk_index = fanout_order[next_fanout_index]; if let Some(disk) = disks.get(disk_index).cloned() { @@ -3264,8 +3345,7 @@ impl SetDisks { } else if bounded_fanout && !defer_pending_inline_data_shard && next_fanout_index < disks.len() - && (!accumulator.can_still_reach_early_stop_with_pending(pending_responses) - || should_hedge_single_pending_data_read) + && !accumulator.can_still_reach_early_stop_with_pending(pending_responses) { let disk_index = fanout_order[next_fanout_index]; if let Some(disk) = disks.get(disk_index).cloned() { @@ -3274,6 +3354,17 @@ impl SetDisks { } next_fanout_index = next_fanout_index.saturating_add(1); } + if bounded_fanout + && !force_full_wait + && !defer_pending_inline_data_shard + && join_set.len() == 1 + && non_inline_candidate_eligible == Some(true) + && !accumulator.candidate_has_read_reserve() + && accumulator.can_still_reach_early_stop_with_pending(join_set.len()) + && next_fanout_index < disks.len() + { + single_pending_hedge_deadline = Some(tokio::time::Instant::now() + NON_INLINE_SINGLE_PENDING_HEDGE_DELAY); + } } let accumulator_miss_reason = accumulator.final_miss_reason(); @@ -7358,6 +7449,90 @@ mod tests { drop(dirs); } + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn metadata_slowtail_fault_gate_stops_before_unneeded_tail() { + const DISKS: usize = 4; + let bucket = "metadata-slowtail-gated-bucket"; + let object = "objects/metadata-slowtail-gated-object"; + let (dirs, disks) = call_counter_local_disks(bucket, DISKS).await; + install_mapped_metadata_fanout_fileinfo(&disks, bucket, object).await; + let order = bounded_metadata_fanout_order(bucket, object, DISKS, 2); + let slow_disk = *order.get(3).expect("four-disk fanout should have a deferred tail disk"); + let slow_disk_env = slow_disk.to_string(); + + temp_env::async_with_vars( + [ + (ENV_RUSTFS_GET_METADATA_TWO_PHASE_READ_PLAN_ENABLE, Some("true")), + (ENV_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE, Some("true")), + (ENV_RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE, Some("true")), + (ENV_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT, Some("true")), + (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DELAY_MS, Some("150")), + (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DISKS, Some(slow_disk_env.as_str())), + (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_BUCKET, Some(bucket)), + (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_OBJECT_PREFIX, Some("objects/")), + ], + async { + let calls = disk_call_counters::observe(object); + let read_with_data = + SetDisks::read_all_fileinfo_observed(&disks, bucket, bucket, object, "", true, false, false, true, 2); + let (parts_metadata, errs, diagnostics) = tokio::time::timeout(Duration::from_millis(500), read_with_data) + .await + .expect("gated metadata read should stop before the deferred slow tail") + .expect("gated metadata fanout should resolve"); + assert!(parts_metadata.iter().filter(|fi| fi.name == object).count() >= 3); + assert!(errs.iter().all(Option::is_none)); + assert!(diagnostics.total_responses() < DISKS); + assert_eq!(calls.total(disk_call_counters::KIND_METADATA_SLOWTAIL_FAULT), 0); + }, + ) + .await; + + drop(dirs); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn metadata_slowtail_fault_gate_hedges_an_initial_slow_data_shard() { + const DISKS: usize = 4; + let bucket = "metadata-slowtail-gated-initial-bucket"; + let object = "objects/metadata-slowtail-gated-initial-object"; + let (dirs, disks) = call_counter_local_disks(bucket, DISKS).await; + install_mapped_metadata_fanout_fileinfo(&disks, bucket, object).await; + let order = bounded_metadata_fanout_order(bucket, object, DISKS, 2); + let slow_disk = *order.get(1).expect("four-disk fanout should have an initial data disk"); + let spare_disk = *order.get(3).expect("four-disk fanout should have a spare disk"); + let slow_disk_env = slow_disk.to_string(); + + temp_env::async_with_vars( + [ + (ENV_RUSTFS_GET_METADATA_TWO_PHASE_READ_PLAN_ENABLE, Some("true")), + (ENV_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE, Some("true")), + (ENV_RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE, Some("true")), + (ENV_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT, Some("true")), + (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DELAY_MS, Some("500")), + (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DISKS, Some(slow_disk_env.as_str())), + (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_BUCKET, Some(bucket)), + (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_OBJECT_PREFIX, Some("objects/")), + ], + async { + let calls = disk_call_counters::observe(object); + let read_with_data = + SetDisks::read_all_fileinfo_observed(&disks, bucket, bucket, object, "", true, false, false, true, 2); + let (parts_metadata, errs, diagnostics) = tokio::time::timeout(Duration::from_millis(300), read_with_data) + .await + .expect("gated metadata read should hedge the initial slow shard") + .expect("gated metadata fanout should resolve"); + assert!(parts_metadata.iter().filter(|fi| fi.name == object).count() >= 3); + assert!(errs.iter().all(Option::is_none)); + assert!(diagnostics.total_responses() < DISKS); + assert_eq!(calls.for_disk(disk_call_counters::KIND_METADATA_SLOWTAIL_FAULT, slow_disk), 1); + assert_eq!(calls.for_disk(disk_call_counters::KIND_READ_VERSION, spare_disk), 1); + }, + ) + .await; + + drop(dirs); + } + /// Demo / regression guard for the backlog#1325 per-disk call counters. /// /// The metadata fan-out issues each `read_version` inside its own @@ -7771,6 +7946,32 @@ mod tests { } } + async fn install_mapped_metadata_fanout_fileinfo(disks: &[Option], bucket: &str, object: &str) { + let version_id = Uuid::new_v4(); + let data_dir = Uuid::new_v4(); + let mod_time = OffsetDateTime::now_utc(); + let distribution = FileInfo::new(&metadata_distribution_key(bucket, object), 2, 2) + .erasure + .distribution; + for (index, disk) in disks + .iter() + .enumerate() + .filter_map(|(index, disk)| disk.as_ref().map(|disk| (index, disk))) + { + 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"); + let mut file_info = valid_metadata_fanout_fileinfo(bucket, object, version_id, data_dir, mod_time); + file_info.erasure.distribution = distribution.clone(); + file_info.erasure.index = *distribution + .get(index) + .expect("mapped metadata distribution should cover every disk"); + disk.write_metadata(bucket, bucket, object, file_info) + .await + .expect("mapped metadata should be installed on every disk"); + } + } + async fn inline_metadata_fanout_fileinfos_with_mode( bucket: &str, object: &str, @@ -10393,6 +10594,41 @@ mod tests { assert_eq!(accumulator.candidate_latest_quorum(&impossible_parity), None); } + #[test] + fn metadata_quorum_accumulator_tracks_mapped_shards_and_requires_a_reserve() { + let version_id = Uuid::new_v4(); + let data_dir = Uuid::new_v4(); + let base = valid_metadata_fanout_fileinfo("bucket", "object", version_id, data_dir, OffsetDateTime::now_utc()); + let distribution = base.erasure.distribution.clone(); + let mut accumulator = MetadataQuorumAccumulator::new(4, 2, true); + + for (disk_index, &erasure_index) in distribution.iter().take(2).enumerate() { + let mut file_info = base.clone(); + file_info.erasure.index = erasure_index; + accumulator.observe_file_info_at(disk_index, &file_info); + } + assert!( + !accumulator.candidate_has_read_reserve(), + "data quorum without parity reserve must not early-stop" + ); + + let mut mismatched = base.clone(); + mismatched.erasure.index = distribution[3]; + accumulator.observe_file_info_at(2, &mismatched); + assert!( + !accumulator.candidate_has_read_reserve(), + "mapped index mismatch must not count as a reserve" + ); + + let mut reserve = base; + reserve.erasure.index = distribution[2]; + accumulator.observe_file_info_at(2, &reserve); + assert!( + accumulator.candidate_has_read_reserve(), + "one matching reserve shard should complete the read reserve" + ); + } + #[test] fn metadata_quorum_accumulator_treats_invalid_default_parity_as_full_fanout() { let accumulator = MetadataQuorumAccumulator::new(2, 2, true); diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 72d137b26..deb8230e3 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -773,10 +773,11 @@ 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. +// Opt-in non-inline data-read quorum early-stop rollout (backlog#1309). The +// existing metadata fanout still reads data-bearing metadata; this gate only +// permits a safe plain single-part candidate to stop before the full fanout. +// Keep it opt-in until the Linux multi-node slow-tail and small-inline cost +// gates are complete. The environment name is retained for compatibility. 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; @@ -1037,14 +1038,19 @@ mod prepared_get_object_metadata_tests { const READ_VERSION_BARRIER_GUARD: std::time::Duration = std::time::Duration::from_secs(10); fn object_with_initial_data_shards(bucket: &str, prefix: &str) -> String { + object_with_initial_data_shards_for_geometry(bucket, prefix, 4, 2) + } + + fn object_with_initial_data_shards_for_geometry(bucket: &str, prefix: &str, total_disks: usize, parity: usize) -> String { (0..1000) .map(|index| format!("{prefix}-{index}.bin")) .find(|name| { - let order = bounded_metadata_fanout_order(bucket, name, 4, 2); - let distribution = FileInfo::new(&[bucket, name].join("/"), 2, 2).erasure.distribution; - let mut seen = [false; 2]; - for disk_index in order.into_iter().take(3) { - if let Some(block_index @ 1..=2) = distribution.get(disk_index).copied() { + let order = bounded_metadata_fanout_order(bucket, name, total_disks, parity); + let data = total_disks.saturating_sub(parity); + let distribution = FileInfo::new(&[bucket, name].join("/"), data, parity).erasure.distribution; + let mut seen = vec![false; data]; + for disk_index in order.into_iter().take(total_disks.saturating_sub(parity).saturating_add(1)) { + if let Some(block_index) = distribution.get(disk_index).copied().filter(|index| *index <= data) { seen[block_index - 1] = true; } } @@ -1177,9 +1183,9 @@ 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() { + async fn non_inline_data_read_early_stop_uses_quorum_plan() { let (_dirs, set_disks) = make_local_set_disks(4, 2).await; - let bucket = "two-phase-read-plan"; + let bucket = "non-inline-read-plan"; let object = object_with_initial_data_shards(bucket, "non-inline-object"); let payload = vec![0x5a; 2 * 1024 * 1024]; let opts = ObjectOptions { @@ -1209,17 +1215,17 @@ mod prepared_get_object_metadata_tests { let mut reader = set_disks .get_object_reader(bucket, &object, None, HeaderMap::new(), &opts) .await - .expect("two-phase GET reader should open"); + .expect("quorum GET reader should open"); let mut restored = Vec::new(); reader .stream .read_to_end(&mut restored) .await - .expect("two-phase GET body should stream"); + .expect("quorum 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" + "non-inline quorum GET should retain a reconstruction reserve" ); }, ) @@ -1228,11 +1234,62 @@ mod prepared_get_object_metadata_tests { #[tokio::test] #[serial_test::serial(body_cache_hook)] - async fn two_phase_read_plan_preserves_inline_early_stop_path() { + async fn non_inline_data_read_early_stop_keeps_reserve_on_unequal_layout() { + let (_dirs, set_disks) = make_local_set_disks(6, 2).await; + let bucket = "non-inline-read-reserve"; + let object = object_with_initial_data_shards_for_geometry(bucket, "reserve-object", 6, 2); + 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); + let mut reader = set_disks + .get_object_reader(bucket, &object, None, HeaderMap::new(), &opts) + .await + .expect("quorum GET reader should open"); + let mut restored = Vec::new(); + reader + .stream + .read_to_end(&mut restored) + .await + .expect("quorum GET body should stream"); + assert_eq!(restored, payload); + assert_eq!( + calls.total(disk_call_counters::KIND_READ_VERSION), + 5, + "the unequal layout should schedule exactly one reserve beyond its data quorum" + ); + }, + ) + .await; + } + + #[tokio::test] + #[serial_test::serial(body_cache_hook)] + async fn non_inline_data_read_early_stop_preserves_inline_path() { let (_dirs, set_disks) = make_local_set_disks(4, 2).await; - let bucket = "two-phase-read-plan-inline"; + let bucket = "non-inline-read-plan-inline"; let object = object_with_initial_data_shards(bucket, "inline-object"); - let payload = b"two-phase inline payload".repeat(256); + let payload = b"quorum inline payload".repeat(256); let opts = ObjectOptions { no_lock: true, ..Default::default() @@ -1259,13 +1316,13 @@ mod prepared_get_object_metadata_tests { let mut reader = set_disks .get_object_reader(bucket, &object, None, HeaderMap::new(), &opts) .await - .expect("two-phase inline GET reader should open"); + .expect("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"); + .expect("inline GET body should stream"); assert_eq!(restored, payload); assert_eq!( test_get_object_reader_path_id(), @@ -1278,6 +1335,69 @@ mod prepared_get_object_metadata_tests { .await; } + #[tokio::test] + #[serial_test::serial(body_cache_hook)] + async fn non_inline_data_read_early_stop_does_not_add_inline_fanout_on_unequal_layout() { + let (_dirs, set_disks) = make_local_set_disks(6, 2).await; + let bucket = "inline-read-plan-unequal"; + let object = object_with_initial_data_shards_for_geometry(bucket, "inline-object", 6, 2); + let payload = b"inline quorum 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"); + + let read_once = |enabled: bool| { + let set_disks = Arc::clone(&set_disks); + let bucket = bucket.to_string(); + let object = object.clone(); + let payload = payload.clone(); + let opts = opts.clone(); + async move { + temp_env::async_with_vars( + [ + ( + "RUSTFS_GET_METADATA_TWO_PHASE_READ_PLAN_ENABLE", + Some(if enabled { "true" } else { "false" }), + ), + ("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("inline GET reader should open"); + let mut restored = Vec::new(); + reader + .stream + .read_to_end(&mut restored) + .await + .expect("inline GET body should stream"); + assert_eq!(restored, payload); + calls.total(disk_call_counters::KIND_READ_VERSION) + }, + ) + .await + } + }; + + let gate_off_calls = read_once(false).await; + let gate_on_calls = read_once(true).await; + assert_eq!(gate_on_calls, gate_off_calls, "inline gate must not add reserve fanout"); + } + #[test] #[serial_test::serial(body_cache_hook)] fn inline_data_read_early_stop_defaults_return_exact_body() { @@ -2024,7 +2144,7 @@ fn is_get_metadata_data_read_early_stop_enabled() -> bool { } } -fn is_get_metadata_two_phase_read_plan_enabled() -> bool { +fn is_get_metadata_non_inline_data_read_early_stop_enabled() -> bool { #[cfg(test)] { rustfs_utils::get_env_bool( diff --git a/crates/ecstore/src/set_disk/read.rs b/crates/ecstore/src/set_disk/read.rs index e42f13110..32ca821f3 100644 --- a/crates/ecstore/src/set_disk/read.rs +++ b/crates/ecstore/src/set_disk/read.rs @@ -128,7 +128,7 @@ 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; +use super::is_get_metadata_non_inline_data_read_early_stop_enabled; #[cfg(test)] use super::is_version_early_stop_enabled; #[cfg(test)] @@ -4272,15 +4272,16 @@ mod tests { } #[test] - fn two_phase_read_plan_gate_defaults_off_and_honors_override() { + #[serial(body_cache_hook)] + fn non_inline_data_read_early_stop_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()); + assert!(!is_get_metadata_non_inline_data_read_early_stop_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()); + assert!(is_get_metadata_non_inline_data_read_early_stop_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()); + assert!(!is_get_metadata_non_inline_data_read_early_stop_enabled()); }); }