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 <noreply@anthropic.com> * 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 <noreply@anthropic.com> * sfu: skip BroadcastRTP allocation assertion under the race detector Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * sfu: group BroadcastRTP and its interfaces after the spreader methods Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> --------- Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
2026-09-16 10:42:36 +00:00 · 2026-09-16 11:53:54 +05:30
parent 4ea0facf85
commit 48700da3f3
5 changed files with 192 additions and 43 deletions
@@ -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,
 				)
@@ -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(
@@ -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(
@@ -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()
+}
@@ -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))
+}