diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index d30de32a4..dc2bb15eb 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -65,6 +65,9 @@ const ( keyFrameIntervalMax = 1000 flushTimeout = 1 * time.Second + // upper bound on NACKed sequence numbers buffered between retransmit worker runs + maxPendingNACKs = 5000 + waitBeforeSendPaddingOnMute = 100 * time.Millisecond maxPaddingOnMuteDuration = 5 * time.Second paddingOnMuteInterval = 100 * time.Millisecond @@ -336,6 +339,12 @@ type DownTrack struct { keyFrameRequesterCh chan struct{} keyFrameRequesterChClosed bool + retransmitChMu sync.RWMutex + retransmitCh chan struct{} + retransmitChClosed bool + pendingNACKsMu sync.Mutex + pendingNACKs []uint16 + createdAt int64 lastUnmutedAt atomic.Time } @@ -363,6 +372,7 @@ func NewDownTrack(params DownTrackParams) (*DownTrack, error) { pacer: params.Pacer, maxLayerNotifierCh: make(chan string, 1), keyFrameRequesterCh: make(chan struct{}, 1), + retransmitCh: make(chan struct{}, 1), createdAt: time.Now().UnixNano(), receiver: params.Receiver, } @@ -422,6 +432,8 @@ func NewDownTrack(params DownTrackParams) (*DownTrack, error) { go d.keyFrameRequester() } + go d.retransmitWorker() + d.params.Receiver.AddOnReady(d.handleReceiverReady) d.rtxSequenceNumber.Store(uint64(rand.Intn(1<<14)) + uint64(1<<15)) // a random number in third quartile of sequence number space d.params.Logger.Debugw("downtrack created", "upstreamCodecs", d.upstreamCodecs) @@ -1461,6 +1473,11 @@ func (d *DownTrack) CloseWithFlush(flush bool, isEnding bool) { close(d.keyFrameRequesterCh) d.keyFrameRequesterChMu.Unlock() + d.retransmitChMu.Lock() + d.retransmitChClosed = true + close(d.retransmitCh) + d.retransmitChMu.Unlock() + d.params.Listener.OnDownTrackClose(!flush) } @@ -2032,7 +2049,7 @@ func (d *DownTrack) handleRTCP(bytes []byte) { numNACKs += uint32(len(packetList)) nacks = append(nacks, packetList...) } - go d.retransmitPackets(nacks) + d.postRetransmitPackets(nacks) } case *rtcp.TransportLayerCC: @@ -2253,6 +2270,49 @@ func (d *DownTrack) retransmitPacket(epm *extPacketMeta, sourcePkt []byte, isPro return headerSize + len(payload), nil } +// postRetransmitPackets buffers the NACKed sequence numbers, capped at +// maxPendingNACKs, and signals the retransmit worker without blocking. +func (d *DownTrack) postRetransmitPackets(nacks []uint16) { + if d.sequencer == nil || len(nacks) == 0 { + return + } + + d.pendingNACKsMu.Lock() + d.pendingNACKs = append(d.pendingNACKs, nacks...) + if len(d.pendingNACKs) > maxPendingNACKs { + // drop the oldest NACKs under a NACK flood + d.pendingNACKs = d.pendingNACKs[len(d.pendingNACKs)-maxPendingNACKs:] + } + d.pendingNACKsMu.Unlock() + + d.retransmitChMu.RLock() + if !d.retransmitChClosed { + select { + case d.retransmitCh <- struct{}{}: + default: + } + } + d.retransmitChMu.RUnlock() +} + +func (d *DownTrack) retransmitWorker() { + for { + _, more := <-d.retransmitCh + if !more { + return + } + + d.pendingNACKsMu.Lock() + nacks := d.pendingNACKs + d.pendingNACKs = nil + d.pendingNACKsMu.Unlock() + + if len(nacks) > 0 { + d.retransmitPackets(nacks) + } + } +} + func (d *DownTrack) retransmitPackets(nacks []uint16) { if d.sequencer == nil { return