diff options
Diffstat (limited to 'library/std/src/sys/sync')
| -rw-r--r-- | library/std/src/sys/sync/mod.rs | 2 | ||||
| -rw-r--r-- | library/std/src/sys/sync/thread_parking/darwin.rs | 132 | ||||
| -rw-r--r-- | library/std/src/sys/sync/thread_parking/futex.rs | 97 | ||||
| -rw-r--r-- | library/std/src/sys/sync/thread_parking/id.rs | 103 | ||||
| -rw-r--r-- | library/std/src/sys/sync/thread_parking/mod.rs | 37 | ||||
| -rw-r--r-- | library/std/src/sys/sync/thread_parking/pthread.rs | 269 | ||||
| -rw-r--r-- | library/std/src/sys/sync/thread_parking/unsupported.rs | 11 | ||||
| -rw-r--r-- | library/std/src/sys/sync/thread_parking/windows.rs | 278 | ||||
| -rw-r--r-- | library/std/src/sys/sync/thread_parking/xous.rs | 112 |
9 files changed, 1041 insertions, 0 deletions
diff --git a/library/std/src/sys/sync/mod.rs b/library/std/src/sys/sync/mod.rs index 623e6bccd51..52fac5902a2 100644 --- a/library/std/src/sys/sync/mod.rs +++ b/library/std/src/sys/sync/mod.rs @@ -2,8 +2,10 @@ mod condvar; mod mutex; mod once; mod rwlock; +mod thread_parking; pub use condvar::Condvar; pub use mutex::Mutex; pub use once::{Once, OnceState}; pub use rwlock::RwLock; +pub use thread_parking::Parker; diff --git a/library/std/src/sys/sync/thread_parking/darwin.rs b/library/std/src/sys/sync/thread_parking/darwin.rs new file mode 100644 index 00000000000..973c08f0317 --- /dev/null +++ b/library/std/src/sys/sync/thread_parking/darwin.rs @@ -0,0 +1,132 @@ +//! Thread parking for Darwin-based systems. +//! +//! Darwin actually has futex syscalls (`__ulock_wait`/`__ulock_wake`), but they +//! cannot be used in `std` because they are non-public (their use will lead to +//! rejection from the App Store). +//! +//! Therefore, we need to look for other synchronization primitives. Luckily, Darwin +//! supports semaphores, which allow us to implement the behaviour we need with +//! only one primitive (as opposed to a mutex-condvar pair). We use the semaphore +//! provided by libdispatch, as the underlying Mach semaphore is only dubiously +//! public. + +#![allow(non_camel_case_types)] + +use crate::pin::Pin; +use crate::sync::atomic::{ + AtomicI8, + Ordering::{Acquire, Release}, +}; +use crate::time::Duration; + +type dispatch_semaphore_t = *mut crate::ffi::c_void; +type dispatch_time_t = u64; + +const DISPATCH_TIME_NOW: dispatch_time_t = 0; +const DISPATCH_TIME_FOREVER: dispatch_time_t = !0; + +// Contained in libSystem.dylib, which is linked by default. +extern "C" { + fn dispatch_time(when: dispatch_time_t, delta: i64) -> dispatch_time_t; + fn dispatch_semaphore_create(val: isize) -> dispatch_semaphore_t; + fn dispatch_semaphore_wait(dsema: dispatch_semaphore_t, timeout: dispatch_time_t) -> isize; + fn dispatch_semaphore_signal(dsema: dispatch_semaphore_t) -> isize; + fn dispatch_release(object: *mut crate::ffi::c_void); +} + +const EMPTY: i8 = 0; +const NOTIFIED: i8 = 1; +const PARKED: i8 = -1; + +pub struct Parker { + semaphore: dispatch_semaphore_t, + state: AtomicI8, +} + +unsafe impl Sync for Parker {} +unsafe impl Send for Parker {} + +impl Parker { + pub unsafe fn new_in_place(parker: *mut Parker) { + let semaphore = dispatch_semaphore_create(0); + assert!( + !semaphore.is_null(), + "failed to create dispatch semaphore for thread synchronization" + ); + parker.write(Parker { semaphore, state: AtomicI8::new(EMPTY) }) + } + + // Does not need `Pin`, but other implementation do. + pub unsafe fn park(self: Pin<&Self>) { + // The semaphore counter must be zero at this point, because unparking + // threads will not actually increase it until we signalled that we + // are waiting. + + // Change NOTIFIED to EMPTY and EMPTY to PARKED. + if self.state.fetch_sub(1, Acquire) == NOTIFIED { + return; + } + + // Another thread may increase the semaphore counter from this point on. + // If it is faster than us, we will decrement it again immediately below. + // If we are faster, we wait. + + // Ensure that the semaphore counter has actually been decremented, even + // if the call timed out for some reason. + while dispatch_semaphore_wait(self.semaphore, DISPATCH_TIME_FOREVER) != 0 {} + + // At this point, the semaphore counter is zero again. + + // We were definitely woken up, so we don't need to check the state. + // Still, we need to reset the state using a swap to observe the state + // change with acquire ordering. + self.state.swap(EMPTY, Acquire); + } + + // Does not need `Pin`, but other implementation do. + pub unsafe fn park_timeout(self: Pin<&Self>, dur: Duration) { + if self.state.fetch_sub(1, Acquire) == NOTIFIED { + return; + } + + let nanos = dur.as_nanos().try_into().unwrap_or(i64::MAX); + let timeout = dispatch_time(DISPATCH_TIME_NOW, nanos); + + let timeout = dispatch_semaphore_wait(self.semaphore, timeout) != 0; + + let state = self.state.swap(EMPTY, Acquire); + if state == NOTIFIED && timeout { + // If the state was NOTIFIED but semaphore_wait returned without + // decrementing the count because of a timeout, it means another + // thread is about to call semaphore_signal. We must wait for that + // to happen to ensure the semaphore count is reset. + while dispatch_semaphore_wait(self.semaphore, DISPATCH_TIME_FOREVER) != 0 {} + } else { + // Either a timeout occurred and we reset the state before any thread + // tried to wake us up, or we were woken up and reset the state, + // making sure to observe the state change with acquire ordering. + // Either way, the semaphore counter is now zero again. + } + } + + // Does not need `Pin`, but other implementation do. + pub fn unpark(self: Pin<&Self>) { + let state = self.state.swap(NOTIFIED, Release); + if state == PARKED { + unsafe { + dispatch_semaphore_signal(self.semaphore); + } + } + } +} + +impl Drop for Parker { + fn drop(&mut self) { + // SAFETY: + // We always ensure that the semaphore count is reset, so this will + // never cause an exception. + unsafe { + dispatch_release(self.semaphore); + } + } +} diff --git a/library/std/src/sys/sync/thread_parking/futex.rs b/library/std/src/sys/sync/thread_parking/futex.rs new file mode 100644 index 00000000000..588e7b27826 --- /dev/null +++ b/library/std/src/sys/sync/thread_parking/futex.rs @@ -0,0 +1,97 @@ +use crate::pin::Pin; +use crate::sync::atomic::AtomicU32; +use crate::sync::atomic::Ordering::{Acquire, Release}; +use crate::sys::futex::{futex_wait, futex_wake}; +use crate::time::Duration; + +const PARKED: u32 = u32::MAX; +const EMPTY: u32 = 0; +const NOTIFIED: u32 = 1; + +pub struct Parker { + state: AtomicU32, +} + +// Notes about memory ordering: +// +// Memory ordering is only relevant for the relative ordering of operations +// between different variables. Even Ordering::Relaxed guarantees a +// monotonic/consistent order when looking at just a single atomic variable. +// +// So, since this parker is just a single atomic variable, we only need to look +// at the ordering guarantees we need to provide to the 'outside world'. +// +// The only memory ordering guarantee that parking and unparking provide, is +// that things which happened before unpark() are visible on the thread +// returning from park() afterwards. Otherwise, it was effectively unparked +// before unpark() was called while still consuming the 'token'. +// +// In other words, unpark() needs to synchronize with the part of park() that +// consumes the token and returns. +// +// This is done with a release-acquire synchronization, by using +// Ordering::Release when writing NOTIFIED (the 'token') in unpark(), and using +// Ordering::Acquire when checking for this state in park(). +impl Parker { + /// Construct the futex parker. The UNIX parker implementation + /// requires this to happen in-place. + pub unsafe fn new_in_place(parker: *mut Parker) { + parker.write(Self { state: AtomicU32::new(EMPTY) }); + } + + // Assumes this is only called by the thread that owns the Parker, + // which means that `self.state != PARKED`. + pub unsafe fn park(self: Pin<&Self>) { + // Change NOTIFIED=>EMPTY or EMPTY=>PARKED, and directly return in the + // first case. + if self.state.fetch_sub(1, Acquire) == NOTIFIED { + return; + } + loop { + // Wait for something to happen, assuming it's still set to PARKED. + futex_wait(&self.state, PARKED, None); + // Change NOTIFIED=>EMPTY and return in that case. + if self.state.compare_exchange(NOTIFIED, EMPTY, Acquire, Acquire).is_ok() { + return; + } else { + // Spurious wake up. We loop to try again. + } + } + } + + // Assumes this is only called by the thread that owns the Parker, + // which means that `self.state != PARKED`. This implementation doesn't + // require `Pin`, but other implementations do. + pub unsafe fn park_timeout(self: Pin<&Self>, timeout: Duration) { + // Change NOTIFIED=>EMPTY or EMPTY=>PARKED, and directly return in the + // first case. + if self.state.fetch_sub(1, Acquire) == NOTIFIED { + return; + } + // Wait for something to happen, assuming it's still set to PARKED. + futex_wait(&self.state, PARKED, Some(timeout)); + // This is not just a store, because we need to establish a + // release-acquire ordering with unpark(). + if self.state.swap(EMPTY, Acquire) == NOTIFIED { + // Woke up because of unpark(). + } else { + // Timeout or spurious wake up. + // We return either way, because we can't easily tell if it was the + // timeout or not. + } + } + + // This implementation doesn't require `Pin`, but other implementations do. + #[inline] + pub fn unpark(self: Pin<&Self>) { + // Change PARKED=>NOTIFIED, EMPTY=>NOTIFIED, or NOTIFIED=>NOTIFIED, and + // wake the thread in the first case. + // + // Note that even NOTIFIED=>NOTIFIED results in a write. This is on + // purpose, to make sure every unpark() has a release-acquire ordering + // with park(). + if self.state.swap(NOTIFIED, Release) == PARKED { + futex_wake(&self.state); + } + } +} diff --git a/library/std/src/sys/sync/thread_parking/id.rs b/library/std/src/sys/sync/thread_parking/id.rs new file mode 100644 index 00000000000..04667439660 --- /dev/null +++ b/library/std/src/sys/sync/thread_parking/id.rs @@ -0,0 +1,103 @@ +//! Thread parking using thread ids. +//! +//! Some platforms (notably NetBSD) have thread parking primitives whose semantics +//! match those offered by `thread::park`, with the difference that the thread to +//! be unparked is referenced by a platform-specific thread id. Since the thread +//! parker is constructed before that id is known, an atomic state variable is used +//! to manage the park state and propagate the thread id. This also avoids platform +//! calls in the case where `unpark` is called before `park`. + +use crate::cell::UnsafeCell; +use crate::pin::Pin; +use crate::sync::atomic::{ + fence, AtomicI8, + Ordering::{Acquire, Relaxed, Release}, +}; +use crate::sys::thread_parking::{current, park, park_timeout, unpark, ThreadId}; +use crate::time::Duration; + +pub struct Parker { + state: AtomicI8, + tid: UnsafeCell<Option<ThreadId>>, +} + +const PARKED: i8 = -1; +const EMPTY: i8 = 0; +const NOTIFIED: i8 = 1; + +impl Parker { + pub fn new() -> Parker { + Parker { state: AtomicI8::new(EMPTY), tid: UnsafeCell::new(None) } + } + + /// Create a new thread parker. UNIX requires this to happen in-place. + pub unsafe fn new_in_place(parker: *mut Parker) { + parker.write(Parker::new()) + } + + /// # Safety + /// * must always be called from the same thread + /// * must be called before the state is set to PARKED + unsafe fn init_tid(&self) { + // The field is only ever written to from this thread, so we don't need + // synchronization to read it here. + if self.tid.get().read().is_none() { + // Because this point is only reached once, before the state is set + // to PARKED for the first time, the non-atomic write here can not + // conflict with reads by other threads. + self.tid.get().write(Some(current())); + // Ensure that the write can be observed by all threads reading the + // state. Synchronizes with the acquire barrier in `unpark`. + fence(Release); + } + } + + pub unsafe fn park(self: Pin<&Self>) { + self.init_tid(); + + // Changes NOTIFIED to EMPTY and EMPTY to PARKED. + let state = self.state.fetch_sub(1, Acquire); + if state == EMPTY { + // Loop to guard against spurious wakeups. + // The state must be reset with acquire ordering to ensure that all + // calls to `unpark` synchronize with this thread. + while self.state.compare_exchange(NOTIFIED, EMPTY, Acquire, Relaxed).is_err() { + park(self.state.as_ptr().addr()); + } + } + } + + pub unsafe fn park_timeout(self: Pin<&Self>, dur: Duration) { + self.init_tid(); + + let state = self.state.fetch_sub(1, Acquire).wrapping_sub(1); + if state == PARKED { + park_timeout(dur, self.state.as_ptr().addr()); + // Swap to ensure that we observe all state changes with acquire + // ordering. + self.state.swap(EMPTY, Acquire); + } + } + + pub fn unpark(self: Pin<&Self>) { + let state = self.state.swap(NOTIFIED, Release); + if state == PARKED { + // Synchronize with the release fence in `init_tid` to observe the + // write to `tid`. + fence(Acquire); + // # Safety + // The thread id is initialized before the state is set to `PARKED` + // for the first time and is not written to from that point on + // (negating the need for an atomic read). + let tid = unsafe { self.tid.get().read().unwrap_unchecked() }; + // It is possible that the waiting thread woke up because of a timeout + // and terminated before this call is made. This call then returns an + // error or wakes up an unrelated thread. The platform API and + // environment does allow this, however. + unpark(tid, self.state.as_ptr().addr()); + } + } +} + +unsafe impl Send for Parker {} +unsafe impl Sync for Parker {} diff --git a/library/std/src/sys/sync/thread_parking/mod.rs b/library/std/src/sys/sync/thread_parking/mod.rs new file mode 100644 index 00000000000..ed1a6437faa --- /dev/null +++ b/library/std/src/sys/sync/thread_parking/mod.rs @@ -0,0 +1,37 @@ +cfg_if::cfg_if! { + if #[cfg(any( + target_os = "linux", + target_os = "android", + all(target_arch = "wasm32", target_feature = "atomics"), + target_os = "freebsd", + target_os = "openbsd", + target_os = "dragonfly", + target_os = "fuchsia", + target_os = "hermit", + ))] { + mod futex; + pub use futex::Parker; + } else if #[cfg(any( + target_os = "netbsd", + all(target_vendor = "fortanix", target_env = "sgx"), + target_os = "solid_asp3", + ))] { + mod id; + pub use id::Parker; + } else if #[cfg(target_os = "windows")] { + mod windows; + pub use windows::Parker; + } else if #[cfg(all(target_vendor = "apple", not(miri)))] { + mod darwin; + pub use darwin::Parker; + } else if #[cfg(target_os = "xous")] { + mod xous; + pub use xous::Parker; + } else if #[cfg(target_family = "unix")] { + mod pthread; + pub use pthread::Parker; + } else { + mod unsupported; + pub use unsupported::Parker; + } +} diff --git a/library/std/src/sys/sync/thread_parking/pthread.rs b/library/std/src/sys/sync/thread_parking/pthread.rs new file mode 100644 index 00000000000..fdac1096dbf --- /dev/null +++ b/library/std/src/sys/sync/thread_parking/pthread.rs @@ -0,0 +1,269 @@ +//! Thread parking without `futex` using the `pthread` synchronization primitives. + +use crate::cell::UnsafeCell; +use crate::marker::PhantomPinned; +use crate::pin::Pin; +use crate::ptr::addr_of_mut; +use crate::sync::atomic::AtomicUsize; +use crate::sync::atomic::Ordering::{Acquire, Relaxed, Release}; +#[cfg(not(target_os = "nto"))] +use crate::sys::time::TIMESPEC_MAX; +#[cfg(target_os = "nto")] +use crate::sys::time::TIMESPEC_MAX_CAPPED; +use crate::time::Duration; + +const EMPTY: usize = 0; +const PARKED: usize = 1; +const NOTIFIED: usize = 2; + +unsafe fn lock(lock: *mut libc::pthread_mutex_t) { + let r = libc::pthread_mutex_lock(lock); + debug_assert_eq!(r, 0); +} + +unsafe fn unlock(lock: *mut libc::pthread_mutex_t) { + let r = libc::pthread_mutex_unlock(lock); + debug_assert_eq!(r, 0); +} + +unsafe fn notify_one(cond: *mut libc::pthread_cond_t) { + let r = libc::pthread_cond_signal(cond); + debug_assert_eq!(r, 0); +} + +unsafe fn wait(cond: *mut libc::pthread_cond_t, lock: *mut libc::pthread_mutex_t) { + let r = libc::pthread_cond_wait(cond, lock); + debug_assert_eq!(r, 0); +} + +unsafe fn wait_timeout( + cond: *mut libc::pthread_cond_t, + lock: *mut libc::pthread_mutex_t, + dur: Duration, +) { + // Use the system clock on systems that do not support pthread_condattr_setclock. + // This unfortunately results in problems when the system time changes. + #[cfg(any(target_os = "espidf", target_os = "horizon", target_vendor = "apple"))] + let (now, dur) = { + use crate::cmp::min; + use crate::sys::time::SystemTime; + + // OSX implementation of `pthread_cond_timedwait` is buggy + // with super long durations. When duration is greater than + // 0x100_0000_0000_0000 seconds, `pthread_cond_timedwait` + // in macOS Sierra return error 316. + // + // This program demonstrates the issue: + // https://gist.github.com/stepancheg/198db4623a20aad2ad7cddb8fda4a63c + // + // To work around this issue, and possible bugs of other OSes, timeout + // is clamped to 1000 years, which is allowable per the API of `park_timeout` + // because of spurious wakeups. + let dur = min(dur, Duration::from_secs(1000 * 365 * 86400)); + let now = SystemTime::now().t; + (now, dur) + }; + // Use the monotonic clock on other systems. + #[cfg(not(any(target_os = "espidf", target_os = "horizon", target_vendor = "apple")))] + let (now, dur) = { + use crate::sys::time::Timespec; + + (Timespec::now(libc::CLOCK_MONOTONIC), dur) + }; + + #[cfg(not(target_os = "nto"))] + let timeout = + now.checked_add_duration(&dur).and_then(|t| t.to_timespec()).unwrap_or(TIMESPEC_MAX); + #[cfg(target_os = "nto")] + let timeout = now + .checked_add_duration(&dur) + .and_then(|t| t.to_timespec_capped()) + .unwrap_or(TIMESPEC_MAX_CAPPED); + let r = libc::pthread_cond_timedwait(cond, lock, &timeout); + debug_assert!(r == libc::ETIMEDOUT || r == 0); +} + +pub struct Parker { + state: AtomicUsize, + lock: UnsafeCell<libc::pthread_mutex_t>, + cvar: UnsafeCell<libc::pthread_cond_t>, + // The `pthread` primitives require a stable address, so make this struct `!Unpin`. + _pinned: PhantomPinned, +} + +impl Parker { + /// Construct the UNIX parker in-place. + /// + /// # Safety + /// The constructed parker must never be moved. + pub unsafe fn new_in_place(parker: *mut Parker) { + // Use the default mutex implementation to allow for simpler initialization. + // This could lead to undefined behaviour when deadlocking. This is avoided + // by not deadlocking. Note in particular the unlocking operation before any + // panic, as code after the panic could try to park again. + addr_of_mut!((*parker).state).write(AtomicUsize::new(EMPTY)); + addr_of_mut!((*parker).lock).write(UnsafeCell::new(libc::PTHREAD_MUTEX_INITIALIZER)); + + cfg_if::cfg_if! { + if #[cfg(any( + target_os = "l4re", + target_os = "android", + target_os = "redox", + target_os = "vita", + target_vendor = "apple", + ))] { + addr_of_mut!((*parker).cvar).write(UnsafeCell::new(libc::PTHREAD_COND_INITIALIZER)); + } else if #[cfg(any(target_os = "espidf", target_os = "horizon"))] { + let r = libc::pthread_cond_init(addr_of_mut!((*parker).cvar).cast(), crate::ptr::null()); + assert_eq!(r, 0); + } else { + use crate::mem::MaybeUninit; + let mut attr = MaybeUninit::<libc::pthread_condattr_t>::uninit(); + let r = libc::pthread_condattr_init(attr.as_mut_ptr()); + assert_eq!(r, 0); + let r = libc::pthread_condattr_setclock(attr.as_mut_ptr(), libc::CLOCK_MONOTONIC); + assert_eq!(r, 0); + let r = libc::pthread_cond_init(addr_of_mut!((*parker).cvar).cast(), attr.as_ptr()); + assert_eq!(r, 0); + let r = libc::pthread_condattr_destroy(attr.as_mut_ptr()); + assert_eq!(r, 0); + } + } + } + + // This implementation doesn't require `unsafe`, but other implementations + // may assume this is only called by the thread that owns the Parker. + // + // For memory ordering, see futex.rs + pub unsafe fn park(self: Pin<&Self>) { + // If we were previously notified then we consume this notification and + // return quickly. + if self.state.compare_exchange(NOTIFIED, EMPTY, Acquire, Relaxed).is_ok() { + return; + } + + // Otherwise we need to coordinate going to sleep + lock(self.lock.get()); + match self.state.compare_exchange(EMPTY, PARKED, Relaxed, Relaxed) { + Ok(_) => {} + Err(NOTIFIED) => { + // We must read here, even though we know it will be `NOTIFIED`. + // This is because `unpark` may have been called again since we read + // `NOTIFIED` in the `compare_exchange` above. We must perform an + // acquire operation that synchronizes with that `unpark` to observe + // any writes it made before the call to unpark. To do that we must + // read from the write it made to `state`. + let old = self.state.swap(EMPTY, Acquire); + + unlock(self.lock.get()); + + assert_eq!(old, NOTIFIED, "park state changed unexpectedly"); + return; + } // should consume this notification, so prohibit spurious wakeups in next park. + Err(_) => { + unlock(self.lock.get()); + + panic!("inconsistent park state") + } + } + + loop { + wait(self.cvar.get(), self.lock.get()); + + match self.state.compare_exchange(NOTIFIED, EMPTY, Acquire, Relaxed) { + Ok(_) => break, // got a notification + Err(_) => {} // spurious wakeup, go back to sleep + } + } + + unlock(self.lock.get()); + } + + // This implementation doesn't require `unsafe`, but other implementations + // may assume this is only called by the thread that owns the Parker. Use + // `Pin` to guarantee a stable address for the mutex and condition variable. + pub unsafe fn park_timeout(self: Pin<&Self>, dur: Duration) { + // Like `park` above we have a fast path for an already-notified thread, and + // afterwards we start coordinating for a sleep. + // return quickly. + if self.state.compare_exchange(NOTIFIED, EMPTY, Acquire, Relaxed).is_ok() { + return; + } + + lock(self.lock.get()); + match self.state.compare_exchange(EMPTY, PARKED, Relaxed, Relaxed) { + Ok(_) => {} + Err(NOTIFIED) => { + // We must read again here, see `park`. + let old = self.state.swap(EMPTY, Acquire); + unlock(self.lock.get()); + + assert_eq!(old, NOTIFIED, "park state changed unexpectedly"); + return; + } // should consume this notification, so prohibit spurious wakeups in next park. + Err(_) => { + unlock(self.lock.get()); + panic!("inconsistent park_timeout state") + } + } + + // Wait with a timeout, and if we spuriously wake up or otherwise wake up + // from a notification we just want to unconditionally set the state back to + // empty, either consuming a notification or un-flagging ourselves as + // parked. + wait_timeout(self.cvar.get(), self.lock.get(), dur); + + match self.state.swap(EMPTY, Acquire) { + NOTIFIED => unlock(self.lock.get()), // got a notification, hurray! + PARKED => unlock(self.lock.get()), // no notification, alas + n => { + unlock(self.lock.get()); + panic!("inconsistent park_timeout state: {n}") + } + } + } + + pub fn unpark(self: Pin<&Self>) { + // To ensure the unparked thread will observe any writes we made + // before this call, we must perform a release operation that `park` + // can synchronize with. To do that we must write `NOTIFIED` even if + // `state` is already `NOTIFIED`. That is why this must be a swap + // rather than a compare-and-swap that returns if it reads `NOTIFIED` + // on failure. + match self.state.swap(NOTIFIED, Release) { + EMPTY => return, // no one was waiting + NOTIFIED => return, // already unparked + PARKED => {} // gotta go wake someone up + _ => panic!("inconsistent state in unpark"), + } + + // There is a period between when the parked thread sets `state` to + // `PARKED` (or last checked `state` in the case of a spurious wake + // up) and when it actually waits on `cvar`. If we were to notify + // during this period it would be ignored and then when the parked + // thread went to sleep it would never wake up. Fortunately, it has + // `lock` locked at this stage so we can acquire `lock` to wait until + // it is ready to receive the notification. + // + // Releasing `lock` before the call to `notify_one` means that when the + // parked thread wakes it doesn't get woken only to have to wait for us + // to release `lock`. + unsafe { + lock(self.lock.get()); + unlock(self.lock.get()); + notify_one(self.cvar.get()); + } + } +} + +impl Drop for Parker { + fn drop(&mut self) { + unsafe { + libc::pthread_cond_destroy(self.cvar.get_mut()); + libc::pthread_mutex_destroy(self.lock.get_mut()); + } + } +} + +unsafe impl Sync for Parker {} +unsafe impl Send for Parker {} diff --git a/library/std/src/sys/sync/thread_parking/unsupported.rs b/library/std/src/sys/sync/thread_parking/unsupported.rs new file mode 100644 index 00000000000..197078bb186 --- /dev/null +++ b/library/std/src/sys/sync/thread_parking/unsupported.rs @@ -0,0 +1,11 @@ +use crate::pin::Pin; +use crate::time::Duration; + +pub struct Parker {} + +impl Parker { + pub unsafe fn new_in_place(_parker: *mut Parker) {} + pub unsafe fn park(self: Pin<&Self>) {} + pub unsafe fn park_timeout(self: Pin<&Self>, _dur: Duration) {} + pub fn unpark(self: Pin<&Self>) {} +} diff --git a/library/std/src/sys/sync/thread_parking/windows.rs b/library/std/src/sys/sync/thread_parking/windows.rs new file mode 100644 index 00000000000..4b8102d505a --- /dev/null +++ b/library/std/src/sys/sync/thread_parking/windows.rs @@ -0,0 +1,278 @@ +// Thread parker implementation for Windows. +// +// This uses WaitOnAddress and WakeByAddressSingle if available (Windows 8+). +// This modern API is exactly the same as the futex syscalls the Linux thread +// parker uses. When These APIs are available, the implementation of this +// thread parker matches the Linux thread parker exactly. +// +// However, when the modern API is not available, this implementation falls +// back to NT Keyed Events, which are similar, but have some important +// differences. These are available since Windows XP. +// +// WaitOnAddress first checks the state of the thread parker to make sure it no +// WakeByAddressSingle calls can be missed between updating the parker state +// and calling the function. +// +// NtWaitForKeyedEvent does not have this option, and unconditionally blocks +// without checking the parker state first. Instead, NtReleaseKeyedEvent +// (unlike WakeByAddressSingle) *blocks* until it woke up a thread waiting for +// it by NtWaitForKeyedEvent. This way, we can be sure no events are missed, +// but we need to be careful not to block unpark() if park_timeout() was woken +// up by a timeout instead of unpark(). +// +// Unlike WaitOnAddress, NtWaitForKeyedEvent/NtReleaseKeyedEvent operate on a +// HANDLE (created with NtCreateKeyedEvent). This means that we can be sure +// a successfully awoken park() was awoken by unpark() and not a +// NtReleaseKeyedEvent call from some other code, as these events are not only +// matched by the key (address of the parker (state)), but also by this HANDLE. +// We lazily allocate this handle the first time it is needed. +// +// The fast path (calling park() after unpark() was already called) and the +// possible states are the same for both implementations. This is used here to +// make sure the fast path does not even check which API to use, but can return +// right away, independent of the used API. Only the slow paths (which will +// actually block/wake a thread) check which API is available and have +// different implementations. +// +// Unfortunately, NT Keyed Events are an undocumented Windows API. However: +// - This API is relatively simple with obvious behaviour, and there are +// several (unofficial) articles documenting the details. [1] +// - `parking_lot` has been using this API for years (on Windows versions +// before Windows 8). [2] Many big projects extensively use parking_lot, +// such as servo and the Rust compiler itself. +// - It is the underlying API used by Windows SRW locks and Windows critical +// sections. [3] [4] +// - The source code of the implementations of Wine, ReactOs, and Windows XP +// are available and match the expected behaviour. +// - The main risk with an undocumented API is that it might change in the +// future. But since we only use it for older versions of Windows, that's not +// a problem. +// - Even if these functions do not block or wake as we expect (which is +// unlikely, see all previous points), this implementation would still be +// memory safe. The NT Keyed Events API is only used to sleep/block in the +// right place. +// +// [1]: http://www.locklessinc.com/articles/keyed_events/ +// [2]: https://github.com/Amanieu/parking_lot/commit/43abbc964e +// [3]: https://docs.microsoft.com/en-us/archive/msdn-magazine/2012/november/windows-with-c-the-evolution-of-synchronization-in-windows-and-c +// [4]: Windows Internals, Part 1, ISBN 9780735671300 + +use crate::pin::Pin; +use crate::sync::atomic::{ + AtomicI8, + Ordering::{Acquire, Release}, +}; +use crate::sys::{c, dur2timeout}; +use crate::time::Duration; + +pub struct Parker { + state: AtomicI8, +} + +const PARKED: i8 = -1; +const EMPTY: i8 = 0; +const NOTIFIED: i8 = 1; + +// Notes about memory ordering: +// +// Memory ordering is only relevant for the relative ordering of operations +// between different variables. Even Ordering::Relaxed guarantees a +// monotonic/consistent order when looking at just a single atomic variable. +// +// So, since this parker is just a single atomic variable, we only need to look +// at the ordering guarantees we need to provide to the 'outside world'. +// +// The only memory ordering guarantee that parking and unparking provide, is +// that things which happened before unpark() are visible on the thread +// returning from park() afterwards. Otherwise, it was effectively unparked +// before unpark() was called while still consuming the 'token'. +// +// In other words, unpark() needs to synchronize with the part of park() that +// consumes the token and returns. +// +// This is done with a release-acquire synchronization, by using +// Ordering::Release when writing NOTIFIED (the 'token') in unpark(), and using +// Ordering::Acquire when reading this state in park() after waking up. +impl Parker { + /// Construct the Windows parker. The UNIX parker implementation + /// requires this to happen in-place. + pub unsafe fn new_in_place(parker: *mut Parker) { + parker.write(Self { state: AtomicI8::new(EMPTY) }); + } + + // Assumes this is only called by the thread that owns the Parker, + // which means that `self.state != PARKED`. This implementation doesn't require `Pin`, + // but other implementations do. + pub unsafe fn park(self: Pin<&Self>) { + // Change NOTIFIED=>EMPTY or EMPTY=>PARKED, and directly return in the + // first case. + if self.state.fetch_sub(1, Acquire) == NOTIFIED { + return; + } + + #[cfg(target_vendor = "win7")] + if c::WaitOnAddress::option().is_none() { + return keyed_events::park(self); + } + + loop { + // Wait for something to happen, assuming it's still set to PARKED. + c::WaitOnAddress(self.ptr(), &PARKED as *const _ as c::LPVOID, 1, c::INFINITE); + // Change NOTIFIED=>EMPTY but leave PARKED alone. + if self.state.compare_exchange(NOTIFIED, EMPTY, Acquire, Acquire).is_ok() { + // Actually woken up by unpark(). + return; + } else { + // Spurious wake up. We loop to try again. + } + } + } + + // Assumes this is only called by the thread that owns the Parker, + // which means that `self.state != PARKED`. This implementation doesn't require `Pin`, + // but other implementations do. + pub unsafe fn park_timeout(self: Pin<&Self>, timeout: Duration) { + // Change NOTIFIED=>EMPTY or EMPTY=>PARKED, and directly return in the + // first case. + if self.state.fetch_sub(1, Acquire) == NOTIFIED { + return; + } + + #[cfg(target_vendor = "win7")] + if c::WaitOnAddress::option().is_none() { + return keyed_events::park_timeout(self, timeout); + } + + // Wait for something to happen, assuming it's still set to PARKED. + c::WaitOnAddress(self.ptr(), &PARKED as *const _ as c::LPVOID, 1, dur2timeout(timeout)); + // Set the state back to EMPTY (from either PARKED or NOTIFIED). + // Note that we don't just write EMPTY, but use swap() to also + // include an acquire-ordered read to synchronize with unpark()'s + // release-ordered write. + if self.state.swap(EMPTY, Acquire) == NOTIFIED { + // Actually woken up by unpark(). + } else { + // Timeout or spurious wake up. + // We return either way, because we can't easily tell if it was the + // timeout or not. + } + } + + // This implementation doesn't require `Pin`, but other implementations do. + pub fn unpark(self: Pin<&Self>) { + // Change PARKED=>NOTIFIED, EMPTY=>NOTIFIED, or NOTIFIED=>NOTIFIED, and + // wake the thread in the first case. + // + // Note that even NOTIFIED=>NOTIFIED results in a write. This is on + // purpose, to make sure every unpark() has a release-acquire ordering + // with park(). + if self.state.swap(NOTIFIED, Release) == PARKED { + unsafe { + #[cfg(target_vendor = "win7")] + if c::WakeByAddressSingle::option().is_none() { + return keyed_events::unpark(self); + } + c::WakeByAddressSingle(self.ptr()); + } + } + } + + fn ptr(&self) -> c::LPVOID { + core::ptr::addr_of!(self.state) as c::LPVOID + } +} + +#[cfg(target_vendor = "win7")] +mod keyed_events { + use super::{Parker, EMPTY, NOTIFIED}; + use crate::sys::c; + use core::pin::Pin; + use core::ptr; + use core::sync::atomic::{ + AtomicPtr, + Ordering::{Acquire, Relaxed}, + }; + use core::time::Duration; + + pub unsafe fn park(parker: Pin<&Parker>) { + // Wait for unpark() to produce this event. + c::NtWaitForKeyedEvent(keyed_event_handle(), parker.ptr(), 0, ptr::null_mut()); + // Set the state back to EMPTY (from either PARKED or NOTIFIED). + // Note that we don't just write EMPTY, but use swap() to also + // include an acquire-ordered read to synchronize with unpark()'s + // release-ordered write. + parker.state.swap(EMPTY, Acquire); + return; + } + pub unsafe fn park_timeout(parker: Pin<&Parker>, timeout: Duration) { + // Need to wait for unpark() using NtWaitForKeyedEvent. + let handle = keyed_event_handle(); + + // NtWaitForKeyedEvent uses a unit of 100ns, and uses negative + // values to indicate a relative time on the monotonic clock. + // This is documented here for the underlying KeWaitForSingleObject function: + // https://docs.microsoft.com/en-us/windows-hardware/drivers/ddi/wdm/nf-wdm-kewaitforsingleobject + let mut timeout = match i64::try_from((timeout.as_nanos() + 99) / 100) { + Ok(t) => -t, + Err(_) => i64::MIN, + }; + + // Wait for unpark() to produce this event. + let unparked = + c::NtWaitForKeyedEvent(handle, parker.ptr(), 0, &mut timeout) == c::STATUS_SUCCESS; + + // Set the state back to EMPTY (from either PARKED or NOTIFIED). + let prev_state = parker.state.swap(EMPTY, Acquire); + + if !unparked && prev_state == NOTIFIED { + // We were awoken by a timeout, not by unpark(), but the state + // was set to NOTIFIED, which means we *just* missed an + // unpark(), which is now blocked on us to wait for it. + // Wait for it to consume the event and unblock that thread. + c::NtWaitForKeyedEvent(handle, parker.ptr(), 0, ptr::null_mut()); + } + } + pub unsafe fn unpark(parker: Pin<&Parker>) { + // If we run NtReleaseKeyedEvent before the waiting thread runs + // NtWaitForKeyedEvent, this (shortly) blocks until we can wake it up. + // If the waiting thread wakes up before we run NtReleaseKeyedEvent + // (e.g. due to a timeout), this blocks until we do wake up a thread. + // To prevent this thread from blocking indefinitely in that case, + // park_impl() will, after seeing the state set to NOTIFIED after + // waking up, call NtWaitForKeyedEvent again to unblock us. + c::NtReleaseKeyedEvent(keyed_event_handle(), parker.ptr(), 0, ptr::null_mut()); + } + + fn keyed_event_handle() -> c::HANDLE { + const INVALID: c::HANDLE = ptr::without_provenance_mut(!0); + static HANDLE: AtomicPtr<crate::ffi::c_void> = AtomicPtr::new(INVALID); + match HANDLE.load(Relaxed) { + INVALID => { + let mut handle = c::INVALID_HANDLE_VALUE; + unsafe { + match c::NtCreateKeyedEvent( + &mut handle, + c::GENERIC_READ | c::GENERIC_WRITE, + ptr::null_mut(), + 0, + ) { + c::STATUS_SUCCESS => {} + r => panic!("Unable to create keyed event handle: error {r}"), + } + } + match HANDLE.compare_exchange(INVALID, handle, Relaxed, Relaxed) { + Ok(_) => handle, + Err(h) => { + // Lost the race to another thread initializing HANDLE before we did. + // Closing our handle and using theirs instead. + unsafe { + c::CloseHandle(handle); + } + h + } + } + } + handle => handle, + } + } +} diff --git a/library/std/src/sys/sync/thread_parking/xous.rs b/library/std/src/sys/sync/thread_parking/xous.rs new file mode 100644 index 00000000000..0bd0462d77d --- /dev/null +++ b/library/std/src/sys/sync/thread_parking/xous.rs @@ -0,0 +1,112 @@ +use crate::os::xous::ffi::{blocking_scalar, scalar}; +use crate::os::xous::services::{ticktimer_server, TicktimerScalar}; +use crate::pin::Pin; +use crate::ptr; +use crate::sync::atomic::{ + AtomicI8, + Ordering::{Acquire, Release}, +}; +use crate::time::Duration; + +const NOTIFIED: i8 = 1; +const EMPTY: i8 = 0; +const PARKED: i8 = -1; + +pub struct Parker { + state: AtomicI8, +} + +impl Parker { + pub unsafe fn new_in_place(parker: *mut Parker) { + unsafe { parker.write(Parker { state: AtomicI8::new(EMPTY) }) } + } + + fn index(&self) -> usize { + ptr::from_ref(self).addr() + } + + pub unsafe fn park(self: Pin<&Self>) { + // Change NOTIFIED to EMPTY and EMPTY to PARKED. + let state = self.state.fetch_sub(1, Acquire); + if state == NOTIFIED { + // The state has gone from NOTIFIED (1) to EMPTY (0) + return; + } + // The state has gone from EMPTY (0) to PARKED (-1) + assert!(state == EMPTY); + + // The state is now PARKED (-1). Wait until the `unpark` wakes us up. + blocking_scalar( + ticktimer_server(), + TicktimerScalar::WaitForCondition(self.index(), 0).into(), + ) + .expect("failed to send WaitForCondition command"); + + let state = self.state.swap(EMPTY, Acquire); + assert!(state == NOTIFIED || state == PARKED); + } + + pub unsafe fn park_timeout(self: Pin<&Self>, timeout: Duration) { + // Change NOTIFIED to EMPTY and EMPTY to PARKED. + let state = self.state.fetch_sub(1, Acquire); + if state == NOTIFIED { + // The state has gone from NOTIFIED (1) to EMPTY (0) + return; + } + // The state has gone from EMPTY (0) to PARKED (-1) + assert!(state == EMPTY); + + // A value of zero indicates an indefinite wait. Clamp the number of + // milliseconds to the allowed range. + let millis = usize::max(timeout.as_millis().try_into().unwrap_or(usize::MAX), 1); + + // The state is now PARKED (-1). Wait until the `unpark` wakes us up, + // or things time out. + let _was_timeout = blocking_scalar( + ticktimer_server(), + TicktimerScalar::WaitForCondition(self.index(), millis).into(), + ) + .expect("failed to send WaitForCondition command")[0] + != 0; + + let state = self.state.swap(EMPTY, Acquire); + assert!(state == PARKED || state == NOTIFIED); + } + + pub fn unpark(self: Pin<&Self>) { + // If the state is already `NOTIFIED`, then another thread has + // indicated it wants to wake up the target thread. + // + // If the state is `EMPTY` then there is nothing to wake up, and + // the target thread will immediately exit from `park()` the + // next time that function is called. + if self.state.swap(NOTIFIED, Release) != PARKED { + return; + } + + // The thread is parked, wake it up. Keep trying until we wake something up. + // This will happen when the `NotifyCondition` call returns the fact that + // 1 condition was notified. + // Alternately, keep going until the state is seen as `EMPTY`, indicating + // the thread woke up and kept going. This can happen when the Park + // times out before we can send the NotifyCondition message. + while blocking_scalar( + ticktimer_server(), + TicktimerScalar::NotifyCondition(self.index(), 1).into(), + ) + .expect("failed to send NotifyCondition command")[0] + == 0 + && self.state.load(Acquire) != EMPTY + { + // The target thread hasn't yet hit the `WaitForCondition` call. + // Yield to let the target thread run some more. + crate::thread::yield_now(); + } + } +} + +impl Drop for Parker { + fn drop(&mut self) { + scalar(ticktimer_server(), TicktimerScalar::FreeCondition(self.index()).into()).ok(); + } +} |