From 85313fc06f89cf07481f5d73ec64ae58ca554646 Mon Sep 17 00:00:00 2001 From: Chris An Date: Fri, 6 Nov 2026 19:05:46 -0500 Subject: [PATCH] fix(runtime): release retained replay and allocator memory (#9211) --- Cargo.lock | 1 + crates/ecstore/src/cluster/rpc/http_auth.rs | 134 ++++++++++++++++++-- docs/operations/multipart-memory.md | 4 + rustfs/Cargo.toml | 2 + rustfs/src/main.rs | 6 + 5 files changed, 138 insertions(+), 9 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 3cfc6334c..59ccbe7a6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9612,6 +9612,7 @@ dependencies = [ "rustfs-log-analyzer", "rustfs-madmin", "rustfs-mimalloc", + "rustfs-mimalloc-sys", "rustfs-notify", "rustfs-object-capacity", "rustfs-object-data-cache", diff --git a/crates/ecstore/src/cluster/rpc/http_auth.rs b/crates/ecstore/src/cluster/rpc/http_auth.rs index 6c5a873d0..aa42e623b 100644 --- a/crates/ecstore/src/cluster/rpc/http_auth.rs +++ b/crates/ecstore/src/cluster/rpc/http_auth.rs @@ -366,10 +366,84 @@ fn resolve_replay_cache_capacity() -> usize { decision.capacity } +const RPC_NONCE_EXPIRATIONS_PER_PAGE: usize = 1_024; + +struct RpcNonceExpiration { + nonce: Uuid, + expiry_offset_ns: u32, + valid_until_offset_secs: i32, +} + +struct RpcNonceExpirationPage { + expires_at: Instant, + valid_until: i64, + entries: VecDeque, +} + +#[derive(Default)] +struct RpcNonceExpirations { + pages: VecDeque, +} + +impl RpcNonceExpirations { + fn push_back(&mut self, expires_at: Instant, valid_until: i64, nonce: Uuid) { + // Use exact offsets and start another page if either offset is too large. + // Fully drained pages can be freed while newer nonces are still valid. + if let Some(page) = self.pages.back_mut() + && page.entries.len() < RPC_NONCE_EXPIRATIONS_PER_PAGE + && let Some(offset) = expires_at.checked_duration_since(page.expires_at) + && let Ok(expiry_offset_ns) = u32::try_from(offset.as_nanos()) + && let Some(offset) = valid_until.checked_sub(page.valid_until) + && let Ok(valid_until_offset_secs) = i32::try_from(offset) + { + page.entries.push_back(RpcNonceExpiration { + nonce, + expiry_offset_ns, + valid_until_offset_secs, + }); + } else { + self.pages.push_back(RpcNonceExpirationPage { + expires_at, + valid_until, + entries: VecDeque::from([RpcNonceExpiration { + nonce, + expiry_offset_ns: 0, + valid_until_offset_secs: 0, + }]), + }); + } + } + + fn pop_expired(&mut self, now: Instant, wall_time: i64) -> Option { + let page = self.pages.front_mut()?; + let entry = page.entries.front()?; + let elapsed = now.checked_duration_since(page.expires_at)?; + let valid_until = page.valid_until + i64::from(entry.valid_until_offset_secs); + if elapsed <= Duration::from_nanos(u64::from(entry.expiry_offset_ns)) || wall_time <= valid_until { + return None; + } + + let nonce = entry.nonce; + page.entries.pop_front(); + if page.entries.is_empty() { + self.pages.pop_front(); + if self.pages.is_empty() { + self.pages.shrink_to_fit(); + } + } + Some(nonce) + } + + #[cfg(test)] + fn capacity(&self) -> usize { + self.pages.iter().map(|page| page.entries.capacity()).sum() + } +} + #[derive(Default)] struct RpcNonceCache { nonces: HashSet, - expirations: VecDeque<(Instant, i64, Uuid)>, + expirations: RpcNonceExpirations, max_wall_time: i64, } @@ -425,16 +499,13 @@ fn publish_nonce_cache_metrics(metrics: Option>) { impl RpcNonceCache { fn remove_expired(&mut self, now: Instant, wall_time: i64) -> usize { let mut removed = 0; - while matches!( - self.expirations.front(), - Some((expires_at, valid_until, _)) if *expires_at < now && *valid_until < wall_time - ) { - let Some((_, _, nonce)) = self.expirations.pop_front() else { - break; - }; + while let Some(nonce) = self.expirations.pop_expired(now, wall_time) { self.nonces.remove(&nonce); removed += 1; } + if removed > 0 && self.nonces.is_empty() { + self.nonces.shrink_to_fit(); + } removed } @@ -468,7 +539,7 @@ impl RpcNonceCache { } self.nonces.insert(record.nonce); self.expirations - .push_back((record.expires_at, record.signed_at.saturating_add(SIGNATURE_VALID_DURATION), record.nonce)); + .push_back(record.expires_at, record.signed_at.saturating_add(SIGNATURE_VALID_DURATION), record.nonce); ( Ok(()), Some(RpcNonceCacheMetrics { @@ -3178,6 +3249,51 @@ mod tests { assert!(cache.nonces.contains(&nonce_b)); } + #[test] + fn nonce_cache_releases_expiration_storage_after_a_burst() { + let start = Instant::now(); + let expiry = start + REPLAY_CACHE_RETENTION; + let mut cache = RpcNonceCache::default(); + + for nonce in 1..=4_096 { + let record = test_nonce_record(Uuid::from_u128(nonce), 100, start, 100, expiry, 4_096); + cache.check_and_record(record).0.expect("first use of nonce"); + } + assert!(cache.expirations.capacity() >= 4_096); + assert_eq!(cache.remove_expired(expiry, 701), 0); + assert_eq!(cache.remove_expired(expiry + Duration::from_nanos(1), 400), 0); + assert_eq!(cache.remove_expired(expiry + Duration::from_nanos(1), 701), 4_096); + assert!(cache.nonces.is_empty()); + assert_eq!(cache.nonces.capacity(), 0); + assert_eq!(cache.expirations.capacity(), 0, "expired entries must release their expiration storage"); + } + + #[test] + fn nonce_expirations_preserve_exact_deadlines_across_pages() { + let expiry = Instant::now(); + let first = Uuid::from_u128(1); + let second = Uuid::from_u128(2); + let third = Uuid::from_u128(3); + let fourth = Uuid::from_u128(4); + let mut queue = super::RpcNonceExpirations::default(); + let next = expiry + Duration::from_nanos(1); + let later = expiry + Duration::from_secs(5); + + queue.push_back(expiry, 501, first); + queue.push_back(next, 500, second); + queue.push_back(later, 505, third); + queue.push_back(later, i64::MAX, fourth); + assert_eq!(queue.pages.len(), 3); + assert_eq!(queue.pop_expired(next, 501), None); + assert_eq!(queue.pop_expired(next, 502), Some(first)); + assert_eq!(queue.pop_expired(next, 502), None); + assert_eq!(queue.pop_expired(later, 502), Some(second)); + assert_eq!(queue.pages.len(), 2); + assert_eq!(queue.pop_expired(later, 506), None); + assert_eq!(queue.pop_expired(later + Duration::from_nanos(1), 506), Some(third)); + assert_eq!(queue.pop_expired(later + Duration::from_nanos(1), i64::MAX), None); + } + #[test] fn nonce_cache_metrics_mark_successful_records_only() { let now = Instant::now(); diff --git a/docs/operations/multipart-memory.md b/docs/operations/multipart-memory.md index 30125f160..e2cc8cb78 100644 --- a/docs/operations/multipart-memory.md +++ b/docs/operations/multipart-memory.md @@ -31,6 +31,10 @@ A previously published gauge can remain visible after a statistic becomes unavai When live requested bytes are unavailable, capture a supported heap profile or allocation/free trace. Preserve attribution to allocation stacks and lifetime; committed bytes and RSS alone cannot establish a leak. Other raw allocator statistics, including page accounting, depend on the allocator build and are not interchangeable with live application bytes. +### Linux transparent huge pages + +Linux RustFS binaries disable transparent huge pages for mimalloc by default. This reduces resident memory when reclaimable allocations occupy only part of a huge page. Check `THP_enabled` in `/proc//status` and `AnonHugePages` in `/proc//smaps_rollup` to verify the running process. Both values should be zero with the default. The Linux binary also defaults to immediately purging free arena ranges. Set `MIMALLOC_PURGE_DELAY` before startup to override the delay in milliseconds. + ### Idle reclaim The reclaim loop waits for request, delete-tail, scanner, heal, EC and GET-buffer activity to become idle. Continuous traffic can skip reclaim. Check the activity gauges, skipped reasons and idle streak rather than assuming that waiting a fixed period guarantees collection. diff --git a/rustfs/Cargo.toml b/rustfs/Cargo.toml index 373e11cda..206074d8f 100644 --- a/rustfs/Cargo.toml +++ b/rustfs/Cargo.toml @@ -388,6 +388,8 @@ rustfs-mimalloc = { workspace = true } [target.'cfg(target_os = "linux")'.dependencies] libsystemd.workspace = true +rustfs-mimalloc = { workspace = true, features = ["no_thp"] } +rustfs-mimalloc-sys = "0.7.3" [target.'cfg(any(all(target_os = "macos", any(target_arch = "x86_64", target_arch = "aarch64")), all(target_os = "linux", target_env = "gnu", any(target_arch = "x86_64", target_arch = "aarch64"))))'.dependencies] pyroscope = { workspace = true, optional = true, features = ["backend-pprof-rs"] } diff --git a/rustfs/src/main.rs b/rustfs/src/main.rs index 36fd78899..92ecdfb3c 100644 --- a/rustfs/src/main.rs +++ b/rustfs/src/main.rs @@ -62,6 +62,12 @@ static GLOBAL: hotpath::CountingAllocator = hotpath::Countin static GLOBAL: rustfs_mimalloc::MiMalloc = rustfs_mimalloc::MiMalloc; fn main() -> std::process::ExitCode { + #[cfg(target_os = "linux")] + #[allow(unsafe_code)] + // SAFETY: The binary entrypoint runs this before RustFS creates any threads. + unsafe { + rustfs_mimalloc::MiMalloc::option_set_default(rustfs_mimalloc_sys::mi_option_t::mi_option_purge_delay, 0); + } let _hotpath_guard = hotpath::HotpathGuardBuilder::new("main").build(); rustfs::startup_entrypoint::run_process()