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