Struct Receiver 

pub struct Receiver<T> { /* private fields */ }

Receives values from the associated Senders.

Created by channel. There is exactly one receiver.

Progress

Wait-free: is_closed, is_empty, len, capacity, max_capacity, sender_strong_count and sender_weak_count: each is a load.

Lock-free: recv, recv_many, try_recv and the two poll_ forms. Registering the waker is wait-free (the cell runs two read-modify-writes and no loop). Taking a value off the message queue is a compare-and-swap no other thread contends for, and never waits for a sender: a value whose send claimed its place and has not written it yet is not there, and its sender wakes the receiver once it is (see the Progress notes of the module). Giving the slot back to the senders is a semaphore release that, when senders wait, serves the queue and is lock-free. close is lock-free; the drop is the same as a receive per value left, since it drops every value still queued, and so takes time proportional to them.

blocking_recv and blocking_recv_many park the calling thread until a value arrives.

Implementations

impl<T> Receiver<T>

pub fn recv(&mut self) -> impl Future<Output = Option<T>> + '_

Receives the next value.

Returns None once the channel is closed and no buffered value is left, which means no value can ever be received again. The channel is closed when every sender has been dropped, or when close is called. If the buffer is empty and the channel is open, it waits for a send or a close.

If close was called while a Permit or an OwnedPermit was still held, the channel does not count as finished for recv until the permit is sent or dropped.

Cancel safety

Cancel safe. If the future is dropped (for instance when another branch of a select wins), no value was taken from the channel.

use nuclei::sync::mpsc;

let (tx, mut rx) = mpsc::channel(100);
tx.try_send("hello").unwrap();
tx.try_send("world").unwrap();
drop(tx);
assert_eq!(now(rx.recv()), Some("hello"));
assert_eq!(now(rx.recv()), Some("world"));
assert_eq!(now(rx.recv()), None);

pub async fn recv_many(&mut self, buffer: &mut Vec<T>, limit: usize) -> usize

Receives the next values and appends them to buffer, taking at most limit of them in one call.

Returns how many values it added. A limit of zero returns 0 at once. Otherwise it waits like recv until at least one value is there, then takes everything that is ready up to limit without waiting again, and returns 0 only when the channel is closed and empty. The capacity of buffer grows as needed.

Cancel safety

Cancel safe, as recv: a dropped future took nothing.

use nuclei::sync::mpsc;

let (tx, mut rx) = mpsc::channel(100);
for word in ["first", "second", "third"] {
    tx.try_send(word).unwrap();
}
let mut buffer = Vec::new();
assert_eq!(now(rx.recv_many(&mut buffer, 2)), 2);
assert_eq!(buffer, ["first", "second"]);
assert_eq!(now(rx.recv_many(&mut buffer, 2)), 1);
drop(tx);
assert_eq!(now(rx.recv_many(&mut buffer, 2)), 0);

pub fn try_recv(&mut self) -> Result<T, TryRecvError>

Tries to receive the next value without waiting.

Errors

TryRecvError::Empty if the channel is empty but a sender or a permit still exists, so a value may yet come. TryRecvError::Disconnected if it is empty and nothing can send: every sender is gone, or the channel was closed and no permit is held.

A value whose send is still in progress on another thread is not in the channel yet, nor are the values sent after it, which come out behind it. This call does not wait for them: the sender wakes the receiver once its value is in.

use nuclei::sync::mpsc::{self, error::TryRecvError};

let (tx, mut rx) = mpsc::channel(100);
tx.try_send("hello").unwrap();
assert_eq!(rx.try_recv(), Ok("hello"));
assert_eq!(rx.try_recv(), Err(TryRecvError::Empty));

tx.try_send("hello").unwrap();
drop(tx);
assert_eq!(rx.try_recv(), Ok("hello"));
assert_eq!(rx.try_recv(), Err(TryRecvError::Disconnected));

pub fn blocking_recv(&mut self) -> Option<T>

Receives the next value from a thread that may block.

Like recv, but it parks the calling thread while the channel is empty and open. On a nuclei shard (inside one of its tasks) it does not park the thread: it runs a nested block_on, so the shard keeps running its other tasks meanwhile. Inside another runtime’s task it blocks that task’s thread. It works whether or not the senders use Sender::blocking_send.

Panics

On wasm32, where no other thread can send, when the channel is empty and open: the call could never end.

use nuclei::sync::mpsc;

let (tx, mut rx) = mpsc::channel::<u8>(10);
let receiver = std::thread::spawn(move || {
    assert_eq!(rx.blocking_recv(), Some(10));
    assert_eq!(rx.blocking_recv(), None);
});
tx.blocking_send(10).unwrap();
drop(tx);
receiver.join().unwrap();

pub fn blocking_recv_many(&mut self, buffer: &mut Vec<T>, limit: usize) -> usize

recv_many for a thread that may block; see blocking_recv.

Panics

As for blocking_recv.

pub fn close(&mut self)

Closes the receiving half without dropping it.

No further value can be sent: a send fails and returns its value. A slot reserved before the close can still be used, and the values already buffered stay receivable, so a clean shutdown closes, then calls recv until it returns None. That happens once every reserved slot is sent or dropped.

use nuclei::sync::mpsc::{self, error::TrySendError};

let (tx, mut rx) = mpsc::channel(20);
tx.try_send(1).unwrap();
rx.close();
assert!(matches!(tx.try_send(2), Err(TrySendError::Closed(2))));
assert_eq!(rx.try_recv(), Ok(1), "the buffered value is still there");

pub fn is_closed(&self) -> bool

Whether the channel is closed: close was called, or every sender is gone. Buffered values may still be received.

use nuclei::sync::mpsc;

let (tx, mut rx) = mpsc::channel::<()>(10);
assert!(!rx.is_closed());
rx.close();
assert!(rx.is_closed());
drop(tx);

pub fn is_empty(&self) -> bool

Whether the channel holds no value.

use nuclei::sync::mpsc;

let (tx, rx) = mpsc::channel(10);
assert!(rx.is_empty());
tx.try_send(0).unwrap();
assert!(!rx.is_empty());

pub fn len(&self) -> usize

How many values the channel holds.

A snapshot: it is exact while no other thread is sending, and while one is it may be off by the values being sent at that moment.

use nuclei::sync::mpsc;

let (tx, rx) = mpsc::channel(10);
assert_eq!(rx.len(), 0);
tx.try_send(0).unwrap();
assert_eq!(rx.len(), 1);

pub fn capacity(&self) -> usize

The channel’s current capacity: the slots that are free.

It goes down when a value is sent or a slot reserved, and up when a value is received or a permit dropped unused. Compare max_capacity, which never changes.

use nuclei::sync::mpsc;

let (tx, mut rx) = mpsc::channel::<()>(5);
assert_eq!(rx.capacity(), 5);
let permit = tx.try_reserve().unwrap();
assert_eq!(rx.capacity(), 4);
assert_eq!(rx.len(), 0);
permit.send(());
assert_eq!(rx.len(), 1);
rx.try_recv().unwrap();
assert_eq!(rx.capacity(), 5);

pub fn max_capacity(&self) -> usize

The channel’s maximum capacity: the buffer it was created with.

use nuclei::sync::mpsc;

let (tx, rx) = mpsc::channel::<()>(5);
let _permit = tx.try_reserve().unwrap();
assert_eq!(rx.max_capacity(), 5);
assert_eq!(rx.capacity(), 4);

pub fn poll_recv(&mut self, cx: &mut Context<'_>) -> Poll<Option<T>>

Polls to receive the next value.

Returns Poll::Pending if no value is available and the channel is open; Poll::Ready(Some(value)) if one is; and Poll::Ready(None) if the channel is closed and every value sent before it closed was received.

On Pending, the waker of cx is woken when a value arrives or the channel closes. Only the waker of the most recent call to poll_recv or poll_recv_many is kept.

pub fn poll_recv_many( &mut self, cx: &mut Context<'_>, buffer: &mut Vec<T>, limit: usize, ) -> Poll<usize>

Polls to receive several values into buffer, at most limit.

Returns Poll::Pending if no value is available and the channel is open; Poll::Ready(count) with the number of values added, which is at least one unless limit is zero or the channel is closed and empty. It takes everything that is ready, up to limit, in one call. The waker of cx is handled as for poll_recv.

pub fn sender_strong_count(&self) -> usize

How many Sender handles exist.

pub fn sender_weak_count(&self) -> usize

How many WeakSender handles exist.

Trait Implementations

impl<T> Debug for Receiver<T>

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl<T> Drop for Receiver<T>

fn drop(&mut self)

Executes the destructor for this type.

fn pin_drop(self: Pin<&mut Self>)

🔬This is a nightly-only experimental API. (pin_ergonomics)

Execute the destructor for this type, but different to Drop::drop, it requires self to be pinned.