mirror of
https://github.com/rustfs/rustfs.git
synced 2026-07-27 08:38:58 +00:00
eebd16d8a4
* feat(cache): add object data cache engine * feat(cache): wire app-layer object cache flow * refactor(cache): streamline app-layer cache flow * refactor(cache): tighten cache flow internals * refactor: address final clippy cleanup * chore(deps): update quick-xml to 0.41.0 * feat(cache): wire object data cache env config * fix(cache): gate materialize fill by cache plan * chore(cache): add object data cache benchmark gate * fix(cache): guard object cache fill size mismatches * refactor(cache): streamline object cache body planning * fix(cache): align object cache rollout config * test(cache): cover buffered object cache benchmark * test(cache): isolate object cache benchmark metrics * test(cache): mark materialize rollout experimental * test(cache): tighten object cache benchmark gate * fix(cache): address review findings for object data cache - singleflight: clean up leader entry on cancellation (Drop impl) so a dropped GET future can no longer wedge all subsequent fills for the same key; switch the fill map to a std Mutex and add a regression test - adapter: honor RUSTFS_OBJECT_DATA_CACHE_ENABLE=true by defaulting to hit_only when no explicit mode is set (explicit mode still wins) - planner: treat nil version UUIDs as "no value" per repo convention so unversioned objects key under the canonical "null" instead of fragmenting the key space - multipart: invalidate the object cache on the quota-exceeded rollback delete after complete-multipart, closing a stale-cache window - layering: move the disabled-cache fallback into app::context and drop the new infra->app layer-dependency baseline entry * fix(cache): close invalidation races and drop full-cache scan on writes - index: make identity-index insert/remove/prune atomic via starshard compute_if_present/compute_if_absent so concurrent fills can no longer drop each other's keys (lost keys made entries unreachable to invalidation until TTL); add a concurrency regression test - fill: register the key in the identity index before the entry becomes visible in the cache and re-check the index afterwards, undoing the fill when an invalidation raced in between (new skipped_invalidation_race fill result) - invalidate: with the index now authoritative, remove the full-cache iter() fallback that made every PUT/DELETE of a never-cached object O(total cache entries) (two scans per PUT, 2N per batch delete) - materialize-fill: fail the GET instead of falling back to the partially consumed stream after a mid-read error (the fallback would send a body missing its prefix under a full-length Content-Length), and log the same size-mismatch warning as the sibling buffering paths Co-Authored-By: heihutu <heihutu@gmail.com> * test(storage): fix media-dependent buffer clamp expectation test_concurrency_manager_multi_factor_strategy_buffer_clamp asserted media_cap.min(MI_B), but the implementation's final safety clamp is [32KiB, media_cap.max(MI_B)] — deliberately so a media cap above 1MiB (NVMe's 2MiB default) stays effective. The test only passed on machines detected as SSD/Unknown (cap == 1MiB) and failed on NVMe-backed CI runners with 2MiB != 1MiB. Assert the media cap itself, which is what the strategy actually guarantees on every environment. Co-Authored-By: heihutu <heihutu@gmail.com> * test(storage): format buffer clamp assertion * chore(logging): update tier guardrail path --------- Signed-off-by: houseme <housemecn@gmail.com> Co-authored-by: cxymds <cxymds@gmail.com> Co-authored-by: overtrue <anzhengchao@gmail.com> Co-authored-by: heihutu <heihutu@gmail.com>
312 lines
10 KiB
Rust
312 lines
10 KiB
Rust
// Copyright 2024 RustFS Team
|
|
//
|
|
// Licensed under the Apache License, Version 2.0 (the "License");
|
|
// you may not use this file except in compliance with the License.
|
|
// You may obtain a copy of the License at
|
|
//
|
|
// http://www.apache.org/licenses/LICENSE-2.0
|
|
//
|
|
// Unless required by applicable law or agreed to in writing, software
|
|
// distributed under the License is distributed on an "AS IS" BASIS,
|
|
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
// See the License for the specific language governing permissions and
|
|
// limitations under the License.
|
|
|
|
use crate::error::ObjectDataCacheConfigError;
|
|
use std::time::Duration;
|
|
use sysinfo::System;
|
|
|
|
const DEFAULT_DERIVED_MAX_MEMORY_PERCENT_CAP: u64 = 10;
|
|
const DEFAULT_DERIVED_MAX_BYTES_CAP: u64 = 64 * 1024 * 1024 * 1024;
|
|
|
|
/// Runtime mode for the object data cache.
|
|
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
|
|
pub enum ObjectDataCacheMode {
|
|
/// Cache is completely disabled.
|
|
#[default]
|
|
Disabled,
|
|
/// Cache lookups are allowed, but cache fill remains disabled.
|
|
HitOnly,
|
|
/// Cache fill is only allowed from an existing buffered body.
|
|
FillBufferedOnly,
|
|
/// Cache fill may materialize the final body stream exactly once.
|
|
FillMaterializeEnabled,
|
|
}
|
|
|
|
impl ObjectDataCacheMode {
|
|
pub(crate) const fn as_metric_label(self) -> &'static str {
|
|
match self {
|
|
Self::Disabled => "disabled",
|
|
Self::HitOnly => "hit_only",
|
|
Self::FillBufferedOnly => "fill_buffered_only",
|
|
Self::FillMaterializeEnabled => "fill_materialize_enabled",
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Object data cache configuration.
|
|
#[derive(Debug, Clone, PartialEq, Eq)]
|
|
pub struct ObjectDataCacheConfig {
|
|
/// Runtime mode gate for the cache engine.
|
|
pub mode: ObjectDataCacheMode,
|
|
/// Explicit byte-capacity override. Zero means derive from memory percent.
|
|
pub max_bytes: u64,
|
|
/// Memory percent used when `max_bytes` is zero.
|
|
pub max_memory_percent: u8,
|
|
/// Maximum cacheable entry size in bytes.
|
|
pub max_entry_bytes: u64,
|
|
/// Time-to-live for a cache entry.
|
|
pub ttl: Duration,
|
|
/// Time-to-idle for a cache entry.
|
|
pub time_to_idle: Duration,
|
|
/// Minimum free memory percent before fill is paused.
|
|
pub min_free_memory_percent: u8,
|
|
/// Fill concurrency multiplier applied to CPU count.
|
|
pub fill_concurrency_per_cpu: u16,
|
|
/// Absolute fill concurrency cap.
|
|
pub fill_concurrency_max: u16,
|
|
/// Conservative cap for keys attached to one object identity.
|
|
pub identity_keys_max: u16,
|
|
}
|
|
|
|
impl Default for ObjectDataCacheConfig {
|
|
fn default() -> Self {
|
|
Self {
|
|
mode: ObjectDataCacheMode::Disabled,
|
|
max_bytes: 0,
|
|
max_memory_percent: 5,
|
|
max_entry_bytes: 1_048_576,
|
|
ttl: Duration::from_secs(60),
|
|
time_to_idle: Duration::from_secs(30),
|
|
min_free_memory_percent: 20,
|
|
fill_concurrency_per_cpu: 1,
|
|
fill_concurrency_max: 32,
|
|
identity_keys_max: 16,
|
|
}
|
|
}
|
|
}
|
|
|
|
impl ObjectDataCacheConfig {
|
|
/// Returns true when the cache is effectively disabled.
|
|
pub const fn is_disabled(&self) -> bool {
|
|
matches!(self.mode, ObjectDataCacheMode::Disabled)
|
|
}
|
|
|
|
/// Returns true when the cache mode allows lookups.
|
|
pub const fn lookup_enabled(&self) -> bool {
|
|
!self.is_disabled()
|
|
}
|
|
|
|
/// Returns true when the cache mode allows fills.
|
|
pub const fn fill_enabled(&self) -> bool {
|
|
matches!(
|
|
self.mode,
|
|
ObjectDataCacheMode::FillBufferedOnly | ObjectDataCacheMode::FillMaterializeEnabled
|
|
)
|
|
}
|
|
|
|
/// Resolves the effective max capacity in bytes for the cache.
|
|
pub fn resolved_max_bytes(&self) -> Result<u64, ObjectDataCacheConfigError> {
|
|
if self.max_bytes > 0 {
|
|
return Ok(self.max_bytes);
|
|
}
|
|
|
|
let mut system = System::new();
|
|
system.refresh_memory();
|
|
let total_memory = system.total_memory();
|
|
let derived = total_memory.saturating_mul(u64::from(self.max_memory_percent)) / 100;
|
|
let resolved = clamp_derived_max_bytes(derived, total_memory, self.max_entry_bytes);
|
|
|
|
if resolved == 0 {
|
|
return Err(ObjectDataCacheConfigError::ZeroResolvedMaxBytes);
|
|
}
|
|
|
|
Ok(resolved)
|
|
}
|
|
|
|
/// Validates the configuration for internal consistency.
|
|
pub fn validate(&self) -> Result<(), ObjectDataCacheConfigError> {
|
|
if self.max_bytes == 0 && (self.max_memory_percent == 0 || self.max_memory_percent > 100) {
|
|
return Err(ObjectDataCacheConfigError::InvalidMaxMemoryPercent);
|
|
}
|
|
|
|
if self.max_entry_bytes == 0 {
|
|
return Err(ObjectDataCacheConfigError::ZeroMaxEntryBytes);
|
|
}
|
|
|
|
if self.ttl.is_zero() {
|
|
return Err(ObjectDataCacheConfigError::ZeroTimeToLiveSecs);
|
|
}
|
|
|
|
if self.time_to_idle.is_zero() {
|
|
return Err(ObjectDataCacheConfigError::ZeroTimeToIdleSecs);
|
|
}
|
|
|
|
if self.min_free_memory_percent == 0 || self.min_free_memory_percent > 100 {
|
|
return Err(ObjectDataCacheConfigError::InvalidMinFreeMemoryPercent);
|
|
}
|
|
|
|
if self.fill_concurrency_per_cpu == 0 {
|
|
return Err(ObjectDataCacheConfigError::ZeroFillConcurrencyPerCpu);
|
|
}
|
|
|
|
if self.fill_concurrency_max == 0 {
|
|
return Err(ObjectDataCacheConfigError::ZeroFillConcurrencyMax);
|
|
}
|
|
|
|
if self.fill_concurrency_max < self.fill_concurrency_per_cpu {
|
|
return Err(ObjectDataCacheConfigError::FillConcurrencyMaxTooSmall);
|
|
}
|
|
|
|
if self.identity_keys_max == 0 {
|
|
return Err(ObjectDataCacheConfigError::ZeroIdentityKeysMax);
|
|
}
|
|
|
|
Ok(())
|
|
}
|
|
}
|
|
|
|
fn clamp_derived_max_bytes(derived: u64, total_memory: u64, max_entry_bytes: u64) -> u64 {
|
|
let percent_cap = total_memory.saturating_mul(DEFAULT_DERIVED_MAX_MEMORY_PERCENT_CAP) / 100;
|
|
let safe_cap = percent_cap.min(DEFAULT_DERIVED_MAX_BYTES_CAP).max(max_entry_bytes);
|
|
|
|
derived.min(safe_cap).max(max_entry_bytes)
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::{DEFAULT_DERIVED_MAX_BYTES_CAP, ObjectDataCacheConfig, ObjectDataCacheMode, clamp_derived_max_bytes};
|
|
use crate::error::ObjectDataCacheConfigError;
|
|
use std::time::Duration;
|
|
|
|
#[test]
|
|
fn default_config_matches_v3_baseline() {
|
|
let config = ObjectDataCacheConfig::default();
|
|
|
|
assert!(matches!(config.mode, ObjectDataCacheMode::Disabled));
|
|
assert_eq!(config.max_bytes, 0);
|
|
assert_eq!(config.max_memory_percent, 5);
|
|
assert_eq!(config.max_entry_bytes, 1_048_576);
|
|
assert_eq!(config.ttl, Duration::from_secs(60));
|
|
assert_eq!(config.time_to_idle, Duration::from_secs(30));
|
|
assert_eq!(config.min_free_memory_percent, 20);
|
|
assert_eq!(config.fill_concurrency_per_cpu, 1);
|
|
assert_eq!(config.fill_concurrency_max, 32);
|
|
assert_eq!(config.identity_keys_max, 16);
|
|
}
|
|
|
|
#[test]
|
|
fn validate_rejects_invalid_memory_percent_when_capacity_is_derived() {
|
|
let config = ObjectDataCacheConfig {
|
|
max_memory_percent: 0,
|
|
..ObjectDataCacheConfig::default()
|
|
};
|
|
|
|
let err = config
|
|
.validate()
|
|
.expect_err("derived capacity requires a non-zero memory percent");
|
|
|
|
assert_eq!(err, ObjectDataCacheConfigError::InvalidMaxMemoryPercent);
|
|
}
|
|
|
|
#[test]
|
|
fn validate_rejects_zero_entry_size() {
|
|
let config = ObjectDataCacheConfig {
|
|
max_entry_bytes: 0,
|
|
..ObjectDataCacheConfig::default()
|
|
};
|
|
|
|
let err = config.validate().expect_err("entry size must stay positive");
|
|
|
|
assert_eq!(err, ObjectDataCacheConfigError::ZeroMaxEntryBytes);
|
|
}
|
|
|
|
#[test]
|
|
fn validate_rejects_zero_ttl() {
|
|
let config = ObjectDataCacheConfig {
|
|
ttl: Duration::ZERO,
|
|
..ObjectDataCacheConfig::default()
|
|
};
|
|
|
|
let err = config.validate().expect_err("ttl must stay positive");
|
|
|
|
assert_eq!(err, ObjectDataCacheConfigError::ZeroTimeToLiveSecs);
|
|
}
|
|
|
|
#[test]
|
|
fn validate_rejects_zero_time_to_idle() {
|
|
let config = ObjectDataCacheConfig {
|
|
time_to_idle: Duration::ZERO,
|
|
..ObjectDataCacheConfig::default()
|
|
};
|
|
|
|
let err = config.validate().expect_err("time-to-idle must stay positive");
|
|
|
|
assert_eq!(err, ObjectDataCacheConfigError::ZeroTimeToIdleSecs);
|
|
}
|
|
|
|
#[test]
|
|
fn validate_rejects_invalid_fill_concurrency_bounds() {
|
|
let config = ObjectDataCacheConfig {
|
|
fill_concurrency_per_cpu: 2,
|
|
fill_concurrency_max: 1,
|
|
..ObjectDataCacheConfig::default()
|
|
};
|
|
|
|
let err = config
|
|
.validate()
|
|
.expect_err("max fill concurrency must not be smaller than per-cpu factor");
|
|
|
|
assert_eq!(err, ObjectDataCacheConfigError::FillConcurrencyMaxTooSmall);
|
|
}
|
|
|
|
#[test]
|
|
fn validate_accepts_explicit_byte_cap() {
|
|
let config = ObjectDataCacheConfig {
|
|
mode: ObjectDataCacheMode::HitOnly,
|
|
max_bytes: 4_194_304,
|
|
max_memory_percent: 0,
|
|
..ObjectDataCacheConfig::default()
|
|
};
|
|
|
|
assert!(config.validate().is_ok());
|
|
}
|
|
|
|
#[test]
|
|
fn resolved_max_bytes_prefers_explicit_cap() {
|
|
let config = ObjectDataCacheConfig {
|
|
max_bytes: 4_194_304,
|
|
..ObjectDataCacheConfig::default()
|
|
};
|
|
|
|
let resolved = config
|
|
.resolved_max_bytes()
|
|
.expect("explicit max_bytes should be returned directly");
|
|
|
|
assert_eq!(resolved, 4_194_304);
|
|
}
|
|
|
|
#[test]
|
|
fn resolved_max_bytes_is_at_least_max_entry_bytes() {
|
|
let config = ObjectDataCacheConfig {
|
|
max_bytes: 0,
|
|
max_memory_percent: 1,
|
|
max_entry_bytes: 8_388_608,
|
|
..ObjectDataCacheConfig::default()
|
|
};
|
|
|
|
let resolved = config.resolved_max_bytes().expect("derived capacity should stay positive");
|
|
|
|
assert!(resolved >= config.max_entry_bytes);
|
|
}
|
|
|
|
#[test]
|
|
fn derived_max_bytes_clamps_to_v3_safe_cap() {
|
|
let one_tib = 1024_u64 * 1024 * 1024 * 1024;
|
|
let derived = one_tib / 2;
|
|
let resolved = clamp_derived_max_bytes(derived, one_tib, 1_048_576);
|
|
|
|
assert_eq!(resolved, DEFAULT_DERIVED_MAX_BYTES_CAP);
|
|
}
|
|
}
|