1 //! Multi-producer, single-consumer FIFO queue communication primitives.
3 //! This module provides message-based communication over channels, concretely
4 //! defined among three types:
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).
14 //! These channels come in two flavors:
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
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.
28 //! [`send`]: Sender::send
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.
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.
42 //! [`unwrap`]: Result::unwrap
50 //! use std::sync::mpsc::channel;
52 //! // Create a simple streaming channel
53 //! let (tx, rx) = channel();
54 //! thread::spawn(move|| {
55 //! tx.send(10).unwrap();
57 //! assert_eq!(rx.recv().unwrap(), 10);
64 //! use std::sync::mpsc::channel;
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();
71 //! let tx = tx.clone();
72 //! thread::spawn(move|| {
73 //! tx.send(i).unwrap();
78 //! let j = rx.recv().unwrap();
79 //! assert!(0 <= j && j < 10);
83 //! Propagating panics:
86 //! use std::sync::mpsc::channel;
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>();
92 //! assert!(rx.recv().is_err());
95 //! Synchronous channels:
99 //! use std::sync::mpsc::sync_channel;
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();
106 //! rx.recv().unwrap();
109 //! Unbounded receive loop:
112 //! use std::sync::mpsc::sync_channel;
115 //! let (tx, rx) = sync_channel(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());
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`.
130 //! // Unbounded receiver waiting for all senders to complete.
131 //! while let Ok(msg) = rx.recv() {
132 //! println!("{msg}");
135 //! println!("completed");
138 #![stable(feature = "rust1", since = "1.0.0")]
140 #[cfg(all(test, not(target_os = "emscripten")))]
143 #[cfg(all(test, not(target_os = "emscripten")))]
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.
153 use crate::sync::mpmc;
154 use crate::time::{Duration, Instant};
156 /// The receiving half of Rust's [`channel`] (or [`sync_channel`]) type.
157 /// This half can only be owned by one thread.
159 /// Messages sent to the channel can be retrieved using [`recv`].
161 /// [`recv`]: Receiver::recv
166 /// use std::sync::mpsc::channel;
168 /// use std::time::Duration;
170 /// let (send, recv) = channel();
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();
178 /// println!("{}", recv.recv().unwrap()); // Received immediately
179 /// println!("Waiting...");
180 /// println!("{}", recv.recv().unwrap()); // Received after 2 seconds
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>,
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> {}
193 #[stable(feature = "rust1", since = "1.0.0")]
194 impl<T> !Sync for Receiver<T> {}
196 /// An iterator over messages on a [`Receiver`], created by [`iter`].
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.
202 /// [`iter`]: Receiver::iter
203 /// [`next`]: Iterator::next
208 /// use std::sync::mpsc::channel;
211 /// let (send, recv) = channel();
213 /// thread::spawn(move || {
214 /// send.send(1u8).unwrap();
215 /// send.send(2u8).unwrap();
216 /// send.send(3u8).unwrap();
219 /// for x in recv.iter() {
220 /// println!("Got: {x}");
223 #[stable(feature = "rust1", since = "1.0.0")]
225 pub struct Iter<'a, T: 'a> {
229 /// An iterator that attempts to yield all pending values for a [`Receiver`],
230 /// created by [`try_iter`].
232 /// [`None`] will be returned when there are no pending values remaining or
233 /// if the corresponding channel has hung up.
235 /// This iterator will never block the caller in order to wait for data to
236 /// become available. Instead, it will return [`None`].
238 /// [`try_iter`]: Receiver::try_iter
243 /// use std::sync::mpsc::channel;
245 /// use std::time::Duration;
247 /// let (sender, receiver) = channel();
249 /// // Nothing is in the buffer yet
250 /// assert!(receiver.try_iter().next().is_none());
251 /// println!("Nothing in the buffer...");
253 /// thread::spawn(move || {
254 /// sender.send(1).unwrap();
255 /// sender.send(2).unwrap();
256 /// sender.send(3).unwrap();
259 /// println!("Going to sleep...");
260 /// thread::sleep(Duration::from_secs(2)); // block for two seconds
262 /// for x in receiver.try_iter() {
263 /// println!("Got: {x}");
266 #[stable(feature = "receiver_try_iter", since = "1.15.0")]
268 pub struct TryIter<'a, T: 'a> {
272 /// An owning iterator over messages on a [`Receiver`],
273 /// created by [`into_iter`].
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.
279 /// [`into_iter`]: Receiver::into_iter
280 /// [`next`]: Iterator::next
285 /// use std::sync::mpsc::channel;
288 /// let (send, recv) = channel();
290 /// thread::spawn(move || {
291 /// send.send(1u8).unwrap();
292 /// send.send(2u8).unwrap();
293 /// send.send(3u8).unwrap();
296 /// for x in recv.into_iter() {
297 /// println!("Got: {x}");
300 #[stable(feature = "receiver_into_iter", since = "1.1.0")]
302 pub struct IntoIter<T> {
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.
309 /// Messages can be sent through this channel with [`send`].
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`].
314 /// [`send`]: Sender::send
319 /// use std::sync::mpsc::channel;
322 /// let (sender, receiver) = channel();
323 /// let sender2 = sender.clone();
325 /// // First thread owns sender
326 /// thread::spawn(move || {
327 /// sender.send(1).unwrap();
330 /// // Second thread owns sender2
331 /// thread::spawn(move || {
332 /// sender2.send(2).unwrap();
335 /// let msg = receiver.recv().unwrap();
336 /// let msg2 = receiver.recv().unwrap();
338 /// assert_eq!(3, msg + msg2);
340 #[stable(feature = "rust1", since = "1.0.0")]
341 pub struct Sender<T> {
342 inner: mpmc::Sender<T>,
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> {}
350 #[stable(feature = "rust1", since = "1.0.0")]
351 impl<T> !Sync for Sender<T> {}
353 /// The sending-half of Rust's synchronous [`sync_channel`] type.
355 /// Messages can be sent through this channel with [`send`] or [`try_send`].
357 /// [`send`] will block if there is no space in the internal buffer.
359 /// [`send`]: SyncSender::send
360 /// [`try_send`]: SyncSender::try_send
365 /// use std::sync::mpsc::sync_channel;
368 /// // Create a sync_channel with buffer size 2
369 /// let (sync_sender, receiver) = sync_channel(2);
370 /// let sync_sender2 = sync_sender.clone();
372 /// // First thread owns sync_sender
373 /// thread::spawn(move || {
374 /// sync_sender.send(1).unwrap();
375 /// sync_sender.send(2).unwrap();
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!");
387 /// msg = receiver.recv().unwrap();
388 /// println!("message {msg} received");
390 /// // "Thread unblocked!" will be printed now
392 /// msg = receiver.recv().unwrap();
393 /// println!("message {msg} received");
395 /// msg = receiver.recv().unwrap();
397 /// println!("message {msg} received");
399 #[stable(feature = "rust1", since = "1.0.0")]
400 pub struct SyncSender<T> {
401 inner: mpmc::Sender<T>,
404 #[stable(feature = "rust1", since = "1.0.0")]
405 unsafe impl<T: Send> Send for SyncSender<T> {}
407 /// An error returned from the [`Sender::send`] or [`SyncSender::send`]
408 /// function on **channel**s.
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);
417 /// An error returned from the [`recv`] function on a [`Receiver`].
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.
423 /// [`recv`]: Receiver::recv
424 #[derive(PartialEq, Eq, Clone, Copy, Debug)]
425 #[stable(feature = "rust1", since = "1.0.0")]
426 pub struct RecvError;
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`].
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")]
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")]
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`].
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")]
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")]
465 /// This enumeration is the list of the possible error outcomes for the
466 /// [`try_send`] method.
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.
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),
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),
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).
494 /// The [`Sender`] can be cloned to [`send`] to the same channel multiple times, but
495 /// only one [`Receiver`] is supported.
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`].
502 /// [`send`]: Sender::send
503 /// [`recv`]: Receiver::recv
508 /// use std::sync::mpsc::channel;
511 /// let (sender, receiver) = channel();
513 /// // Spawn off an expensive computation
514 /// thread::spawn(move|| {
515 /// # fn expensive_computation() {}
516 /// sender.send(expensive_computation()).unwrap();
519 /// // Do some useful work for awhile
521 /// // Let's see what that answer was
522 /// println!("{:?}", receiver.recv().unwrap());
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 })
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.
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.
543 /// The [`SyncSender`] can be cloned to [`send`] to the same channel multiple
544 /// times, but only one [`Receiver`] is supported.
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`].
551 /// [`send`]: SyncSender::send
552 /// [`recv`]: Receiver::recv
557 /// use std::sync::mpsc::sync_channel;
560 /// let (sender, receiver) = sync_channel(1);
562 /// // this returns immediately
563 /// sender.send(1).unwrap();
565 /// thread::spawn(move|| {
566 /// // this will block until the previous message has been received
567 /// sender.send(2).unwrap();
570 /// assert_eq!(receiver.recv().unwrap(), 1);
571 /// assert_eq!(receiver.recv().unwrap(), 2);
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 })
580 ////////////////////////////////////////////////////////////////////////////////
582 ////////////////////////////////////////////////////////////////////////////////
585 /// Attempts to send a value on this channel, returning it back if it could
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`].
596 /// This method will never block the current thread.
601 /// use std::sync::mpsc::channel;
603 /// let (tx, rx) = channel();
605 /// // This send is always successful
606 /// tx.send(1).unwrap();
608 /// // This send will fail because the receiver is gone
610 /// assert_eq!(tx.send(1).unwrap_err().0, 1);
612 #[stable(feature = "rust1", since = "1.0.0")]
613 pub fn send(&self, t: T) -> Result<(), SendError<T>> {
618 #[stable(feature = "rust1", since = "1.0.0")]
619 impl<T> Clone for Sender<T> {
620 /// Clone a sender to send to other threads.
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() }
630 #[stable(feature = "rust1", since = "1.0.0")]
631 impl<T> Drop for Sender<T> {
632 fn drop(&mut self) {}
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()
642 ////////////////////////////////////////////////////////////////////////////////
644 ////////////////////////////////////////////////////////////////////////////////
646 impl<T> SyncSender<T> {
647 /// Sends a value on this synchronous channel.
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.
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.
659 /// This function will never panic, but it may return [`Err`] if the
660 /// [`Receiver`] has disconnected and is no longer able to receive
666 /// use std::sync::mpsc::sync_channel;
669 /// // Create a rendezvous sync_channel with buffer size 0
670 /// let (sync_sender, receiver) = sync_channel(0);
672 /// thread::spawn(move || {
673 /// println!("sending message...");
674 /// sync_sender.send(1).unwrap();
675 /// // Thread is now blocked until the message is received
677 /// println!("...message received!");
680 /// let msg = receiver.recv().unwrap();
681 /// assert_eq!(1, msg);
683 #[stable(feature = "rust1", since = "1.0.0")]
684 pub fn send(&self, t: T) -> Result<(), SendError<T>> {
688 /// Attempts to send a value on this channel without blocking.
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).
695 /// See [`send`] for notes about guarantees of whether the
696 /// receiver has received the data or not if this function is successful.
698 /// [`send`]: Self::send
703 /// use std::sync::mpsc::sync_channel;
706 /// // Create a sync_channel with buffer size 1
707 /// let (sync_sender, receiver) = sync_channel(1);
708 /// let sync_sender2 = sync_sender.clone();
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
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);
725 /// msg = receiver.recv().unwrap();
726 /// println!("message {msg} received");
728 /// msg = receiver.recv().unwrap();
729 /// println!("message {msg} received");
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"),
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)
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() }
750 #[stable(feature = "rust1", since = "1.0.0")]
751 impl<T> Drop for SyncSender<T> {
752 fn drop(&mut self) {}
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()
762 ////////////////////////////////////////////////////////////////////////////////
764 ////////////////////////////////////////////////////////////////////////////////
766 impl<T> Receiver<T> {
767 /// Attempts to return a pending value on this receiver without blocking.
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.
773 /// This is useful for a flavor of "optimistic check" before deciding to
774 /// block on a receiver.
776 /// Compared with [`recv`], this function has two failure cases instead of one
777 /// (one for disconnection, one for an empty buffer).
779 /// [`recv`]: Self::recv
784 /// use std::sync::mpsc::{Receiver, channel};
786 /// let (_, receiver): (_, Receiver<i32>) = channel();
788 /// assert!(receiver.try_recv().is_err());
790 #[stable(feature = "rust1", since = "1.0.0")]
791 pub fn try_recv(&self) -> Result<T, TryRecvError> {
792 self.inner.try_recv()
795 /// Attempts to wait for a value on this receiver, returning an error if the
796 /// corresponding channel has hung up.
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
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.
813 /// use std::sync::mpsc;
816 /// let (send, recv) = mpsc::channel();
817 /// let handle = thread::spawn(move || {
818 /// send.send(1u8).unwrap();
821 /// handle.join().unwrap();
823 /// assert_eq!(Ok(1), recv.recv());
826 /// Buffering behavior:
829 /// use std::sync::mpsc;
831 /// use std::sync::mpsc::RecvError;
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();
841 /// // wait for the thread to join so we ensure the sender is dropped
842 /// handle.join().unwrap();
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());
849 #[stable(feature = "rust1", since = "1.0.0")]
850 pub fn recv(&self) -> Result<T, RecvError> {
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`.
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
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.
871 /// Successfully receiving value before encountering timeout:
875 /// use std::time::Duration;
876 /// use std::sync::mpsc;
878 /// let (send, recv) = mpsc::channel();
880 /// thread::spawn(move || {
881 /// send.send('a').unwrap();
885 /// recv.recv_timeout(Duration::from_millis(400)),
890 /// Receiving an error upon reaching timeout:
894 /// use std::time::Duration;
895 /// use std::sync::mpsc;
897 /// let (send, recv) = mpsc::channel();
899 /// thread::spawn(move || {
900 /// thread::sleep(Duration::from_millis(800));
901 /// send.send('a').unwrap();
905 /// recv.recv_timeout(Duration::from_millis(400)),
906 /// Err(mpsc::RecvTimeoutError::Timeout)
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)
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.
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.
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.
930 /// Successfully receiving value before reaching deadline:
933 /// #![feature(deadline_api)]
935 /// use std::time::{Duration, Instant};
936 /// use std::sync::mpsc;
938 /// let (send, recv) = mpsc::channel();
940 /// thread::spawn(move || {
941 /// send.send('a').unwrap();
945 /// recv.recv_deadline(Instant::now() + Duration::from_millis(400)),
950 /// Receiving an error upon reaching deadline:
953 /// #![feature(deadline_api)]
955 /// use std::time::{Duration, Instant};
956 /// use std::sync::mpsc;
958 /// let (send, recv) = mpsc::channel();
960 /// thread::spawn(move || {
961 /// thread::sleep(Duration::from_millis(800));
962 /// send.send('a').unwrap();
966 /// recv.recv_deadline(Instant::now() + Duration::from_millis(400)),
967 /// Err(mpsc::RecvTimeoutError::Timeout)
970 #[unstable(feature = "deadline_api", issue = "46316")]
971 pub fn recv_deadline(&self, deadline: Instant) -> Result<T, RecvTimeoutError> {
972 self.inner.recv_deadline(deadline)
975 /// Returns an iterator that will block waiting for messages, but never
976 /// [`panic!`]. It will return [`None`] when the channel has hung up.
981 /// use std::sync::mpsc::channel;
984 /// let (send, recv) = channel();
986 /// thread::spawn(move || {
987 /// send.send(1).unwrap();
988 /// send.send(2).unwrap();
989 /// send.send(3).unwrap();
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);
998 #[stable(feature = "rust1", since = "1.0.0")]
999 pub fn iter(&self) -> Iter<'_, T> {
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.
1011 /// use std::sync::mpsc::channel;
1012 /// use std::thread;
1013 /// use std::time::Duration;
1015 /// let (sender, receiver) = channel();
1017 /// // nothing is in the buffer yet
1018 /// assert!(receiver.try_iter().next().is_none());
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();
1027 /// // nothing is in the buffer yet
1028 /// assert!(receiver.try_iter().next().is_none());
1030 /// // block for two seconds
1031 /// thread::sleep(Duration::from_secs(2));
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);
1039 #[stable(feature = "receiver_try_iter", since = "1.15.0")]
1040 pub fn try_iter(&self) -> TryIter<'_, T> {
1041 TryIter { rx: self }
1045 #[stable(feature = "rust1", since = "1.0.0")]
1046 impl<'a, T> Iterator for Iter<'a, T> {
1049 fn next(&mut self) -> Option<T> {
1054 #[stable(feature = "receiver_try_iter", since = "1.15.0")]
1055 impl<'a, T> Iterator for TryIter<'a, T> {
1058 fn next(&mut self) -> Option<T> {
1059 self.rx.try_recv().ok()
1063 #[stable(feature = "receiver_into_iter", since = "1.1.0")]
1064 impl<'a, T> IntoIterator for &'a Receiver<T> {
1066 type IntoIter = Iter<'a, T>;
1068 fn into_iter(self) -> Iter<'a, T> {
1073 #[stable(feature = "receiver_into_iter", since = "1.1.0")]
1074 impl<T> Iterator for IntoIter<T> {
1076 fn next(&mut self) -> Option<T> {
1081 #[stable(feature = "receiver_into_iter", since = "1.1.0")]
1082 impl<T> IntoIterator for Receiver<T> {
1084 type IntoIter = IntoIter<T>;
1086 fn into_iter(self) -> IntoIter<T> {
1087 IntoIter { rx: self }
1091 #[stable(feature = "rust1", since = "1.0.0")]
1092 impl<T> Drop for Receiver<T> {
1093 fn drop(&mut self) {}
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()
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()
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)
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"
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 {
1129 TrySendError::Full(..) => "Full(..)".fmt(f),
1130 TrySendError::Disconnected(..) => "Disconnected(..)".fmt(f),
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 {
1139 TrySendError::Full(..) => "sending on a full channel".fmt(f),
1140 TrySendError::Disconnected(..) => "sending on a closed channel".fmt(f),
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 {
1150 TrySendError::Full(..) => "sending on a full channel",
1151 TrySendError::Disconnected(..) => "sending on a closed channel",
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>`.
1160 /// This conversion always returns a `TrySendError::Disconnected` containing the data in the `SendError<T>`.
1162 /// No data is allocated on the heap.
1163 fn from(err: SendError<T>) -> TrySendError<T> {
1165 SendError(t) => TrySendError::Disconnected(t),
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)
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"
1185 #[stable(feature = "rust1", since = "1.0.0")]
1186 impl fmt::Display for TryRecvError {
1187 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1189 TryRecvError::Empty => "receiving on an empty channel".fmt(f),
1190 TryRecvError::Disconnected => "receiving on a closed channel".fmt(f),
1195 #[stable(feature = "rust1", since = "1.0.0")]
1196 impl error::Error for TryRecvError {
1197 #[allow(deprecated)]
1198 fn description(&self) -> &str {
1200 TryRecvError::Empty => "receiving on an empty channel",
1201 TryRecvError::Disconnected => "receiving on a closed channel",
1206 #[stable(feature = "mpsc_error_conversions", since = "1.24.0")]
1207 impl From<RecvError> for TryRecvError {
1208 /// Converts a `RecvError` into a `TryRecvError`.
1210 /// This conversion always returns `TryRecvError::Disconnected`.
1212 /// No data is allocated on the heap.
1213 fn from(err: RecvError) -> TryRecvError {
1215 RecvError => TryRecvError::Disconnected,
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 {
1224 RecvTimeoutError::Timeout => "timed out waiting on channel".fmt(f),
1225 RecvTimeoutError::Disconnected => "channel is empty and sending half is closed".fmt(f),
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 {
1235 RecvTimeoutError::Timeout => "timed out waiting on channel",
1236 RecvTimeoutError::Disconnected => "channel is empty and sending half is closed",
1241 #[stable(feature = "mpsc_error_conversions", since = "1.24.0")]
1242 impl From<RecvError> for RecvTimeoutError {
1243 /// Converts a `RecvError` into a `RecvTimeoutError`.
1245 /// This conversion always returns `RecvTimeoutError::Disconnected`.
1247 /// No data is allocated on the heap.
1248 fn from(err: RecvError) -> RecvTimeoutError {
1250 RecvError => RecvTimeoutError::Disconnected,