fix(runtime): release retained replay and allocator memory (#9211)

This commit is contained in:
Chris An
2026-11-06 19:05:46 -05:00
committed by GitHub
parent b5f3207378
commit 85313fc06f
5 changed files with 138 additions and 9 deletions
Generated
+1
View File
@@ -9612,6 +9612,7 @@ dependencies = [
"rustfs-log-analyzer",
"rustfs-madmin",
"rustfs-mimalloc",
"rustfs-mimalloc-sys",
"rustfs-notify",
"rustfs-object-capacity",
"rustfs-object-data-cache",
+125 -9
View File
@@ -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<RpcNonceExpiration>,
}
#[derive(Default)]
struct RpcNonceExpirations {
pages: VecDeque<RpcNonceExpirationPage>,
}
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<Uuid> {
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<Uuid>,
expirations: VecDeque<(Instant, i64, Uuid)>,
expirations: RpcNonceExpirations,
max_wall_time: i64,
}
@@ -425,16 +499,13 @@ fn publish_nonce_cache_metrics(metrics: Option<RpcNonceCacheMetrics<'_>>) {
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();
+4
View File
@@ -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/<pid>/status` and `AnonHugePages` in `/proc/<pid>/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.
+2
View File
@@ -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"] }
+6
View File
@@ -62,6 +62,12 @@ static GLOBAL: hotpath::CountingAllocator<std::alloc::System> = 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()