Concurrency Primitives

Datum's concurrency primitives are stream-native handles for shared state, producer/consumer handoff, and pub-sub broadcast. They are ordinary Datum sources on the consumer side, so they keep the same blueprint-vs-materialization contract as Source, Flow, and Sink: constructing a handle or source blueprint does not start a stream; materialization starts execution.

This page is the overview. Each primitive has a dedicated deep-dive guide, and the design rationale has its own concepts page:

• Signal & Subscription — the state cell: coalesced vs lossless change feeds, get() semantics, overflow policies, pitfalls, and integration patterns.
• Channel — bounded MPSC handoff: send/try_send, graceful close, and positioning vs SourceQueue/MergeHub.
• Topic — pub-sub broadcast: choosing an overflow policy, ordering, churn, and positioning vs BroadcastHub.
• Concurrency Design — why they are built this way: the two-plane architecture, the measured evidence, and the benchmark-first methodology.

Which primitive to use

Primitive	Use it for	Delivery contract	Backpressure shape
Signal<T>	Latest-value state such as health, config, progress, or leader state	Current value plus a coalesced changes() feed; slow subscribers see the newest value and may skip intermediate values	Writers are never backpressured by subscribers
Subscription<T>	State where every accepted change matters, such as offsets, audit states, or ordered state transitions	Current value plus a bounded every-change changes() feed	SubscriptionOverflow::Backpressure preserves every change; DropNew and Fail are explicit non-lossless policies
Channel<T>	Bounded MPSC handoff from many producers into one consumer stream	Every accepted item is delivered once to the single active consumer	send().await waits for capacity; try_send() returns immediately
Topic<T>	Many-publisher, many-subscriber broadcast	Subscribers see elements published after their registration, in one global order, subject to the overflow policy	Per-subscriber Backpressure, Sliding, or Dropping

Use Signal when "latest wins" is the contract. Use Subscription when the state cell is also a lossless change log. Use Channel when work should move from producers to one downstream consumer. Use Topic when every subscriber should receive the broadcast stream independently.

Two-plane design

All four primitives follow the M9 two-plane rule:

• The actor control plane owns lifecycle and registry decisions: subscribe, unsubscribe, close, terminal state, and published subscriber-table snapshots.
• The lock-free data plane carries hot elements or snapshots directly through ArcSwap, ArrayQueue, sequence rings, and wake/permit protocols.

That distinction is load-bearing. Datum uses Ractor where serialization is useful, but does not send an actor message per stream element on the hot path.

Signal: latest-value state

Signal<T> stores an immutable Arc<T> snapshot. get() returns an owned Arc<T> without an actor hop; get_cloned() clones the current value through the guarded ArcSwap load path. The changes() source starts with the current value and then emits coalesced updates.

use datum::{Signal, testkit::TestSink};

let signal = Signal::new("idle").unwrap();
let sink = signal.changes().run_with(TestSink::probe()).unwrap();

// A Signal subscriber starts with the current value.
sink.request(1);
sink.assert_next("idle");

// Slow subscribers see the newest value and may skip intermediate states.
signal.set("running").unwrap();
signal.set("draining").unwrap();
sink.request(1);
sink.assert_next("draining");

// Reads are synchronous snapshots, not actor messages.
assert_eq!(signal.get_cloned(), "draining");

// Closing re-emits the final value before completion.
signal.close_with("done").unwrap();
sink.request(2);
sink.assert_next("done");
sink.expect_complete();

Reach for set() / update() when you want synchronous publication and error reporting. The set_eventually() / update_eventually() names remain available for source compatibility; in the current two-plane implementation they also publish on the caller thread before returning.

Subscription: every accepted change

Subscription<T> has the same latest-value read side as Signal, but its changes() feed is a bounded sequence feed. Under SubscriptionOverflow::Backpressure, a producer waits outside the actor until active subscribers have capacity, preserving every accepted change.

use datum::{Subscription, SubscriptionOverflow, testkit::TestSink};

let subscription = Subscription::new(0_u64, 4, SubscriptionOverflow::Backpressure).unwrap();
let sink = subscription.changes().run_with(TestSink::probe()).unwrap();

// A Subscription also starts with the current value.
sink.request(1);
sink.assert_next(0);

// With Backpressure, accepted changes are delivered in order without gaps.
subscription.set(1).unwrap();
subscription.set(2).unwrap();
sink.request(2);
sink.assert_next_n([1, 2]);

subscription.close_with(3).unwrap();
sink.request(2);
sink.assert_next(3);
sink.expect_complete();

Overflow policies are explicit:

Policy	Behavior
SubscriptionOverflow::Backpressure	Preserve every change by waiting for subscriber capacity
SubscriptionOverflow::DropNew	Apply the state transition, but let full subscribers skip the new feed item
SubscriptionOverflow::Fail	Apply the state transition, fail full subscribers after accepted items drain, and return an error to the producer

Channel: MPSC handoff

Channel<T> is a closeable bounded many-producer, single-consumer source. Clone the handle for producers. send().await waits for capacity; try_send() returns TrySendError::Full(value) if the ring is full and TrySendError::Closed(value) after close. close() is graceful: buffered items drain before the source completes.

use datum::{Channel, Sink};

let channel = Channel::bounded(8);
let completion = channel.source().run_with(Sink::collect()).unwrap();

channel.try_send("red").unwrap();

let producer = channel.clone();
producer.try_send("blue").unwrap();

// close() is graceful: buffered elements drain before the source completes.
channel.close();

let mut items = completion.wait().unwrap();
items.sort_unstable();
assert_eq!(items, vec!["blue", "red"]);

The explicit style is Channel::bounded(capacity) plus channel.source(). The convenience style is Source::channel(capacity), which materializes the Channel<T> handle from the source blueprint:

use datum::prelude::*;

let (channel, completion) = Source::<u64>::channel(4)
    .to_mat(Sink::collect(), Keep::both)
    .run()
    .unwrap();

channel.try_send(10).unwrap();
channel.try_send(20).unwrap();
channel.close();

assert_eq!(completion.wait().unwrap(), vec![10, 20]);

Both styles create the same kind of channel. The convenience form is useful when the channel is part of a larger graph and you want the handle as the source's materialized value.

Topic: pub-sub broadcast

Topic<T> is a stream-native broadcast primitive. Each materialized subscriber receives elements published after its registration. Publishing with zero subscribers succeeds and drops the element, matching FS2 Topic semantics. Closing the topic lets current subscribers drain queued elements and then complete.

use datum::{Topic, TopicOverflow, testkit::TestSink};

let topic = Topic::new(8, TopicOverflow::Backpressure).unwrap();
let left = topic.subscribe().run_with(TestSink::probe()).unwrap();
let right = topic.subscribe().run_with(TestSink::probe()).unwrap();

assert_eq!(topic.subscriber_count(), 2);

topic.try_publish("alpha").unwrap();
topic.try_publish("beta").unwrap();

left.request(2);
right.request(2);
left.assert_next_n(["alpha", "beta"]);
right.assert_next_n(["alpha", "beta"]);

topic.close().unwrap();
left.request(1);
right.request(1);
left.expect_complete();
right.expect_complete();

Overflow is per subscriber:

Policy	Behavior
TopicOverflow::Backpressure	publish(value).await waits until every active subscriber in the publish snapshot has room; try_publish(value) returns TopicTryPublishError::Full(value) instead
TopicOverflow::Sliding	A full subscriber drops its oldest queued element, then receives the new one
TopicOverflow::Dropping	A full subscriber skips the new element; other subscribers are unaffected

try_publish() can also return TopicTryPublishError::Busy(value) when another publisher owns an earlier global publish turn.

Explicit and convenient imports

The explicit import style names the handles and policies you use:

use datum::{Channel, Signal, Subscription, SubscriptionOverflow, Topic, TopicOverflow};

The prelude style is equivalent and convenient in examples or modules already centered on Datum:

use datum::prelude::*;

Channel send errors are re-exported from the crate root as ChannelSendError and TrySendError. Topic publish errors are TopicPublishError<T> and TopicTryPublishError<T>.

Performance summary

Numbers below come from the same-host M9 comparison record in roadmap/benchmarks/concurrency-primitives.md captured on 2026-07-02. The comparison target is the best JVM row per scenario family and sweep point, not a single fixed library. Datum passes every frozen wall-clock and CPU target in this record.

Scenario family	Best JVM target	Datum result
channel_mpsc_send_1024xN	ZIO Queue at 1 producer; Akka MergeHub from 16 to 1024 producers	2.54-43.44x faster wall; 2.60-370.23x lower CPU
topic_fanout_1024xN	ZIO Hub	3.98-34.42x faster wall; 1.44-237.77x lower CPU
topic_overflow_sliding/dropping	ZIO Hub dropping policy	1.23-21.33x faster wall; 3.51-7.02x lower CPU
signal_get	ZIO SubscriptionRef	3.59-7.16x faster wall; 5.61-6.15x lower CPU
signal_propagation_1024xN	FS2 SignallingRef.discrete for wall; best CPU row per sweep	2.42-93.49x faster wall; 2.09-343.24x lower CPU
subscription_lossless_1024xN	FS2 Topic-fed state at 1 subscriber; ZIO SubscriptionRef.changes from 16 to 256 subscribers	1.02-47.23x faster wall; 1.67-271.70x lower CPU

The full benchmark file includes p50, p99, allocation, RSS, correctness counters, and the implementation notes behind each row.