From acce8b22536aa6ade42514b5f42a000cba24cf7a Mon Sep 17 00:00:00 2001 From: Miguel Amador Date: Mon, 3 Aug 2026 15:08:48 +0100 Subject: [PATCH] fix(lock): let waiters hear releases and let acquisition succeed past registered waiters (#5670) * fix(lock): let waiters hear releases and let acquisition succeed past registered waiters Same-key write contention scaled superlinearly with writer count: 8 concurrent conditional PUTs on one key cost ~340-460 ms, 16 cost ~700 ms, 32 cost ~5 s, against ~4 ms per uncontended write and ~10 ms actual lock holds (measured via RUSTFS_OBJECT_LOCK_DIAG at 1 ms thresholds). Outcomes were always correct; the cost was pure waiting. Two coupled defects in fast_lock caused it: 1. The slow path's early retries slept without subscribing to anything. notify_writer()/notify_readers() are gated on the waiter counters, which a sleeper never increments, so a release during the backoff woke nobody. The lock sat free while every loser slept out its full backoff, and the ladder compounded: successive acquires landed at the cumulative ladder offsets (10+20+40+80+100... ms). 2. try_acquire_exclusive demanded the entire packed state word be zero, including the readers_waiting/writers_waiting counter bits. A lock with registered waiters could be acquired by no one - including the waiters themselves, each blocked by the others' registration - so contended acquisition only succeeded in windows where every waiter happened to be unregistered. This is also why (1) could not be fixed by simply registering the sleepers: registration alone deadlocks acquisition until the acquire deadline. try_acquire_shared already masks correctly and preserves the counter bits in its CAS; the exclusive path now mirrors it. The fix: mask the acquisition CAS to ownership bits only (writer flag, active readers), and turn the early-retry sleep into a notification wait bounded by the same backoff, so a release wakes a waiter immediately while the bound still protects against lost or stolen wakeups exactly as NOTIFY_WAIT_CAP does for the post-retry wait. With both changes, 8 concurrent same-key CAS writers resolve in 17-29 ms (was 340-460 ms) and 32 resolve in 20-53 ms (was ~5 s), with per-racer cost now decreasing in N. Outcomes remain exactly one winner, N-1 precondition failures, zero errors at every width. cargo test -p rustfs-lock passes 113/113 at pristine-parity runtime, including test_concurrent_write_lock_contention, which previously only passed because sleepers were invisible to it. * test(lock): pin both halves of the waiter-starvation fix The fix commit touched only production files, so reverting either half left the suite green: test_concurrent_write_lock_contention only waits for five writers to finish and never asserts that acquisition happens before the backoff ladder runs out. Three tests, one per revert: * exclusive_acquisition_ignores_registered_waiters (state.rs) - a free lock with registered waiters must be acquirable, and the CAS must preserve the counters. Fails against the all-zero `expected`. * early_retry_registers_as_waiter (shard.rs) - a waiter in the early-retry backoff must appear in the writer waiter count within the ~750ms early-retry phase, since notify_writer/notify_readers are gated on those counters. Fails against a bare `sleep`, which registers nowhere. * contended_writers_drain_promptly_after_release (tests.rs) - 16 same-key writers, all registered behind one holder, must drain within 1s of the release rather than sit out their 5s acquire deadlines. Fails against the all-zero `expected` end to end. Wakeup latency is deliberately not asserted anywhere. NOTIFY_POOL is a process-global of 128 Notify slots shared by every lock, so a waiter in a concurrently-running test can consume another's notify_one and push it to the end of its rung: a 24-key latency probe measured ~150us in isolation and ~92ms - a full unexpired rung - alongside the existing 64-key missed-wakeup test. That is the stolen wakeup NOTIFY_WAIT_CAP already exists to bound, and it makes any in-suite latency budget flaky. cargo test -p rustfs-lock: 116/116. Signed-off-by: Miguel Amador --------- Signed-off-by: Miguel Amador --- crates/lock/src/fast_lock/shard.rs | 86 +++++++++++++++++++++++++++++- crates/lock/src/fast_lock/state.rs | 69 +++++++++++++++++++++--- crates/lock/src/fast_lock/tests.rs | 63 +++++++++++++++++++++- 3 files changed, 209 insertions(+), 9 deletions(-) diff --git a/crates/lock/src/fast_lock/shard.rs b/crates/lock/src/fast_lock/shard.rs index 769075063..a0ca7eaa7 100644 --- a/crates/lock/src/fast_lock/shard.rs +++ b/crates/lock/src/fast_lock/shard.rs @@ -215,12 +215,30 @@ impl LockShard { let remaining = deadline - Instant::now(); if retry_count < MAX_RETRIES && remaining > Duration::from_millis(10) { - // For early retries, use a brief exponential backoff instead of full notification wait + // For early retries, wait for a release notification bounded by + // an exponential backoff. The bound (not a bare sleep) matters: + // a plain `sleep` subscribes to nothing, so a release during + // the backoff wakes nobody — `notify_writer`/`notify_readers` + // are gated on the waiter counters, which a sleeper never + // increments. Under N-writer same-key contention the lock sits + // free while every loser sleeps out its full backoff, and the + // ladder compounds superlinearly with N. The backoff cap still + // protects against lost/stolen wakeups, exactly as + // NOTIFY_WAIT_CAP does for the post-retry wait below. let backoff_ms = std::cmp::min(10 << retry_count, 100); // 10ms, 20ms, 40ms, 80ms, 100ms max let backoff_duration = Duration::from_millis(backoff_ms); if backoff_duration < remaining { - tokio::time::sleep(backoff_duration).await; + match request.mode { + LockMode::Shared => { + let _waiter_guard = WaiterCounterGuard::new(state.clone(), LockMode::Shared); + let _ = timeout(backoff_duration, state.optimized_notify.wait_for_read()).await; + } + LockMode::Exclusive => { + let _waiter_guard = WaiterCounterGuard::new(state.clone(), LockMode::Exclusive); + let _ = timeout(backoff_duration, state.optimized_notify.wait_for_write()).await; + } + } retry_count += 1; continue; } @@ -853,6 +871,70 @@ mod tests { assert!(shard.release_lock(&key, &owner1, LockMode::Exclusive)); } + // Regression for the waiter-preserving early retry (rustfs#5657). + // + // The early retries used a bare `sleep`, which subscribes to nothing. + // `notify_writer`/`notify_readers` are gated on the waiter counters, so a + // sleeping waiter is invisible to every release: the lock sits free while + // each loser sleeps out its full 10/20/40/80/100ms rung, and under N-writer + // same-key contention that ladder — not the hold — is what the wait costs. + // Registration is what lets a release reach the waiter at all; the wakeup + // itself is covered by `write_lock_waiter_is_not_stranded_by_missed_wakeup`. + // + // MAX_RETRIES rungs total ~750ms, so the window sampled here sits entirely + // inside the early-retry phase, where a sleeping waiter registers nowhere. + // + // Wakeup *latency* is deliberately not asserted: NOTIFY_POOL is a global of + // 128 `Notify`s shared by every lock in the process, so a waiter in another + // concurrently-running test can consume this one's `notify_one` and push it + // out to the end of its rung. That is the same stolen wakeup NOTIFY_WAIT_CAP + // exists to bound, and it makes any latency budget flaky in-suite. + #[tokio::test(flavor = "multi_thread")] + async fn early_retry_registers_as_waiter() { + let shard = Arc::new(LockShard::new(0)); + let key = ObjectKey::new("bucket", "object"); + let holder: Arc = Arc::from("holder"); + let waiter: Arc = Arc::from("waiter"); + + let request = |owner: Arc| ObjectLockRequest { + key: key.clone(), + mode: LockMode::Exclusive, + owner, + acquire_timeout: Duration::from_secs(5), + lock_timeout: Duration::from_secs(30), + priority: LockPriority::Normal, + }; + + assert!(shard.acquire_lock(&request(holder.clone())).await.is_ok()); + + let waiter_shard = shard.clone(); + let waiter_request = request(waiter.clone()); + let waiter_task = tokio::spawn(async move { waiter_shard.acquire_lock(&waiter_request).await }); + + let sample_until = Instant::now() + Duration::from_millis(200); + let mut registered = false; + while !registered && Instant::now() < sample_until { + registered = shard + .objects + .read() + .get(&key) + .is_some_and(|state| state.atomic_state.writers_waiting_count() > 0); + tokio::time::sleep(Duration::from_millis(1)).await; + } + + assert!( + registered, + "a waiter in the early-retry backoff must be registered in the writer waiter count, \ + otherwise releases cannot reach it" + ); + + assert!(shard.release_lock(&key, &holder, LockMode::Exclusive)); + waiter_task + .await + .expect("waiter task should not panic") + .expect("waiter must acquire once the holder releases"); + } + #[test] fn test_adaptive_timeout_does_not_exceed_request_acquire_timeout() { let shard = LockShard::new(0); diff --git a/crates/lock/src/fast_lock/state.rs b/crates/lock/src/fast_lock/state.rs index c7cb43115..da6611a58 100644 --- a/crates/lock/src/fast_lock/state.rs +++ b/crates/lock/src/fast_lock/state.rs @@ -110,13 +110,31 @@ impl AtomicLockState { pub fn try_acquire_exclusive(&self) -> bool { self.update_access_time(); - // Must be completely unlocked to acquire exclusive - let expected = 0; - let new_state = WRITER_FLAG_MASK; + loop { + let current = self.state.load(Ordering::Acquire); - self.state - .compare_exchange(expected, new_state, Ordering::AcqRel, Ordering::Relaxed) - .is_ok() + // Only ownership bits may block acquisition: no writer flag, no + // active readers. The waiting counters are preserved, not + // required to be zero — demanding a fully-zero word means a lock + // with registered waiters can be acquired by *no one*, including + // the waiters themselves (each sees the others' registration), + // so contended acquisition only succeeds in windows where every + // waiter happens to be unregistered. `try_acquire_shared` above + // already masks correctly; this mirrors it. + if (current & (WRITER_FLAG_MASK | READERS_MASK)) != 0 { + return false; + } + + let new_state = current | WRITER_FLAG_MASK; + + if self + .state + .compare_exchange_weak(current, new_state, Ordering::AcqRel, Ordering::Relaxed) + .is_ok() + { + return true; + } + } } /// Release shared lock @@ -530,6 +548,45 @@ mod tests { assert!(!state.release_exclusive()); } + // Regression for the waiter-preserving exclusive CAS. + // + // The acquisition CAS used to demand a fully-zero state word, which + // includes the readers_waiting/writers_waiting counters. A free lock with + // registered waiters was then acquirable by *no one* — including the + // waiters themselves, each blocked by the others' registration — so + // contended acquisition only succeeded in windows where every waiter + // happened to be unregistered. Reverting to `expected = 0` must fail here. + #[test] + fn exclusive_acquisition_ignores_registered_waiters() { + let state = AtomicLockState::new(); + + // Waiters register while the lock is held, then the holder releases. + assert!(state.try_acquire_exclusive()); + assert!(state.inc_writers_waiting()); + assert!(state.inc_readers_waiting()); + assert!(state.release_exclusive()); + + // The lock is now free — only the waiting counters are set. + assert!( + state.try_acquire_exclusive(), + "registered waiters must not block acquisition of a free lock" + ); + // ...and the CAS must preserve those counters, not clobber them. + assert_eq!(state.writers_waiting_count(), 1); + assert_eq!(state.readers_waiting_count(), 1); + + assert!(state.release_exclusive()); + state.dec_writers_waiting(); + + // Ownership bits still block, registered waiters or not. + assert!(state.try_acquire_shared()); + assert!(!state.try_acquire_exclusive(), "an active reader must still block"); + assert!(state.release_shared()); + + state.dec_readers_waiting(); + assert!(state.is_free()); + } + #[test] fn test_object_lock_state() { let state = ObjectLockState::new(); diff --git a/crates/lock/src/fast_lock/tests.rs b/crates/lock/src/fast_lock/tests.rs index 62c062927..7a2543a84 100644 --- a/crates/lock/src/fast_lock/tests.rs +++ b/crates/lock/src/fast_lock/tests.rs @@ -18,7 +18,7 @@ mod fast_lock_tests { use crate::fast_lock::types::{LockConfig, LockMode, LockPriority, LockResult, ObjectKey, ObjectLockRequest}; use crate::fast_lock::{DEFAULT_SHARD_COUNT, FastObjectLockManager}; use std::sync::Arc; - use std::time::Duration; + use std::time::{Duration, Instant}; use tokio::time::sleep; /// Helper function to create a test lock manager @@ -470,6 +470,67 @@ mod fast_lock_tests { } } + // Regression for the waiter-preserving exclusive CAS, end to end + // (rustfs#5657 same-key write contention). + // + // `try_acquire_exclusive` used to demand a fully-zero state word, which + // includes the waiting counters. Once the slow path's retries register as + // waiters — as they must, to hear a release — a lock with waiters becomes + // acquirable by no one, each waiter blocked by the others' registration, so + // every waiter here sits out its full acquire deadline instead of draining. + // + // The holder is held long enough for all waiters to be registered before + // the single release. After it they only serialize on each other, holding + // nothing, so they should drain in tens of milliseconds. + #[tokio::test(flavor = "multi_thread")] + async fn contended_writers_drain_promptly_after_release() { + let manager = Arc::new(create_test_manager()); + let key = ObjectKey::new("bucket", "hot-object"); + const WAITERS: usize = 16; + const HOLD: Duration = Duration::from_millis(300); + // Generous next to the ~20ms the fixed path needs for these waiters, and + // far below the 5s acquire deadline a zero-state CAS makes them all + // sit out. + const DRAIN_BUDGET: Duration = Duration::from_millis(1000); + + let mut holder = manager + .acquire_write_lock(key.clone(), "holder") + .await + .expect("holder should acquire immediately"); + + let mut handles = Vec::new(); + for i in 0..WAITERS { + let manager = manager.clone(); + let key = key.clone(); + handles.push(tokio::spawn(async move { + let request = + ObjectLockRequest::new_write(key, format!("waiter-{i}")).with_acquire_timeout(Duration::from_secs(5)); + let mut guard = manager + .acquire_lock(request) + .await + .expect("every waiter must acquire once the holder releases"); + assert!(guard.release()); + })); + } + + // Let every waiter fail its fast path and register in the retry ladder. + sleep(HOLD).await; + + let released_at = Instant::now(); + assert!(holder.release()); + + for handle in handles { + handle.await.expect("waiter task should not panic"); + } + + let drain = released_at.elapsed(); + assert!( + drain < DRAIN_BUDGET, + "{WAITERS} waiters took {drain:?} to drain after the release (budget {DRAIN_BUDGET:?}) - \ + registered waiters are blocking acquisition" + ); + } + #[tokio::test] async fn test_lock_timeout() { let manager = create_test_manager();