Inside the concurrency primitives

This page explains why Datum's concurrency primitives — Signal, Subscription, Channel, and Topic — are built the way they are. It is a design record, not an API reference: for the API and a "which one do I use" table, read the Concurrency Primitives guide first.

The audience here is a senior engineer deciding whether to trust these primitives on a hot path. So the tone is precise and honest: every architectural choice is tied to a measured number or to what the reference implementations (ZIO, Akka Streams, FS2) actually do internally, and the trade-offs we accepted are stated plainly.

The one rule: two planes

Every primitive is split into two planes with different jobs, different rates, and different correctness needs.

                        ┌───────────────────────────────────────────────┐
   subscribe / close    │            CONTROL PLANE  (Ractor actor)        │
   registry / terminal  │  • serializes rare, correctness-critical ops    │
   ───────────────────► │  • owns the subscriber table + lifecycle        │
                        │  • publishes an immutable ArcSwap snapshot      │
                        └───────────────────────┬───────────────────────┘
                                                │ publishes snapshot (rare)
                                                ▼
                        ┌───────────────────────────────────────────────┐
   set / send / publish │             DATA PLANE  (lock-free)             │
   get / element pull   │  • ArcSwap mirror, crossbeam ArrayQueue rings   │
   ───────────────────► │  • sequence cursors + waker/permit protocol     │
   (element-rate,       │  • NO actor message per element                 │
    the hot path)       └───────────────────────────────────────────────┘

• The control plane is an actor. Subscribe, unsubscribe, close, terminal delivery, and the subscriber registry run through a Ractor actor, because these operations demand serialization and happen at a low rate (once per subscriber lifetime, once per close). The actor is the single writer of the registry, which is what makes "subscribe with no get-then-subscribe gap" fall out for free — the actor seeds the new subscriber and registers it in the same handler turn.

• The data plane is lock-free. The per-element path — get, set, send, publish, and the subscriber pull — never sends an actor message. It uses an arc_swap::ArcSwap read mirror, crossbeam_queue::ArrayQueue rings, atomic sequence cursors, and an explicit waker/permit protocol.

Concretely, per primitive:

Primitive	Control plane (actor)	Data plane (lock-free)
Signal	subscribe / unsubscribe / close / terminal	ArcSwap<T> mirror; per-slot coalescing 1-slot mailbox; caller-thread writes
Subscription	subscribe / unsubscribe / close / terminal	ArcSwap<T> mirror; per-subscriber bounded sequence ring; caller-thread writes
Channel	consumer-slot claim; close state	ArrayQueue<T> ring; batched permit handoff; condvar/Notify wakeups
Topic	subscribe / unsubscribe / close; subscriber table	per-subscriber ArrayQueue<Arc<T>>; global sequence; ArcSwap table snapshot

Signal and Subscription writes are the interesting case: they run on the caller thread, not through the actor. The actor only owns registration and lifecycle. That is deliberate, and the next two sections explain why it had to be.

Why not an actor per element

"Actor-based" is not the same as "an actor message per element." The distinction is the whole design.

What the reference implementations actually do

We read the sources before choosing (full notes and links in .agents/notes/m9-substrate-internals.md). None of the high-throughput primitives put a fiber/actor message on the element path:

• ZIO Hub is an atomic ring. BoundedHubPow2 / BoundedHubArb / BoundedHubSingle reserve a slot with a CAS, write the array, and expose per-slot subscriber counters; fibers only park on a Promise around the structure when empty or backpressured. Publishing is CAS-into-ring, not a message.
• Akka Streams hubs (MergeHub / BroadcastHub / PartitionHub) are GraphStages over an internal lock-free AbstractNodeQueue (MPSC) or a power-of-two ring plus a consumer wheel. Control traffic (Register, Advance, NeedWakeup) goes through AsyncCallback; elements go through the queue/ring.
• FS2 Channel / Topic are cats-effect Ref + Deferred CAS state machines. Ref is a non-blocking wrapper over AtomicReference; fibers park on a Deferred only when they must wait.

And the cautionary counterexample: Akka Typed Topic (akka.actor.typed.pubsub.Topic) is actor pub-sub — one command per publish, then one ! to every subscriber actor. Its own docs position it for coarse, clustered, eventually-consistent events with no delivery guarantees, and explicitly not for stream-hub throughput. It is the shape to avoid for a data path, and it shows that the actor-per-element instinct is a known trap, not a novel one.

The Datum-specific forcing function

There is a second, harder reason unique to Datum: Ractor boxes every delivered message. Ractor 0.15.13 does a Box::pin per message in its processing loop (measured at ~848 B/element, and pinned upstream — not fixable through the public ActorRef<Msg> API). A topic fanout implemented as "tell each subscriber actor" would bake roughly 848 MB of allocation per 1M elements, times the fan-out, into the hottest path — before any payload. A preallocated ArrayQueue offer/poll is normally allocation-free and in the tens-of-nanoseconds class uncontended. For a library whose defining goal is beating warmed JVM implementations on wall and CPU, actor-per-element is simply the wrong substrate.

   actor-per-element (rejected)          two-plane (chosen)
   ───────────────────────────          ──────────────────────────────
   set/publish ─► actor msg (box ~1KB)   set/publish ─► ArcSwap store / ring offer
              ─► handler schedules                   ─► wake only parked slots
              ─► deliver to each sub                 (control actor untouched on the
              (allocation × fan-out,                  hot path; ~56 B, caller thread)
               control plane on hot path)

The measured evidence

The design was not chosen on principle alone; it was forced by numbers. All figures below are from the same-host M9 round tables in roadmap/benchmarks/concurrency-primitives.md (AMD EPYC 7R13, captured 2026-07-02). Nothing here is estimated.

1. The write-path floor — why Signal/Subscription writes skip the actor

A dedicated diagnostic measured the cost of a single state transition, comparing "route it through the actor" against the two-plane caller-thread write:

Path	Wall ns/transition	CPU ns/transition	Alloc B/transition
Ractor mailbox enqueue while actor held	114	120	81
Ractor enqueue + drain no-op handler	435	760	999
Signal::set (two-plane, no subscribers)	192	200	56
Signal::set (two-plane, one blocked subscriber)	270	540	56

The actor round-trip alone — enqueue plus a no-op handler — costs ~435 ns and ~1 KB of allocation per transition, dominated by that per-message Box::pin. For comparison, FS2's whole per-set cost in the single-subscriber coalesced propagation row is 336 µs/op ÷ 1024 sets ≈ 0.33 µs — i.e. FS2 delivers an entire set including subscriber observation for less wall time than one empty Ractor hop, and with far less allocation. Routing writes through the actor would have put Datum behind before it did any useful work.

The two-plane write drops the transition to 192 ns and 56 B with no subscribers — below the FS2 per-set target — because the actor never touches it. That single measurement is why writes live on the data plane and the actor owns only lifecycle.

2. Refcount contention — load_full() vs the guarded load

The first signal_get benchmark used get() -> Arc<T> (a full ArcSwap::load_full()) even for a scalar u64. It measured Arc refcount contention, not read cost:

Readers	load_full() on the hot read	after: guarded get_cloned()	Best JVM (ZIO)
16	51,203 µs/op · 628,000 CPU	383 µs/op · 5,500 CPU	2,743 · 30,847
64	55,109 µs/op · 2,886,500 CPU	462 µs/op · 7,000 CPU	1,657 · 43,048

Sixteen readers hammering load_full() cost ~628,000 CPU µs/op; the guarded ArcSwap::load() path (exposed as get_cloned() for cheap Copy/Clone values) cut that by two-to-three orders of magnitude. The lesson baked into the API: get() returns an owned Arc<T> snapshot for callers that want to share it, and get_cloned() exists precisely so scalar readers avoid the shared-refcount cache line. Both are lock-free reads; the difference is which one you should reach for.

3. Wake discipline — CPU ≫ wall is the thundering-herd signature

The most instructive misses were in signal propagation. Early rounds woke every subscriber on every set:

Subscribers	Early (wake-all) wall / CPU	Final (wake-discipline) wall / CPU	Best JVM (FS2)
64	133,187 / 532,000	668 / 1,000	28,617 / 162,804
1,024	2,425,854 / 9,633,333	6,254 / 53,333	522,147 / 7,768,200

Look at the 1,024-subscriber early row: CPU (9.6M) is ~4× the wall time (2.4M). Whenever CPU runs far ahead of wall-clock, that gap is a thundering herd — cores burning on redundant wakeups and re-checks, not on work. The fix was strict wake discipline, and it is the shared rule across all four primitives:

• Wake a subscriber slot only on a real edge — a consumed → fresh transition, or a slot actually parked at the current head. A slot that already has pending data is never re-woken.
• Producers scan the slot table only when a subscriber is genuinely parked (a parked_slots counter gates the scan away entirely when nobody is waiting).
• Drain in batches. Consumers pull up to 256 ring entries per wake; Channel releases those 256 slots as one batch and grants them to parked producers through producer-local permits, instead of waking one task per freed slot.

The Channel MPSC path shows the same story from the producer side — the first implementation woke one task per released slot:

Producers	First cut wall / CPU	After batched permits wall / CPU	Best JVM (Akka)
64	96,592 / 3,239,500	4,434 / 8,750	18,682 / 90,538
1,024	1,692,298 / 16,302,500	38,953 / 70,000	293,274 / 1,435,369

Batched permit handoff collapsed CPU from millions of µs/op to tens of thousands. Topic fanout was fixed the same way (running subscriber consumers on a current-thread runtime and only awaiting when the bounded topic actually backpressures), taking the 256-subscriber row from 373,869/1,056,000 to 11,014/10,000.

The takeaway for a reviewer: these primitives keep CPU close to wall-clock deliberately, and the benchmark record reports the CPU column on every row so a busy-spin win cannot hide.

Two types, not one flag: Signal vs Subscription

Signal and Subscription share a read side (an ArcSwap latest-value cell) but expose feeds with sharply different contracts:

	Signal.changes()	Subscription.changes()
Delivery	Coalesced — newest wins, intermediates may be skipped	Lossless (by default) — every accepted change, in order
Subscriber memory	One conflating slot (O(1))	Bounded per-subscriber ring (O(capacity))
Writer backpressure	Never — writers are decoupled from readers	Backpressure policy parks the writer until subscribers have room
Failure surface	none from the feed	DropNew (silent skip) or Fail (feed errors) when full

These are not one behavior with a knob. They differ in memory, in whether a slow reader can stall a writer, and in what happens on overflow. A single subscribe(mode) call would hide those consequences behind an innocent-looking argument — the caller who picks Lossless on a hot signal would silently opt into unbounded-ish buffering and writer backpressure without seeing it at the call site. Making them distinct types forces the choice to be explicit and legible in the code (maintainer decision, confirmed after a GPT-5.5 design review). The compiler-friendly, explicit-over-clever ordering is the same one described in Design Principles.

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

// Coalesced state cell. A slow subscriber sees the newest value and may skip
// intermediate states — "latest wins".
let signal = Signal::new(0_u64).unwrap();
let coalesced = signal.changes().run_with(TestSink::probe()).unwrap();
coalesced.request(1);
coalesced.assert_next(0); // the feed is seeded with the current value

signal.set(1).unwrap();
signal.set(2).unwrap();
coalesced.request(1);
coalesced.assert_next(2); // 1 was coalesced away; there is no signal.changes(mode)

// Lossless state cell: the SAME latest-value read side, but a *different type*
// whose feed delivers every accepted change in order. Coalescing vs losslessness
// are separate contracts, so they are separate types — never one flag.
let subscription = Subscription::new(0_u64, 8, SubscriptionOverflow::Backpressure).unwrap();
let lossless = subscription.changes().run_with(TestSink::probe()).unwrap();
lossless.request(1);
lossless.assert_next(0);

subscription.set(1).unwrap();
subscription.set(2).unwrap();
lossless.request(2);
lossless.assert_next_n([1, 2]); // nothing is skipped

No version counters, no MVCC object

An early proposal stored Arc<Versioned<T>> { version, value, terminal } in one ArcSwap so a reader could version-compare and decide whether to reload — a classic MVCC object guarding against a version/value split-brain race (observe a bumped version, load the old snapshot, then suppress the real new value).

That race only exists in a poll-style design where a reader independently checks a version and then loads a value. Datum's changes() feed is push: the writer publishes and then wakes; the subscriber slot state itself is the edge. There is nothing for a subscriber to poll and nothing to version-compare, so the race has no surface to occur on. The Versioned<T> object was solving a problem the push design does not have — so it was cut (maintainer simplification, 2026-07-02).

What remains is a single load-bearing ordering rule inside the write path:

   apply transition to owned state
        ▼
   store the ArcSwap mirror        ◄── value is visible here
        ▼
   deliver to subscriber slots
        ▼
   ack the write                   ◄── acked set() ⇒ read-your-writes

Mirror-before-ack gives read-your-writes for an acknowledged set(); a fire-and-forget write is documented as eventually-visible. Terminal/close is the same transition shape: publish the final snapshot, then the terminal, to the slots before acking the close.

When to reconsider: if Datum ever adds a poll-style watch-like API (read-and-compare instead of push-and-wake), the split-brain race reappears and version counters (or the Versioned<T> MVCC object) become necessary. They are deliberately deferred until then — not added speculatively — and this is the documented trigger to revisit.

Safety: lock-free without unsafe

datum-core is #![forbid(unsafe_code)], and the concurrency primitives do not carve out an exception (see Design Principles). The lock-free structures come entirely from safe, audited dependencies:

• arc-swap for the atomic Arc mirror and subscriber-table snapshots,
• crossbeam-queue (ArrayQueue) for the MPSC and per-subscriber rings.

The correctness argument for the read side is worth stating precisely, because it is what lets the data plane be both lock-free and safe:

Immutability + refcount = MVCC by construction. Every published state is an immutable Arc<T>; a new state is a new Arc, never a mutation of the old one. A reader that holds an Arc<T> keeps that exact snapshot alive by refcount until it drops the handle. So reads cannot tear (there is no in-place mutation to observe half-done), and there is no use-after-free (the refcount is the lifetime). Multi-version concurrency is achieved without a version field, a lock, or a single line of unsafe.

The snippet below is the observable form of that guarantee — an older snapshot stays valid and untorn across a concurrent write:

use datum::Signal;

let signal = Signal::new(vec![1_u64, 2, 3]).unwrap();

// get() returns an owned Arc<T> snapshot with no actor message — a plain
// ArcSwap load on the data plane. An old reader keeps its snapshot until it
// drops the Arc, so reads never tear and never block a writer.
let before = signal.get();
signal.set(vec![4, 5, 6]).unwrap();
let after = signal.get();

assert_eq!(*before, vec![1, 2, 3]); // the older snapshot is still valid
assert_eq!(*after, vec![4, 5, 6]);

// get_cloned() uses the guarded read path for cheap Copy/Clone values, and an
// acknowledged set() is read-your-writes on a later read.
let counter = Signal::new(0_u64).unwrap();
counter.set(42).unwrap();
assert_eq!(counter.get_cloned(), 42);

One honest caveat: Arc<T> cannot prevent user-chosen interior mutability inside T (e.g. a Mutex or atomic you put in the value). The primitives treat T as logically immutable and you should too; the "no torn reads" guarantee is about Datum's own publication, not about mutation you introduce inside the value.

Benchmark-first, and root-cause every miss

The methodology is as much a part of the design as the substrate:

• Frozen JVM baselines first. The comparison harness (WP-B0) captured FS2, ZIO, and Akka target numbers before any Datum implementation existed, under fair JMH warmup (-wi 5 -i 5 -w 1s -r 1s). Each Datum row is measured against a frozen target it cannot retroactively move. The bar is parity-or-better against the best listed competitor per row, on both wall and CPU.
• Correctness counters on every benchmark. Each scenario asserts its invariant — total sends/receives, per-subscriber counts, final observed value, or that the overflow policy actually engaged — so a fast-but-wrong run cannot pass. A benchmark that loses elements is a failing benchmark, not a fast one.
• Root-cause every miss. The round tables above are kept in the record on purpose: the first cut usually missed, and each round names the lever it moved (refcount path, wake discipline, permit batching, current-thread runtime). "It's slow" was never accepted as an answer; the CPU column points at the herd, and the diagnostic points at the floor.

Performance at a glance

A few representative rows — Datum against the best JVM competitor for that scenario and sweep point. Full p50/p99, allocation, RSS, and per-round notes are in the benchmark record.

Scenario	Datum wall / CPU (µs/op)	Best JVM wall / CPU (µs/op)
topic_fanout 1024 × 1024 subs	61,891 / 70,000	ZIO Hub 1,430,782 / 11,034,842
channel_mpsc 1024 × 1024 producers	38,953 / 70,000	Akka MergeHub 293,274 / 1,435,369
signal_propagation 1024 × 64 subs	668 / 1,000	FS2 28,617 / 162,804
subscription_lossless 1024 × 64 subs	1,347 / 2,000	ZIO 63,620 / 543,396
signal_get 64 readers	462 / 7,000	ZIO SubscriptionRef 1,657 / 43,048

Datum passes every frozen wall-clock and CPU target in the record. The wins are largest exactly where the two-plane rule pays off most: high fan-out, where an actor-per-element design would have been paying ~1 KB and a mailbox hop per delivered message.

Related

• Concurrency Primitives guide — the API and a "which primitive" table.
• Execution Model — the thread pool, spin-then-park, and the Ractor allocation profile referenced above.
• Design Principles — #![forbid(unsafe_code)], blueprint-first construction, and the Result-based error model these primitives follow.