mpsc.rs - source
Description: Source of the Rust file `library/std/src/sync/mpsc.rs`.

#
std/sync/
mpsc.rs

    1//! Multi-producer, single-consumer FIFO queue communication primitives.
    2//!
    3//! This module provides message-based communication over channels, concretely
    4//! defined among three types:
    5//!
    6//! * [`Sender`]
    7//! * [`SyncSender`]
    8//! * [`Receiver`]
    9//!
    10//! A [`Sender`] or [`SyncSender`] is used to send data to a [`Receiver`]. Both
    11//! senders are clone-able (multi-producer) such that many threads can send
    12//! simultaneously to one receiver (single-consumer).
    13//!
    14//! These channels come in two flavors:
    15//!
    16//! 1. An asynchronous, infinitely buffered channel. The [`channel`] function
    17//!    will return a `(Sender, Receiver)` tuple where all sends will be
    18//!    **asynchronous** (they never block). The channel conceptually has an
    19//!    infinite buffer.
    20//!
    21//! 2. A synchronous, bounded channel. The [`sync_channel`] function will
    22//!    return a `(SyncSender, Receiver)` tuple where the storage for pending
    23//!    messages is a pre-allocated buffer of a fixed size. All sends will be
    24//!    **synchronous** by blocking until there is buffer space available. Note
    25//!    that a bound of 0 is allowed, causing the channel to become a "rendezvous"
    26//!    channel where each sender atomically hands off a message to a receiver.
    27//!
    28//! [`send`]: Sender::send
    29//!
    30//! ## Disconnection
    31//!
    32//! The send and receive operations on channels will all return a [`Result`]
    33//! indicating whether the operation succeeded or not. An unsuccessful operation
    34//! is normally indicative of the other half of a channel having "hung up" by
    35//! being dropped in its corresponding thread.
    36//!
    37//! Once half of a channel has been deallocated, most operations can no longer
    38//! continue to make progress, so [`Err`] will be returned. Many applications
    39//! will continue to [`unwrap`] the results returned from this module,
    40//! instigating a propagation of failure among threads if one unexpectedly dies.
    41//!
    42//! [`unwrap`]: Result::unwrap
    43//!
    44//! # Examples
    45//!
    46//! Simple usage:
    47//!
    48//! ```
    49//! use std::thread;
    50//! use std::sync::mpsc::channel;
    51//!
    52//! // Create a simple streaming channel
    53//! let (tx, rx) = channel();
    54//! thread::spawn(move || {
    55//!     tx.send(10).unwrap();
    56//! });
    57//! assert_eq!(rx.recv().unwrap(), 10);
    58//! ```
    59//!
    60//! Shared usage:
    61//!
    62//! ```
    63//! use std::thread;
    64//! use std::sync::mpsc::channel;
    65//!
    66//! // Create a shared channel that can be sent along from many threads
    67//! // where tx is the sending half (tx for transmission), and rx is the receiving
    68//! // half (rx for receiving).
    69//! let (tx, rx) = channel();
    70//! for i in 0..10 {
    71//!     let tx = tx.clone();
    72//!     thread::spawn(move || {
    73//!         tx.send(i).unwrap();
    74//!     });
    75//! }
    76//!
    77//! for _ in 0..10 {
    78//!     let j = rx.recv().unwrap();
    79//!     assert!(0 <= j && j < 10);
    80//! }
    81//! ```
    82//!
    83//! Propagating panics:
    84//!
    85//! ```
    86//! use std::sync::mpsc::channel;
    87//!
    88//! // The call to recv() will return an error because the channel has already
    89//! // hung up (or been deallocated)
    90//! let (tx, rx) = channel::<i32>();
    91//! drop(tx);
    92//! assert!(rx.recv().is_err());
    93//! ```
    94//!
    95//! Synchronous channels:
    96//!
    97//! ```
    98//! use std::thread;
    99//! use std::sync::mpsc::sync_channel;
    100//!
    101//! let (tx, rx) = sync_channel::<i32>(0);
    102//! thread::spawn(move || {
    103//!     // This will wait for the parent thread to start receiving
    104//!     tx.send(53).unwrap();
    105//! });
    106//! rx.recv().unwrap();
    107//! ```
    108//!
    109//! Unbounded receive loop:
    110//!
    111//! ```
    112//! use std::sync::mpsc::sync_channel;
    113//! use std::thread;
    114//!
    115//! let (tx, rx) = sync_channel(3);
    116//!
    117//! for _ in 0..3 {
    118//!     // It would be the same without thread and clone here
    119//!     // since there will still be one `tx` left.
    120//!     let tx = tx.clone();
    121//!     // cloned tx dropped within thread
    122//!     thread::spawn(move || tx.send("ok").unwrap());
    123//! }
    124//!
    125//! // Drop the last sender to stop `rx` waiting for message.
    126//! // The program will not complete if we comment this out.
    127//! // **All** `tx` needs to be dropped for `rx` to have `Err`.
    128//! drop(tx);
    129//!
    130//! // Unbounded receiver waiting for all senders to complete.
    131//! while let Ok(msg) = rx.recv() {
    132//!     println!("{msg}");
    133//! }
    134//!
    135//! println!("completed");
    136//! ```
    137
    138#![stable(feature = "rust1", since = "1.0.0")]
    139
    140// MPSC channels are built as a wrapper around MPMC channels, which
    141// were ported from the `crossbeam-channel` crate. MPMC channels are
    142// not exposed publicly, but if you are curious about the implementation,
    143// that's where everything is.
    144
    145use crate::sync::mpmc;
    146use crate::time::{Duration, Instant};
    147use crate::{error, fmt};
    148
    149/// The receiving half of Rust's [`channel`] (or [`sync_channel`]) type.
    150/// This half can only be owned by one thread.
    151///
    152/// Messages sent to the channel can be retrieved using [`recv`].
    153///
    154/// [`recv`]: Receiver::recv
    155///
    156/// # Examples
    157///
    158/// ```rust
    159/// use std::sync::mpsc::channel;
    160/// use std::thread;
    161/// use std::time::Duration;
    162///
    163/// let (send, recv) = channel();
    164///
    165/// thread::spawn(move || {
    166///     send.send("Hello world!").unwrap();
    167///     thread::sleep(Duration::from_secs(2)); // block for two seconds
    168///     send.send("Delayed for 2 seconds").unwrap();
    169/// });
    170///
    171/// println!("{}", recv.recv().unwrap()); // Received immediately
    172/// println!("Waiting...");
    173/// println!("{}", recv.recv().unwrap()); // Received after 2 seconds
    174/// ```
    175#[stable(feature = "rust1", since = "1.0.0")]
    176#[cfg_attr(not(test), rustc_diagnostic_item = "Receiver")]
    177pub struct Receiver<T> {
    178    inner: mpmc::Receiver<T>,
    179}
    180
    181// The receiver port can be sent from place to place, so long as it
    182// is not used to receive non-sendable things.
    183#[stable(feature = "rust1", since = "1.0.0")]
    184unsafe impl<T: Send> Send for Receiver<T> {}
    185
    186#[stable(feature = "rust1", since = "1.0.0")]
    187impl<T> !Sync for Receiver<T> {}
    188
    189/// An iterator over messages on a [`Receiver`], created by [`iter`].
    190///
    191/// This iterator will block whenever [`next`] is called,
    192/// waiting for a new message, and [`None`] will be returned
    193/// when the corresponding channel has hung up.
    194///
    195/// [`iter`]: Receiver::iter
    196/// [`next`]: Iterator::next
    197///
    198/// # Examples
    199///
    200/// ```rust
    201/// use std::sync::mpsc::channel;
    202/// use std::thread;
    203///
    204/// let (send, recv) = channel();
    205///
    206/// thread::spawn(move || {
    207///     send.send(1u8).unwrap();
    208///     send.send(2u8).unwrap();
    209///     send.send(3u8).unwrap();
    210/// });
    211///
    212/// for x in recv.iter() {
    213///     println!("Got: {x}");
    214/// }
    215/// ```
    216#[stable(feature = "rust1", since = "1.0.0")]
    217#[derive(Debug)]
    218pub struct Iter<'a, T: 'a> {
    219    rx: &'a Receiver<T>,
    220}
    221
    222/// An iterator that attempts to yield all pending values for a [`Receiver`],
    223/// created by [`try_iter`].
    224///
    225/// [`None`] will be returned when there are no pending values remaining or
    226/// if the corresponding channel has hung up.
    227///
    228/// This iterator will never block the caller in order to wait for data to
    229/// become available. Instead, it will return [`None`].
    230///
    231/// [`try_iter`]: Receiver::try_iter
    232///
    233/// # Examples
    234///
    235/// ```rust
    236/// use std::sync::mpsc::channel;
    237/// use std::thread;
    238/// use std::time::Duration;
    239///
    240/// let (sender, receiver) = channel();
    241///
    242/// // Nothing is in the buffer yet
    243/// assert!(receiver.try_iter().next().is_none());
    244/// println!("Nothing in the buffer...");
    245///
    246/// thread::spawn(move || {
    247///     sender.send(1).unwrap();
    248///     sender.send(2).unwrap();
    249///     sender.send(3).unwrap();
    250/// });
    251///
    252/// println!("Going to sleep...");
    253/// thread::sleep(Duration::from_secs(2)); // block for two seconds
    254///
    255/// for x in receiver.try_iter() {
    256///     println!("Got: {x}");
    257/// }
    258/// ```
    259#[stable(feature = "receiver_try_iter", since = "1.15.0")]
    260#[derive(Debug)]
    261pub struct TryIter<'a, T: 'a> {
    262    rx: &'a Receiver<T>,
    263}
    264
    265/// An owning iterator over messages on a [`Receiver`],
    266/// created by [`into_iter`].
    267///
    268/// This iterator will block whenever [`next`]
    269/// is called, waiting for a new message, and [`None`] will be
    270/// returned if the corresponding channel has hung up.
    271///
    272/// [`into_iter`]: Receiver::into_iter
    273/// [`next`]: Iterator::next
    274///
    275/// # Examples
    276///
    277/// ```rust
    278/// use std::sync::mpsc::channel;
    279/// use std::thread;
    280///
    281/// let (send, recv) = channel();
    282///
    283/// thread::spawn(move || {
    284///     send.send(1u8).unwrap();
    285///     send.send(2u8).unwrap();
    286///     send.send(3u8).unwrap();
    287/// });
    288///
    289/// for x in recv.into_iter() {
    290///     println!("Got: {x}");
    291/// }
    292/// ```
    293#[stable(feature = "receiver_into_iter", since = "1.1.0")]
    294#[derive(Debug)]
    295pub struct IntoIter<T> {
    296    rx: Receiver<T>,
    297}
    298
    299/// The sending-half of Rust's asynchronous [`channel`] type.
    300///
    301/// Messages can be sent through this channel with [`send`].
    302///
    303/// Note: all senders (the original and its clones) need to be dropped for the receiver
    304/// to stop blocking to receive messages with [`Receiver::recv`].
    305///
    306/// [`send`]: Sender::send
    307///
    308/// # Examples
    309///
    310/// ```rust
    311/// use std::sync::mpsc::channel;
    312/// use std::thread;
    313///
    314/// let (sender, receiver) = channel();
    315/// let sender2 = sender.clone();
    316///
    317/// // First thread owns sender
    318/// thread::spawn(move || {
    319///     sender.send(1).unwrap();
    320/// });
    321///
    322/// // Second thread owns sender2
    323/// thread::spawn(move || {
    324///     sender2.send(2).unwrap();
    325/// });
    326///
    327/// let msg = receiver.recv().unwrap();
    328/// let msg2 = receiver.recv().unwrap();
    329///
    330/// assert_eq!(3, msg + msg2);
    331/// ```
    332#[stable(feature = "rust1", since = "1.0.0")]
    333pub struct Sender<T> {
    334    inner: mpmc::Sender<T>,
    335}
    336
    337// The send port can be sent from place to place, so long as it
    338// is not used to send non-sendable things.
    339#[stable(feature = "rust1", since = "1.0.0")]
    340unsafe impl<T: Send> Send for Sender<T> {}
    341
    342#[stable(feature = "mpsc_sender_sync", since = "1.72.0")]
    343unsafe impl<T: Send> Sync for Sender<T> {}
    344
    345/// The sending-half of Rust's synchronous [`sync_channel`] type.
    346///
    347/// Messages can be sent through this channel with [`send`] or [`try_send`].
    348///
    349/// [`send`] will block if there is no space in the internal buffer.
    350///
    351/// [`send`]: SyncSender::send
    352/// [`try_send`]: SyncSender::try_send
    353///
    354/// # Examples
    355///
    356/// ```rust
    357/// use std::sync::mpsc::sync_channel;
    358/// use std::thread;
    359///
    360/// // Create a sync_channel with buffer size 2
    361/// let (sync_sender, receiver) = sync_channel(2);
    362/// let sync_sender2 = sync_sender.clone();
    363///
    364/// // First thread owns sync_sender
    365/// thread::spawn(move || {
    366///     sync_sender.send(1).unwrap();
    367///     sync_sender.send(2).unwrap();
    368/// });
    369///
    370/// // Second thread owns sync_sender2
    371/// thread::spawn(move || {
    372///     sync_sender2.send(3).unwrap();
    373///     // thread will now block since the buffer is full
    374///     println!("Thread unblocked!");
    375/// });
    376///
    377/// let mut msg;
    378///
    379/// msg = receiver.recv().unwrap();
    380/// println!("message {msg} received");
    381///
    382/// // "Thread unblocked!" will be printed now
    383///
    384/// msg = receiver.recv().unwrap();
    385/// println!("message {msg} received");
    386///
    387/// msg = receiver.recv().unwrap();
    388///
    389/// println!("message {msg} received");
    390/// ```
    391#[stable(feature = "rust1", since = "1.0.0")]
    392pub struct SyncSender<T> {
    393    inner: mpmc::Sender<T>,
    394}
    395
    396#[stable(feature = "rust1", since = "1.0.0")]
    397unsafe impl<T: Send> Send for SyncSender<T> {}
    398
    399/// An error returned from the [`Sender::send`] or [`SyncSender::send`]
    400/// function on **channel**s.
    401///
    402/// A **send** operation can only fail if the receiving end of a channel is
    403/// disconnected, implying that the data could never be received. The error
    404/// contains the data being sent as a payload so it can be recovered.
    405#[stable(feature = "rust1", since = "1.0.0")]
    406#[derive(PartialEq, Eq, Clone, Copy)]
    407pub struct SendError<T>(#[stable(feature = "rust1", since = "1.0.0")] pub T);
    408
    409/// An error returned from the [`recv`] function on a [`Receiver`].
    410///
    411/// The [`recv`] operation can only fail if the sending half of a
    412/// [`channel`] (or [`sync_channel`]) is disconnected, implying that no further
    413/// messages will ever be received.
    414///
    415/// [`recv`]: Receiver::recv
    416#[derive(PartialEq, Eq, Clone, Copy, Debug)]
    417#[stable(feature = "rust1", since = "1.0.0")]
    418pub struct RecvError;
    419
    420/// This enumeration is the list of the possible reasons that [`try_recv`] could
    421/// not return data when called. This can occur with both a [`channel`] and
    422/// a [`sync_channel`].
    423///
    424/// [`try_recv`]: Receiver::try_recv
    425#[derive(PartialEq, Eq, Clone, Copy, Debug)]
    426#[stable(feature = "rust1", since = "1.0.0")]
    427pub enum TryRecvError {
    428    /// This **channel** is currently empty, but the **Sender**(s) have not yet
    429    /// disconnected, so data may yet become available.
    430    #[stable(feature = "rust1", since = "1.0.0")]
    431    Empty,
    432
    433    /// The **channel**'s sending half has become disconnected, and there will
    434    /// never be any more data received on it.
    435    #[stable(feature = "rust1", since = "1.0.0")]
    436    Disconnected,
    437}
    438
    439/// This enumeration is the list of possible errors that made [`recv_timeout`]
    440/// unable to return data when called. This can occur with both a [`channel`] and
    441/// a [`sync_channel`].
    442///
    443/// [`recv_timeout`]: Receiver::recv_timeout
    444#[derive(PartialEq, Eq, Clone, Copy, Debug)]
    445#[stable(feature = "mpsc_recv_timeout", since = "1.12.0")]
    446pub enum RecvTimeoutError {
    447    /// This **channel** is currently empty, but the **Sender**(s) have not yet
    448    /// disconnected, so data may yet become available.
    449    #[stable(feature = "mpsc_recv_timeout", since = "1.12.0")]
    450    Timeout,
    451    /// The **channel**'s sending half has become disconnected, and there will
    452    /// never be any more data received on it.
    453    #[stable(feature = "mpsc_recv_timeout", since = "1.12.0")]
    454    Disconnected,
    455}
    456
    457/// This enumeration is the list of the possible error outcomes for the
    458/// [`try_send`] method.
    459///
    460/// [`try_send`]: SyncSender::try_send
    461#[stable(feature = "rust1", since = "1.0.0")]
    462#[derive(PartialEq, Eq, Clone, Copy)]
    463pub enum TrySendError<T> {
    464    /// The data could not be sent on the [`sync_channel`] because it would require that
    465    /// the callee block to send the data.
    466    ///
    467    /// If this is a buffered channel, then the buffer is full at this time. If
    468    /// this is not a buffered channel, then there is no [`Receiver`] available to
    469    /// acquire the data.
    470    #[stable(feature = "rust1", since = "1.0.0")]
    471    Full(#[stable(feature = "rust1", since = "1.0.0")] T),
    472
    473    /// This [`sync_channel`]'s receiving half has disconnected, so the data could not be
    474    /// sent. The data is returned back to the callee in this case.
    475    #[stable(feature = "rust1", since = "1.0.0")]
    476    Disconnected(#[stable(feature = "rust1", since = "1.0.0")] T),
    477}
    478
    479/// Creates a new asynchronous channel, returning the sender/receiver halves.
    480///
    481/// All data sent on the [`Sender`] will become available on the [`Receiver`] in
    482/// the same order as it was sent, and no [`send`] will block the calling thread
    483/// (this channel has an "infinite buffer", unlike [`sync_channel`], which will
    484/// block after its buffer limit is reached). [`recv`] will block until a message
    485/// is available while there is at least one [`Sender`] alive (including clones).
    486///
    487/// The [`Sender`] can be cloned to [`send`] to the same channel multiple times, but
    488/// only one [`Receiver`] is supported.
    489///
    490/// If the [`Receiver`] is disconnected while trying to [`send`] with the
    491/// [`Sender`], the [`send`] method will return a [`SendError`]. Similarly, if the
    492/// [`Sender`] is disconnected while trying to [`recv`], the [`recv`] method will
    493/// return a [`RecvError`].
    494///
    495/// [`send`]: Sender::send
    496/// [`recv`]: Receiver::recv
    497///
    498/// # Examples
    499///
    500/// ```
    501/// use std::sync::mpsc::channel;
    502/// use std::thread;
    503///
    504/// let (sender, receiver) = channel();
    505///
    506/// // Spawn off an expensive computation
    507/// thread::spawn(move || {
    508/// #   fn expensive_computation() {}
    509///     sender.send(expensive_computation()).unwrap();
    510/// });
    511///
    512/// // Do some useful work for a while
    513///
    514/// // Let's see what that answer was
    515/// println!("{:?}", receiver.recv().unwrap());
    516/// ```
    517#[must_use]
    518#[stable(feature = "rust1", since = "1.0.0")]
    519pub fn channel<T>() -> (Sender<T>, Receiver<T>) {
    520    let (tx, rx) = mpmc::channel();
    521    (Sender { inner: tx }, Receiver { inner: rx })
    522}
    523
    524/// Creates a new synchronous, bounded channel.
    525///
    526/// All data sent on the [`SyncSender`] will become available on the [`Receiver`]
    527/// in the same order as it was sent. Like asynchronous [`channel`]s, the
    528/// [`Receiver`] will block until a message becomes available. `sync_channel`
    529/// differs greatly in the semantics of the sender, however.
    530///
    531/// This channel has an internal buffer on which messages will be queued.
    532/// `bound` specifies the buffer size. When the internal buffer becomes full,
    533/// future sends will *block* waiting for the buffer to open up. Note that a
    534/// buffer size of 0 is valid, in which case this becomes "rendezvous channel"
    535/// where each [`send`] will not return until a [`recv`] is paired with it.
    536///
    537/// The [`SyncSender`] can be cloned to [`send`] to the same channel multiple
    538/// times, but only one [`Receiver`] is supported.
    539///
    540/// Like asynchronous channels, if the [`Receiver`] is disconnected while trying
    541/// to [`send`] with the [`SyncSender`], the [`send`] method will return a
    542/// [`SendError`]. Similarly, If the [`SyncSender`] is disconnected while trying
    543/// to [`recv`], the [`recv`] method will return a [`RecvError`].
    544///
    545/// [`send`]: SyncSender::send
    546/// [`recv`]: Receiver::recv
    547///
    548/// # Examples
    549///
    550/// ```
    551/// use std::sync::mpsc::sync_channel;
    552/// use std::thread;
    553///
    554/// let (sender, receiver) = sync_channel(1);
    555///
    556/// // this returns immediately
    557/// sender.send(1).unwrap();
    558///
    559/// thread::spawn(move || {
    560///     // this will block until the previous message has been received
    561///     sender.send(2).unwrap();
    562/// });
    563///
    564/// assert_eq!(receiver.recv().unwrap(), 1);
    565/// assert_eq!(receiver.recv().unwrap(), 2);
    566/// ```
    567#[must_use]
    568#[stable(feature = "rust1", since = "1.0.0")]
    569pub fn sync_channel<T>(bound: usize) -> (SyncSender<T>, Receiver<T>) {
    570    let (tx, rx) = mpmc::sync_channel(bound);
    571    (SyncSender { inner: tx }, Receiver { inner: rx })
    572}
    573
    574////////////////////////////////////////////////////////////////////////////////
    575// Sender
    576////////////////////////////////////////////////////////////////////////////////
    577
    578impl<T> Sender<T> {
    579    /// Attempts to send a value on this channel, returning it back if it could
    580    /// not be sent.
    581    ///
    582    /// A successful send occurs when it is determined that the other end of
    583    /// the channel has not hung up already. An unsuccessful send would be one
    584    /// where the corresponding receiver has already been deallocated. Note
    585    /// that a return value of [`Err`] means that the data will never be
    586    /// received, but a return value of [`Ok`] does *not* mean that the data
    587    /// will be received. It is possible for the corresponding receiver to
    588    /// hang up immediately after this function returns [`Ok`].
    589    ///
    590    /// This method will never block the current thread.
    591    ///
    592    /// # Examples
    593    ///
    594    /// ```
    595    /// use std::sync::mpsc::channel;
    596    ///
    597    /// let (tx, rx) = channel();
    598    ///
    599    /// // This send is always successful
    600    /// tx.send(1).unwrap();
    601    ///
    602    /// // This send will fail because the receiver is gone
    603    /// drop(rx);
    604    /// assert_eq!(tx.send(1).unwrap_err().0, 1);
    605    /// ```
    606    #[stable(feature = "rust1", since = "1.0.0")]
    607    pub fn send(&self, t: T) -> Result<(), SendError<T>> {
    608        self.inner.send(t)
    609    }
    610}
    611
    612#[stable(feature = "rust1", since = "1.0.0")]
    613impl<T> Clone for Sender<T> {
    614    /// Clone a sender to send to other threads.
    615    ///
    616    /// Note, be aware of the lifetime of the sender because all senders
    617    /// (including the original) need to be dropped in order for
    618    /// [`Receiver::recv`] to stop blocking.
    619    fn clone(&self) -> Sender<T> {
    620        Sender { inner: self.inner.clone() }
    621    }
    622}
    623
    624#[stable(feature = "mpsc_debug", since = "1.8.0")]
    625impl<T> fmt::Debug for Sender<T> {
    626    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
    627        f.debug_struct("Sender").finish_non_exhaustive()
    628    }
    629}
    630
    631////////////////////////////////////////////////////////////////////////////////
    632// SyncSender
    633////////////////////////////////////////////////////////////////////////////////
    634
    635impl<T> SyncSender<T> {
    636    /// Sends a value on this synchronous channel.
    637    ///
    638    /// This function will *block* until space in the internal buffer becomes
    639    /// available or a receiver is available to hand off the message to.
    640    ///
    641    /// Note that a successful send does *not* guarantee that the receiver will
    642    /// ever see the data if there is a buffer on this channel. Items may be
    643    /// enqueued in the internal buffer for the receiver to receive at a later
    644    /// time. If the buffer size is 0, however, the channel becomes a rendezvous
    645    /// channel and it guarantees that the receiver has indeed received
    646    /// the data if this function returns success.
    647    ///
    648    /// This function will never panic, but it may return [`Err`] if the
    649    /// [`Receiver`] has disconnected and is no longer able to receive
    650    /// information.
    651    ///
    652    /// # Examples
    653    ///
    654    /// ```rust
    655    /// use std::sync::mpsc::sync_channel;
    656    /// use std::thread;
    657    ///
    658    /// // Create a rendezvous sync_channel with buffer size 0
    659    /// let (sync_sender, receiver) = sync_channel(0);
    660    ///
    661    /// thread::spawn(move || {
    662    ///    println!("sending message...");
    663    ///    sync_sender.send(1).unwrap();
    664    ///    // Thread is now blocked until the message is received
    665    ///
    666    ///    println!("...message received!");
    667    /// });
    668    ///
    669    /// let msg = receiver.recv().unwrap();
    670    /// assert_eq!(1, msg);
    671    /// ```
    672    #[stable(feature = "rust1", since = "1.0.0")]
    673    pub fn send(&self, t: T) -> Result<(), SendError<T>> {
    674        self.inner.send(t)
    675    }
    676
    677    /// Attempts to send a value on this channel without blocking.
    678    ///
    679    /// This method differs from [`send`] by returning immediately if the
    680    /// channel's buffer is full or no receiver is waiting to acquire some
    681    /// data. Compared with [`send`], this function has two failure cases
    682    /// instead of one (one for disconnection, one for a full buffer).
    683    ///
    684    /// See [`send`] for notes about guarantees of whether the
    685    /// receiver has received the data or not if this function is successful.
    686    ///
    687    /// [`send`]: Self::send
    688    ///
    689    /// # Examples
    690    ///
    691    /// ```rust
    692    /// use std::sync::mpsc::sync_channel;
    693    /// use std::thread;
    694    ///
    695    /// // Create a sync_channel with buffer size 1
    696    /// let (sync_sender, receiver) = sync_channel(1);
    697    /// let sync_sender2 = sync_sender.clone();
    698    ///
    699    /// // First thread owns sync_sender
    700    /// let handle1 = thread::spawn(move || {
    701    ///     sync_sender.send(1).unwrap();
    702    ///     sync_sender.send(2).unwrap();
    703    ///     // Thread blocked
    704    /// });
    705    ///
    706    /// // Second thread owns sync_sender2
    707    /// let handle2 = thread::spawn(move || {
    708    ///     // This will return an error and send
    709    ///     // no message if the buffer is full
    710    ///     let _ = sync_sender2.try_send(3);
    711    /// });
    712    ///
    713    /// let mut msg;
    714    /// msg = receiver.recv().unwrap();
    715    /// println!("message {msg} received");
    716    ///
    717    /// msg = receiver.recv().unwrap();
    718    /// println!("message {msg} received");
    719    ///
    720    /// // Third message may have never been sent
    721    /// match receiver.try_recv() {
    722    ///     Ok(msg) => println!("message {msg} received"),
    723    ///     Err(_) => println!("the third message was never sent"),
    724    /// }
    725    ///
    726    /// // Wait for threads to complete
    727    /// handle1.join().unwrap();
    728    /// handle2.join().unwrap();
    729    /// ```
    730    #[stable(feature = "rust1", since = "1.0.0")]
    731    pub fn try_send(&self, t: T) -> Result<(), TrySendError<T>> {
    732        self.inner.try_send(t)
    733    }
    734
    735    // Attempts to send for a value on this receiver, returning an error if the
    736    // corresponding channel has hung up, or if it waits more than `timeout`.
    737    //
    738    // This method is currently only used for tests.
    739    #[unstable(issue = "none", feature = "std_internals")]
    740    #[doc(hidden)]
    741    pub fn send_timeout(&self, t: T, timeout: Duration) -> Result<(), mpmc::SendTimeoutError<T>> {
    742        self.inner.send_timeout(t, timeout)
    743    }
    744}
    745
    746#[stable(feature = "rust1", since = "1.0.0")]
    747impl<T> Clone for SyncSender<T> {
    748    fn clone(&self) -> SyncSender<T> {
    749        SyncSender { inner: self.inner.clone() }
    750    }
    751}
    752
    753#[stable(feature = "mpsc_debug", since = "1.8.0")]
    754impl<T> fmt::Debug for SyncSender<T> {
    755    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
    756        f.debug_struct("SyncSender").finish_non_exhaustive()
    757    }
    758}
    759
    760////////////////////////////////////////////////////////////////////////////////
    761// Receiver
    762////////////////////////////////////////////////////////////////////////////////
    763
    764impl<T> Receiver<T> {
    765    /// Attempts to return a pending value on this receiver without blocking.
    766    ///
    767    /// This method will never block the caller in order to wait for data to
    768    /// become available. Instead, this will always return immediately with a
    769    /// possible option of pending data on the channel.
    770    ///
    771    /// This is useful for a flavor of "optimistic check" before deciding to
    772    /// block on a receiver.
    773    ///
    774    /// Compared with [`recv`], this function has two failure cases instead of one
    775    /// (one for disconnection, one for an empty buffer).
    776    ///
    777    /// [`recv`]: Self::recv
    778    ///
    779    /// # Examples
    780    ///
    781    /// ```rust
    782    /// use std::sync::mpsc::{Receiver, channel};
    783    ///
    784    /// let (_, receiver): (_, Receiver<i32>) = channel();
    785    ///
    786    /// assert!(receiver.try_recv().is_err());
    787    /// ```
    788    #[stable(feature = "rust1", since = "1.0.0")]
    789    pub fn try_recv(&self) -> Result<T, TryRecvError> {
    790        self.inner.try_recv()
    791    }
    792
    793    /// Attempts to wait for a value on this receiver, returning an error if the
    794    /// corresponding channel has hung up.
    795    ///
    796    /// This function will always block the current thread if there is no data
    797    /// available and it's possible for more data to be sent (at least one sender
    798    /// still exists). Once a message is sent to the corresponding [`Sender`]
    799    /// (or [`SyncSender`]), this receiver will wake up and return that
    800    /// message.
    801    ///
    802    /// If the corresponding [`Sender`] has disconnected, or it disconnects while
    803    /// this call is blocking, this call will wake up and return [`Err`] to
    804    /// indicate that no more messages can ever be received on this channel.
    805    /// However, since channels are buffered, messages sent before the disconnect
    806    /// will still be properly received.
    807    ///
    808    /// # Examples
    809    ///
    810    /// ```
    811    /// use std::sync::mpsc;
    812    /// use std::thread;
    813    ///
    814    /// let (send, recv) = mpsc::channel();
    815    /// let handle = thread::spawn(move || {
    816    ///     send.send(1u8).unwrap();
    817    /// });
    818    ///
    819    /// handle.join().unwrap();
    820    ///
    821    /// assert_eq!(Ok(1), recv.recv());
    822    /// ```
    823    ///
    824    /// Buffering behavior:
    825    ///
    826    /// ```
    827    /// use std::sync::mpsc;
    828    /// use std::thread;
    829    /// use std::sync::mpsc::RecvError;
    830    ///
    831    /// let (send, recv) = mpsc::channel();
    832    /// let handle = thread::spawn(move || {
    833    ///     send.send(1u8).unwrap();
    834    ///     send.send(2).unwrap();
    835    ///     send.send(3).unwrap();
    836    ///     drop(send);
    837    /// });
    838    ///
    839    /// // wait for the thread to join so we ensure the sender is dropped
    840    /// handle.join().unwrap();
    841    ///
    842    /// assert_eq!(Ok(1), recv.recv());
    843    /// assert_eq!(Ok(2), recv.recv());
    844    /// assert_eq!(Ok(3), recv.recv());
    845    /// assert_eq!(Err(RecvError), recv.recv());
    846    /// ```
    847    #[stable(feature = "rust1", since = "1.0.0")]
    848    pub fn recv(&self) -> Result<T, RecvError> {
    849        self.inner.recv()
    850    }
    851
    852    /// Attempts to wait for a value on this receiver, returning an error if the
    853    /// corresponding channel has hung up, or if it waits more than `timeout`.
    854    ///
    855    /// This function will always block the current thread if there is no data
    856    /// available and it's possible for more data to be sent (at least one sender
    857    /// still exists). Once a message is sent to the corresponding [`Sender`]
    858    /// (or [`SyncSender`]), this receiver will wake up and return that
    859    /// message.
    860    ///
    861    /// If the corresponding [`Sender`] has disconnected, or it disconnects while
    862    /// this call is blocking, this call will wake up and return [`Err`] to
    863    /// indicate that no more messages can ever be received on this channel.
    864    /// However, since channels are buffered, messages sent before the disconnect
    865    /// will still be properly received.
    866    ///
    867    /// # Examples
    868    ///
    869    /// Successfully receiving value before encountering timeout:
    870    ///
    871    /// ```no_run
    872    /// use std::thread;
    873    /// use std::time::Duration;
    874    /// use std::sync::mpsc;
    875    ///
    876    /// let (send, recv) = mpsc::channel();
    877    ///
    878    /// thread::spawn(move || {
    879    ///     send.send('a').unwrap();
    880    /// });
    881    ///
    882    /// assert_eq!(
    883    ///     recv.recv_timeout(Duration::from_millis(400)),
    884    ///     Ok('a')
    885    /// );
    886    /// ```
    887    ///
    888    /// Receiving an error upon reaching timeout:
    889    ///
    890    /// ```no_run
    891    /// use std::thread;
    892    /// use std::time::Duration;
    893    /// use std::sync::mpsc;
    894    ///
    895    /// let (send, recv) = mpsc::channel();
    896    ///
    897    /// thread::spawn(move || {
    898    ///     thread::sleep(Duration::from_millis(800));
    899    ///     send.send('a').unwrap();
    900    /// });
    901    ///
    902    /// assert_eq!(
    903    ///     recv.recv_timeout(Duration::from_millis(400)),
    904    ///     Err(mpsc::RecvTimeoutError::Timeout)
    905    /// );
    906    /// ```
    907    #[stable(feature = "mpsc_recv_timeout", since = "1.12.0")]
    908    pub fn recv_timeout(&self, timeout: Duration) -> Result<T, RecvTimeoutError> {
    909        self.inner.recv_timeout(timeout)
    910    }
    911
    912    /// Attempts to wait for a value on this receiver, returning an error if the
    913    /// corresponding channel has hung up, or if `deadline` is reached.
    914    ///
    915    /// This function will always block the current thread if there is no data
    916    /// available and it's possible for more data to be sent. Once a message is
    917    /// sent to the corresponding [`Sender`] (or [`SyncSender`]), then this
    918    /// receiver will wake up and return that message.
    919    ///
    920    /// If the corresponding [`Sender`] has disconnected, or it disconnects while
    921    /// this call is blocking, this call will wake up and return [`Err`] to
    922    /// indicate that no more messages can ever be received on this channel.
    923    /// However, since channels are buffered, messages sent before the disconnect
    924    /// will still be properly received.
    925    ///
    926    /// # Examples
    927    ///
    928    /// Successfully receiving value before reaching deadline:
    929    ///
    930    /// ```no_run
    931    /// #![feature(deadline_api)]
    932    /// use std::thread;
    933    /// use std::time::{Duration, Instant};
    934    /// use std::sync::mpsc;
    935    ///
    936    /// let (send, recv) = mpsc::channel();
    937    ///
    938    /// thread::spawn(move || {
    939    ///     send.send('a').unwrap();
    940    /// });
    941    ///
    942    /// assert_eq!(
    943    ///     recv.recv_deadline(Instant::now() + Duration::from_millis(400)),
    944    ///     Ok('a')
    945    /// );
    946    /// ```
    947    ///
    948    /// Receiving an error upon reaching deadline:
    949    ///
    950    /// ```no_run
    951    /// #![feature(deadline_api)]
    952    /// use std::thread;
    953    /// use std::time::{Duration, Instant};
    954    /// use std::sync::mpsc;
    955    ///
    956    /// let (send, recv) = mpsc::channel();
    957    ///
    958    /// thread::spawn(move || {
    959    ///     thread::sleep(Duration::from_millis(800));
    960    ///     send.send('a').unwrap();
    961    /// });
    962    ///
    963    /// assert_eq!(
    964    ///     recv.recv_deadline(Instant::now() + Duration::from_millis(400)),
    965    ///     Err(mpsc::RecvTimeoutError::Timeout)
    966    /// );
    967    /// ```
    968    #[unstable(feature = "deadline_api", issue = "46316")]
    969    pub fn recv_deadline(&self, deadline: Instant) -> Result<T, RecvTimeoutError> {
    970        self.inner.recv_deadline(deadline)
    971    }
    972
    973    /// Returns an iterator that will block waiting for messages, but never
    974    /// [`panic!`]. It will return [`None`] when the channel has hung up.
    975    ///
    976    /// # Examples
    977    ///
    978    /// ```rust
    979    /// use std::sync::mpsc::channel;
    980    /// use std::thread;
    981    ///
    982    /// let (send, recv) = channel();
    983    ///
    984    /// thread::spawn(move || {
    985    ///     send.send(1).unwrap();
    986    ///     send.send(2).unwrap();
    987    ///     send.send(3).unwrap();
    988    /// });
    989    ///
    990    /// let mut iter = recv.iter();
    991    /// assert_eq!(iter.next(), Some(1));
    992    /// assert_eq!(iter.next(), Some(2));
    993    /// assert_eq!(iter.next(), Some(3));
    994    /// assert_eq!(iter.next(), None);
    995    /// ```
    996    #[stable(feature = "rust1", since = "1.0.0")]
    997    pub fn iter(&self) -> Iter<'_, T> {
    998        Iter { rx: self }
    999    }
    1000
    1001    /// Returns an iterator that will attempt to yield all pending values.
    1002    /// It will return `None` if there are no more pending values or if the
    1003    /// channel has hung up. The iterator will never [`panic!`] or block the
    1004    /// user by waiting for values.
    1005    ///
    1006    /// # Examples
    1007    ///
    1008    /// ```no_run
    1009    /// use std::sync::mpsc::channel;
    1010    /// use std::thread;
    1011    /// use std::time::Duration;
    1012    ///
    1013    /// let (sender, receiver) = channel();
    1014    ///
    1015    /// // nothing is in the buffer yet
    1016    /// assert!(receiver.try_iter().next().is_none());
    1017    ///
    1018    /// thread::spawn(move || {
    1019    ///     thread::sleep(Duration::from_secs(1));
    1020    ///     sender.send(1).unwrap();
    1021    ///     sender.send(2).unwrap();
    1022    ///     sender.send(3).unwrap();
    1023    /// });
    1024    ///
    1025    /// // nothing is in the buffer yet
    1026    /// assert!(receiver.try_iter().next().is_none());
    1027    ///
    1028    /// // block for two seconds
    1029    /// thread::sleep(Duration::from_secs(2));
    1030    ///
    1031    /// let mut iter = receiver.try_iter();
    1032    /// assert_eq!(iter.next(), Some(1));
    1033    /// assert_eq!(iter.next(), Some(2));
    1034    /// assert_eq!(iter.next(), Some(3));
    1035    /// assert_eq!(iter.next(), None);
    1036    /// ```
    1037    #[stable(feature = "receiver_try_iter", since = "1.15.0")]
    1038    pub fn try_iter(&self) -> TryIter<'_, T> {
    1039        TryIter { rx: self }
    1040    }
    1041}
    1042
    1043#[stable(feature = "rust1", since = "1.0.0")]
    1044impl<'a, T> Iterator for Iter<'a, T> {
    1045    type Item = T;
    1046
    1047    fn next(&mut self) -> Option<T> {
    1048        self.rx.recv().ok()
    1049    }
    1050}
    1051
    1052#[stable(feature = "receiver_try_iter", since = "1.15.0")]
    1053impl<'a, T> Iterator for TryIter<'a, T> {
    1054    type Item = T;
    1055
    1056    fn next(&mut self) -> Option<T> {
    1057        self.rx.try_recv().ok()
    1058    }
    1059}
    1060
    1061#[stable(feature = "receiver_into_iter", since = "1.1.0")]
    1062impl<'a, T> IntoIterator for &'a Receiver<T> {
    1063    type Item = T;
    1064    type IntoIter = Iter<'a, T>;
    1065
    1066    fn into_iter(self) -> Iter<'a, T> {
    1067        self.iter()
    1068    }
    1069}
    1070
    1071#[stable(feature = "receiver_into_iter", since = "1.1.0")]
    1072impl<T> Iterator for IntoIter<T> {
    1073    type Item = T;
    1074    fn next(&mut self) -> Option<T> {
    1075        self.rx.recv().ok()
    1076    }
    1077}
    1078
    1079#[stable(feature = "receiver_into_iter", since = "1.1.0")]
    1080impl<T> IntoIterator for Receiver<T> {
    1081    type Item = T;
    1082    type IntoIter = IntoIter<T>;
    1083
    1084    fn into_iter(self) -> IntoIter<T> {
    1085        IntoIter { rx: self }
    1086    }
    1087}
    1088
    1089#[stable(feature = "mpsc_debug", since = "1.8.0")]
    1090impl<T> fmt::Debug for Receiver<T> {
    1091    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
    1092        f.debug_struct("Receiver").finish_non_exhaustive()
    1093    }
    1094}
    1095
    1096#[stable(feature = "rust1", since = "1.0.0")]
    1097impl<T> fmt::Debug for SendError<T> {
    1098    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
    1099        f.debug_struct("SendError").finish_non_exhaustive()
    1100    }
    1101}
    1102
    1103#[stable(feature = "rust1", since = "1.0.0")]
    1104impl<T> fmt::Display for SendError<T> {
    1105    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
    1106        "sending on a closed channel".fmt(f)
    1107    }
    1108}
    1109
    1110#[stable(feature = "rust1", since = "1.0.0")]
    1111impl<T> error::Error for SendError<T> {}
    1112
    1113#[stable(feature = "rust1", since = "1.0.0")]
    1114impl<T> fmt::Debug for TrySendError<T> {
    1115    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
    1116        match *self {
    1117            TrySendError::Full(..) => f.debug_tuple("TrySendError::Full").finish_non_exhaustive(),
    1118            TrySendError::Disconnected(..) => {
    1119                f.debug_tuple("TrySendError::Disconnected").finish_non_exhaustive()
    1120            }
    1121        }
    1122    }
    1123}
    1124
    1125#[stable(feature = "rust1", since = "1.0.0")]
    1126impl<T> fmt::Display for TrySendError<T> {
    1127    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
    1128        match *self {
    1129            TrySendError::Full(..) => "sending on a full channel".fmt(f),
    1130            TrySendError::Disconnected(..) => "sending on a closed channel".fmt(f),
    1131        }
    1132    }
    1133}
    1134
    1135#[stable(feature = "rust1", since = "1.0.0")]
    1136impl<T> error::Error for TrySendError<T> {}
    1137
    1138#[stable(feature = "mpsc_error_conversions", since = "1.24.0")]
    1139impl<T> From<SendError<T>> for TrySendError<T> {
    1140    /// Converts a `SendError<T>` into a `TrySendError<T>`.
    1141    ///
    1142    /// This conversion always returns a `TrySendError::Disconnected` containing the data in the `SendError<T>`.
    1143    ///
    1144    /// No data is allocated on the heap.
    1145    fn from(err: SendError<T>) -> TrySendError<T> {
    1146        match err {
    1147            SendError(t) => TrySendError::Disconnected(t),
    1148        }
    1149    }
    1150}
    1151
    1152#[stable(feature = "rust1", since = "1.0.0")]
    1153impl fmt::Display for RecvError {
    1154    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
    1155        "receiving on a closed channel".fmt(f)
    1156    }
    1157}
    1158
    1159#[stable(feature = "rust1", since = "1.0.0")]
    1160impl error::Error for RecvError {}
    1161
    1162#[stable(feature = "rust1", since = "1.0.0")]
    1163impl fmt::Display for TryRecvError {
    1164    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
    1165        match *self {
    1166            TryRecvError::Empty => "receiving on an empty channel".fmt(f),
    1167            TryRecvError::Disconnected => "receiving on a closed channel".fmt(f),
    1168        }
    1169    }
    1170}
    1171
    1172#[stable(feature = "rust1", since = "1.0.0")]
    1173impl error::Error for TryRecvError {}
    1174
    1175#[stable(feature = "mpsc_error_conversions", since = "1.24.0")]
    1176impl From<RecvError> for TryRecvError {
    1177    /// Converts a `RecvError` into a `TryRecvError`.
    1178    ///
    1179    /// This conversion always returns `TryRecvError::Disconnected`.
    1180    ///
    1181    /// No data is allocated on the heap.
    1182    fn from(err: RecvError) -> TryRecvError {
    1183        match err {
    1184            RecvError => TryRecvError::Disconnected,
    1185        }
    1186    }
    1187}
    1188
    1189#[stable(feature = "mpsc_recv_timeout_error", since = "1.15.0")]
    1190impl fmt::Display for RecvTimeoutError {
    1191    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
    1192        match *self {
    1193            RecvTimeoutError::Timeout => "timed out waiting on channel".fmt(f),
    1194            RecvTimeoutError::Disconnected => "channel is empty and sending half is closed".fmt(f),
    1195        }
    1196    }
    1197}
    1198
    1199#[stable(feature = "mpsc_recv_timeout_error", since = "1.15.0")]
    1200impl error::Error for RecvTimeoutError {}
    1201
    1202#[stable(feature = "mpsc_error_conversions", since = "1.24.0")]
    1203impl From<RecvError> for RecvTimeoutError {
    1204    /// Converts a `RecvError` into a `RecvTimeoutError`.
    1205    ///
    1206    /// This conversion always returns `RecvTimeoutError::Disconnected`.
    1207    ///
    1208    /// No data is allocated on the heap.
    1209    fn from(err: RecvError) -> RecvTimeoutError {
    1210        match err {
    1211            RecvError => RecvTimeoutError::Disconnected,
    1212        }
    1213    }
    1214}