From 698f8c2f01ea549d77d7dc3338a12e04c11057b9 Mon Sep 17 00:00:00 2001 From: Daniel Baumann Date: Wed, 17 Apr 2024 14:02:58 +0200 Subject: Adding upstream version 1.64.0+dfsg1. Signed-off-by: Daniel Baumann --- library/std/src/sync/mpsc/mod.rs | 1669 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 1669 insertions(+) create mode 100644 library/std/src/sync/mpsc/mod.rs (limited to 'library/std/src/sync/mpsc/mod.rs') diff --git a/library/std/src/sync/mpsc/mod.rs b/library/std/src/sync/mpsc/mod.rs new file mode 100644 index 000000000..e85a87239 --- /dev/null +++ b/library/std/src/sync/mpsc/mod.rs @@ -0,0 +1,1669 @@ +//! Multi-producer, single-consumer FIFO queue communication primitives. +//! +//! This module provides message-based communication over channels, concretely +//! defined among three types: +//! +//! * [`Sender`] +//! * [`SyncSender`] +//! * [`Receiver`] +//! +//! A [`Sender`] or [`SyncSender`] is used to send data to a [`Receiver`]. Both +//! senders are clone-able (multi-producer) such that many threads can send +//! simultaneously to one receiver (single-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: +//! +//! ``` +//! use std::thread; +//! use std::sync::mpsc::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: +//! +//! ``` +//! use std::thread; +//! use std::sync::mpsc::channel; +//! +//! // 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(); +//! thread::spawn(move|| { +//! tx.send(i).unwrap(); +//! }); +//! } +//! +//! for _ in 0..10 { +//! let j = rx.recv().unwrap(); +//! assert!(0 <= j && j < 10); +//! } +//! ``` +//! +//! Propagating panics: +//! +//! ``` +//! use std::sync::mpsc::channel; +//! +//! // The call to recv() will return an error because the channel has already +//! // hung up (or been deallocated) +//! let (tx, rx) = channel::(); +//! drop(tx); +//! assert!(rx.recv().is_err()); +//! ``` +//! +//! Synchronous channels: +//! +//! ``` +//! use std::thread; +//! use std::sync::mpsc::sync_channel; +//! +//! let (tx, rx) = sync_channel::(0); +//! thread::spawn(move|| { +//! // This will wait for the parent thread to start receiving +//! tx.send(53).unwrap(); +//! }); +//! rx.recv().unwrap(); +//! ``` +//! +//! Unbounded receive loop: +//! +//! ``` +//! use std::sync::mpsc::sync_channel; +//! use std::thread; +//! +//! let (tx, rx) = sync_channel(3); +//! +//! for _ in 0..3 { +//! // It would be the same without thread and clone here +//! // since there will still be one `tx` left. +//! let tx = tx.clone(); +//! // cloned tx dropped within thread +//! thread::spawn(move || tx.send("ok").unwrap()); +//! } +//! +//! // Drop the last sender to stop `rx` waiting for message. +//! // The program will not complete if we comment this out. +//! // **All** `tx` needs to be dropped for `rx` to have `Err`. +//! drop(tx); +//! +//! // Unbounded receiver waiting for all senders to complete. +//! while let Ok(msg) = rx.recv() { +//! println!("{msg}"); +//! } +//! +//! println!("completed"); +//! ``` + +#![stable(feature = "rust1", since = "1.0.0")] + +#[cfg(all(test, not(target_os = "emscripten")))] +mod tests; + +#[cfg(all(test, not(target_os = "emscripten")))] +mod sync_tests; + +// A description of how Rust's channel implementation works +// +// Channels are supposed to be the basic building block for all other +// concurrent primitives that are used in Rust. As a result, the channel type +// needs to be highly optimized, flexible, and broad enough for use everywhere. +// +// The choice of implementation of all channels is to be built on lock-free data +// structures. The channels themselves are then consequently also lock-free data +// structures. As always with lock-free code, this is a very "here be dragons" +// territory, especially because I'm unaware of any academic papers that have +// gone into great length about channels of these flavors. +// +// ## Flavors of channels +// +// From the perspective of a consumer of this library, there is only one flavor +// of channel. This channel can be used as a stream and cloned to allow multiple +// senders. Under the hood, however, there are actually three flavors of +// channels in play. +// +// * Flavor::Oneshots - these channels are highly optimized for the one-send use +// case. They contain as few atomics as possible and +// involve one and exactly one allocation. +// * Streams - these channels are optimized for the non-shared use case. They +// use a different concurrent queue that is more tailored for this +// use case. The initial allocation of this flavor of channel is not +// optimized. +// * Shared - this is the most general form of channel that this module offers, +// a channel with multiple senders. This type is as optimized as it +// can be, but the previous two types mentioned are much faster for +// their use-cases. +// +// ## Concurrent queues +// +// The basic idea of Rust's Sender/Receiver types is that send() never blocks, +// but recv() obviously blocks. This means that under the hood there must be +// some shared and concurrent queue holding all of the actual data. +// +// With two flavors of channels, two flavors of queues are also used. We have +// chosen to use queues from a well-known author that are abbreviated as SPSC +// and MPSC (single producer, single consumer and multiple producer, single +// consumer). SPSC queues are used for streams while MPSC queues are used for +// shared channels. +// +// ### SPSC optimizations +// +// The SPSC queue found online is essentially a linked list of nodes where one +// half of the nodes are the "queue of data" and the other half of nodes are a +// cache of unused nodes. The unused nodes are used such that an allocation is +// not required on every push() and a free doesn't need to happen on every +// pop(). +// +// As found online, however, the cache of nodes is of an infinite size. This +// means that if a channel at one point in its life had 50k items in the queue, +// then the queue will always have the capacity for 50k items. I believed that +// this was an unnecessary limitation of the implementation, so I have altered +// the queue to optionally have a bound on the cache size. +// +// By default, streams will have an unbounded SPSC queue with a small-ish cache +// size. The hope is that the cache is still large enough to have very fast +// send() operations while not too large such that millions of channels can +// coexist at once. +// +// ### MPSC optimizations +// +// Right now the MPSC queue has not been optimized. Like the SPSC queue, it uses +// a linked list under the hood to earn its unboundedness, but I have not put +// forth much effort into having a cache of nodes similar to the SPSC queue. +// +// For now, I believe that this is "ok" because shared channels are not the most +// common type, but soon we may wish to revisit this queue choice and determine +// another candidate for backend storage of shared channels. +// +// ## Overview of the Implementation +// +// Now that there's a little background on the concurrent queues used, it's +// worth going into much more detail about the channels themselves. The basic +// pseudocode for a send/recv are: +// +// +// send(t) recv() +// queue.push(t) return if queue.pop() +// if increment() == -1 deschedule { +// wakeup() if decrement() > 0 +// cancel_deschedule() +// } +// queue.pop() +// +// As mentioned before, there are no locks in this implementation, only atomic +// instructions are used. +// +// ### The internal atomic counter +// +// Every channel has a shared counter with each half to keep track of the size +// of the queue. This counter is used to abort descheduling by the receiver and +// to know when to wake up on the sending side. +// +// As seen in the pseudocode, senders will increment this count and receivers +// will decrement the count. The theory behind this is that if a sender sees a +// -1 count, it will wake up the receiver, and if the receiver sees a 1+ count, +// then it doesn't need to block. +// +// The recv() method has a beginning call to pop(), and if successful, it needs +// to decrement the count. It is a crucial implementation detail that this +// decrement does *not* happen to the shared counter. If this were the case, +// then it would be possible for the counter to be very negative when there were +// no receivers waiting, in which case the senders would have to determine when +// it was actually appropriate to wake up a receiver. +// +// Instead, the "steal count" is kept track of separately (not atomically +// because it's only used by receivers), and then the decrement() call when +// descheduling will lump in all of the recent steals into one large decrement. +// +// The implication of this is that if a sender sees a -1 count, then there's +// guaranteed to be a waiter waiting! +// +// ## Native Implementation +// +// A major goal of these channels is to work seamlessly on and off the runtime. +// All of the previous race conditions have been worded in terms of +// scheduler-isms (which is obviously not available without the runtime). +// +// For now, native usage of channels (off the runtime) will fall back onto +// mutexes/cond vars for descheduling/atomic decisions. The no-contention path +// is still entirely lock-free, the "deschedule" blocks above are surrounded by +// a mutex and the "wakeup" blocks involve grabbing a mutex and signaling on a +// condition variable. +// +// ## Select +// +// Being able to support selection over channels has greatly influenced this +// design, and not only does selection need to work inside the runtime, but also +// outside the runtime. +// +// The implementation is fairly straightforward. The goal of select() is not to +// return some data, but only to return which channel can receive data without +// blocking. The implementation is essentially the entire blocking procedure +// followed by an increment as soon as its woken up. The cancellation procedure +// involves an increment and swapping out of to_wake to acquire ownership of the +// thread to unblock. +// +// Sadly this current implementation requires multiple allocations, so I have +// seen the throughput of select() be much worse than it should be. I do not +// believe that there is anything fundamental that needs to change about these +// channels, however, in order to support a more efficient select(). +// +// FIXME: Select is now removed, so these factors are ready to be cleaned up! +// +// # Conclusion +// +// And now that you've seen all the races that I found and attempted to fix, +// here's the code for you to find some more! + +use crate::cell::UnsafeCell; +use crate::error; +use crate::fmt; +use crate::mem; +use crate::sync::Arc; +use crate::time::{Duration, Instant}; + +mod blocking; +mod mpsc_queue; +mod oneshot; +mod shared; +mod spsc_queue; +mod stream; +mod sync; + +mod cache_aligned; + +/// The receiving half of Rust's [`channel`] (or [`sync_channel`]) type. +/// This half can only be owned by one thread. +/// +/// Messages sent to the channel can be retrieved using [`recv`]. +/// +/// [`recv`]: Receiver::recv +/// +/// # Examples +/// +/// ```rust +/// use std::sync::mpsc::channel; +/// use std::thread; +/// use std::time::Duration; +/// +/// let (send, recv) = channel(); +/// +/// 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(); +/// }); +/// +/// println!("{}", recv.recv().unwrap()); // Received immediately +/// println!("Waiting..."); +/// println!("{}", recv.recv().unwrap()); // Received after 2 seconds +/// ``` +#[stable(feature = "rust1", since = "1.0.0")] +#[cfg_attr(not(test), rustc_diagnostic_item = "Receiver")] +pub struct Receiver { + inner: UnsafeCell>, +} + +// The receiver port can be sent from place to place, so long as it +// is not used to receive non-sendable things. +#[stable(feature = "rust1", since = "1.0.0")] +unsafe impl Send for Receiver {} + +#[stable(feature = "rust1", since = "1.0.0")] +impl !Sync for Receiver {} + +/// 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 +/// use std::sync::mpsc::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}"); +/// } +/// ``` +#[stable(feature = "rust1", since = "1.0.0")] +#[derive(Debug)] +pub struct Iter<'a, T: 'a> { + rx: &'a Receiver, +} + +/// 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 +/// use std::sync::mpsc::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}"); +/// } +/// ``` +#[stable(feature = "receiver_try_iter", since = "1.15.0")] +#[derive(Debug)] +pub struct TryIter<'a, T: 'a> { + rx: &'a Receiver, +} + +/// 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 +/// use std::sync::mpsc::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}"); +/// } +/// ``` +#[stable(feature = "receiver_into_iter", since = "1.1.0")] +#[derive(Debug)] +pub struct IntoIter { + rx: Receiver, +} + +/// The sending-half of Rust's asynchronous [`channel`] type. This half can only be +/// owned by one thread, but it can be cloned to send to other threads. +/// +/// Messages can be sent through this channel with [`send`]. +/// +/// Note: all senders (the original and the clones) need to be dropped for the receiver +/// to stop blocking to receive messages with [`Receiver::recv`]. +/// +/// [`send`]: Sender::send +/// +/// # Examples +/// +/// ```rust +/// use std::sync::mpsc::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); +/// ``` +#[stable(feature = "rust1", since = "1.0.0")] +pub struct Sender { + inner: UnsafeCell>, +} + +// The send port can be sent from place to place, so long as it +// is not used to send non-sendable things. +#[stable(feature = "rust1", since = "1.0.0")] +unsafe impl Send for Sender {} + +#[stable(feature = "rust1", since = "1.0.0")] +impl !Sync for Sender {} + +/// The sending-half of Rust's synchronous [`sync_channel`] type. +/// +/// Messages can be sent through this channel with [`send`] or [`try_send`]. +/// +/// [`send`] will block if there is no space in the internal buffer. +/// +/// [`send`]: SyncSender::send +/// [`try_send`]: SyncSender::try_send +/// +/// # Examples +/// +/// ```rust +/// use std::sync::mpsc::sync_channel; +/// use std::thread; +/// +/// // Create a sync_channel with buffer size 2 +/// let (sync_sender, receiver) = sync_channel(2); +/// let sync_sender2 = sync_sender.clone(); +/// +/// // First thread owns sync_sender +/// thread::spawn(move || { +/// sync_sender.send(1).unwrap(); +/// sync_sender.send(2).unwrap(); +/// }); +/// +/// // Second thread owns sync_sender2 +/// thread::spawn(move || { +/// sync_sender2.send(3).unwrap(); +/// // thread will now block since the buffer is full +/// println!("Thread unblocked!"); +/// }); +/// +/// let mut msg; +/// +/// msg = receiver.recv().unwrap(); +/// println!("message {msg} received"); +/// +/// // "Thread unblocked!" will be printed now +/// +/// msg = receiver.recv().unwrap(); +/// println!("message {msg} received"); +/// +/// msg = receiver.recv().unwrap(); +/// +/// println!("message {msg} received"); +/// ``` +#[stable(feature = "rust1", since = "1.0.0")] +pub struct SyncSender { + inner: Arc>, +} + +#[stable(feature = "rust1", since = "1.0.0")] +unsafe impl Send for SyncSender {} + +/// An error returned from the [`Sender::send`] or [`SyncSender::send`] +/// function on **channel**s. +/// +/// A **send** operation can only fail if the receiving end of a channel is +/// disconnected, implying that the data could never be received. The error +/// contains the data being sent as a payload so it can be recovered. +#[stable(feature = "rust1", since = "1.0.0")] +#[derive(PartialEq, Eq, Clone, Copy)] +pub struct SendError(#[stable(feature = "rust1", since = "1.0.0")] pub T); + +/// An error returned from the [`recv`] function on a [`Receiver`]. +/// +/// The [`recv`] operation can only fail if the sending half of a +/// [`channel`] (or [`sync_channel`]) is disconnected, implying that no further +/// messages will ever be received. +/// +/// [`recv`]: Receiver::recv +#[derive(PartialEq, Eq, Clone, Copy, Debug)] +#[stable(feature = "rust1", since = "1.0.0")] +pub struct RecvError; + +/// This enumeration is the list of the possible reasons that [`try_recv`] could +/// not return data when called. This can occur with both a [`channel`] and +/// a [`sync_channel`]. +/// +/// [`try_recv`]: Receiver::try_recv +#[derive(PartialEq, Eq, Clone, Copy, Debug)] +#[stable(feature = "rust1", since = "1.0.0")] +pub enum TryRecvError { + /// This **channel** is currently empty, but the **Sender**(s) have not yet + /// disconnected, so data may yet become available. + #[stable(feature = "rust1", since = "1.0.0")] + Empty, + + /// The **channel**'s sending half has become disconnected, and there will + /// never be any more data received on it. + #[stable(feature = "rust1", since = "1.0.0")] + Disconnected, +} + +/// This enumeration is the list of possible errors that made [`recv_timeout`] +/// unable to return data when called. This can occur with both a [`channel`] and +/// a [`sync_channel`]. +/// +/// [`recv_timeout`]: Receiver::recv_timeout +#[derive(PartialEq, Eq, Clone, Copy, Debug)] +#[stable(feature = "mpsc_recv_timeout", since = "1.12.0")] +pub enum RecvTimeoutError { + /// This **channel** is currently empty, but the **Sender**(s) have not yet + /// disconnected, so data may yet become available. + #[stable(feature = "mpsc_recv_timeout", since = "1.12.0")] + Timeout, + /// The **channel**'s sending half has become disconnected, and there will + /// never be any more data received on it. + #[stable(feature = "mpsc_recv_timeout", since = "1.12.0")] + Disconnected, +} + +/// This enumeration is the list of the possible error outcomes for the +/// [`try_send`] method. +/// +/// [`try_send`]: SyncSender::try_send +#[stable(feature = "rust1", since = "1.0.0")] +#[derive(PartialEq, Eq, Clone, Copy)] +pub enum TrySendError { + /// The data could not be sent on the [`sync_channel`] because it would require that + /// the callee block to send the data. + /// + /// If this is a buffered channel, then the buffer is full at this time. If + /// this is not a buffered channel, then there is no [`Receiver`] available to + /// acquire the data. + #[stable(feature = "rust1", since = "1.0.0")] + Full(#[stable(feature = "rust1", since = "1.0.0")] T), + + /// This [`sync_channel`]'s receiving half has disconnected, so the data could not be + /// sent. The data is returned back to the callee in this case. + #[stable(feature = "rust1", since = "1.0.0")] + Disconnected(#[stable(feature = "rust1", since = "1.0.0")] T), +} + +enum Flavor { + Oneshot(Arc>), + Stream(Arc>), + Shared(Arc>), + Sync(Arc>), +} + +#[doc(hidden)] +trait UnsafeFlavor { + fn inner_unsafe(&self) -> &UnsafeCell>; + unsafe fn inner_mut(&self) -> &mut Flavor { + &mut *self.inner_unsafe().get() + } + unsafe fn inner(&self) -> &Flavor { + &*self.inner_unsafe().get() + } +} +impl UnsafeFlavor for Sender { + fn inner_unsafe(&self) -> &UnsafeCell> { + &self.inner + } +} +impl UnsafeFlavor for Receiver { + fn inner_unsafe(&self) -> &UnsafeCell> { + &self.inner + } +} + +/// 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). +/// +/// The [`Sender`] can be cloned to [`send`] to the same channel multiple times, but +/// only one [`Receiver`] is supported. +/// +/// 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::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] +#[stable(feature = "rust1", since = "1.0.0")] +pub fn channel() -> (Sender, Receiver) { + let a = Arc::new(oneshot::Packet::new()); + (Sender::new(Flavor::Oneshot(a.clone())), Receiver::new(Flavor::Oneshot(a))) +} + +/// Creates a new synchronous, bounded channel. +/// All data sent on the [`SyncSender`] 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 [`SyncSender`] can be cloned to [`send`] to the same channel multiple +/// times, but only one [`Receiver`] is supported. +/// +/// Like asynchronous channels, if the [`Receiver`] is disconnected while trying +/// to [`send`] with the [`SyncSender`], the [`send`] method will return a +/// [`SendError`]. Similarly, If the [`SyncSender`] is disconnected while trying +/// to [`recv`], the [`recv`] method will return a [`RecvError`]. +/// +/// [`send`]: SyncSender::send +/// [`recv`]: Receiver::recv +/// +/// # Examples +/// +/// ``` +/// use std::sync::mpsc::sync_channel; +/// use std::thread; +/// +/// let (sender, receiver) = sync_channel(1); +/// +/// // this returns immediately +/// sender.send(1).unwrap(); +/// +/// 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] +#[stable(feature = "rust1", since = "1.0.0")] +pub fn sync_channel(bound: usize) -> (SyncSender, Receiver) { + let a = Arc::new(sync::Packet::new(bound)); + (SyncSender::new(a.clone()), Receiver::new(Flavor::Sync(a))) +} + +//////////////////////////////////////////////////////////////////////////////// +// Sender +//////////////////////////////////////////////////////////////////////////////// + +impl Sender { + fn new(inner: Flavor) -> Sender { + Sender { inner: UnsafeCell::new(inner) } + } + + /// Attempts to send a value on this channel, returning it back if it could + /// not be sent. + /// + /// 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`]. + /// + /// This method will never block the current thread. + /// + /// # Examples + /// + /// ``` + /// use std::sync::mpsc::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_eq!(tx.send(1).unwrap_err().0, 1); + /// ``` + #[stable(feature = "rust1", since = "1.0.0")] + pub fn send(&self, t: T) -> Result<(), SendError> { + let (new_inner, ret) = match *unsafe { self.inner() } { + Flavor::Oneshot(ref p) => { + if !p.sent() { + return p.send(t).map_err(SendError); + } else { + let a = Arc::new(stream::Packet::new()); + let rx = Receiver::new(Flavor::Stream(a.clone())); + match p.upgrade(rx) { + oneshot::UpSuccess => { + let ret = a.send(t); + (a, ret) + } + oneshot::UpDisconnected => (a, Err(t)), + oneshot::UpWoke(token) => { + // This send cannot panic because the thread is + // asleep (we're looking at it), so the receiver + // can't go away. + a.send(t).ok().unwrap(); + token.signal(); + (a, Ok(())) + } + } + } + } + Flavor::Stream(ref p) => return p.send(t).map_err(SendError), + Flavor::Shared(ref p) => return p.send(t).map_err(SendError), + Flavor::Sync(..) => unreachable!(), + }; + + unsafe { + let tmp = Sender::new(Flavor::Stream(new_inner)); + mem::swap(self.inner_mut(), tmp.inner_mut()); + } + ret.map_err(SendError) + } +} + +#[stable(feature = "rust1", since = "1.0.0")] +impl Clone for Sender { + /// Clone a sender to send to other threads. + /// + /// Note, be aware of the lifetime of the sender because all senders + /// (including the original) need to be dropped in order for + /// [`Receiver::recv`] to stop blocking. + fn clone(&self) -> Sender { + let packet = match *unsafe { self.inner() } { + Flavor::Oneshot(ref p) => { + let a = Arc::new(shared::Packet::new()); + { + let guard = a.postinit_lock(); + let rx = Receiver::new(Flavor::Shared(a.clone())); + let sleeper = match p.upgrade(rx) { + oneshot::UpSuccess | oneshot::UpDisconnected => None, + oneshot::UpWoke(task) => Some(task), + }; + a.inherit_blocker(sleeper, guard); + } + a + } + Flavor::Stream(ref p) => { + let a = Arc::new(shared::Packet::new()); + { + let guard = a.postinit_lock(); + let rx = Receiver::new(Flavor::Shared(a.clone())); + let sleeper = match p.upgrade(rx) { + stream::UpSuccess | stream::UpDisconnected => None, + stream::UpWoke(task) => Some(task), + }; + a.inherit_blocker(sleeper, guard); + } + a + } + Flavor::Shared(ref p) => { + p.clone_chan(); + return Sender::new(Flavor::Shared(p.clone())); + } + Flavor::Sync(..) => unreachable!(), + }; + + unsafe { + let tmp = Sender::new(Flavor::Shared(packet.clone())); + mem::swap(self.inner_mut(), tmp.inner_mut()); + } + Sender::new(Flavor::Shared(packet)) + } +} + +#[stable(feature = "rust1", since = "1.0.0")] +impl Drop for Sender { + fn drop(&mut self) { + match *unsafe { self.inner() } { + Flavor::Oneshot(ref p) => p.drop_chan(), + Flavor::Stream(ref p) => p.drop_chan(), + Flavor::Shared(ref p) => p.drop_chan(), + Flavor::Sync(..) => unreachable!(), + } + } +} + +#[stable(feature = "mpsc_debug", since = "1.8.0")] +impl fmt::Debug for Sender { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("Sender").finish_non_exhaustive() + } +} + +//////////////////////////////////////////////////////////////////////////////// +// SyncSender +//////////////////////////////////////////////////////////////////////////////// + +impl SyncSender { + fn new(inner: Arc>) -> SyncSender { + SyncSender { inner } + } + + /// Sends a value on this synchronous channel. + /// + /// This function will *block* until space in the internal buffer becomes + /// available or a receiver is available to hand off the message to. + /// + /// Note that a successful send does *not* guarantee that the receiver will + /// ever see the data if there is a buffer on this channel. Items may be + /// enqueued in the internal buffer for the receiver to receive at a later + /// time. If the buffer size is 0, however, the channel becomes a rendezvous + /// channel and it guarantees that the receiver has indeed received + /// the data if this function returns success. + /// + /// This function will never panic, but it may return [`Err`] if the + /// [`Receiver`] has disconnected and is no longer able to receive + /// information. + /// + /// # Examples + /// + /// ```rust + /// use std::sync::mpsc::sync_channel; + /// use std::thread; + /// + /// // Create a rendezvous sync_channel with buffer size 0 + /// let (sync_sender, receiver) = sync_channel(0); + /// + /// thread::spawn(move || { + /// println!("sending message..."); + /// sync_sender.send(1).unwrap(); + /// // Thread is now blocked until the message is received + /// + /// println!("...message received!"); + /// }); + /// + /// let msg = receiver.recv().unwrap(); + /// assert_eq!(1, msg); + /// ``` + #[stable(feature = "rust1", since = "1.0.0")] + pub fn send(&self, t: T) -> Result<(), SendError> { + self.inner.send(t).map_err(SendError) + } + + /// Attempts to send a value on this channel without blocking. + /// + /// This method differs from [`send`] by returning immediately if the + /// channel's buffer is full or no receiver is waiting to acquire some + /// data. Compared with [`send`], this function has two failure cases + /// instead of one (one for disconnection, one for a full buffer). + /// + /// See [`send`] for notes about guarantees of whether the + /// receiver has received the data or not if this function is successful. + /// + /// [`send`]: Self::send + /// + /// # Examples + /// + /// ```rust + /// use std::sync::mpsc::sync_channel; + /// use std::thread; + /// + /// // Create a sync_channel with buffer size 1 + /// let (sync_sender, receiver) = sync_channel(1); + /// let sync_sender2 = sync_sender.clone(); + /// + /// // First thread owns sync_sender + /// thread::spawn(move || { + /// sync_sender.send(1).unwrap(); + /// sync_sender.send(2).unwrap(); + /// // Thread blocked + /// }); + /// + /// // Second thread owns sync_sender2 + /// thread::spawn(move || { + /// // This will return an error and send + /// // no message if the buffer is full + /// let _ = sync_sender2.try_send(3); + /// }); + /// + /// let mut msg; + /// msg = receiver.recv().unwrap(); + /// println!("message {msg} received"); + /// + /// msg = receiver.recv().unwrap(); + /// println!("message {msg} received"); + /// + /// // Third message may have never been sent + /// match receiver.try_recv() { + /// Ok(msg) => println!("message {msg} received"), + /// Err(_) => println!("the third message was never sent"), + /// } + /// ``` + #[stable(feature = "rust1", since = "1.0.0")] + pub fn try_send(&self, t: T) -> Result<(), TrySendError> { + self.inner.try_send(t) + } +} + +#[stable(feature = "rust1", since = "1.0.0")] +impl Clone for SyncSender { + fn clone(&self) -> SyncSender { + self.inner.clone_chan(); + SyncSender::new(self.inner.clone()) + } +} + +#[stable(feature = "rust1", since = "1.0.0")] +impl Drop for SyncSender { + fn drop(&mut self) { + self.inner.drop_chan(); + } +} + +#[stable(feature = "mpsc_debug", since = "1.8.0")] +impl fmt::Debug for SyncSender { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("SyncSender").finish_non_exhaustive() + } +} + +//////////////////////////////////////////////////////////////////////////////// +// Receiver +//////////////////////////////////////////////////////////////////////////////// + +impl Receiver { + fn new(inner: Flavor) -> Receiver { + Receiver { inner: UnsafeCell::new(inner) } + } + + /// Attempts to return a pending value on this receiver without blocking. + /// + /// 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. + /// + /// 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 + /// use std::sync::mpsc::{Receiver, channel}; + /// + /// let (_, receiver): (_, Receiver) = channel(); + /// + /// assert!(receiver.try_recv().is_err()); + /// ``` + #[stable(feature = "rust1", since = "1.0.0")] + pub fn try_recv(&self) -> Result { + loop { + let new_port = match *unsafe { self.inner() } { + Flavor::Oneshot(ref p) => match p.try_recv() { + Ok(t) => return Ok(t), + Err(oneshot::Empty) => return Err(TryRecvError::Empty), + Err(oneshot::Disconnected) => return Err(TryRecvError::Disconnected), + Err(oneshot::Upgraded(rx)) => rx, + }, + Flavor::Stream(ref p) => match p.try_recv() { + Ok(t) => return Ok(t), + Err(stream::Empty) => return Err(TryRecvError::Empty), + Err(stream::Disconnected) => return Err(TryRecvError::Disconnected), + Err(stream::Upgraded(rx)) => rx, + }, + Flavor::Shared(ref p) => match p.try_recv() { + Ok(t) => return Ok(t), + Err(shared::Empty) => return Err(TryRecvError::Empty), + Err(shared::Disconnected) => return Err(TryRecvError::Disconnected), + }, + Flavor::Sync(ref p) => match p.try_recv() { + Ok(t) => return Ok(t), + Err(sync::Empty) => return Err(TryRecvError::Empty), + Err(sync::Disconnected) => return Err(TryRecvError::Disconnected), + }, + }; + unsafe { + mem::swap(self.inner_mut(), new_port.inner_mut()); + } + } + } + + /// Attempts to wait for a value on this receiver, returning an error if the + /// corresponding channel has hung up. + /// + /// 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`] + /// (or [`SyncSender`]), 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 + /// + /// ``` + /// use std::sync::mpsc; + /// use std::thread; + /// + /// let (send, recv) = mpsc::channel(); + /// let handle = thread::spawn(move || { + /// send.send(1u8).unwrap(); + /// }); + /// + /// handle.join().unwrap(); + /// + /// assert_eq!(Ok(1), recv.recv()); + /// ``` + /// + /// Buffering behavior: + /// + /// ``` + /// use std::sync::mpsc; + /// use std::thread; + /// use std::sync::mpsc::RecvError; + /// + /// let (send, recv) = mpsc::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()); + /// ``` + #[stable(feature = "rust1", since = "1.0.0")] + pub fn recv(&self) -> Result { + loop { + let new_port = match *unsafe { self.inner() } { + Flavor::Oneshot(ref p) => match p.recv(None) { + Ok(t) => return Ok(t), + Err(oneshot::Disconnected) => return Err(RecvError), + Err(oneshot::Upgraded(rx)) => rx, + Err(oneshot::Empty) => unreachable!(), + }, + Flavor::Stream(ref p) => match p.recv(None) { + Ok(t) => return Ok(t), + Err(stream::Disconnected) => return Err(RecvError), + Err(stream::Upgraded(rx)) => rx, + Err(stream::Empty) => unreachable!(), + }, + Flavor::Shared(ref p) => match p.recv(None) { + Ok(t) => return Ok(t), + Err(shared::Disconnected) => return Err(RecvError), + Err(shared::Empty) => unreachable!(), + }, + Flavor::Sync(ref p) => return p.recv(None).map_err(|_| RecvError), + }; + unsafe { + mem::swap(self.inner_mut(), new_port.inner_mut()); + } + } + } + + /// 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`] + /// (or [`SyncSender`]), 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. + /// + /// # Known Issues + /// + /// There is currently a known issue (see [`#39364`]) that causes `recv_timeout` + /// to panic unexpectedly with the following example: + /// + /// ```no_run + /// use std::sync::mpsc::channel; + /// use std::thread; + /// use std::time::Duration; + /// + /// let (tx, rx) = channel::(); + /// + /// thread::spawn(move || { + /// let d = Duration::from_millis(10); + /// loop { + /// println!("recv"); + /// let _r = rx.recv_timeout(d); + /// } + /// }); + /// + /// thread::sleep(Duration::from_millis(100)); + /// let _c1 = tx.clone(); + /// + /// thread::sleep(Duration::from_secs(1)); + /// ``` + /// + /// [`#39364`]: https://github.com/rust-lang/rust/issues/39364 + /// + /// # Examples + /// + /// Successfully receiving value before encountering timeout: + /// + /// ```no_run + /// use std::thread; + /// use std::time::Duration; + /// use std::sync::mpsc; + /// + /// let (send, recv) = mpsc::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 + /// use std::thread; + /// use std::time::Duration; + /// use std::sync::mpsc; + /// + /// let (send, recv) = mpsc::channel(); + /// + /// thread::spawn(move || { + /// thread::sleep(Duration::from_millis(800)); + /// send.send('a').unwrap(); + /// }); + /// + /// assert_eq!( + /// recv.recv_timeout(Duration::from_millis(400)), + /// Err(mpsc::RecvTimeoutError::Timeout) + /// ); + /// ``` + #[stable(feature = "mpsc_recv_timeout", since = "1.12.0")] + pub fn recv_timeout(&self, timeout: Duration) -> Result { + // Do an optimistic try_recv to avoid the performance impact of + // Instant::now() in the full-channel case. + match self.try_recv() { + Ok(result) => Ok(result), + Err(TryRecvError::Disconnected) => Err(RecvTimeoutError::Disconnected), + Err(TryRecvError::Empty) => match Instant::now().checked_add(timeout) { + Some(deadline) => self.recv_deadline(deadline), + // So far in the future that it's practically the same as waiting indefinitely. + None => self.recv().map_err(RecvTimeoutError::from), + }, + } + } + + /// 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`] (or [`SyncSender`]), 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(deadline_api)] + /// use std::thread; + /// use std::time::{Duration, Instant}; + /// use std::sync::mpsc; + /// + /// let (send, recv) = mpsc::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(deadline_api)] + /// use std::thread; + /// use std::time::{Duration, Instant}; + /// use std::sync::mpsc; + /// + /// let (send, recv) = mpsc::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(mpsc::RecvTimeoutError::Timeout) + /// ); + /// ``` + #[unstable(feature = "deadline_api", issue = "46316")] + pub fn recv_deadline(&self, deadline: Instant) -> Result { + use self::RecvTimeoutError::*; + + loop { + let port_or_empty = match *unsafe { self.inner() } { + Flavor::Oneshot(ref p) => match p.recv(Some(deadline)) { + Ok(t) => return Ok(t), + Err(oneshot::Disconnected) => return Err(Disconnected), + Err(oneshot::Upgraded(rx)) => Some(rx), + Err(oneshot::Empty) => None, + }, + Flavor::Stream(ref p) => match p.recv(Some(deadline)) { + Ok(t) => return Ok(t), + Err(stream::Disconnected) => return Err(Disconnected), + Err(stream::Upgraded(rx)) => Some(rx), + Err(stream::Empty) => None, + }, + Flavor::Shared(ref p) => match p.recv(Some(deadline)) { + Ok(t) => return Ok(t), + Err(shared::Disconnected) => return Err(Disconnected), + Err(shared::Empty) => None, + }, + Flavor::Sync(ref p) => match p.recv(Some(deadline)) { + Ok(t) => return Ok(t), + Err(sync::Disconnected) => return Err(Disconnected), + Err(sync::Empty) => None, + }, + }; + + if let Some(new_port) = port_or_empty { + unsafe { + mem::swap(self.inner_mut(), new_port.inner_mut()); + } + } + + // If we're already passed the deadline, and we're here without + // data, return a timeout, else try again. + if Instant::now() >= deadline { + return Err(Timeout); + } + } + } + + /// Returns an iterator that will block waiting for messages, but never + /// [`panic!`]. It will return [`None`] when the channel has hung up. + /// + /// # Examples + /// + /// ```rust + /// use std::sync::mpsc::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); + /// ``` + #[stable(feature = "rust1", since = "1.0.0")] + pub fn iter(&self) -> Iter<'_, T> { + Iter { rx: self } + } + + /// 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 + /// use std::sync::mpsc::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); + /// ``` + #[stable(feature = "receiver_try_iter", since = "1.15.0")] + pub fn try_iter(&self) -> TryIter<'_, T> { + TryIter { rx: self } + } +} + +#[stable(feature = "rust1", since = "1.0.0")] +impl<'a, T> Iterator for Iter<'a, T> { + type Item = T; + + fn next(&mut self) -> Option { + self.rx.recv().ok() + } +} + +#[stable(feature = "receiver_try_iter", since = "1.15.0")] +impl<'a, T> Iterator for TryIter<'a, T> { + type Item = T; + + fn next(&mut self) -> Option { + self.rx.try_recv().ok() + } +} + +#[stable(feature = "receiver_into_iter", since = "1.1.0")] +impl<'a, T> IntoIterator for &'a Receiver { + type Item = T; + type IntoIter = Iter<'a, T>; + + fn into_iter(self) -> Iter<'a, T> { + self.iter() + } +} + +#[stable(feature = "receiver_into_iter", since = "1.1.0")] +impl Iterator for IntoIter { + type Item = T; + fn next(&mut self) -> Option { + self.rx.recv().ok() + } +} + +#[stable(feature = "receiver_into_iter", since = "1.1.0")] +impl IntoIterator for Receiver { + type Item = T; + type IntoIter = IntoIter; + + fn into_iter(self) -> IntoIter { + IntoIter { rx: self } + } +} + +#[stable(feature = "rust1", since = "1.0.0")] +impl Drop for Receiver { + fn drop(&mut self) { + match *unsafe { self.inner() } { + Flavor::Oneshot(ref p) => p.drop_port(), + Flavor::Stream(ref p) => p.drop_port(), + Flavor::Shared(ref p) => p.drop_port(), + Flavor::Sync(ref p) => p.drop_port(), + } + } +} + +#[stable(feature = "mpsc_debug", since = "1.8.0")] +impl fmt::Debug for Receiver { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("Receiver").finish_non_exhaustive() + } +} + +#[stable(feature = "rust1", since = "1.0.0")] +impl fmt::Debug for SendError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("SendError").finish_non_exhaustive() + } +} + +#[stable(feature = "rust1", since = "1.0.0")] +impl fmt::Display for SendError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + "sending on a closed channel".fmt(f) + } +} + +#[stable(feature = "rust1", since = "1.0.0")] +impl error::Error for SendError { + #[allow(deprecated)] + fn description(&self) -> &str { + "sending on a closed channel" + } +} + +#[stable(feature = "rust1", since = "1.0.0")] +impl fmt::Debug for TrySendError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match *self { + TrySendError::Full(..) => "Full(..)".fmt(f), + TrySendError::Disconnected(..) => "Disconnected(..)".fmt(f), + } + } +} + +#[stable(feature = "rust1", since = "1.0.0")] +impl fmt::Display for TrySendError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match *self { + TrySendError::Full(..) => "sending on a full channel".fmt(f), + TrySendError::Disconnected(..) => "sending on a closed channel".fmt(f), + } + } +} + +#[stable(feature = "rust1", since = "1.0.0")] +impl error::Error for TrySendError { + #[allow(deprecated)] + fn description(&self) -> &str { + match *self { + TrySendError::Full(..) => "sending on a full channel", + TrySendError::Disconnected(..) => "sending on a closed channel", + } + } +} + +#[stable(feature = "mpsc_error_conversions", since = "1.24.0")] +impl From> for TrySendError { + /// Converts a `SendError` into a `TrySendError`. + /// + /// This conversion always returns a `TrySendError::Disconnected` containing the data in the `SendError`. + /// + /// No data is allocated on the heap. + fn from(err: SendError) -> TrySendError { + match err { + SendError(t) => TrySendError::Disconnected(t), + } + } +} + +#[stable(feature = "rust1", since = "1.0.0")] +impl fmt::Display for RecvError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + "receiving on a closed channel".fmt(f) + } +} + +#[stable(feature = "rust1", since = "1.0.0")] +impl error::Error for RecvError { + #[allow(deprecated)] + fn description(&self) -> &str { + "receiving on a closed channel" + } +} + +#[stable(feature = "rust1", since = "1.0.0")] +impl fmt::Display for TryRecvError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match *self { + TryRecvError::Empty => "receiving on an empty channel".fmt(f), + TryRecvError::Disconnected => "receiving on a closed channel".fmt(f), + } + } +} + +#[stable(feature = "rust1", since = "1.0.0")] +impl error::Error for TryRecvError { + #[allow(deprecated)] + fn description(&self) -> &str { + match *self { + TryRecvError::Empty => "receiving on an empty channel", + TryRecvError::Disconnected => "receiving on a closed channel", + } + } +} + +#[stable(feature = "mpsc_error_conversions", since = "1.24.0")] +impl From for TryRecvError { + /// Converts a `RecvError` into a `TryRecvError`. + /// + /// This conversion always returns `TryRecvError::Disconnected`. + /// + /// No data is allocated on the heap. + fn from(err: RecvError) -> TryRecvError { + match err { + RecvError => TryRecvError::Disconnected, + } + } +} + +#[stable(feature = "mpsc_recv_timeout_error", since = "1.15.0")] +impl fmt::Display for RecvTimeoutError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match *self { + RecvTimeoutError::Timeout => "timed out waiting on channel".fmt(f), + RecvTimeoutError::Disconnected => "channel is empty and sending half is closed".fmt(f), + } + } +} + +#[stable(feature = "mpsc_recv_timeout_error", since = "1.15.0")] +impl error::Error for RecvTimeoutError { + #[allow(deprecated)] + fn description(&self) -> &str { + match *self { + RecvTimeoutError::Timeout => "timed out waiting on channel", + RecvTimeoutError::Disconnected => "channel is empty and sending half is closed", + } + } +} + +#[stable(feature = "mpsc_error_conversions", since = "1.24.0")] +impl From for RecvTimeoutError { + /// Converts a `RecvError` into a `RecvTimeoutError`. + /// + /// This conversion always returns `RecvTimeoutError::Disconnected`. + /// + /// No data is allocated on the heap. + fn from(err: RecvError) -> RecvTimeoutError { + match err { + RecvError => RecvTimeoutError::Disconnected, + } + } +} -- cgit v1.2.3