//! Futex-backed synchronization (SYS_FUTEX_WAIT / SYS_FUTEX_WAKE). use core::cell::UnsafeCell; use core::ops::{Deref, DerefMut}; use core::sync::atomic::{AtomicU32, Ordering}; use core::time::Duration; use crate::sys::{self, Status}; /// Block while `*word == expected`. `Ok(())` means woken by [`futex_wake`]; /// `Err(WouldBlock)` means the word had already changed; `Err(TimedOut)` /// means the deadline passed. Always re-check your condition afterwards. pub fn futex_wait(word: &AtomicU32, expected: u32, timeout: Option) -> Result<(), Status> { let ns = match timeout { None => sys::FUTEX_WAIT_FOREVER, Some(d) => d.as_nanos().min((sys::FUTEX_WAIT_FOREVER - 1) as u128) as u64, }; let rc = unsafe { sys::syscall3(sys::SYS_FUTEX_WAIT, word.as_ptr() as u64, expected as u64, ns) }; Status::from_result(rc).map(|_| ()) } /// Wake up to `count` waiters on `word`, longest-waiting first. /// Returns how many were woken. pub fn futex_wake(word: &AtomicU32, count: u32) -> u32 { let rc = unsafe { sys::syscall2(sys::SYS_FUTEX_WAKE, word.as_ptr() as u64, count as u64) }; Status::from_result(rc).unwrap_or(0) as u32 } // ==================================================================== // Mutex: Drepper's three-state futex mutex ("Futexes Are Tricky", #3). // 0 = unlocked, 1 = locked, 2 = locked with possible waiters. // ==================================================================== pub struct Mutex { state: AtomicU32, data: UnsafeCell, } unsafe impl Send for Mutex {} unsafe impl Sync for Mutex {} impl Mutex { pub const fn new(value: T) -> Self { Mutex { state: AtomicU32::new(0), data: UnsafeCell::new(value) } } pub fn into_inner(self) -> T { self.data.into_inner() } } impl Mutex { pub fn lock(&self) -> MutexGuard<'_, T> { if self.state.compare_exchange(0, 1, Ordering::Acquire, Ordering::Relaxed).is_err() { self.lock_contended(); } MutexGuard { mutex: self } } pub fn try_lock(&self) -> Option> { self.state .compare_exchange(0, 1, Ordering::Acquire, Ordering::Relaxed) .ok() .map(|_| MutexGuard { mutex: self }) } #[cold] fn lock_contended(&self) { // Brief spin first: most critical sections are short. for _ in 0..100 { if self.state.load(Ordering::Relaxed) == 0 && self.state.compare_exchange(0, 1, Ordering::Acquire, Ordering::Relaxed).is_ok() { return; } core::hint::spin_loop(); } while self.state.swap(2, Ordering::Acquire) != 0 { let _ = futex_wait(&self.state, 2, None); } } fn unlock(&self) { if self.state.swap(0, Ordering::Release) == 2 { futex_wake(&self.state, 1); } } pub fn get_mut(&mut self) -> &mut T { self.data.get_mut() } } impl Default for Mutex { fn default() -> Self { Mutex::new(T::default()) } } pub struct MutexGuard<'a, T: ?Sized> { mutex: &'a Mutex, } unsafe impl Sync for MutexGuard<'_, T> {} impl Deref for MutexGuard<'_, T> { type Target = T; fn deref(&self) -> &T { unsafe { &*self.mutex.data.get() } } } impl DerefMut for MutexGuard<'_, T> { fn deref_mut(&mut self) -> &mut T { unsafe { &mut *self.mutex.data.get() } } } impl Drop for MutexGuard<'_, T> { fn drop(&mut self) { self.mutex.unlock(); } }