From f18df85c3071475fd2ee8be2cbf1096a6c17b6b9 Mon Sep 17 00:00:00 2001 From: houseme Date: Thu, 13 Aug 2026 22:49:38 +0800 Subject: [PATCH] perf(ecstore): scale inline threshold by EC layout Co-Authored-By: heihutu --- crates/ecstore/src/config/storageclass.rs | 67 ++++++++++++++++------- crates/ecstore/src/runtime/sources.rs | 4 -- crates/ecstore/src/set_disk/mod.rs | 4 +- crates/ecstore/src/set_disk/ops/object.rs | 45 ++++++++++++++- 4 files changed, 93 insertions(+), 27 deletions(-) diff --git a/crates/ecstore/src/config/storageclass.rs b/crates/ecstore/src/config/storageclass.rs index 11af87adf..3e1814334 100644 --- a/crates/ecstore/src/config/storageclass.rs +++ b/crates/ecstore/src/config/storageclass.rs @@ -101,6 +101,7 @@ const DEFAULT_RRS_STORAGE_CLASS: &str = "EC:1"; const ZERO_SET_DRIVE_COUNT_ERROR: &str = "set drive count must be greater than zero"; pub static DEFAULT_INLINE_BLOCK: usize = 128 * 1024; +const DEFAULT_INLINE_OBJECT_BUDGET: usize = 2 * DEFAULT_INLINE_BLOCK; pub static DEFAULT_KVS: LazyLock = LazyLock::new(|| { let kvs = vec![ @@ -151,6 +152,8 @@ pub struct Config { inline_block: usize, initialized: bool, #[serde(skip)] + inline_block_explicit: bool, + #[serde(skip)] standard_parities: Vec, #[serde(skip)] rrs_parities: Vec, @@ -233,17 +236,19 @@ impl Config { .map(|(pool_index, pool)| (pool_index, pool.drives_per_set)) } - pub fn should_inline(&self, shard_size: i64, versioned: bool) -> bool { - if shard_size < 0 { + pub fn should_inline(&self, shard_size: i64, data_shards: usize, versioned: bool) -> bool { + if shard_size < 0 || data_shards == 0 { return false; } let shard_size = shard_size as usize; - - let mut inline_block = DEFAULT_INLINE_BLOCK; - if self.initialized { - inline_block = self.inline_block; - } + // Keep the historical two-data-shard object budget while preventing + // wider EC layouts from multiplying the maximum inline object size. + let inline_block = if self.initialized && self.inline_block_explicit { + self.inline_block + } else { + (DEFAULT_INLINE_OBJECT_BUDGET / data_shards).min(DEFAULT_INLINE_BLOCK) + }; if versioned { shard_size <= inline_block / 8 @@ -392,6 +397,7 @@ fn lookup_config_for_pools_with_env( } let optimize = overrides.optimize; + let inline_block_explicit = overrides.inline_block.is_some(); let inline_block = if let Some(value) = overrides.inline_block { let block = value .parse::() @@ -424,6 +430,7 @@ fn lookup_config_for_pools_with_env( optimize, inline_block, initialized: true, + inline_block_explicit, standard_parities, rrs_parities, }) @@ -541,22 +548,26 @@ mod tests { } #[test] - fn should_inline_preserves_exact_default_shard_boundaries() { - let config = Config::default(); + fn should_inline_scales_default_threshold_by_data_shards() { + let config = lookup_config_for_pools_with_env(&KVS::new(), &[3, 12], no_env_overrides()) + .expect("default inline policy should resolve for EC2+1 and EC8+4"); - for (case, shard_size, versioned, expected) in [ - ("unversioned below", 128 * 1024 - 1, false, true), - ("unversioned exact", 128 * 1024, false, true), - ("unversioned above", 128 * 1024 + 1, false, false), - ("versioned below", 16 * 1024 - 1, true, true), - ("versioned exact", 16 * 1024, true, true), - ("versioned above", 16 * 1024 + 1, true, false), - ("negative", -1, false, false), + for (case, shard_size, data_shards, versioned, expected) in [ + ("EC2+1 unversioned exact", 128 * 1024, 2, false, true), + ("EC2+1 unversioned above", 128 * 1024 + 1, 2, false, false), + ("EC2+1 versioned exact", 16 * 1024, 2, true, true), + ("EC2+1 versioned above", 16 * 1024 + 1, 2, true, false), + ("EC8+4 unversioned exact", 32 * 1024, 8, false, true), + ("EC8+4 unversioned above", 32 * 1024 + 1, 8, false, false), + ("EC8+4 versioned exact", 4 * 1024, 8, true, true), + ("EC8+4 versioned above", 4 * 1024 + 1, 8, true, false), + ("negative", -1, 2, false, false), + ("zero data shards", 0, 0, false, false), ] { assert_eq!( - config.should_inline(shard_size, versioned), + config.should_inline(shard_size, data_shards, versioned), expected, - "{case}: shard_size={shard_size}, versioned={versioned}" + "{case}: shard_size={shard_size}, data_shards={data_shards}, versioned={versioned}" ); } } @@ -577,13 +588,28 @@ mod tests { let shard_size = erasure.shard_file_size(object_size); assert_eq!(shard_size, expected_shard_size, "{case}: object_size={object_size}"); assert_eq!( - config.should_inline(shard_size, versioned), + config.should_inline(shard_size, erasure.data_shards, versioned), expected, "{case}: object_size={object_size}, shard_size={shard_size}, versioned={versioned}" ); } } + #[test] + fn explicit_inline_block_preserves_fixed_per_shard_rollback() { + let overrides = StorageClassEnvOverrides { + inline_block: Some("128KiB".to_string()), + ..Default::default() + }; + let config = lookup_config_for_pools_with_env(&KVS::new(), &[12], overrides) + .expect("explicit inline block should resolve for EC8+4"); + + assert!(config.should_inline(128 * 1024, 8, false)); + assert!(!config.should_inline(128 * 1024 + 1, 8, false)); + assert!(config.should_inline(16 * 1024, 8, true)); + assert!(!config.should_inline(16 * 1024 + 1, 8, true)); + } + #[test] fn write_capability_contract_only_accepts_implemented_layouts() { assert_eq!(SUPPORTED_WRITE_CLASSES, [STANDARD, RRS]); @@ -777,6 +803,7 @@ mod tests { let encoded = serde_json::to_string(&cfg).expect("config should serialize"); assert!(!encoded.contains("standard_parities")); assert!(!encoded.contains("rrs_parities")); + assert!(!encoded.contains("inline_block_explicit")); let decoded: Config = serde_json::from_str(&encoded).expect("legacy scalar config should deserialize"); assert_eq!(decoded.get_parity_for_sc(STANDARD), Some(2)); diff --git a/crates/ecstore/src/runtime/sources.rs b/crates/ecstore/src/runtime/sources.rs index ed35c14d3..6c0b96c3b 100644 --- a/crates/ecstore/src/runtime/sources.rs +++ b/crates/ecstore/src/runtime/sources.rs @@ -236,10 +236,6 @@ pub(crate) fn storage_class_parity(storage_class: Option<&str>) -> Option get_global_storage_class_snapshot().get_parity_for_sc(storage_class.unwrap_or_default()) } -pub(crate) fn storage_class_should_inline(shard_size: i64, versioned: bool) -> bool { - get_global_storage_class_snapshot().should_inline(shard_size, versioned) -} - pub(crate) fn deployment_upload_id(upload_id: &str) -> String { base64_simd::URL_SAFE_NO_PAD .encode_to_string(format!("{}.{}", get_global_deployment_id().unwrap_or_default(), upload_id).as_bytes()) diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 35cec2f28..4abbfe749 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -2741,12 +2741,12 @@ mod write_layout_tests { let held_layout = resolve_write_layout(&held, 0, 4, 2, None, false).expect("held snapshot should remain valid"); assert_eq!(held_layout.parity_drives, 2); - assert!(held.should_inline(512, false)); + assert!(held.should_inline(512, held_layout.data_drives, false)); let current = published.load_full(); let current_layout = resolve_write_layout(¤t, 0, 4, 2, None, false).expect("new snapshot should resolve"); assert_eq!(current_layout.parity_drives, 1); - assert!(!current.should_inline(512, false)); + assert!(!current.should_inline(512, current_layout.data_drives, false)); } } diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index 2b65f79ac..3a5ae4904 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -1188,7 +1188,8 @@ impl SetDisks { let erasure = Arc::new(erasure_from_file_info(&fi, false)?); let put_object_size = known_put_object_storage_size(data.size()); - let is_inline_buffer = storage_class_config.should_inline(erasure.shard_file_size(put_object_size), opts.versioned); + let is_inline_buffer = + storage_class_config.should_inline(erasure.shard_file_size(put_object_size), erasure.data_shards, opts.versioned); let collect_stage_timing = rustfs_io_metrics::put_stage_metrics_enabled() || issue3031_diag_enabled(); let shard_file_size = erasure.shard_file_size(put_object_size); @@ -5921,8 +5922,10 @@ pub(in crate::set_disk::ops) mod hermetic_set_disks_support { mod inline_put_commit_path_tests { use super::hermetic_set_disks_support::hermetic_set_disks_isolated as hermetic_set_disks; use super::*; + use crate::config::storageclass::lookup_config_for_pools_without_env; use crate::disk::{DiskAPI as _, ReadOptions}; use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations as _}; + use rustfs_config::server_config::KVS; use tokio::io::AsyncReadExt; async fn make_bucket(disks: &[DiskStore], bucket: &str) { @@ -5986,6 +5989,46 @@ mod inline_put_commit_path_tests { assert_eq!(restored, payload); } + #[tokio::test] + async fn ec_8_4_default_budget_keeps_large_inline_candidate_out_of_xl_meta() { + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(12).await; + set_disks.set_test_storage_class_config( + lookup_config_for_pools_without_env(&KVS::new(), &[12]).expect("EC8+4 storage class should resolve"), + ); + let bucket = "ec-8-4-inline-budget"; + let object = "object.bin"; + let payload = vec![0x5c; 300 * 1024]; + make_bucket(&disk_stores, bucket).await; + + let mut reader = PutObjReader::from_vec(payload.clone()); + set_disks + .put_object(bucket, object, &mut reader, &ObjectOptions::default()) + .await + .expect("EC8+4 PUT should commit through the non-inline path"); + + for (disk_index, disk) in disk_stores.iter().enumerate() { + let file_info = disk + .read_version("", bucket, object, "", &ReadOptions::default()) + .await + .unwrap_or_else(|err| panic!("disk {disk_index} should persist EC8+4 metadata: {err}")); + assert_eq!(file_info.erasure.data_blocks, 8); + assert_eq!(file_info.erasure.parity_blocks, 4); + assert!(!file_info.inline_data(), "disk {disk_index} must keep the shard outside xl.meta"); + } + + let mut object_reader = set_disks + .get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default()) + .await + .expect("non-inline EC8+4 object should remain readable"); + let mut restored = Vec::new(); + object_reader + .stream + .read_to_end(&mut restored) + .await + .expect("non-inline EC8+4 object should stream"); + assert_eq!(restored, payload); + } + #[tokio::test] async fn inline_put_direct_commit_accepts_exact_quorum_and_rejects_quorum_minus_one() { let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;