diff options
Diffstat (limited to '')
16 files changed, 336 insertions, 244 deletions
diff --git a/library/std/src/sys_common/remutex.rs b/library/std/src/sync/remutex.rs index 4c054da64..4c054da64 100644 --- a/library/std/src/sys_common/remutex.rs +++ b/library/std/src/sync/remutex.rs diff --git a/library/std/src/sys_common/backtrace.rs b/library/std/src/sys_common/backtrace.rs index 8807077cb..f1d804ef4 100644 --- a/library/std/src/sys_common/backtrace.rs +++ b/library/std/src/sys_common/backtrace.rs @@ -20,7 +20,7 @@ pub fn lock() -> impl Drop { /// Prints the current backtrace. pub fn print(w: &mut dyn Write, format: PrintFmt) -> io::Result<()> { // There are issues currently linking libbacktrace into tests, and in - // general during libstd's own unit tests we're not testing this path. In + // general during std's own unit tests we're not testing this path. In // test mode immediately return here to optimize away any references to the // libbacktrace symbols if cfg!(test) { @@ -111,7 +111,7 @@ unsafe fn _print_fmt(fmt: &mut fmt::Formatter<'_>, print_fmt: PrintFmt) -> fmt:: } /// Fixed frame used to clean the backtrace with `RUST_BACKTRACE=1`. Note that -/// this is only inline(never) when backtraces in libstd are enabled, otherwise +/// this is only inline(never) when backtraces in std are enabled, otherwise /// it's fine to optimize away. #[cfg_attr(feature = "backtrace", inline(never))] pub fn __rust_begin_short_backtrace<F, T>(f: F) -> T @@ -127,7 +127,7 @@ where } /// Fixed frame used to clean the backtrace with `RUST_BACKTRACE=1`. Note that -/// this is only inline(never) when backtraces in libstd are enabled, otherwise +/// this is only inline(never) when backtraces in std are enabled, otherwise /// it's fine to optimize away. #[cfg_attr(feature = "backtrace", inline(never))] pub fn __rust_end_short_backtrace<F, T>(f: F) -> T diff --git a/library/std/src/sys_common/io.rs b/library/std/src/sys_common/io.rs index d1e9fed41..4a42ff3c6 100644 --- a/library/std/src/sys_common/io.rs +++ b/library/std/src/sys_common/io.rs @@ -39,9 +39,10 @@ pub mod test { } } + #[track_caller] // for `test_rng` pub fn tmpdir() -> TempDir { let p = env::temp_dir(); - let mut r = rand::thread_rng(); + let mut r = crate::test_helpers::test_rng(); let ret = p.join(&format!("rust-{}", r.next_u32())); fs::create_dir(&ret).unwrap(); TempDir(ret) diff --git a/library/std/src/sys_common/mod.rs b/library/std/src/sys_common/mod.rs index b1987aa0f..6b24b0e9a 100644 --- a/library/std/src/sys_common/mod.rs +++ b/library/std/src/sys_common/mod.rs @@ -27,11 +27,10 @@ pub mod lazy_box; pub mod memchr; pub mod once; pub mod process; -pub mod remutex; pub mod thread; pub mod thread_info; pub mod thread_local_dtor; -pub mod thread_parker; +pub mod thread_parking; pub mod wstr; pub mod wtf8; diff --git a/library/std/src/sys_common/once/mod.rs b/library/std/src/sys_common/once/mod.rs index 8742e68cc..359697d83 100644 --- a/library/std/src/sys_common/once/mod.rs +++ b/library/std/src/sys_common/once/mod.rs @@ -6,22 +6,6 @@ // As a result, we end up implementing it ourselves in the standard library. // This also gives us the opportunity to optimize the implementation a bit which // should help the fast path on call sites. -// -// So to recap, the guarantees of a Once are that it will call the -// initialization closure at most once, and it will never return until the one -// that's running has finished running. This means that we need some form of -// blocking here while the custom callback is running at the very least. -// Additionally, we add on the restriction of **poisoning**. Whenever an -// initialization closure panics, the Once enters a "poisoned" state which means -// that all future calls will immediately panic as well. -// -// So to implement this, one might first reach for a `Mutex`, but those cannot -// be put into a `static`. It also gets a lot harder with poisoning to figure -// out when the mutex needs to be deallocated because it's not after the closure -// finishes, but after the first successful closure finishes. -// -// All in all, this is instead implemented with atomics and lock-free -// operations! Whee! cfg_if::cfg_if! { if #[cfg(any( @@ -36,8 +20,15 @@ cfg_if::cfg_if! { ))] { mod futex; pub use futex::{Once, OnceState}; + } else if #[cfg(any( + windows, + target_family = "unix", + all(target_vendor = "fortanix", target_env = "sgx"), + target_os = "solid_asp3", + ))] { + mod queue; + pub use queue::{Once, OnceState}; } else { - mod generic; - pub use generic::{Once, OnceState}; + pub use crate::sys::once::{Once, OnceState}; } } diff --git a/library/std/src/sys_common/once/generic.rs b/library/std/src/sys_common/once/queue.rs index d953a6745..d953a6745 100644 --- a/library/std/src/sys_common/once/generic.rs +++ b/library/std/src/sys_common/once/queue.rs diff --git a/library/std/src/sys_common/process.rs b/library/std/src/sys_common/process.rs index 9f978789a..18883048d 100644 --- a/library/std/src/sys_common/process.rs +++ b/library/std/src/sys_common/process.rs @@ -4,10 +4,13 @@ use crate::collections::BTreeMap; use crate::env; use crate::ffi::{OsStr, OsString}; -use crate::sys::process::EnvKey; +use crate::fmt; +use crate::io; +use crate::sys::pipe::read2; +use crate::sys::process::{EnvKey, ExitStatus, Process, StdioPipes}; // Stores a set of changes to an environment -#[derive(Clone, Debug)] +#[derive(Clone)] pub struct CommandEnv { clear: bool, saw_path: bool, @@ -20,6 +23,14 @@ impl Default for CommandEnv { } } +impl fmt::Debug for CommandEnv { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + let mut debug_command_env = f.debug_struct("CommandEnv"); + debug_command_env.field("clear", &self.clear).field("vars", &self.vars); + debug_command_env.finish() + } +} + impl CommandEnv { // Capture the current environment with these changes applied pub fn capture(&self) -> BTreeMap<EnvKey, OsString> { @@ -117,3 +128,30 @@ impl<'a> ExactSizeIterator for CommandEnvs<'a> { self.iter.is_empty() } } + +pub fn wait_with_output( + mut process: Process, + mut pipes: StdioPipes, +) -> io::Result<(ExitStatus, Vec<u8>, Vec<u8>)> { + drop(pipes.stdin.take()); + + let (mut stdout, mut stderr) = (Vec::new(), Vec::new()); + match (pipes.stdout.take(), pipes.stderr.take()) { + (None, None) => {} + (Some(out), None) => { + let res = out.read_to_end(&mut stdout); + res.unwrap(); + } + (None, Some(err)) => { + let res = err.read_to_end(&mut stderr); + res.unwrap(); + } + (Some(out), Some(err)) => { + let res = read2(out, &mut stdout, err, &mut stderr); + res.unwrap(); + } + } + + let status = process.wait()?; + Ok((status, stdout, stderr)) +} diff --git a/library/std/src/sys_common/remutex/tests.rs b/library/std/src/sys_common/remutex/tests.rs deleted file mode 100644 index 8e97ce11c..000000000 --- a/library/std/src/sys_common/remutex/tests.rs +++ /dev/null @@ -1,60 +0,0 @@ -use crate::cell::RefCell; -use crate::sync::Arc; -use crate::sys_common::remutex::{ReentrantMutex, ReentrantMutexGuard}; -use crate::thread; - -#[test] -fn smoke() { - let m = ReentrantMutex::new(()); - { - let a = m.lock(); - { - let b = m.lock(); - { - let c = m.lock(); - assert_eq!(*c, ()); - } - assert_eq!(*b, ()); - } - assert_eq!(*a, ()); - } -} - -#[test] -fn is_mutex() { - let m = Arc::new(ReentrantMutex::new(RefCell::new(0))); - let m2 = m.clone(); - let lock = m.lock(); - let child = thread::spawn(move || { - let lock = m2.lock(); - assert_eq!(*lock.borrow(), 4950); - }); - for i in 0..100 { - let lock = m.lock(); - *lock.borrow_mut() += i; - } - drop(lock); - child.join().unwrap(); -} - -#[test] -fn trylock_works() { - let m = Arc::new(ReentrantMutex::new(())); - let m2 = m.clone(); - let _lock = m.try_lock(); - let _lock2 = m.try_lock(); - thread::spawn(move || { - let lock = m2.try_lock(); - assert!(lock.is_none()); - }) - .join() - .unwrap(); - let _lock3 = m.try_lock(); -} - -pub struct Answer<'a>(pub ReentrantMutexGuard<'a, RefCell<u32>>); -impl Drop for Answer<'_> { - fn drop(&mut self) { - *self.0.borrow_mut() = 42; - } -} diff --git a/library/std/src/sys_common/thread_local_key.rs b/library/std/src/sys_common/thread_local_key.rs index 747579f17..2672a2a75 100644 --- a/library/std/src/sys_common/thread_local_key.rs +++ b/library/std/src/sys_common/thread_local_key.rs @@ -117,10 +117,14 @@ pub struct Key { /// This value specifies no destructor by default. pub const INIT: StaticKey = StaticKey::new(None); +// Define a sentinel value that is unlikely to be returned +// as a TLS key (but it may be returned). +const KEY_SENTVAL: usize = 0; + impl StaticKey { #[rustc_const_unstable(feature = "thread_local_internals", issue = "none")] pub const fn new(dtor: Option<unsafe extern "C" fn(*mut u8)>) -> StaticKey { - StaticKey { key: atomic::AtomicUsize::new(0), dtor } + StaticKey { key: atomic::AtomicUsize::new(KEY_SENTVAL), dtor } } /// Gets the value associated with this TLS key @@ -144,31 +148,36 @@ impl StaticKey { #[inline] unsafe fn key(&self) -> imp::Key { match self.key.load(Ordering::Relaxed) { - 0 => self.lazy_init() as imp::Key, + KEY_SENTVAL => self.lazy_init() as imp::Key, n => n as imp::Key, } } unsafe fn lazy_init(&self) -> usize { - // POSIX allows the key created here to be 0, but the compare_exchange - // below relies on using 0 as a sentinel value to check who won the + // POSIX allows the key created here to be KEY_SENTVAL, but the compare_exchange + // below relies on using KEY_SENTVAL as a sentinel value to check who won the // race to set the shared TLS key. As far as I know, there is no // guaranteed value that cannot be returned as a posix_key_create key, // so there is no value we can initialize the inner key with to // prove that it has not yet been set. As such, we'll continue using a - // value of 0, but with some gyrations to make sure we have a non-0 + // value of KEY_SENTVAL, but with some gyrations to make sure we have a non-KEY_SENTVAL // value returned from the creation routine. // FIXME: this is clearly a hack, and should be cleaned up. let key1 = imp::create(self.dtor); - let key = if key1 != 0 { + let key = if key1 as usize != KEY_SENTVAL { key1 } else { let key2 = imp::create(self.dtor); imp::destroy(key1); key2 }; - rtassert!(key != 0); - match self.key.compare_exchange(0, key as usize, Ordering::SeqCst, Ordering::SeqCst) { + rtassert!(key as usize != KEY_SENTVAL); + match self.key.compare_exchange( + KEY_SENTVAL, + key as usize, + Ordering::SeqCst, + Ordering::SeqCst, + ) { // The CAS succeeded, so we've created the actual key Ok(_) => key as usize, // If someone beat us to the punch, use their key instead diff --git a/library/std/src/sys_common/thread_parker/generic.rs b/library/std/src/sys_common/thread_parker/generic.rs deleted file mode 100644 index f3d8b34d3..000000000 --- a/library/std/src/sys_common/thread_parker/generic.rs +++ /dev/null @@ -1,125 +0,0 @@ -//! Parker implementation based on a Mutex and Condvar. - -use crate::pin::Pin; -use crate::sync::atomic::AtomicUsize; -use crate::sync::atomic::Ordering::SeqCst; -use crate::sync::{Condvar, Mutex}; -use crate::time::Duration; - -const EMPTY: usize = 0; -const PARKED: usize = 1; -const NOTIFIED: usize = 2; - -pub struct Parker { - state: AtomicUsize, - lock: Mutex<()>, - cvar: Condvar, -} - -impl Parker { - /// Construct the generic parker. The UNIX parker implementation - /// requires this to happen in-place. - pub unsafe fn new(parker: *mut Parker) { - parker.write(Parker { - state: AtomicUsize::new(EMPTY), - lock: Mutex::new(()), - cvar: Condvar::new(), - }); - } - - // This implementation doesn't require `unsafe` and `Pin`, but other implementations do. - 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 - let mut m = self.lock.lock().unwrap(); - 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); - assert_eq!(old, NOTIFIED, "park state changed unexpectedly"); - return; - } // should consume this notification, so prohibit spurious wakeups in next park. - Err(_) => panic!("inconsistent park state"), - } - loop { - m = self.cvar.wait(m).unwrap(); - match self.state.compare_exchange(NOTIFIED, EMPTY, SeqCst, SeqCst) { - Ok(_) => return, // got a notification - Err(_) => {} // spurious wakeup, go back to sleep - } - } - } - - // This implementation doesn't require `unsafe` and `Pin`, but other implementations do. - 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; - } - let m = self.lock.lock().unwrap(); - 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); - assert_eq!(old, NOTIFIED, "park state changed unexpectedly"); - return; - } // should consume this notification, so prohibit spurious wakeups in next park. - Err(_) => 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. - let (_m, _result) = self.cvar.wait_timeout(m, dur).unwrap(); - match self.state.swap(EMPTY, SeqCst) { - NOTIFIED => {} // got a notification, hurray! - PARKED => {} // no notification, alas - n => panic!("inconsistent park_timeout state: {n}"), - } - } - - // This implementation doesn't require `Pin`, but other implementations do. - 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`. - drop(self.lock.lock().unwrap()); - self.cvar.notify_one() - } -} diff --git a/library/std/src/sys_common/thread_parker/mod.rs b/library/std/src/sys_common/thread_parker/mod.rs deleted file mode 100644 index f86a9a555..000000000 --- a/library/std/src/sys_common/thread_parker/mod.rs +++ /dev/null @@ -1,23 +0,0 @@ -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(target_os = "solid_asp3")] { - mod wait_flag; - pub use wait_flag::Parker; - } else if #[cfg(any(windows, target_family = "unix"))] { - pub use crate::sys::thread_parker::Parker; - } else { - mod generic; - pub use generic::Parker; - } -} diff --git a/library/std/src/sys_common/thread_parker/futex.rs b/library/std/src/sys_common/thread_parking/futex.rs index d9e2f39e3..588e7b278 100644 --- a/library/std/src/sys_common/thread_parker/futex.rs +++ b/library/std/src/sys_common/thread_parking/futex.rs @@ -35,7 +35,7 @@ pub struct Parker { impl Parker { /// Construct the futex 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: AtomicU32::new(EMPTY) }); } diff --git a/library/std/src/sys_common/thread_parking/generic.rs b/library/std/src/sys_common/thread_parking/generic.rs new file mode 100644 index 000000000..3209bffe3 --- /dev/null +++ b/library/std/src/sys_common/thread_parking/generic.rs @@ -0,0 +1,125 @@ +//! Parker implementation based on a Mutex and Condvar. + +use crate::pin::Pin; +use crate::sync::atomic::AtomicUsize; +use crate::sync::atomic::Ordering::SeqCst; +use crate::sync::{Condvar, Mutex}; +use crate::time::Duration; + +const EMPTY: usize = 0; +const PARKED: usize = 1; +const NOTIFIED: usize = 2; + +pub struct Parker { + state: AtomicUsize, + lock: Mutex<()>, + cvar: Condvar, +} + +impl Parker { + /// Construct the generic parker. The UNIX parker implementation + /// requires this to happen in-place. + pub unsafe fn new_in_place(parker: *mut Parker) { + parker.write(Parker { + state: AtomicUsize::new(EMPTY), + lock: Mutex::new(()), + cvar: Condvar::new(), + }); + } + + // This implementation doesn't require `unsafe` and `Pin`, but other implementations do. + 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 + let mut m = self.lock.lock().unwrap(); + 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); + assert_eq!(old, NOTIFIED, "park state changed unexpectedly"); + return; + } // should consume this notification, so prohibit spurious wakeups in next park. + Err(_) => panic!("inconsistent park state"), + } + loop { + m = self.cvar.wait(m).unwrap(); + match self.state.compare_exchange(NOTIFIED, EMPTY, SeqCst, SeqCst) { + Ok(_) => return, // got a notification + Err(_) => {} // spurious wakeup, go back to sleep + } + } + } + + // This implementation doesn't require `unsafe` and `Pin`, but other implementations do. + 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; + } + let m = self.lock.lock().unwrap(); + 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); + assert_eq!(old, NOTIFIED, "park state changed unexpectedly"); + return; + } // should consume this notification, so prohibit spurious wakeups in next park. + Err(_) => 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. + let (_m, _result) = self.cvar.wait_timeout(m, dur).unwrap(); + match self.state.swap(EMPTY, SeqCst) { + NOTIFIED => {} // got a notification, hurray! + PARKED => {} // no notification, alas + n => panic!("inconsistent park_timeout state: {n}"), + } + } + + // This implementation doesn't require `Pin`, but other implementations do. + 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`. + drop(self.lock.lock().unwrap()); + self.cvar.notify_one() + } +} diff --git a/library/std/src/sys_common/thread_parking/id.rs b/library/std/src/sys_common/thread_parking/id.rs new file mode 100644 index 000000000..e98169597 --- /dev/null +++ b/library/std/src/sys_common/thread_parking/id.rs @@ -0,0 +1,108 @@ +//! 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 mut state = self.state.fetch_sub(1, Acquire).wrapping_sub(1); + if state == PARKED { + // Loop to guard against spurious wakeups. + while state == PARKED { + park(self.state.as_mut_ptr().addr()); + state = self.state.load(Acquire); + } + + // Since the state change has already been observed with acquire + // ordering, the state can be reset with a relaxed store instead + // of a swap. + self.state.store(EMPTY, Relaxed); + } + } + + 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_mut_ptr().addr()); + // Swap to ensure that we observe all state changes with acquire + // ordering, even if the state has been changed after the timeout + // occured. + 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_mut_ptr().addr()); + } + } +} + +unsafe impl Send for Parker {} +unsafe impl Sync for Parker {} diff --git a/library/std/src/sys_common/thread_parking/mod.rs b/library/std/src/sys_common/thread_parking/mod.rs new file mode 100644 index 000000000..0ead6633c --- /dev/null +++ b/library/std/src/sys_common/thread_parking/mod.rs @@ -0,0 +1,29 @@ +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"), + ))] { + mod id; + pub use id::Parker; + } else if #[cfg(target_os = "solid_asp3")] { + mod wait_flag; + pub use wait_flag::Parker; + } else if #[cfg(any(windows, target_family = "unix"))] { + pub use crate::sys::thread_parking::Parker; + } else { + mod generic; + pub use generic::Parker; + } +} diff --git a/library/std/src/sys_common/thread_parker/wait_flag.rs b/library/std/src/sys_common/thread_parking/wait_flag.rs index 6561c1866..d0f8899a9 100644 --- a/library/std/src/sys_common/thread_parker/wait_flag.rs +++ b/library/std/src/sys_common/thread_parking/wait_flag.rs @@ -41,7 +41,7 @@ pub struct Parker { impl Parker { /// Construct a parker for the current thread. 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(Parker { state: AtomicI8::new(EMPTY), wait_flag: WaitFlag::new() }) } |