lean_ctx/server/
bounded_lock.rs

1use std::sync::Arc;
2use std::time::Duration;
3use tokio::sync::{OwnedRwLockReadGuard, OwnedRwLockWriteGuard, RwLock};
4
5const BASE_READ_TIMEOUT: Duration = Duration::from_secs(10);
6const BASE_WRITE_TIMEOUT: Duration = Duration::from_secs(10);
7
8/// Spin interval between `try_*` attempts. Kept short enough for responsiveness
9/// but long enough to avoid busy-spinning on Windows where thread scheduling
10/// quanta are ~15ms.
11const SPIN_INTERVAL: Duration = Duration::from_millis(20);
12
13/// Acquire a read lock via a non-blocking spin loop with an adaptive timeout.
14///
15/// Unlike the previous `Handle::block_on` approach, this never parks a blocking
16/// thread waiting for the async runtime to make progress — eliminating the
17/// stall-under-load anti-pattern on Windows (#1018). The loop yields the thread
18/// between attempts so it does not starve other work on the blocking pool.
19///
20/// Returns `None` on timeout (caller must provide graceful fallback).
21pub fn read<T: Send + Sync + 'static>(
22    lock: &Arc<RwLock<T>>,
23    context: &str,
24) -> Option<OwnedRwLockReadGuard<T>> {
25    let timeout = crate::core::io_health::adaptive_timeout(BASE_READ_TIMEOUT);
26    let deadline = std::time::Instant::now() + timeout;
27
28    loop {
29        if let Ok(guard) = lock.clone().try_read_owned() {
30            return Some(guard);
31        }
32        if std::time::Instant::now() >= deadline {
33            crate::core::io_health::record_freeze();
34            tracing::warn!(
35                "bounded_lock: read timeout ({}ms) for {context}; degrading gracefully",
36                timeout.as_millis()
37            );
38            return None;
39        }
40        std::thread::sleep(SPIN_INTERVAL);
41    }
42}
43
44/// Acquire a write lock via a non-blocking spin loop with an adaptive timeout.
45/// See `read()` for design rationale (#1018).
46///
47/// Returns `None` on timeout (caller must provide graceful fallback).
48pub fn write<T: Send + Sync + 'static>(
49    lock: &Arc<RwLock<T>>,
50    context: &str,
51) -> Option<OwnedRwLockWriteGuard<T>> {
52    let timeout = crate::core::io_health::adaptive_timeout(BASE_WRITE_TIMEOUT);
53    let deadline = std::time::Instant::now() + timeout;
54
55    loop {
56        if let Ok(guard) = lock.clone().try_write_owned() {
57            return Some(guard);
58        }
59        if std::time::Instant::now() >= deadline {
60            crate::core::io_health::record_freeze();
61            tracing::warn!(
62                "bounded_lock: write timeout ({}ms) for {context}; degrading gracefully",
63                timeout.as_millis()
64            );
65            return None;
66        }
67        std::thread::sleep(SPIN_INTERVAL);
68    }
69}
70
71#[cfg(test)]
72mod tests {
73    use super::*;
74
75    #[test]
76    fn read_succeeds_on_uncontested_lock() {
77        let lock = Arc::new(RwLock::new(42u32));
78        let guard = read(&lock, "test").expect("uncontested read must succeed");
79        assert_eq!(*guard, 42);
80    }
81
82    #[test]
83    fn write_succeeds_on_uncontested_lock() {
84        let lock = Arc::new(RwLock::new(0u32));
85        let mut guard = write(&lock, "test").expect("uncontested write must succeed");
86        *guard = 7;
87        assert_eq!(*guard, 7);
88    }
89
90    #[test]
91    fn multiple_readers_concurrent() {
92        let lock = Arc::new(RwLock::new(99u32));
93        let g1 = read(&lock, "r1").expect("first reader");
94        let g2 = read(&lock, "r2").expect("second reader");
95        assert_eq!(*g1, 99);
96        assert_eq!(*g2, 99);
97    }
98
99    #[test]
100    fn write_excludes_readers() {
101        let lock = Arc::new(RwLock::new(0u32));
102        let _hold = lock.clone().try_write_owned().unwrap();
103        assert!(lock.clone().try_read_owned().is_err());
104    }
105}
lean_ctx/server/bounded_lock.rs

lean_ctx/server/
bounded_lock.rs
