diff --git a/crates/lock/src/client/local.rs b/crates/lock/src/client/local.rs index 8c9305e112..6a5fee9482 100644 --- a/crates/lock/src/client/local.rs +++ b/crates/lock/src/client/local.rs @@ -30,19 +30,92 @@ use crate::{ const DEFAULT_GUARD_SHARD_COUNT: usize = 64; const EVENT_NAMESPACE_LOCK_ACQUIRE: &str = "namespace_lock_acquire"; const EVENT_NAMESPACE_LOCK_RELEASE: &str = "namespace_lock_release"; +const EVENT_LOCK_GUARD_RECLAIMED: &str = "lock_guard_reclaimed"; const LOG_COMPONENT_LOCK: &str = "lock"; const LOG_SUBSYSTEM_NAMESPACE: &str = "namespace"; -type GuardMap = HashMap; type GuardShard = Arc; type GuardStorage = Arc>; +#[derive(Debug, Default)] +struct GuardRegistryState { + entries: HashMap, + // A lower bound on every registered lease deadline. Removal and refresh + // may leave it early, so a due sweep recomputes it from the surviving leases. + next_expiry: Option, + #[cfg(test)] + expiry_scans: usize, +} + +impl GuardRegistryState { + fn insert(&mut self, lock_id: LockId, entry: LocalGuardEntry) -> Option { + self.next_expiry = Some( + self.next_expiry + .map_or(entry.deadline, |deadline| deadline.min(entry.deadline)), + ); + self.entries.insert(lock_id, entry) + } + + fn remove(&mut self, lock_id: &LockId) -> Option { + let entry = self.entries.remove(lock_id); + if self.entries.is_empty() { + self.next_expiry = None; + } + entry + } + + fn refresh(&mut self, lock_id: &LockId) -> bool { + let Some(entry) = self.entries.get_mut(lock_id) else { + return false; + }; + if entry.is_expired() { + return false; + } + entry.refresh(); + self.next_expiry = Some( + self.next_expiry + .map_or(entry.deadline, |deadline| deadline.min(entry.deadline)), + ); + true + } + + fn sweep_due(&self) -> bool { + self.next_expiry.is_some_and(|deadline| deadline <= Instant::now()) + } + + fn extract_expired(&mut self) -> Vec { + if !self.sweep_due() { + return Vec::new(); + } + #[cfg(test)] + { + self.expiry_scans += 1; + } + let now = Instant::now(); + let mut next_expiry: Option = None; + let expired = self + .entries + .extract_if(|_, entry| { + if entry.deadline <= now { + true + } else { + next_expiry = Some(next_expiry.map_or(entry.deadline, |deadline| deadline.min(entry.deadline))); + false + } + }) + .map(|(_, entry)| entry) + .collect(); + self.next_expiry = next_expiry; + expired + } +} + /// Registry access never reserves the table for a task that has not resumed. /// Only short synchronous map operations hold the native lock; contention /// waits on a notification without blocking an executor thread. #[derive(Debug, Default)] struct GuardRegistry { - entries: RwLock, + entries: RwLock, changed: Notify, } @@ -83,7 +156,7 @@ impl GuardRegistry { } } - async fn read(&self) -> GuardRegistryAccess<'_, RwLockReadGuard<'_, GuardMap>> { + async fn read(&self) -> GuardRegistryAccess<'_, RwLockReadGuard<'_, GuardRegistryState>> { tokio::task::consume_budget().await; if let Some(guard) = self.entries.try_read() { return self.access(guard); @@ -100,7 +173,7 @@ impl GuardRegistry { } } - async fn write(&self) -> GuardRegistryAccess<'_, RwLockWriteGuard<'_, GuardMap>> { + async fn write(&self) -> GuardRegistryAccess<'_, RwLockWriteGuard<'_, GuardRegistryState>> { tokio::task::consume_budget().await; if let Some(guard) = self.entries.try_write() { return self.access(guard); @@ -284,8 +357,8 @@ impl LocalClient { replaced_entry, registration_wait, mutation_started.elapsed(), - guards.len(), - guards.capacity(), + guards.entries.len(), + guards.entries.capacity(), ) }; // Repeated shared requests can replace an entry. Releasing @@ -415,8 +488,8 @@ impl LocalClient { self.manager.clone().unwrap_or_else(crate::get_global_lock_manager) } - /// Guard entries for one resource share a shard so expiration checks do not - /// need to take every shard's write lock on the acquisition hot path. + /// Guard entries for one resource share a shard so acquisition only checks + /// that shard's expiration deadline. fn get_resource_shard_index(&self, resource: &crate::ObjectKey) -> usize { resource.shard_index(self.shard_mask) } @@ -429,46 +502,32 @@ impl LocalClient { async fn reclaim_expired_guards_for_resource(&self, resource: &crate::ObjectKey) -> usize { let shard = &self.guard_storage[self.get_resource_shard_index(resource)]; - let has_expired = { - let guards = shard.read().await; - guards - .iter() - .any(|(lock_id, entry)| &lock_id.resource == resource && entry.is_expired()) - }; - if !has_expired { + if !shard.read().await.sweep_due() { return 0; } let expired_entries = { let mut guards = shard.write().await; - guards - .extract_if(|lock_id, entry| &lock_id.resource == resource && entry.is_expired()) - .map(|(_, entry)| entry) - .collect::>() + guards.extract_expired() }; - Self::release_reclaimed_guards(expired_entries, Some(resource)) + let reclaimed_for_resource = expired_entries.iter().filter(|entry| entry.guard.key() == resource).count(); + Self::release_reclaimed_guards(expired_entries); + reclaimed_for_resource } async fn extract_expired_guards(storage: &GuardStorage) -> Vec { let mut expired_entries = Vec::new(); for shard in storage.iter() { - let has_expired = { - let guards = shard.read().await; - guards.values().any(LocalGuardEntry::is_expired) - }; - if !has_expired { + if !shard.read().await.sweep_due() { continue; } let mut guards = shard.write().await; - expired_entries.extend(guards.extract_if(|_, entry| entry.is_expired()).map(|(_, entry)| entry)); + expired_entries.extend(guards.extract_expired()); } expired_entries } - fn release_reclaimed_guards( - entries: impl IntoIterator, - resource: Option<&crate::ObjectKey>, - ) -> usize { + fn release_reclaimed_guards(entries: impl IntoIterator) -> usize { let mut reclaimed = 0; for mut entry in entries { let _ = entry.guard.release(); @@ -476,11 +535,14 @@ impl LocalClient { reclaimed += 1; } if reclaimed > 0 { - if let Some(resource) = resource { - tracing::debug!(event = "lock_guard_reclaimed", resource = %resource, count = reclaimed, "expired lock guards reclaimed"); - } else { - tracing::debug!(event = "lock_guard_reaper_sweep", count = reclaimed, "expired lock guards reclaimed"); - } + tracing::debug!( + event = EVENT_LOCK_GUARD_RECLAIMED, + component = LOG_COMPONENT_LOCK, + subsystem = LOG_SUBSYSTEM_NAMESPACE, + state = "reclaimed", + count = reclaimed, + "Expired lock guards reclaimed" + ); } reclaimed } @@ -499,7 +561,7 @@ impl LocalClient { break; }; let expired_entries = Self::extract_expired_guards(&storage).await; - Self::release_reclaimed_guards(expired_entries, None); + Self::release_reclaimed_guards(expired_entries); } }); } @@ -585,19 +647,17 @@ impl LockClient for LocalClient { let shard = self.get_shard(lock_id); let expired_entry = { let mut guards = shard.write().await; - let Some(entry) = guards.get_mut(lock_id) else { - return Ok(false); - }; - if entry.is_expired() { - guards.remove(lock_id) - } else { - entry.refresh(); + if guards.refresh(lock_id) { None + } else if let Some(entry) = guards.remove(lock_id) { + Some(entry) + } else { + return Ok(false); } }; if let Some(entry) = expired_entry { - Self::release_reclaimed_guards([entry], Some(&lock_id.resource)); + Self::release_reclaimed_guards([entry]); Ok(false) } else { Ok(true) @@ -611,7 +671,7 @@ impl LockClient for LocalClient { async fn check_status(&self, lock_id: &LockId) -> Result> { let shard = self.get_shard(lock_id); let guards = shard.read().await; - if let Some(entry) = guards.get(lock_id) { + if let Some(entry) = guards.entries.get(lock_id) { // We have an active guard for this lock let lock_type = match entry.guard.mode() { crate::LockMode::Shared => LockType::Shared, @@ -644,8 +704,8 @@ impl LockClient for LocalClient { let mut leases = Vec::new(); for shard in self.guard_storage.iter() { let guards = shard.read().await; - leases.reserve(guards.len()); - leases.extend(guards.iter().map(|(lock_id, entry)| crate::LockLeaseInfo { + leases.reserve(guards.entries.len()); + leases.extend(guards.entries.iter().map(|(lock_id, entry)| crate::LockLeaseInfo { resource: lock_id.resource.clone(), lock_type: match entry.guard.mode() { crate::LockMode::Shared => LockType::Shared, @@ -851,6 +911,139 @@ mod tests { .with_acquire_timeout(Duration::from_millis(80)) } + #[tokio::test(start_paused = true)] + async fn unexpired_shared_acquisitions_and_reaper_do_not_scan_guard_entries() { + let client = LocalClient::with_manager(Arc::new(GlobalLockManager::new())); + client.reaper_started.store(true, Ordering::Release); + let resource = crate::ObjectKey::new("bucket", "unexpired-shared-registry"); + let mut requests = Vec::with_capacity(512); + for _ in 0..512 { + let request = LockRequest::new(resource.clone(), LockType::Shared, "reader"); + assert!(client.acquire_lock(&request).await.expect("register shared lease").success); + requests.push(request); + } + assert!(LocalClient::extract_expired_guards(&client.guard_storage).await.is_empty()); + { + let registry = client.get_shard(&requests[0].lock_id).read().await; + assert_eq!(registry.entries.len(), 512); + assert_eq!(registry.expiry_scans, 0, "live lease acquisition and reaping must use the deadline hint"); + assert!(registry.next_expiry.is_some()); + } + for request in &requests { + assert!(client.release(&request.lock_id).await.expect("release shared lease")); + } + let registry = client.get_shard(&requests[0].lock_id).read().await; + assert!(registry.entries.is_empty()); + assert_eq!(registry.next_expiry, None, "the final release must clear the deadline hint"); + } + + #[tokio::test(start_paused = true)] + async fn same_shard_expiration_sweep_only_counts_the_requested_resource() { + let manager = Arc::new(GlobalLockManager::new()); + let client = LocalClient::with_manager(manager.clone()); + client.reaper_started.store(true, Ordering::Release); + let resource = crate::ObjectKey::new("bucket", "live-sweep-trigger"); + let shard_index = client.get_resource_shard_index(&resource); + let other_resource = (0..4096) + .map(|index| crate::ObjectKey::new("bucket", format!("expired-same-shard-{index}"))) + .find(|candidate| client.get_resource_shard_index(candidate) == shard_index) + .expect("find an independent resource in the same guard shard"); + let live = request(resource.clone(), "live", Duration::from_secs(30)); + let expired = request(other_resource.clone(), "expired", Duration::from_secs(5)); + assert!(client.acquire_lock(&live).await.expect("register live lease").success); + assert!(client.acquire_lock(&expired).await.expect("register earlier lease").success); + tokio::time::advance(Duration::from_secs(5)).await; + + assert_eq!(client.reclaim_expired_guards_for_resource(&resource).await, 0); + assert!( + client + .check_status(&expired.lock_id) + .await + .expect("expired lease status") + .is_none() + ); + assert!(client.check_status(&live.lock_id).await.expect("live lease status").is_some()); + let replacement = manager + .acquire_lock(crate::ObjectLockRequest::new_write(other_resource, "replacement")) + .await + .expect("the sweep must release the other resource's expired physical guard"); + drop(replacement); + assert_eq!(client.reclaim_expired_guards_for_resource(&resource).await, 0); + assert_eq!(client.get_shard(&live.lock_id).read().await.expiry_scans, 1); + assert!(client.release(&live.lock_id).await.expect("release live lease")); + } + + #[tokio::test(start_paused = true)] + async fn refreshed_and_removed_earliest_leases_only_trigger_one_sweep_per_deadline() { + let client = LocalClient::with_manager(Arc::new(GlobalLockManager::new())); + client.reaper_started.store(true, Ordering::Release); + let resource = crate::ObjectKey::new("bucket", "refresh-expiry-hint"); + let short = LockRequest::new(resource.clone(), LockType::Shared, "short").with_ttl(Duration::from_secs(10)); + let long = LockRequest::new(resource.clone(), LockType::Shared, "long").with_ttl(Duration::from_secs(30)); + assert!(client.acquire_lock(&short).await.expect("short lease").success); + assert!(client.acquire_lock(&long).await.expect("long lease").success); + tokio::time::advance(Duration::from_secs(5)).await; + assert!(client.refresh(&short.lock_id).await.expect("extend the earliest deadline")); + tokio::time::advance(Duration::from_secs(5)).await; + assert_eq!(client.reclaim_expired_guards_for_resource(&resource).await, 0); + assert_eq!(client.reclaim_expired_guards_for_resource(&resource).await, 0); + assert_eq!(client.get_shard(&short.lock_id).read().await.expiry_scans, 1); + + assert!( + client + .release(&short.lock_id) + .await + .expect("remove the earliest surviving lease") + ); + tokio::time::advance(Duration::from_secs(5)).await; + assert_eq!(client.reclaim_expired_guards_for_resource(&resource).await, 0); + assert_eq!(client.reclaim_expired_guards_for_resource(&resource).await, 0); + assert_eq!(client.get_shard(&long.lock_id).read().await.expiry_scans, 2); + + tokio::time::advance(Duration::from_secs(15)).await; + assert_eq!(client.reclaim_expired_guards_for_resource(&resource).await, 1); + let registry = client.get_shard(&long.lock_id).read().await; + assert_eq!(registry.expiry_scans, 3); + assert!(registry.entries.is_empty()); + assert_eq!(registry.next_expiry, None); + } + + #[tokio::test(start_paused = true)] + async fn repeated_shared_id_replacement_preserves_earlier_expiration_and_reaper_cleanup() { + let manager = Arc::new(GlobalLockManager::new()); + let client = LocalClient::with_manager(manager.clone()); + client.reaper_started.store(true, Ordering::Release); + let resource = crate::ObjectKey::new("bucket", "replace-expiry-hint"); + let original = LockRequest::new(resource.clone(), LockType::Shared, "owner").with_ttl(Duration::from_secs(30)); + assert!(client.acquire_lock(&original).await.expect("register original lease").success); + let replacement = original.clone().with_ttl(Duration::from_secs(5)); + assert!( + client + .acquire_lock(&replacement) + .await + .expect("replace the same grant identity") + .success + ); + assert_eq!(client.get_shard(&original.lock_id).read().await.entries.len(), 1); + let backend = manager.as_fast_lock_manager().expect("enabled physical lock backend"); + assert_eq!(backend.get_shard(&resource).active_guard_count(), 1); + + tokio::time::advance(Duration::from_secs(5)).await; + let expired = LocalClient::extract_expired_guards(&client.guard_storage).await; + assert_eq!(LocalClient::release_reclaimed_guards(expired), 1); + assert_eq!(backend.get_shard(&resource).active_guard_count(), 0); + assert!( + client + .check_status(&original.lock_id) + .await + .expect("replaced lease status") + .is_none() + ); + let registry = client.get_shard(&original.lock_id).read().await; + assert_eq!(registry.expiry_scans, 1); + assert_eq!(registry.next_expiry, None); + } + fn hold_guard_shard(shard: GuardShard) -> (std::sync::mpsc::Sender<()>, std::thread::JoinHandle<()>) { let (entered_tx, entered_rx) = std::sync::mpsc::sync_channel(0); let (release_tx, release_rx) = std::sync::mpsc::channel(); @@ -1037,7 +1230,7 @@ mod tests { let access = timeout(Duration::from_millis(100), survivor) .await .expect("a cancelled waiter must not consume another waiter's unlock notification"); - assert!(access.is_empty()); + assert!(access.entries.is_empty()); } #[tokio::test] diff --git a/crates/lock/src/fast_lock/state.rs b/crates/lock/src/fast_lock/state.rs index 4dbc423728..c55f75cc13 100644 --- a/crates/lock/src/fast_lock/state.rs +++ b/crates/lock/src/fast_lock/state.rs @@ -21,10 +21,14 @@ use crate::fast_lock::optimized_notify::OptimizedNotify; use crate::fast_lock::types::{LockMode, LockPriority}; /// Optimized atomic lock state encoding in u64 -/// Bits: [63:48] reserved | [47:32] writers_waiting | [31:16] readers_waiting | [15:8] readers_count | [7:1] flags | [0] writer_flag +/// Bits: [63:48] readers_count | [47:32] writers_waiting | [31:16] readers_waiting | [15:1] reserved | [0] writer_flag +/// +/// The reader count is 16 bits wide because distributed bucket-scoped read locks (for example the +/// per-bucket lifecycle and metadata-transaction fences) are held by every in-flight request into that +/// bucket on every node, so a single key routinely has hundreds of concurrent holders. const WRITER_FLAG_MASK: u64 = 0x1; -const READERS_SHIFT: u8 = 8; -const READERS_MASK: u64 = 0xFF << READERS_SHIFT; // Support up to 255 concurrent readers +const READERS_SHIFT: u8 = 48; +const READERS_MASK: u64 = 0xFFFF << READERS_SHIFT; const READERS_WAITING_SHIFT: u8 = 16; const READERS_WAITING_MASK: u64 = 0xFFFF << READERS_WAITING_SHIFT; const WRITERS_WAITING_SHIFT: u8 = 32; @@ -96,9 +100,7 @@ impl AtomicLockState { return false; } - let readers = self.readers_count(current); - if readers == 0xFF { - // Updated limit to 255 + if self.readers_count(current) == u16::MAX { return false; // Too many readers } @@ -303,8 +305,8 @@ impl AtomicLockState { self.last_accessed.store(now, Ordering::Relaxed); } - fn readers_count(&self, state: u64) -> u8 { - ((state & READERS_MASK) >> READERS_SHIFT) as u8 + fn readers_count(&self, state: u64) -> u16 { + ((state & READERS_MASK) >> READERS_SHIFT) as u16 } fn readers_waiting(&self, state: u64) -> u16 { @@ -622,6 +624,50 @@ mod tests { assert!(state.is_free()); } + // Bucket-scoped fences are read-locked by every in-flight request into the bucket, so one key + // needs far more than 255 concurrent shared holders. The old 8-bit reader count refused the + // 256th reader outright. + #[test] + fn shared_lock_admits_more_than_255_holders() { + let state = ObjectLockState::new(); + let timeout = Duration::from_secs(30); + let owners: Vec> = (0..1024).map(|i| Arc::from(format!("owner-{i}"))).collect(); + + for owner in &owners { + assert!( + state.try_acquire_shared_fast(owner, timeout), + "shared acquisition for {owner} must succeed" + ); + } + assert_eq!(state.atomic_state.holder_and_waiter_counts(), (1024, 0)); + assert!(!state.try_acquire_exclusive_fast(&Arc::from("writer"), timeout)); + + for owner in &owners { + assert!(state.release_shared(owner), "shared release for {owner} must succeed"); + } + assert!(state.try_acquire_exclusive_fast(&Arc::from("writer"), timeout)); + } + + #[test] + fn shared_lock_refuses_readers_beyond_counter_capacity() { + let state = AtomicLockState::new(); + assert!(state.inc_readers_waiting()); + + for _ in 0..u16::MAX { + assert!(state.try_acquire_shared()); + } + assert!(!state.try_acquire_shared(), "the reader count must not overflow"); + assert_eq!(state.holder_and_waiter_counts(), (u64::from(u16::MAX), 1)); + assert_eq!( + state.readers_waiting_count(), + 1, + "a full reader count must not spill into the waiter counters" + ); + + assert!(state.release_shared()); + assert!(state.try_acquire_shared(), "a released slot must be reusable"); + } + #[test] fn test_object_lock_state() { let state = ObjectLockState::new(); diff --git a/crates/lock/src/namespace/tests.rs b/crates/lock/src/namespace/tests.rs index b584f7e239..67e9e9a260 100644 --- a/crates/lock/src/namespace/tests.rs +++ b/crates/lock/src/namespace/tests.rs @@ -300,6 +300,43 @@ fn create_test_object_key(bucket: &str, object: &str) -> ObjectKey { } } +#[tokio::test(start_paused = true)] +async fn distributed_namespace_shared_readers_exceed_255() { + let managers = (0..3).map(|_| Arc::new(GlobalLockManager::new())).collect::>(); + let clients = managers + .iter() + .map(|manager| Arc::new(LocalClient::with_manager(Arc::clone(manager))) as Arc) + .collect(); + let lock = NamespaceLock::with_clients("shared-reader-capacity".to_string(), clients); + let resource = create_test_object_key(".rustfs.sys", "bucket-targets/bucket/transaction.lock"); + let mut guards = Vec::with_capacity(512); + + for reader in 0..512 { + let guard = lock + .get_read_lock(resource.clone(), &format!("reader-{reader}"), Duration::from_secs(1)) + .await + .unwrap_or_else(|error| panic!("shared reader {reader} must reach quorum: {error}")); + guards.push(guard); + } + assert_eq!(guards.len(), 512); + assert!(guards.iter().all(|guard| !guard.is_lock_lost())); + let contended_writer = lock + .lock_guard(resource.clone(), "blocked-writer", Duration::from_millis(100), Duration::from_secs(30)) + .await + .expect("writer contention must preserve the acquisition result"); + assert!( + contended_writer.is_none(), + "a writer must remain excluded while shared readers hold quorum" + ); + + drop(guards); + wait_until_all_managers_can_write(&managers, resource.clone()).await; + let _writer = lock + .get_write_lock(resource, "writer-after-readers", Duration::from_secs(1)) + .await + .expect("all reader grants must drain before the next writer acquires quorum"); +} + async fn wait_until_all_managers_can_write(managers: &[Arc], resource: ObjectKey) { let deadline = tokio::time::Instant::now() + Duration::from_secs(2);