From 07647857548c06b4230eaa22c7d9faf9f5fdf4e9 Mon Sep 17 00:00:00 2001 From: Hauser Date: Sun, 4 Oct 2026 21:04:19 +0800 Subject: [PATCH] perf(scanner): reduce scan and checkpoint peak allocations (#8346) * perf(scanner): reduce scan and checkpoint peak allocations * test(scanner): isolate partial commit recovery scenarios --- crates/scanner/src/data_usage_define.rs | 2 +- .../src/data_usage_define/persistence.rs | 25 +++++-- .../src/data_usage_define/tests/cache_cost.rs | 1 + crates/scanner/src/scanner/backlog.rs | 68 +++++++++++++++++-- crates/scanner/src/scanner_folder.rs | 45 ++++++------ crates/scanner/src/scanner_folder/tests.rs | 13 ++++ 6 files changed, 121 insertions(+), 33 deletions(-) diff --git a/crates/scanner/src/data_usage_define.rs b/crates/scanner/src/data_usage_define.rs index b6d2da017..f711f11fc 100644 --- a/crates/scanner/src/data_usage_define.rs +++ b/crates/scanner/src/data_usage_define.rs @@ -43,7 +43,7 @@ use crate::storage_api::owner::HTTPPreconditions; use crate::{ BUCKET_META_PREFIX, EcstoreError as Error, EcstoreResult as StorageResult, RUSTFS_META_BUCKET, ReplicationConfig, SCANNER_PUBLICATION_EPOCH_CHANGED, ScannerObjectInfo as ObjectInfo, ScannerObjectOptions as ObjectOptions, StorageError, - TRANSITION_COMPLETE, save_config, save_config_with_preconditions, scanner_publication_admission_for_epoch, storageclass, + TRANSITION_COMPLETE, save_config, scanner_publication_admission_for_epoch, storageclass, }; use crate::{ScannerConfigObjectDelete, ScannerObjectIO}; diff --git a/crates/scanner/src/data_usage_define/persistence.rs b/crates/scanner/src/data_usage_define/persistence.rs index b7c70cff1..c31c8f2a4 100644 --- a/crates/scanner/src/data_usage_define/persistence.rs +++ b/crates/scanner/src/data_usage_define/persistence.rs @@ -13,6 +13,7 @@ // limitations under the License. use super::*; +use bytes::Bytes; use std::sync::Arc; #[derive(Debug)] @@ -477,7 +478,7 @@ impl DataUsageCache { async fn save_path_with_retry( store: Arc, path: &str, - buf: &[u8], + buf: &Bytes, timeout_duration: Duration, max_retries: u32, revision: Option, @@ -486,11 +487,14 @@ impl DataUsageCache { Self::ensure_cache_save_metrics_registered(); let path_type = Self::cache_path_type(path); let path = path.to_string(); + let sha256hex = (revision.is_some() && !buf.is_empty()) + .then(|| hex_simd::encode_to_string(Sha256::digest(buf), hex_simd::AsciiCase::Lower)); let save_result = Self::retry_save_op(path_type, timeout_duration, max_retries, || { let store_clone = store.clone(); let path_clone = path.clone(); - let buf_clone = buf.to_vec(); + let buf_clone = buf.clone(); + let sha256hex = sha256hex.clone(); let revision = revision.clone(); async move { let publication_admission = match expected_epoch { @@ -505,9 +509,18 @@ impl DataUsageCache { }); }; if let Some(revision) = revision { - save_config_with_preconditions(store_clone, &path_clone, buf_clone, revision.preconditions()).await?; + crate::save_config_shared_with_preconditions_and_lease_fence_and_scope( + store_clone, + &path_clone, + buf_clone, + sha256hex, + revision.preconditions(), + None, + None, + ) + .await?; } else { - save_config(store_clone, &path_clone, buf_clone).await?; + save_config(store_clone, &path_clone, buf_clone.to_vec()).await?; } Ok::<(), StorageError>(()) } @@ -539,7 +552,7 @@ impl DataUsageCache { }, ) .await?; - Ok::(reader.read_all().await? == buf) + Ok::(reader.read_all().await?.as_slice() == buf.as_ref()) }) .await; if matches!(reconcile, Ok(Ok(true))) { @@ -590,6 +603,8 @@ impl DataUsageCache { ) -> StorageResult<()> { let mut buf = Vec::new(); self.serialize(&mut rmp_serde::Serializer::new(&mut buf))?; + // Primary, backup, and retries read the same immutable encoded body. + let buf = Bytes::from(buf); let timeout_duration = Self::cache_save_timeout(); let path = path_join_buf(&[BUCKET_META_PREFIX, name]); diff --git a/crates/scanner/src/data_usage_define/tests/cache_cost.rs b/crates/scanner/src/data_usage_define/tests/cache_cost.rs index cc1990bbf..d9b79a47e 100644 --- a/crates/scanner/src/data_usage_define/tests/cache_cost.rs +++ b/crates/scanner/src/data_usage_define/tests/cache_cost.rs @@ -75,6 +75,7 @@ impl ObjectIO for CountingStore { options: &Self::ObjectOptions, ) -> StorageResult { assert_eq!(bucket, RUSTFS_META_BUCKET); + assert!(options.max_parity, "cache writes must preserve the metadata parity policy"); let started = WallInstant::now(); let mut bytes = Vec::new(); (&mut data.stream).take(MAX_WIRE_BYTES + 1).read_to_end(&mut bytes).await?; diff --git a/crates/scanner/src/scanner/backlog.rs b/crates/scanner/src/scanner/backlog.rs index ceea2255e..ba5df0feb 100644 --- a/crates/scanner/src/scanner/backlog.rs +++ b/crates/scanner/src/scanner/backlog.rs @@ -2164,6 +2164,14 @@ mod tests { } fn run_native_retirement_test(case: C) + where + C: FnOnce() -> F + Send + 'static, + F: std::future::Future + 'static, + { + run_native_retirement_test_with_budget(Duration::from_secs(180), case); + } + + fn run_native_retirement_test_with_budget(budget: Duration, case: C) where C: FnOnce() -> F + Send + 'static, F: std::future::Future + 'static, @@ -2177,7 +2185,7 @@ mod tests { .build() .expect("native retirement test runtime") .block_on(async { - tokio::time::timeout(Duration::from_secs(180), case()) + tokio::time::timeout(budget, case()) .await .expect("native retirement scenario must finish within its fixed budget"); }); @@ -2612,6 +2620,7 @@ mod tests { register_scanner_pause_backlog_retirement(); let root = tempfile::tempdir().unwrap(); + eprintln!("native expansion fixture old_pools={old_pool_count} stage=initialize_old_store"); let old = super::super::tests::setup_scanner_cycle_store_at_path_with_layout_and_disk_preinit( root.path(), false, @@ -2621,6 +2630,7 @@ mod tests { false, ) .await; + eprintln!("native expansion fixture old_pools={old_pool_count} stage=seed_old_authority"); let fault = unstable_source .then(|| NativeScannerPauseBacklogWriteFault::fail_before_write(Arc::clone(&old.pools[0].disk_set[0]), "publish", 2)); let now = unix_now(); @@ -2655,7 +2665,9 @@ mod tests { old.pool_meta_write_status() .await .expect("healthy pool metadata keeps the background recovery loop read-only"); + eprintln!("native expansion fixture old_pools={old_pool_count} stage=shutdown_old_store"); shutdown_native_retirement_store(old).await; + eprintln!("native expansion fixture old_pools={old_pool_count} stage=initialize_expanded_store"); let expanded = super::super::tests::setup_scanner_cycle_store_at_path_with_layout_and_disk_preinit( root.path(), false, @@ -2690,7 +2702,16 @@ mod tests { async fn assert_native_writer_expansion_recovers(old_pool_count: usize, failed_pool: usize, failed_write: usize) { use crate::storage_api::owner::NativeScannerPauseBacklogWriteFault; + let started = std::time::Instant::now(); + let stage = |name: &str| { + eprintln!( + "native expansion old_pools={old_pool_count} failed_pool={failed_pool} failed_write={failed_write} stage={name} elapsed={:?}", + started.elapsed() + ); + }; + stage("create_expanded_fixture"); let (root, store, original) = native_expanded_backlog_store(old_pool_count, false).await; + stage("load_old_authority"); let before = load_scanner_pause_backlog(Arc::clone(&store)) .await .expect("old cohort is authoritative before expansion publication"); @@ -2713,6 +2734,7 @@ mod tests { "publish", failed_write, ); + stage("publish_with_fault"); let result = ScannerPauseBacklogController::claim(Arc::clone(&store), now).await; if failed_write == 3 { let controller = result.expect("a complete new commit survives failed stabilization"); @@ -2724,6 +2746,7 @@ mod tests { } drop(fault); + stage("load_partial_commit"); let recovered = load_scanner_pause_backlog(Arc::clone(&store)) .await .expect("partial publication must leave a native authority or stable rollback point"); @@ -2758,8 +2781,11 @@ mod tests { assert_eq!(record.committed.as_ref(), Some(expected_commit)); } + stage("check_pool_metadata"); store.pool_meta_write_status().await.expect("healthy metadata before restart"); + stage("shutdown"); shutdown_native_retirement_store(store).await; + stage("restart"); let restarted = super::super::tests::setup_scanner_cycle_store_at_path_with_layout_and_disk_preinit( root.path(), false, @@ -2769,16 +2795,19 @@ mod tests { false, ) .await; + stage("load_after_restart"); let reloaded = load_scanner_pause_backlog(Arc::clone(&restarted)) .await .expect("a new store must recover from disk without the failed controller"); assert_eq!(&reloaded.ledger, expected); let retry_now = now.saturating_add(1); let retry_ledger = claim_scanner_pause_backlog_writer(expected, retry_now).expect("retry writer generation"); + stage("retry_writer_claim"); let retried = ScannerPauseBacklogController::claim(Arc::clone(&restarted), retry_now) .await .expect("retry must converge the entire expanded membership"); assert!(!retried.loaded.requires_reload); + stage("verify_recovered_membership"); assert_current_native_writer_ledger(&restarted, &retry_ledger).await; assert_eq!(retry_ledger.pending_full_scan, original.pending_full_scan); assert_eq!(retry_ledger.dirty_usage_buckets, original.dirty_usage_buckets); @@ -2786,7 +2815,9 @@ mod tests { assert_eq!(retry_ledger.last_finished_attempt_serial, original.current_attempt_serial); assert_eq!(retry_ledger.consecutive_failures, original.consecutive_failures + 1); drop(retried); + stage("final_shutdown"); shutdown_native_retirement_store(restarted).await; + stage("complete"); } async fn assert_current_native_writer_ledger(store: &Arc, expected: &ScannerPauseBacklogLedger) { @@ -2866,12 +2897,35 @@ mod tests { #[test] #[serial_test::serial] fn native_writer_expansion_partial_commit_recovers_after_restart() { - run_native_retirement_test(async || { - for old_pool_count in [1, 2] { - for failed_pool in [0, 2] { - assert_native_writer_expansion_recovers(old_pool_count, failed_pool, 2).await; - } - } + // Independently runnable cells let nextest isolate fixture workers + // and identify the failing topology. Divide the original 180s + // matrix budget across all four cells. + run_native_retirement_test_with_budget(Duration::from_secs(45), async || { + assert_native_writer_expansion_recovers(1, 0, 2).await; + }); + } + + #[test] + #[serial_test::serial] + fn native_writer_expansion_partial_commit_recovers_after_restart_one_pool_new_member() { + run_native_retirement_test_with_budget(Duration::from_secs(45), async || { + assert_native_writer_expansion_recovers(1, 2, 2).await; + }); + } + + #[test] + #[serial_test::serial] + fn native_writer_expansion_partial_commit_recovers_after_restart_two_pools_old_member() { + run_native_retirement_test_with_budget(Duration::from_secs(45), async || { + assert_native_writer_expansion_recovers(2, 0, 2).await; + }); + } + + #[test] + #[serial_test::serial] + fn native_writer_expansion_partial_commit_recovers_after_restart_two_pools_new_member() { + run_native_retirement_test_with_budget(Duration::from_secs(45), async || { + assert_native_writer_expansion_recovers(2, 2, 2).await; }); } diff --git a/crates/scanner/src/scanner_folder.rs b/crates/scanner/src/scanner_folder.rs index 2f44a941e..8569f2180 100644 --- a/crates/scanner/src/scanner_folder.rs +++ b/crates/scanner/src/scanner_folder.rs @@ -2806,18 +2806,23 @@ pub(crate) async fn scan_data_folder_scoped( cache.info.tier_registry_generation = Some(tier_registry.generation); let resume_frontier = cache.validated_scan_frontier().map(str::to_owned); + let bucket_name = cache.info.name.clone(); + let next_cycle = cache.info.next_cycle; + let had_scan_checkpoint = cache.info.scan_checkpoint.is_some(); + let new_cache = DataUsageCache { + info: cache.info.clone(), + ..Default::default() + }; + let update_cache = DataUsageCache { + info: cache.info.clone(), + ..Default::default() + }; // Create folder scanner let mut scanner = FolderScanner { root: base_path, - old_cache: cache.clone(), - new_cache: DataUsageCache { - info: cache.info.clone(), - ..Default::default() - }, - update_cache: DataUsageCache { - info: cache.info.clone(), - ..Default::default() - }, + old_cache: cache, + new_cache, + update_cache, data_usage_scanner_debug: false, heal_object_select, scan_mode, @@ -2871,7 +2876,7 @@ pub(crate) async fn scan_data_folder_scoped( // Read top level in bucket let mut root = DataUsageEntry::default(); let folder = CachedFolder { - name: cache.info.name.clone(), + name: bucket_name.clone(), parent: None, object_heal_prob_div: 1, }; @@ -2885,7 +2890,7 @@ pub(crate) async fn scan_data_folder_scoped( let new_cache = scanner.as_mut_new_cache(); new_cache.force_compact(DATA_SCANNER_COMPACT_AT_CHILDREN); new_cache.info.last_update = Some(SystemTime::now()); - new_cache.info.next_cycle = cache.info.next_cycle; + new_cache.info.next_cycle = next_cycle; let unresolved_objects = coverage_gap || root.failed_objects > 0 || !new_cache.info.failed_objects.is_empty() @@ -2906,7 +2911,7 @@ pub(crate) async fn scan_data_folder_scoped( new_cache.info.scan_plan_digest = None; } } - let had_scan_checkpoint = cache.info.scan_checkpoint.is_some() || new_cache.info.scan_checkpoint.is_some(); + let had_scan_checkpoint = had_scan_checkpoint || new_cache.info.scan_checkpoint.is_some(); new_cache.info.scan_resume_after = None; new_cache.info.scan_checkpoint = None; new_cache.info.scan_raw_enumeration_cursor = None; @@ -2918,20 +2923,20 @@ pub(crate) async fn scan_data_folder_scoped( close_disk_guard.close().await; if unresolved_objects || mixed_coverage { - Err(ScannerError::PartialCache(Box::new(new_cache.clone()))) + Err(ScannerError::PartialCache(Box::new(std::mem::take(new_cache)))) } else { - Ok(new_cache.clone()) + Ok(std::mem::take(new_cache)) } } Err(e) => { if ctx.is_cancelled() { - let root_hash = hash_path(&cache.info.name); + let root_hash = hash_path(&bucket_name); let root_has_progress = data_usage_root_has_progress(&root); let pending_heals_changed = scanner.pending_heals_changed; let (raw_enumeration_cursor, raw_enumeration_page_index) = scanner.take_raw_enumeration_resume_state(); let carry_forward_cache = ((raw_enumeration_cursor.is_some() || raw_enumeration_page_index.is_some()) && !root_has_progress) - .then(|| scanner.old_cache.cache.clone()); + .then(|| std::mem::take(&mut scanner.old_cache.cache)); if root_has_progress { scanner.carry_forward_old_children(&root_hash, &mut root); } @@ -2963,7 +2968,7 @@ pub(crate) async fn scan_data_folder_scoped( new_cache.force_compact(DATA_SCANNER_COMPACT_AT_CHILDREN); } new_cache.info.last_update = Some(SystemTime::now()); - new_cache.info.next_cycle = cache.info.next_cycle; + new_cache.info.next_cycle = next_cycle; new_cache.info.snapshot_complete = false; if root_has_progress { set_scan_checkpoint(new_cache, checkpoint_reason_from_budget(budget.reason())); @@ -2981,13 +2986,13 @@ pub(crate) async fn scan_data_folder_scoped( } } close_disk_guard.close().await; - return Err(ScannerError::PartialCache(Box::new(new_cache.clone()))); + return Err(ScannerError::PartialCache(Box::new(std::mem::take(new_cache)))); } } if matches!(&e, ScannerError::Io(io) if io.kind() == ErrorKind::NotFound) { - let mut partial_cache = scanner.old_cache.clone(); + let mut partial_cache = scanner.old_cache; partial_cache.info.last_update = Some(SystemTime::now()); - partial_cache.info.next_cycle = cache.info.next_cycle; + partial_cache.info.next_cycle = next_cycle; partial_cache.info.snapshot_complete = false; close_disk_guard.close().await; return Err(ScannerError::NamespaceNotFoundCache(Box::new(partial_cache))); diff --git a/crates/scanner/src/scanner_folder/tests.rs b/crates/scanner/src/scanner_folder/tests.rs index 4c8f89bc3..bde002039 100644 --- a/crates/scanner/src/scanner_folder/tests.rs +++ b/crates/scanner/src/scanner_folder/tests.rs @@ -2979,6 +2979,8 @@ async fn test_scan_data_folder_returns_raw_cursor_on_enumeration_cancel_without_ cache.prepare_bucket_checkpoint("bucket", 7, 3, source, plan, identity), crate::data_usage_define::DataUsageCachePrepareOutcome::Reset ); + cache.replace("bucket", "", DataUsageEntry::default()); + let retained_key_ptr = cache.cache.keys().next().expect("existing root key").as_ptr() as usize; let parent = CancellationToken::new(); let budget = ScannerCycleBudget::new_with_progress_tracking(&parent, Default::default()); @@ -3000,6 +3002,11 @@ async fn test_scan_data_folder_returns_raw_cursor_on_enumeration_cancel_without_ Err(ScannerError::PartialCache(partial_cache)) => partial_cache, other => panic!("expected raw enumeration partial cache after cancellation, got {other:?}"), }; + assert_eq!( + partial_cache.cache.keys().next().expect("retained root key").as_ptr() as usize, + retained_key_ptr, + "enumeration cancellation must transfer the existing cache allocation" + ); assert!( partial_cache @@ -3192,6 +3199,7 @@ async fn scan_data_folder_missing_bucket_returns_partial() { ..Default::default() }, ); + let retained_key_ptr = cache.cache.keys().next().expect("durable root key").as_ptr() as usize; let result = scan_data_folder( budget.token(), @@ -3209,6 +3217,11 @@ async fn scan_data_folder_missing_bucket_returns_partial() { Err(ScannerError::NamespaceNotFoundCache(partial)) => partial, other => panic!("missing bucket should keep the scan incomplete, got {other:?}"), }; + assert_eq!( + partial.cache.keys().next().expect("returned durable root key").as_ptr() as usize, + retained_key_ptr, + "a missing namespace must return the owned cache without a deep copy" + ); assert!(!partial.info.snapshot_complete); assert_eq!(partial.info.next_cycle, 9); let root = partial