From 64d98f8ee037282c35007b64c2649055c56af1db Mon Sep 17 00:00:00 2001 From: Daniel Baumann Date: Wed, 17 Apr 2024 14:19:03 +0200 Subject: Merging upstream version 1.68.2+dfsg1. Signed-off-by: Daniel Baumann --- library/std/src/sys/unix/thread_parking/darwin.rs | 131 ++++++++++ library/std/src/sys/unix/thread_parking/mod.rs | 32 +++ library/std/src/sys/unix/thread_parking/netbsd.rs | 52 ++++ library/std/src/sys/unix/thread_parking/pthread.rs | 271 +++++++++++++++++++++ 4 files changed, 486 insertions(+) create mode 100644 library/std/src/sys/unix/thread_parking/darwin.rs create mode 100644 library/std/src/sys/unix/thread_parking/mod.rs create mode 100644 library/std/src/sys/unix/thread_parking/netbsd.rs create mode 100644 library/std/src/sys/unix/thread_parking/pthread.rs (limited to 'library/std/src/sys/unix/thread_parking') diff --git a/library/std/src/sys/unix/thread_parking/darwin.rs b/library/std/src/sys/unix/thread_parking/darwin.rs new file mode 100644 index 000000000..b709fada3 --- /dev/null +++ b/library/std/src/sys/unix/thread_parking/darwin.rs @@ -0,0 +1,131 @@ +//! 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) and because they are only available starting +//! with macOS version 10.12, even though the minimum target version is 10.7. +//! +//! 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. + +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/unix/thread_parking/mod.rs b/library/std/src/sys/unix/thread_parking/mod.rs new file mode 100644 index 000000000..185333c07 --- /dev/null +++ b/library/std/src/sys/unix/thread_parking/mod.rs @@ -0,0 +1,32 @@ +//! Thread parking on systems without futex support. + +#![cfg(not(any( + target_os = "linux", + target_os = "android", + all(target_os = "emscripten", target_feature = "atomics"), + target_os = "freebsd", + target_os = "openbsd", + target_os = "dragonfly", + target_os = "fuchsia", +)))] + +cfg_if::cfg_if! { + if #[cfg(all( + any( + target_os = "macos", + target_os = "ios", + target_os = "watchos", + target_os = "tvos", + ), + not(miri), + ))] { + mod darwin; + pub use darwin::Parker; + } else if #[cfg(target_os = "netbsd")] { + mod netbsd; + pub use netbsd::{current, park, park_timeout, unpark, ThreadId}; + } else { + mod pthread; + pub use pthread::Parker; + } +} diff --git a/library/std/src/sys/unix/thread_parking/netbsd.rs b/library/std/src/sys/unix/thread_parking/netbsd.rs new file mode 100644 index 000000000..3be081221 --- /dev/null +++ b/library/std/src/sys/unix/thread_parking/netbsd.rs @@ -0,0 +1,52 @@ +use crate::ffi::{c_int, c_void}; +use crate::ptr; +use crate::time::Duration; +use libc::{_lwp_self, clockid_t, lwpid_t, time_t, timespec, CLOCK_MONOTONIC}; + +extern "C" { + fn ___lwp_park60( + clock_id: clockid_t, + flags: c_int, + ts: *mut timespec, + unpark: lwpid_t, + hint: *const c_void, + unparkhint: *const c_void, + ) -> c_int; + fn _lwp_unpark(lwp: lwpid_t, hint: *const c_void) -> c_int; +} + +pub type ThreadId = lwpid_t; + +#[inline] +pub fn current() -> ThreadId { + unsafe { _lwp_self() } +} + +#[inline] +pub fn park(hint: usize) { + unsafe { + ___lwp_park60(0, 0, ptr::null_mut(), 0, ptr::invalid(hint), ptr::null()); + } +} + +pub fn park_timeout(dur: Duration, hint: usize) { + let mut timeout = timespec { + // Saturate so that the operation will definitely time out + // (even if it is after the heat death of the universe). + tv_sec: dur.as_secs().try_into().ok().unwrap_or(time_t::MAX), + tv_nsec: dur.subsec_nanos().into(), + }; + + // Timeout needs to be mutable since it is modified on NetBSD 9.0 and + // above. + unsafe { + ___lwp_park60(CLOCK_MONOTONIC, 0, &mut timeout, 0, ptr::invalid(hint), ptr::null()); + } +} + +#[inline] +pub fn unpark(tid: ThreadId, hint: usize) { + unsafe { + _lwp_unpark(tid, ptr::invalid(hint)); + } +} diff --git a/library/std/src/sys/unix/thread_parking/pthread.rs b/library/std/src/sys/unix/thread_parking/pthread.rs new file mode 100644 index 000000000..082d25e68 --- /dev/null +++ b/library/std/src/sys/unix/thread_parking/pthread.rs @@ -0,0 +1,271 @@ +//! 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::SeqCst; +use crate::sys::time::TIMESPEC_MAX; +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 = "macos", + target_os = "ios", + target_os = "watchos", + target_os = "espidf", + target_os = "horizon", + ))] + 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 = "macos", + target_os = "ios", + target_os = "watchos", + target_os = "espidf", + target_os = "horizon", + )))] + let (now, dur) = { + use crate::sys::time::Timespec; + + (Timespec::now(libc::CLOCK_MONOTONIC), dur) + }; + + let timeout = + now.checked_add_duration(&dur).and_then(|t| t.to_timespec()).unwrap_or(TIMESPEC_MAX); + let r = libc::pthread_cond_timedwait(cond, lock, &timeout); + debug_assert!(r == libc::ETIMEDOUT || r == 0); +} + +pub struct Parker { + state: AtomicUsize, + lock: UnsafeCell, + cvar: UnsafeCell, + // 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 = "macos", + target_os = "ios", + target_os = "watchos", + target_os = "l4re", + target_os = "android", + target_os = "redox" + ))] { + 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::::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. + 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, SeqCst, SeqCst).is_ok() { + return; + } + + // Otherwise we need to coordinate going to sleep + lock(self.lock.get()); + match self.state.compare_exchange(EMPTY, PARKED, SeqCst, SeqCst) { + 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, SeqCst); + + 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, SeqCst, SeqCst) { + 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, SeqCst, SeqCst).is_ok() { + return; + } + + lock(self.lock.get()); + match self.state.compare_exchange(EMPTY, PARKED, SeqCst, SeqCst) { + Ok(_) => {} + Err(NOTIFIED) => { + // We must read again here, see `park`. + let old = self.state.swap(EMPTY, SeqCst); + 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, SeqCst) { + 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, SeqCst) { + 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 {} -- cgit v1.2.3