Dispatcher Architecture

Dispatcher is the core component in Monibuca V6 responsible for distributing frame data from a Publisher to all Subscribers. Its design goal is to eliminate redundant reads, achieving true O(1) read + O(N) broadcast.

Design Motivation

Problems with the Traditional N-Reader Approach

In traditional streaming servers, each subscriber independently reads data from the buffer:

flowchart LR
  RB["RingBuffer"]
  R1["Reader 1"] --> S1["Subscriber 1"]
  R2["Reader 2"] --> S2["Subscriber 2"]
  R3["Reader 3"] --> S3["Subscriber 3"]
  RN["Reader N"] --> SN["Subscriber N"]
  RB --> R1 & R2 & R3 & RN

Problems with the traditional approach: N subscribers = N reads of the same frame; each Reader tracks its own state; high concurrency causes heavy atomic contention.

The Dispatcher Approach

flowchart TB
  Pub["Publisher"] --> RB["RingBuffer"]
  RB --> Disp["Dispatcher<br/>single-thread read once · Arc zero-copy · try_send"]
  Disp --> Q1["Queue 1 bounded"] & Q2["Queue 2 bounded"] & Q3["Queue 3 bounded"] & QN["Queue N bounded"]
  Q1 --> W1["Writer 1 task"]
  Q2 --> W2["Writer 2 task"]
  Q3 --> W3["Writer 3 task"]
  QN --> WN["Writer N task"]

Core advantages:

Metric	Traditional Approach	Dispatcher Approach
RingBuffer reads per frame	N times	1 time
Read lock contention	O(N)	O(1)
Frame data copies	0 (Arc shared)	0 (Arc shared)
Slow subscriber impact	May block	Drops frames, no blocking

Core Components

Dispatcher Structure

pub struct Dispatcher {
    stream_path: String,
    // Subscriber list - ArcSwap lock-free reads (COW mode)
    subscribers: ArcSwap<Vec<DispatchSubscriber>>,
    next_id: AtomicU64,
    running: AtomicBool,
    queue_capacity: usize,    // Default 150
    total_dispatched: AtomicU64,
}

Each Stream corresponds to one Dispatcher instance.

Subscriber Queues

Each subscriber owns a bounded channel (default capacity 150):

Queue capacity = 150 frames (≈ 1.8 seconds @ 80fps mixed video + audio), enough to tolerate network jitter.

When the queue is full, new frames are dropped (try_send returns Full) instead of blocking other subscribers. A dropped frame counter is automatically incremented for monitoring purposes.

Frame Message Types

The Dispatcher sends the following frame types through the channel:

pub enum DispatchFrame {
    Video(Arc<AVFrame>),         // Video frame
    Audio(Arc<AVFrame>),         // Audio frame
    VideoSeqHeader(Bytes),       // Video sequence header (AVC/HEVC decoder config)
    AudioSeqHeader(Bytes),       // Audio sequence header (AAC decoder config)
    Eos,                         // End of stream signal
}

Bounded Channel Backpressure

flowchart LR
  Pub["Publisher 30fps"] --> Disp["Dispatcher"]
  Disp -->|try_send OK| Qok["Queue partially full<br/>Subscriber normal"]
  Disp -->|try_send FULL| Qfull["Queue full<br/>drop frame · frames_dropped++ · no blocking"]

Design principle: Slow subscribers only affect themselves, never dragging down the entire system.

• Short-lived queue buildup from network fluctuations can be absorbed by the buffer
• Sustained slow consumption leads to frame drops; clients can typically recover on their own
• Per-subscriber dropped frame counts can be monitored via API

Lock-Free Subscriber Management

The subscriber list uses ArcSwap<Vec<DispatchSubscriber>> for lock-free management:

Adding Subscribers (COW Write)

// Clone-on-Write: does not block ongoing dispatch
self.subscribers.rcu(|old| {
    let mut new = (**old).clone();
    new.push(subscriber);
    new
});

Frame Dispatch (Lock-Free Read)

// Atomic load - lock-free
let subs = self.subscribers.load();
for sub in subs.iter() {
    if sub.receive_video && !sub.is_closed() {
        sub.try_send(DispatchFrame::Video(frame.clone()));
    }
}

In a scenario with 1000 subscribers @ 60fps, dispatch operations are completely free of lock contention.

DispatcherPool

When the server needs to handle a large number of concurrent streams (>100), DispatcherPool mode can be enabled.

Architecture

flowchart TB
  Pool["DispatcherPool<br/>Manages N Workers"]
  W0["Worker 0<br/>Stream A / B"]
  W1["Worker 1<br/>Stream C / D"]
  WN["Worker N-1<br/>Stream E / F"]
  Pool --> W0 & W1 & WN

Consistent Hashing

Streams are assigned to Workers via consistent hashing:

fn worker_index(&self, stream_path: &str) -> usize {
    let mut hasher = DefaultHasher::new();
    stream_path.hash(&mut hasher);
    (hasher.finish() as usize) % self.num_workers
}

The same stream path is always assigned to the same Worker, ensuring processing continuity for each stream.

Two Operating Modes

Controlled by the dispatcher_workers configuration:

Value	Mode	Description
0	Per-Stream	One independent Dispatcher task per stream (default)
N	Pool	N Workers, each handling multiple streams

# Configuration example
stream:
  dispatcher_workers: 0    # Per-Stream mode (default)
  dispatcher_workers: 4    # Pool mode with 4 Workers
  dispatcher_workers: 8    # Pool mode with 8 Workers

Selection guidelines:

• Few high-concurrency streams (< 100): Use 0 (Per-Stream), each stream gets its own task
• Many streams (> 100): Use Pool mode to reduce task overhead
• Recommended Worker count: 1–2x the number of CPU cores

Complete Data Flow Sequence

sequenceDiagram
  participant Pub as Publisher
  participant Track as VideoTrack.buffer
  participant Notify as frame_notify watch
  participant Disp as Dispatcher
  participant Sub as Subscriber
  participant Enc as Protocol encoder

  Pub->>Track: write_video(frame)
  Track->>Notify: send(count)
  Notify->>Disp: changed().await
  Disp->>Track: read_next() once → Arc AVFrame
  Disp->>Sub: try_send Video frame.clone
  Sub->>Sub: recv().await
  Sub->>Enc: RTMP / FLV / HLS / WebRTC

Notification Mechanism

The Dispatcher uses Tokio’s watch::channel to receive frame notifications. The characteristic of a watch channel is: even if multiple writes occur, the receiver is woken only once and then reads all new frames. This avoids excessive wakeups in high frame rate scenarios.

Periodic Cleanup

The Dispatcher performs cleanup every 100 dispatch cycles, removing closed subscribers:

cleanup_counter += 1;
if cleanup_counter % 100 == 0 {
    self.cleanup_closed();  // COW mode removes closed subscribers
}

When a stream ends, the Dispatcher sends an Eos (End of Stream) signal to all subscribers.