Struct Sender 

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

Sends values to the associated Receiver.

Created by channel. Cloning it is a reference-count increment.

Progress

Wait-free (a bounded number of the caller’s own atomic steps whatever other threads do): is_closed, capacity, max_capacity, same_channel, strong_count, weak_count, downgrade, clone and drop. The drop of the last sender ends with a wake of the receiver’s waker cell, which is two atomic steps; what the waker then does belongs to the executor.

Lock-free, not wait-free: try_send, the try_reserve family, and every poll of send and the reserve family. A permit is taken with a compare-and-swap that retries when another thread changed the semaphore first; a wait that has to queue pushes a node on the semaphore’s lock-free arrival stack and may serve the queue (one pass, plus one for each operation of another thread that lands meanwhile); and the push of a value onto the message queue retries when it loses a slot to another sender and allocates a segment when the last one is full. Some thread’s call completes whenever any retries. closed is lock-free for the same reason.

A waiting reservation is granted by whichever thread serves the queue once its slots are free (the receiver taking a value, a permit dropped), so a thread the scheduler stops while it serves the queue delays that grant until it runs again; for that moment a try_ attempt fails as well, even if the last waiting reservation has just left. No call waits for it.

blocking_send parks the calling thread until there is room.

Implementations

impl<T> Sender<T>

pub fn send( &self, value: T, ) -> impl Future<Output = Result<(), SendError<T>>> + '_

Sends a value, waiting until there is capacity.

A successful send means the other end had not hung up when the value was published. Ok does not mean the value will be received: the receiver can hang up right after.

Errors

If the receiving half is closed, by Receiver::close or by being dropped, the function returns an error that holds the value passed to send.

Cancel safety

If the returned future is dropped before it completes, the message was not sent, and the value is dropped. To keep the value, reserve capacity with reserve and send through the Permit.

Waiting sends and reservations are served in the order they began to wait. Dropping the future gives up its place, and a slot it was handed meanwhile goes on to the next one.

use nuclei::sync::mpsc;

let (tx, mut rx) = mpsc::channel(1);
now(tx.send(1)).unwrap();
assert_eq!(rx.try_recv(), Ok(1));

drop(rx);
let err = now(tx.send(2)).unwrap_err();
assert_eq!(err.0, 2);

pub async fn send_timeout( &self, value: T, timeout: Duration, ) -> Result<(), SendTimeoutError<T>>

Sends a value, waiting for room at most timeout.

As send, with a deadline: the value comes back in SendTimeoutError::Timeout if the channel stayed full until then, and in SendTimeoutError::Closed if the receiver is gone. The wait is on nuclei’s timers: on a thread that runs no nuclei shard, the default runtime fires it.

Cancel safety

As send: a dropped future sent nothing, and drops its value.

use std::time::Duration;
use nuclei::sync::mpsc::{self, error::SendTimeoutError};

nuclei::runtime::block_on(async {
    let (tx, _rx) = mpsc::channel(1);
    tx.send_timeout(1, Duration::from_millis(10)).await.unwrap();
    let full = tx.send_timeout(2, Duration::from_millis(10)).await;
    assert!(matches!(full, Err(SendTimeoutError::Timeout(2))));
});

pub async fn closed(&self)

Completes when the receiver has been closed or dropped.

This lets a producer learn that nobody wants its values any more and stop work early.

Cancel safety

Cancel safe. Once the channel is closed it stays closed, and every later call completes at once.

use nuclei::sync::mpsc;

let (tx, rx) = mpsc::channel::<()>(1);
let mut cx = Context::from_waker(Waker::noop());
let mut closed = pin!(tx.closed());
assert_eq!(closed.as_mut().poll(&mut cx), Poll::Pending);
drop(rx);
assert_eq!(closed.as_mut().poll(&mut cx), Poll::Ready(()));

pub fn try_send(&self, message: T) -> Result<(), TrySendError<T>>

Sends a value at once, or says why it cannot be sent.

Unlike send it never waits: a full buffer is an error. It never takes a slot ahead of a send or reservation that is waiting.

Errors

TrySendError::Full if no slot is free or a send or reservation is waiting, TrySendError::Closed if the receiver closed or was dropped. Both hold the value back.

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

let (tx, mut rx) = mpsc::channel(1);
tx.try_send("first").unwrap();
assert!(matches!(tx.try_send("second"), Err(TrySendError::Full("second"))));
assert_eq!(rx.try_recv(), Ok("first"));

pub fn blocking_send(&self, value: T) -> Result<(), SendError<T>>

Sends a value from a thread that may block.

It parks the calling thread until there is room. 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 receiver uses Receiver::blocking_recv.

Panics

On wasm32, where no other thread can make room, when the channel is full: the call could never end.

use nuclei::sync::mpsc;

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

pub fn is_closed(&self) -> bool

Whether the channel is closed: the receiver was closed or dropped.

use nuclei::sync::mpsc;

let (tx, rx) = mpsc::channel::<()>(42);
let tx2 = tx.clone();
assert!(!tx.is_closed());
drop(rx);
assert!(tx.is_closed() && tx2.is_closed());

pub async fn reserve(&self) -> Result<Permit<'_, T>, SendError<()>>

Waits for capacity for one message and reserves it.

A Permit tracks the reserved slot; Permit::send uses it, and dropping the permit unused gives the slot back.

Errors

A SendError if the receiver is closed or dropped, now or while waiting.

Cancel safety

Dropping the future gives up its place in the queue of waiters, and a slot it was handed meanwhile goes on to the next one. No slot is held by a dropped future.

use nuclei::sync::mpsc;

let (tx, mut rx) = mpsc::channel(1);
let permit = now(tx.reserve()).unwrap();
assert!(tx.try_send(123).is_err(), "the only slot is reserved");
permit.send(456);
assert_eq!(rx.try_recv(), Ok(456));

pub async fn reserve_many( &self, n: usize, ) -> Result<PermitIterator<'_, T>, SendError<()>>

Waits for capacity for n messages and reserves them.

The returned PermitIterator yields n permits; dropping it before it is exhausted gives the rest back. Waiting requests are served in the order they began to wait, whatever each asks for: the slots that come free are set aside for the oldest one until it has all n, and nothing behind it is served first. Slots set aside for a request do not count in capacity.

Errors

A SendError if the receiver is closed or dropped, now or while waiting, or if n is more than the channel can ever hold (max_capacity). n == 0 succeeds with an empty iterator unless the channel is closed.

Cancel safety

As for reserve: the slots set aside for a dropped request go on to the next one.

use nuclei::sync::mpsc;

let (tx, mut rx) = mpsc::channel(2);
let mut permits = now(tx.reserve_many(2)).unwrap();
assert!(tx.try_send(123).is_err());
permits.next().unwrap().send(456);
permits.next().unwrap().send(457);
assert!(permits.next().is_none());
assert_eq!(rx.try_recv(), Ok(456));
assert_eq!(rx.try_recv(), Ok(457));

pub async fn reserve_owned(self) -> Result<OwnedPermit<T>, SendError<()>>

Waits for capacity for one message, moving the sender into the returned OwnedPermit.

Unlike reserve, the permit borrows nothing, so it can be stored or moved into another task. Cloning a sender first keeps a handle to send with meanwhile. OwnedPermit::send and OwnedPermit::release hand the sender back.

Errors

A SendError if the receiver is closed or dropped, now or while waiting.

Cancel safety

Dropping the future gives up its place in the queue of waiters, passes on a slot it was handed meanwhile, and drops the sender it holds.

use nuclei::sync::mpsc;

let (tx, mut rx) = mpsc::channel(1);
let permit = now(tx.reserve_owned()).unwrap();
let tx = permit.send(123);
assert_eq!(rx.try_recv(), Ok(123));
tx.try_send(456).unwrap();

pub fn try_reserve(&self) -> Result<Permit<'_, T>, TrySendError<()>>

Reserves capacity for one message if there is any, without waiting. It never takes a slot ahead of a send or reservation that is waiting.

Errors

TrySendError::Full when no slot is free or a send or reservation is waiting, TrySendError::Closed when the receiver closed or was dropped.

use nuclei::sync::mpsc;

let (tx, mut rx) = mpsc::channel(1);
let permit = tx.try_reserve().unwrap();
assert!(tx.try_reserve().is_err(), "no slot is left");
permit.send("foo");
assert_eq!(rx.try_recv(), Ok("foo"));

pub fn try_reserve_many( &self, n: usize, ) -> Result<PermitIterator<'_, T>, TrySendError<()>>

Reserves capacity for n messages if there is enough, without waiting.

Errors

TrySendError::Full when fewer than n slots are free, a send or reservation is waiting, or n is more than the channel can ever hold; TrySendError::Closed when the receiver closed or was dropped. n == 0 succeeds with an empty iterator unless the channel is closed.

use nuclei::sync::mpsc;

let (tx, mut rx) = mpsc::channel(2);
let mut permits = tx.try_reserve_many(2).unwrap();
assert!(tx.try_send(123).is_err());
permits.next().unwrap().send(456);
permits.next().unwrap().send(457);
assert_eq!(rx.try_recv(), Ok(456));
assert_eq!(rx.try_recv(), Ok(457));

assert!(tx.try_reserve_many(3).is_err(), "more than the channel holds");

pub fn try_reserve_owned(self) -> Result<OwnedPermit<T>, TrySendError<Self>>

Reserves capacity for one message if there is any, moving the sender into the returned OwnedPermit.

Errors

On failure the sender is handed back inside the error: TrySendError::Full when no slot is free or a send or reservation is waiting, TrySendError::Closed when the receiver closed or was dropped.

use nuclei::sync::mpsc;

let (tx, mut rx) = mpsc::channel(1);
let permit = tx.clone().try_reserve_owned().unwrap();
assert!(tx.try_send(123).is_err());
permit.send(456);
assert_eq!(rx.try_recv(), Ok(456));

pub fn same_channel(&self, other: &Self) -> bool

Whether two senders belong to the same channel.

use nuclei::sync::mpsc;

let (tx, _rx) = mpsc::channel::<()>(1);
let tx2 = tx.clone();
assert!(tx.same_channel(&tx2));

let (tx3, _rx3) = mpsc::channel::<()>(1);
assert!(!tx3.same_channel(&tx2));

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 is reserved, and up when the receiver takes a value or a permit is dropped unused. A slot set aside for a waiting reserve_many is not free. Compare max_capacity, which never changes.

use nuclei::sync::mpsc;

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

pub fn downgrade(&self) -> WeakSender<T>

Converts the sender into a WeakSender, which does not keep the channel open. The sender itself stays.

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!(tx.max_capacity(), 5);
assert_eq!(tx.capacity(), 4);

pub fn strong_count(&self) -> usize

How many Sender handles exist (an OwnedPermit holds one).

pub fn weak_count(&self) -> usize

How many WeakSender handles exist.

Trait Implementations

impl<T> Clone for Sender<T>

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl<T> Debug for Sender<T>

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

Formats the value using the given formatter.

impl<T> Drop for Sender<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.