lockfree/broadcast

Concurrency Topology: MPMC / SPMC Multicast Pub-Sub Ring Buffer (BroadcastRing[T] / TopicBus[T])

Dimension	Architectural Specification
Topologies	SPMC (Single-Publisher Multi-Consumer) / MPMC (Multi-Publisher Multicast)
Delivery Model	1-to-N Fan-Out Broadcast (Every message delivered to ALL active readers)
Buffer Geometry	Power-of-two circular ring buffer (2^k) with monotonic sequence counters
Consumer Tracking	Independent reader cursors (BroadcastCursor[T]) dynamically registered
Overflow Policies	omDropOldest (Lossy real-time ring) vs omBackoff (Lossless backpressure)
Multiplexer	TopicBus[T] mapping topic keys (string) to dedicated BroadcastRings
Slot Lifetime	Atomic refcounted payload box (VBox[T]) with safe ARC/ORC lifecycle
Cache Alignment	Slots, tail counters, and cursor descriptors aligned to CacheLineBytes
Progress Guarantees	Publishers: Lock-Free (omDropOldest) / Flow-Controlled (omBackoff)
	Readers: Wait-Free Population (omDropOldest) / Lock-Free (omBackoff)

Overview

BroadcastRing[T] is a high-performance lock-free multicast broadcast ring buffer. In contrast to point-to-point queues where each item is consumed by exactly one reader, BroadcastRing[T] delivers every message to all active registered subscribers.

TopicBus[T] provides a multiplexed Pub-Sub bus routing messages by topic string to automatically provisioned BroadcastRing[T] instances backed by Ctrie.

Types

OverflowMode

OverflowMode = enum
  omDropOldest,             ## Overwrite oldest unread items; slow consumers lag and skip
  omBackoff                  ## Stalls publisher until slowest active cursor advances

SubscriptionOrigin

SubscriptionOrigin = enum
  soFromLatest,             ## Start reading from messages published AFTER subscription
  soFromEarliest             ## Replay available ring history up to current capacity

PollResultKind

PollResultKind = enum
  prSuccess,                ## Item successfully read
  prEmpty,                  ## No new messages available at this cursor
  prLagged                   ## Reader lagged behind overwritten horizon; skipped messages

PollResult

PollResult[T] = object
  case kind*: PollResultKind
  of prSuccess:
    val*: T
  of prEmpty:
    nil
  of prLagged:
    skippedCount*: uint64

VBox

VBox[T] = object
  rc*: Atomic[int32]
  val*: T

BroadcastSlot

BroadcastSlot[T] = object
  seq* {.align: 128.}: Atomic[uint64] ## Monotonic sequence number published into this slot
  writeSeq*: Atomic[uint64]  ## Pre-write sequence stamp for inline seqlock
  vbox*: Atomic[ptr VBox[T]] ## Pointer to heap-allocated payload box
  retiredBox*: Atomic[ptr VBox[T]] ## Quarantined previous box from prior lap
  inlineBits*: Atomic[uint64] ## Atomic inline storage for trivial <=64-bit types

BroadcastRingConfig

BroadcastRingConfig = object
  capacity*: int             ## Power of two (e.g. 1024, 65536)
  overflowMode*: OverflowMode ## omDropOldest or omBackoff
  maxReaders*: int           ## Maximum concurrent registered cursors (default 64)

BroadcastRingCore

BroadcastRingCore[T] = object
  capacity*: int
  mask*: uint64
  overflowMode*: OverflowMode
  maxReaders*: int
  publisherTail*: Atomic[uint64] ## Monotonically incrementing publication ticket
  publishedHead*: Atomic[uint64] ## Highest fully published sequence number
  slots*: BroadcastSlotBuffer[T]
  cursors*: CursorDescriptorBuffer
  activeCursorMask*: Atomic[uint64] ## Bitmask of active cursors (0..<min(maxReaders, 64))
  activeCursorCount*: Atomic[int32]
  rc*: Atomic[int]

BroadcastRing

BroadcastRing[T] = object
  core*: ptr BroadcastRingCore[T]

BroadcastCursor

BroadcastCursor[T] = object
  cursorId*: int             ## Slot in the ring's cursor table
  readSeq*: uint64           ## Reader's local non-atomic sequence counter
  lagCount*: uint64          ## Total number of messages skipped due to overflow
  ring*: BroadcastRing[T]    ## Handle keeping ring alive

TopicBusConfig

TopicBusConfig = object
  defaultRingCapacity*: int
  defaultOverflowMode*: OverflowMode
  defaultMaxReaders*: int

TopicBus

TopicBus[T] = object
  core*: ptr TopicBusCore[T]

Constants

DefaultRingCapacity

DefaultRingCapacity = 1024

DefaultMaxReaders

DefaultMaxReaders = 64

CacheLine

CacheLine = 128

moAcqRel

moAcqRel = moAcquireRelease

moSeqCst

moSeqCst = moSequentiallyConsistent

Procedures & Functions

initBroadcastRing

proc initBroadcastRing[T](capacity: int = DefaultRingCapacity;
                          overflowMode: OverflowMode = omDropOldest;
                          maxReaders: int = DefaultMaxReaders): BroadcastRing[T]

Constructs a new concurrent multicast broadcast ring.

=destroy

proc `=destroy`[T](self: var BroadcastRing[T]) {.gcsafe, raises: [].}

=copy

proc `=copy`[T](dest: var BroadcastRing[T]; src: BroadcastRing[T])

=sink

proc `=sink`[T](dest: var BroadcastRing[T]; src: BroadcastRing[T])

isNil

proc isNil[T](self: BroadcastRing[T]): bool {.inline.}

capacity

proc capacity[T](self: BroadcastRing[T]): int {.inline.}

subscriberCount

proc subscriberCount[T](self: BroadcastRing[T]): int {.inline.}

len

proc len[T](self: BroadcastRing[T]): int {.inline.}

Returns the number of published messages currently residing in the ring.

publish

proc publish[T](self: BroadcastRing[T]; item: sink T)

Publishes a message to all active subscribers. In omDropOldest, overwrites oldest messages if the ring is full without blocking. In omBackoff, stalls/spins until the slowest active subscriber advances.

tryPublish

proc tryPublish[T](self: BroadcastRing[T]; item: sink T): bool

Non-blocking publish attempt. Returns true if published. In omBackoff, returns false if the ring is currently full.

subscribe

proc subscribe[T](self: BroadcastRing[T];
                  origin: SubscriptionOrigin = soFromLatest): BroadcastCursor[T]

Dynamically registers a new reader cursor.

unsubscribe

proc unsubscribe[T](cursor: var BroadcastCursor[T])

Deregisters the cursor, freeing its slot in the cursor table.

=destroy

proc `=destroy`[T](self: var BroadcastCursor[T]) {.gcsafe, raises: [].}

isNil

proc isNil[T](cursor: BroadcastCursor[T]): bool {.inline.}

lag

proc lag[T](cursor: BroadcastCursor[T]): uint64 {.inline.}

poll

proc poll[T](cursor: var BroadcastCursor[T]): PollResult[T]

Polls the next message from this cursor’s position. Returns: prSuccess with the value if available. prEmpty if the publisher has not yet published this sequence. prLagged with skippedCount if the reader fell behind the ring horizon (omDropOldest).

tryRead

proc tryRead[T](cursor: var BroadcastCursor[T]; outVal: var T): bool

Convenience read helper. If a message is available, writes to outVal and returns true. If empty, returns false. If lagged, advances cursor and returns false.

initTopicBus

proc initTopicBus[T](config: TopicBusConfig = TopicBusConfig(
    defaultRingCapacity: 1024, defaultOverflowMode: omDropOldest,
    defaultMaxReaders: 64)): TopicBus[T]

Initializes a new TopicBus multiplexing messages across topic-specific broadcast rings.

=destroy

proc `=destroy`[T](bus: var TopicBus[T]) {.raises: [].}

=copy

proc `=copy`[T](dest: var TopicBus[T]; src: TopicBus[T])

=sink

proc `=sink`[T](dest: var TopicBus[T]; src: TopicBus[T])

isNil

proc isNil[T](bus: TopicBus[T]): bool {.inline.}

getOrCreateRing

proc getOrCreateRing[T](bus: TopicBus[T]; topic: string): BroadcastRing[T]

publish

proc publish[T](bus: TopicBus[T]; topic: string; item: sink T)

Publishes a message to the broadcast ring associated with topic.

subscribe

proc subscribe[T](bus: TopicBus[T]; topic: string;
                  origin: SubscriptionOrigin = soFromLatest): BroadcastCursor[T]

Dynamically subscribes to topic, returning an independent reader cursor.

topicCount

proc topicCount[T](bus: TopicBus[T]): int {.inline.}

hasTopic

proc hasTopic[T](bus: TopicBus[T]; topic: string): bool {.inline.}

Templates

isInlineBroadcastType

template isInlineBroadcastType(T: typedesc): bool