From 48700da3f3f2c3b514f673308bed861ec09fd729 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Wed, 16 Sep 2026 11:53:54 +0530 Subject: [PATCH] sfu: broadcast RTP to down tracks without per-packet allocations (#4875) * sfu: broadcast RTP to down tracks without a per-packet closure Every forwarded packet allocated a closure capturing the packet and layer, and an atomic write counter that escaped with it, even for tracks with a single subscriber. Add BroadcastRTP, which carries the packet and layer in the worker state and sums the writes per worker. The serial path no longer allocates; the parallel path allocates the shared state and the worker funcval only. Co-Authored-By: Claude Fable 5.1 * sfu: reuse the forwarded packet copy in the RED receivers Both RED receivers copied the ExtPacket (and the opus one the rtp.Packet) per forwarded packet, and the copies moved to the heap because the broadcast takes their address. Keep them on the receiver like the existing redPayloadBuf: ForwardRTP runs on one goroutine and down tracks do not keep the packet past WriteRTP. Co-Authored-By: Claude Fable 5.1 * sfu: skip BroadcastRTP allocation assertion under the race detector Co-Authored-By: Claude Fable 5.1 * sfu: group BroadcastRTP and its interfaces after the spreader methods Co-Authored-By: Claude Fable 5.1 --------- Co-authored-by: Claude Fable 5.1 --- pkg/sfu/receiver_base.go | 11 ++- pkg/sfu/redprimaryreceiver.go | 21 +++--- pkg/sfu/redreceiver.go | 24 +++---- pkg/sfu/utils/downtrackspreader.go | 95 +++++++++++++++++++++---- pkg/sfu/utils/downtrackspreader_test.go | 84 ++++++++++++++++++++++ 5 files changed, 192 insertions(+), 43 deletions(-) create mode 100644 pkg/sfu/utils/downtrackspreader_test.go diff --git a/pkg/sfu/receiver_base.go b/pkg/sfu/receiver_base.go index 80874a893..617d7fc64 100644 --- a/pkg/sfu/receiver_base.go +++ b/pkg/sfu/receiver_base.go @@ -962,12 +962,9 @@ func (r *ReceiverBase) forwardRTP( continue } - var writeCount atomic.Int32 - r.downTrackSpreader.Broadcast(func(dt TrackSender) { - writeCount.Add(dt.WriteRTP(extPkt, spatialLayer)) - }) + writeCount := sfuutils.BroadcastRTP(r.downTrackSpreader, extPkt, spatialLayer) if rt := r.loadREDTransformer(); rt != nil { - writeCount.Add(rt.ForwardRTP(extPkt, spatialLayer)) + writeCount += rt.ForwardRTP(extPkt, spatialLayer) } // track delay/jitter @@ -977,13 +974,13 @@ func (r *ReceiverBase) forwardRTP( // delivered back-to-back) which the single forwarder goroutine drains // serially, inflating the measured transit for the tail of the burst. That // reflects loss recovery rather than steady-state forwarding health. - if writeCount.Load() > 0 && r.forwardStats != nil && !extPkt.IsBuffered && !extPkt.IsOutOfOrder { + if writeCount > 0 && r.forwardStats != nil && !extPkt.IsBuffered && !extPkt.IsOutOfOrder { if latency, isHigh := r.forwardStats.Update(extPkt.Arrival, mono.UnixNano()); isHigh { r.params.Logger.Debugw( "high forwarding latency", "latency", time.Duration(latency), "queuingLatency", time.Duration(dequeuedAt-extPkt.Arrival), - "writeCount", writeCount.Load(), + "writeCount", writeCount, "isOutOfOrder", extPkt.IsOutOfOrder, "layer", layer, ) diff --git a/pkg/sfu/redprimaryreceiver.go b/pkg/sfu/redprimaryreceiver.go index 17b639abb..a34ef8586 100644 --- a/pkg/sfu/redprimaryreceiver.go +++ b/pkg/sfu/redprimaryreceiver.go @@ -50,6 +50,10 @@ type RedPrimaryReceiver struct { // bitset for upstream packet receive history [lastSeq-8, lastSeq-1], bit 1 represents packet received pktHistory byte + + // forwarded packet, reused since ForwardRTP runs on one goroutine and + // down tracks do not keep the packet past WriteRTP + sendExtPkt buffer.ExtPacket } func NewRedPrimaryReceiver(receiver TrackReceiver, dsp utils.DownTrackSpreaderParams) REDTransformer { @@ -73,11 +77,7 @@ func (r *RedPrimaryReceiver) ForwardRTP(pkt *buffer.ExtPacket, spatialLayer int3 if pkt.Packet.PayloadType != r.redPT { // forward non-red packet directly - var writeCount atomic.Int32 - r.downTrackSpreader.Broadcast(func(dt TrackSender) { - writeCount.Add(dt.WriteRTP(pkt, spatialLayer)) - }) - return writeCount.Load() + return utils.BroadcastRTP(r.downTrackSpreader, pkt, spatialLayer) } pkts, err := r.getSendPktsFromRed(pkt.Packet) @@ -86,9 +86,10 @@ func (r *RedPrimaryReceiver) ForwardRTP(pkt *buffer.ExtPacket, spatialLayer int3 return 0 } - var writeCount atomic.Int32 + var writeCount int32 for i, sendPkt := range pkts { - pPkt := *pkt + pPkt := &r.sendExtPkt + *pPkt = *pkt if i != len(pkts)-1 { // patch extended sequence number and time stamp for all but the last packet, // last packet is the primary payload @@ -129,11 +130,9 @@ func (r *RedPrimaryReceiver) ForwardRTP(pkt *buffer.ExtPacket, spatialLayer int3 // not modify the ExtPacket.RawPacket here for performance since it is not used by the DownTrack, // otherwise it should be set to the correct value (marshal the primary rtp packet) - r.downTrackSpreader.Broadcast(func(dt TrackSender) { - writeCount.Add(dt.WriteRTP(&pPkt, spatialLayer)) - }) + writeCount += utils.BroadcastRTP(r.downTrackSpreader, pPkt, spatialLayer) } - return writeCount.Load() + return writeCount } func (r *RedPrimaryReceiver) ForwardRTCPSenderReport( diff --git a/pkg/sfu/redreceiver.go b/pkg/sfu/redreceiver.go index 959a54e96..4977878bf 100644 --- a/pkg/sfu/redreceiver.go +++ b/pkg/sfu/redreceiver.go @@ -51,6 +51,10 @@ type RedReceiver struct { closed atomic.Bool pktBuff [maxRedCount]*rtp.Packet redPayloadBuf [mtuSize]byte + // forwarded packet, reused like redPayloadBuf since ForwardRTP runs on one goroutine + // and down tracks do not keep the packet past WriteRTP + redExtPkt buffer.ExtPacket + redRtpPkt rtp.Packet } func NewRedReceiver(receiver TrackReceiver, dsp utils.DownTrackSpreaderParams) REDTransformer { @@ -73,11 +77,7 @@ func (r *RedReceiver) ForwardRTP(pkt *buffer.ExtPacket, spatialLayer int32) int3 // fallback to primary codec if payload size exceeds redundant block length if len(pkt.Packet.Payload) >= maxRedPayload { - var writeCount atomic.Int32 - r.downTrackSpreader.Broadcast(func(dt TrackSender) { - writeCount.Add(dt.WriteRTP(pkt, spatialLayer)) - }) - return writeCount.Load() + return utils.BroadcastRTP(r.downTrackSpreader, pkt, spatialLayer) } redLen, err := r.encodeRedForPrimary(pkt.Packet, r.redPayloadBuf[:]) @@ -86,19 +86,17 @@ func (r *RedReceiver) ForwardRTP(pkt *buffer.ExtPacket, spatialLayer int32) int3 return 0 } - pPkt := *pkt - redRtpPacket := *pkt.Packet + redRtpPacket := &r.redRtpPkt + *redRtpPacket = *pkt.Packet redRtpPacket.PayloadType = opusRedPT redRtpPacket.Payload = r.redPayloadBuf[:redLen] - pPkt.Packet = &redRtpPacket + pPkt := &r.redExtPkt + *pPkt = *pkt + pPkt.Packet = redRtpPacket // not modify the ExtPacket.RawPacket here for performance since it is not used by the DownTrack, // otherwise it should be set to the correct value (marshal the primary rtp packet) - var writeCount atomic.Int32 - r.downTrackSpreader.Broadcast(func(dt TrackSender) { - writeCount.Add(dt.WriteRTP(&pPkt, spatialLayer)) - }) - return writeCount.Load() + return utils.BroadcastRTP(r.downTrackSpreader, pPkt, spatialLayer) } func (r *RedReceiver) ForwardRTCPSenderReport( diff --git a/pkg/sfu/utils/downtrackspreader.go b/pkg/sfu/utils/downtrackspreader.go index e06f6c945..29ba07f2d 100644 --- a/pkg/sfu/utils/downtrackspreader.go +++ b/pkg/sfu/utils/downtrackspreader.go @@ -15,16 +15,18 @@ package utils import ( + "runtime" "sync" + "sync/atomic" "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/logger" "github.com/livekit/protocol/utils" ) -type sender interface { - SubscriberID() livekit.ParticipantID -} +// 100µs is enough to amortize the overhead and provide sufficient load balancing. +// WriteRTP takes about 50µs on average, so we write to 2 down tracks per loop. +const broadcastStep = 2 type DownTrackSpreaderParams struct { Threshold int @@ -90,23 +92,26 @@ func (d *DownTrackSpreader[T]) HasDownTrack(subscriberID livekit.ParticipantID) return ok } -func (d *DownTrackSpreader[T]) Broadcast(writer func(T)) { +// snapshot returns the current down tracks and the parallelization threshold +func (d *DownTrackSpreader[T]) snapshot() ([]T, int) { d.downTrackMu.RLock() downTracks := d.downTracksShadow - threshold := uint64(d.params.Threshold) + threshold := d.params.Threshold d.downTrackMu.RUnlock() - if len(downTracks) == 0 { - return - } if threshold == 0 { threshold = 1000000 } + return downTracks, threshold +} - // 100µs is enough to amortize the overhead and provide sufficient load balancing. - // WriteRTP takes about 50µs on average, so we write to 2 down tracks per loop. - step := uint64(2) - utils.ParallelExec(downTracks, threshold, step, writer) +func (d *DownTrackSpreader[T]) Broadcast(writer func(T)) { + downTracks, threshold := d.snapshot() + if len(downTracks) == 0 { + return + } + + utils.ParallelExec(downTracks, uint64(threshold), broadcastStep, writer) } func (d *DownTrackSpreader[T]) DownTrackCount() int { @@ -127,3 +132,69 @@ func (d *DownTrackSpreader[T]) SetThreshold(threshold int) { d.params.Threshold = threshold d.downTrackMu.Unlock() } + +// ------------------------------------------------ + +type sender interface { + SubscriberID() livekit.ParticipantID +} + +type rtpWriter[P any] interface { + WriteRTP(pkt P, layer int32) int32 +} + +// rtpBroadcast is the shared state of one parallel BroadcastRTP, it carries the +// packet and layer so that no closure has to be allocated per packet +type rtpBroadcast[T rtpWriter[P], P any] struct { + downTracks []T + pkt P + layer int32 + next atomic.Uint64 + written atomic.Int32 + wg sync.WaitGroup +} + +func (b *rtpBroadcast[T, P]) run() { + defer b.wg.Done() + + var written int32 + end := uint64(len(b.downTracks)) + for { + n := b.next.Add(broadcastStep) + if n >= end+broadcastStep { + break + } + for i := n - broadcastStep; i < n && i < end; i++ { + written += b.downTracks[i].WriteRTP(b.pkt, b.layer) + } + } + b.written.Add(written) +} + +// BroadcastRTP writes pkt to every down track and returns how many accepted it. +// Below the threshold it runs on the caller without allocating; above it, the +// only allocations are the shared state and the worker funcval. +func BroadcastRTP[T interface { + sender + rtpWriter[P] +}, P any](d *DownTrackSpreader[T], pkt P, layer int32) int32 { + downTracks, threshold := d.snapshot() + if len(downTracks) < threshold { + var written int32 + for _, dt := range downTracks { + written += dt.WriteRTP(pkt, layer) + } + return written + } + + numWorkers := min(runtime.NumCPU(), len(downTracks)) + b := &rtpBroadcast[T, P]{downTracks: downTracks, pkt: pkt, layer: layer} + b.wg.Add(numWorkers) + worker := b.run + for i := 0; i < numWorkers; i++ { + go worker() + } + b.wg.Wait() + + return b.written.Load() +} diff --git a/pkg/sfu/utils/downtrackspreader_test.go b/pkg/sfu/utils/downtrackspreader_test.go new file mode 100644 index 000000000..1de2d96ce --- /dev/null +++ b/pkg/sfu/utils/downtrackspreader_test.go @@ -0,0 +1,84 @@ +// Copyright 2026 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package utils + +import ( + "fmt" + "sync/atomic" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/livekit/protocol/livekit" +) + +type testPacket struct{ seq int } + +type testSender struct { + id livekit.ParticipantID + writes atomic.Int32 + accept int32 +} + +func (s *testSender) SubscriberID() livekit.ParticipantID { return s.id } + +func (s *testSender) WriteRTP(pkt *testPacket, layer int32) int32 { + s.writes.Add(1) + return s.accept +} + +func newTestSpreader(numDownTracks, threshold int) (*DownTrackSpreader[*testSender], []*testSender) { + d := NewDownTrackSpreader[*testSender](DownTrackSpreaderParams{Threshold: threshold}) + senders := make([]*testSender, numDownTracks) + for i := range senders { + senders[i] = &testSender{id: livekit.ParticipantID(fmt.Sprintf("p%d", i)), accept: 1} + d.Store(senders[i]) + } + return d, senders +} + +func TestBroadcastRTP(t *testing.T) { + for _, tc := range []struct { + name string + numDownTracks int + threshold int + maxAllocs float64 + }{ + {"serial", 5, 20, 0}, + {"parallel", 50, 20, 2}, // shared state and worker funcval + } { + t.Run(tc.name, func(t *testing.T) { + d, senders := newTestSpreader(tc.numDownTracks, tc.threshold) + senders[0].accept = 0 // one down track that drops + pkt := &testPacket{} + + written := BroadcastRTP(d, pkt, 2) + require.EqualValues(t, tc.numDownTracks-1, written) + for _, s := range senders { + require.EqualValues(t, 1, s.writes.Load()) + } + + if !RaceEnabled { + allocs := testing.AllocsPerRun(1000, func() { BroadcastRTP(d, pkt, 2) }) + require.LessOrEqual(t, allocs, tc.maxAllocs) + } + }) + } +} + +func TestBroadcastRTPEmpty(t *testing.T) { + d := NewDownTrackSpreader[*testSender](DownTrackSpreaderParams{Threshold: 20}) + require.EqualValues(t, 0, BroadcastRTP(d, &testPacket{}, 0)) +}