gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-10-12 23:52:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

perf(scanner): reduce scan and checkpoint peak allocations (#8346)

Browse Source 
* perf(scanner): reduce scan and checkpoint peak allocations

* test(scanner): isolate partial commit recovery scenarios
This commit is contained in:
Hauser
2026-10-04 21:04:19 +08:00
committed by GitHub
parent 182b2bfe98
commit 0764785754
6 changed files with 121 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/scanner/src/data_usage_define.rs
+1 -1
View File
@@ -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};
crates/scanner/src/data_usage_define/persistence.rs
+20 -5
View File
@@ -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<S: ScannerObjectIO + ScannerConfigObjectDelete>(
store: Arc<S>,
path: &str,
buf: &[u8],
buf: &Bytes,
timeout_duration: Duration,
max_retries: u32,
revision: Option<DataUsageCacheRevision>,
@@ -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::<bool, StorageError>(reader.read_all().await? == buf)
Ok::<bool, StorageError>(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]);
crates/scanner/src/data_usage_define/tests/cache_cost.rs
+1
View File
@@ -75,6 +75,7 @@ impl ObjectIO for CountingStore {
options: &Self::ObjectOptions,
) -> StorageResult<Self::ObjectInfo> {
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?;
crates/scanner/src/scanner/backlog.rs
+61 -7
View File
@@ -2164,6 +2164,14 @@ mod tests {
}
fn run_native_retirement_test<C, F>(case: C)
where
C: FnOnce() -> F + Send + 'static,
F: std::future::Future<Output = ()> + 'static,
{
run_native_retirement_test_with_budget(Duration::from_secs(180), case);
}
fn run_native_retirement_test_with_budget<C, F>(budget: Duration, case: C)
where
C: FnOnce() -> F + Send + 'static,
F: std::future::Future<Output = ()> + '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<ECStore>, 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;
});
}
crates/scanner/src/scanner_folder.rs
+25 -20
View File
@@ -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)));
crates/scanner/src/scanner_folder/tests.rs
+13
View File
@@ -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