From cf7ff15a0ddfe0713dd794cd39eb3b3b58ba8d27 Mon Sep 17 00:00:00 2001 From: Paul Menage Date: Thu, 17 Oct 2024 12:21:53 -0700 Subject: [PATCH] Abstract the state type for futexes In the same way that we expose SmallAtomic and SmallPrimitive to allow Windows to use a value other than an AtomicU32 for its futex state, this patch switches the primary futex state type from AtomicU32 to futex::Atomic. The futex::Atomic type should be usable as an atomic value with underlying primitive type equal to futex::Primitive. This allows supporting the futex API on systems where the underlying kernel futex implementation requires more state than simply an AtomicU32. All in-tree futex implementations simply define {Atomic,Primitive} directly as {AtomicU32,u32}. --- library/std/src/sys/pal/hermit/futex.rs | 9 +++- library/std/src/sys/pal/unix/futex.rs | 9 +++- library/std/src/sys/pal/wasm/atomics/futex.rs | 9 +++- library/std/src/sys/pal/windows/futex.rs | 35 +++++++++------- library/std/src/sys/sync/condvar/futex.rs | 7 ++-- library/std/src/sys/sync/mutex/futex.rs | 6 +-- library/std/src/sys/sync/once/futex.rs | 25 ++++++----- library/std/src/sys/sync/once/queue.rs | 2 +- library/std/src/sys/sync/rwlock/futex.rs | 41 +++++++++---------- .../std/src/sys/sync/thread_parking/futex.rs | 6 +-- 10 files changed, 83 insertions(+), 66 deletions(-) diff --git a/library/std/src/sys/pal/hermit/futex.rs b/library/std/src/sys/pal/hermit/futex.rs index 21c5facd52f..670383b45ac 100644 --- a/library/std/src/sys/pal/hermit/futex.rs +++ b/library/std/src/sys/pal/hermit/futex.rs @@ -3,9 +3,14 @@ use crate::sync::atomic::AtomicU32; use crate::time::Duration; +/// An atomic for use as a futex that is at least 32-bits but may be larger +pub type Futex = AtomicU32; +/// Must be the underlying type of Futex +pub type Primitive = u32; + /// An atomic for use as a futex that is at least 8-bits but may be larger. -pub type SmallAtomic = AtomicU32; -/// Must be the underlying type of SmallAtomic +pub type SmallFutex = AtomicU32; +/// Must be the underlying type of SmallFutex pub type SmallPrimitive = u32; pub fn futex_wait(futex: &AtomicU32, expected: u32, timeout: Option) -> bool { diff --git a/library/std/src/sys/pal/unix/futex.rs b/library/std/src/sys/pal/unix/futex.rs index cc725045c48..0fc765dc87a 100644 --- a/library/std/src/sys/pal/unix/futex.rs +++ b/library/std/src/sys/pal/unix/futex.rs @@ -11,9 +11,14 @@ use crate::sync::atomic::AtomicU32; use crate::time::Duration; +/// An atomic for use as a futex that is at least 32-bits but may be larger +pub type Futex = AtomicU32; +/// Must be the underlying type of Futex +pub type Primitive = u32; + /// An atomic for use as a futex that is at least 8-bits but may be larger. -pub type SmallAtomic = AtomicU32; -/// Must be the underlying type of SmallAtomic +pub type SmallFutex = AtomicU32; +/// Must be the underlying type of SmallFutex pub type SmallPrimitive = u32; /// Waits for a `futex_wake` operation to wake us. diff --git a/library/std/src/sys/pal/wasm/atomics/futex.rs b/library/std/src/sys/pal/wasm/atomics/futex.rs index 42913a99ee9..bdad0da73f0 100644 --- a/library/std/src/sys/pal/wasm/atomics/futex.rs +++ b/library/std/src/sys/pal/wasm/atomics/futex.rs @@ -6,9 +6,14 @@ use crate::sync::atomic::AtomicU32; use crate::time::Duration; +/// An atomic for use as a futex that is at least 32-bits but may be larger +pub type Futex = AtomicU32; +/// Must be the underlying type of Futex +pub type Primitive = u32; + /// An atomic for use as a futex that is at least 8-bits but may be larger. -pub type SmallAtomic = AtomicU32; -/// Must be the underlying type of SmallAtomic +pub type SmallFutex = AtomicU32; +/// Must be the underlying type of SmallFutex pub type SmallPrimitive = u32; /// Wait for a futex_wake operation to wake us. diff --git a/library/std/src/sys/pal/windows/futex.rs b/library/std/src/sys/pal/windows/futex.rs index 4d6c4df9a5a..38afb8c043b 100644 --- a/library/std/src/sys/pal/windows/futex.rs +++ b/library/std/src/sys/pal/windows/futex.rs @@ -9,22 +9,27 @@ use super::api::{self, WinError}; use crate::sys::{c, dur2timeout}; +/// An atomic for use as a futex that is at least 32-bits but may be larger +pub type Futex = AtomicU32; +/// Must be the underlying type of Futex +pub type Primitive = u32; + /// An atomic for use as a futex that is at least 8-bits but may be larger. -pub type SmallAtomic = AtomicU8; -/// Must be the underlying type of SmallAtomic +pub type SmallFutex = AtomicU8; +/// Must be the underlying type of SmallFutex pub type SmallPrimitive = u8; -pub unsafe trait Futex {} +pub unsafe trait Futexable {} pub unsafe trait Waitable { - type Atomic; + type Futex; } macro_rules! unsafe_waitable_int { ($(($int:ty, $atomic:ty)),*$(,)?) => { $( unsafe impl Waitable for $int { - type Atomic = $atomic; + type Futex = $atomic; } - unsafe impl Futex for $atomic {} + unsafe impl Futexable for $atomic {} )* }; } @@ -42,15 +47,15 @@ unsafe impl Futex for $atomic {} (usize, AtomicUsize), } unsafe impl Waitable for *const T { - type Atomic = AtomicPtr; + type Futex = AtomicPtr; } unsafe impl Waitable for *mut T { - type Atomic = AtomicPtr; + type Futex = AtomicPtr; } -unsafe impl Futex for AtomicPtr {} +unsafe impl Futexable for AtomicPtr {} pub fn wait_on_address( - address: &W::Atomic, + address: &W::Futex, compare: W, timeout: Option, ) -> bool { @@ -63,30 +68,30 @@ pub fn wait_on_address( } } -pub fn wake_by_address_single(address: &T) { +pub fn wake_by_address_single(address: &T) { unsafe { let addr = ptr::from_ref(address).cast::(); c::WakeByAddressSingle(addr); } } -pub fn wake_by_address_all(address: &T) { +pub fn wake_by_address_all(address: &T) { unsafe { let addr = ptr::from_ref(address).cast::(); c::WakeByAddressAll(addr); } } -pub fn futex_wait(futex: &W::Atomic, expected: W, timeout: Option) -> bool { +pub fn futex_wait(futex: &W::Futex, expected: W, timeout: Option) -> bool { // return false only on timeout wait_on_address(futex, expected, timeout) || api::get_last_error() != WinError::TIMEOUT } -pub fn futex_wake(futex: &T) -> bool { +pub fn futex_wake(futex: &T) -> bool { wake_by_address_single(futex); false } -pub fn futex_wake_all(futex: &T) { +pub fn futex_wake_all(futex: &T) { wake_by_address_all(futex) } diff --git a/library/std/src/sys/sync/condvar/futex.rs b/library/std/src/sys/sync/condvar/futex.rs index 39cd97c01ea..0d0c5f0dbe7 100644 --- a/library/std/src/sys/sync/condvar/futex.rs +++ b/library/std/src/sys/sync/condvar/futex.rs @@ -1,6 +1,5 @@ -use crate::sync::atomic::AtomicU32; use crate::sync::atomic::Ordering::Relaxed; -use crate::sys::futex::{futex_wait, futex_wake, futex_wake_all}; +use crate::sys::futex::{Futex, futex_wait, futex_wake, futex_wake_all}; use crate::sys::sync::Mutex; use crate::time::Duration; @@ -8,13 +7,13 @@ pub struct Condvar { // The value of this atomic is simply incremented on every notification. // This is used by `.wait()` to not miss any notifications after // unlocking the mutex and before waiting for notifications. - futex: AtomicU32, + futex: Futex, } impl Condvar { #[inline] pub const fn new() -> Self { - Self { futex: AtomicU32::new(0) } + Self { futex: Futex::new(0) } } // All the memory orderings here are `Relaxed`, diff --git a/library/std/src/sys/sync/mutex/futex.rs b/library/std/src/sys/sync/mutex/futex.rs index 81afa94b147..ce9b2daa5f8 100644 --- a/library/std/src/sys/sync/mutex/futex.rs +++ b/library/std/src/sys/sync/mutex/futex.rs @@ -1,11 +1,11 @@ use crate::sync::atomic::Ordering::{Acquire, Relaxed, Release}; use crate::sys::futex::{self, futex_wait, futex_wake}; -type Atomic = futex::SmallAtomic; +type Futex = futex::SmallFutex; type State = futex::SmallPrimitive; pub struct Mutex { - futex: Atomic, + futex: Futex, } const UNLOCKED: State = 0; @@ -15,7 +15,7 @@ pub struct Mutex { impl Mutex { #[inline] pub const fn new() -> Self { - Self { futex: Atomic::new(UNLOCKED) } + Self { futex: Futex::new(UNLOCKED) } } #[inline] diff --git a/library/std/src/sys/sync/once/futex.rs b/library/std/src/sys/sync/once/futex.rs index 25588a4217b..10bfa81a6d7 100644 --- a/library/std/src/sys/sync/once/futex.rs +++ b/library/std/src/sys/sync/once/futex.rs @@ -1,39 +1,38 @@ use crate::cell::Cell; use crate::sync as public; -use crate::sync::atomic::AtomicU32; use crate::sync::atomic::Ordering::{Acquire, Relaxed, Release}; use crate::sync::once::ExclusiveState; -use crate::sys::futex::{futex_wait, futex_wake_all}; +use crate::sys::futex::{Futex, Primitive, futex_wait, futex_wake_all}; // On some platforms, the OS is very nice and handles the waiter queue for us. // This means we only need one atomic value with 4 states: /// No initialization has run yet, and no thread is currently using the Once. -const INCOMPLETE: u32 = 0; +const INCOMPLETE: Primitive = 0; /// Some thread has previously attempted to initialize the Once, but it panicked, /// so the Once is now poisoned. There are no other threads currently accessing /// this Once. -const POISONED: u32 = 1; +const POISONED: Primitive = 1; /// Some thread is currently attempting to run initialization. It may succeed, /// so all future threads need to wait for it to finish. -const RUNNING: u32 = 2; +const RUNNING: Primitive = 2; /// Initialization has completed and all future calls should finish immediately. -const COMPLETE: u32 = 3; +const COMPLETE: Primitive = 3; // An additional bit indicates whether there are waiting threads: /// May only be set if the state is not COMPLETE. -const QUEUED: u32 = 4; +const QUEUED: Primitive = 4; // Threads wait by setting the QUEUED bit and calling `futex_wait` on the state // variable. When the running thread finishes, it will wake all waiting threads using // `futex_wake_all`. -const STATE_MASK: u32 = 0b11; +const STATE_MASK: Primitive = 0b11; pub struct OnceState { poisoned: bool, - set_state_to: Cell, + set_state_to: Cell, } impl OnceState { @@ -49,8 +48,8 @@ pub fn poison(&self) { } struct CompletionGuard<'a> { - state_and_queued: &'a AtomicU32, - set_state_on_drop_to: u32, + state_and_queued: &'a Futex, + set_state_on_drop_to: Primitive, } impl<'a> Drop for CompletionGuard<'a> { @@ -65,13 +64,13 @@ fn drop(&mut self) { } pub struct Once { - state_and_queued: AtomicU32, + state_and_queued: Futex, } impl Once { #[inline] pub const fn new() -> Once { - Once { state_and_queued: AtomicU32::new(INCOMPLETE) } + Once { state_and_queued: Futex::new(INCOMPLETE) } } #[inline] diff --git a/library/std/src/sys/sync/once/queue.rs b/library/std/src/sys/sync/once/queue.rs index 17abaf0bf26..3e83a4a088f 100644 --- a/library/std/src/sys/sync/once/queue.rs +++ b/library/std/src/sys/sync/once/queue.rs @@ -23,7 +23,7 @@ // You'll find a few more details in the implementation, but that's the gist of // it! // -// Atomic orderings: +// Futex orderings: // When running `Once` we deal with multiple atomics: // `Once.state_and_queue` and an unknown number of `Waiter.signaled`. // * `state_and_queue` is used (1) as a state flag, (2) for synchronizing the diff --git a/library/std/src/sys/sync/rwlock/futex.rs b/library/std/src/sys/sync/rwlock/futex.rs index 75ecc2ab5c5..df22c36dd5a 100644 --- a/library/std/src/sys/sync/rwlock/futex.rs +++ b/library/std/src/sys/sync/rwlock/futex.rs @@ -1,6 +1,5 @@ -use crate::sync::atomic::AtomicU32; use crate::sync::atomic::Ordering::{Acquire, Relaxed, Release}; -use crate::sys::futex::{futex_wait, futex_wake, futex_wake_all}; +use crate::sys::futex::{Futex, Primitive, futex_wait, futex_wake, futex_wake_all}; pub struct RwLock { // The state consists of a 30-bit reader counter, a 'readers waiting' flag, and a 'writers waiting' flag. @@ -10,41 +9,41 @@ pub struct RwLock { // 0x3FFF_FFFF: Write locked // Bit 30: Readers are waiting on this futex. // Bit 31: Writers are waiting on the writer_notify futex. - state: AtomicU32, + state: Futex, // The 'condition variable' to notify writers through. // Incremented on every signal. - writer_notify: AtomicU32, + writer_notify: Futex, } -const READ_LOCKED: u32 = 1; -const MASK: u32 = (1 << 30) - 1; -const WRITE_LOCKED: u32 = MASK; -const MAX_READERS: u32 = MASK - 1; -const READERS_WAITING: u32 = 1 << 30; -const WRITERS_WAITING: u32 = 1 << 31; +const READ_LOCKED: Primitive = 1; +const MASK: Primitive = (1 << 30) - 1; +const WRITE_LOCKED: Primitive = MASK; +const MAX_READERS: Primitive = MASK - 1; +const READERS_WAITING: Primitive = 1 << 30; +const WRITERS_WAITING: Primitive = 1 << 31; #[inline] -fn is_unlocked(state: u32) -> bool { +fn is_unlocked(state: Primitive) -> bool { state & MASK == 0 } #[inline] -fn is_write_locked(state: u32) -> bool { +fn is_write_locked(state: Primitive) -> bool { state & MASK == WRITE_LOCKED } #[inline] -fn has_readers_waiting(state: u32) -> bool { +fn has_readers_waiting(state: Primitive) -> bool { state & READERS_WAITING != 0 } #[inline] -fn has_writers_waiting(state: u32) -> bool { +fn has_writers_waiting(state: Primitive) -> bool { state & WRITERS_WAITING != 0 } #[inline] -fn is_read_lockable(state: u32) -> bool { +fn is_read_lockable(state: Primitive) -> bool { // This also returns false if the counter could overflow if we tried to read lock it. // // We don't allow read-locking if there's readers waiting, even if the lock is unlocked @@ -55,14 +54,14 @@ fn is_read_lockable(state: u32) -> bool { } #[inline] -fn has_reached_max_readers(state: u32) -> bool { +fn has_reached_max_readers(state: Primitive) -> bool { state & MASK == MAX_READERS } impl RwLock { #[inline] pub const fn new() -> Self { - Self { state: AtomicU32::new(0), writer_notify: AtomicU32::new(0) } + Self { state: Futex::new(0), writer_notify: Futex::new(0) } } #[inline] @@ -225,7 +224,7 @@ fn write_contended(&self) { /// If both are waiting, this will wake up only one writer, but will fall /// back to waking up readers if there was no writer to wake up. #[cold] - fn wake_writer_or_readers(&self, mut state: u32) { + fn wake_writer_or_readers(&self, mut state: Primitive) { assert!(is_unlocked(state)); // The readers waiting bit might be turned on at any point now, @@ -290,7 +289,7 @@ fn wake_writer(&self) -> bool { /// Spin for a while, but stop directly at the given condition. #[inline] - fn spin_until(&self, f: impl Fn(u32) -> bool) -> u32 { + fn spin_until(&self, f: impl Fn(Primitive) -> bool) -> Primitive { let mut spin = 100; // Chosen by fair dice roll. loop { let state = self.state.load(Relaxed); @@ -303,13 +302,13 @@ fn spin_until(&self, f: impl Fn(u32) -> bool) -> u32 { } #[inline] - fn spin_write(&self) -> u32 { + fn spin_write(&self) -> Primitive { // Stop spinning when it's unlocked or when there's waiting writers, to keep things somewhat fair. self.spin_until(|state| is_unlocked(state) || has_writers_waiting(state)) } #[inline] - fn spin_read(&self) -> u32 { + fn spin_read(&self) -> Primitive { // Stop spinning when it's unlocked or read locked, or when there's waiting threads. self.spin_until(|state| { !is_write_locked(state) || has_readers_waiting(state) || has_writers_waiting(state) diff --git a/library/std/src/sys/sync/thread_parking/futex.rs b/library/std/src/sys/sync/thread_parking/futex.rs index ce852eaadc4..c8f7f26386a 100644 --- a/library/std/src/sys/sync/thread_parking/futex.rs +++ b/library/std/src/sys/sync/thread_parking/futex.rs @@ -4,7 +4,7 @@ use crate::sys::futex::{self, futex_wait, futex_wake}; use crate::time::Duration; -type Atomic = futex::SmallAtomic; +type Futex = futex::SmallFutex; type State = futex::SmallPrimitive; const PARKED: State = State::MAX; @@ -12,7 +12,7 @@ const NOTIFIED: State = 1; pub struct Parker { - state: Atomic, + state: Futex, } // Notes about memory ordering: @@ -39,7 +39,7 @@ impl Parker { /// Constructs the futex parker. The UNIX parker implementation /// requires this to happen in-place. pub unsafe fn new_in_place(parker: *mut Parker) { - unsafe { parker.write(Self { state: Atomic::new(EMPTY) }) }; + unsafe { parker.write(Self { state: Futex::new(EMPTY) }) }; } // Assumes this is only called by the thread that owns the Parker,