gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-10-27 06:18:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

perf(scanner): compact completed leaf entries in place

Browse Source 
Keep completed-folder eligibility and frontier handling in one compaction
boundary. Borrow leaf statistics and update the cached sampling flag without
cloning, deleting, or reinserting the entry; aggregate only non-leaf nodes.

Pin allocation identity, complete metadata, failure guards and caller state
with a regression that fails under the previous compaction implementation.

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>
This commit is contained in:
Hauser
2026-10-27 00:53:01 +08:00
parent 076ffc7a5f
commit 934238cb48
2 changed files with 114 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/scanner/src/scanner_folder.rs
+55 -42
View File
@@ -12,6 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use std::borrow::Cow;
use std::collections::{HashMap, HashSet};
use std::fs::FileType;
use std::io::ErrorKind;
@@ -1671,6 +1672,59 @@ impl FolderScanner {
self.data_usage_scanner_debug = debug;
}
fn compact_completed_folder(&mut self, folder: &CachedFolder, hash: &DataUsageHash, entry: &DataUsageEntry) {
// Completed healthy subtrees can keep aggregate usage. Coverage gaps
// and failed-object debt still need child entries for precise resume
// and to avoid rereading healthy siblings on a retry.
let forward_sweep = self.old_cache.info.scan_progress.is_some();
if entry.compacted || self.new_cache.info.name == folder.name || (forward_sweep && self.coverage_gap) {
return;
}
let summary = if entry.children.is_empty() {
Cow::Borrowed(entry)
} else if let Some(flat) = self.new_cache.size_recursive(&hash.0) {
Cow::Owned(flat)
} else {
return;
};
if forward_sweep && summary.failed_objects != 0 {
return;
}
let should_compact = summary.objects < DATA_SCANNER_COMPACT_LEAST_OBJECT
|| entry.children.iter().all(|key| {
self.new_cache
.cache
.get(key)
.is_none_or(|child| child.children.is_empty() && child.objects <= 1)
});
if !should_compact {
return;
}
// Heal revisits also reach this path without a parent advancing
// coverage. Keep the endpoint addressable when its completed
// subtree is replaced by an aggregate.
if self.coverage_frontier.as_deref().is_some_and(|frontier| {
frontier
.strip_prefix(&folder.name)
.is_some_and(|suffix| suffix.starts_with('/'))
}) {
self.record_completed_child(&folder.name);
}
match summary {
Cow::Borrowed(_) => {
// The leaf is already cached; only its sampling mode changes.
if let Some(cached) = self.new_cache.cache.get_mut(&hash.0) {
cached.compacted = true;
}
}
Cow::Owned(mut flat) => {
flat.compacted = true;
self.new_cache.delete_recursive(hash);
self.new_cache.replace_hashed(hash, &folder.parent, &flat);
}
}
}
fn refresh_usage_subtree(&mut self, hash: &DataUsageHash, parent: &Option<DataUsageHash>) {
if let Some(updates) = &mut self.usage_updates {
updates.cache.delete_recursive(hash);
@@ -3025,48 +3079,7 @@ impl FolderScanner {
self.new_cache.replace_hashed(&this_hash, &folder.parent, into);
}
// Completed healthy subtrees can keep aggregate usage. Coverage gaps
// and failed-object debt still need child entries for precise resume
// and to avoid rereading healthy siblings on a retry.
if !into.compacted
&& (self.old_cache.info.scan_progress.is_none() || !self.coverage_gap)
&& self.new_cache.info.name != folder.name
&& let Some(mut flat) = self.new_cache.size_recursive(&this_hash.key())
&& (self.old_cache.info.scan_progress.is_none() || flat.failed_objects == 0)
{
flat.compacted = true;
let mut should_compact = false;
if flat.objects < DATA_SCANNER_COMPACT_LEAST_OBJECT {
should_compact = true;
} else {
// Compact if we only have objects as children...
should_compact = true;
for k in &into.children {
if let Some(v) = self.new_cache.cache.get(k)
&& (!v.children.is_empty() || v.objects > 1)
{
should_compact = false;
break;
}
}
}
if should_compact {
// Heal revisits also reach this path without a parent advancing
// coverage. Keep the endpoint addressable when its completed
// subtree is replaced by an aggregate.
if self.coverage_frontier.as_deref().is_some_and(|frontier| {
frontier
.strip_prefix(&folder.name)
.is_some_and(|suffix| suffix.starts_with('/'))
}) {
self.record_completed_child(&folder.name);
}
self.new_cache.delete_recursive(&this_hash);
self.new_cache.replace_hashed(&this_hash, &folder.parent, &flat);
}
}
self.compact_completed_folder(&folder, &this_hash, into);
// Compact if too many children...
if !into.compacted {
crates/scanner/src/scanner_folder/tests/checkpoint_fixture/completed_compaction.rs
+59
View File
@@ -3,6 +3,65 @@
use super::*;
#[tokio::test]
async fn completed_leaf_compaction_reuses_cached_storage() {
let (mut scanner, root) = build_test_scanner().await;
let _cleanup = TestGuard { temp_dir: Some(root) };
scanner.old_cache = bound_checkpoint().0;
let folder = CachedFolder {
name: "bucket/object".to_string(),
parent: Some(hash_path("bucket")),
object_heal_prob_div: 1,
};
let hash = hash_path(&folder.name);
for (coverage_gap, failed_objects, should_compact) in [(false, 0, true), (true, 0, false), (false, 1, false)] {
scanner.coverage_gap = coverage_gap;
scanner.new_cache = DataUsageCache {
info: scanner.old_cache.info.clone(),
..Default::default()
};
let mut target = String::with_capacity(4096);
target.push_str("arn:test:replica");
let allocation = target.as_ptr();
let capacity = target.capacity();
let mut leaf = DataUsageEntry {
size: 2048,
objects: 1,
versions: 3,
failed_objects,
replication_stats: Some(rustfs_data_usage::ReplicationAllStats {
targets: HashMap::from([(target, rustfs_data_usage::ReplicationTargetUsage::default())]),
..Default::default()
}),
all_tier_stats: Some(rustfs_data_usage::AllTierStats::new()),
..Default::default()
};
leaf.obj_sizes.add(2048);
leaf.obj_versions.add(3);
let accumulator = leaf.clone();
let mut expected = serde_json::to_value(&leaf).expect("leaf metadata");
expected["compacted"] = serde_json::json!(should_compact);
scanner.new_cache.replace(&folder.name, "bucket", leaf);
scanner.compact_completed_folder(&folder, &hash, &accumulator);
let cached = scanner.new_cache.find(&folder.name).expect("retained leaf");
assert_eq!(serde_json::to_value(cached).expect("compacted metadata"), expected);
let target = cached
.replication_stats
.as_ref()
.expect("retained replication statistics")
.targets
.keys()
.next()
.expect("retained replication target");
assert_eq!(target.capacity(), capacity, "leaf compaction must preserve owned allocations");
assert_eq!(target.as_ptr(), allocation);
assert!(!accumulator.compacted, "cache compaction must not change the caller's accounting mode");
}
}
async fn scan(
disk: &Arc<Disk>,
cache: DataUsageCache,