diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 8bb71dcda..a95b158d6 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -140,7 +140,7 @@ use rustfs_lock::local_lock::LocalLock; use rustfs_lock::{FastLockGuard, LockManager, NamespaceLock, NamespaceLockGuard, NamespaceLockWrapper, ObjectKey}; use rustfs_madmin::heal_commands::{HealDriveInfo, HealResultItem, Infos}; use rustfs_object_capacity::capacity_scope::{ - CapacityScope, CapacityScopeDisk, record_capacity_scope, record_global_dirty_scope, + CapacityScope, CapacityScopeDisk, current_dirty_generation, record_capacity_scope, record_global_dirty_scope, }; use rustfs_s3_types::EventName; use rustfs_utils::http::headers::AMZ_OBJECT_TAGGING; @@ -164,6 +164,7 @@ use std::hash::Hash; use std::mem::{self}; use std::pin::Pin; use std::sync::OnceLock; +use std::sync::atomic::{AtomicU64, Ordering}; use std::task::{Context, Poll}; use std::time::{Instant, SystemTime, UNIX_EPOCH}; use std::{ @@ -349,10 +350,145 @@ fn should_persist_encryption_original_size(metadata: &HashMap) - || metadata.contains_key(SSEC_KEY_MD5_HEADER) } +/// Per-set memoized capacity dirty scope. +/// +/// The disk endpoint/path identity of each slot in a set is immutable for the +/// set's lifetime (heal replaces the [`DiskStore`] instance but keeps the same +/// endpoint and root), so the dirty scope only needs to be built once instead +/// of allocating a `String` per online disk on every successful write +/// (backlog#1315). Slots are filled lazily as disks are observed online; once +/// every slot has contributed, `complete` latches and the hot path returns the +/// shared `Arc` without any scan. +#[derive(Default, Debug)] +struct CapacityScopeCache { + /// Per-slot resolved disk identity; `None` slots have not been seen online. + slots: Vec>, + /// Deduplicated scope covering every slot resolved so far. + scope: Option>, + /// Latches once every slot is resolved; the scope is then stable. + complete: bool, +} + +impl CapacityScopeCache { + /// Whether `disks` contains an online disk whose slot has not been recorded + /// yet. Read-only, allocation-free (used under the read lock). + fn has_unresolved_slot(&self, disks: &[Option]) -> bool { + disks + .iter() + .enumerate() + .any(|(idx, disk)| disk.is_some() && self.slots.get(idx).is_none_or(|slot| slot.is_none())) + } +} + +impl SetDisks { + /// Return the memoized dirty scope for this set, resolving any newly-online + /// slots first. Steady-state writes hit the fast path (a read lock and an + /// `Arc` clone) with no per-disk `String` allocation. + fn capacity_scope(&self, disks: &[Option]) -> Arc { + { + let cache = self.capacity_scope_cache.read().unwrap_or_else(|p| p.into_inner()); + if let Some(scope) = cache.scope.as_ref() + && (cache.complete || !cache.has_unresolved_slot(disks)) + { + return scope.clone(); + } + } + self.resolve_capacity_scope(disks) + } + + /// Slow path: fill newly-observed slots and rebuild the deduplicated scope. + fn resolve_capacity_scope(&self, disks: &[Option]) -> Arc { + let mut cache = self.capacity_scope_cache.write().unwrap_or_else(|p| p.into_inner()); + if cache.slots.len() < self.set_drive_count { + cache.slots.resize(self.set_drive_count, None); + } + + let mut changed = false; + for (idx, disk) in disks.iter().enumerate() { + let Some(slot) = cache.slots.get_mut(idx) else { + break; + }; + if slot.is_some() { + continue; + } + if let Some(disk) = disk { + *slot = Some(CapacityScopeDisk { + endpoint: disk.endpoint().to_string(), + drive_path: disk.to_string(), + }); + changed = true; + } + } + + if changed || cache.scope.is_none() { + let mut unique = HashSet::with_capacity(cache.slots.len()); + let mut scoped_disks = Vec::with_capacity(cache.slots.len()); + for slot in cache.slots.iter().flatten() { + if unique.insert(slot.clone()) { + scoped_disks.push(slot.clone()); + } + } + cache.complete = !cache.slots.is_empty() && cache.slots.iter().all(|slot| slot.is_some()); + cache.scope = Some(Arc::new(CapacityScope { disks: scoped_disks })); + } + + cache.scope.clone().unwrap_or_else(|| Arc::new(CapacityScope::default())) + } + + /// Record the set's dirty scope after a successful write. + /// + /// The global dirty registry is only upgraded on the first write of each + /// registry generation: once this set has marked its disks, subsequent + /// writes skip the global mutex until a refresh drain advances the + /// generation, forcing a re-mark (backlog#1315). The scope-token registry + /// (multipart completion) still records per token so the app-side write + /// settle can consume it. + fn record_capacity_scope_if_needed(&self, scope_token: Option, disks: &[Option]) { + let scope = self.capacity_scope(disks); + if scope.disks.is_empty() { + return; + } + + let generation = current_dirty_generation(); + if self.capacity_dirty_generation.load(Ordering::Acquire) != generation { + // First write of this generation (or after a drain): upgrade the + // global registry and cache the generation observed under its lock. + let observed = record_global_dirty_scope((*scope).clone()); + self.capacity_dirty_generation.store(observed, Ordering::Release); + } + + if let Some(token) = scope_token { + record_capacity_scope(token, (*scope).clone()); + } + } + + /// Mark healed disks dirty from the (infrequent) heal path. + /// + /// Heal passes disks in erasure-distribution order (`shuffle_disks`), not + /// physical-slot order, so they must not seed the slot-indexed memo — doing + /// so could record a disk under the wrong slot and drop another from the + /// steady-state scope. Heal therefore builds an ad-hoc scope from the disks + /// it actually rewrote and marks the global registry directly. This runs at + /// heal frequency, so it does not use the per-generation skip fast-path + /// (backlog#1315). + fn record_healed_capacity_scope(&self, disks: &[Option]) { + let scope = capacity_scope_from_disks(disks); + if scope.disks.is_empty() { + return; + } + // Do not advance the set's generation marker here: this ad-hoc scope is + // only the subset of disks heal rewrote, whereas the marker asserts the + // full set was marked. Leaving the marker untouched keeps the next write + // free to upgrade the full-set scope if it has not been marked yet. + let _ = record_global_dirty_scope(scope); + } +} + +/// Build an ad-hoc, deduplicated dirty scope from `disks`. Used by the heal +/// path where disks are not in physical-slot order (backlog#1315). fn capacity_scope_from_disks(disks: &[Option]) -> CapacityScope { let mut unique = HashSet::with_capacity(disks.len()); let mut scoped_disks = Vec::with_capacity(disks.len()); - for disk in disks.iter().flatten() { let scope_disk = CapacityScopeDisk { endpoint: disk.endpoint().to_string(), @@ -362,23 +498,9 @@ fn capacity_scope_from_disks(disks: &[Option]) -> CapacityScope { scoped_disks.push(scope_disk); } } - CapacityScope { disks: scoped_disks } } -fn record_capacity_scope_if_needed(scope_token: Option, disks: &[Option]) { - let scope = capacity_scope_from_disks(disks); - if scope.disks.is_empty() { - return; - } - - record_global_dirty_scope(scope.clone()); - - if let Some(token) = scope_token { - record_capacity_scope(token, scope); - } -} - /// Get the duplex buffer size from environment variable or use default. /// /// This function reads `RUSTFS_DUPLEX_BUFFER_SIZE` environment variable @@ -1628,6 +1750,15 @@ pub struct SetDisks { /// Slice 3 sources `local_lock_manager` from this context to give each /// instance its own lock namespace. ctx: Arc, + /// Memoized capacity dirty scope so successful writes reuse a prebuilt + /// `Arc` instead of allocating a `String` per online disk (backlog#1315). + /// `Arc` so clones of a set share one memo. + capacity_scope_cache: Arc>, + /// Last global dirty-registry generation this set marked. `u64::MAX` until + /// the first write; equality with the current generation lets steady-state + /// writes skip the global registry mutex (backlog#1315). `Arc` so clones of + /// a set share one generation marker. + capacity_dirty_generation: Arc, } #[derive(Clone, Debug, Eq, Hash, PartialEq)] @@ -1828,6 +1959,8 @@ impl SetDisks { // process lock-manager singleton, so this is unchanged. local_lock_manager: ctx.lock_manager(), ctx, + capacity_scope_cache: Arc::new(std::sync::RwLock::new(CapacityScopeCache::default())), + capacity_dirty_generation: Arc::new(AtomicU64::new(u64::MAX)), }) } @@ -4628,6 +4761,109 @@ mod tests { .await } + // backlog#1315: the memoized dirty scope must be byte-for-byte identical to + // the ad-hoc scope the previous per-write construction produced, otherwise + // dirty-disk keys diverge from the disk-cache keys and capacity counts drift. + #[tokio::test] + #[serial] + async fn capacity_scope_memo_matches_adhoc_and_is_reused() { + use rustfs_object_capacity::capacity_scope::drain_global_dirty_scopes; + + let (dir_a, disk_a) = make_single_local_disk().await; + let (dir_b, disk_b) = make_single_local_disk().await; + let disks = vec![Some(disk_a), Some(disk_b)]; + let set = make_set_disks_with(disks.clone()).await; + + let expected = capacity_scope_from_disks(&disks); + let first = set.capacity_scope(&disks); + assert_eq!(*first, expected, "memoized scope must equal the ad-hoc per-disk construction bit-for-bit"); + + // Second call on a fully-resolved set returns the very same Arc (no new + // allocation): proving steady-state writes do not rebuild String/HashSet. + let second = set.capacity_scope(&disks); + assert!(Arc::ptr_eq(&first, &second), "steady-state scope must reuse the cached Arc"); + + let _ = drain_global_dirty_scopes(); + drop((dir_a, dir_b)); + } + + // backlog#1315: the global registry mutex must be upgraded only on the first + // write of each generation; steady-state writes skip it. Reverting the + // generation skip makes the upgrade count grow per write and fails this test. + #[tokio::test] + #[serial] + async fn record_capacity_scope_upgrades_registry_once_per_generation() { + use rustfs_object_capacity::capacity_scope::{drain_global_dirty_scopes, global_dirty_upgrade_count}; + + let (dir_a, disk_a) = make_single_local_disk().await; + let (dir_b, disk_b) = make_single_local_disk().await; + let disks = vec![Some(disk_a), Some(disk_b)]; + let set = make_set_disks_with(disks.clone()).await; + + // Start from a clean registry generation. + let _ = drain_global_dirty_scopes(); + let before = global_dirty_upgrade_count(); + + // First write of this generation upgrades the registry exactly once. + set.record_capacity_scope_if_needed(None, &disks); + assert_eq!( + global_dirty_upgrade_count(), + before + 1, + "first write of a generation must upgrade the global registry" + ); + + // Subsequent writes in the same generation must not touch the mutex. + for _ in 0..16 { + set.record_capacity_scope_if_needed(None, &disks); + } + assert_eq!( + global_dirty_upgrade_count(), + before + 1, + "steady-state writes must reuse the generation mark, not re-upgrade" + ); + + // A drain advances the generation; the next write must re-mark so the + // disks it wrote are captured by the following refresh (no lost update). + let drained = drain_global_dirty_scopes(); + assert_eq!(drained.len(), 2, "both set disks must have been recorded dirty"); + set.record_capacity_scope_if_needed(None, &disks); + assert_eq!( + global_dirty_upgrade_count(), + before + 2, + "the first write after a drain must re-upgrade the registry" + ); + + let _ = drain_global_dirty_scopes(); + drop((dir_a, dir_b)); + } + + // backlog#1315: an offline slot must not force the per-write slow path, and + // the resolved scope must still cover every online disk. + #[tokio::test] + #[serial] + async fn capacity_scope_tolerates_offline_slot_without_reallocating() { + use rustfs_object_capacity::capacity_scope::drain_global_dirty_scopes; + + let (dir_a, disk_a) = make_single_local_disk().await; + // Slot 1 is permanently offline (None); the set never resolves it. + let disks = vec![Some(disk_a), None]; + let set = make_set_disks_with(disks.clone()).await; + + let first = set.capacity_scope(&disks); + assert_eq!(first.disks.len(), 1, "only the online disk contributes to the scope"); + // Even though the set is not "complete" (slot 1 unresolved), repeated + // writes with the same online set reuse the cached Arc via the + // no-unresolved-slot fast path. + let second = set.capacity_scope(&disks); + assert!( + Arc::ptr_eq(&first, &second), + "a stable online subset must reuse the cached Arc even with an offline slot" + ); + + let _ = drain_global_dirty_scopes(); + drop(dir_a); + } + // issue #4189: an orphan directory tree (empty dirs, no xl.meta) must be purged. #[tokio::test] async fn purge_orphan_dir_object_removes_empty_tree() { diff --git a/crates/ecstore/src/set_disk/ops/heal.rs b/crates/ecstore/src/set_disk/ops/heal.rs index 639f7d08b..d10b145b6 100644 --- a/crates/ecstore/src/set_disk/ops/heal.rs +++ b/crates/ecstore/src/set_disk/ops/heal.rs @@ -682,7 +682,7 @@ impl SetDisks { )); } - record_capacity_scope_if_needed(None, &out_dated_disks); + self.record_healed_capacity_scope(&out_dated_disks); // The object is healthy here; sweep any data dirs left behind // by pre-#3510 unversioned overwrites, which the dangling paths diff --git a/crates/ecstore/src/set_disk/ops/multipart.rs b/crates/ecstore/src/set_disk/ops/multipart.rs index 38704c3ff..9d4d2d1d9 100644 --- a/crates/ecstore/src/set_disk/ops/multipart.rs +++ b/crates/ecstore/src/set_disk/ops/multipart.rs @@ -1542,7 +1542,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { } } - record_capacity_scope_if_needed(opts.capacity_scope_token, &online_disks); + self.record_capacity_scope_if_needed(opts.capacity_scope_token, &online_disks); fi.is_latest = true; diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index 0d3df49fe..884321530 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -1076,7 +1076,7 @@ impl SetDisks { if fi.is_compressed() { record_compression_total_memory(actual_size as u64, w_size as u64).await; } - record_capacity_scope_if_needed(opts.capacity_scope_token, &online_disks); + self.record_capacity_scope_if_needed(opts.capacity_scope_token, &online_disks); fi.replication_state_internal = Some(replication_state_to_filemeta(&opts.put_replication_state())); @@ -1801,7 +1801,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { } } - record_capacity_scope_if_needed(opts.capacity_scope_token, &disks); + self.record_capacity_scope_if_needed(opts.capacity_scope_token, &disks); let mut rollback_futures = Vec::new(); for fi_vers in &vers { @@ -1997,7 +1997,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { .map_err(|e| to_object_err(e, vec![bucket, object]))?; let disks = self.disk_inventory().await; - record_capacity_scope_if_needed(opts.capacity_scope_token, &disks); + self.record_capacity_scope_if_needed(opts.capacity_scope_token, &disks); let mut oi = ObjectInfo::from_file_info(&fi, bucket, object, opts.versioned || opts.version_suspended); oi.replication_decision = goi.replication_decision; @@ -2027,7 +2027,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { .map_err(|e| to_object_err(e, vec![bucket, object]))?; let disks = self.disk_inventory().await; - record_capacity_scope_if_needed(opts.capacity_scope_token, &disks); + self.record_capacity_scope_if_needed(opts.capacity_scope_token, &disks); let mut obj_info = ObjectInfo::from_file_info(&dfi, bucket, object, opts.versioned || opts.version_suspended); obj_info.size = goi.size; @@ -2326,7 +2326,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { "transition completed on remote tier but source cleanup failed; skipping external lifecycle transition notification" ); } else { - record_capacity_scope_if_needed(opts.capacity_scope_token, &disks); + self.record_capacity_scope_if_needed(opts.capacity_scope_token, &disks); } for disk in disks.iter() { diff --git a/crates/object-capacity/src/capacity_manager.rs b/crates/object-capacity/src/capacity_manager.rs index 509349eca..876fa9be6 100644 --- a/crates/object-capacity/src/capacity_manager.rs +++ b/crates/object-capacity/src/capacity_manager.rs @@ -37,6 +37,7 @@ use std::collections::{HashMap, HashSet}; use std::future::Future; use std::panic::AssertUnwindSafe; use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; use std::time::{Duration, Instant}; use tokio::sync::{Mutex, RwLock, watch}; use tracing::{debug, info, warn}; @@ -450,21 +451,81 @@ const REFRESH_JOINER_WAIT_TIMEOUT: Duration = Duration::from_secs(300); /// never overflow. const MAX_BACKGROUND_INTERVAL: Duration = Duration::from_secs(30 * 24 * 60 * 60); -#[derive(Clone, Copy, Debug, Default)] -struct WriteBucket { - second: u64, - count: usize, +/// A single second-granular write bucket packed into one `u64` so it can be +/// updated with a lock-free compare-and-swap. The high 32 bits hold the +/// monotonic-second key, the low 32 bits hold the write count for that second. +/// A per-second count never approaches `u32::MAX`, and the count saturates +/// rather than wrapping into the second key (backlog#1315). +#[derive(Default)] +struct AtomicWriteBucket(AtomicU64); + +const WRITE_BUCKET_COUNT_MASK: u64 = 0xFFFF_FFFF; + +impl AtomicWriteBucket { + #[inline] + fn unpack(packed: u64) -> (u64, usize) { + let second = packed >> 32; + let count = (packed & WRITE_BUCKET_COUNT_MASK) as usize; + (second, count) + } + + #[inline] + fn pack(second: u64, count: usize) -> u64 { + let count = (count as u64).min(WRITE_BUCKET_COUNT_MASK); + (second << 32) | count + } + + /// Record one write for `now_second`, resetting the bucket if it holds a + /// different (older or future) second. Lock-free CAS loop tolerating + /// concurrent writers landing on the same bucket. + fn record(&self, now_second: u64) { + loop { + let current = self.0.load(Ordering::Acquire); + let (second, count) = Self::unpack(current); + let next = if second == now_second { + Self::pack(now_second, count.saturating_add(1)) + } else { + Self::pack(now_second, 1) + }; + if self + .0 + .compare_exchange_weak(current, next, Ordering::AcqRel, Ordering::Acquire) + .is_ok() + { + return; + } + } + } + + fn snapshot(&self) -> (u64, usize) { + Self::unpack(self.0.load(Ordering::Acquire)) + } + + #[cfg(test)] + fn store(&self, second: u64, count: usize) { + self.0.store(Self::pack(second, count), Ordering::Release); + } } -/// Write record for tracking write operations -#[derive(Debug)] +/// Lock-free write record for tracking write operations. +/// +/// Previously guarded by an async `RwLock` taken on every successful write in +/// the PUT response path; the write side is now a set of relaxed/CAS atomic +/// updates so concurrent small-object PUTs no longer serialize on a single +/// writer lock (backlog#1315). Counters use saturating arithmetic and monotonic +/// second keys, preserving the exact debounce/frequency semantics the refresh +/// scheduler relies on. pub struct WriteRecord { - /// Last write time - pub last_write_time: Option, - /// Write count - pub write_count: usize, + /// Last write time, encoded as monotonic nanoseconds since `epoch`. + last_write_nanos: AtomicU64, + /// Whether any write has been recorded (`0` == never). Kept separate from + /// `last_write_nanos` so a genuine near-zero-nanos first write is still + /// distinguished from "no write yet". + has_write: AtomicU64, + /// Total write count (saturating). Observability only. + write_count: AtomicU64, /// Fixed-size time buckets for the recent write window. - write_buckets: [WriteBucket; WRITE_WINDOW_BUCKETS], + write_buckets: [AtomicWriteBucket; WRITE_WINDOW_BUCKETS], /// Monotonic origin for bucket keys. Wall-clock keys made an NTP step /// backwards mark recent buckets as "future" and silently suppress /// write-triggered refreshes until the clock catches up (backlog#1022 S32). @@ -474,9 +535,10 @@ pub struct WriteRecord { impl WriteRecord { fn new() -> Self { Self { - last_write_time: None, - write_count: 0, - write_buckets: [WriteBucket::default(); WRITE_WINDOW_BUCKETS], + last_write_nanos: AtomicU64::new(0), + has_write: AtomicU64::new(0), + write_count: AtomicU64::new(0), + write_buckets: std::array::from_fn(|_| AtomicWriteBucket::default()), epoch: Instant::now(), } } @@ -485,31 +547,48 @@ impl WriteRecord { self.epoch.elapsed().as_secs() } + /// Total number of writes recorded (saturating). Observability only. + fn total_write_count(&self) -> u64 { + self.write_count.load(Ordering::Relaxed) + } + + /// Elapsed time since the last write, or `None` if no write has been + /// recorded yet. + fn time_since_last_write(&self) -> Option { + if self.has_write.load(Ordering::Acquire) == 0 { + return None; + } + let last = self.last_write_nanos.load(Ordering::Acquire); + let now = self.epoch.elapsed().as_nanos() as u64; + Some(Duration::from_nanos(now.saturating_sub(last))) + } + fn recent_write_count(&self, now_second: u64) -> usize { self.write_buckets .iter() - .filter(|bucket| { - bucket.count > 0 && bucket.second <= now_second && now_second.saturating_sub(bucket.second) < WRITE_WINDOW_SECS + .filter_map(|bucket| { + let (second, count) = bucket.snapshot(); + if count > 0 && second <= now_second && now_second.saturating_sub(second) < WRITE_WINDOW_SECS { + Some(count) + } else { + None + } }) - .map(|bucket| bucket.count) .sum() } - fn record_write(&mut self, now: Instant) -> usize { + fn record_write(&self, _now: Instant) -> usize { let now_second = self.monotonic_second(); let bucket_idx = (now_second % WRITE_WINDOW_BUCKETS as u64) as usize; - let bucket = &mut self.write_buckets[bucket_idx]; + self.write_buckets[bucket_idx].record(now_second); - if bucket.second != now_second { - *bucket = WriteBucket { - second: now_second, - count: 0, - }; - } - - bucket.count = bucket.count.saturating_add(1); - self.last_write_time = Some(now); - self.write_count = self.write_count.saturating_add(1); + self.last_write_nanos + .store(self.epoch.elapsed().as_nanos() as u64, Ordering::Release); + self.has_write.store(1, Ordering::Release); + // Single atomic add (wraps only after 2^64 writes; effectively saturating + // for any real deployment). Observability only — the frequency window + // used for refresh decisions is tracked per-bucket above. + self.write_count.fetch_add(1, Ordering::Relaxed); self.recent_write_count(now_second) } @@ -641,8 +720,9 @@ impl Drop for RefreshLeaderGuard { pub struct HybridCapacityManager { /// Capacity cache cache: Arc>>, - /// Write record - write_record: Arc>, + /// Write record. Lock-free atomics (backlog#1315): the PUT response path no + /// longer takes an async writer lock to record a write. + write_record: Arc, /// Dirty disks recorded from write-side scope propagation, keyed to the /// instant they were last marked so a commit only clears marks that /// predate its scan (backlog#1020 S19). @@ -680,7 +760,7 @@ impl HybridCapacityManager { pub fn new(config: HybridStrategyConfig) -> Self { Self { cache: Arc::new(RwLock::new(None)), - write_record: Arc::new(RwLock::new(WriteRecord::new())), + write_record: Arc::new(WriteRecord::new()), dirty_disks: Arc::new(RwLock::new(HashMap::new())), disk_cache: Arc::new(RwLock::new(HashMap::new())), disk_cache_complete: Arc::new(RwLock::new(false)), @@ -815,9 +895,8 @@ impl HybridCapacityManager { /// Record write operation pub async fn record_write_operation(&self) { - let mut record = self.write_record.write().await; let now = Instant::now(); - let recent_write_count = record.record_write(now); + let recent_write_count = self.write_record.record_write(now); record_capacity_write_operation(recent_write_count); debug!( @@ -825,7 +904,7 @@ impl HybridCapacityManager { component = LOG_COMPONENT_CAPACITY, subsystem = LOG_SUBSYSTEM_REFRESH, state = "recorded", - total_writes = record.write_count, + total_writes = self.write_record.total_write_count(), recent_writes = recent_write_count, "capacity refresh write recorded" ); @@ -873,15 +952,13 @@ impl HybridCapacityManager { return false; } - let write_record = self.write_record.read().await; + let write_record = &self.write_record; let write_frequency = write_record.recent_write_count(write_record.monotonic_second()); if write_frequency <= self.config.write_frequency_threshold { return false; } - if let Some(last_write_time) = write_record.last_write_time { - let time_since_write = last_write_time.elapsed(); - + if let Some(time_since_write) = write_record.time_since_last_write() { if time_since_write < self.config.write_trigger_delay { debug!( event = EVENT_CAPACITY_REFRESH_DEBOUNCE_STATE, @@ -923,7 +1000,7 @@ impl HybridCapacityManager { /// Get write frequency (writes/minute) #[allow(dead_code)] pub async fn get_write_frequency(&self) -> usize { - let record = self.write_record.read().await; + let record = &self.write_record; record.recent_write_count(record.monotonic_second()) } @@ -1465,15 +1542,9 @@ mod tests { #[test] #[serial] fn test_recent_write_count_ignores_future_buckets() { - let mut record = WriteRecord { - last_write_time: None, - write_count: 1, - write_buckets: [WriteBucket::default(); WRITE_WINDOW_BUCKETS], - epoch: Instant::now(), - }; - - record.write_buckets[0] = WriteBucket { second: 120, count: 3 }; - record.write_buckets[1] = WriteBucket { second: 90, count: 2 }; + let record = WriteRecord::new(); + record.write_buckets[0].store(120, 3); + record.write_buckets[1].store(90, 2); assert_eq!( record.recent_write_count(100), @@ -1616,6 +1687,35 @@ mod tests { assert_eq!(manager.get_write_frequency().await, 10); } + // backlog#1315: the lock-free write record must not drop concurrent writes. + // The previous async RwLock serialized every write; the CAS bucket must be + // exact under heavy same-second contention or the frequency window (and the + // write-trigger decision) would undercount. + #[tokio::test(flavor = "multi_thread", worker_threads = 8)] + #[serial] + async fn test_record_write_operation_lock_free_is_exact_under_contention() { + let manager = Arc::new(HybridCapacityManager::from_env()); + let mut handles = Vec::new(); + const WRITERS: usize = 16; + const PER_WRITER: usize = 64; + + for _ in 0..WRITERS { + let mgr = manager.clone(); + handles.push(tokio::spawn(async move { + for _ in 0..PER_WRITER { + mgr.record_write_operation().await; + } + })); + } + for handle in handles { + handle.await.unwrap(); + } + + // All writes land in the same monotonic second (the test runs well under + // one second), so the recent-window frequency must equal the exact total. + assert_eq!(manager.get_write_frequency().await, WRITERS * PER_WRITER); + } + #[tokio::test] #[serial] async fn test_performance_overhead() { diff --git a/crates/object-capacity/src/capacity_scope.rs b/crates/object-capacity/src/capacity_scope.rs index e6d1c9da8..e5269cfa1 100644 --- a/crates/object-capacity/src/capacity_scope.rs +++ b/crates/object-capacity/src/capacity_scope.rs @@ -13,6 +13,7 @@ // limitations under the License. use std::collections::{HashMap, HashSet}; +use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Mutex, OnceLock}; use std::time::{Duration, Instant}; use uuid::Uuid; @@ -48,6 +49,36 @@ fn global_dirty_scope_registry() -> &'static Mutex> { REGISTRY.get_or_init(|| Mutex::new(HashSet::new())) } +/// Monotonic generation of the global dirty-scope registry. +/// +/// Advanced every time a non-empty drain removes disks from the registry. A +/// storage-set that recorded its disks at generation `g` can safely skip the +/// registry mutex on subsequent writes as long as the generation is still `g`: +/// any drain that could have removed its disks would have advanced the counter, +/// forcing the set to re-mark (backlog#1315). The generation is loaded and +/// advanced only while the registry mutex is held, so a set that observes +/// `generation == g` at record time is guaranteed its disks are still present +/// until the next drain. +static DIRTY_GENERATION: AtomicU64 = AtomicU64::new(0); + +/// Test-only counter of how many times the global registry mutex was upgraded +/// to insert dirty disks. Steady-state writes must reuse an existing generation +/// mark and leave this untouched; only the first write of each generation +/// bumps it. Used by white-box tests to prove the per-generation skip holds and +/// to fail closed if the optimization regresses (backlog#1315). +static GLOBAL_DIRTY_UPGRADE_COUNT: AtomicU64 = AtomicU64::new(0); + +/// Current global dirty-scope generation. See [`DIRTY_GENERATION`]. +pub fn current_dirty_generation() -> u64 { + DIRTY_GENERATION.load(Ordering::Acquire) +} + +/// Number of times the global dirty registry mutex was upgraded to record +/// disks. Test/observability hook (backlog#1315). +pub fn global_dirty_upgrade_count() -> u64 { + GLOBAL_DIRTY_UPGRADE_COUNT.load(Ordering::Relaxed) +} + fn prune_expired_entries(entries: &mut HashMap, now: Instant) { entries.retain(|_, entry| now.duration_since(entry.recorded_at) <= CAPACITY_SCOPE_TTL); } @@ -107,17 +138,43 @@ pub fn take_capacity_scope(token: Uuid) -> Option { Some(entry.scope) } -pub fn record_global_dirty_scope(scope: CapacityScope) { +/// Record dirty disks in the global registry, returning the registry generation +/// observed while the mutex was held. +/// +/// Callers cache the returned generation and, on subsequent writes, compare it +/// against [`current_dirty_generation`]: while it is unchanged the disks are +/// still queued for the next drain and the registry mutex can be skipped +/// entirely (backlog#1315). An empty scope is a no-op that returns the current +/// generation without touching the mutex. +pub fn record_global_dirty_scope(scope: CapacityScope) -> u64 { if scope.disks.is_empty() { - return; + return current_dirty_generation(); } let mut dirty_scopes = global_dirty_scope_registry().lock().unwrap_or_else(|p| p.into_inner()); dirty_scopes.extend(scope.disks); + GLOBAL_DIRTY_UPGRADE_COUNT.fetch_add(1, Ordering::Relaxed); + // Load under the registry lock so the value is coherent with any concurrent + // drain (which advances the generation under the same lock). A set that + // stores this value only re-marks once a later drain moves past it. + DIRTY_GENERATION.load(Ordering::Acquire) } pub fn drain_global_dirty_scopes() -> Vec { let mut dirty_scopes = global_dirty_scope_registry().lock().unwrap_or_else(|p| p.into_inner()); + if dirty_scopes.is_empty() { + // Nothing to drain: leave the generation untouched so sets that already + // marked keep their skip fast-path. Advancing here would only force + // redundant re-marks without changing correctness. + return Vec::new(); + } + // Advance before draining so the new generation is visible under the lock; + // any set that recorded at the old generation will observe the change and + // re-mark on its next write, and the disks it just wrote are read by the + // refresh that consumes this drain (the write commits before it records the + // scope, and this bump is ordered after that record in the registry's + // modification order). + DIRTY_GENERATION.fetch_add(1, Ordering::AcqRel); dirty_scopes.drain().collect() } @@ -141,6 +198,8 @@ mod tests { .lock() .unwrap_or_else(|poisoned| poisoned.into_inner()) .clear(); + DIRTY_GENERATION.store(0, Ordering::Release); + GLOBAL_DIRTY_UPGRADE_COUNT.store(0, Ordering::Release); } fn poison_capacity_scope_registry_for_test() { @@ -329,6 +388,65 @@ mod tests { clear_capacity_scope_registry_for_test(); } + #[test] + fn record_global_dirty_scope_generation_is_stable_until_drain() { + let _guard = test_lock().lock().expect("test lock poisoned"); + clear_capacity_scope_registry_for_test(); + + let disk = CapacityScopeDisk { + endpoint: "node-a".to_string(), + drive_path: "/tmp/disk-a".to_string(), + }; + let scope = CapacityScope { + disks: vec![disk.clone()], + }; + + // First record upgrades the registry and returns the current generation. + let gen0 = record_global_dirty_scope(scope.clone()); + assert_eq!(gen0, current_dirty_generation()); + assert_eq!(global_dirty_upgrade_count(), 1); + + // A subsequent record while the generation is unchanged still returns + // the same generation; the caller's skip fast-path keys off equality. + let gen1 = record_global_dirty_scope(scope); + assert_eq!(gen1, gen0); + assert_eq!(global_dirty_upgrade_count(), 2); + + // Draining advances the generation so a caller that cached gen0 is + // forced to re-mark on its next write. + let drained = drain_global_dirty_scopes(); + assert_eq!(drained, vec![disk]); + assert_ne!(current_dirty_generation(), gen0); + assert_eq!(current_dirty_generation(), gen0 + 1); + + clear_capacity_scope_registry_for_test(); + } + + #[test] + fn drain_empty_registry_does_not_advance_generation() { + let _guard = test_lock().lock().expect("test lock poisoned"); + clear_capacity_scope_registry_for_test(); + + let before = current_dirty_generation(); + assert!(drain_global_dirty_scopes().is_empty()); + assert_eq!(current_dirty_generation(), before); + + clear_capacity_scope_registry_for_test(); + } + + #[test] + fn record_empty_scope_is_noop_without_upgrade() { + let _guard = test_lock().lock().expect("test lock poisoned"); + clear_capacity_scope_registry_for_test(); + + let observed = record_global_dirty_scope(CapacityScope::default()); + assert_eq!(observed, current_dirty_generation()); + assert_eq!(global_dirty_upgrade_count(), 0); + assert!(drain_global_dirty_scopes().is_empty()); + + clear_capacity_scope_registry_for_test(); + } + #[test] fn record_global_dirty_scope_recovers_from_poisoned_registry() { let _guard = test_lock().lock().expect("test lock poisoned");