Rollup merge of #105903 - joboet:unify_parking, r=m-ou-se

Unify id-based thread parking implementations

Multiple platforms currently use thread-id-based parking implementations (NetBSD and SGX[^1]). Even though the strategy does not differ, these are duplicated for each platform, as the id is encoded into an atomic thread variable in different ways for each platform.

Since `park` is only called by one thread, it is possible to move the thread id into a separate field. By ensuring that the field is only written to once, before any other threads access it, these accesses can be unsynchronized, removing any restrictions on the size and niches of the thread id.

This PR also renames the internal `thread_parker` modules to `thread_parking`, as that name now better reflects their contents. I hope this does not add too much reviewing noise.

r? `@m-ou-se`

`@rustbot` label +T-libs

[^1]: SOLID supports this as well, I will switch it over in a follow-up PR.

12 files changed, 86 insertions, 231 deletions

diff --git a/library/std/src/sys/sgx/mod.rs b/library/std/src/sys/sgx/mod.rs
index 63e070207cd..9865a945bad 100644
--- a/library/std/src/sys/sgx/mod.rs
+++ b/library/std/src/sys/sgx/mod.rs
@@ -34,7 +34,7 @@ pub mod process;
 pub mod stdio;
 pub mod thread;
 pub mod thread_local_key;
-pub mod thread_parker;
+pub mod thread_parking;
 pub mod time;
 
 mod condvar;
diff --git a/library/std/src/sys/sgx/thread.rs b/library/std/src/sys/sgx/thread.rs
index 579f758c6cc..1608b8cb642 100644
--- a/library/std/src/sys/sgx/thread.rs
+++ b/library/std/src/sys/sgx/thread.rs
@@ -65,9 +65,9 @@ mod task_queue {
 /// execution. The signal is sent once all TLS destructors have finished at
 /// which point no new thread locals should be created.
 pub mod wait_notify {
-    use super::super::thread_parker::Parker;
     use crate::pin::Pin;
     use crate::sync::Arc;
+    use crate::sys_common::thread_parking::Parker;
 
     pub struct Notifier(Arc<Parker>);
 
@@ -87,14 +87,14 @@ pub mod wait_notify {
         /// called, this will return immediately, otherwise the current thread
         /// is blocked until notified.
         pub fn wait(self) {
-            // This is not actually `unsafe`, but it uses the `Parker` API,
-            // which needs `unsafe` on some platforms.
+            // SAFETY:
+            // This is only ever called on one thread.
             unsafe { Pin::new(&*self.0).park() }
         }
     }
 
     pub fn new() -> (Notifier, Waiter) {
-        let inner = Arc::new(Parker::new_internal());
+        let inner = Arc::new(Parker::new());
         (Notifier(inner.clone()), Waiter(inner))
     }
 }
diff --git a/library/std/src/sys/sgx/thread_parker.rs b/library/std/src/sys/sgx/thread_parker.rs
deleted file mode 100644
index 1c55bcffb1e..00000000000
--- a/library/std/src/sys/sgx/thread_parker.rs
+++ /dev/null
@@ -1,107 +0,0 @@
-//! Thread parking based on SGX events.
-
-use super::abi::{thread, usercalls};
-use crate::io::ErrorKind;
-use crate::pin::Pin;
-use crate::ptr::{self, NonNull};
-use crate::sync::atomic::AtomicPtr;
-use crate::sync::atomic::Ordering::{Acquire, Relaxed, Release};
-use crate::time::Duration;
-use fortanix_sgx_abi::{EV_UNPARK, WAIT_INDEFINITE};
-
-// The TCS structure must be page-aligned (this is checked by EENTER), so these cannot
-// be valid pointers
-const EMPTY: *mut u8 = ptr::invalid_mut(1);
-const NOTIFIED: *mut u8 = ptr::invalid_mut(2);
-
-pub struct Parker {
-    /// The park state. One of EMPTY, NOTIFIED or a TCS address.
-    /// A state change to NOTIFIED must be done with release ordering
-    /// and be observed with acquire ordering so that operations after
-    /// `thread::park` returns will not occur before the unpark message
-    /// was sent.
-    state: AtomicPtr<u8>,
-}
-
-impl Parker {
-    /// Construct the thread parker. The UNIX parker implementation
-    /// requires this to happen in-place.
-    pub unsafe fn new(parker: *mut Parker) {
-        unsafe { parker.write(Parker::new_internal()) }
-    }
-
-    pub(super) fn new_internal() -> Parker {
-        Parker { state: AtomicPtr::new(EMPTY) }
-    }
-
-    // This implementation doesn't require `unsafe` and `Pin`, but other implementations do.
-    pub unsafe fn park(self: Pin<&Self>) {
-        if self.state.load(Acquire) != NOTIFIED {
-            let mut prev = EMPTY;
-            loop {
-                // Guard against changing TCS addresses by always setting the state to
-                // the current value.
-                let tcs = thread::current().as_ptr();
-                if self.state.compare_exchange(prev, tcs, Relaxed, Acquire).is_ok() {
-                    let event = usercalls::wait(EV_UNPARK, WAIT_INDEFINITE).unwrap();
-                    assert!(event & EV_UNPARK == EV_UNPARK);
-                    prev = tcs;
-                } else {
-                    // The state was definitely changed by another thread at this point.
-                    // The only time this occurs is when the state is changed to NOTIFIED.
-                    // We observed this change with acquire ordering, so we can simply
-                    // change the state to EMPTY with a relaxed store.
-                    break;
-                }
-            }
-        }
-
-        // At this point, the token was definately read with acquire ordering,
-        // so this can be a relaxed store.
-        self.state.store(EMPTY, Relaxed);
-    }
-
-    // This implementation doesn't require `unsafe` and `Pin`, but other implementations do.
-    pub unsafe fn park_timeout(self: Pin<&Self>, dur: Duration) {
-        let timeout = u128::min(dur.as_nanos(), WAIT_INDEFINITE as u128 - 1) as u64;
-        let tcs = thread::current().as_ptr();
-
-        if self.state.load(Acquire) != NOTIFIED {
-            if self.state.compare_exchange(EMPTY, tcs, Relaxed, Acquire).is_ok() {
-                match usercalls::wait(EV_UNPARK, timeout) {
-                    Ok(event) => assert!(event & EV_UNPARK == EV_UNPARK),
-                    Err(e) => {
-                        assert!(matches!(e.kind(), ErrorKind::TimedOut | ErrorKind::WouldBlock))
-                    }
-                }
-
-                // Swap to provide acquire ordering even if the timeout occurred
-                // before the token was set. This situation can result in spurious
-                // wakeups on the next call to `park_timeout`, but it is better to let
-                // those be handled by the user than do some perhaps unnecessary, but
-                // always expensive guarding.
-                self.state.swap(EMPTY, Acquire);
-                return;
-            }
-        }
-
-        // The token was already read with `acquire` ordering, this can be a store.
-        self.state.store(EMPTY, Relaxed);
-    }
-
-    // This implementation doesn't require `Pin`, but other implementations do.
-    pub fn unpark(self: Pin<&Self>) {
-        let state = self.state.swap(NOTIFIED, Release);
-
-        if !matches!(state, EMPTY | NOTIFIED) {
-            // There is a thread waiting, wake it up.
-            let tcs = NonNull::new(state).unwrap();
-            // This will fail if the thread has already terminated or its TCS is destroyed
-            // by the time the signal is sent, but that is fine. If another thread receives
-            // the same TCS, it will receive this notification as a spurious wakeup, but
-            // all users of `wait` should and (internally) do guard against those where
-            // necessary.
-            let _ = usercalls::send(EV_UNPARK, Some(tcs));
-        }
-    }
-}
diff --git a/library/std/src/sys/sgx/thread_parking.rs b/library/std/src/sys/sgx/thread_parking.rs
new file mode 100644
index 00000000000..0006cd4f1be
--- /dev/null
+++ b/library/std/src/sys/sgx/thread_parking.rs
@@ -0,0 +1,23 @@
+use super::abi::usercalls;
+use crate::io::ErrorKind;
+use crate::time::Duration;
+use fortanix_sgx_abi::{EV_UNPARK, WAIT_INDEFINITE};
+
+pub type ThreadId = fortanix_sgx_abi::Tcs;
+
+pub use super::abi::thread::current;
+
+pub fn park(_hint: usize) {
+    usercalls::wait(EV_UNPARK, WAIT_INDEFINITE).unwrap();
+}
+
+pub fn park_timeout(dur: Duration, _hint: usize) {
+    let timeout = u128::min(dur.as_nanos(), WAIT_INDEFINITE as u128 - 1) as u64;
+    if let Err(e) = usercalls::wait(EV_UNPARK, timeout) {
+        assert!(matches!(e.kind(), ErrorKind::TimedOut | ErrorKind::WouldBlock))
+    }
+}
+
+pub fn unpark(tid: ThreadId, _hint: usize) {
+    let _ = usercalls::send(EV_UNPARK, Some(tid));
+}
diff --git a/library/std/src/sys/unix/mod.rs b/library/std/src/sys/unix/mod.rs
index 9d7a06852b7..30a96be1430 100644
--- a/library/std/src/sys/unix/mod.rs
+++ b/library/std/src/sys/unix/mod.rs
@@ -40,7 +40,7 @@ pub mod stdio;
 pub mod thread;
 pub mod thread_local_dtor;
 pub mod thread_local_key;
-pub mod thread_parker;
+pub mod thread_parking;
 pub mod time;
 
 #[cfg(target_os = "espidf")]
diff --git a/library/std/src/sys/unix/thread_parker/netbsd.rs b/library/std/src/sys/unix/thread_parker/netbsd.rs
deleted file mode 100644
index 7657605b52f..00000000000
--- a/library/std/src/sys/unix/thread_parker/netbsd.rs
+++ /dev/null
@@ -1,113 +0,0 @@
-use crate::ffi::{c_int, c_void};
-use crate::pin::Pin;
-use crate::ptr::{null, null_mut};
-use crate::sync::atomic::{
-    AtomicU64,
-    Ordering::{Acquire, Relaxed, Release},
-};
-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;
-}
-
-/// The thread is not parked and the token is not available.
-///
-/// Zero cannot be a valid LWP id, since it is used as empty value for the unpark
-/// argument in _lwp_park.
-const EMPTY: u64 = 0;
-/// The token is available. Do not park anymore.
-const NOTIFIED: u64 = u64::MAX;
-
-pub struct Parker {
-    /// The parker state. Contains either one of the two state values above or the LWP
-    /// id of the parked thread.
-    state: AtomicU64,
-}
-
-impl Parker {
-    pub unsafe fn new(parker: *mut Parker) {
-        parker.write(Parker { state: AtomicU64::new(EMPTY) })
-    }
-
-    // Does not actually need `unsafe` or `Pin`, but the pthread implementation does.
-    pub unsafe fn park(self: Pin<&Self>) {
-        // If the token has already been made available, we can skip
-        // a bit of work, so check for it here.
-        if self.state.load(Acquire) != NOTIFIED {
-            let parked = _lwp_self() as u64;
-            let hint = self.state.as_mut_ptr().cast();
-            if self.state.compare_exchange(EMPTY, parked, Relaxed, Acquire).is_ok() {
-                // Loop to guard against spurious wakeups.
-                loop {
-                    ___lwp_park60(0, 0, null_mut(), 0, hint, null());
-                    if self.state.load(Acquire) == NOTIFIED {
-                        break;
-                    }
-                }
-            }
-        }
-
-        // At this point, the change to NOTIFIED has always been observed with acquire
-        // ordering, so we can just use a relaxed store here (instead of a swap).
-        self.state.store(EMPTY, Relaxed);
-    }
-
-    // Does not actually need `unsafe` or `Pin`, but the pthread implementation does.
-    pub unsafe fn park_timeout(self: Pin<&Self>, dur: Duration) {
-        if self.state.load(Acquire) != NOTIFIED {
-            let parked = _lwp_self() as u64;
-            let hint = self.state.as_mut_ptr().cast();
-            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(),
-            };
-
-            if self.state.compare_exchange(EMPTY, parked, Relaxed, Acquire).is_ok() {
-                // Timeout needs to be mutable since it is modified on NetBSD 9.0 and
-                // above.
-                ___lwp_park60(CLOCK_MONOTONIC, 0, &mut timeout, 0, hint, null());
-                // Use a swap to get acquire ordering even if the token was set after
-                // the timeout occurred.
-                self.state.swap(EMPTY, Acquire);
-                return;
-            }
-        }
-
-        self.state.store(EMPTY, Relaxed);
-    }
-
-    // Does not actually need `Pin`, but the pthread implementation does.
-    pub fn unpark(self: Pin<&Self>) {
-        let state = self.state.swap(NOTIFIED, Release);
-        if !matches!(state, EMPTY | NOTIFIED) {
-            let lwp = state as lwpid_t;
-            let hint = self.state.as_mut_ptr().cast();
-
-            // If the parking thread terminated and did not actually park, this will
-            // probably return an error, which is OK. In the worst case, another
-            // thread has received the same LWP id. It will then receive a spurious
-            // wakeup, but those are allowable per the API contract. The same reasoning
-            // applies if a timeout occurred before this call, but the state was not
-            // yet reset.
-
-            // SAFETY:
-            // The syscall has no invariants to hold. Only unsafe because it is an
-            // extern function.
-            unsafe {
-                _lwp_unpark(lwp, hint);
-            }
-        }
-    }
-}
diff --git a/library/std/src/sys/unix/thread_parker/darwin.rs b/library/std/src/sys/unix/thread_parking/darwin.rs
index 2f5356fe227..b709fada3b4 100644
--- a/library/std/src/sys/unix/thread_parker/darwin.rs
+++ b/library/std/src/sys/unix/thread_parking/darwin.rs
@@ -46,7 +46,7 @@ unsafe impl Sync for Parker {}
 unsafe impl Send for Parker {}
 
 impl Parker {
-    pub unsafe fn new(parker: *mut Parker) {
+    pub unsafe fn new_in_place(parker: *mut Parker) {
         let semaphore = dispatch_semaphore_create(0);
         assert!(
             !semaphore.is_null(),
diff --git a/library/std/src/sys/unix/thread_parker/mod.rs b/library/std/src/sys/unix/thread_parking/mod.rs
index 35f1e68a87e..185333c072f 100644
--- a/library/std/src/sys/unix/thread_parker/mod.rs
+++ b/library/std/src/sys/unix/thread_parking/mod.rs
@@ -24,7 +24,7 @@ cfg_if::cfg_if! {
         pub use darwin::Parker;
     } else if #[cfg(target_os = "netbsd")] {
         mod netbsd;
-        pub use netbsd::Parker;
+        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 00000000000..3be08122138
--- /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_parker/pthread.rs b/library/std/src/sys/unix/thread_parking/pthread.rs
index 510168a010f..082d25e68f5 100644
--- a/library/std/src/sys/unix/thread_parker/pthread.rs
+++ b/library/std/src/sys/unix/thread_parking/pthread.rs
@@ -99,7 +99,7 @@ impl Parker {
     ///
     /// # Safety
     /// The constructed parker must never be moved.
-    pub unsafe fn new(parker: *mut Parker) {
+    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
diff --git a/library/std/src/sys/windows/mod.rs b/library/std/src/sys/windows/mod.rs
index e67411e1686..77359abe429 100644
--- a/library/std/src/sys/windows/mod.rs
+++ b/library/std/src/sys/windows/mod.rs
@@ -33,7 +33,7 @@ pub mod stdio;
 pub mod thread;
 pub mod thread_local_dtor;
 pub mod thread_local_key;
-pub mod thread_parker;
+pub mod thread_parking;
 pub mod time;
 cfg_if::cfg_if! {
     if #[cfg(not(target_vendor = "uwp"))] {
diff --git a/library/std/src/sys/windows/thread_parker.rs b/library/std/src/sys/windows/thread_parking.rs
index 2f7ae863b6a..5d43676adbb 100644
--- a/library/std/src/sys/windows/thread_parker.rs
+++ b/library/std/src/sys/windows/thread_parking.rs
@@ -97,7 +97,7 @@ const NOTIFIED: i8 = 1;
 impl Parker {
     /// Construct the Windows parker. The UNIX parker implementation
     /// requires this to happen in-place.
-    pub unsafe fn new(parker: *mut Parker) {
+    pub unsafe fn new_in_place(parker: *mut Parker) {
         parker.write(Self { state: AtomicI8::new(EMPTY) });
     }