diff options
Diffstat (limited to 'library/std/src/sync')
| -rw-r--r-- | library/std/src/sync/barrier/tests.rs | 4 | ||||
| -rw-r--r-- | library/std/src/sync/condvar.rs | 2 | ||||
| -rw-r--r-- | library/std/src/sync/condvar/tests.rs | 16 | ||||
| -rw-r--r-- | library/std/src/sync/lazy_lock/tests.rs | 6 | ||||
| -rw-r--r-- | library/std/src/sync/mod.rs | 18 | ||||
| -rw-r--r-- | library/std/src/sync/mpmc/context.rs | 2 | ||||
| -rw-r--r-- | library/std/src/sync/mpmc/error.rs | 5 | ||||
| -rw-r--r-- | library/std/src/sync/mpmc/mod.rs | 1037 | ||||
| -rw-r--r-- | library/std/src/sync/mpmc/tests.rs | 728 | ||||
| -rw-r--r-- | library/std/src/sync/mpmc/waker.rs | 2 | ||||
| -rw-r--r-- | library/std/src/sync/mpmc/zero.rs | 14 | ||||
| -rw-r--r-- | library/std/src/sync/mpsc/mod.rs | 4 | ||||
| -rw-r--r-- | library/std/src/sync/mutex.rs | 4 | ||||
| -rw-r--r-- | library/std/src/sync/once.rs | 2 | ||||
| -rw-r--r-- | library/std/src/sync/once_lock/tests.rs | 9 | ||||
| -rw-r--r-- | library/std/src/sync/reentrant_lock.rs | 4 | ||||
| -rw-r--r-- | library/std/src/sync/rwlock.rs | 4 |
17 files changed, 1771 insertions, 90 deletions
diff --git a/library/std/src/sync/barrier/tests.rs b/library/std/src/sync/barrier/tests.rs index 834a3e75158..0fbcd998812 100644 --- a/library/std/src/sync/barrier/tests.rs +++ b/library/std/src/sync/barrier/tests.rs @@ -1,9 +1,9 @@ -use crate::sync::mpsc::{channel, TryRecvError}; +use crate::sync::mpsc::{TryRecvError, channel}; use crate::sync::{Arc, Barrier}; use crate::thread; #[test] -#[cfg_attr(target_os = "emscripten", ignore)] +#[cfg_attr(any(target_os = "emscripten", target_os = "wasi"), ignore)] // no threads fn test_barrier() { const N: usize = 10; diff --git a/library/std/src/sync/condvar.rs b/library/std/src/sync/condvar.rs index e41cbc1a65c..44ffcb528d9 100644 --- a/library/std/src/sync/condvar.rs +++ b/library/std/src/sync/condvar.rs @@ -2,7 +2,7 @@ mod tests; use crate::fmt; -use crate::sync::{mutex, poison, LockResult, MutexGuard, PoisonError}; +use crate::sync::{LockResult, MutexGuard, PoisonError, mutex, poison}; use crate::sys::sync as sys; use crate::time::{Duration, Instant}; diff --git a/library/std/src/sync/condvar/tests.rs b/library/std/src/sync/condvar/tests.rs index 12d13a6b20b..f9e9066bc92 100644 --- a/library/std/src/sync/condvar/tests.rs +++ b/library/std/src/sync/condvar/tests.rs @@ -12,7 +12,7 @@ fn smoke() { } #[test] -#[cfg_attr(target_os = "emscripten", ignore)] +#[cfg_attr(any(target_os = "emscripten", target_os = "wasi"), ignore)] // no threads fn notify_one() { let m = Arc::new(Mutex::new(())); let m2 = m.clone(); @@ -29,7 +29,7 @@ fn notify_one() { } #[test] -#[cfg_attr(target_os = "emscripten", ignore)] +#[cfg_attr(any(target_os = "emscripten", target_os = "wasi"), ignore)] // no threads fn notify_all() { const N: usize = 10; @@ -66,7 +66,7 @@ fn notify_all() { } #[test] -#[cfg_attr(target_os = "emscripten", ignore)] +#[cfg_attr(any(target_os = "emscripten", target_os = "wasi"), ignore)] // no threads fn wait_while() { let pair = Arc::new((Mutex::new(false), Condvar::new())); let pair2 = pair.clone(); @@ -87,7 +87,7 @@ fn wait_while() { } #[test] -#[cfg_attr(target_os = "emscripten", ignore)] +#[cfg_attr(any(target_os = "emscripten", target_os = "wasi"), ignore)] // condvar wait not supported fn wait_timeout_wait() { let m = Arc::new(Mutex::new(())); let c = Arc::new(Condvar::new()); @@ -106,7 +106,7 @@ fn wait_timeout_wait() { } #[test] -#[cfg_attr(target_os = "emscripten", ignore)] +#[cfg_attr(any(target_os = "emscripten", target_os = "wasi"), ignore)] // condvar wait not supported fn wait_timeout_while_wait() { let m = Arc::new(Mutex::new(())); let c = Arc::new(Condvar::new()); @@ -118,7 +118,7 @@ fn wait_timeout_while_wait() { } #[test] -#[cfg_attr(target_os = "emscripten", ignore)] +#[cfg_attr(any(target_os = "emscripten", target_os = "wasi"), ignore)] // condvar wait not supported fn wait_timeout_while_instant_satisfy() { let m = Arc::new(Mutex::new(())); let c = Arc::new(Condvar::new()); @@ -130,7 +130,7 @@ fn wait_timeout_while_instant_satisfy() { } #[test] -#[cfg_attr(target_os = "emscripten", ignore)] +#[cfg_attr(any(target_os = "emscripten", target_os = "wasi"), ignore)] // no threads fn wait_timeout_while_wake() { let pair = Arc::new((Mutex::new(false), Condvar::new())); let pair_copy = pair.clone(); @@ -153,7 +153,7 @@ fn wait_timeout_while_wake() { } #[test] -#[cfg_attr(target_os = "emscripten", ignore)] +#[cfg_attr(any(target_os = "emscripten", target_os = "wasi"), ignore)] // no threads fn wait_timeout_wake() { let m = Arc::new(Mutex::new(())); let c = Arc::new(Condvar::new()); diff --git a/library/std/src/sync/lazy_lock/tests.rs b/library/std/src/sync/lazy_lock/tests.rs index 94044368305..7d7dde54349 100644 --- a/library/std/src/sync/lazy_lock/tests.rs +++ b/library/std/src/sync/lazy_lock/tests.rs @@ -34,6 +34,7 @@ fn lazy_default() { } #[test] +#[cfg_attr(not(panic = "unwind"), ignore = "test requires unwinding support")] fn lazy_poisoning() { let x: LazyCell<String> = LazyCell::new(|| panic!("kaboom")); for _ in 0..2 { @@ -43,7 +44,7 @@ fn lazy_poisoning() { } #[test] -#[cfg_attr(target_os = "emscripten", ignore)] +#[cfg_attr(any(target_os = "emscripten", target_os = "wasi"), ignore)] // no threads fn sync_lazy_new() { static CALLED: AtomicUsize = AtomicUsize::new(0); static SYNC_LAZY: LazyLock<i32> = LazyLock::new(|| { @@ -90,7 +91,7 @@ fn sync_lazy_default() { } #[test] -#[cfg_attr(target_os = "emscripten", ignore)] +#[cfg_attr(any(target_os = "emscripten", target_os = "wasi"), ignore)] // no threads fn static_sync_lazy() { static XS: LazyLock<Vec<i32>> = LazyLock::new(|| { let mut xs = Vec::new(); @@ -123,6 +124,7 @@ fn static_sync_lazy_via_fn() { } #[test] +#[cfg_attr(not(panic = "unwind"), ignore = "test requires unwinding support")] fn sync_lazy_poisoning() { let x: LazyLock<String> = LazyLock::new(|| panic!("kaboom")); for _ in 0..2 { diff --git a/library/std/src/sync/mod.rs b/library/std/src/sync/mod.rs index d0ba8cc3b47..0fb77331293 100644 --- a/library/std/src/sync/mod.rs +++ b/library/std/src/sync/mod.rs @@ -9,6 +9,9 @@ //! Consider the following code, operating on some global static variables: //! //! ```rust +//! // FIXME(static_mut_refs): Do not allow `static_mut_refs` lint +//! #![allow(static_mut_refs)] +//! //! static mut A: u32 = 0; //! static mut B: u32 = 0; //! static mut C: u32 = 0; @@ -130,6 +133,11 @@ //! inter-thread synchronisation mechanism, at the cost of some //! extra memory. //! +//! - [`mpmc`]: Multi-producer, multi-consumer queues, used for +//! message-based communication. Can provide a lightweight +//! inter-thread synchronisation mechanism, at the cost of some +//! extra memory. +//! //! - [`Mutex`]: Mutual Exclusion mechanism, which ensures that at //! most one thread at a time is able to access some data. //! @@ -150,6 +158,7 @@ //! [`Arc`]: crate::sync::Arc //! [`Barrier`]: crate::sync::Barrier //! [`Condvar`]: crate::sync::Condvar +//! [`mpmc`]: crate::sync::mpmc //! [`mpsc`]: crate::sync::mpsc //! [`Mutex`]: crate::sync::Mutex //! [`Once`]: crate::sync::Once @@ -158,10 +167,10 @@ #![stable(feature = "rust1", since = "1.0.0")] -#[stable(feature = "rust1", since = "1.0.0")] -pub use core::sync::atomic; #[unstable(feature = "exclusive_wrapper", issue = "98407")] pub use core::sync::Exclusive; +#[stable(feature = "rust1", since = "1.0.0")] +pub use core::sync::atomic; #[stable(feature = "rust1", since = "1.0.0")] pub use alloc_crate::sync::{Arc, Weak}; @@ -178,7 +187,7 @@ pub use self::mutex::MappedMutexGuard; pub use self::mutex::{Mutex, MutexGuard}; #[stable(feature = "rust1", since = "1.0.0")] #[allow(deprecated)] -pub use self::once::{Once, OnceState, ONCE_INIT}; +pub use self::once::{ONCE_INIT, Once, OnceState}; #[stable(feature = "once_cell", since = "1.70.0")] pub use self::once_lock::OnceLock; #[stable(feature = "rust1", since = "1.0.0")] @@ -190,12 +199,13 @@ pub use self::rwlock::{MappedRwLockReadGuard, MappedRwLockWriteGuard}; #[stable(feature = "rust1", since = "1.0.0")] pub use self::rwlock::{RwLock, RwLockReadGuard, RwLockWriteGuard}; +#[unstable(feature = "mpmc_channel", issue = "126840")] +pub mod mpmc; pub mod mpsc; mod barrier; mod condvar; mod lazy_lock; -mod mpmc; mod mutex; pub(crate) mod once; mod once_lock; diff --git a/library/std/src/sync/mpmc/context.rs b/library/std/src/sync/mpmc/context.rs index 8db3c9896eb..2371d32d4ea 100644 --- a/library/std/src/sync/mpmc/context.rs +++ b/library/std/src/sync/mpmc/context.rs @@ -4,8 +4,8 @@ use super::select::Selected; use super::waker::current_thread_id; use crate::cell::Cell; use crate::ptr; -use crate::sync::atomic::{AtomicPtr, AtomicUsize, Ordering}; use crate::sync::Arc; +use crate::sync::atomic::{AtomicPtr, AtomicUsize, Ordering}; use crate::thread::{self, Thread}; use crate::time::Instant; diff --git a/library/std/src/sync/mpmc/error.rs b/library/std/src/sync/mpmc/error.rs index e3aec7e7623..e34b56d0831 100644 --- a/library/std/src/sync/mpmc/error.rs +++ b/library/std/src/sync/mpmc/error.rs @@ -7,6 +7,7 @@ use crate::{error, fmt}; /// /// [`send_timeout`]: super::Sender::send_timeout #[derive(PartialEq, Eq, Clone, Copy)] +#[unstable(feature = "mpmc_channel", issue = "126840")] pub enum SendTimeoutError<T> { /// The message could not be sent because the channel is full and the operation timed out. /// @@ -18,12 +19,14 @@ pub enum SendTimeoutError<T> { Disconnected(T), } +#[unstable(feature = "mpmc_channel", issue = "126840")] impl<T> fmt::Debug for SendTimeoutError<T> { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { "SendTimeoutError(..)".fmt(f) } } +#[unstable(feature = "mpmc_channel", issue = "126840")] impl<T> fmt::Display for SendTimeoutError<T> { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { match *self { @@ -33,8 +36,10 @@ impl<T> fmt::Display for SendTimeoutError<T> { } } +#[unstable(feature = "mpmc_channel", issue = "126840")] impl<T> error::Error for SendTimeoutError<T> {} +#[unstable(feature = "mpmc_channel", issue = "126840")] impl<T> From<SendError<T>> for SendTimeoutError<T> { fn from(err: SendError<T>) -> SendTimeoutError<T> { match err { diff --git a/library/std/src/sync/mpmc/mod.rs b/library/std/src/sync/mpmc/mod.rs index c640e07348e..44e146a89ba 100644 --- a/library/std/src/sync/mpmc/mod.rs +++ b/library/std/src/sync/mpmc/mod.rs @@ -1,8 +1,114 @@ -//! Multi-producer multi-consumer channels. +//! Multi-producer, multi-consumer FIFO queue communication primitives. +//! +//! This module provides message-based communication over channels, concretely +//! defined by two types: +//! +//! * [`Sender`] +//! * [`Receiver`] +//! +//! [`Sender`]s are used to send data to a set of [`Receiver`]s. Both +//! sender and receiver are cloneable (multi-producer) such that many threads can send +//! simultaneously to receivers (multi-consumer). +//! +//! These channels come in two flavors: +//! +//! 1. An asynchronous, infinitely buffered channel. The [`channel`] function +//! will return a `(Sender, Receiver)` tuple where all sends will be +//! **asynchronous** (they never block). The channel conceptually has an +//! infinite buffer. +//! +//! 2. A synchronous, bounded channel. The [`sync_channel`] function will +//! return a `(SyncSender, Receiver)` tuple where the storage for pending +//! messages is a pre-allocated buffer of a fixed size. All sends will be +//! **synchronous** by blocking until there is buffer space available. Note +//! that a bound of 0 is allowed, causing the channel to become a "rendezvous" +//! channel where each sender atomically hands off a message to a receiver. +//! +//! [`send`]: Sender::send +//! +//! ## Disconnection +//! +//! The send and receive operations on channels will all return a [`Result`] +//! indicating whether the operation succeeded or not. An unsuccessful operation +//! is normally indicative of the other half of a channel having "hung up" by +//! being dropped in its corresponding thread. +//! +//! Once half of a channel has been deallocated, most operations can no longer +//! continue to make progress, so [`Err`] will be returned. Many applications +//! will continue to [`unwrap`] the results returned from this module, +//! instigating a propagation of failure among threads if one unexpectedly dies. +//! +//! [`unwrap`]: Result::unwrap +//! +//! # Examples +//! +//! Simple usage: +//! +//! ``` +//! #![feature(mpmc_channel)] +//! +//! use std::thread; +//! use std::sync::mpmc::channel; +//! +//! // Create a simple streaming channel +//! let (tx, rx) = channel(); +//! thread::spawn(move || { +//! tx.send(10).unwrap(); +//! }); +//! assert_eq!(rx.recv().unwrap(), 10); +//! ``` +//! +//! Shared usage: +//! +//! ``` +//! #![feature(mpmc_channel)] +//! +//! use std::thread; +//! use std::sync::mpmc::channel; +//! +//! thread::scope(|s| { +//! // Create a shared channel that can be sent along from many threads +//! // where tx is the sending half (tx for transmission), and rx is the receiving +//! // half (rx for receiving). +//! let (tx, rx) = channel(); +//! for i in 0..10 { +//! let tx = tx.clone(); +//! s.spawn(move || { +//! tx.send(i).unwrap(); +//! }); +//! } +//! +//! for _ in 0..5 { +//! let rx1 = rx.clone(); +//! let rx2 = rx.clone(); +//! s.spawn(move || { +//! let j = rx1.recv().unwrap(); +//! assert!(0 <= j && j < 10); +//! }); +//! s.spawn(move || { +//! let j = rx2.recv().unwrap(); +//! assert!(0 <= j && j < 10); +//! }); +//! } +//! }) +//! ``` +//! +//! Propagating panics: +//! +//! ``` +//! #![feature(mpmc_channel)] +//! +//! use std::sync::mpmc::channel; +//! +//! // The call to recv() will return an error because the channel has already +//! // hung up (or been deallocated) +//! let (tx, rx) = channel::<i32>(); +//! drop(tx); +//! assert!(rx.recv().is_err()); +//! ``` -// This module is not currently exposed publicly, but is used -// as the implementation for the channels in `sync::mpsc`. The -// implementation comes from the crossbeam-channel crate: +// This module is used as the implementation for the channels in `sync::mpsc`. +// The implementation comes from the crossbeam-channel crate: // // Copyright (c) 2019 The Crossbeam Project Developers // @@ -46,9 +152,47 @@ use crate::fmt; use crate::panic::{RefUnwindSafe, UnwindSafe}; use crate::time::{Duration, Instant}; -/// Creates a channel of unbounded capacity. +/// Creates a new asynchronous channel, returning the sender/receiver halves. +/// All data sent on the [`Sender`] will become available on the [`Receiver`] in +/// the same order as it was sent, and no [`send`] will block the calling thread +/// (this channel has an "infinite buffer", unlike [`sync_channel`], which will +/// block after its buffer limit is reached). [`recv`] will block until a message +/// is available while there is at least one [`Sender`] alive (including clones). /// -/// This channel has a growable buffer that can hold any number of messages at a time. +/// The [`Sender`] can be cloned to [`send`] to the same channel multiple times. +/// The [`Receiver`] also can be cloned to have multi receivers. +/// +/// If the [`Receiver`] is disconnected while trying to [`send`] with the +/// [`Sender`], the [`send`] method will return a [`SendError`]. Similarly, if the +/// [`Sender`] is disconnected while trying to [`recv`], the [`recv`] method will +/// return a [`RecvError`]. +/// +/// [`send`]: Sender::send +/// [`recv`]: Receiver::recv +/// +/// # Examples +/// +/// ``` +/// #![feature(mpmc_channel)] +/// +/// use std::sync::mpmc::channel; +/// use std::thread; +/// +/// let (sender, receiver) = channel(); +/// +/// // Spawn off an expensive computation +/// thread::spawn(move || { +/// # fn expensive_computation() {} +/// sender.send(expensive_computation()).unwrap(); +/// }); +/// +/// // Do some useful work for awhile +/// +/// // Let's see what that answer was +/// println!("{:?}", receiver.recv().unwrap()); +/// ``` +#[must_use] +#[unstable(feature = "mpmc_channel", issue = "126840")] pub fn channel<T>() -> (Sender<T>, Receiver<T>) { let (s, r) = counter::new(list::Channel::new()); let s = Sender { flavor: SenderFlavor::List(s) }; @@ -56,12 +200,50 @@ pub fn channel<T>() -> (Sender<T>, Receiver<T>) { (s, r) } -/// Creates a channel of bounded capacity. +/// Creates a new synchronous, bounded channel. +/// All data sent on the [`Sender`] will become available on the [`Receiver`] +/// in the same order as it was sent. Like asynchronous [`channel`]s, the +/// [`Receiver`] will block until a message becomes available. `sync_channel` +/// differs greatly in the semantics of the sender, however. +/// +/// This channel has an internal buffer on which messages will be queued. +/// `bound` specifies the buffer size. When the internal buffer becomes full, +/// future sends will *block* waiting for the buffer to open up. Note that a +/// buffer size of 0 is valid, in which case this becomes "rendezvous channel" +/// where each [`send`] will not return until a [`recv`] is paired with it. +/// +/// The [`Sender`] can be cloned to [`send`] to the same channel multiple +/// times. The [`Receiver`] also can be cloned to have multi receivers. +/// +/// Like asynchronous channels, if the [`Receiver`] is disconnected while trying +/// to [`send`] with the [`Sender`], the [`send`] method will return a +/// [`SendError`]. Similarly, If the [`Sender`] is disconnected while trying +/// to [`recv`], the [`recv`] method will return a [`RecvError`]. +/// +/// [`send`]: Sender::send +/// [`recv`]: Receiver::recv +/// +/// # Examples +/// +/// ``` +/// use std::sync::mpsc::sync_channel; +/// use std::thread; +/// +/// let (sender, receiver) = sync_channel(1); /// -/// This channel has a buffer that can hold at most `cap` messages at a time. +/// // this returns immediately +/// sender.send(1).unwrap(); /// -/// A special case is zero-capacity channel, which cannot hold any messages. Instead, send and -/// receive operations must appear at the same time in order to pair up and pass the message over. +/// thread::spawn(move || { +/// // this will block until the previous message has been received +/// sender.send(2).unwrap(); +/// }); +/// +/// assert_eq!(receiver.recv().unwrap(), 1); +/// assert_eq!(receiver.recv().unwrap(), 2); +/// ``` +#[must_use] +#[unstable(feature = "mpmc_channel", issue = "126840")] pub fn sync_channel<T>(cap: usize) -> (Sender<T>, Receiver<T>) { if cap == 0 { let (s, r) = counter::new(zero::Channel::new()); @@ -76,7 +258,42 @@ pub fn sync_channel<T>(cap: usize) -> (Sender<T>, Receiver<T>) { } } -/// The sending side of a channel. +/// The sending-half of Rust's synchronous [`channel`] type. +/// +/// Messages can be sent through this channel with [`send`]. +/// +/// Note: all senders (the original and its clones) need to be dropped for the receiver +/// to stop blocking to receive messages with [`Receiver::recv`]. +/// +/// [`send`]: Sender::send +/// +/// # Examples +/// +/// ```rust +/// #![feature(mpmc_channel)] +/// +/// use std::sync::mpmc::channel; +/// use std::thread; +/// +/// let (sender, receiver) = channel(); +/// let sender2 = sender.clone(); +/// +/// // First thread owns sender +/// thread::spawn(move || { +/// sender.send(1).unwrap(); +/// }); +/// +/// // Second thread owns sender2 +/// thread::spawn(move || { +/// sender2.send(2).unwrap(); +/// }); +/// +/// let msg = receiver.recv().unwrap(); +/// let msg2 = receiver.recv().unwrap(); +/// +/// assert_eq!(3, msg + msg2); +/// ``` +#[unstable(feature = "mpmc_channel", issue = "126840")] pub struct Sender<T> { flavor: SenderFlavor<T>, } @@ -93,10 +310,14 @@ enum SenderFlavor<T> { Zero(counter::Sender<zero::Channel<T>>), } +#[unstable(feature = "mpmc_channel", issue = "126840")] unsafe impl<T: Send> Send for Sender<T> {} +#[unstable(feature = "mpmc_channel", issue = "126840")] unsafe impl<T: Send> Sync for Sender<T> {} +#[unstable(feature = "mpmc_channel", issue = "126840")] impl<T> UnwindSafe for Sender<T> {} +#[unstable(feature = "mpmc_channel", issue = "126840")] impl<T> RefUnwindSafe for Sender<T> {} impl<T> Sender<T> { @@ -107,6 +328,19 @@ impl<T> Sender<T> { /// /// If called on a zero-capacity channel, this method will send the message only if there /// happens to be a receive operation on the other side of the channel at the same time. + /// + /// # Examples + /// + /// ```rust + /// #![feature(mpmc_channel)] + /// + /// use std::sync::mpmc::{channel, Receiver, Sender}; + /// + /// let (sender, _receiver): (Sender<i32>, Receiver<i32>) = channel(); + /// + /// assert!(sender.try_send(1).is_ok()); + /// ``` + #[unstable(feature = "mpmc_channel", issue = "126840")] pub fn try_send(&self, msg: T) -> Result<(), TrySendError<T>> { match &self.flavor { SenderFlavor::Array(chan) => chan.try_send(msg), @@ -115,14 +349,36 @@ impl<T> Sender<T> { } } - /// Blocks the current thread until a message is sent or the channel is disconnected. + /// Attempts to send a value on this channel, returning it back if it could + /// not be sent. /// - /// If the channel is full and not disconnected, this call will block until the send operation - /// can proceed. If the channel becomes disconnected, this call will wake up and return an - /// error. The returned error contains the original message. + /// A successful send occurs when it is determined that the other end of + /// the channel has not hung up already. An unsuccessful send would be one + /// where the corresponding receiver has already been deallocated. Note + /// that a return value of [`Err`] means that the data will never be + /// received, but a return value of [`Ok`] does *not* mean that the data + /// will be received. It is possible for the corresponding receiver to + /// hang up immediately after this function returns [`Ok`]. /// - /// If called on a zero-capacity channel, this method will wait for a receive operation to - /// appear on the other side of the channel. + /// This method will never block the current thread. + /// + /// # Examples + /// + /// ``` + /// #![feature(mpmc_channel)] + /// + /// use std::sync::mpmc::channel; + /// + /// let (tx, rx) = channel(); + /// + /// // This send is always successful + /// tx.send(1).unwrap(); + /// + /// // This send will fail because the receiver is gone + /// drop(rx); + /// assert!(tx.send(1).is_err()); + /// ``` + #[unstable(feature = "mpmc_channel", issue = "126840")] pub fn send(&self, msg: T) -> Result<(), SendError<T>> { match &self.flavor { SenderFlavor::Array(chan) => chan.send(msg, None), @@ -136,10 +392,6 @@ impl<T> Sender<T> { } } -// The methods below are not used by `sync::mpsc`, but -// are useful and we'll likely want to expose them -// eventually -#[allow(unused)] impl<T> Sender<T> { /// Waits for a message to be sent into the channel, but only for a limited time. /// @@ -149,6 +401,20 @@ impl<T> Sender<T> { /// /// If called on a zero-capacity channel, this method will wait for a receive operation to /// appear on the other side of the channel. + /// + /// # Examples + /// + /// ``` + /// #![feature(mpmc_channel)] + /// + /// use std::sync::mpmc::channel; + /// use std::time::Duration; + /// + /// let (tx, rx) = channel(); + /// + /// tx.send_timeout(1, Duration::from_millis(400)).unwrap(); + /// ``` + #[unstable(feature = "mpmc_channel", issue = "126840")] pub fn send_timeout(&self, msg: T, timeout: Duration) -> Result<(), SendTimeoutError<T>> { match Instant::now().checked_add(timeout) { Some(deadline) => self.send_deadline(msg, deadline), @@ -165,6 +431,21 @@ impl<T> Sender<T> { /// /// If called on a zero-capacity channel, this method will wait for a receive operation to /// appear on the other side of the channel. + /// + /// # Examples + /// + /// ``` + /// #![feature(mpmc_channel)] + /// + /// use std::sync::mpmc::channel; + /// use std::time::{Duration, Instant}; + /// + /// let (tx, rx) = channel(); + /// + /// let t = Instant::now() + Duration::from_millis(400); + /// tx.send_deadline(1, t).unwrap(); + /// ``` + #[unstable(feature = "mpmc_channel", issue = "126840")] pub fn send_deadline(&self, msg: T, deadline: Instant) -> Result<(), SendTimeoutError<T>> { match &self.flavor { SenderFlavor::Array(chan) => chan.send(msg, Some(deadline)), @@ -176,6 +457,31 @@ impl<T> Sender<T> { /// Returns `true` if the channel is empty. /// /// Note: Zero-capacity channels are always empty. + /// + /// # Examples + /// + /// ``` + /// #![feature(mpmc_channel)] + /// + /// use std::sync::mpmc; + /// use std::thread; + /// + /// let (send, _recv) = mpmc::channel(); + /// + /// let tx1 = send.clone(); + /// let tx2 = send.clone(); + /// + /// assert!(tx1.is_empty()); + /// + /// let handle = thread::spawn(move || { + /// tx2.send(1u8).unwrap(); + /// }); + /// + /// handle.join().unwrap(); + /// + /// assert!(!tx1.is_empty()); + /// ``` + #[unstable(feature = "mpmc_channel", issue = "126840")] pub fn is_empty(&self) -> bool { match &self.flavor { SenderFlavor::Array(chan) => chan.is_empty(), @@ -187,6 +493,29 @@ impl<T> Sender<T> { /// Returns `true` if the channel is full. /// /// Note: Zero-capacity channels are always full. + /// + /// # Examples + /// + /// ``` + /// #![feature(mpmc_channel)] + /// + /// use std::sync::mpmc; + /// use std::thread; + /// + /// let (send, _recv) = mpmc::sync_channel(1); + /// + /// let (tx1, tx2) = (send.clone(), send.clone()); + /// assert!(!tx1.is_full()); + /// + /// let handle = thread::spawn(move || { + /// tx2.send(1u8).unwrap(); + /// }); + /// + /// handle.join().unwrap(); + /// + /// assert!(tx1.is_full()); + /// ``` + #[unstable(feature = "mpmc_channel", issue = "126840")] pub fn is_full(&self) -> bool { match &self.flavor { SenderFlavor::Array(chan) => chan.is_full(), @@ -196,6 +525,29 @@ impl<T> Sender<T> { } /// Returns the number of messages in the channel. + /// + /// # Examples + /// + /// ``` + /// #![feature(mpmc_channel)] + /// + /// use std::sync::mpmc; + /// use std::thread; + /// + /// let (send, _recv) = mpmc::channel(); + /// let (tx1, tx2) = (send.clone(), send.clone()); + /// + /// assert_eq!(tx1.len(), 0); + /// + /// let handle = thread::spawn(move || { + /// tx2.send(1u8).unwrap(); + /// }); + /// + /// handle.join().unwrap(); + /// + /// assert_eq!(tx1.len(), 1); + /// ``` + #[unstable(feature = "mpmc_channel", issue = "126840")] pub fn len(&self) -> usize { match &self.flavor { SenderFlavor::Array(chan) => chan.len(), @@ -205,6 +557,29 @@ impl<T> Sender<T> { } /// If the channel is bounded, returns its capacity. + /// + /// # Examples + /// + /// ``` + /// #![feature(mpmc_channel)] + /// + /// use std::sync::mpmc; + /// use std::thread; + /// + /// let (send, _recv) = mpmc::sync_channel(3); + /// let (tx1, tx2) = (send.clone(), send.clone()); + /// + /// assert_eq!(tx1.capacity(), Some(3)); + /// + /// let handle = thread::spawn(move || { + /// tx2.send(1u8).unwrap(); + /// }); + /// + /// handle.join().unwrap(); + /// + /// assert_eq!(tx1.capacity(), Some(3)); + /// ``` + #[unstable(feature = "mpmc_channel", issue = "126840")] pub fn capacity(&self) -> Option<usize> { match &self.flavor { SenderFlavor::Array(chan) => chan.capacity(), @@ -214,6 +589,21 @@ impl<T> Sender<T> { } /// Returns `true` if senders belong to the same channel. + /// + /// # Examples + /// + /// ``` + /// #![feature(mpmc_channel)] + /// + /// use std::sync::mpmc; + /// + /// let (tx1, _) = mpmc::channel::<i32>(); + /// let (tx2, _) = mpmc::channel::<i32>(); + /// + /// assert!(tx1.same_channel(&tx1)); + /// assert!(!tx1.same_channel(&tx2)); + /// ``` + #[unstable(feature = "mpmc_channel", issue = "126840")] pub fn same_channel(&self, other: &Sender<T>) -> bool { match (&self.flavor, &other.flavor) { (SenderFlavor::Array(ref a), SenderFlavor::Array(ref b)) => a == b, @@ -224,6 +614,7 @@ impl<T> Sender<T> { } } +#[unstable(feature = "mpmc_channel", issue = "126840")] impl<T> Drop for Sender<T> { fn drop(&mut self) { unsafe { @@ -236,6 +627,7 @@ impl<T> Drop for Sender<T> { } } +#[unstable(feature = "mpmc_channel", issue = "126840")] impl<T> Clone for Sender<T> { fn clone(&self) -> Self { let flavor = match &self.flavor { @@ -248,17 +640,216 @@ impl<T> Clone for Sender<T> { } } +#[unstable(feature = "mpmc_channel", issue = "126840")] impl<T> fmt::Debug for Sender<T> { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.pad("Sender { .. }") } } -/// The receiving side of a channel. +/// The receiving half of Rust's [`channel`] (or [`sync_channel`]) type. +/// Different threads can share this [`Sender`] by cloning it. +/// +/// Messages sent to the channel can be retrieved using [`recv`]. +/// +/// [`recv`]: Receiver::recv +/// +/// # Examples +/// +/// ```rust +/// #![feature(mpmc_channel)] +/// +/// use std::sync::mpmc::channel; +/// use std::thread; +/// use std::time::Duration; +/// +/// let (send, recv) = channel(); +/// +/// let tx_thread = thread::spawn(move || { +/// send.send("Hello world!").unwrap(); +/// thread::sleep(Duration::from_secs(2)); // block for two seconds +/// send.send("Delayed for 2 seconds").unwrap(); +/// }); +/// +/// let (rx1, rx2) = (recv.clone(), recv.clone()); +/// let rx_thread_1 = thread::spawn(move || { +/// println!("{}", rx1.recv().unwrap()); // Received immediately +/// }); +/// let rx_thread_2 = thread::spawn(move || { +/// println!("{}", rx2.recv().unwrap()); // Received after 2 seconds +/// }); +/// +/// tx_thread.join().unwrap(); +/// rx_thread_1.join().unwrap(); +/// rx_thread_2.join().unwrap(); +/// ``` +#[unstable(feature = "mpmc_channel", issue = "126840")] pub struct Receiver<T> { flavor: ReceiverFlavor<T>, } +/// An iterator over messages on a [`Receiver`], created by [`iter`]. +/// +/// This iterator will block whenever [`next`] is called, +/// waiting for a new message, and [`None`] will be returned +/// when the corresponding channel has hung up. +/// +/// [`iter`]: Receiver::iter +/// [`next`]: Iterator::next +/// +/// # Examples +/// +/// ```rust +/// #![feature(mpmc_channel)] +/// +/// use std::sync::mpmc::channel; +/// use std::thread; +/// +/// let (send, recv) = channel(); +/// +/// thread::spawn(move || { +/// send.send(1u8).unwrap(); +/// send.send(2u8).unwrap(); +/// send.send(3u8).unwrap(); +/// }); +/// +/// for x in recv.iter() { +/// println!("Got: {x}"); +/// } +/// ``` +#[unstable(feature = "mpmc_channel", issue = "126840")] +#[derive(Debug)] +pub struct Iter<'a, T: 'a> { + rx: &'a Receiver<T>, +} + +/// An iterator that attempts to yield all pending values for a [`Receiver`], +/// created by [`try_iter`]. +/// +/// [`None`] will be returned when there are no pending values remaining or +/// if the corresponding channel has hung up. +/// +/// This iterator will never block the caller in order to wait for data to +/// become available. Instead, it will return [`None`]. +/// +/// [`try_iter`]: Receiver::try_iter +/// +/// # Examples +/// +/// ```rust +/// #![feature(mpmc_channel)] +/// +/// use std::sync::mpmc::channel; +/// use std::thread; +/// use std::time::Duration; +/// +/// let (sender, receiver) = channel(); +/// +/// // Nothing is in the buffer yet +/// assert!(receiver.try_iter().next().is_none()); +/// println!("Nothing in the buffer..."); +/// +/// thread::spawn(move || { +/// sender.send(1).unwrap(); +/// sender.send(2).unwrap(); +/// sender.send(3).unwrap(); +/// }); +/// +/// println!("Going to sleep..."); +/// thread::sleep(Duration::from_secs(2)); // block for two seconds +/// +/// for x in receiver.try_iter() { +/// println!("Got: {x}"); +/// } +/// ``` +#[unstable(feature = "mpmc_channel", issue = "126840")] +#[derive(Debug)] +pub struct TryIter<'a, T: 'a> { + rx: &'a Receiver<T>, +} + +/// An owning iterator over messages on a [`Receiver`], +/// created by [`into_iter`]. +/// +/// This iterator will block whenever [`next`] +/// is called, waiting for a new message, and [`None`] will be +/// returned if the corresponding channel has hung up. +/// +/// [`into_iter`]: Receiver::into_iter +/// [`next`]: Iterator::next +/// +/// # Examples +/// +/// ```rust +/// #![feature(mpmc_channel)] +/// +/// use std::sync::mpmc::channel; +/// use std::thread; +/// +/// let (send, recv) = channel(); +/// +/// thread::spawn(move || { +/// send.send(1u8).unwrap(); +/// send.send(2u8).unwrap(); +/// send.send(3u8).unwrap(); +/// }); +/// +/// for x in recv.into_iter() { +/// println!("Got: {x}"); +/// } +/// ``` +#[unstable(feature = "mpmc_channel", issue = "126840")] +#[derive(Debug)] +pub struct IntoIter<T> { + rx: Receiver<T>, +} + +#[unstable(feature = "mpmc_channel", issue = "126840")] +impl<'a, T> Iterator for Iter<'a, T> { + type Item = T; + + fn next(&mut self) -> Option<T> { + self.rx.recv().ok() + } +} + +#[unstable(feature = "mpmc_channel", issue = "126840")] +impl<'a, T> Iterator for TryIter<'a, T> { + type Item = T; + + fn next(&mut self) -> Option<T> { + self.rx.try_recv().ok() + } +} + +#[unstable(feature = "mpmc_channel", issue = "126840")] +impl<'a, T> IntoIterator for &'a Receiver<T> { + type Item = T; + type IntoIter = Iter<'a, T>; + + fn into_iter(self) -> Iter<'a, T> { + self.iter() + } +} + +#[unstable(feature = "mpmc_channel", issue = "126840")] +impl<T> Iterator for IntoIter<T> { + type Item = T; + fn next(&mut self) -> Option<T> { + self.rx.recv().ok() + } +} + +#[unstable(feature = "mpmc_channel", issue = "126840")] +impl<T> IntoIterator for Receiver<T> { + type Item = T; + type IntoIter = IntoIter<T>; + + fn into_iter(self) -> IntoIter<T> { + IntoIter { rx: self } + } +} + /// Receiver flavors. enum ReceiverFlavor<T> { /// Bounded channel based on a preallocated array. @@ -271,20 +862,46 @@ enum ReceiverFlavor<T> { Zero(counter::Receiver<zero::Channel<T>>), } +#[unstable(feature = "mpmc_channel", issue = "126840")] unsafe impl<T: Send> Send for Receiver<T> {} +#[unstable(feature = "mpmc_channel", issue = "126840")] unsafe impl<T: Send> Sync for Receiver<T> {} +#[unstable(feature = "mpmc_channel", issue = "126840")] impl<T> UnwindSafe for Receiver<T> {} +#[unstable(feature = "mpmc_channel", issue = "126840")] impl<T> RefUnwindSafe for Receiver<T> {} impl<T> Receiver<T> { /// Attempts to receive a message from the channel without blocking. /// - /// This method will either receive a message from the channel immediately or return an error - /// if the channel is empty. + /// This method will never block the caller in order to wait for data to + /// become available. Instead, this will always return immediately with a + /// possible option of pending data on the channel. /// /// If called on a zero-capacity channel, this method will receive a message only if there /// happens to be a send operation on the other side of the channel at the same time. + /// + /// This is useful for a flavor of "optimistic check" before deciding to + /// block on a receiver. + /// + /// Compared with [`recv`], this function has two failure cases instead of one + /// (one for disconnection, one for an empty buffer). + /// + /// [`recv`]: Self::recv + /// + /// # Examples + /// + /// ```rust + /// #![feature(mpmc_channel)] + /// + /// use std::sync::mpmc::{Receiver, channel}; + /// + /// let (_, receiver): (_, Receiver<i32>) = channel(); + /// + /// assert!(receiver.try_recv().is_err()); + /// ``` + #[unstable(feature = "mpmc_channel", issue = "126840")] pub fn try_recv(&self) -> Result<T, TryRecvError> { match &self.flavor { ReceiverFlavor::Array(chan) => chan.try_recv(), @@ -293,15 +910,64 @@ impl<T> Receiver<T> { } } - /// Blocks the current thread until a message is received or the channel is empty and - /// disconnected. + /// Attempts to wait for a value on this receiver, returning an error if the + /// corresponding channel has hung up. /// - /// If the channel is empty and not disconnected, this call will block until the receive - /// operation can proceed. If the channel is empty and becomes disconnected, this call will - /// wake up and return an error. + /// This function will always block the current thread if there is no data + /// available and it's possible for more data to be sent (at least one sender + /// still exists). Once a message is sent to the corresponding [`Sender`], + /// this receiver will wake up and return that message. /// - /// If called on a zero-capacity channel, this method will wait for a send operation to appear - /// on the other side of the channel. + /// If the corresponding [`Sender`] has disconnected, or it disconnects while + /// this call is blocking, this call will wake up and return [`Err`] to + /// indicate that no more messages can ever be received on this channel. + /// However, since channels are buffered, messages sent before the disconnect + /// will still be properly received. + /// + /// # Examples + /// + /// ``` + /// #![feature(mpmc_channel)] + /// + /// use std::sync::mpmc; + /// use std::thread; + /// + /// let (send, recv) = mpmc::channel(); + /// let handle = thread::spawn(move || { + /// send.send(1u8).unwrap(); + /// }); + /// + /// handle.join().unwrap(); + /// + /// assert_eq!(Ok(1), recv.recv()); + /// ``` + /// + /// Buffering behavior: + /// + /// ``` + /// #![feature(mpmc_channel)] + /// + /// use std::sync::mpmc; + /// use std::thread; + /// use std::sync::mpmc::RecvError; + /// + /// let (send, recv) = mpmc::channel(); + /// let handle = thread::spawn(move || { + /// send.send(1u8).unwrap(); + /// send.send(2).unwrap(); + /// send.send(3).unwrap(); + /// drop(send); + /// }); + /// + /// // wait for the thread to join so we ensure the sender is dropped + /// handle.join().unwrap(); + /// + /// assert_eq!(Ok(1), recv.recv()); + /// assert_eq!(Ok(2), recv.recv()); + /// assert_eq!(Ok(3), recv.recv()); + /// assert_eq!(Err(RecvError), recv.recv()); + /// ``` + #[unstable(feature = "mpmc_channel", issue = "126840")] pub fn recv(&self) -> Result<T, RecvError> { match &self.flavor { ReceiverFlavor::Array(chan) => chan.recv(None), @@ -311,14 +977,65 @@ impl<T> Receiver<T> { .map_err(|_| RecvError) } - /// Waits for a message to be received from the channel, but only for a limited time. + /// Attempts to wait for a value on this receiver, returning an error if the + /// corresponding channel has hung up, or if it waits more than `timeout`. + /// + /// This function will always block the current thread if there is no data + /// available and it's possible for more data to be sent (at least one sender + /// still exists). Once a message is sent to the corresponding [`Sender`], + /// this receiver will wake up and return that message. + /// + /// If the corresponding [`Sender`] has disconnected, or it disconnects while + /// this call is blocking, this call will wake up and return [`Err`] to + /// indicate that no more messages can ever be received on this channel. + /// However, since channels are buffered, messages sent before the disconnect + /// will still be properly received. + /// + /// # Examples + /// + /// Successfully receiving value before encountering timeout: /// - /// If the channel is empty and not disconnected, this call will block until the receive - /// operation can proceed or the operation times out. If the channel is empty and becomes - /// disconnected, this call will wake up and return an error. + /// ```no_run + /// #![feature(mpmc_channel)] /// - /// If called on a zero-capacity channel, this method will wait for a send operation to appear - /// on the other side of the channel. + /// use std::thread; + /// use std::time::Duration; + /// use std::sync::mpmc; + /// + /// let (send, recv) = mpmc::channel(); + /// + /// thread::spawn(move || { + /// send.send('a').unwrap(); + /// }); + /// + /// assert_eq!( + /// recv.recv_timeout(Duration::from_millis(400)), + /// Ok('a') + /// ); + /// ``` + /// + /// Receiving an error upon reaching timeout: + /// + /// ```no_run + /// #![feature(mpmc_channel)] + /// + /// use std::thread; + /// use std::time::Duration; + /// use std::sync::mpmc; + /// + /// let (send, recv) = mpmc::channel(); + /// + /// thread::spawn(move || { + /// thread::sleep(Duration::from_millis(800)); + /// send.send('a').unwrap(); + /// }); + /// + /// assert_eq!( + /// recv.recv_timeout(Duration::from_millis(400)), + /// Err(mpmc::RecvTimeoutError::Timeout) + /// ); + /// ``` + #[unstable(feature = "mpmc_channel", issue = "126840")] pub fn recv_timeout(&self, timeout: Duration) -> Result<T, RecvTimeoutError> { match Instant::now().checked_add(timeout) { Some(deadline) => self.recv_deadline(deadline), @@ -327,14 +1044,65 @@ impl<T> Receiver<T> { } } - /// Waits for a message to be received from the channel, but only for a limited time. + /// Attempts to wait for a value on this receiver, returning an error if the + /// corresponding channel has hung up, or if `deadline` is reached. + /// + /// This function will always block the current thread if there is no data + /// available and it's possible for more data to be sent. Once a message is + /// sent to the corresponding [`Sender`], then this receiver will wake up + /// and return that message. + /// + /// If the corresponding [`Sender`] has disconnected, or it disconnects while + /// this call is blocking, this call will wake up and return [`Err`] to + /// indicate that no more messages can ever be received on this channel. + /// However, since channels are buffered, messages sent before the disconnect + /// will still be properly received. + /// + /// # Examples + /// + /// Successfully receiving value before reaching deadline: + /// + /// ```no_run + /// #![feature(mpmc_channel)] + /// + /// use std::thread; + /// use std::time::{Duration, Instant}; + /// use std::sync::mpmc; /// - /// If the channel is empty and not disconnected, this call will block until the receive - /// operation can proceed or the operation times out. If the channel is empty and becomes - /// disconnected, this call will wake up and return an error. + /// let (send, recv) = mpmc::channel(); /// - /// If called on a zero-capacity channel, this method will wait for a send operation to appear - /// on the other side of the channel. + /// thread::spawn(move || { + /// send.send('a').unwrap(); + /// }); + /// + /// assert_eq!( + /// recv.recv_deadline(Instant::now() + Duration::from_millis(400)), + /// Ok('a') + /// ); + /// ``` + /// + /// Receiving an error upon reaching deadline: + /// + /// ```no_run + /// #![feature(mpmc_channel)] + /// + /// use std::thread; + /// use std::time::{Duration, Instant}; + /// use std::sync::mpmc; + /// + /// let (send, recv) = mpmc::channel(); + /// + /// thread::spawn(move || { + /// thread::sleep(Duration::from_millis(800)); + /// send.send('a').unwrap(); + /// }); + /// + /// assert_eq!( + /// recv.recv_deadline(Instant::now() + Duration::from_millis(400)), + /// Err(mpmc::RecvTimeoutError::Timeout) + /// ); + /// ``` + #[unstable(feature = "mpmc_channel", issue = "126840")] pub fn recv_deadline(&self, deadline: Instant) -> Result<T, RecvTimeoutError> { match &self.flavor { ReceiverFlavor::Array(chan) => chan.recv(Some(deadline)), @@ -342,16 +1110,77 @@ impl<T> Receiver<T> { ReceiverFlavor::Zero(chan) => chan.recv(Some(deadline)), } } + + /// Returns an iterator that will attempt to yield all pending values. + /// It will return `None` if there are no more pending values or if the + /// channel has hung up. The iterator will never [`panic!`] or block the + /// user by waiting for values. + /// + /// # Examples + /// + /// ```no_run + /// #![feature(mpmc_channel)] + /// + /// use std::sync::mpmc::channel; + /// use std::thread; + /// use std::time::Duration; + /// + /// let (sender, receiver) = channel(); + /// + /// // nothing is in the buffer yet + /// assert!(receiver.try_iter().next().is_none()); + /// + /// thread::spawn(move || { + /// thread::sleep(Duration::from_secs(1)); + /// sender.send(1).unwrap(); + /// sender.send(2).unwrap(); + /// sender.send(3).unwrap(); + /// }); + /// + /// // nothing is in the buffer yet + /// assert!(receiver.try_iter().next().is_none()); + /// + /// // block for two seconds + /// thread::sleep(Duration::from_secs(2)); + /// + /// let mut iter = receiver.try_iter(); + /// assert_eq!(iter.next(), Some(1)); + /// assert_eq!(iter.next(), Some(2)); + /// assert_eq!(iter.next(), Some(3)); + /// assert_eq!(iter.next(), None); + /// ``` + #[unstable(feature = "mpmc_channel", issue = "126840")] + pub fn try_iter(&self) -> TryIter<'_, T> { + TryIter { rx: self } + } } -// The methods below are not used by `sync::mpsc`, but -// are useful and we'll likely want to expose them -// eventually -#[allow(unused)] impl<T> Receiver<T> { /// Returns `true` if the channel is empty. /// /// Note: Zero-capacity channels are always empty. + /// + /// # Examples + /// + /// ``` + /// #![feature(mpmc_channel)] + /// + /// use std::sync::mpmc; + /// use std::thread; + /// + /// let (send, recv) = mpmc::channel(); + /// + /// assert!(recv.is_empty()); + /// + /// let handle = thread::spawn(move || { + /// send.send(1u8).unwrap(); + /// }); + /// + /// handle.join().unwrap(); + /// + /// assert!(!recv.is_empty()); + /// ``` + #[unstable(feature = "mpmc_channel", issue = "126840")] pub fn is_empty(&self) -> bool { match &self.flavor { ReceiverFlavor::Array(chan) => chan.is_empty(), @@ -363,6 +1192,28 @@ impl<T> Receiver<T> { /// Returns `true` if the channel is full. /// /// Note: Zero-capacity channels are always full. + /// + /// # Examples + /// + /// ``` + /// #![feature(mpmc_channel)] + /// + /// use std::sync::mpmc; + /// use std::thread; + /// + /// let (send, recv) = mpmc::sync_channel(1); + /// + /// assert!(!recv.is_full()); + /// + /// let handle = thread::spawn(move || { + /// send.send(1u8).unwrap(); + /// }); + /// + /// handle.join().unwrap(); + /// + /// assert!(recv.is_full()); + /// ``` + #[unstable(feature = "mpmc_channel", issue = "126840")] pub fn is_full(&self) -> bool { match &self.flavor { ReceiverFlavor::Array(chan) => chan.is_full(), @@ -372,6 +1223,28 @@ impl<T> Receiver<T> { } /// Returns the number of messages in the channel. + /// + /// # Examples + /// + /// ``` + /// #![feature(mpmc_channel)] + /// + /// use std::sync::mpmc; + /// use std::thread; + /// + /// let (send, recv) = mpmc::channel(); + /// + /// assert_eq!(recv.len(), 0); + /// + /// let handle = thread::spawn(move || { + /// send.send(1u8).unwrap(); + /// }); + /// + /// handle.join().unwrap(); + /// + /// assert_eq!(recv.len(), 1); + /// ``` + #[unstable(feature = "mpmc_channel", issue = "126840")] pub fn len(&self) -> usize { match &self.flavor { ReceiverFlavor::Array(chan) => chan.len(), @@ -381,6 +1254,28 @@ impl<T> Receiver<T> { } /// If the channel is bounded, returns its capacity. + /// + /// # Examples + /// + /// ``` + /// #![feature(mpmc_channel)] + /// + /// use std::sync::mpmc; + /// use std::thread; + /// + /// let (send, recv) = mpmc::sync_channel(3); + /// + /// assert_eq!(recv.capacity(), Some(3)); + /// + /// let handle = thread::spawn(move || { + /// send.send(1u8).unwrap(); + /// }); + /// + /// handle.join().unwrap(); + /// + /// assert_eq!(recv.capacity(), Some(3)); + /// ``` + #[unstable(feature = "mpmc_channel", issue = "126840")] pub fn capacity(&self) -> Option<usize> { match &self.flavor { ReceiverFlavor::Array(chan) => chan.capacity(), @@ -390,6 +1285,21 @@ impl<T> Receiver<T> { } /// Returns `true` if receivers belong to the same channel. + /// + /// # Examples + /// + /// ``` + /// #![feature(mpmc_channel)] + /// + /// use std::sync::mpmc; + /// + /// let (_, rx1) = mpmc::channel::<i32>(); + /// let (_, rx2) = mpmc::channel::<i32>(); + /// + /// assert!(rx1.same_channel(&rx1)); + /// assert!(!rx1.same_channel(&rx2)); + /// ``` + #[unstable(feature = "mpmc_channel", issue = "126840")] pub fn same_channel(&self, other: &Receiver<T>) -> bool { match (&self.flavor, &other.flavor) { (ReceiverFlavor::Array(a), ReceiverFlavor::Array(b)) => a == b, @@ -398,8 +1308,39 @@ impl<T> Receiver<T> { _ => false, } } + + /// Returns an iterator that will block waiting for messages, but never + /// [`panic!`]. It will return [`None`] when the channel has hung up. + /// + /// # Examples + /// + /// ```rust + /// #![feature(mpmc_channel)] + /// + /// use std::sync::mpmc::channel; + /// use std::thread; + /// + /// let (send, recv) = channel(); + /// + /// thread::spawn(move || { + /// send.send(1).unwrap(); + /// send.send(2).unwrap(); + /// send.send(3).unwrap(); + /// }); + /// + /// let mut iter = recv.iter(); + /// assert_eq!(iter.next(), Some(1)); + /// assert_eq!(iter.next(), Some(2)); + /// assert_eq!(iter.next(), Some(3)); + /// assert_eq!(iter.next(), None); + /// ``` + #[unstable(feature = "mpmc_channel", issue = "126840")] + pub fn iter(&self) -> Iter<'_, T> { + Iter { rx: self } + } } +#[unstable(feature = "mpmc_channel", issue = "126840")] impl<T> Drop for Receiver<T> { fn drop(&mut self) { unsafe { @@ -412,6 +1353,7 @@ impl<T> Drop for Receiver<T> { } } +#[unstable(feature = "mpmc_channel", issue = "126840")] impl<T> Clone for Receiver<T> { fn clone(&self) -> Self { let flavor = match &self.flavor { @@ -424,6 +1366,7 @@ impl<T> Clone for Receiver<T> { } } +#[unstable(feature = "mpmc_channel", issue = "126840")] impl<T> fmt::Debug for Receiver<T> { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.pad("Receiver { .. }") diff --git a/library/std/src/sync/mpmc/tests.rs b/library/std/src/sync/mpmc/tests.rs new file mode 100644 index 00000000000..ab14050df6c --- /dev/null +++ b/library/std/src/sync/mpmc/tests.rs @@ -0,0 +1,728 @@ +use super::*; +use crate::{env, thread}; + +pub fn stress_factor() -> usize { + match env::var("RUST_TEST_STRESS") { + Ok(val) => val.parse().unwrap(), + Err(..) => 1, + } +} + +#[test] +fn smoke() { + let (tx, rx) = channel::<i32>(); + tx.send(1).unwrap(); + assert_eq!(rx.recv().unwrap(), 1); +} + +#[test] +fn drop_full() { + let (tx, _rx) = channel::<Box<isize>>(); + tx.send(Box::new(1)).unwrap(); +} + +#[test] +fn drop_full_shared() { + let (tx, _rx) = channel::<Box<isize>>(); + drop(tx.clone()); + drop(tx.clone()); + tx.send(Box::new(1)).unwrap(); +} + +#[test] +fn smoke_shared() { + let (tx, rx) = channel::<i32>(); + tx.send(1).unwrap(); + assert_eq!(rx.recv().unwrap(), 1); + let tx = tx.clone(); + tx.send(1).unwrap(); + assert_eq!(rx.recv().unwrap(), 1); +} + +#[test] +fn smoke_threads() { + let (tx, rx) = channel::<i32>(); + let t1 = thread::spawn(move || { + for i in 0..2 { + tx.send(i).unwrap(); + } + }); + let t2 = thread::spawn(move || { + assert_eq!(rx.recv().unwrap(), 0); + assert_eq!(rx.recv().unwrap(), 1); + }); + t1.join().unwrap(); + t2.join().unwrap(); +} + +#[test] +fn smoke_port_gone() { + let (tx, rx) = channel::<i32>(); + drop(rx); + assert!(tx.send(1).is_err()); +} + +#[test] +fn smoke_shared_port_gone() { + let (tx, rx) = channel::<i32>(); + drop(rx); + assert!(tx.send(1).is_err()) +} + +#[test] +fn smoke_shared_port_gone2() { + let (tx, rx) = channel::<i32>(); + drop(rx); + let tx2 = tx.clone(); + drop(tx); + assert!(tx2.send(1).is_err()); +} + +#[test] +fn port_gone_concurrent() { + let (tx, rx) = channel::<i32>(); + let _t = thread::spawn(move || { + rx.recv().unwrap(); + }); + while tx.send(1).is_ok() {} +} + +#[test] +fn port_gone_concurrent_shared() { + let (tx, rx) = channel::<i32>(); + let tx2 = tx.clone(); + let _t = thread::spawn(move || { + rx.recv().unwrap(); + }); + while tx.send(1).is_ok() && tx2.send(1).is_ok() {} +} + +#[test] +fn smoke_chan_gone() { + let (tx, rx) = channel::<i32>(); + drop(tx); + assert!(rx.recv().is_err()); +} + +#[test] +fn smoke_chan_gone_shared() { + let (tx, rx) = channel::<()>(); + let tx2 = tx.clone(); + drop(tx); + drop(tx2); + assert!(rx.recv().is_err()); +} + +#[test] +fn chan_gone_concurrent() { + let (tx, rx) = channel::<i32>(); + let _t = thread::spawn(move || { + tx.send(1).unwrap(); + tx.send(1).unwrap(); + }); + while rx.recv().is_ok() {} +} + +#[test] +fn stress() { + let count = if cfg!(miri) { 100 } else { 10000 }; + let (tx, rx) = channel::<i32>(); + let t = thread::spawn(move || { + for _ in 0..count { + tx.send(1).unwrap(); + } + }); + for _ in 0..count { + assert_eq!(rx.recv().unwrap(), 1); + } + t.join().ok().expect("thread panicked"); +} + +#[test] +fn stress_shared() { + const AMT: u32 = if cfg!(miri) { 100 } else { 10000 }; + const NTHREADS: u32 = 8; + let (tx, rx) = channel::<i32>(); + + let t = thread::spawn(move || { + for _ in 0..AMT * NTHREADS { + assert_eq!(rx.recv().unwrap(), 1); + } + match rx.try_recv() { + Ok(..) => panic!(), + _ => {} + } + }); + + for _ in 0..NTHREADS { + let tx = tx.clone(); + thread::spawn(move || { + for _ in 0..AMT { + tx.send(1).unwrap(); + } + }); + } + drop(tx); + t.join().ok().expect("thread panicked"); +} + +#[test] +fn send_from_outside_runtime() { + let (tx1, rx1) = channel::<()>(); + let (tx2, rx2) = channel::<i32>(); + let t1 = thread::spawn(move || { + tx1.send(()).unwrap(); + for _ in 0..40 { + assert_eq!(rx2.recv().unwrap(), 1); + } + }); + rx1.recv().unwrap(); + let t2 = thread::spawn(move || { + for _ in 0..40 { + tx2.send(1).unwrap(); + } + }); + t1.join().ok().expect("thread panicked"); + t2.join().ok().expect("thread panicked"); +} + +#[test] +fn recv_from_outside_runtime() { + let (tx, rx) = channel::<i32>(); + let t = thread::spawn(move || { + for _ in 0..40 { + assert_eq!(rx.recv().unwrap(), 1); + } + }); + for _ in 0..40 { + tx.send(1).unwrap(); + } + t.join().ok().expect("thread panicked"); +} + +#[test] +fn no_runtime() { + let (tx1, rx1) = channel::<i32>(); + let (tx2, rx2) = channel::<i32>(); + let t1 = thread::spawn(move || { + assert_eq!(rx1.recv().unwrap(), 1); + tx2.send(2).unwrap(); + }); + let t2 = thread::spawn(move || { + tx1.send(1).unwrap(); + assert_eq!(rx2.recv().unwrap(), 2); + }); + t1.join().ok().expect("thread panicked"); + t2.join().ok().expect("thread panicked"); +} + +#[test] +fn oneshot_single_thread_close_port_first() { + // Simple test of closing without sending + let (_tx, rx) = channel::<i32>(); + drop(rx); +} + +#[test] +fn oneshot_single_thread_close_chan_first() { + // Simple test of closing without sending + let (tx, _rx) = channel::<i32>(); + drop(tx); +} + +#[test] +fn oneshot_single_thread_send_port_close() { + // Testing that the sender cleans up the payload if receiver is closed + let (tx, rx) = channel::<Box<i32>>(); + drop(rx); + assert!(tx.send(Box::new(0)).is_err()); +} + +#[test] +fn oneshot_single_thread_recv_chan_close() { + // Receiving on a closed chan will panic + let res = thread::spawn(move || { + let (tx, rx) = channel::<i32>(); + drop(tx); + rx.recv().unwrap(); + }) + .join(); + // What is our res? + assert!(res.is_err()); +} + +#[test] +fn oneshot_single_thread_send_then_recv() { + let (tx, rx) = channel::<Box<i32>>(); + tx.send(Box::new(10)).unwrap(); + assert!(*rx.recv().unwrap() == 10); +} + +#[test] +fn oneshot_single_thread_try_send_open() { + let (tx, rx) = channel::<i32>(); + assert!(tx.send(10).is_ok()); + assert!(rx.recv().unwrap() == 10); +} + +#[test] +fn oneshot_single_thread_try_send_closed() { + let (tx, rx) = channel::<i32>(); + drop(rx); + assert!(tx.send(10).is_err()); +} + +#[test] +fn oneshot_single_thread_try_recv_open() { + let (tx, rx) = channel::<i32>(); + tx.send(10).unwrap(); + assert!(rx.recv() == Ok(10)); +} + +#[test] +fn oneshot_single_thread_try_recv_closed() { + let (tx, rx) = channel::<i32>(); + drop(tx); + assert!(rx.recv().is_err()); +} + +#[test] +fn oneshot_single_thread_peek_data() { + let (tx, rx) = channel::<i32>(); + assert_eq!(rx.try_recv(), Err(TryRecvError::Empty)); + tx.send(10).unwrap(); + assert_eq!(rx.try_recv(), Ok(10)); +} + +#[test] +fn oneshot_single_thread_peek_close() { + let (tx, rx) = channel::<i32>(); + drop(tx); + assert_eq!(rx.try_recv(), Err(TryRecvError::Disconnected)); + assert_eq!(rx.try_recv(), Err(TryRecvError::Disconnected)); +} + +#[test] +fn oneshot_single_thread_peek_open() { + let (_tx, rx) = channel::<i32>(); + assert_eq!(rx.try_recv(), Err(TryRecvError::Empty)); +} + +#[test] +fn oneshot_multi_task_recv_then_send() { + let (tx, rx) = channel::<Box<i32>>(); + let _t = thread::spawn(move || { + assert!(*rx.recv().unwrap() == 10); + }); + + tx.send(Box::new(10)).unwrap(); +} + +#[test] +fn oneshot_multi_task_recv_then_close() { + let (tx, rx) = channel::<Box<i32>>(); + let _t = thread::spawn(move || { + drop(tx); + }); + let res = thread::spawn(move || { + assert!(*rx.recv().unwrap() == 10); + }) + .join(); + assert!(res.is_err()); +} + +#[test] +fn oneshot_multi_thread_close_stress() { + for _ in 0..stress_factor() { + let (tx, rx) = channel::<i32>(); + let _t = thread::spawn(move || { + drop(rx); + }); + drop(tx); + } +} + +#[test] +fn oneshot_multi_thread_send_close_stress() { + for _ in 0..stress_factor() { + let (tx, rx) = channel::<i32>(); + let _t = thread::spawn(move || { + drop(rx); + }); + let _ = thread::spawn(move || { + tx.send(1).unwrap(); + }) + .join(); + } +} + +#[test] +fn oneshot_multi_thread_recv_close_stress() { + for _ in 0..stress_factor() { + let (tx, rx) = channel::<i32>(); + thread::spawn(move || { + let res = thread::spawn(move || { + rx.recv().unwrap(); + }) + .join(); + assert!(res.is_err()); + }); + let _t = thread::spawn(move || { + thread::spawn(move || { + drop(tx); + }); + }); + } +} + +#[test] +fn oneshot_multi_thread_send_recv_stress() { + for _ in 0..stress_factor() { + let (tx, rx) = channel::<Box<isize>>(); + let _t = thread::spawn(move || { + tx.send(Box::new(10)).unwrap(); + }); + assert!(*rx.recv().unwrap() == 10); + } +} + +#[test] +fn stream_send_recv_stress() { + for _ in 0..stress_factor() { + let (tx, rx) = channel(); + + send(tx, 0); + recv(rx, 0); + + fn send(tx: Sender<Box<i32>>, i: i32) { + if i == 10 { + return; + } + + thread::spawn(move || { + tx.send(Box::new(i)).unwrap(); + send(tx, i + 1); + }); + } + + fn recv(rx: Receiver<Box<i32>>, i: i32) { + if i == 10 { + return; + } + + thread::spawn(move || { + assert!(*rx.recv().unwrap() == i); + recv(rx, i + 1); + }); + } + } +} + +#[test] +fn oneshot_single_thread_recv_timeout() { + let (tx, rx) = channel(); + tx.send(()).unwrap(); + assert_eq!(rx.recv_timeout(Duration::from_millis(1)), Ok(())); + assert_eq!(rx.recv_timeout(Duration::from_millis(1)), Err(RecvTimeoutError::Timeout)); + tx.send(()).unwrap(); + assert_eq!(rx.recv_timeout(Duration::from_millis(1)), Ok(())); +} + +#[test] +fn stress_recv_timeout_two_threads() { + let (tx, rx) = channel(); + let stress = stress_factor() + 100; + let timeout = Duration::from_millis(100); + + thread::spawn(move || { + for i in 0..stress { + if i % 2 == 0 { + thread::sleep(timeout * 2); + } + tx.send(1usize).unwrap(); + } + }); + + let mut recv_count = 0; + loop { + match rx.recv_timeout(timeout) { + Ok(n) => { + assert_eq!(n, 1usize); + recv_count += 1; + } + Err(RecvTimeoutError::Timeout) => continue, + Err(RecvTimeoutError::Disconnected) => break, + } + } + + assert_eq!(recv_count, stress); +} + +#[test] +fn recv_timeout_upgrade() { + let (tx, rx) = channel::<()>(); + let timeout = Duration::from_millis(1); + let _tx_clone = tx.clone(); + + let start = Instant::now(); + assert_eq!(rx.recv_timeout(timeout), Err(RecvTimeoutError::Timeout)); + assert!(Instant::now() >= start + timeout); +} + +#[test] +fn stress_recv_timeout_shared() { + let (tx, rx) = channel(); + let stress = stress_factor() + 100; + + for i in 0..stress { + let tx = tx.clone(); + thread::spawn(move || { + thread::sleep(Duration::from_millis(i as u64 * 10)); + tx.send(1usize).unwrap(); + }); + } + + drop(tx); + + let mut recv_count = 0; + loop { + match rx.recv_timeout(Duration::from_millis(10)) { + Ok(n) => { + assert_eq!(n, 1usize); + recv_count += 1; + } + Err(RecvTimeoutError::Timeout) => continue, + Err(RecvTimeoutError::Disconnected) => break, + } + } + + assert_eq!(recv_count, stress); +} + +#[test] +fn very_long_recv_timeout_wont_panic() { + let (tx, rx) = channel::<()>(); + let join_handle = thread::spawn(move || rx.recv_timeout(Duration::from_secs(u64::MAX))); + thread::sleep(Duration::from_secs(1)); + assert!(tx.send(()).is_ok()); + assert_eq!(join_handle.join().unwrap(), Ok(())); +} + +#[test] +fn recv_a_lot() { + let count = if cfg!(miri) { 1000 } else { 10000 }; + // Regression test that we don't run out of stack in scheduler context + let (tx, rx) = channel(); + for _ in 0..count { + tx.send(()).unwrap(); + } + for _ in 0..count { + rx.recv().unwrap(); + } +} + +#[test] +fn shared_recv_timeout() { + let (tx, rx) = channel(); + let total = 5; + for _ in 0..total { + let tx = tx.clone(); + thread::spawn(move || { + tx.send(()).unwrap(); + }); + } + + for _ in 0..total { + rx.recv().unwrap(); + } + + assert_eq!(rx.recv_timeout(Duration::from_millis(1)), Err(RecvTimeoutError::Timeout)); + tx.send(()).unwrap(); + assert_eq!(rx.recv_timeout(Duration::from_millis(1)), Ok(())); +} + +#[test] +fn shared_chan_stress() { + let (tx, rx) = channel(); + let total = stress_factor() + 100; + for _ in 0..total { + let tx = tx.clone(); + thread::spawn(move || { + tx.send(()).unwrap(); + }); + } + + for _ in 0..total { + rx.recv().unwrap(); + } +} + +#[test] +fn test_nested_recv_iter() { + let (tx, rx) = channel::<i32>(); + let (total_tx, total_rx) = channel::<i32>(); + + let _t = thread::spawn(move || { + let mut acc = 0; + for x in rx.iter() { + acc += x; + } + total_tx.send(acc).unwrap(); + }); + + tx.send(3).unwrap(); + tx.send(1).unwrap(); + tx.send(2).unwrap(); + drop(tx); + assert_eq!(total_rx.recv().unwrap(), 6); +} + +#[test] +fn test_recv_iter_break() { + let (tx, rx) = channel::<i32>(); + let (count_tx, count_rx) = channel(); + + let _t = thread::spawn(move || { + let mut count = 0; + for x in rx.iter() { + if count >= 3 { + break; + } else { + count += x; + } + } + count_tx.send(count).unwrap(); + }); + + tx.send(2).unwrap(); + tx.send(2).unwrap(); + tx.send(2).unwrap(); + let _ = tx.send(2); + drop(tx); + assert_eq!(count_rx.recv().unwrap(), 4); +} + +#[test] +fn test_recv_try_iter() { + let (request_tx, request_rx) = channel(); + let (response_tx, response_rx) = channel(); + + // Request `x`s until we have `6`. + let t = thread::spawn(move || { + let mut count = 0; + loop { + for x in response_rx.try_iter() { + count += x; + if count == 6 { + return count; + } + } + request_tx.send(()).unwrap(); + } + }); + + for _ in request_rx.iter() { + if response_tx.send(2).is_err() { + break; + } + } + + assert_eq!(t.join().unwrap(), 6); +} + +#[test] +fn test_recv_into_iter_owned() { + let mut iter = { + let (tx, rx) = channel::<i32>(); + tx.send(1).unwrap(); + tx.send(2).unwrap(); + + rx.into_iter() + }; + assert_eq!(iter.next().unwrap(), 1); + assert_eq!(iter.next().unwrap(), 2); + assert_eq!(iter.next().is_none(), true); +} + +#[test] +fn test_recv_into_iter_borrowed() { + let (tx, rx) = channel::<i32>(); + tx.send(1).unwrap(); + tx.send(2).unwrap(); + drop(tx); + let mut iter = (&rx).into_iter(); + assert_eq!(iter.next().unwrap(), 1); + assert_eq!(iter.next().unwrap(), 2); + assert_eq!(iter.next().is_none(), true); +} + +#[test] +fn try_recv_states() { + let (tx1, rx1) = channel::<i32>(); + let (tx2, rx2) = channel::<()>(); + let (tx3, rx3) = channel::<()>(); + let _t = thread::spawn(move || { + rx2.recv().unwrap(); + tx1.send(1).unwrap(); + tx3.send(()).unwrap(); + rx2.recv().unwrap(); + drop(tx1); + tx3.send(()).unwrap(); + }); + + assert_eq!(rx1.try_recv(), Err(TryRecvError::Empty)); + tx2.send(()).unwrap(); + rx3.recv().unwrap(); + assert_eq!(rx1.try_recv(), Ok(1)); + assert_eq!(rx1.try_recv(), Err(TryRecvError::Empty)); + tx2.send(()).unwrap(); + rx3.recv().unwrap(); + assert_eq!(rx1.try_recv(), Err(TryRecvError::Disconnected)); +} + +// This bug used to end up in a livelock inside of the Receiver destructor +// because the internal state of the Shared packet was corrupted +#[test] +fn destroy_upgraded_shared_port_when_sender_still_active() { + let (tx, rx) = channel(); + let (tx2, rx2) = channel(); + let _t = thread::spawn(move || { + rx.recv().unwrap(); // wait on a oneshot + drop(rx); // destroy a shared + tx2.send(()).unwrap(); + }); + // make sure the other thread has gone to sleep + for _ in 0..5000 { + thread::yield_now(); + } + + // upgrade to a shared chan and send a message + let t = tx.clone(); + drop(tx); + t.send(()).unwrap(); + + // wait for the child thread to exit before we exit + rx2.recv().unwrap(); +} + +#[test] +fn issue_32114() { + let (tx, _) = channel(); + let _ = tx.send(123); + assert_eq!(tx.send(123), Err(SendError(123))); +} + +#[test] +fn issue_39364() { + let (tx, rx) = channel::<()>(); + let t = thread::spawn(move || { + thread::sleep(Duration::from_millis(300)); + let _ = tx.clone(); + // Don't drop; hand back to caller. + tx + }); + + let _ = rx.recv_timeout(Duration::from_millis(500)); + let _tx = t.join().unwrap(); // delay dropping until end of test + let _ = rx.recv_timeout(Duration::from_millis(500)); +} diff --git a/library/std/src/sync/mpmc/waker.rs b/library/std/src/sync/mpmc/waker.rs index fb877887f9c..1895466f95d 100644 --- a/library/std/src/sync/mpmc/waker.rs +++ b/library/std/src/sync/mpmc/waker.rs @@ -3,8 +3,8 @@ use super::context::Context; use super::select::{Operation, Selected}; use crate::ptr; -use crate::sync::atomic::{AtomicBool, Ordering}; use crate::sync::Mutex; +use crate::sync::atomic::{AtomicBool, Ordering}; /// Represents a thread blocked on a specific channel operation. pub(crate) struct Entry { diff --git a/library/std/src/sync/mpmc/zero.rs b/library/std/src/sync/mpmc/zero.rs index 2b82eeda3d5..446881291e6 100644 --- a/library/std/src/sync/mpmc/zero.rs +++ b/library/std/src/sync/mpmc/zero.rs @@ -9,8 +9,8 @@ use super::utils::Backoff; use super::waker::Waker; use crate::cell::UnsafeCell; use crate::marker::PhantomData; -use crate::sync::atomic::{AtomicBool, Ordering}; use crate::sync::Mutex; +use crate::sync::atomic::{AtomicBool, Ordering}; use crate::time::Instant; use crate::{fmt, ptr}; @@ -185,11 +185,7 @@ impl<T> Channel<T> { // Prepare for blocking until a receiver wakes us up. let oper = Operation::hook(token); let mut packet = Packet::<T>::message_on_stack(msg); - inner.senders.register_with_packet( - oper, - core::ptr::addr_of_mut!(packet) as *mut (), - cx, - ); + inner.senders.register_with_packet(oper, (&raw mut packet) as *mut (), cx); inner.receivers.notify(); drop(inner); @@ -256,11 +252,7 @@ impl<T> Channel<T> { // Prepare for blocking until a sender wakes us up. let oper = Operation::hook(token); let mut packet = Packet::<T>::empty_on_stack(); - inner.receivers.register_with_packet( - oper, - core::ptr::addr_of_mut!(packet) as *mut (), - cx, - ); + inner.receivers.register_with_packet(oper, (&raw mut packet) as *mut (), cx); inner.senders.notify(); drop(inner); diff --git a/library/std/src/sync/mpsc/mod.rs b/library/std/src/sync/mpsc/mod.rs index 26d5b9515a2..83a93a06369 100644 --- a/library/std/src/sync/mpsc/mod.rs +++ b/library/std/src/sync/mpsc/mod.rs @@ -137,10 +137,10 @@ #![stable(feature = "rust1", since = "1.0.0")] -#[cfg(all(test, not(target_os = "emscripten")))] +#[cfg(all(test, not(any(target_os = "emscripten", target_os = "wasi"))))] mod tests; -#[cfg(all(test, not(target_os = "emscripten")))] +#[cfg(all(test, not(any(target_os = "emscripten", target_os = "wasi"))))] mod sync_tests; // MPSC channels are built as a wrapper around MPMC channels, which diff --git a/library/std/src/sync/mutex.rs b/library/std/src/sync/mutex.rs index d417034f5af..fe2aca031a2 100644 --- a/library/std/src/sync/mutex.rs +++ b/library/std/src/sync/mutex.rs @@ -1,4 +1,4 @@ -#[cfg(all(test, not(target_os = "emscripten")))] +#[cfg(all(test, not(any(target_os = "emscripten", target_os = "wasi"))))] mod tests; use crate::cell::UnsafeCell; @@ -7,7 +7,7 @@ use crate::marker::PhantomData; use crate::mem::ManuallyDrop; use crate::ops::{Deref, DerefMut}; use crate::ptr::NonNull; -use crate::sync::{poison, LockResult, TryLockError, TryLockResult}; +use crate::sync::{LockResult, TryLockError, TryLockResult, poison}; use crate::sys::sync as sys; /// A mutual exclusion primitive useful for protecting shared data diff --git a/library/std/src/sync/once.rs b/library/std/src/sync/once.rs index 5a1cd7d0b8b..993df9314fc 100644 --- a/library/std/src/sync/once.rs +++ b/library/std/src/sync/once.rs @@ -3,7 +3,7 @@ //! This primitive is meant to be used to run one-time initialization. An //! example use case would be for initializing an FFI library. -#[cfg(all(test, not(target_os = "emscripten")))] +#[cfg(all(test, not(any(target_os = "emscripten", target_os = "wasi"))))] mod tests; use crate::fmt; diff --git a/library/std/src/sync/once_lock/tests.rs b/library/std/src/sync/once_lock/tests.rs index 176830c6748..5113d436c3c 100644 --- a/library/std/src/sync/once_lock/tests.rs +++ b/library/std/src/sync/once_lock/tests.rs @@ -1,7 +1,7 @@ +use crate::sync::OnceLock; use crate::sync::atomic::AtomicUsize; use crate::sync::atomic::Ordering::SeqCst; use crate::sync::mpsc::channel; -use crate::sync::OnceLock; use crate::{panic, thread}; fn spawn_and_wait<R: Send + 'static>(f: impl FnOnce() -> R + Send + 'static) -> R { @@ -9,7 +9,7 @@ fn spawn_and_wait<R: Send + 'static>(f: impl FnOnce() -> R + Send + 'static) -> } #[test] -#[cfg_attr(target_os = "emscripten", ignore)] +#[cfg_attr(any(target_os = "emscripten", target_os = "wasi"), ignore)] // no threads fn sync_once_cell() { static ONCE_CELL: OnceLock<i32> = OnceLock::new(); @@ -43,7 +43,7 @@ fn sync_once_cell_get_unchecked() { } #[test] -#[cfg_attr(target_os = "emscripten", ignore)] +#[cfg_attr(any(target_os = "emscripten", target_os = "wasi"), ignore)] // no threads fn sync_once_cell_drop() { static DROP_CNT: AtomicUsize = AtomicUsize::new(0); struct Dropper; @@ -81,6 +81,7 @@ fn clone() { } #[test] +#[cfg_attr(not(panic = "unwind"), ignore = "test requires unwinding support")] fn get_or_try_init() { let cell: OnceLock<String> = OnceLock::new(); assert!(cell.get().is_none()); @@ -154,7 +155,7 @@ fn eval_once_macro() { } #[test] -#[cfg_attr(target_os = "emscripten", ignore)] +#[cfg_attr(any(target_os = "emscripten", target_os = "wasi"), ignore)] // no threads fn sync_once_cell_does_not_leak_partially_constructed_boxes() { static ONCE_CELL: OnceLock<String> = OnceLock::new(); diff --git a/library/std/src/sync/reentrant_lock.rs b/library/std/src/sync/reentrant_lock.rs index 0b23681e907..0140e0d2129 100644 --- a/library/std/src/sync/reentrant_lock.rs +++ b/library/std/src/sync/reentrant_lock.rs @@ -1,4 +1,4 @@ -#[cfg(all(test, not(target_os = "emscripten")))] +#[cfg(all(test, not(any(target_os = "emscripten", target_os = "wasi"))))] mod tests; use cfg_if::cfg_if; @@ -8,7 +8,7 @@ use crate::fmt; use crate::ops::Deref; use crate::panic::{RefUnwindSafe, UnwindSafe}; use crate::sys::sync as sys; -use crate::thread::{current_id, ThreadId}; +use crate::thread::{ThreadId, current_id}; /// A re-entrant mutual exclusion lock /// diff --git a/library/std/src/sync/rwlock.rs b/library/std/src/sync/rwlock.rs index d995a16e056..da2da6f9dfc 100644 --- a/library/std/src/sync/rwlock.rs +++ b/library/std/src/sync/rwlock.rs @@ -1,4 +1,4 @@ -#[cfg(all(test, not(target_os = "emscripten")))] +#[cfg(all(test, not(any(target_os = "emscripten", target_os = "wasi"))))] mod tests; use crate::cell::UnsafeCell; @@ -7,7 +7,7 @@ use crate::marker::PhantomData; use crate::mem::ManuallyDrop; use crate::ops::{Deref, DerefMut}; use crate::ptr::NonNull; -use crate::sync::{poison, LockResult, TryLockError, TryLockResult}; +use crate::sync::{LockResult, TryLockError, TryLockResult, poison}; use crate::sys::sync as sys; /// A reader-writer lock |