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")))]
141mod tests;
142
143#[cfg(all(test, not(target_os = "emscripten")))]
144mod 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
151use crate::error;
152use crate::fmt;
153use crate::sync::mpmc;
154use 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")]
184pub 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")]
191unsafe impl<T: Send> Send for Receiver<T> {}
192
193#[stable(feature = "rust1", since = "1.0.0")]
194impl<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)]
225pub 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)]
268pub 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)]
302pub struct IntoIter<T> {
303 rx: Receiver<T>,
304}
305
306/// The sending-half of Rust's asynchronous [`channel`] type.
307///
308/// Messages can be sent through this channel with [`send`].
309///
310/// Note: all senders (the original and its clones) need to be dropped for the receiver
311/// to stop blocking to receive messages with [`Receiver::recv`].
312///
313/// [`send`]: Sender::send
314///
315/// # Examples
316///
317/// ```rust
318/// use std::sync::mpsc::channel;
319/// use std::thread;
320///
321/// let (sender, receiver) = channel();
322/// let sender2 = sender.clone();
323///
324/// // First thread owns sender
325/// thread::spawn(move || {
326/// sender.send(1).unwrap();
327/// });
328///
329/// // Second thread owns sender2
330/// thread::spawn(move || {
331/// sender2.send(2).unwrap();
332/// });
333///
334/// let msg = receiver.recv().unwrap();
335/// let msg2 = receiver.recv().unwrap();
336///
337/// assert_eq!(3, msg + msg2);
338/// ```
339#[stable(feature = "rust1", since = "1.0.0")]
340pub struct Sender<T> {
341 inner: mpmc::Sender<T>,
342}
343
344// The send port can be sent from place to place, so long as it
345// is not used to send non-sendable things.
346#[stable(feature = "rust1", since = "1.0.0")]
347unsafe impl<T: Send> Send for Sender<T> {}
348
349#[stable(feature = "mpsc_sender_sync", since = "1.72.0")]
350unsafe impl<T: Send> Sync for Sender<T> {}
351
352/// The sending-half of Rust's synchronous [`sync_channel`] type.
353///
354/// Messages can be sent through this channel with [`send`] or [`try_send`].
355///
356/// [`send`] will block if there is no space in the internal buffer.
357///
358/// [`send`]: SyncSender::send
359/// [`try_send`]: SyncSender::try_send
360///
361/// # Examples
362///
363/// ```rust
364/// use std::sync::mpsc::sync_channel;
365/// use std::thread;
366///
367/// // Create a sync_channel with buffer size 2
368/// let (sync_sender, receiver) = sync_channel(2);
369/// let sync_sender2 = sync_sender.clone();
370///
371/// // First thread owns sync_sender
372/// thread::spawn(move || {
373/// sync_sender.send(1).unwrap();
374/// sync_sender.send(2).unwrap();
375/// });
376///
377/// // Second thread owns sync_sender2
378/// thread::spawn(move || {
379/// sync_sender2.send(3).unwrap();
380/// // thread will now block since the buffer is full
381/// println!("Thread unblocked!");
382/// });
383///
384/// let mut msg;
385///
386/// msg = receiver.recv().unwrap();
387/// println!("message {msg} received");
388///
389/// // "Thread unblocked!" will be printed now
390///
391/// msg = receiver.recv().unwrap();
392/// println!("message {msg} received");
393///
394/// msg = receiver.recv().unwrap();
395///
396/// println!("message {msg} received");
397/// ```
398#[stable(feature = "rust1", since = "1.0.0")]
399pub struct SyncSender<T> {
400 inner: mpmc::Sender<T>,
401}
402
403#[stable(feature = "rust1", since = "1.0.0")]
404unsafe impl<T: Send> Send for SyncSender<T> {}
405
406/// An error returned from the [`Sender::send`] or [`SyncSender::send`]
407/// function on **channel**s.
408///
409/// A **send** operation can only fail if the receiving end of a channel is
410/// disconnected, implying that the data could never be received. The error
411/// contains the data being sent as a payload so it can be recovered.
412#[stable(feature = "rust1", since = "1.0.0")]
413#[derive(PartialEq, Eq, Clone, Copy)]
414pub struct SendError<T>(#[stable(feature = "rust1", since = "1.0.0")] pub T);
415
416/// An error returned from the [`recv`] function on a [`Receiver`].
417///
418/// The [`recv`] operation can only fail if the sending half of a
419/// [`channel`] (or [`sync_channel`]) is disconnected, implying that no further
420/// messages will ever be received.
421///
422/// [`recv`]: Receiver::recv
423#[derive(PartialEq, Eq, Clone, Copy, Debug)]
424#[stable(feature = "rust1", since = "1.0.0")]
425pub struct RecvError;
426
427/// This enumeration is the list of the possible reasons that [`try_recv`] could
428/// not return data when called. This can occur with both a [`channel`] and
429/// a [`sync_channel`].
430///
431/// [`try_recv`]: Receiver::try_recv
432#[derive(PartialEq, Eq, Clone, Copy, Debug)]
433#[stable(feature = "rust1", since = "1.0.0")]
434pub enum TryRecvError {
435 /// This **channel** is currently empty, but the **Sender**(s) have not yet
436 /// disconnected, so data may yet become available.
437 #[stable(feature = "rust1", since = "1.0.0")]
438 Empty,
439
440 /// The **channel**'s sending half has become disconnected, and there will
441 /// never be any more data received on it.
442 #[stable(feature = "rust1", since = "1.0.0")]
443 Disconnected,
444}
445
446/// This enumeration is the list of possible errors that made [`recv_timeout`]
447/// unable to return data when called. This can occur with both a [`channel`] and
448/// a [`sync_channel`].
449///
450/// [`recv_timeout`]: Receiver::recv_timeout
451#[derive(PartialEq, Eq, Clone, Copy, Debug)]
452#[stable(feature = "mpsc_recv_timeout", since = "1.12.0")]
453pub enum RecvTimeoutError {
454 /// This **channel** is currently empty, but the **Sender**(s) have not yet
455 /// disconnected, so data may yet become available.
456 #[stable(feature = "mpsc_recv_timeout", since = "1.12.0")]
457 Timeout,
458 /// The **channel**'s sending half has become disconnected, and there will
459 /// never be any more data received on it.
460 #[stable(feature = "mpsc_recv_timeout", since = "1.12.0")]
461 Disconnected,
462}
463
464/// This enumeration is the list of the possible error outcomes for the
465/// [`try_send`] method.
466///
467/// [`try_send`]: SyncSender::try_send
468#[stable(feature = "rust1", since = "1.0.0")]
469#[derive(PartialEq, Eq, Clone, Copy)]
470pub enum TrySendError<T> {
471 /// The data could not be sent on the [`sync_channel`] because it would require that
472 /// the callee block to send the data.
473 ///
474 /// If this is a buffered channel, then the buffer is full at this time. If
475 /// this is not a buffered channel, then there is no [`Receiver`] available to
476 /// acquire the data.
477 #[stable(feature = "rust1", since = "1.0.0")]
478 Full(#[stable(feature = "rust1", since = "1.0.0")] T),
479
480 /// This [`sync_channel`]'s receiving half has disconnected, so the data could not be
481 /// sent. The data is returned back to the callee in this case.
482 #[stable(feature = "rust1", since = "1.0.0")]
483 Disconnected(#[stable(feature = "rust1", since = "1.0.0")] T),
484}
485
486/// Creates a new asynchronous channel, returning the sender/receiver halves.
487/// All data sent on the [`Sender`] will become available on the [`Receiver`] in
488/// the same order as it was sent, and no [`send`] will block the calling thread
489/// (this channel has an "infinite buffer", unlike [`sync_channel`], which will
490/// block after its buffer limit is reached). [`recv`] will block until a message
491/// is available while there is at least one [`Sender`] alive (including clones).
492///
493/// The [`Sender`] can be cloned to [`send`] to the same channel multiple times, but
494/// only one [`Receiver`] is supported.
495///
496/// If the [`Receiver`] is disconnected while trying to [`send`] with the
497/// [`Sender`], the [`send`] method will return a [`SendError`]. Similarly, if the
498/// [`Sender`] is disconnected while trying to [`recv`], the [`recv`] method will
499/// return a [`RecvError`].
500///
501/// [`send`]: Sender::send
502/// [`recv`]: Receiver::recv
503///
504/// # Examples
505///
506/// ```
507/// use std::sync::mpsc::channel;
508/// use std::thread;
509///
510/// let (sender, receiver) = channel();
511///
512/// // Spawn off an expensive computation
513/// thread::spawn(move|| {
514/// # fn expensive_computation() {}
515/// sender.send(expensive_computation()).unwrap();
516/// });
517///
518/// // Do some useful work for awhile
519///
520/// // Let's see what that answer was
521/// println!("{:?}", receiver.recv().unwrap());
522/// ```
523#[must_use]
524#[stable(feature = "rust1", since = "1.0.0")]
525pub fn channel<T>() -> (Sender<T>, Receiver<T>) {
526 let (tx: Sender, rx: Receiver) = mpmc::channel();
527 (Sender { inner: tx }, Receiver { inner: rx })
528}
529
530/// Creates a new synchronous, bounded channel.
531/// All data sent on the [`SyncSender`] will become available on the [`Receiver`]
532/// in the same order as it was sent. Like asynchronous [`channel`]s, the
533/// [`Receiver`] will block until a message becomes available. `sync_channel`
534/// differs greatly in the semantics of the sender, however.
535///
536/// This channel has an internal buffer on which messages will be queued.
537/// `bound` specifies the buffer size. When the internal buffer becomes full,
538/// future sends will *block* waiting for the buffer to open up. Note that a
539/// buffer size of 0 is valid, in which case this becomes "rendezvous channel"
540/// where each [`send`] will not return until a [`recv`] is paired with it.
541///
542/// The [`SyncSender`] can be cloned to [`send`] to the same channel multiple
543/// times, but only one [`Receiver`] is supported.
544///
545/// Like asynchronous channels, if the [`Receiver`] is disconnected while trying
546/// to [`send`] with the [`SyncSender`], the [`send`] method will return a
547/// [`SendError`]. Similarly, If the [`SyncSender`] is disconnected while trying
548/// to [`recv`], the [`recv`] method will return a [`RecvError`].
549///
550/// [`send`]: SyncSender::send
551/// [`recv`]: Receiver::recv
552///
553/// # Examples
554///
555/// ```
556/// use std::sync::mpsc::sync_channel;
557/// use std::thread;
558///
559/// let (sender, receiver) = sync_channel(1);
560///
561/// // this returns immediately
562/// sender.send(1).unwrap();
563///
564/// thread::spawn(move|| {
565/// // this will block until the previous message has been received
566/// sender.send(2).unwrap();
567/// });
568///
569/// assert_eq!(receiver.recv().unwrap(), 1);
570/// assert_eq!(receiver.recv().unwrap(), 2);
571/// ```
572#[must_use]
573#[stable(feature = "rust1", since = "1.0.0")]
574pub fn sync_channel<T>(bound: usize) -> (SyncSender<T>, Receiver<T>) {
575 let (tx: Sender, rx: Receiver) = mpmc::sync_channel(cap:bound);
576 (SyncSender { inner: tx }, Receiver { inner: rx })
577}
578
579////////////////////////////////////////////////////////////////////////////////
580// Sender
581////////////////////////////////////////////////////////////////////////////////
582
583impl<T> Sender<T> {
584 /// Attempts to send a value on this channel, returning it back if it could
585 /// not be sent.
586 ///
587 /// A successful send occurs when it is determined that the other end of
588 /// the channel has not hung up already. An unsuccessful send would be one
589 /// where the corresponding receiver has already been deallocated. Note
590 /// that a return value of [`Err`] means that the data will never be
591 /// received, but a return value of [`Ok`] does *not* mean that the data
592 /// will be received. It is possible for the corresponding receiver to
593 /// hang up immediately after this function returns [`Ok`].
594 ///
595 /// This method will never block the current thread.
596 ///
597 /// # Examples
598 ///
599 /// ```
600 /// use std::sync::mpsc::channel;
601 ///
602 /// let (tx, rx) = channel();
603 ///
604 /// // This send is always successful
605 /// tx.send(1).unwrap();
606 ///
607 /// // This send will fail because the receiver is gone
608 /// drop(rx);
609 /// assert_eq!(tx.send(1).unwrap_err().0, 1);
610 /// ```
611 #[stable(feature = "rust1", since = "1.0.0")]
612 pub fn send(&self, t: T) -> Result<(), SendError<T>> {
613 self.inner.send(t)
614 }
615}
616
617#[stable(feature = "rust1", since = "1.0.0")]
618impl<T> Clone for Sender<T> {
619 /// Clone a sender to send to other threads.
620 ///
621 /// Note, be aware of the lifetime of the sender because all senders
622 /// (including the original) need to be dropped in order for
623 /// [`Receiver::recv`] to stop blocking.
624 fn clone(&self) -> Sender<T> {
625 Sender { inner: self.inner.clone() }
626 }
627}
628
629#[stable(feature = "mpsc_debug", since = "1.8.0")]
630impl<T> fmt::Debug for Sender<T> {
631 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
632 f.debug_struct(name:"Sender").finish_non_exhaustive()
633 }
634}
635
636////////////////////////////////////////////////////////////////////////////////
637// SyncSender
638////////////////////////////////////////////////////////////////////////////////
639
640impl<T> SyncSender<T> {
641 /// Sends a value on this synchronous channel.
642 ///
643 /// This function will *block* until space in the internal buffer becomes
644 /// available or a receiver is available to hand off the message to.
645 ///
646 /// Note that a successful send does *not* guarantee that the receiver will
647 /// ever see the data if there is a buffer on this channel. Items may be
648 /// enqueued in the internal buffer for the receiver to receive at a later
649 /// time. If the buffer size is 0, however, the channel becomes a rendezvous
650 /// channel and it guarantees that the receiver has indeed received
651 /// the data if this function returns success.
652 ///
653 /// This function will never panic, but it may return [`Err`] if the
654 /// [`Receiver`] has disconnected and is no longer able to receive
655 /// information.
656 ///
657 /// # Examples
658 ///
659 /// ```rust
660 /// use std::sync::mpsc::sync_channel;
661 /// use std::thread;
662 ///
663 /// // Create a rendezvous sync_channel with buffer size 0
664 /// let (sync_sender, receiver) = sync_channel(0);
665 ///
666 /// thread::spawn(move || {
667 /// println!("sending message...");
668 /// sync_sender.send(1).unwrap();
669 /// // Thread is now blocked until the message is received
670 ///
671 /// println!("...message received!");
672 /// });
673 ///
674 /// let msg = receiver.recv().unwrap();
675 /// assert_eq!(1, msg);
676 /// ```
677 #[stable(feature = "rust1", since = "1.0.0")]
678 pub fn send(&self, t: T) -> Result<(), SendError<T>> {
679 self.inner.send(t)
680 }
681
682 /// Attempts to send a value on this channel without blocking.
683 ///
684 /// This method differs from [`send`] by returning immediately if the
685 /// channel's buffer is full or no receiver is waiting to acquire some
686 /// data. Compared with [`send`], this function has two failure cases
687 /// instead of one (one for disconnection, one for a full buffer).
688 ///
689 /// See [`send`] for notes about guarantees of whether the
690 /// receiver has received the data or not if this function is successful.
691 ///
692 /// [`send`]: Self::send
693 ///
694 /// # Examples
695 ///
696 /// ```rust
697 /// use std::sync::mpsc::sync_channel;
698 /// use std::thread;
699 ///
700 /// // Create a sync_channel with buffer size 1
701 /// let (sync_sender, receiver) = sync_channel(1);
702 /// let sync_sender2 = sync_sender.clone();
703 ///
704 /// // First thread owns sync_sender
705 /// thread::spawn(move || {
706 /// sync_sender.send(1).unwrap();
707 /// sync_sender.send(2).unwrap();
708 /// // Thread blocked
709 /// });
710 ///
711 /// // Second thread owns sync_sender2
712 /// thread::spawn(move || {
713 /// // This will return an error and send
714 /// // no message if the buffer is full
715 /// let _ = sync_sender2.try_send(3);
716 /// });
717 ///
718 /// let mut msg;
719 /// msg = receiver.recv().unwrap();
720 /// println!("message {msg} received");
721 ///
722 /// msg = receiver.recv().unwrap();
723 /// println!("message {msg} received");
724 ///
725 /// // Third message may have never been sent
726 /// match receiver.try_recv() {
727 /// Ok(msg) => println!("message {msg} received"),
728 /// Err(_) => println!("the third message was never sent"),
729 /// }
730 /// ```
731 #[stable(feature = "rust1", since = "1.0.0")]
732 pub fn try_send(&self, t: T) -> Result<(), TrySendError<T>> {
733 self.inner.try_send(t)
734 }
735
736 // Attempts to send for a value on this receiver, returning an error if the
737 // corresponding channel has hung up, or if it waits more than `timeout`.
738 //
739 // This method is currently private and only used for tests.
740 #[allow(unused)]
741 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(name:"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(name:"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(name:"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 #[allow(deprecated)]
1113 fn description(&self) -> &str {
1114 "sending on a closed channel"
1115 }
1116}
1117
1118#[stable(feature = "rust1", since = "1.0.0")]
1119impl<T> fmt::Debug for TrySendError<T> {
1120 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1121 match *self {
1122 TrySendError::Full(..) => "Full(..)".fmt(f),
1123 TrySendError::Disconnected(..) => "Disconnected(..)".fmt(f),
1124 }
1125 }
1126}
1127
1128#[stable(feature = "rust1", since = "1.0.0")]
1129impl<T> fmt::Display for TrySendError<T> {
1130 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1131 match *self {
1132 TrySendError::Full(..) => "sending on a full channel".fmt(f),
1133 TrySendError::Disconnected(..) => "sending on a closed channel".fmt(f),
1134 }
1135 }
1136}
1137
1138#[stable(feature = "rust1", since = "1.0.0")]
1139impl<T> error::Error for TrySendError<T> {
1140 #[allow(deprecated)]
1141 fn description(&self) -> &str {
1142 match *self {
1143 TrySendError::Full(..) => "sending on a full channel",
1144 TrySendError::Disconnected(..) => "sending on a closed channel",
1145 }
1146 }
1147}
1148
1149#[stable(feature = "mpsc_error_conversions", since = "1.24.0")]
1150impl<T> From<SendError<T>> for TrySendError<T> {
1151 /// Converts a `SendError<T>` into a `TrySendError<T>`.
1152 ///
1153 /// This conversion always returns a `TrySendError::Disconnected` containing the data in the `SendError<T>`.
1154 ///
1155 /// No data is allocated on the heap.
1156 fn from(err: SendError<T>) -> TrySendError<T> {
1157 match err {
1158 SendError(t: T) => TrySendError::Disconnected(t),
1159 }
1160 }
1161}
1162
1163#[stable(feature = "rust1", since = "1.0.0")]
1164impl fmt::Display for RecvError {
1165 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1166 "receiving on a closed channel".fmt(f)
1167 }
1168}
1169
1170#[stable(feature = "rust1", since = "1.0.0")]
1171impl error::Error for RecvError {
1172 #[allow(deprecated)]
1173 fn description(&self) -> &str {
1174 "receiving on a closed channel"
1175 }
1176}
1177
1178#[stable(feature = "rust1", since = "1.0.0")]
1179impl fmt::Display for TryRecvError {
1180 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1181 match *self {
1182 TryRecvError::Empty => "receiving on an empty channel".fmt(f),
1183 TryRecvError::Disconnected => "receiving on a closed channel".fmt(f),
1184 }
1185 }
1186}
1187
1188#[stable(feature = "rust1", since = "1.0.0")]
1189impl error::Error for TryRecvError {
1190 #[allow(deprecated)]
1191 fn description(&self) -> &str {
1192 match *self {
1193 TryRecvError::Empty => "receiving on an empty channel",
1194 TryRecvError::Disconnected => "receiving on a closed channel",
1195 }
1196 }
1197}
1198
1199#[stable(feature = "mpsc_error_conversions", since = "1.24.0")]
1200impl From<RecvError> for TryRecvError {
1201 /// Converts a `RecvError` into a `TryRecvError`.
1202 ///
1203 /// This conversion always returns `TryRecvError::Disconnected`.
1204 ///
1205 /// No data is allocated on the heap.
1206 fn from(err: RecvError) -> TryRecvError {
1207 match err {
1208 RecvError => TryRecvError::Disconnected,
1209 }
1210 }
1211}
1212
1213#[stable(feature = "mpsc_recv_timeout_error", since = "1.15.0")]
1214impl fmt::Display for RecvTimeoutError {
1215 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1216 match *self {
1217 RecvTimeoutError::Timeout => "timed out waiting on channel".fmt(f),
1218 RecvTimeoutError::Disconnected => "channel is empty and sending half is closed".fmt(f),
1219 }
1220 }
1221}
1222
1223#[stable(feature = "mpsc_recv_timeout_error", since = "1.15.0")]
1224impl error::Error for RecvTimeoutError {
1225 #[allow(deprecated)]
1226 fn description(&self) -> &str {
1227 match *self {
1228 RecvTimeoutError::Timeout => "timed out waiting on channel",
1229 RecvTimeoutError::Disconnected => "channel is empty and sending half is closed",
1230 }
1231 }
1232}
1233
1234#[stable(feature = "mpsc_error_conversions", since = "1.24.0")]
1235impl From<RecvError> for RecvTimeoutError {
1236 /// Converts a `RecvError` into a `RecvTimeoutError`.
1237 ///
1238 /// This conversion always returns `RecvTimeoutError::Disconnected`.
1239 ///
1240 /// No data is allocated on the heap.
1241 fn from(err: RecvError) -> RecvTimeoutError {
1242 match err {
1243 RecvError => RecvTimeoutError::Disconnected,
1244 }
1245 }
1246}
1247