rtp

type JitterBuffer ¶

type JitterBuffer struct {
	Overflows  atomic.Int64
	SSRCResets atomic.Int64
	IdleResets atomic.Int64
	// GapRuns{N} counts contiguous sequence-gap runs observed by the
	// skip-missing branch of popReadyLocked. Each gap is bucketed once
	// at the point the consumer first skips into it — subsequent
	// single-step cursor advances over the same run are not re-counted.
	// The buckets distinguish isolated loss (GapRuns1, GapRuns2to5)
	// from long contiguous bursts (GapRuns11to20 and up) so operators
	// can tell whether FEC/redundancy or masking is the right recovery
	// strategy. MaxDepth caps any single measured run at maxDepth, so
	// the larger buckets will only fire if MaxDepth is raised.
	GapRuns1      atomic.Int64
	GapRuns2to5   atomic.Int64
	GapRuns6to10  atomic.Int64
	GapRuns11to20 atomic.Int64
	GapRuns21to50 atomic.Int64
	GapRunsOver50 atomic.Int64
	// contains filtered or unexported fields
}

JitterBuffer is a sequence-number-ordered buffer for RTP audio payloads. It provides prebuffering, late-packet dropping, gap detection for PLC, and SSRC-change/idle-gap detection so a new talker is never silently dropped because their starting sequence number lies in the "past half" of the previous talker's frozen sequence cursor.

Internally, frames are stored in a fixed-size circular array indexed by (seq % maxDepth), eliminating all map allocations on the hot path.

notifyCh is the optional edge-triggered "frame available" wakeup used by consumers that prefer push-driven scheduling over a polling ticker (see EnableNotify and webPlayoutLoop). It is nil until a consumer opts in; malgo-driven consumers, which are clocked by the audio hardware, do not call EnableNotify and pay nothing on the push hot path beyond the nil-check.

func NewJitterBuffer(prebuffer, maxDepth int) *JitterBuffer

func (jb *JitterBuffer) AdvancePast()

func (jb *JitterBuffer) EnableNotify() <-chan struct{}

func (jb *JitterBuffer) PopOrConceal(recentWindow time.Duration) (payload []byte, conceal bool)

func (jb *JitterBuffer) PopReady() (payload []byte, ready bool, skippedMissing bool)

func (jb *JitterBuffer) Push(seq uint16, payload []byte) bool

func (jb *JitterBuffer) PushWithSSRC(ssrc uint32, seq uint16, payload []byte, onSSRCChange func(oldSSRC, newSSRC uint32)) bool

func (jb *JitterBuffer) ReleasePayload(p []byte)

func (jb *JitterBuffer) Reset()

func (jb *JitterBuffer) ShouldConceal(recentWindow time.Duration) bool

func (jb *JitterBuffer) Snapshot(dst *JitterBufferSnapshot)

type JitterBufferSnapshot ¶

type JitterBufferSnapshot struct {
	// Overflows is the monotonic count of incoming packets the jitter
	// buffer rejected because it was full. Sustained non-zero overflow
	// deltas suggest the receiver is behind the sender or the network
	// is bursting.
	Overflows int64 `json:"overflows"`
	// SSRCResets is the monotonic count of mid-stream SSRC transitions
	// the jitter buffer handled by resetting and re-initializing. A
	// high value suggests multiple talkers or a sender restart.
	SSRCResets int64 `json:"ssrc_resets"`
	// IdleResets is the monotonic count of gap-driven buffer resets
	// (the "same SSRC but the stream has been silent for longer than
	// the idle threshold" safety net).
	IdleResets int64 `json:"idle_resets"`
	// GapRuns{N} is the monotonic count of contiguous sequence-gap
	// runs observed at the jitter buffer's skip-missing branch,
	// bucketed by run length in frames (20 ms each). A single gap is
	// counted exactly once at the skip point, not once per skipped
	// frame. Distribution across buckets distinguishes isolated loss
	// (bucket 1, 2–5) from long contiguous bursts (11–20 and up) and
	// drives the choice of recovery strategy (FEC/RED vs masking).
	GapRuns1      int64 `json:"gap_runs_1"`
	GapRuns2to5   int64 `json:"gap_runs_2_5"`
	GapRuns6to10  int64 `json:"gap_runs_6_10"`
	GapRuns11to20 int64 `json:"gap_runs_11_20"`
	GapRuns21to50 int64 `json:"gap_runs_21_50"`
	GapRunsOver50 int64 `json:"gap_runs_over_50"`
}

JitterBufferSnapshot is a point-in-time copy of the per-port receive jitter buffer's monotonic counters. Field semantics are documented in docs/instrumentation-snapshot.md — keep that file in sync when adding or renaming fields here.

Note that depth/occupancy are intentionally omitted. Reading them would require acquiring jb.mu, which the snapshot path never holds. The three fields below are read via atomic loads and so observe producer writes without contention.

type PacketReader ¶

type PacketReader interface {
	ReadFromUDP(b []byte) (int, *net.UDPAddr, error)
	Close() error
}

PacketReader abstracts the UDP receive path so receiveLoop can be exercised with pre-seeded byte sequences in tests.

type PacketWriter ¶

type PacketWriter interface {
	Write(b []byte) (int, error)
}

PacketWriter abstracts the UDP send path so the broadcast callback can be tested without a live socket.

type Sender ¶

type Sender interface {
	Send(payload []byte) error
}

Sender is the interface the broadcast encoder uses to ship an encoded Opus frame over the network. Backed by Session in production; tests inject mockRTPSender.

type Session ¶

type Session struct {
	// contains filtered or unexported fields
}

Session wraps a pion Packetizer and an interceptor chain that generates periodic RTCP Sender Reports. One session represents one local SSRC (the node running this software).

The RTCP path is one-way outbound: the SR generator fires every 5 seconds and writes to the provided rtcpTransport. Inbound RTP is parsed externally with ParseIncoming; the session does not handle receive stats (each remote SSRC in a multicast group would require a separate receiver).

Concurrency: Send is the sole writer to packetizer and rtpWriter. In production each *Session is owned by exactly one PortChannel, and Send is only called from the per-encoder broadcastEncoder.encodeLoop goroutine (one writer per encodeLoop, one encodeLoop per process). The pion SenderInterceptor's RTCP timer runs on its own goroutine but only writes to the RTCP transport (bound separately at NewSession), not the RTP packetizer or rtpWriter — so there is no contention with Send. Adding a second concurrent Send caller without external synchronization would be a bug; the call invariant is enforced by code review, not a mutex.

func NewSession(
	ssrc uint32,
	rtpTransport PacketWriter,
	rtcpTransport PacketWriter,
	log zerolog.Logger,
) (*Session, error)

func (s *Session) Close() error

func (s *Session) Send(payload []byte) error

type SwappableReceiver ¶

type SwappableReceiver struct {
	// contains filtered or unexported fields
}

SwappableReceiver wraps a PacketReader so it can be atomically replaced at runtime without races with the blocking receive loop.

ReadFromUDP snapshots the current implementation under a read lock and then releases the lock before blocking, so a concurrent swap is never blocked by an in-flight I/O call. Closing the old connection after the swap unblocks any in-progress ReadFromUDP on the old socket, causing receiveLoop to loop back and immediately read from the new socket.

func NewSwappableReceiver(r PacketReader) *SwappableReceiver

func (r *SwappableReceiver) Close() error

func (r *SwappableReceiver) ReadFromUDP(b []byte) (int, *net.UDPAddr, error)

func (r *SwappableReceiver) Swap(newR PacketReader) PacketReader

type SwappableSender ¶

type SwappableSender struct {
	// contains filtered or unexported fields
}

SwappableSender wraps a PacketWriter so it can be atomically replaced at runtime without races with in-flight writes.

Concurrency model:

• The hot path (Write) is lock-free: it atomically loads the current PacketWriter pointer and performs the single underlying UDP sendto(2). No mutex and no WaitGroup increment, so concurrent writers do not contend against each other or against swappers.
• swap() serializes concurrent swappers via swapMu, publishes the new PacketWriter via atomic.Pointer.Store(), and returns the previous one. The caller is responsible for closing the previous PacketWriter.
• Because the hot path holds no lock, swap cannot prove that writers are done with the previous pointer before it returns. Callers that close the returned PacketWriter synchronously risk closing a socket out from under an in-flight sendto(2). To avoid this, swap returns a deferred closer via swapAndDeferClose which schedules the underlying close after SwapCloseGrace — long enough for any in-flight write syscall to complete. This is the tradeoff the refactor plan explicitly allows as the fallback to a WaitGroup-based drain (the WaitGroup approach fails with "Add called concurrently with Wait" because writers do not take any lock that could be ordered against Wait).
• Close takes a snapshot and closes it if it implements io.Closer.

func NewSwappableSender(w PacketWriter) *SwappableSender

func (s *SwappableSender) Close() error

func (s *SwappableSender) Swap(newW PacketWriter) PacketWriter

func (s *SwappableSender) SwapAndDeferClose(newW PacketWriter)

func (s *SwappableSender) Write(b []byte) (int, error)