RingBuffer

RingBuffer is the core data structure of the Monibuca V6 streaming engine, responsible for efficient frame storage and concurrent multi-consumer reads.

Design Overview

flowchart TB
  WP["write_pos (atomic)"]
  subgraph Slots["RingBuffer slots"]
    direction LR
    S0["Slot 0<br/>Arc Frame"]
    S1["Slot 1<br/>Arc Frame"]
    S2["Slot 2<br/>Arc Frame"]
    S3["Slot 3…N"]
  end
  RA["Reader A<br/>read_pos=0"]
  RBr["Reader B<br/>read_pos=1"]
  RC["Reader C<br/>read_pos=2"]
  WP --> Slots
  S0 --> RA
  S1 --> RBr
  S2 --> RC

RingBuffer uses an SPMC (Single Producer, Multiple Consumer) pattern:

• Single producer: One Publisher writes frame data
• Multiple consumers: Multiple Subscribers (via RingReader) read concurrently
• Fixed capacity: Automatically wraps around and overwrites old data when full

Core Features

Writes Never Block

Write operations use fetch_add atomic instructions to increment write_pos, requiring no locks:

// Atomically fetch and advance the write position
let pos = self.write_pos.fetch_add(1, Ordering::AcqRel) % self.capacity;
let slot = &self.slots[pos];

// ArcSwap atomically swaps the frame data
let arc_frame = Arc::new(frame);
slot.frame.store(Some(arc_frame));

// Update the version number (used to detect overwrites)
slot.version.fetch_add(1, Ordering::Release);

Key design: Even if a consumer reads slowly, write operations are never blocked. The writer simply advances write_pos forward, and old data is naturally overwritten.

Arc Zero-Copy Sharing

Each frame is stored as an Arc<AVFrame>:

flowchart LR
  subgraph Write["Publisher writes"]
    F["AVFrame"] --> ArcN["Arc::new(frame)"] --> Slot["Slot.frame<br/>ArcSwapOption"]
  end
  subgraph Read["Subscriber reads"]
    Load["Slot.frame.load_full()"] --> Shared["Arc AVFrame<br/>ref count +1"]
    Shared --> SA["Subscriber A"]
    Shared --> SB["Subscriber B"]
    Shared --> SC["Subscriber C"]
  end
  Slot -.-> Load

• All subscribers share the same Arc<AVFrame>, no data copying needed
• Frame data is automatically reclaimed when all references are released
• ArcSwapOption provides lock-free atomic swap capability

Version Number Validation

Each Slot contains an atomic version number version: AtomicU64:

pub struct RingSlot {
    frame: ArcSwapOption<AVFrame>,    // Frame data
    version: AtomicU64,               // Version number
    sequence: AtomicU64,              // Frame sequence number
    written: AtomicBool,              // Whether data has been written
}

• version is incremented on each write
• Readers compare version to detect whether data has been overwritten
• If the sequence number doesn’t match, the reader automatically seeks to the nearest IDR frame to resynchronize

IDR Frame Tracking

Video streams need to start playback from a keyframe (IDR). RingBuffer maintains a list of IDR frame positions:

flowchart TB
  IDR["IDR List<br/>ArcSwap Vec IDRNode"]
  Nodes["IDR@pos=0, IDR@pos=30, IDR@pos=60"]
  Ptr["Atomic pointer"]
  IDR --> Nodes
  Ptr -.-> IDR

  subgraph COW["When writing an IDR frame (COW)"]
    direction TB
    C1["1. Acquire lock idr_write_lock"] --> C2["2. Clone old list"]
    C2 --> C3["3. Add new IDR node"]
    C3 --> C4["4. Atomically swap new list"]
    C4 --> C5["5. Release lock"]
  end

The IDR list uses an ArcSwap + COW (Copy-on-Write) pattern:

• Read (seek to keyframe): Completely lock-free, loaded via atomic pointer
• Write (new keyframe arrives): Clone-modify-swap, does not block read operations
• This is a read-heavy, write-light optimization — keyframes typically appear every 1–2 seconds, while seek operations occur more frequently

Each IDR node records:

pub struct IDRNode {
    pub index: usize,       // Position in the RingBuffer
    pub sequence: u64,      // Frame sequence number
    pub timestamp: Duration, // Frame timestamp (milliseconds)
}

A maximum of 16 IDR frames are tracked (default value, configurable via with_idr_capacity).

RingReader

Each subscriber owns an independent RingReader instance to track its own read position:

pub struct RingReader {
    buffer: Arc<RingBuffer>,  // Shared buffer reference
    read_pos: usize,          // Current read position
    expected_seq: u64,        // Expected next frame sequence number
    frames_read: u64,         // Number of frames read
    state: ReaderState,       // Reader state
}

Reader State Machine

stateDiagram-v2
  [*] --> Init
  Init --> WaitingKeyframe
  WaitingKeyframe --> Normal: IDR found
  Normal --> WaitingKeyframe: overwritten / sequence mismatch
  Normal --> CatchingUp: fallen too far behind
  CatchingUp --> Normal: re-seek to IDR

State	Description
Init	Initial state, waiting for first seek
WaitingKeyframe	Waiting for a keyframe
Normal	Normal reading
CatchingUp	Fallen too far behind, needs to jump to the latest IDR

Read Flow

// 1. Check if we've reached the write position (no new data)
if self.read_pos == write_pos {
    return None;
}

// 2. Check if we've fallen too far behind (distance > capacity - 2)
if distance > capacity - 2 {
    // Re-seek to the latest IDR
    self.seek_to_latest_idr();
}

// 3. Check if the sequence number matches
if frame.sequence != self.expected_seq {
    // Data has been overwritten, seek to the latest IDR
    self.seek_to_latest_idr();
}

// 4. Read the frame (acquire Arc reference)
let arc_frame = self.buffer.read(self.read_pos);
self.read_pos = (self.read_pos + 1) % capacity;

Default Capacities

Track	Default Capacity	Description
VideoTrack	1024 slots	~34 seconds of buffer (at 30fps)
AudioTrack	64 slots	~1.3 seconds of buffer (at 48kHz, 1024 samples/frame)

Capacities are configurable via PublisherConfig:

PublisherConfig {
    video_buffer_capacity: 1024,
    audio_buffer_capacity: 64,
    ..Default::default()
}

Minimum Capacity

The minimum RingBuffer capacity is 8 slots. Even if a smaller value is configured, it will be automatically raised to 8.

Performance Characteristics

Operation	Latency	Lock Contention
Write frame	~100ns	Lock-free (atomic fetch_add)
Read frame	~50ns	Lock-free (ArcSwap load)
Seek IDR	~100ns	Lock-free (ArcSwap load)
Add IDR	~200ns	Brief Mutex (serializes IDR writes only)

Multi-Reader Concurrency

flowchart TB
  RB["RingBuffer (shared)"]
  RA["RingReader Sub A<br/>read_pos=10"]
  RBr["RingReader Sub B<br/>read_pos=8"]
  RC["RingReader Sub C<br/>read_pos=12"]
  RB --> RA & RBr & RC

• Each Reader independently tracks its position
• Read operations do not interfere with each other
• Write operations never block any read operations

Supports thousands of concurrent readers with zero contention between them. The only shared operation between readers is the reference count on the underlying Arc<AVFrame> (hardware-level atomic instructions).