From 526d6f667eeb5d2fa99393442cd9834be7928083 Mon Sep 17 00:00:00 2001 From: houseme Date: Sun, 16 Aug 2026 09:44:00 +0800 Subject: [PATCH] perf(ecstore): defer pending inline data shards (#6137) --- .../src/set_disk/core/io_primitives.rs | 216 +++++++++++++++++- 1 file changed, 214 insertions(+), 2 deletions(-) diff --git a/crates/ecstore/src/set_disk/core/io_primitives.rs b/crates/ecstore/src/set_disk/core/io_primitives.rs index fff99a3cc..43057104d 100644 --- a/crates/ecstore/src/set_disk/core/io_primitives.rs +++ b/crates/ecstore/src/set_disk/core/io_primitives.rs @@ -749,6 +749,63 @@ pub(in crate::set_disk) async fn data_read_early_stop_inline_body_miss_reason( } } +fn data_read_inline_missing_shards_are_pending( + candidate: &FileInfo, + parts_metadata: &[FileInfo], + errors: &[Option], + disks: &[Option], + fanout_order: &[usize], + scheduled_fanout_len: usize, +) -> bool { + 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; + }; + let distribution = &candidate.erasure.distribution; + let mut data_shards_seen_or_pending = vec![false; erasure.data_shards]; + let mut missing_pending_data_shards = 0usize; + + for (disk_index, file_info) in parts_metadata.iter().enumerate() { + let Some(&block_index) = distribution.get(disk_index) else { + return false; + }; + if block_index == 0 || block_index > erasure.data_shards { + continue; + } + if !disks.get(disk_index).is_some_and(Option::is_some) { + return false; + } + + let data_slot = block_index - 1; + if file_info.name.is_empty() { + let scheduled_and_not_failed = fanout_order + .get(..scheduled_fanout_len) + .is_some_and(|scheduled_disks| scheduled_disks.contains(&disk_index)) + && errors.get(disk_index).is_some_and(Option::is_none); + if scheduled_and_not_failed { + data_shards_seen_or_pending[data_slot] = true; + missing_pending_data_shards = missing_pending_data_shards.saturating_add(1); + continue; + } + return false; + } + if file_info.erasure.index != block_index + || !file_info.has_valid_erasure_geometry() + || !metadata_early_stop_candidate_matches(file_info, candidate) + || file_info.data.as_ref().is_none_or(|data| data.is_empty()) + { + return false; + } + data_shards_seen_or_pending[data_slot] = true; + } + + missing_pending_data_shards > 0 && data_shards_seen_or_pending.into_iter().all(|seen_or_pending| seen_or_pending) +} + pub(in crate::set_disk) fn classify_metadata_response_error(err: &DiskError) -> &'static str { match err { DiskError::FileNotFound | DiskError::VolumeNotFound => GET_METADATA_RESPONSE_NOT_FOUND, @@ -2532,6 +2589,7 @@ impl SetDisks { } while let Some(result) = join_set.join_next().await { + let mut defer_pending_inline_data_shard = false; match result { Ok((index, res, elapsed)) => match res { Ok(file_info) => { @@ -2574,8 +2632,22 @@ impl SetDisks { { None => true, Some(reason) => { - force_full_wait = true; final_miss_reason_override = Some(reason); + if bounded_fanout + && reason == GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_SHARD + && data_read_inline_missing_shards_are_pending( + candidate, + &ress, + &errors, + disks, + &fanout_order, + next_fanout_index, + ) + { + defer_pending_inline_data_shard = true; + } else { + force_full_wait = true; + } false } }, @@ -2621,6 +2693,7 @@ impl SetDisks { let pending_responses = join_set.len(); let should_hedge_single_pending_data_read = read_data && !force_full_wait + && !defer_pending_inline_data_shard && pending_responses == 1 && accumulator.can_still_reach_early_stop_with_pending(pending_responses); if bounded_fanout && force_full_wait { @@ -2633,6 +2706,7 @@ impl SetDisks { next_fanout_index = next_fanout_index.saturating_add(1); } } 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) @@ -5790,9 +5864,20 @@ mod tests { object: &str, payload: &[u8], uses_legacy_checksum: bool, + ) -> Vec { + inline_metadata_fanout_fileinfos_with_geometry(bucket, object, payload, uses_legacy_checksum, 2, 2).await + } + + async fn inline_metadata_fanout_fileinfos_with_geometry( + bucket: &str, + object: &str, + payload: &[u8], + uses_legacy_checksum: bool, + data_shards: usize, + parity_shards: usize, ) -> Vec { let distribution_key = metadata_distribution_key(bucket, object); - let mut base = FileInfo::new(&distribution_key, 2, 2); + let mut base = FileInfo::new(&distribution_key, data_shards, parity_shards); base.volume = bucket.to_string(); base.name = object.to_string(); base.size = i64::try_from(payload.len()).expect("test payload should fit i64"); @@ -5855,6 +5940,21 @@ mod tests { install_inline_metadata_fanout_files(disks, bucket, object, files).await; } + async fn install_inline_metadata_fanout_fileinfo_with_geometry( + disks: &[Option], + bucket: &str, + object: &str, + payload: &[u8], + data_shards: usize, + parity_shards: usize, + mutate: impl FnOnce(&mut [FileInfo]), + ) { + let mut files = + inline_metadata_fanout_fileinfos_with_geometry(bucket, object, payload, false, data_shards, parity_shards).await; + mutate(&mut files); + install_inline_metadata_fanout_files(disks, bucket, object, files).await; + } + async fn install_inline_metadata_fanout_files(disks: &[Option], bucket: &str, object: &str, files: Vec) { let distribution = files .first() @@ -6075,6 +6175,118 @@ mod tests { drop(dirs); } + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn bounded_metadata_early_stop_waits_for_pending_inline_data_shard() { + const DISKS: usize = 6; + const DATA_SHARDS: usize = 4; + const PARITY_SHARDS: usize = 2; + let bucket = "bounded-inline-data-get-pending-shard-bucket"; + let object = + object_with_initial_data_shards(bucket, "bounded-inline-data-get-pending-shard-object", DATA_SHARDS, DATA_SHARDS); + let (dirs, disks) = call_counter_local_disks(bucket, DISKS).await; + install_inline_metadata_fanout_fileinfo_with_geometry( + &disks, + bucket, + &object, + b"verified inline payload", + DATA_SHARDS, + PARITY_SHARDS, + |_| {}, + ) + .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 fanout_order = bounded_metadata_fanout_order(bucket, &object, DISKS, PARITY_SHARDS); + let distribution_key = metadata_distribution_key(bucket, &object); + let distribution = FileInfo::new(&distribution_key, DATA_SHARDS, PARITY_SHARDS) + .erasure + .distribution; + let paused_data_disk = *fanout_order + .iter() + .take(DATA_SHARDS) + .find(|disk_index| { + distribution + .get(**disk_index) + .is_some_and(|block_index| (1..=DATA_SHARDS).contains(block_index)) + }) + .expect("initial fanout should include a data shard to pause"); + let hedged_parity_disk = fanout_order[DATA_SHARDS]; + let unscheduled_parity_disk = fanout_order[DATA_SHARDS + 1]; + + let barrier = rename_fanout_barrier::arm(&object, paused_data_disk, rename_fanout_barrier::PHASE_READ_VERSION); + let tracker = rename_fanout_barrier::observe_tasks(&object); + let calls = disk_call_counters::observe(&object); + let disks_for_read = disks.clone(); + let object_for_read = object.clone(); + let mut read = tokio::spawn(async move { + SetDisks::read_all_fileinfo_observed( + &disks_for_read, + bucket, + bucket, + &object_for_read, + "", + true, + false, + false, + true, + PARITY_SHARDS, + ) + .await + }); + + tokio::time::timeout(BARRIER_PAUSE_GUARD, barrier.wait_until_paused()) + .await + .expect("initial data shard should pause before returning"); + tokio::time::timeout(BARRIER_PAUSE_GUARD, async { + while calls.for_disk(disk_call_counters::KIND_READ_VERSION, hedged_parity_disk) == 0 { + tokio::task::yield_now().await; + } + }) + .await + .expect("bounded fanout should hedge one parity disk while the data shard is pending"); + + assert!( + tokio::time::timeout(BARRIER_PAUSE_GUARD, &mut read).await.is_err(), + "inline data-read early-stop must wait for a scheduled missing data shard instead of forcing full wait" + ); + + barrier.release(); + let (parts_metadata, errs, diagnostics) = read + .await + .expect("metadata read task should not panic") + .expect("pending data shard should let the inline verifier finish"); + + assert_eq!( + calls.total(disk_call_counters::KIND_READ_VERSION), + 5, + "pending data-shard defer should not schedule the final parity disk" + ); + assert_eq!( + calls.for_disk(disk_call_counters::KIND_READ_VERSION, unscheduled_parity_disk), + 0, + "the remaining parity disk must stay unissued when pending data verification succeeds" + ); + assert_eq!( + tracker.running(), + 0, + "early-stop should drain spawned read_version tasks before returning" + ); + assert_eq!(diagnostics.total_responses(), 5); + assert_eq!(parts_metadata.iter().filter(|fi| fi.name == object).count(), 5); + assert!(errs.iter().all(Option::is_none)); + }, + ) + .await; + + drop(dirs); + } + #[tokio::test] async fn data_read_early_stop_verifies_legacy_inline_checksum_payload() { let bucket = "legacy-inline-data-get-fanout-bucket";