126 lines
3.6 KiB
Rust
126 lines
3.6 KiB
Rust
//! 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<Duration>) -> 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<T: ?Sized> {
|
|
state: AtomicU32,
|
|
data: UnsafeCell<T>,
|
|
}
|
|
|
|
unsafe impl<T: ?Sized + Send> Send for Mutex<T> {}
|
|
unsafe impl<T: ?Sized + Send> Sync for Mutex<T> {}
|
|
|
|
impl<T> Mutex<T> {
|
|
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<T: ?Sized> Mutex<T> {
|
|
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<MutexGuard<'_, T>> {
|
|
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<T: Default> Default for Mutex<T> {
|
|
fn default() -> Self {
|
|
Mutex::new(T::default())
|
|
}
|
|
}
|
|
|
|
pub struct MutexGuard<'a, T: ?Sized> {
|
|
mutex: &'a Mutex<T>,
|
|
}
|
|
|
|
unsafe impl<T: ?Sized + Sync> Sync for MutexGuard<'_, T> {}
|
|
|
|
impl<T: ?Sized> Deref for MutexGuard<'_, T> {
|
|
type Target = T;
|
|
fn deref(&self) -> &T {
|
|
unsafe { &*self.mutex.data.get() }
|
|
}
|
|
}
|
|
|
|
impl<T: ?Sized> DerefMut for MutexGuard<'_, T> {
|
|
fn deref_mut(&mut self) -> &mut T {
|
|
unsafe { &mut *self.mutex.data.get() }
|
|
}
|
|
}
|
|
|
|
impl<T: ?Sized> Drop for MutexGuard<'_, T> {
|
|
fn drop(&mut self) {
|
|
self.mutex.unlock();
|
|
}
|
|
}
|