Rollup merge of #128184 - joboet:refactor_pthread_sync, r=workingjubilee
std: refactor `pthread`-based synchronization The non-trivial code for `pthread_condvar` is duplicated across the thread parking and the `Mutex`/`Condvar` implementations. This PR moves that code into `sys::pal`, which now exposes an `unsafe` wrapper type for `pthread_mutex_t` and `pthread_condvar_t`.
This commit is contained in:
Matthias Krüger
GitHub
commit
fe4c6e8657
11 changed files with 469 additions and 442 deletions
|
|
@ -27,6 +27,14 @@ pub mod thread;
|
|||
#[path = "../unix/time.rs"]
|
||||
pub mod time;
|
||||
|
||||
#[path = "../unix/sync"]
|
||||
pub mod sync {
|
||||
mod condvar;
|
||||
mod mutex;
|
||||
pub use condvar::Condvar;
|
||||
pub use mutex::Mutex;
|
||||
}
|
||||
|
||||
use crate::io::ErrorKind;
|
||||
|
||||
pub fn abort_internal() -> ! {
|
||||
|
|
|
|||
|
|
@ -27,6 +27,7 @@ pub mod pipe;
|
|||
pub mod process;
|
||||
pub mod stack_overflow;
|
||||
pub mod stdio;
|
||||
pub mod sync;
|
||||
pub mod thread;
|
||||
pub mod thread_parking;
|
||||
pub mod time;
|
||||
|
|
|
|||
172
library/std/src/sys/pal/unix/sync/condvar.rs
Normal file
172
library/std/src/sys/pal/unix/sync/condvar.rs
Normal file
|
|
@ -0,0 +1,172 @@
|
|||
use super::Mutex;
|
||||
use crate::cell::UnsafeCell;
|
||||
use crate::pin::Pin;
|
||||
#[cfg(not(target_os = "nto"))]
|
||||
use crate::sys::pal::time::TIMESPEC_MAX;
|
||||
#[cfg(target_os = "nto")]
|
||||
use crate::sys::pal::time::TIMESPEC_MAX_CAPPED;
|
||||
use crate::sys::pal::time::Timespec;
|
||||
use crate::time::Duration;
|
||||
|
||||
pub struct Condvar {
|
||||
inner: UnsafeCell<libc::pthread_cond_t>,
|
||||
}
|
||||
|
||||
impl Condvar {
|
||||
pub fn new() -> Condvar {
|
||||
Condvar { inner: UnsafeCell::new(libc::PTHREAD_COND_INITIALIZER) }
|
||||
}
|
||||
|
||||
#[inline]
|
||||
fn raw(&self) -> *mut libc::pthread_cond_t {
|
||||
self.inner.get()
|
||||
}
|
||||
|
||||
/// # Safety
|
||||
/// `init` must have been called on this instance.
|
||||
#[inline]
|
||||
pub unsafe fn notify_one(self: Pin<&Self>) {
|
||||
let r = unsafe { libc::pthread_cond_signal(self.raw()) };
|
||||
debug_assert_eq!(r, 0);
|
||||
}
|
||||
|
||||
/// # Safety
|
||||
/// `init` must have been called on this instance.
|
||||
#[inline]
|
||||
pub unsafe fn notify_all(self: Pin<&Self>) {
|
||||
let r = unsafe { libc::pthread_cond_broadcast(self.raw()) };
|
||||
debug_assert_eq!(r, 0);
|
||||
}
|
||||
|
||||
/// # Safety
|
||||
/// * `init` must have been called on this instance.
|
||||
/// * `mutex` must be locked by the current thread.
|
||||
/// * This condition variable may only be used with the same mutex.
|
||||
#[inline]
|
||||
pub unsafe fn wait(self: Pin<&Self>, mutex: Pin<&Mutex>) {
|
||||
let r = unsafe { libc::pthread_cond_wait(self.raw(), mutex.raw()) };
|
||||
debug_assert_eq!(r, 0);
|
||||
}
|
||||
|
||||
/// # Safety
|
||||
/// * `init` must have been called on this instance.
|
||||
/// * `mutex` must be locked by the current thread.
|
||||
/// * This condition variable may only be used with the same mutex.
|
||||
pub unsafe fn wait_timeout(&self, mutex: Pin<&Mutex>, dur: Duration) -> bool {
|
||||
let mutex = mutex.raw();
|
||||
|
||||
// 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 returns error 316.
|
||||
//
|
||||
// This program demonstrates the issue:
|
||||
// https://gist.github.com/stepancheg/198db4623a20aad2ad7cddb8fda4a63c
|
||||
//
|
||||
// To work around this issue, the timeout is clamped to 1000 years.
|
||||
#[cfg(target_vendor = "apple")]
|
||||
let dur = Duration::min(dur, Duration::from_secs(1000 * 365 * 86400));
|
||||
|
||||
let timeout = Timespec::now(Self::CLOCK).checked_add_duration(&dur);
|
||||
|
||||
#[cfg(not(target_os = "nto"))]
|
||||
let timeout = timeout.and_then(|t| t.to_timespec()).unwrap_or(TIMESPEC_MAX);
|
||||
|
||||
#[cfg(target_os = "nto")]
|
||||
let timeout = timeout.and_then(|t| t.to_timespec_capped()).unwrap_or(TIMESPEC_MAX_CAPPED);
|
||||
|
||||
let r = unsafe { libc::pthread_cond_timedwait(self.raw(), mutex, &timeout) };
|
||||
assert!(r == libc::ETIMEDOUT || r == 0);
|
||||
r == 0
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(not(any(
|
||||
target_os = "android",
|
||||
target_vendor = "apple",
|
||||
target_os = "espidf",
|
||||
target_os = "horizon",
|
||||
target_os = "l4re",
|
||||
target_os = "redox",
|
||||
target_os = "teeos",
|
||||
)))]
|
||||
impl Condvar {
|
||||
pub const PRECISE_TIMEOUT: bool = true;
|
||||
const CLOCK: libc::clockid_t = libc::CLOCK_MONOTONIC;
|
||||
|
||||
/// # Safety
|
||||
/// May only be called once per instance of `Self`.
|
||||
pub unsafe fn init(self: Pin<&mut Self>) {
|
||||
use crate::mem::MaybeUninit;
|
||||
|
||||
struct AttrGuard<'a>(pub &'a mut MaybeUninit<libc::pthread_condattr_t>);
|
||||
impl Drop for AttrGuard<'_> {
|
||||
fn drop(&mut self) {
|
||||
unsafe {
|
||||
let result = libc::pthread_condattr_destroy(self.0.as_mut_ptr());
|
||||
assert_eq!(result, 0);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
unsafe {
|
||||
let mut attr = MaybeUninit::<libc::pthread_condattr_t>::uninit();
|
||||
let r = libc::pthread_condattr_init(attr.as_mut_ptr());
|
||||
assert_eq!(r, 0);
|
||||
let attr = AttrGuard(&mut attr);
|
||||
let r = libc::pthread_condattr_setclock(attr.0.as_mut_ptr(), Self::CLOCK);
|
||||
assert_eq!(r, 0);
|
||||
let r = libc::pthread_cond_init(self.raw(), attr.0.as_ptr());
|
||||
assert_eq!(r, 0);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// `pthread_condattr_setclock` is unfortunately not supported on these platforms.
|
||||
#[cfg(any(
|
||||
target_os = "android",
|
||||
target_vendor = "apple",
|
||||
target_os = "espidf",
|
||||
target_os = "horizon",
|
||||
target_os = "l4re",
|
||||
target_os = "redox",
|
||||
target_os = "teeos",
|
||||
))]
|
||||
impl Condvar {
|
||||
pub const PRECISE_TIMEOUT: bool = false;
|
||||
const CLOCK: libc::clockid_t = libc::CLOCK_REALTIME;
|
||||
|
||||
/// # Safety
|
||||
/// May only be called once per instance of `Self`.
|
||||
pub unsafe fn init(self: Pin<&mut Self>) {
|
||||
if cfg!(any(target_os = "espidf", target_os = "horizon", target_os = "teeos")) {
|
||||
// NOTE: ESP-IDF's PTHREAD_COND_INITIALIZER support is not released yet
|
||||
// So on that platform, init() should always be called.
|
||||
//
|
||||
// Similar story for the 3DS (horizon) and for TEEOS.
|
||||
let r = unsafe { libc::pthread_cond_init(self.raw(), crate::ptr::null()) };
|
||||
assert_eq!(r, 0);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl !Unpin for Condvar {}
|
||||
|
||||
unsafe impl Sync for Condvar {}
|
||||
unsafe impl Send for Condvar {}
|
||||
|
||||
impl Drop for Condvar {
|
||||
#[inline]
|
||||
fn drop(&mut self) {
|
||||
let r = unsafe { libc::pthread_cond_destroy(self.raw()) };
|
||||
if cfg!(target_os = "dragonfly") {
|
||||
// On DragonFly pthread_cond_destroy() returns EINVAL if called on
|
||||
// a condvar that was just initialized with
|
||||
// libc::PTHREAD_COND_INITIALIZER. Once it is used or
|
||||
// pthread_cond_init() is called, this behaviour no longer occurs.
|
||||
debug_assert!(r == 0 || r == libc::EINVAL);
|
||||
} else {
|
||||
debug_assert_eq!(r, 0);
|
||||
}
|
||||
}
|
||||
}
|
||||
16
library/std/src/sys/pal/unix/sync/mod.rs
Normal file
16
library/std/src/sys/pal/unix/sync/mod.rs
Normal file
|
|
@ -0,0 +1,16 @@
|
|||
#![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",
|
||||
)))]
|
||||
#![forbid(unsafe_op_in_unsafe_fn)]
|
||||
|
||||
mod condvar;
|
||||
mod mutex;
|
||||
|
||||
pub use condvar::Condvar;
|
||||
pub use mutex::Mutex;
|
||||
135
library/std/src/sys/pal/unix/sync/mutex.rs
Normal file
135
library/std/src/sys/pal/unix/sync/mutex.rs
Normal file
|
|
@ -0,0 +1,135 @@
|
|||
use super::super::cvt_nz;
|
||||
use crate::cell::UnsafeCell;
|
||||
use crate::io::Error;
|
||||
use crate::mem::MaybeUninit;
|
||||
use crate::pin::Pin;
|
||||
|
||||
pub struct Mutex {
|
||||
inner: UnsafeCell<libc::pthread_mutex_t>,
|
||||
}
|
||||
|
||||
impl Mutex {
|
||||
pub fn new() -> Mutex {
|
||||
Mutex { inner: UnsafeCell::new(libc::PTHREAD_MUTEX_INITIALIZER) }
|
||||
}
|
||||
|
||||
pub(super) fn raw(&self) -> *mut libc::pthread_mutex_t {
|
||||
self.inner.get()
|
||||
}
|
||||
|
||||
/// # Safety
|
||||
/// May only be called once per instance of `Self`.
|
||||
pub unsafe fn init(self: Pin<&mut Self>) {
|
||||
// Issue #33770
|
||||
//
|
||||
// A pthread mutex initialized with PTHREAD_MUTEX_INITIALIZER will have
|
||||
// a type of PTHREAD_MUTEX_DEFAULT, which has undefined behavior if you
|
||||
// try to re-lock it from the same thread when you already hold a lock
|
||||
// (https://pubs.opengroup.org/onlinepubs/9699919799/functions/pthread_mutex_init.html).
|
||||
// This is the case even if PTHREAD_MUTEX_DEFAULT == PTHREAD_MUTEX_NORMAL
|
||||
// (https://github.com/rust-lang/rust/issues/33770#issuecomment-220847521) -- in that
|
||||
// case, `pthread_mutexattr_settype(PTHREAD_MUTEX_DEFAULT)` will of course be the same
|
||||
// as setting it to `PTHREAD_MUTEX_NORMAL`, but not setting any mode will result in
|
||||
// a Mutex where re-locking is UB.
|
||||
//
|
||||
// In practice, glibc takes advantage of this undefined behavior to
|
||||
// implement hardware lock elision, which uses hardware transactional
|
||||
// memory to avoid acquiring the lock. While a transaction is in
|
||||
// progress, the lock appears to be unlocked. This isn't a problem for
|
||||
// other threads since the transactional memory will abort if a conflict
|
||||
// is detected, however no abort is generated when re-locking from the
|
||||
// same thread.
|
||||
//
|
||||
// Since locking the same mutex twice will result in two aliasing &mut
|
||||
// references, we instead create the mutex with type
|
||||
// PTHREAD_MUTEX_NORMAL which is guaranteed to deadlock if we try to
|
||||
// re-lock it from the same thread, thus avoiding undefined behavior.
|
||||
unsafe {
|
||||
let mut attr = MaybeUninit::<libc::pthread_mutexattr_t>::uninit();
|
||||
cvt_nz(libc::pthread_mutexattr_init(attr.as_mut_ptr())).unwrap();
|
||||
let attr = AttrGuard(&mut attr);
|
||||
cvt_nz(libc::pthread_mutexattr_settype(
|
||||
attr.0.as_mut_ptr(),
|
||||
libc::PTHREAD_MUTEX_NORMAL,
|
||||
))
|
||||
.unwrap();
|
||||
cvt_nz(libc::pthread_mutex_init(self.raw(), attr.0.as_ptr())).unwrap();
|
||||
}
|
||||
}
|
||||
|
||||
/// # Safety
|
||||
/// * If `init` was not called on this instance, reentrant locking causes
|
||||
/// undefined behaviour.
|
||||
/// * Destroying a locked mutex causes undefined behaviour.
|
||||
pub unsafe fn lock(self: Pin<&Self>) {
|
||||
#[cold]
|
||||
#[inline(never)]
|
||||
fn fail(r: i32) -> ! {
|
||||
let error = Error::from_raw_os_error(r);
|
||||
panic!("failed to lock mutex: {error}");
|
||||
}
|
||||
|
||||
let r = unsafe { libc::pthread_mutex_lock(self.raw()) };
|
||||
// As we set the mutex type to `PTHREAD_MUTEX_NORMAL` above, we expect
|
||||
// the lock call to never fail. Unfortunately however, some platforms
|
||||
// (Solaris) do not conform to the standard, and instead always provide
|
||||
// deadlock detection. How kind of them! Unfortunately that means that
|
||||
// we need to check the error code here. To save us from UB on other
|
||||
// less well-behaved platforms in the future, we do it even on "good"
|
||||
// platforms like macOS. See #120147 for more context.
|
||||
if r != 0 {
|
||||
fail(r)
|
||||
}
|
||||
}
|
||||
|
||||
/// # Safety
|
||||
/// * If `init` was not called on this instance, reentrant locking causes
|
||||
/// undefined behaviour.
|
||||
/// * Destroying a locked mutex causes undefined behaviour.
|
||||
pub unsafe fn try_lock(self: Pin<&Self>) -> bool {
|
||||
unsafe { libc::pthread_mutex_trylock(self.raw()) == 0 }
|
||||
}
|
||||
|
||||
/// # Safety
|
||||
/// The mutex must be locked by the current thread.
|
||||
pub unsafe fn unlock(self: Pin<&Self>) {
|
||||
let r = unsafe { libc::pthread_mutex_unlock(self.raw()) };
|
||||
debug_assert_eq!(r, 0);
|
||||
}
|
||||
}
|
||||
|
||||
impl !Unpin for Mutex {}
|
||||
|
||||
unsafe impl Send for Mutex {}
|
||||
unsafe impl Sync for Mutex {}
|
||||
|
||||
impl Drop for Mutex {
|
||||
fn drop(&mut self) {
|
||||
// SAFETY:
|
||||
// If `lock` or `init` was called, the mutex must have been pinned, so
|
||||
// it is still at the same location. Otherwise, `inner` must contain
|
||||
// `PTHREAD_MUTEX_INITIALIZER`, which is valid at all locations. Thus,
|
||||
// this call always destroys a valid mutex.
|
||||
let r = unsafe { libc::pthread_mutex_destroy(self.raw()) };
|
||||
if cfg!(target_os = "dragonfly") {
|
||||
// On DragonFly pthread_mutex_destroy() returns EINVAL if called on a
|
||||
// mutex that was just initialized with libc::PTHREAD_MUTEX_INITIALIZER.
|
||||
// Once it is used (locked/unlocked) or pthread_mutex_init() is called,
|
||||
// this behaviour no longer occurs.
|
||||
debug_assert!(r == 0 || r == libc::EINVAL);
|
||||
} else {
|
||||
debug_assert_eq!(r, 0);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
struct AttrGuard<'a>(pub &'a mut MaybeUninit<libc::pthread_mutexattr_t>);
|
||||
|
||||
impl Drop for AttrGuard<'_> {
|
||||
fn drop(&mut self) {
|
||||
unsafe {
|
||||
let result = libc::pthread_mutexattr_destroy(self.0.as_mut_ptr());
|
||||
assert_eq!(result, 0);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -1,196 +1,88 @@
|
|||
use crate::cell::UnsafeCell;
|
||||
use crate::ptr;
|
||||
use crate::sync::atomic::AtomicPtr;
|
||||
use crate::sync::atomic::Ordering::Relaxed;
|
||||
use crate::sys::sync::{Mutex, OnceBox};
|
||||
#[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;
|
||||
#![forbid(unsafe_op_in_unsafe_fn)]
|
||||
|
||||
struct AllocatedCondvar(UnsafeCell<libc::pthread_cond_t>);
|
||||
use crate::pin::Pin;
|
||||
use crate::ptr;
|
||||
use crate::sync::atomic::AtomicUsize;
|
||||
use crate::sync::atomic::Ordering::Relaxed;
|
||||
use crate::sys::pal::sync as pal;
|
||||
use crate::sys::sync::{Mutex, OnceBox};
|
||||
use crate::time::{Duration, Instant};
|
||||
|
||||
pub struct Condvar {
|
||||
inner: OnceBox<AllocatedCondvar>,
|
||||
mutex: AtomicPtr<libc::pthread_mutex_t>,
|
||||
}
|
||||
|
||||
unsafe impl Send for AllocatedCondvar {}
|
||||
unsafe impl Sync for AllocatedCondvar {}
|
||||
|
||||
impl AllocatedCondvar {
|
||||
fn new() -> Box<Self> {
|
||||
let condvar = Box::new(AllocatedCondvar(UnsafeCell::new(libc::PTHREAD_COND_INITIALIZER)));
|
||||
|
||||
cfg_if::cfg_if! {
|
||||
if #[cfg(any(
|
||||
target_os = "l4re",
|
||||
target_os = "android",
|
||||
target_os = "redox",
|
||||
target_vendor = "apple",
|
||||
))] {
|
||||
// `pthread_condattr_setclock` is unfortunately not supported on these platforms.
|
||||
} else if #[cfg(any(target_os = "espidf", target_os = "horizon", target_os = "teeos"))] {
|
||||
// NOTE: ESP-IDF's PTHREAD_COND_INITIALIZER support is not released yet
|
||||
// So on that platform, init() should always be called
|
||||
// Moreover, that platform does not have pthread_condattr_setclock support,
|
||||
// hence that initialization should be skipped as well
|
||||
//
|
||||
// Similar story for the 3DS (horizon).
|
||||
let r = unsafe { libc::pthread_cond_init(condvar.0.get(), crate::ptr::null()) };
|
||||
assert_eq!(r, 0);
|
||||
} else {
|
||||
use crate::mem::MaybeUninit;
|
||||
let mut attr = MaybeUninit::<libc::pthread_condattr_t>::uninit();
|
||||
let r = unsafe { libc::pthread_condattr_init(attr.as_mut_ptr()) };
|
||||
assert_eq!(r, 0);
|
||||
let r = unsafe { libc::pthread_condattr_setclock(attr.as_mut_ptr(), libc::CLOCK_MONOTONIC) };
|
||||
assert_eq!(r, 0);
|
||||
let r = unsafe { libc::pthread_cond_init(condvar.0.get(), attr.as_ptr()) };
|
||||
assert_eq!(r, 0);
|
||||
let r = unsafe { libc::pthread_condattr_destroy(attr.as_mut_ptr()) };
|
||||
assert_eq!(r, 0);
|
||||
}
|
||||
}
|
||||
|
||||
condvar
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for AllocatedCondvar {
|
||||
#[inline]
|
||||
fn drop(&mut self) {
|
||||
let r = unsafe { libc::pthread_cond_destroy(self.0.get()) };
|
||||
if cfg!(target_os = "dragonfly") {
|
||||
// On DragonFly pthread_cond_destroy() returns EINVAL if called on
|
||||
// a condvar that was just initialized with
|
||||
// libc::PTHREAD_COND_INITIALIZER. Once it is used or
|
||||
// pthread_cond_init() is called, this behavior no longer occurs.
|
||||
debug_assert!(r == 0 || r == libc::EINVAL);
|
||||
} else {
|
||||
debug_assert_eq!(r, 0);
|
||||
}
|
||||
}
|
||||
cvar: OnceBox<pal::Condvar>,
|
||||
mutex: AtomicUsize,
|
||||
}
|
||||
|
||||
impl Condvar {
|
||||
pub const fn new() -> Condvar {
|
||||
Condvar { inner: OnceBox::new(), mutex: AtomicPtr::new(ptr::null_mut()) }
|
||||
}
|
||||
|
||||
fn get(&self) -> *mut libc::pthread_cond_t {
|
||||
self.inner.get_or_init(AllocatedCondvar::new).0.get()
|
||||
Condvar { cvar: OnceBox::new(), mutex: AtomicUsize::new(0) }
|
||||
}
|
||||
|
||||
#[inline]
|
||||
fn verify(&self, mutex: *mut libc::pthread_mutex_t) {
|
||||
// Relaxed is okay here because we never read through `self.addr`, and only use it to
|
||||
fn get(&self) -> Pin<&pal::Condvar> {
|
||||
self.cvar.get_or_init(|| {
|
||||
let mut cvar = Box::pin(pal::Condvar::new());
|
||||
// SAFETY: we only call `init` once per `pal::Condvar`, namely here.
|
||||
unsafe { cvar.as_mut().init() };
|
||||
cvar
|
||||
})
|
||||
}
|
||||
|
||||
#[inline]
|
||||
fn verify(&self, mutex: Pin<&pal::Mutex>) {
|
||||
let addr = ptr::from_ref::<pal::Mutex>(&mutex).addr();
|
||||
// Relaxed is okay here because we never read through `self.mutex`, and only use it to
|
||||
// compare addresses.
|
||||
match self.mutex.compare_exchange(ptr::null_mut(), mutex, Relaxed, Relaxed) {
|
||||
Ok(_) => {} // Stored the address
|
||||
Err(n) if n == mutex => {} // Lost a race to store the same address
|
||||
match self.mutex.compare_exchange(0, addr, Relaxed, Relaxed) {
|
||||
Ok(_) => {} // Stored the address
|
||||
Err(n) if n == addr => {} // Lost a race to store the same address
|
||||
_ => panic!("attempted to use a condition variable with two mutexes"),
|
||||
}
|
||||
}
|
||||
|
||||
#[inline]
|
||||
pub fn notify_one(&self) {
|
||||
let r = unsafe { libc::pthread_cond_signal(self.get()) };
|
||||
debug_assert_eq!(r, 0);
|
||||
// SAFETY: we called `init` above.
|
||||
unsafe { self.get().notify_one() }
|
||||
}
|
||||
|
||||
#[inline]
|
||||
pub fn notify_all(&self) {
|
||||
let r = unsafe { libc::pthread_cond_broadcast(self.get()) };
|
||||
debug_assert_eq!(r, 0);
|
||||
// SAFETY: we called `init` above.
|
||||
unsafe { self.get().notify_all() }
|
||||
}
|
||||
|
||||
#[inline]
|
||||
pub unsafe fn wait(&self, mutex: &Mutex) {
|
||||
let mutex = mutex.get_assert_locked();
|
||||
// SAFETY: the caller guarantees that the lock is owned, thus the mutex
|
||||
// must have been initialized already.
|
||||
let mutex = unsafe { mutex.pal.get_unchecked() };
|
||||
self.verify(mutex);
|
||||
let r = libc::pthread_cond_wait(self.get(), mutex);
|
||||
debug_assert_eq!(r, 0);
|
||||
// SAFETY: we called `init` above, we verified that this condition
|
||||
// variable is only used with `mutex` and the caller guarantees that
|
||||
// `mutex` is locked by the current thread.
|
||||
unsafe { self.get().wait(mutex) }
|
||||
}
|
||||
|
||||
// This implementation is used on systems that support pthread_condattr_setclock
|
||||
// where we configure condition variable to use monotonic clock (instead of
|
||||
// default system clock). This approach avoids all problems that result
|
||||
// from changes made to the system time.
|
||||
#[cfg(not(any(
|
||||
target_os = "android",
|
||||
target_os = "espidf",
|
||||
target_os = "horizon",
|
||||
target_vendor = "apple",
|
||||
)))]
|
||||
pub unsafe fn wait_timeout(&self, mutex: &Mutex, dur: Duration) -> bool {
|
||||
use crate::sys::time::Timespec;
|
||||
|
||||
let mutex = mutex.get_assert_locked();
|
||||
// SAFETY: the caller guarantees that the lock is owned, thus the mutex
|
||||
// must have been initialized already.
|
||||
let mutex = unsafe { mutex.pal.get_unchecked() };
|
||||
self.verify(mutex);
|
||||
|
||||
#[cfg(not(target_os = "nto"))]
|
||||
let timeout = Timespec::now(libc::CLOCK_MONOTONIC)
|
||||
.checked_add_duration(&dur)
|
||||
.and_then(|t| t.to_timespec())
|
||||
.unwrap_or(TIMESPEC_MAX);
|
||||
|
||||
#[cfg(target_os = "nto")]
|
||||
let timeout = Timespec::now(libc::CLOCK_MONOTONIC)
|
||||
.checked_add_duration(&dur)
|
||||
.and_then(|t| t.to_timespec_capped())
|
||||
.unwrap_or(TIMESPEC_MAX_CAPPED);
|
||||
|
||||
let r = libc::pthread_cond_timedwait(self.get(), mutex, &timeout);
|
||||
assert!(r == libc::ETIMEDOUT || r == 0);
|
||||
r == 0
|
||||
}
|
||||
|
||||
// This implementation is modeled after libcxx's condition_variable
|
||||
// https://github.com/llvm-mirror/libcxx/blob/release_35/src/condition_variable.cpp#L46
|
||||
// https://github.com/llvm-mirror/libcxx/blob/release_35/include/__mutex_base#L367
|
||||
#[cfg(any(
|
||||
target_os = "android",
|
||||
target_os = "espidf",
|
||||
target_os = "horizon",
|
||||
target_vendor = "apple",
|
||||
))]
|
||||
pub unsafe fn wait_timeout(&self, mutex: &Mutex, dur: Duration) -> bool {
|
||||
use crate::sys::time::SystemTime;
|
||||
use crate::time::Instant;
|
||||
|
||||
let mutex = mutex.get_assert_locked();
|
||||
self.verify(mutex);
|
||||
|
||||
// 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 returns 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 `wait_timeout`
|
||||
// because of spurious wakeups.
|
||||
let dur = Duration::min(dur, Duration::from_secs(1000 * 365 * 86400));
|
||||
|
||||
// pthread_cond_timedwait uses system time, but we want to report timeout
|
||||
// based on stable time.
|
||||
let now = Instant::now();
|
||||
|
||||
let timeout = SystemTime::now()
|
||||
.t
|
||||
.checked_add_duration(&dur)
|
||||
.and_then(|t| t.to_timespec())
|
||||
.unwrap_or(TIMESPEC_MAX);
|
||||
|
||||
let r = libc::pthread_cond_timedwait(self.get(), mutex, &timeout);
|
||||
debug_assert!(r == libc::ETIMEDOUT || r == 0);
|
||||
|
||||
// ETIMEDOUT is not a totally reliable method of determining timeout due
|
||||
// to clock shifts, so do the check ourselves
|
||||
now.elapsed() < dur
|
||||
if pal::Condvar::PRECISE_TIMEOUT {
|
||||
// SAFETY: we called `init` above, we verified that this condition
|
||||
// variable is only used with `mutex` and the caller guarantees that
|
||||
// `mutex` is locked by the current thread.
|
||||
unsafe { self.get().wait_timeout(mutex, dur) }
|
||||
} else {
|
||||
// Timeout reports are not reliable, so do the check ourselves.
|
||||
let now = Instant::now();
|
||||
// SAFETY: we called `init` above, we verified that this condition
|
||||
// variable is only used with `mutex` and the caller guarantees that
|
||||
// `mutex` is locked by the current thread.
|
||||
let woken = unsafe { self.get().wait_timeout(mutex, dur) };
|
||||
woken || now.elapsed() < dur
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -13,17 +13,19 @@ impl Condvar {
|
|||
}
|
||||
|
||||
fn get(&self) -> &SpinMutex<WaitVariable<()>> {
|
||||
self.inner.get_or_init(|| Box::new(SpinMutex::new(WaitVariable::new(()))))
|
||||
self.inner.get_or_init(|| Box::pin(SpinMutex::new(WaitVariable::new(())))).get_ref()
|
||||
}
|
||||
|
||||
#[inline]
|
||||
pub fn notify_one(&self) {
|
||||
let _ = WaitQueue::notify_one(self.get().lock());
|
||||
let guard = self.get().lock();
|
||||
let _ = WaitQueue::notify_one(guard);
|
||||
}
|
||||
|
||||
#[inline]
|
||||
pub fn notify_all(&self) {
|
||||
let _ = WaitQueue::notify_all(self.get().lock());
|
||||
let guard = self.get().lock();
|
||||
let _ = WaitQueue::notify_all(guard);
|
||||
}
|
||||
|
||||
pub unsafe fn wait(&self, mutex: &Mutex) {
|
||||
|
|
|
|||
|
|
@ -1,163 +1,66 @@
|
|||
use crate::cell::UnsafeCell;
|
||||
use crate::io::Error;
|
||||
use crate::mem::{MaybeUninit, forget};
|
||||
use crate::sys::cvt_nz;
|
||||
#![forbid(unsafe_op_in_unsafe_fn)]
|
||||
|
||||
use crate::mem::forget;
|
||||
use crate::pin::Pin;
|
||||
use crate::sys::pal::sync as pal;
|
||||
use crate::sys::sync::OnceBox;
|
||||
|
||||
struct AllocatedMutex(UnsafeCell<libc::pthread_mutex_t>);
|
||||
|
||||
pub struct Mutex {
|
||||
inner: OnceBox<AllocatedMutex>,
|
||||
}
|
||||
|
||||
unsafe impl Send for AllocatedMutex {}
|
||||
unsafe impl Sync for AllocatedMutex {}
|
||||
|
||||
impl AllocatedMutex {
|
||||
fn new() -> Box<Self> {
|
||||
let mutex = Box::new(AllocatedMutex(UnsafeCell::new(libc::PTHREAD_MUTEX_INITIALIZER)));
|
||||
|
||||
// Issue #33770
|
||||
//
|
||||
// A pthread mutex initialized with PTHREAD_MUTEX_INITIALIZER will have
|
||||
// a type of PTHREAD_MUTEX_DEFAULT, which has undefined behavior if you
|
||||
// try to re-lock it from the same thread when you already hold a lock
|
||||
// (https://pubs.opengroup.org/onlinepubs/9699919799/functions/pthread_mutex_init.html).
|
||||
// This is the case even if PTHREAD_MUTEX_DEFAULT == PTHREAD_MUTEX_NORMAL
|
||||
// (https://github.com/rust-lang/rust/issues/33770#issuecomment-220847521) -- in that
|
||||
// case, `pthread_mutexattr_settype(PTHREAD_MUTEX_DEFAULT)` will of course be the same
|
||||
// as setting it to `PTHREAD_MUTEX_NORMAL`, but not setting any mode will result in
|
||||
// a Mutex where re-locking is UB.
|
||||
//
|
||||
// In practice, glibc takes advantage of this undefined behavior to
|
||||
// implement hardware lock elision, which uses hardware transactional
|
||||
// memory to avoid acquiring the lock. While a transaction is in
|
||||
// progress, the lock appears to be unlocked. This isn't a problem for
|
||||
// other threads since the transactional memory will abort if a conflict
|
||||
// is detected, however no abort is generated when re-locking from the
|
||||
// same thread.
|
||||
//
|
||||
// Since locking the same mutex twice will result in two aliasing &mut
|
||||
// references, we instead create the mutex with type
|
||||
// PTHREAD_MUTEX_NORMAL which is guaranteed to deadlock if we try to
|
||||
// re-lock it from the same thread, thus avoiding undefined behavior.
|
||||
unsafe {
|
||||
let mut attr = MaybeUninit::<libc::pthread_mutexattr_t>::uninit();
|
||||
cvt_nz(libc::pthread_mutexattr_init(attr.as_mut_ptr())).unwrap();
|
||||
let attr = PthreadMutexAttr(&mut attr);
|
||||
cvt_nz(libc::pthread_mutexattr_settype(
|
||||
attr.0.as_mut_ptr(),
|
||||
libc::PTHREAD_MUTEX_NORMAL,
|
||||
))
|
||||
.unwrap();
|
||||
cvt_nz(libc::pthread_mutex_init(mutex.0.get(), attr.0.as_ptr())).unwrap();
|
||||
}
|
||||
|
||||
mutex
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for AllocatedMutex {
|
||||
#[inline]
|
||||
fn drop(&mut self) {
|
||||
let r = unsafe { libc::pthread_mutex_destroy(self.0.get()) };
|
||||
if cfg!(target_os = "dragonfly") {
|
||||
// On DragonFly pthread_mutex_destroy() returns EINVAL if called on a
|
||||
// mutex that was just initialized with libc::PTHREAD_MUTEX_INITIALIZER.
|
||||
// Once it is used (locked/unlocked) or pthread_mutex_init() is called,
|
||||
// this behavior no longer occurs.
|
||||
debug_assert!(r == 0 || r == libc::EINVAL);
|
||||
} else {
|
||||
debug_assert_eq!(r, 0);
|
||||
}
|
||||
}
|
||||
pub pal: OnceBox<pal::Mutex>,
|
||||
}
|
||||
|
||||
impl Mutex {
|
||||
#[inline]
|
||||
pub const fn new() -> Mutex {
|
||||
Mutex { inner: OnceBox::new() }
|
||||
}
|
||||
|
||||
/// Gets access to the pthread mutex under the assumption that the mutex is
|
||||
/// locked.
|
||||
///
|
||||
/// This allows skipping the initialization check, as the mutex can only be
|
||||
/// locked if it is already initialized, and allows relaxing the ordering
|
||||
/// on the pointer load, since the allocation cannot have been modified
|
||||
/// since the `lock` and the lock must have occurred on the current thread.
|
||||
///
|
||||
/// # Safety
|
||||
/// Causes undefined behavior if the mutex is not locked.
|
||||
#[inline]
|
||||
pub(crate) unsafe fn get_assert_locked(&self) -> *mut libc::pthread_mutex_t {
|
||||
unsafe { self.inner.get_unchecked().0.get() }
|
||||
Mutex { pal: OnceBox::new() }
|
||||
}
|
||||
|
||||
#[inline]
|
||||
fn get(&self) -> *mut libc::pthread_mutex_t {
|
||||
// If initialization fails, the mutex is destroyed. This is always sound,
|
||||
// however, as the mutex cannot have been locked yet.
|
||||
self.inner.get_or_init(AllocatedMutex::new).0.get()
|
||||
fn get(&self) -> Pin<&pal::Mutex> {
|
||||
// If the initialization race is lost, the new mutex is destroyed.
|
||||
// This is sound however, as it cannot have been locked.
|
||||
self.pal.get_or_init(|| {
|
||||
let mut pal = Box::pin(pal::Mutex::new());
|
||||
// SAFETY: we only call `init` once per `pal::Mutex`, namely here.
|
||||
unsafe { pal.as_mut().init() };
|
||||
pal
|
||||
})
|
||||
}
|
||||
|
||||
#[inline]
|
||||
pub fn lock(&self) {
|
||||
#[cold]
|
||||
#[inline(never)]
|
||||
fn fail(r: i32) -> ! {
|
||||
let error = Error::from_raw_os_error(r);
|
||||
panic!("failed to lock mutex: {error}");
|
||||
}
|
||||
|
||||
let r = unsafe { libc::pthread_mutex_lock(self.get()) };
|
||||
// As we set the mutex type to `PTHREAD_MUTEX_NORMAL` above, we expect
|
||||
// the lock call to never fail. Unfortunately however, some platforms
|
||||
// (Solaris) do not conform to the standard, and instead always provide
|
||||
// deadlock detection. How kind of them! Unfortunately that means that
|
||||
// we need to check the error code here. To save us from UB on other
|
||||
// less well-behaved platforms in the future, we do it even on "good"
|
||||
// platforms like macOS. See #120147 for more context.
|
||||
if r != 0 {
|
||||
fail(r)
|
||||
}
|
||||
// SAFETY: we call `init` above, therefore reentrant locking is safe.
|
||||
// In `drop` we ensure that the mutex is not destroyed while locked.
|
||||
unsafe { self.get().lock() }
|
||||
}
|
||||
|
||||
#[inline]
|
||||
pub unsafe fn unlock(&self) {
|
||||
let r = libc::pthread_mutex_unlock(self.get_assert_locked());
|
||||
debug_assert_eq!(r, 0);
|
||||
// SAFETY: the mutex can only be locked if it is already initialized
|
||||
// and we observed this initialization since we observed the locking.
|
||||
unsafe { self.pal.get_unchecked().unlock() }
|
||||
}
|
||||
|
||||
#[inline]
|
||||
pub fn try_lock(&self) -> bool {
|
||||
unsafe { libc::pthread_mutex_trylock(self.get()) == 0 }
|
||||
// SAFETY: we call `init` above, therefore reentrant locking is safe.
|
||||
// In `drop` we ensure that the mutex is not destroyed while locked.
|
||||
unsafe { self.get().try_lock() }
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for Mutex {
|
||||
fn drop(&mut self) {
|
||||
let Some(mutex) = self.inner.take() else { return };
|
||||
let Some(pal) = self.pal.take() else { return };
|
||||
// We're not allowed to pthread_mutex_destroy a locked mutex,
|
||||
// so check first if it's unlocked.
|
||||
if unsafe { libc::pthread_mutex_trylock(mutex.0.get()) == 0 } {
|
||||
unsafe { libc::pthread_mutex_unlock(mutex.0.get()) };
|
||||
drop(mutex);
|
||||
if unsafe { pal.as_ref().try_lock() } {
|
||||
unsafe { pal.as_ref().unlock() };
|
||||
drop(pal)
|
||||
} else {
|
||||
// The mutex is locked. This happens if a MutexGuard is leaked.
|
||||
// In this case, we just leak the Mutex too.
|
||||
forget(mutex);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub(super) struct PthreadMutexAttr<'a>(pub &'a mut MaybeUninit<libc::pthread_mutexattr_t>);
|
||||
|
||||
impl Drop for PthreadMutexAttr<'_> {
|
||||
fn drop(&mut self) {
|
||||
unsafe {
|
||||
let result = libc::pthread_mutexattr_destroy(self.0.as_mut_ptr());
|
||||
debug_assert_eq!(result, 0);
|
||||
forget(pal)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -13,7 +13,7 @@ impl Mutex {
|
|||
}
|
||||
|
||||
fn get(&self) -> &SpinMutex<WaitVariable<bool>> {
|
||||
self.inner.get_or_init(|| Box::new(SpinMutex::new(WaitVariable::new(false))))
|
||||
self.inner.get_or_init(|| Box::pin(SpinMutex::new(WaitVariable::new(false)))).get_ref()
|
||||
}
|
||||
|
||||
#[inline]
|
||||
|
|
@ -33,7 +33,7 @@ impl Mutex {
|
|||
pub unsafe fn unlock(&self) {
|
||||
// SAFETY: the mutex was locked by the current thread, so it has been
|
||||
// initialized already.
|
||||
let guard = unsafe { self.inner.get_unchecked().lock() };
|
||||
let guard = unsafe { self.inner.get_unchecked().get_ref().lock() };
|
||||
if let Err(mut guard) = WaitQueue::notify_one(guard) {
|
||||
// No other waiters, unlock
|
||||
*guard.lock_var_mut() = false;
|
||||
|
|
|
|||
|
|
@ -6,6 +6,7 @@
|
|||
#![allow(dead_code)] // Only used on some platforms.
|
||||
|
||||
use crate::mem::replace;
|
||||
use crate::pin::Pin;
|
||||
use crate::ptr::null_mut;
|
||||
use crate::sync::atomic::AtomicPtr;
|
||||
use crate::sync::atomic::Ordering::{Acquire, Relaxed, Release};
|
||||
|
|
@ -27,46 +28,46 @@ impl<T> OnceBox<T> {
|
|||
/// pointer load in this function can be performed with relaxed ordering,
|
||||
/// potentially allowing the optimizer to turn code like this:
|
||||
/// ```rust, ignore
|
||||
/// once_box.get_or_init(|| Box::new(42));
|
||||
/// once_box.get_or_init(|| Box::pin(42));
|
||||
/// unsafe { once_box.get_unchecked() }
|
||||
/// ```
|
||||
/// into
|
||||
/// ```rust, ignore
|
||||
/// once_box.get_or_init(|| Box::new(42))
|
||||
/// once_box.get_or_init(|| Box::pin(42))
|
||||
/// ```
|
||||
///
|
||||
/// # Safety
|
||||
/// This causes undefined behavior if the assumption above is violated.
|
||||
#[inline]
|
||||
pub unsafe fn get_unchecked(&self) -> &T {
|
||||
unsafe { &*self.ptr.load(Relaxed) }
|
||||
pub unsafe fn get_unchecked(&self) -> Pin<&T> {
|
||||
unsafe { Pin::new_unchecked(&*self.ptr.load(Relaxed)) }
|
||||
}
|
||||
|
||||
#[inline]
|
||||
pub fn get_or_init(&self, f: impl FnOnce() -> Box<T>) -> &T {
|
||||
pub fn get_or_init(&self, f: impl FnOnce() -> Pin<Box<T>>) -> Pin<&T> {
|
||||
let ptr = self.ptr.load(Acquire);
|
||||
match unsafe { ptr.as_ref() } {
|
||||
Some(val) => val,
|
||||
Some(val) => unsafe { Pin::new_unchecked(val) },
|
||||
None => self.initialize(f),
|
||||
}
|
||||
}
|
||||
|
||||
#[inline]
|
||||
pub fn take(&mut self) -> Option<Box<T>> {
|
||||
pub fn take(&mut self) -> Option<Pin<Box<T>>> {
|
||||
let ptr = replace(self.ptr.get_mut(), null_mut());
|
||||
if !ptr.is_null() { Some(unsafe { Box::from_raw(ptr) }) } else { None }
|
||||
if !ptr.is_null() { Some(unsafe { Pin::new_unchecked(Box::from_raw(ptr)) }) } else { None }
|
||||
}
|
||||
|
||||
#[cold]
|
||||
fn initialize(&self, f: impl FnOnce() -> Box<T>) -> &T {
|
||||
let new_ptr = Box::into_raw(f());
|
||||
fn initialize(&self, f: impl FnOnce() -> Pin<Box<T>>) -> Pin<&T> {
|
||||
let new_ptr = Box::into_raw(unsafe { Pin::into_inner_unchecked(f()) });
|
||||
match self.ptr.compare_exchange(null_mut(), new_ptr, Release, Acquire) {
|
||||
Ok(_) => unsafe { &*new_ptr },
|
||||
Ok(_) => unsafe { Pin::new_unchecked(&*new_ptr) },
|
||||
Err(ptr) => {
|
||||
// Lost the race to another thread.
|
||||
// Drop the value we created, and use the one from the other thread instead.
|
||||
drop(unsafe { Box::from_raw(new_ptr) });
|
||||
unsafe { &*ptr }
|
||||
unsafe { Pin::new_unchecked(&*ptr) }
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,93 +1,19 @@
|
|||
//! Thread parking without `futex` using the `pthread` synchronization primitives.
|
||||
|
||||
use crate::cell::UnsafeCell;
|
||||
use crate::marker::PhantomPinned;
|
||||
use crate::pin::Pin;
|
||||
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::sys::pal::sync::{Condvar, Mutex};
|
||||
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,
|
||||
lock: Mutex,
|
||||
cvar: Condvar,
|
||||
}
|
||||
|
||||
impl Parker {
|
||||
|
|
@ -96,38 +22,21 @@ impl Parker {
|
|||
/// # 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 behavior 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.
|
||||
(&raw mut (*parker).state).write(AtomicUsize::new(EMPTY));
|
||||
(&raw mut (*parker).lock).write(UnsafeCell::new(libc::PTHREAD_MUTEX_INITIALIZER));
|
||||
parker.write(Parker {
|
||||
state: AtomicUsize::new(EMPTY),
|
||||
lock: Mutex::new(),
|
||||
cvar: Condvar::new(),
|
||||
});
|
||||
|
||||
cfg_if::cfg_if! {
|
||||
if #[cfg(any(
|
||||
target_os = "l4re",
|
||||
target_os = "android",
|
||||
target_os = "redox",
|
||||
target_os = "vita",
|
||||
target_vendor = "apple",
|
||||
))] {
|
||||
(&raw 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((&raw 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((&raw 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);
|
||||
}
|
||||
}
|
||||
Pin::new_unchecked(&mut (*parker).cvar).init();
|
||||
}
|
||||
|
||||
fn lock(self: Pin<&Self>) -> Pin<&Mutex> {
|
||||
unsafe { self.map_unchecked(|p| &p.lock) }
|
||||
}
|
||||
|
||||
fn cvar(self: Pin<&Self>) -> Pin<&Condvar> {
|
||||
unsafe { self.map_unchecked(|p| &p.cvar) }
|
||||
}
|
||||
|
||||
// This implementation doesn't require `unsafe`, but other implementations
|
||||
|
|
@ -142,7 +51,7 @@ impl Parker {
|
|||
}
|
||||
|
||||
// Otherwise we need to coordinate going to sleep
|
||||
lock(self.lock.get());
|
||||
self.lock().lock();
|
||||
match self.state.compare_exchange(EMPTY, PARKED, Relaxed, Relaxed) {
|
||||
Ok(_) => {}
|
||||
Err(NOTIFIED) => {
|
||||
|
|
@ -154,20 +63,20 @@ impl Parker {
|
|||
// read from the write it made to `state`.
|
||||
let old = self.state.swap(EMPTY, Acquire);
|
||||
|
||||
unlock(self.lock.get());
|
||||
self.lock().unlock();
|
||||
|
||||
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());
|
||||
self.lock().unlock();
|
||||
|
||||
panic!("inconsistent park state")
|
||||
}
|
||||
}
|
||||
|
||||
loop {
|
||||
wait(self.cvar.get(), self.lock.get());
|
||||
self.cvar().wait(self.lock());
|
||||
|
||||
match self.state.compare_exchange(NOTIFIED, EMPTY, Acquire, Relaxed) {
|
||||
Ok(_) => break, // got a notification
|
||||
|
|
@ -175,7 +84,7 @@ impl Parker {
|
|||
}
|
||||
}
|
||||
|
||||
unlock(self.lock.get());
|
||||
self.lock().unlock();
|
||||
}
|
||||
|
||||
// This implementation doesn't require `unsafe`, but other implementations
|
||||
|
|
@ -189,19 +98,19 @@ impl Parker {
|
|||
return;
|
||||
}
|
||||
|
||||
lock(self.lock.get());
|
||||
self.lock().lock();
|
||||
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());
|
||||
self.lock().unlock();
|
||||
|
||||
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());
|
||||
self.lock().unlock();
|
||||
panic!("inconsistent park_timeout state")
|
||||
}
|
||||
}
|
||||
|
|
@ -210,13 +119,13 @@ impl Parker {
|
|||
// 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);
|
||||
self.cvar().wait_timeout(self.lock(), dur);
|
||||
|
||||
match self.state.swap(EMPTY, Acquire) {
|
||||
NOTIFIED => unlock(self.lock.get()), // got a notification, hurray!
|
||||
PARKED => unlock(self.lock.get()), // no notification, alas
|
||||
NOTIFIED => self.lock().unlock(), // got a notification, hurray!
|
||||
PARKED => self.lock().unlock(), // no notification, alas
|
||||
n => {
|
||||
unlock(self.lock.get());
|
||||
self.lock().unlock();
|
||||
panic!("inconsistent park_timeout state: {n}")
|
||||
}
|
||||
}
|
||||
|
|
@ -248,21 +157,9 @@ impl Parker {
|
|||
// 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());
|
||||
self.lock().lock();
|
||||
self.lock().unlock();
|
||||
self.cvar().notify_one();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
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 {}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue