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)) +}