mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-21 11:56:38 +00:00
perf(ecstore): scale inline threshold by EC layout
Co-Authored-By: heihutu <heihutu@gmail.com>
This commit is contained in:
@@ -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";
|
const ZERO_SET_DRIVE_COUNT_ERROR: &str = "set drive count must be greater than zero";
|
||||||
|
|
||||||
pub static DEFAULT_INLINE_BLOCK: usize = 128 * 1024;
|
pub static DEFAULT_INLINE_BLOCK: usize = 128 * 1024;
|
||||||
|
const DEFAULT_INLINE_OBJECT_BUDGET: usize = 2 * DEFAULT_INLINE_BLOCK;
|
||||||
|
|
||||||
pub static DEFAULT_KVS: LazyLock<KVS> = LazyLock::new(|| {
|
pub static DEFAULT_KVS: LazyLock<KVS> = LazyLock::new(|| {
|
||||||
let kvs = vec![
|
let kvs = vec![
|
||||||
@@ -151,6 +152,8 @@ pub struct Config {
|
|||||||
inline_block: usize,
|
inline_block: usize,
|
||||||
initialized: bool,
|
initialized: bool,
|
||||||
#[serde(skip)]
|
#[serde(skip)]
|
||||||
|
inline_block_explicit: bool,
|
||||||
|
#[serde(skip)]
|
||||||
standard_parities: Vec<PoolParity>,
|
standard_parities: Vec<PoolParity>,
|
||||||
#[serde(skip)]
|
#[serde(skip)]
|
||||||
rrs_parities: Vec<PoolParity>,
|
rrs_parities: Vec<PoolParity>,
|
||||||
@@ -233,17 +236,19 @@ impl Config {
|
|||||||
.map(|(pool_index, pool)| (pool_index, pool.drives_per_set))
|
.map(|(pool_index, pool)| (pool_index, pool.drives_per_set))
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn should_inline(&self, shard_size: i64, versioned: bool) -> bool {
|
pub fn should_inline(&self, shard_size: i64, data_shards: usize, versioned: bool) -> bool {
|
||||||
if shard_size < 0 {
|
if shard_size < 0 || data_shards == 0 {
|
||||||
return false;
|
return false;
|
||||||
}
|
}
|
||||||
|
|
||||||
let shard_size = shard_size as usize;
|
let shard_size = shard_size as usize;
|
||||||
|
// Keep the historical two-data-shard object budget while preventing
|
||||||
let mut inline_block = DEFAULT_INLINE_BLOCK;
|
// wider EC layouts from multiplying the maximum inline object size.
|
||||||
if self.initialized {
|
let inline_block = if self.initialized && self.inline_block_explicit {
|
||||||
inline_block = self.inline_block;
|
self.inline_block
|
||||||
}
|
} else {
|
||||||
|
(DEFAULT_INLINE_OBJECT_BUDGET / data_shards).min(DEFAULT_INLINE_BLOCK)
|
||||||
|
};
|
||||||
|
|
||||||
if versioned {
|
if versioned {
|
||||||
shard_size <= inline_block / 8
|
shard_size <= inline_block / 8
|
||||||
@@ -392,6 +397,7 @@ fn lookup_config_for_pools_with_env(
|
|||||||
}
|
}
|
||||||
|
|
||||||
let optimize = overrides.optimize;
|
let optimize = overrides.optimize;
|
||||||
|
let inline_block_explicit = overrides.inline_block.is_some();
|
||||||
let inline_block = if let Some(value) = overrides.inline_block {
|
let inline_block = if let Some(value) = overrides.inline_block {
|
||||||
let block = value
|
let block = value
|
||||||
.parse::<bytesize::ByteSize>()
|
.parse::<bytesize::ByteSize>()
|
||||||
@@ -424,6 +430,7 @@ fn lookup_config_for_pools_with_env(
|
|||||||
optimize,
|
optimize,
|
||||||
inline_block,
|
inline_block,
|
||||||
initialized: true,
|
initialized: true,
|
||||||
|
inline_block_explicit,
|
||||||
standard_parities,
|
standard_parities,
|
||||||
rrs_parities,
|
rrs_parities,
|
||||||
})
|
})
|
||||||
@@ -541,22 +548,26 @@ mod tests {
|
|||||||
}
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn should_inline_preserves_exact_default_shard_boundaries() {
|
fn should_inline_scales_default_threshold_by_data_shards() {
|
||||||
let config = Config::default();
|
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 [
|
for (case, shard_size, data_shards, versioned, expected) in [
|
||||||
("unversioned below", 128 * 1024 - 1, false, true),
|
("EC2+1 unversioned exact", 128 * 1024, 2, false, true),
|
||||||
("unversioned exact", 128 * 1024, false, true),
|
("EC2+1 unversioned above", 128 * 1024 + 1, 2, false, false),
|
||||||
("unversioned above", 128 * 1024 + 1, false, false),
|
("EC2+1 versioned exact", 16 * 1024, 2, true, true),
|
||||||
("versioned below", 16 * 1024 - 1, true, true),
|
("EC2+1 versioned above", 16 * 1024 + 1, 2, true, false),
|
||||||
("versioned exact", 16 * 1024, true, true),
|
("EC8+4 unversioned exact", 32 * 1024, 8, false, true),
|
||||||
("versioned above", 16 * 1024 + 1, true, false),
|
("EC8+4 unversioned above", 32 * 1024 + 1, 8, false, false),
|
||||||
("negative", -1, 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!(
|
assert_eq!(
|
||||||
config.should_inline(shard_size, versioned),
|
config.should_inline(shard_size, data_shards, versioned),
|
||||||
expected,
|
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);
|
let shard_size = erasure.shard_file_size(object_size);
|
||||||
assert_eq!(shard_size, expected_shard_size, "{case}: object_size={object_size}");
|
assert_eq!(shard_size, expected_shard_size, "{case}: object_size={object_size}");
|
||||||
assert_eq!(
|
assert_eq!(
|
||||||
config.should_inline(shard_size, versioned),
|
config.should_inline(shard_size, erasure.data_shards, versioned),
|
||||||
expected,
|
expected,
|
||||||
"{case}: object_size={object_size}, shard_size={shard_size}, versioned={versioned}"
|
"{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]
|
#[test]
|
||||||
fn write_capability_contract_only_accepts_implemented_layouts() {
|
fn write_capability_contract_only_accepts_implemented_layouts() {
|
||||||
assert_eq!(SUPPORTED_WRITE_CLASSES, [STANDARD, RRS]);
|
assert_eq!(SUPPORTED_WRITE_CLASSES, [STANDARD, RRS]);
|
||||||
@@ -777,6 +803,7 @@ mod tests {
|
|||||||
let encoded = serde_json::to_string(&cfg).expect("config should serialize");
|
let encoded = serde_json::to_string(&cfg).expect("config should serialize");
|
||||||
assert!(!encoded.contains("standard_parities"));
|
assert!(!encoded.contains("standard_parities"));
|
||||||
assert!(!encoded.contains("rrs_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");
|
let decoded: Config = serde_json::from_str(&encoded).expect("legacy scalar config should deserialize");
|
||||||
assert_eq!(decoded.get_parity_for_sc(STANDARD), Some(2));
|
assert_eq!(decoded.get_parity_for_sc(STANDARD), Some(2));
|
||||||
|
|||||||
@@ -236,10 +236,6 @@ pub(crate) fn storage_class_parity(storage_class: Option<&str>) -> Option<usize>
|
|||||||
get_global_storage_class_snapshot().get_parity_for_sc(storage_class.unwrap_or_default())
|
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 {
|
pub(crate) fn deployment_upload_id(upload_id: &str) -> String {
|
||||||
base64_simd::URL_SAFE_NO_PAD
|
base64_simd::URL_SAFE_NO_PAD
|
||||||
.encode_to_string(format!("{}.{}", get_global_deployment_id().unwrap_or_default(), upload_id).as_bytes())
|
.encode_to_string(format!("{}.{}", get_global_deployment_id().unwrap_or_default(), upload_id).as_bytes())
|
||||||
|
|||||||
@@ -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");
|
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_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 = published.load_full();
|
||||||
let current_layout = resolve_write_layout(¤t, 0, 4, 2, None, false).expect("new snapshot should resolve");
|
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_eq!(current_layout.parity_drives, 1);
|
||||||
assert!(!current.should_inline(512, false));
|
assert!(!current.should_inline(512, current_layout.data_drives, false));
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -1188,7 +1188,8 @@ impl SetDisks {
|
|||||||
let erasure = Arc::new(erasure_from_file_info(&fi, false)?);
|
let erasure = Arc::new(erasure_from_file_info(&fi, false)?);
|
||||||
|
|
||||||
let put_object_size = known_put_object_storage_size(data.size());
|
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 collect_stage_timing = rustfs_io_metrics::put_stage_metrics_enabled() || issue3031_diag_enabled();
|
||||||
let shard_file_size = erasure.shard_file_size(put_object_size);
|
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 {
|
mod inline_put_commit_path_tests {
|
||||||
use super::hermetic_set_disks_support::hermetic_set_disks_isolated as hermetic_set_disks;
|
use super::hermetic_set_disks_support::hermetic_set_disks_isolated as hermetic_set_disks;
|
||||||
use super::*;
|
use super::*;
|
||||||
|
use crate::config::storageclass::lookup_config_for_pools_without_env;
|
||||||
use crate::disk::{DiskAPI as _, ReadOptions};
|
use crate::disk::{DiskAPI as _, ReadOptions};
|
||||||
use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations as _};
|
use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations as _};
|
||||||
|
use rustfs_config::server_config::KVS;
|
||||||
use tokio::io::AsyncReadExt;
|
use tokio::io::AsyncReadExt;
|
||||||
|
|
||||||
async fn make_bucket(disks: &[DiskStore], bucket: &str) {
|
async fn make_bucket(disks: &[DiskStore], bucket: &str) {
|
||||||
@@ -5986,6 +5989,46 @@ mod inline_put_commit_path_tests {
|
|||||||
assert_eq!(restored, payload);
|
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]
|
#[tokio::test]
|
||||||
async fn inline_put_direct_commit_accepts_exact_quorum_and_rejects_quorum_minus_one() {
|
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;
|
let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
|
||||||
|
|||||||
Reference in New Issue
Block a user