nuclei/sync/local/mpsc/
mod.rs

1//! Multi-producer, single-consumer channels for the tasks of one shard.
2//!
3//! The names, signatures and error types follow [`sync::mpsc`](crate::sync::mpsc),
4//! so code switches by changing the path. The state sits in one `Rc` and
5//! plain cells, so the halves are `!Send` and no operation touches an atomic.
6//!
7//! The bounded flavor ([`channel`]) holds at most a fixed number of messages
8//! and serves waiting sends strictly in arrival order; the unbounded one
9//! ([`unbounded_channel`]) never makes a send wait.
10//!
11//! What one sender sends comes out in the order it was sent. When every
12//! sender is dropped the stream ends: once the buffered values are received,
13//! `recv` returns `None`. If the receiver is dropped, its unread values are
14//! dropped with it and every later send fails with an error that holds the
15//! value; [`Receiver::close`] first stops further sends and keeps the
16//! buffered values, and the slots reserved before it, usable.
17//!
18//! Left out, because a local channel never leaves its thread: the
19//! `blocking_*` methods (a blocking call there could only wait for itself)
20//! and `WeakSender`.
21//!
22//! Receives, and sends that complete at their first attempt, spend the
23//! cooperative budget (see [`task::coop`](crate::task::coop)) as the atomic
24//! channels do.
25
26/// The receiving methods both flavors share.
27macro_rules! receiver_methods {
28    () => {
29        /// Receives the next value, waiting for one if the channel is open
30        /// and empty. Answers `None` once the channel is closed and drained
31        /// (no sender, and no reserved slot still to be used).
32        ///
33        /// Cancel safe: a dropped future loses no value.
34        pub async fn recv(&mut self) -> Option<T> {
35            core::future::poll_fn(|cx| self.poll_recv(cx)).await
36        }
37
38        /// Receives up to `limit` values into `buf` and answers how many,
39        /// waiting for at least one. Answers 0 when `limit` is 0 or the
40        /// channel ended.
41        ///
42        /// Cancel safe: a dropped future loses no value.
43        pub async fn recv_many(&mut self, buf: &mut Vec<T>, limit: usize) -> usize {
44            core::future::poll_fn(|cx| self.poll_recv_many(cx, buf, limit)).await
45        }
46
47        /// Polls to receive the next value; the waker of `cx` is woken when
48        /// a value arrives or the channel ends.
49        pub fn poll_recv(
50            &mut self,
51            cx: &mut core::task::Context<'_>,
52        ) -> core::task::Poll<Option<T>> {
53            self.chan.poll_recv(cx)
54        }
55
56        /// Polls to receive up to `limit` values into `buf`; see
57        /// [`recv_many`](Self::recv_many).
58        pub fn poll_recv_many(
59            &mut self,
60            cx: &mut core::task::Context<'_>,
61            buf: &mut Vec<T>,
62            limit: usize,
63        ) -> core::task::Poll<usize> {
64            self.chan.poll_recv_many(cx, buf, limit)
65        }
66
67        /// Receives the next value without waiting.
68        ///
69        /// # Errors
70        ///
71        /// [`TryRecvError::Empty`](crate::sync::mpsc::error::TryRecvError::Empty)
72        /// if a value may yet come,
73        /// [`TryRecvError::Disconnected`](crate::sync::mpsc::error::TryRecvError::Disconnected)
74        /// if the channel ended.
75        pub fn try_recv(&mut self) -> Result<T, crate::sync::mpsc::error::TryRecvError> {
76            self.chan.try_recv()
77        }
78
79        /// Closes the receiving half without dropping it: no further value
80        /// can be sent, and sends that wait fail. The values already
81        /// buffered, and the slots reserved before the close, stay usable.
82        pub fn close(&mut self) {
83            self.chan.close_rx();
84        }
85
86        /// Whether the channel is closed: [`close`](Self::close) was called,
87        /// or every sender is gone. Buffered values may still be received.
88        pub fn is_closed(&self) -> bool {
89            self.chan.rx_is_closed()
90        }
91
92        /// Whether the channel holds no value.
93        pub fn is_empty(&self) -> bool {
94            self.chan.len() == 0
95        }
96
97        /// How many values the channel holds.
98        pub fn len(&self) -> usize {
99            self.chan.len()
100        }
101    };
102}
103
104mod bounded;
105mod chan;
106mod unbounded;
107
108pub use self::bounded::{Permit, Receiver, Sender, channel};
109pub use self::unbounded::{UnboundedReceiver, UnboundedSender, unbounded_channel};
nuclei/sync/local/mpsc/mod.rs

nuclei/sync/local/mpsc/
mod.rs
