From 218caa410aa38c29984be31a5229b9fa717560ee Mon Sep 17 00:00:00 2001 From: Daniel Baumann Date: Wed, 17 Apr 2024 14:19:13 +0200 Subject: Merging upstream version 1.68.2+dfsg1. Signed-off-by: Daniel Baumann --- library/std/src/sys_common/backtrace.rs | 6 +- library/std/src/sys_common/io.rs | 3 +- library/std/src/sys_common/mod.rs | 3 +- library/std/src/sys_common/once/generic.rs | 283 --------------------- library/std/src/sys_common/once/mod.rs | 27 +- library/std/src/sys_common/once/queue.rs | 283 +++++++++++++++++++++ library/std/src/sys_common/process.rs | 42 ++- library/std/src/sys_common/remutex.rs | 178 ------------- library/std/src/sys_common/remutex/tests.rs | 60 ----- library/std/src/sys_common/thread_local_key.rs | 25 +- library/std/src/sys_common/thread_parker/futex.rs | 97 ------- .../std/src/sys_common/thread_parker/generic.rs | 125 --------- library/std/src/sys_common/thread_parker/mod.rs | 23 -- .../std/src/sys_common/thread_parker/wait_flag.rs | 102 -------- library/std/src/sys_common/thread_parking/futex.rs | 97 +++++++ .../std/src/sys_common/thread_parking/generic.rs | 125 +++++++++ library/std/src/sys_common/thread_parking/id.rs | 108 ++++++++ library/std/src/sys_common/thread_parking/mod.rs | 29 +++ .../std/src/sys_common/thread_parking/wait_flag.rs | 102 ++++++++ 19 files changed, 816 insertions(+), 902 deletions(-) delete mode 100644 library/std/src/sys_common/once/generic.rs create mode 100644 library/std/src/sys_common/once/queue.rs delete mode 100644 library/std/src/sys_common/remutex.rs delete mode 100644 library/std/src/sys_common/remutex/tests.rs delete mode 100644 library/std/src/sys_common/thread_parker/futex.rs delete mode 100644 library/std/src/sys_common/thread_parker/generic.rs delete mode 100644 library/std/src/sys_common/thread_parker/mod.rs delete mode 100644 library/std/src/sys_common/thread_parker/wait_flag.rs create mode 100644 library/std/src/sys_common/thread_parking/futex.rs create mode 100644 library/std/src/sys_common/thread_parking/generic.rs create mode 100644 library/std/src/sys_common/thread_parking/id.rs create mode 100644 library/std/src/sys_common/thread_parking/mod.rs create mode 100644 library/std/src/sys_common/thread_parking/wait_flag.rs (limited to 'library/std/src/sys_common') 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: 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: 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/generic.rs b/library/std/src/sys_common/once/generic.rs deleted file mode 100644 index d953a6745..000000000 --- a/library/std/src/sys_common/once/generic.rs +++ /dev/null @@ -1,283 +0,0 @@ -// Each `Once` has one word of atomic state, and this state is CAS'd on to -// determine what to do. There are four possible state of a `Once`: -// -// * Incomplete - no initialization has run yet, and no thread is currently -// using the Once. -// * Poisoned - some thread has previously attempted to initialize the Once, but -// it panicked, so the Once is now poisoned. There are no other -// threads currently accessing this Once. -// * Running - some thread is currently attempting to run initialization. It may -// succeed, so all future threads need to wait for it to finish. -// Note that this state is accompanied with a payload, described -// below. -// * Complete - initialization has completed and all future calls should finish -// immediately. -// -// With 4 states we need 2 bits to encode this, and we use the remaining bits -// in the word we have allocated as a queue of threads waiting for the thread -// responsible for entering the RUNNING state. This queue is just a linked list -// of Waiter nodes which is monotonically increasing in size. Each node is -// allocated on the stack, and whenever the running closure finishes it will -// consume the entire queue and notify all waiters they should try again. -// -// You'll find a few more details in the implementation, but that's the gist of -// it! -// -// Atomic orderings: -// When running `Once` we deal with multiple atomics: -// `Once.state_and_queue` and an unknown number of `Waiter.signaled`. -// * `state_and_queue` is used (1) as a state flag, (2) for synchronizing the -// result of the `Once`, and (3) for synchronizing `Waiter` nodes. -// - At the end of the `call` function we have to make sure the result -// of the `Once` is acquired. So every load which can be the only one to -// load COMPLETED must have at least acquire ordering, which means all -// three of them. -// - `WaiterQueue::drop` is the only place that may store COMPLETED, and -// must do so with release ordering to make the result available. -// - `wait` inserts `Waiter` nodes as a pointer in `state_and_queue`, and -// needs to make the nodes available with release ordering. The load in -// its `compare_exchange` can be relaxed because it only has to compare -// the atomic, not to read other data. -// - `WaiterQueue::drop` must see the `Waiter` nodes, so it must load -// `state_and_queue` with acquire ordering. -// - There is just one store where `state_and_queue` is used only as a -// state flag, without having to synchronize data: switching the state -// from INCOMPLETE to RUNNING in `call`. This store can be Relaxed, -// but the read has to be Acquire because of the requirements mentioned -// above. -// * `Waiter.signaled` is both used as a flag, and to protect a field with -// interior mutability in `Waiter`. `Waiter.thread` is changed in -// `WaiterQueue::drop` which then sets `signaled` with release ordering. -// After `wait` loads `signaled` with acquire ordering and sees it is true, -// it needs to see the changes to drop the `Waiter` struct correctly. -// * There is one place where the two atomics `Once.state_and_queue` and -// `Waiter.signaled` come together, and might be reordered by the compiler or -// processor. Because both use acquire ordering such a reordering is not -// allowed, so no need for `SeqCst`. - -use crate::cell::Cell; -use crate::fmt; -use crate::ptr; -use crate::sync as public; -use crate::sync::atomic::{AtomicBool, AtomicPtr, Ordering}; -use crate::thread::{self, Thread}; - -type Masked = (); - -pub struct Once { - state_and_queue: AtomicPtr, -} - -pub struct OnceState { - poisoned: bool, - set_state_on_drop_to: Cell<*mut Masked>, -} - -// Four states that a Once can be in, encoded into the lower bits of -// `state_and_queue` in the Once structure. -const INCOMPLETE: usize = 0x0; -const POISONED: usize = 0x1; -const RUNNING: usize = 0x2; -const COMPLETE: usize = 0x3; - -// Mask to learn about the state. All other bits are the queue of waiters if -// this is in the RUNNING state. -const STATE_MASK: usize = 0x3; - -// Representation of a node in the linked list of waiters, used while in the -// RUNNING state. -// Note: `Waiter` can't hold a mutable pointer to the next thread, because then -// `wait` would both hand out a mutable reference to its `Waiter` node, and keep -// a shared reference to check `signaled`. Instead we hold shared references and -// use interior mutability. -#[repr(align(4))] // Ensure the two lower bits are free to use as state bits. -struct Waiter { - thread: Cell>, - signaled: AtomicBool, - next: *const Waiter, -} - -// Head of a linked list of waiters. -// Every node is a struct on the stack of a waiting thread. -// Will wake up the waiters when it gets dropped, i.e. also on panic. -struct WaiterQueue<'a> { - state_and_queue: &'a AtomicPtr, - set_state_on_drop_to: *mut Masked, -} - -impl Once { - #[inline] - #[rustc_const_stable(feature = "const_once_new", since = "1.32.0")] - pub const fn new() -> Once { - Once { state_and_queue: AtomicPtr::new(ptr::invalid_mut(INCOMPLETE)) } - } - - #[inline] - pub fn is_completed(&self) -> bool { - // An `Acquire` load is enough because that makes all the initialization - // operations visible to us, and, this being a fast path, weaker - // ordering helps with performance. This `Acquire` synchronizes with - // `Release` operations on the slow path. - self.state_and_queue.load(Ordering::Acquire).addr() == COMPLETE - } - - // This is a non-generic function to reduce the monomorphization cost of - // using `call_once` (this isn't exactly a trivial or small implementation). - // - // Additionally, this is tagged with `#[cold]` as it should indeed be cold - // and it helps let LLVM know that calls to this function should be off the - // fast path. Essentially, this should help generate more straight line code - // in LLVM. - // - // Finally, this takes an `FnMut` instead of a `FnOnce` because there's - // currently no way to take an `FnOnce` and call it via virtual dispatch - // without some allocation overhead. - #[cold] - #[track_caller] - pub fn call(&self, ignore_poisoning: bool, init: &mut dyn FnMut(&public::OnceState)) { - let mut state_and_queue = self.state_and_queue.load(Ordering::Acquire); - loop { - match state_and_queue.addr() { - COMPLETE => break, - POISONED if !ignore_poisoning => { - // Panic to propagate the poison. - panic!("Once instance has previously been poisoned"); - } - POISONED | INCOMPLETE => { - // Try to register this thread as the one RUNNING. - let exchange_result = self.state_and_queue.compare_exchange( - state_and_queue, - ptr::invalid_mut(RUNNING), - Ordering::Acquire, - Ordering::Acquire, - ); - if let Err(old) = exchange_result { - state_and_queue = old; - continue; - } - // `waiter_queue` will manage other waiting threads, and - // wake them up on drop. - let mut waiter_queue = WaiterQueue { - state_and_queue: &self.state_and_queue, - set_state_on_drop_to: ptr::invalid_mut(POISONED), - }; - // Run the initialization function, letting it know if we're - // poisoned or not. - let init_state = public::OnceState { - inner: OnceState { - poisoned: state_and_queue.addr() == POISONED, - set_state_on_drop_to: Cell::new(ptr::invalid_mut(COMPLETE)), - }, - }; - init(&init_state); - waiter_queue.set_state_on_drop_to = init_state.inner.set_state_on_drop_to.get(); - break; - } - _ => { - // All other values must be RUNNING with possibly a - // pointer to the waiter queue in the more significant bits. - assert!(state_and_queue.addr() & STATE_MASK == RUNNING); - wait(&self.state_and_queue, state_and_queue); - state_and_queue = self.state_and_queue.load(Ordering::Acquire); - } - } - } - } -} - -fn wait(state_and_queue: &AtomicPtr, mut current_state: *mut Masked) { - // Note: the following code was carefully written to avoid creating a - // mutable reference to `node` that gets aliased. - loop { - // Don't queue this thread if the status is no longer running, - // otherwise we will not be woken up. - if current_state.addr() & STATE_MASK != RUNNING { - return; - } - - // Create the node for our current thread. - let node = Waiter { - thread: Cell::new(Some(thread::current())), - signaled: AtomicBool::new(false), - next: current_state.with_addr(current_state.addr() & !STATE_MASK) as *const Waiter, - }; - let me = &node as *const Waiter as *const Masked as *mut Masked; - - // Try to slide in the node at the head of the linked list, making sure - // that another thread didn't just replace the head of the linked list. - let exchange_result = state_and_queue.compare_exchange( - current_state, - me.with_addr(me.addr() | RUNNING), - Ordering::Release, - Ordering::Relaxed, - ); - if let Err(old) = exchange_result { - current_state = old; - continue; - } - - // We have enqueued ourselves, now lets wait. - // It is important not to return before being signaled, otherwise we - // would drop our `Waiter` node and leave a hole in the linked list - // (and a dangling reference). Guard against spurious wakeups by - // reparking ourselves until we are signaled. - while !node.signaled.load(Ordering::Acquire) { - // If the managing thread happens to signal and unpark us before we - // can park ourselves, the result could be this thread never gets - // unparked. Luckily `park` comes with the guarantee that if it got - // an `unpark` just before on an unparked thread it does not park. - thread::park(); - } - break; - } -} - -#[stable(feature = "std_debug", since = "1.16.0")] -impl fmt::Debug for Once { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - f.debug_struct("Once").finish_non_exhaustive() - } -} - -impl Drop for WaiterQueue<'_> { - fn drop(&mut self) { - // Swap out our state with however we finished. - let state_and_queue = - self.state_and_queue.swap(self.set_state_on_drop_to, Ordering::AcqRel); - - // We should only ever see an old state which was RUNNING. - assert_eq!(state_and_queue.addr() & STATE_MASK, RUNNING); - - // Walk the entire linked list of waiters and wake them up (in lifo - // order, last to register is first to wake up). - unsafe { - // Right after setting `node.signaled = true` the other thread may - // free `node` if there happens to be has a spurious wakeup. - // So we have to take out the `thread` field and copy the pointer to - // `next` first. - let mut queue = - state_and_queue.with_addr(state_and_queue.addr() & !STATE_MASK) as *const Waiter; - while !queue.is_null() { - let next = (*queue).next; - let thread = (*queue).thread.take().unwrap(); - (*queue).signaled.store(true, Ordering::Release); - // ^- FIXME (maybe): This is another case of issue #55005 - // `store()` has a potentially dangling ref to `signaled`. - queue = next; - thread.unpark(); - } - } - } -} - -impl OnceState { - #[inline] - pub fn is_poisoned(&self) -> bool { - self.poisoned - } - - #[inline] - pub fn poison(&self) { - self.set_state_on_drop_to.set(ptr::invalid_mut(POISONED)); - } -} 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/queue.rs b/library/std/src/sys_common/once/queue.rs new file mode 100644 index 000000000..d953a6745 --- /dev/null +++ b/library/std/src/sys_common/once/queue.rs @@ -0,0 +1,283 @@ +// Each `Once` has one word of atomic state, and this state is CAS'd on to +// determine what to do. There are four possible state of a `Once`: +// +// * Incomplete - no initialization has run yet, and no thread is currently +// using the Once. +// * Poisoned - some thread has previously attempted to initialize the Once, but +// it panicked, so the Once is now poisoned. There are no other +// threads currently accessing this Once. +// * Running - some thread is currently attempting to run initialization. It may +// succeed, so all future threads need to wait for it to finish. +// Note that this state is accompanied with a payload, described +// below. +// * Complete - initialization has completed and all future calls should finish +// immediately. +// +// With 4 states we need 2 bits to encode this, and we use the remaining bits +// in the word we have allocated as a queue of threads waiting for the thread +// responsible for entering the RUNNING state. This queue is just a linked list +// of Waiter nodes which is monotonically increasing in size. Each node is +// allocated on the stack, and whenever the running closure finishes it will +// consume the entire queue and notify all waiters they should try again. +// +// You'll find a few more details in the implementation, but that's the gist of +// it! +// +// Atomic orderings: +// When running `Once` we deal with multiple atomics: +// `Once.state_and_queue` and an unknown number of `Waiter.signaled`. +// * `state_and_queue` is used (1) as a state flag, (2) for synchronizing the +// result of the `Once`, and (3) for synchronizing `Waiter` nodes. +// - At the end of the `call` function we have to make sure the result +// of the `Once` is acquired. So every load which can be the only one to +// load COMPLETED must have at least acquire ordering, which means all +// three of them. +// - `WaiterQueue::drop` is the only place that may store COMPLETED, and +// must do so with release ordering to make the result available. +// - `wait` inserts `Waiter` nodes as a pointer in `state_and_queue`, and +// needs to make the nodes available with release ordering. The load in +// its `compare_exchange` can be relaxed because it only has to compare +// the atomic, not to read other data. +// - `WaiterQueue::drop` must see the `Waiter` nodes, so it must load +// `state_and_queue` with acquire ordering. +// - There is just one store where `state_and_queue` is used only as a +// state flag, without having to synchronize data: switching the state +// from INCOMPLETE to RUNNING in `call`. This store can be Relaxed, +// but the read has to be Acquire because of the requirements mentioned +// above. +// * `Waiter.signaled` is both used as a flag, and to protect a field with +// interior mutability in `Waiter`. `Waiter.thread` is changed in +// `WaiterQueue::drop` which then sets `signaled` with release ordering. +// After `wait` loads `signaled` with acquire ordering and sees it is true, +// it needs to see the changes to drop the `Waiter` struct correctly. +// * There is one place where the two atomics `Once.state_and_queue` and +// `Waiter.signaled` come together, and might be reordered by the compiler or +// processor. Because both use acquire ordering such a reordering is not +// allowed, so no need for `SeqCst`. + +use crate::cell::Cell; +use crate::fmt; +use crate::ptr; +use crate::sync as public; +use crate::sync::atomic::{AtomicBool, AtomicPtr, Ordering}; +use crate::thread::{self, Thread}; + +type Masked = (); + +pub struct Once { + state_and_queue: AtomicPtr, +} + +pub struct OnceState { + poisoned: bool, + set_state_on_drop_to: Cell<*mut Masked>, +} + +// Four states that a Once can be in, encoded into the lower bits of +// `state_and_queue` in the Once structure. +const INCOMPLETE: usize = 0x0; +const POISONED: usize = 0x1; +const RUNNING: usize = 0x2; +const COMPLETE: usize = 0x3; + +// Mask to learn about the state. All other bits are the queue of waiters if +// this is in the RUNNING state. +const STATE_MASK: usize = 0x3; + +// Representation of a node in the linked list of waiters, used while in the +// RUNNING state. +// Note: `Waiter` can't hold a mutable pointer to the next thread, because then +// `wait` would both hand out a mutable reference to its `Waiter` node, and keep +// a shared reference to check `signaled`. Instead we hold shared references and +// use interior mutability. +#[repr(align(4))] // Ensure the two lower bits are free to use as state bits. +struct Waiter { + thread: Cell>, + signaled: AtomicBool, + next: *const Waiter, +} + +// Head of a linked list of waiters. +// Every node is a struct on the stack of a waiting thread. +// Will wake up the waiters when it gets dropped, i.e. also on panic. +struct WaiterQueue<'a> { + state_and_queue: &'a AtomicPtr, + set_state_on_drop_to: *mut Masked, +} + +impl Once { + #[inline] + #[rustc_const_stable(feature = "const_once_new", since = "1.32.0")] + pub const fn new() -> Once { + Once { state_and_queue: AtomicPtr::new(ptr::invalid_mut(INCOMPLETE)) } + } + + #[inline] + pub fn is_completed(&self) -> bool { + // An `Acquire` load is enough because that makes all the initialization + // operations visible to us, and, this being a fast path, weaker + // ordering helps with performance. This `Acquire` synchronizes with + // `Release` operations on the slow path. + self.state_and_queue.load(Ordering::Acquire).addr() == COMPLETE + } + + // This is a non-generic function to reduce the monomorphization cost of + // using `call_once` (this isn't exactly a trivial or small implementation). + // + // Additionally, this is tagged with `#[cold]` as it should indeed be cold + // and it helps let LLVM know that calls to this function should be off the + // fast path. Essentially, this should help generate more straight line code + // in LLVM. + // + // Finally, this takes an `FnMut` instead of a `FnOnce` because there's + // currently no way to take an `FnOnce` and call it via virtual dispatch + // without some allocation overhead. + #[cold] + #[track_caller] + pub fn call(&self, ignore_poisoning: bool, init: &mut dyn FnMut(&public::OnceState)) { + let mut state_and_queue = self.state_and_queue.load(Ordering::Acquire); + loop { + match state_and_queue.addr() { + COMPLETE => break, + POISONED if !ignore_poisoning => { + // Panic to propagate the poison. + panic!("Once instance has previously been poisoned"); + } + POISONED | INCOMPLETE => { + // Try to register this thread as the one RUNNING. + let exchange_result = self.state_and_queue.compare_exchange( + state_and_queue, + ptr::invalid_mut(RUNNING), + Ordering::Acquire, + Ordering::Acquire, + ); + if let Err(old) = exchange_result { + state_and_queue = old; + continue; + } + // `waiter_queue` will manage other waiting threads, and + // wake them up on drop. + let mut waiter_queue = WaiterQueue { + state_and_queue: &self.state_and_queue, + set_state_on_drop_to: ptr::invalid_mut(POISONED), + }; + // Run the initialization function, letting it know if we're + // poisoned or not. + let init_state = public::OnceState { + inner: OnceState { + poisoned: state_and_queue.addr() == POISONED, + set_state_on_drop_to: Cell::new(ptr::invalid_mut(COMPLETE)), + }, + }; + init(&init_state); + waiter_queue.set_state_on_drop_to = init_state.inner.set_state_on_drop_to.get(); + break; + } + _ => { + // All other values must be RUNNING with possibly a + // pointer to the waiter queue in the more significant bits. + assert!(state_and_queue.addr() & STATE_MASK == RUNNING); + wait(&self.state_and_queue, state_and_queue); + state_and_queue = self.state_and_queue.load(Ordering::Acquire); + } + } + } + } +} + +fn wait(state_and_queue: &AtomicPtr, mut current_state: *mut Masked) { + // Note: the following code was carefully written to avoid creating a + // mutable reference to `node` that gets aliased. + loop { + // Don't queue this thread if the status is no longer running, + // otherwise we will not be woken up. + if current_state.addr() & STATE_MASK != RUNNING { + return; + } + + // Create the node for our current thread. + let node = Waiter { + thread: Cell::new(Some(thread::current())), + signaled: AtomicBool::new(false), + next: current_state.with_addr(current_state.addr() & !STATE_MASK) as *const Waiter, + }; + let me = &node as *const Waiter as *const Masked as *mut Masked; + + // Try to slide in the node at the head of the linked list, making sure + // that another thread didn't just replace the head of the linked list. + let exchange_result = state_and_queue.compare_exchange( + current_state, + me.with_addr(me.addr() | RUNNING), + Ordering::Release, + Ordering::Relaxed, + ); + if let Err(old) = exchange_result { + current_state = old; + continue; + } + + // We have enqueued ourselves, now lets wait. + // It is important not to return before being signaled, otherwise we + // would drop our `Waiter` node and leave a hole in the linked list + // (and a dangling reference). Guard against spurious wakeups by + // reparking ourselves until we are signaled. + while !node.signaled.load(Ordering::Acquire) { + // If the managing thread happens to signal and unpark us before we + // can park ourselves, the result could be this thread never gets + // unparked. Luckily `park` comes with the guarantee that if it got + // an `unpark` just before on an unparked thread it does not park. + thread::park(); + } + break; + } +} + +#[stable(feature = "std_debug", since = "1.16.0")] +impl fmt::Debug for Once { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("Once").finish_non_exhaustive() + } +} + +impl Drop for WaiterQueue<'_> { + fn drop(&mut self) { + // Swap out our state with however we finished. + let state_and_queue = + self.state_and_queue.swap(self.set_state_on_drop_to, Ordering::AcqRel); + + // We should only ever see an old state which was RUNNING. + assert_eq!(state_and_queue.addr() & STATE_MASK, RUNNING); + + // Walk the entire linked list of waiters and wake them up (in lifo + // order, last to register is first to wake up). + unsafe { + // Right after setting `node.signaled = true` the other thread may + // free `node` if there happens to be has a spurious wakeup. + // So we have to take out the `thread` field and copy the pointer to + // `next` first. + let mut queue = + state_and_queue.with_addr(state_and_queue.addr() & !STATE_MASK) as *const Waiter; + while !queue.is_null() { + let next = (*queue).next; + let thread = (*queue).thread.take().unwrap(); + (*queue).signaled.store(true, Ordering::Release); + // ^- FIXME (maybe): This is another case of issue #55005 + // `store()` has a potentially dangling ref to `signaled`. + queue = next; + thread.unpark(); + } + } + } +} + +impl OnceState { + #[inline] + pub fn is_poisoned(&self) -> bool { + self.poisoned + } + + #[inline] + pub fn poison(&self) { + self.set_state_on_drop_to.set(ptr::invalid_mut(POISONED)); + } +} 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 { @@ -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, Vec)> { + 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.rs b/library/std/src/sys_common/remutex.rs deleted file mode 100644 index 4c054da64..000000000 --- a/library/std/src/sys_common/remutex.rs +++ /dev/null @@ -1,178 +0,0 @@ -#[cfg(all(test, not(target_os = "emscripten")))] -mod tests; - -use crate::cell::UnsafeCell; -use crate::ops::Deref; -use crate::panic::{RefUnwindSafe, UnwindSafe}; -use crate::sync::atomic::{AtomicUsize, Ordering::Relaxed}; -use crate::sys::locks as sys; - -/// A re-entrant mutual exclusion -/// -/// This mutex will block *other* threads waiting for the lock to become -/// available. The thread which has already locked the mutex can lock it -/// multiple times without blocking, preventing a common source of deadlocks. -/// -/// This is used by stdout().lock() and friends. -/// -/// ## Implementation details -/// -/// The 'owner' field tracks which thread has locked the mutex. -/// -/// We use current_thread_unique_ptr() as the thread identifier, -/// which is just the address of a thread local variable. -/// -/// If `owner` is set to the identifier of the current thread, -/// we assume the mutex is already locked and instead of locking it again, -/// we increment `lock_count`. -/// -/// When unlocking, we decrement `lock_count`, and only unlock the mutex when -/// it reaches zero. -/// -/// `lock_count` is protected by the mutex and only accessed by the thread that has -/// locked the mutex, so needs no synchronization. -/// -/// `owner` can be checked by other threads that want to see if they already -/// hold the lock, so needs to be atomic. If it compares equal, we're on the -/// same thread that holds the mutex and memory access can use relaxed ordering -/// since we're not dealing with multiple threads. If it compares unequal, -/// synchronization is left to the mutex, making relaxed memory ordering for -/// the `owner` field fine in all cases. -pub struct ReentrantMutex { - mutex: sys::Mutex, - owner: AtomicUsize, - lock_count: UnsafeCell, - data: T, -} - -unsafe impl Send for ReentrantMutex {} -unsafe impl Sync for ReentrantMutex {} - -impl UnwindSafe for ReentrantMutex {} -impl RefUnwindSafe for ReentrantMutex {} - -/// An RAII implementation of a "scoped lock" of a mutex. When this structure is -/// dropped (falls out of scope), the lock will be unlocked. -/// -/// The data protected by the mutex can be accessed through this guard via its -/// Deref implementation. -/// -/// # Mutability -/// -/// Unlike `MutexGuard`, `ReentrantMutexGuard` does not implement `DerefMut`, -/// because implementation of the trait would violate Rust’s reference aliasing -/// rules. Use interior mutability (usually `RefCell`) in order to mutate the -/// guarded data. -#[must_use = "if unused the ReentrantMutex will immediately unlock"] -pub struct ReentrantMutexGuard<'a, T: 'a> { - lock: &'a ReentrantMutex, -} - -impl !Send for ReentrantMutexGuard<'_, T> {} - -impl ReentrantMutex { - /// Creates a new reentrant mutex in an unlocked state. - pub const fn new(t: T) -> ReentrantMutex { - ReentrantMutex { - mutex: sys::Mutex::new(), - owner: AtomicUsize::new(0), - lock_count: UnsafeCell::new(0), - data: t, - } - } - - /// Acquires a mutex, blocking the current thread until it is able to do so. - /// - /// This function will block the caller until it is available to acquire the mutex. - /// Upon returning, the thread is the only thread with the mutex held. When the thread - /// calling this method already holds the lock, the call shall succeed without - /// blocking. - /// - /// # Errors - /// - /// If another user of this mutex panicked while holding the mutex, then - /// this call will return failure if the mutex would otherwise be - /// acquired. - pub fn lock(&self) -> ReentrantMutexGuard<'_, T> { - let this_thread = current_thread_unique_ptr(); - // Safety: We only touch lock_count when we own the lock. - unsafe { - if self.owner.load(Relaxed) == this_thread { - self.increment_lock_count(); - } else { - self.mutex.lock(); - self.owner.store(this_thread, Relaxed); - debug_assert_eq!(*self.lock_count.get(), 0); - *self.lock_count.get() = 1; - } - } - ReentrantMutexGuard { lock: self } - } - - /// Attempts to acquire this lock. - /// - /// If the lock could not be acquired at this time, then `Err` is returned. - /// Otherwise, an RAII guard is returned. - /// - /// This function does not block. - /// - /// # Errors - /// - /// If another user of this mutex panicked while holding the mutex, then - /// this call will return failure if the mutex would otherwise be - /// acquired. - pub fn try_lock(&self) -> Option> { - let this_thread = current_thread_unique_ptr(); - // Safety: We only touch lock_count when we own the lock. - unsafe { - if self.owner.load(Relaxed) == this_thread { - self.increment_lock_count(); - Some(ReentrantMutexGuard { lock: self }) - } else if self.mutex.try_lock() { - self.owner.store(this_thread, Relaxed); - debug_assert_eq!(*self.lock_count.get(), 0); - *self.lock_count.get() = 1; - Some(ReentrantMutexGuard { lock: self }) - } else { - None - } - } - } - - unsafe fn increment_lock_count(&self) { - *self.lock_count.get() = (*self.lock_count.get()) - .checked_add(1) - .expect("lock count overflow in reentrant mutex"); - } -} - -impl Deref for ReentrantMutexGuard<'_, T> { - type Target = T; - - fn deref(&self) -> &T { - &self.lock.data - } -} - -impl Drop for ReentrantMutexGuard<'_, T> { - #[inline] - fn drop(&mut self) { - // Safety: We own the lock. - unsafe { - *self.lock.lock_count.get() -= 1; - if *self.lock.lock_count.get() == 0 { - self.lock.owner.store(0, Relaxed); - self.lock.mutex.unlock(); - } - } - } -} - -/// Get an address that is unique per running thread. -/// -/// This can be used as a non-null usize-sized ID. -pub fn current_thread_unique_ptr() -> usize { - // Use a non-drop type to make sure it's still available during thread destruction. - thread_local! { static X: u8 = const { 0 } } - X.with(|x| <*const _>::addr(x)) -} 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>); -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) -> 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/futex.rs b/library/std/src/sys_common/thread_parker/futex.rs deleted file mode 100644 index d9e2f39e3..000000000 --- a/library/std/src/sys_common/thread_parker/futex.rs +++ /dev/null @@ -1,97 +0,0 @@ -use crate::pin::Pin; -use crate::sync::atomic::AtomicU32; -use crate::sync::atomic::Ordering::{Acquire, Release}; -use crate::sys::futex::{futex_wait, futex_wake}; -use crate::time::Duration; - -const PARKED: u32 = u32::MAX; -const EMPTY: u32 = 0; -const NOTIFIED: u32 = 1; - -pub struct Parker { - state: AtomicU32, -} - -// Notes about memory ordering: -// -// Memory ordering is only relevant for the relative ordering of operations -// between different variables. Even Ordering::Relaxed guarantees a -// monotonic/consistent order when looking at just a single atomic variable. -// -// So, since this parker is just a single atomic variable, we only need to look -// at the ordering guarantees we need to provide to the 'outside world'. -// -// The only memory ordering guarantee that parking and unparking provide, is -// that things which happened before unpark() are visible on the thread -// returning from park() afterwards. Otherwise, it was effectively unparked -// before unpark() was called while still consuming the 'token'. -// -// In other words, unpark() needs to synchronize with the part of park() that -// consumes the token and returns. -// -// This is done with a release-acquire synchronization, by using -// Ordering::Release when writing NOTIFIED (the 'token') in unpark(), and using -// Ordering::Acquire when checking for this state in park(). -impl Parker { - /// Construct the futex parker. The UNIX parker implementation - /// requires this to happen in-place. - pub unsafe fn new(parker: *mut Parker) { - parker.write(Self { state: AtomicU32::new(EMPTY) }); - } - - // Assumes this is only called by the thread that owns the Parker, - // which means that `self.state != PARKED`. - pub unsafe fn park(self: Pin<&Self>) { - // Change NOTIFIED=>EMPTY or EMPTY=>PARKED, and directly return in the - // first case. - if self.state.fetch_sub(1, Acquire) == NOTIFIED { - return; - } - loop { - // Wait for something to happen, assuming it's still set to PARKED. - futex_wait(&self.state, PARKED, None); - // Change NOTIFIED=>EMPTY and return in that case. - if self.state.compare_exchange(NOTIFIED, EMPTY, Acquire, Acquire).is_ok() { - return; - } else { - // Spurious wake up. We loop to try again. - } - } - } - - // Assumes this is only called by the thread that owns the Parker, - // which means that `self.state != PARKED`. This implementation doesn't - // require `Pin`, but other implementations do. - pub unsafe fn park_timeout(self: Pin<&Self>, timeout: Duration) { - // Change NOTIFIED=>EMPTY or EMPTY=>PARKED, and directly return in the - // first case. - if self.state.fetch_sub(1, Acquire) == NOTIFIED { - return; - } - // Wait for something to happen, assuming it's still set to PARKED. - futex_wait(&self.state, PARKED, Some(timeout)); - // This is not just a store, because we need to establish a - // release-acquire ordering with unpark(). - if self.state.swap(EMPTY, Acquire) == NOTIFIED { - // Woke up because of unpark(). - } else { - // Timeout or spurious wake up. - // We return either way, because we can't easily tell if it was the - // timeout or not. - } - } - - // This implementation doesn't require `Pin`, but other implementations do. - #[inline] - pub fn unpark(self: Pin<&Self>) { - // Change PARKED=>NOTIFIED, EMPTY=>NOTIFIED, or NOTIFIED=>NOTIFIED, and - // wake the thread in the first case. - // - // Note that even NOTIFIED=>NOTIFIED results in a write. This is on - // purpose, to make sure every unpark() has a release-acquire ordering - // with park(). - if self.state.swap(NOTIFIED, Release) == PARKED { - futex_wake(&self.state); - } - } -} diff --git a/library/std/src/sys_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/wait_flag.rs b/library/std/src/sys_common/thread_parker/wait_flag.rs deleted file mode 100644 index 6561c1866..000000000 --- a/library/std/src/sys_common/thread_parker/wait_flag.rs +++ /dev/null @@ -1,102 +0,0 @@ -//! A wait-flag-based thread parker. -//! -//! Some operating systems provide low-level parking primitives like wait counts, -//! event flags or semaphores which are not susceptible to race conditions (meaning -//! the wakeup can occur before the wait operation). To implement the `std` thread -//! parker on top of these primitives, we only have to ensure that parking is fast -//! when the thread token is available, the atomic ordering guarantees are maintained -//! and spurious wakeups are minimized. -//! -//! To achieve this, this parker uses an atomic variable with three states: `EMPTY`, -//! `PARKED` and `NOTIFIED`: -//! * `EMPTY` means the token has not been made available, but the thread is not -//! currently waiting on it. -//! * `PARKED` means the token is not available and the thread is parked. -//! * `NOTIFIED` means the token is available. -//! -//! `park` and `park_timeout` change the state from `EMPTY` to `PARKED` and from -//! `NOTIFIED` to `EMPTY`. If the state was `NOTIFIED`, the thread was unparked and -//! execution can continue without calling into the OS. If the state was `EMPTY`, -//! the token is not available and the thread waits on the primitive (here called -//! "wait flag"). -//! -//! `unpark` changes the state to `NOTIFIED`. If the state was `PARKED`, the thread -//! is or will be sleeping on the wait flag, so we raise it. - -use crate::pin::Pin; -use crate::sync::atomic::AtomicI8; -use crate::sync::atomic::Ordering::{Acquire, Relaxed, Release}; -use crate::sys::wait_flag::WaitFlag; -use crate::time::Duration; - -const EMPTY: i8 = 0; -const PARKED: i8 = -1; -const NOTIFIED: i8 = 1; - -pub struct Parker { - state: AtomicI8, - wait_flag: WaitFlag, -} - -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) { - parker.write(Parker { state: AtomicI8::new(EMPTY), wait_flag: WaitFlag::new() }) - } - - // This implementation doesn't require `unsafe` and `Pin`, but other implementations do. - pub unsafe fn park(self: Pin<&Self>) { - match self.state.fetch_sub(1, Acquire) { - // NOTIFIED => EMPTY - NOTIFIED => return, - // EMPTY => PARKED - EMPTY => (), - _ => panic!("inconsistent park state"), - } - - // Avoid waking up from spurious wakeups (these are quite likely, see below). - loop { - self.wait_flag.wait(); - - match self.state.compare_exchange(NOTIFIED, EMPTY, Acquire, Relaxed) { - Ok(_) => return, - Err(PARKED) => (), - Err(_) => panic!("inconsistent park state"), - } - } - } - - // This implementation doesn't require `unsafe` and `Pin`, but other implementations do. - pub unsafe fn park_timeout(self: Pin<&Self>, dur: Duration) { - match self.state.fetch_sub(1, Acquire) { - NOTIFIED => return, - EMPTY => (), - _ => panic!("inconsistent park state"), - } - - self.wait_flag.wait_timeout(dur); - - // Either a wakeup or a timeout occurred. Wakeups may be spurious, as there can be - // a race condition when `unpark` is performed between receiving the timeout and - // resetting the state, resulting in the eventflag being set unnecessarily. `park` - // is protected against this by looping until the token is actually given, but - // here we cannot easily tell. - - // Use `swap` to provide acquire ordering. - match self.state.swap(EMPTY, Acquire) { - NOTIFIED => (), - PARKED => (), - _ => panic!("inconsistent park state"), - } - } - - // This implementation doesn't require `Pin`, but other implementations do. - pub fn unpark(self: Pin<&Self>) { - let state = self.state.swap(NOTIFIED, Release); - - if state == PARKED { - self.wait_flag.raise(); - } - } -} diff --git a/library/std/src/sys_common/thread_parking/futex.rs b/library/std/src/sys_common/thread_parking/futex.rs new file mode 100644 index 000000000..588e7b278 --- /dev/null +++ b/library/std/src/sys_common/thread_parking/futex.rs @@ -0,0 +1,97 @@ +use crate::pin::Pin; +use crate::sync::atomic::AtomicU32; +use crate::sync::atomic::Ordering::{Acquire, Release}; +use crate::sys::futex::{futex_wait, futex_wake}; +use crate::time::Duration; + +const PARKED: u32 = u32::MAX; +const EMPTY: u32 = 0; +const NOTIFIED: u32 = 1; + +pub struct Parker { + state: AtomicU32, +} + +// Notes about memory ordering: +// +// Memory ordering is only relevant for the relative ordering of operations +// between different variables. Even Ordering::Relaxed guarantees a +// monotonic/consistent order when looking at just a single atomic variable. +// +// So, since this parker is just a single atomic variable, we only need to look +// at the ordering guarantees we need to provide to the 'outside world'. +// +// The only memory ordering guarantee that parking and unparking provide, is +// that things which happened before unpark() are visible on the thread +// returning from park() afterwards. Otherwise, it was effectively unparked +// before unpark() was called while still consuming the 'token'. +// +// In other words, unpark() needs to synchronize with the part of park() that +// consumes the token and returns. +// +// This is done with a release-acquire synchronization, by using +// Ordering::Release when writing NOTIFIED (the 'token') in unpark(), and using +// Ordering::Acquire when checking for this state in park(). +impl Parker { + /// Construct the futex parker. The UNIX parker implementation + /// requires this to happen in-place. + pub unsafe fn new_in_place(parker: *mut Parker) { + parker.write(Self { state: AtomicU32::new(EMPTY) }); + } + + // Assumes this is only called by the thread that owns the Parker, + // which means that `self.state != PARKED`. + pub unsafe fn park(self: Pin<&Self>) { + // Change NOTIFIED=>EMPTY or EMPTY=>PARKED, and directly return in the + // first case. + if self.state.fetch_sub(1, Acquire) == NOTIFIED { + return; + } + loop { + // Wait for something to happen, assuming it's still set to PARKED. + futex_wait(&self.state, PARKED, None); + // Change NOTIFIED=>EMPTY and return in that case. + if self.state.compare_exchange(NOTIFIED, EMPTY, Acquire, Acquire).is_ok() { + return; + } else { + // Spurious wake up. We loop to try again. + } + } + } + + // Assumes this is only called by the thread that owns the Parker, + // which means that `self.state != PARKED`. This implementation doesn't + // require `Pin`, but other implementations do. + pub unsafe fn park_timeout(self: Pin<&Self>, timeout: Duration) { + // Change NOTIFIED=>EMPTY or EMPTY=>PARKED, and directly return in the + // first case. + if self.state.fetch_sub(1, Acquire) == NOTIFIED { + return; + } + // Wait for something to happen, assuming it's still set to PARKED. + futex_wait(&self.state, PARKED, Some(timeout)); + // This is not just a store, because we need to establish a + // release-acquire ordering with unpark(). + if self.state.swap(EMPTY, Acquire) == NOTIFIED { + // Woke up because of unpark(). + } else { + // Timeout or spurious wake up. + // We return either way, because we can't easily tell if it was the + // timeout or not. + } + } + + // This implementation doesn't require `Pin`, but other implementations do. + #[inline] + pub fn unpark(self: Pin<&Self>) { + // Change PARKED=>NOTIFIED, EMPTY=>NOTIFIED, or NOTIFIED=>NOTIFIED, and + // wake the thread in the first case. + // + // Note that even NOTIFIED=>NOTIFIED results in a write. This is on + // purpose, to make sure every unpark() has a release-acquire ordering + // with park(). + if self.state.swap(NOTIFIED, Release) == PARKED { + futex_wake(&self.state); + } + } +} diff --git a/library/std/src/sys_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>, +} + +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_parking/wait_flag.rs b/library/std/src/sys_common/thread_parking/wait_flag.rs new file mode 100644 index 000000000..d0f8899a9 --- /dev/null +++ b/library/std/src/sys_common/thread_parking/wait_flag.rs @@ -0,0 +1,102 @@ +//! A wait-flag-based thread parker. +//! +//! Some operating systems provide low-level parking primitives like wait counts, +//! event flags or semaphores which are not susceptible to race conditions (meaning +//! the wakeup can occur before the wait operation). To implement the `std` thread +//! parker on top of these primitives, we only have to ensure that parking is fast +//! when the thread token is available, the atomic ordering guarantees are maintained +//! and spurious wakeups are minimized. +//! +//! To achieve this, this parker uses an atomic variable with three states: `EMPTY`, +//! `PARKED` and `NOTIFIED`: +//! * `EMPTY` means the token has not been made available, but the thread is not +//! currently waiting on it. +//! * `PARKED` means the token is not available and the thread is parked. +//! * `NOTIFIED` means the token is available. +//! +//! `park` and `park_timeout` change the state from `EMPTY` to `PARKED` and from +//! `NOTIFIED` to `EMPTY`. If the state was `NOTIFIED`, the thread was unparked and +//! execution can continue without calling into the OS. If the state was `EMPTY`, +//! the token is not available and the thread waits on the primitive (here called +//! "wait flag"). +//! +//! `unpark` changes the state to `NOTIFIED`. If the state was `PARKED`, the thread +//! is or will be sleeping on the wait flag, so we raise it. + +use crate::pin::Pin; +use crate::sync::atomic::AtomicI8; +use crate::sync::atomic::Ordering::{Acquire, Relaxed, Release}; +use crate::sys::wait_flag::WaitFlag; +use crate::time::Duration; + +const EMPTY: i8 = 0; +const PARKED: i8 = -1; +const NOTIFIED: i8 = 1; + +pub struct Parker { + state: AtomicI8, + wait_flag: WaitFlag, +} + +impl Parker { + /// Construct a parker for the current thread. The UNIX parker + /// implementation requires this to happen in-place. + pub unsafe fn new_in_place(parker: *mut Parker) { + parker.write(Parker { state: AtomicI8::new(EMPTY), wait_flag: WaitFlag::new() }) + } + + // This implementation doesn't require `unsafe` and `Pin`, but other implementations do. + pub unsafe fn park(self: Pin<&Self>) { + match self.state.fetch_sub(1, Acquire) { + // NOTIFIED => EMPTY + NOTIFIED => return, + // EMPTY => PARKED + EMPTY => (), + _ => panic!("inconsistent park state"), + } + + // Avoid waking up from spurious wakeups (these are quite likely, see below). + loop { + self.wait_flag.wait(); + + match self.state.compare_exchange(NOTIFIED, EMPTY, Acquire, Relaxed) { + Ok(_) => return, + Err(PARKED) => (), + Err(_) => panic!("inconsistent park state"), + } + } + } + + // This implementation doesn't require `unsafe` and `Pin`, but other implementations do. + pub unsafe fn park_timeout(self: Pin<&Self>, dur: Duration) { + match self.state.fetch_sub(1, Acquire) { + NOTIFIED => return, + EMPTY => (), + _ => panic!("inconsistent park state"), + } + + self.wait_flag.wait_timeout(dur); + + // Either a wakeup or a timeout occurred. Wakeups may be spurious, as there can be + // a race condition when `unpark` is performed between receiving the timeout and + // resetting the state, resulting in the eventflag being set unnecessarily. `park` + // is protected against this by looping until the token is actually given, but + // here we cannot easily tell. + + // Use `swap` to provide acquire ordering. + match self.state.swap(EMPTY, Acquire) { + NOTIFIED => (), + PARKED => (), + _ => panic!("inconsistent park state"), + } + } + + // This implementation doesn't require `Pin`, but other implementations do. + pub fn unpark(self: Pin<&Self>) { + let state = self.state.swap(NOTIFIED, Release); + + if state == PARKED { + self.wait_flag.raise(); + } + } +} -- cgit v1.2.3