Which channel
| Channel | For |
|---|---|
mpsc::channel(n) | Many senders, one receiver, at most n messages waiting: a send waits for room, so a fast sender cannot run away. |
mpsc::unbounded_channel() | The same with no bound: a send never waits. |
oneshot::channel() | One value, once: a reply to a request. |
broadcast::channel(n) | Every receiver sees every message; one that falls n behind is told how many it missed. |
watch::channel(v) | One value many tasks read, and the news that it changed. |
local::mpsc::channel(n), local::oneshot::channel() | mpsc and oneshot for the tasks of one shard: not Send, with no atomic instruction on any path (Sync). |
mesh::mesh(n, capacity) | A bounded mpsc channel between every pair of n parties, such as one party per worker (below). |
Like everything in nuclei::sync, they work under any executor, and their receives are safe to drop halfway. On a Nuclei shard a receive that completes spends one operation of the task's budget, so a loop on a channel that is never empty still lets the shard's other tasks run.
mpsc
Senders clone; the channel closes when the last one drops, and recv then gives None. recv_many takes up to a limit at once; the receiver's close stops new sends and lets it drain what is there.
use nuclei::sync::mpsc;
#[nuclei::main]
async fn main() {
// Bounded: a send waits while the channel holds 16 messages.
let (tx, mut rx) = mpsc::channel::<String>(16);
for worker in 0..3 {
let tx = tx.clone();
nuclei::spawn(async move {
tx.send(format!("hello from {worker}")).await.unwrap();
})
.detach();
}
// The channel closes when the last sender drops, and recv gives None.
drop(tx);
let mut got = Vec::new();
while let Some(message) = rx.recv().await {
got.push(message);
}
got.sort();
assert_eq!(got, ["hello from 0", "hello from 1", "hello from 2"]);
}Permits, timeouts, weak senders
reserve takes a slot before the value exists, so the send that uses the permit cannot fail; reserve_many takes several, reserve_owned one that moves. try_send never waits, and send_timeout gives up after a while and gives the value back. A weak sender does not keep the channel open.
use std::time::Duration;
use nuclei::sync::mpsc::{self, error::TrySendError};
#[nuclei::main]
async fn main() {
let (tx, mut rx) = mpsc::channel::<u32>(1);
// reserve takes a slot first; the send that uses it cannot fail.
let permit = tx.reserve().await.unwrap();
permit.send(1);
// The one slot is taken: try_send says so instead of waiting.
assert!(matches!(tx.try_send(2), Err(TrySendError::Full(2))));
// send_timeout gives up after a while, and gives the value back.
let late = tx.send_timeout(3, Duration::from_millis(10)).await;
assert_eq!(late.unwrap_err().into_inner(), 3);
assert_eq!(rx.recv().await, Some(1));
// A weak sender does not keep the channel open.
let weak = tx.downgrade();
drop(tx);
assert!(weak.upgrade().is_none());
assert_eq!(rx.recv().await, None);
}oneshot
use nuclei::sync::oneshot;
#[nuclei::main]
async fn main() {
// One value, once: a reply to a request.
let (reply, answer) = oneshot::channel::<u64>();
nuclei::spawn(async move {
let _ = reply.send(6 * 7);
})
.detach();
assert_eq!(answer.await.unwrap(), 42);
// A sender dropped before it sent: the receiver gets an error.
let (reply, answer) = oneshot::channel::<u64>();
drop(reply);
assert!(answer.await.is_err());
}The sender can wait for the receiver to go away (closed), to stop work nobody will read.
broadcast
use nuclei::sync::broadcast::{self, error::RecvError};
#[nuclei::main]
async fn main() {
// Every receiver sees every message sent after it subscribed.
let (tx, mut first) = broadcast::channel::<&str>(2);
let mut second = tx.subscribe();
tx.send("deploy").unwrap();
assert_eq!(first.recv().await, Ok("deploy"));
assert_eq!(second.recv().await, Ok("deploy"));
// A receiver that falls more than the capacity behind is told how far.
for event in ["a", "b", "c"] {
tx.send(event).unwrap();
}
assert_eq!(first.recv().await, Err(RecvError::Lagged(1)));
assert_eq!(first.recv().await, Ok("b"));
// A weak sender: it can send again once upgraded, while a strong one lives.
let weak = tx.downgrade();
assert!(weak.upgrade().is_some());
drop(tx);
assert!(weak.upgrade().is_none());
}subscribe on the sender, and resubscribe on a receiver, add a receiver that takes what is sent from then on. A WeakSender sends only once upgraded, and does not keep the channel open.
watch
use nuclei::sync::watch;
#[nuclei::main]
async fn main() {
// One value, read by many; each receiver sees the latest.
let (tx, mut rx) = watch::channel(1u32);
let watcher = nuclei::spawn(async move {
// changed waits for a value this receiver has not seen.
rx.changed().await.unwrap();
*rx.borrow_and_update()
});
tx.send_modify(|v| *v += 1);
assert_eq!(watcher.await, 2);
assert_eq!(*tx.borrow(), 2);
}borrow reads the current value; changed waits for one this receiver has not seen, and wait_for for one that passes a test. send_modify changes the value in place, and send_if_modified tells the receivers only when the change says so.
A mesh between workers
mesh(n, capacity) joins n parties, each ordered pair by a bounded mpsc channel of capacity messages, and gives each its Party: its senders to every party and its receivers from every party, indexed by party. A party may send to itself, so a send to whoever owns a key needs no special case. The mesh only builds channels: put each party in a task of its own, here one per worker with spawn_on_worker.
use nuclei::RuntimeHandle;
use nuclei::sync::mesh::mesh;
#[nuclei::main(threads = 3)]
async fn main() -> std::io::Result<()> {
let runtime = RuntimeHandle::current();
// One party per worker, each pair joined by a channel of 8 messages each way.
let parties = mesh::<usize>(runtime.workers(), 8);
let mut tasks = Vec::new();
for (worker, mut party) in parties.into_iter().enumerate() {
tasks.push(runtime.spawn_on_worker(worker, async move {
// Every party sends its index to every party, itself included.
for to in 0..party.parties() {
party.send_to(to, party.index()).await.unwrap();
}
// recv ends once every party, this one too, has closed its senders.
party.close_senders();
let mut heard = Vec::new();
while let Some((from, index)) = party.recv().await {
assert_eq!(from, index);
heard.push(from);
}
heard.sort_unstable();
heard
})?);
}
for task in tasks {
assert_eq!(task.await, [0, 1, 2]);
}
Ok(())
}send_to(to, value)waits for room on the channel totoalone: a full channel holds up no other pair.try_send_tonever waits.recv()takes from any party, with its index, round robin from the one after the last that delivered, so a party that floods cannot starve a quiet one.recv_from(from)takes from one party.recv()ends once every channel into the party has ended. The party's sender to itself keeps one open, so a party callsclose_senderswhen it is done sending.nparties maken * nchannels.
Cancellation tokens
A CancellationToken tells any number of tasks to stop. Clones share one state; a child token is cancelled with its parent, and on its own without touching the parent.
use std::time::Duration;
use nuclei::sync::CancellationToken;
use nuclei::time::sleep;
#[nuclei::main]
async fn main() {
let shutdown = CancellationToken::new();
// A child token is cancelled with its parent, and alone too.
let worker_token = shutdown.child_token();
let worker = nuclei::spawn(async move {
// The work, or None once the token is cancelled first.
worker_token
.run_until_cancelled(sleep(Duration::from_secs(60)))
.await
});
shutdown.cancel();
assert_eq!(worker.await, None);
assert!(shutdown.is_cancelled());
// A drop guard cancels its token when it drops, on every way out of a scope.
let token = CancellationToken::new();
{
let _guard = token.clone().drop_guard();
}
token.cancelled().await;
}cancelled().awaitwaits for it;is_cancelled()checks.run_until_cancelled(future)givesSome(output), orNoneonce the token is cancelled first, dropping the future.drop_guard()cancels the token when the guard drops, however the scope ends.
A clean shutdown puts it to work.