From d04f4d12d1ba0345439d4281f19ef0e07018e833 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Thu, 17 Feb 2022 13:33:44 +0530 Subject: [PATCH] Throttle RTX under certain conditions to prevent RTX storm (#440) * WIP commit * WIP commit * WIP commit * WIP commit * WIP commit * WIP commit * WIP commit * WIP commit * Clean up * Remove debug * Remove unneeded change * fix test * Remove incorrect comment * WIP commit * Reset probe after estimate trends down * WIP commit * variable name change * Clean up * Remove debug logs * gofmt --- pkg/sfu/downtrack.go | 63 ++++++++++++++++++++++++++++++-------- pkg/sfu/forwarder.go | 21 +++++++++---- pkg/sfu/rtpmunger.go | 31 +++++++++++++++++++ pkg/sfu/streamallocator.go | 3 ++ 4 files changed, 99 insertions(+), 19 deletions(-) diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index aeaca9b06..1b97d537c 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -38,6 +38,8 @@ const ( RTPBlankFramesMax = 6 firstKeyFramePLIInterval = 500 * time.Millisecond + + FlagStopRTXOnPLI = true ) var ( @@ -104,6 +106,8 @@ type DownTrack struct { lastRTP atomicInt64 pktsDropped atomicUint32 + isNACKThrottled atomicBool + // RTCP callbacks onREMB func(dt *DownTrack, remb *rtcp.ReceiverEstimatedMaximumBitrate) onTransportCCFeedback func(dt *DownTrack, cc *rtcp.TransportLayerCC) @@ -353,6 +357,9 @@ func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) error { } d.updatePrimaryStats(pktSize, hdr.Marker) + if extPkt.KeyFrame { + d.isNACKThrottled.set(false) + } } else { d.logger.Errorw("writing rtp packet err", err) d.pktsDropped.add(1) @@ -643,7 +650,9 @@ func (d *DownTrack) DistanceToDesired() int32 { } func (d *DownTrack) Allocate(availableChannelCapacity int64, allowPause bool) VideoAllocation { - return d.forwarder.Allocate(availableChannelCapacity, allowPause, d.receiver.GetBitrateTemporalCumulative()) + allocation := d.forwarder.Allocate(availableChannelCapacity, allowPause, d.receiver.GetBitrateTemporalCumulative()) + d.logger.Debugw("stream: allocation", "channel", availableChannelCapacity, "allocation", allocation) + return allocation } func (d *DownTrack) ProvisionalAllocatePrepare() { @@ -655,15 +664,21 @@ func (d *DownTrack) ProvisionalAllocate(availableChannelCapacity int64, layers V } func (d *DownTrack) ProvisionalAllocateGetCooperativeTransition() VideoTransition { - return d.forwarder.ProvisionalAllocateGetCooperativeTransition() + transition := d.forwarder.ProvisionalAllocateGetCooperativeTransition() + d.logger.Debugw("stream: cooperative transition", "transition", transition) + return transition } func (d *DownTrack) ProvisionalAllocateGetBestWeightedTransition() VideoTransition { - return d.forwarder.ProvisionalAllocateGetBestWeightedTransition() + transition := d.forwarder.ProvisionalAllocateGetBestWeightedTransition() + d.logger.Debugw("stream: best weighted transition", "transition", transition) + return transition } func (d *DownTrack) ProvisionalAllocateCommit() VideoAllocation { - return d.forwarder.ProvisionalAllocateCommit() + allocation := d.forwarder.ProvisionalAllocateCommit() + d.logger.Debugw("stream: allocation commit", "allocation", allocation) + return allocation } func (d *DownTrack) FinalizeAllocate() VideoAllocation { @@ -671,15 +686,21 @@ func (d *DownTrack) FinalizeAllocate() VideoAllocation { } func (d *DownTrack) AllocateNextHigher(availableChannelCapacity int64) (VideoAllocation, bool) { - return d.forwarder.AllocateNextHigher(availableChannelCapacity, d.receiver.GetBitrateTemporalCumulative()) + allocation, available := d.forwarder.AllocateNextHigher(availableChannelCapacity, d.receiver.GetBitrateTemporalCumulative()) + d.logger.Debugw("stream: allocation next higher layer", "allocation", allocation, "available", available) + return allocation, available } func (d *DownTrack) GetNextHigherTransition() (VideoTransition, bool) { - return d.forwarder.GetNextHigherTransition(d.receiver.GetBitrateTemporalCumulative()) + transition, available := d.forwarder.GetNextHigherTransition(d.receiver.GetBitrateTemporalCumulative()) + d.logger.Debugw("stream: get next higher layer", "transition", transition, "available", available) + return transition, available } func (d *DownTrack) Pause() VideoAllocation { - return d.forwarder.Pause(d.receiver.GetBitrateTemporalCumulative()) + allocation := d.forwarder.Pause(d.receiver.GetBitrateTemporalCumulative()) + d.logger.Debugw("stream: pause", "allocation", allocation) + return allocation } func (d *DownTrack) Resync() { @@ -882,6 +903,7 @@ func (d *DownTrack) handleRTCP(bytes []byte) { if targetLayers != InvalidLayers { d.lastPli.set(time.Now().UnixNano()) d.receiver.SendPLI(targetLayers.spatial) + d.isNACKThrottled.set(true) pliOnce = false } } @@ -942,12 +964,12 @@ func (d *DownTrack) handleRTCP(bytes []byte) { } case *rtcp.TransportLayerNack: - var nackedPackets []packetMeta + var nacks []uint16 for _, pair := range p.Nacks { - nackedPackets = append(nackedPackets, d.sequencer.getSeqNoPairs(pair.PacketList())...) + nacks = append(nacks, pair.PacketList()...) } - go d.retransmitPackets(nackedPackets) - numNACKs += uint32(len(nackedPackets)) + go d.retransmitPackets(nacks) + numNACKs += uint32(len(nacks)) case *rtcp.TransportLayerCC: if p.MediaSSRC == d.ssrc && d.onTransportCCFeedback != nil { @@ -967,7 +989,22 @@ func (d *DownTrack) handleRTCP(bytes []byte) { } } -func (d *DownTrack) retransmitPackets(nackedPackets []packetMeta) { +func (d *DownTrack) retransmitPackets(nacks []uint16) { + if FlagStopRTXOnPLI && d.isNACKThrottled.get() { + return + } + + filtered, disallowedLayers := d.forwarder.FilterRTX(nacks) + if len(filtered) == 0 { + return + } + + if d.sequencer == nil { + return + } + + nackedPackets := d.sequencer.getSeqNoPairs(filtered) + var pool *[]byte defer func() { if pool != nil { @@ -980,7 +1017,7 @@ func (d *DownTrack) retransmitPackets(nackedPackets []packetMeta) { defer PacketFactory.Put(src) for _, meta := range nackedPackets { - if !d.forwarder.IsRtxAllowed(int32(meta.layer)) { + if disallowedLayers[meta.layer] { continue } diff --git a/pkg/sfu/forwarder.go b/pkg/sfu/forwarder.go index 7b303d1ca..1323010b5 100644 --- a/pkg/sfu/forwarder.go +++ b/pkg/sfu/forwarder.go @@ -17,6 +17,7 @@ import ( // const ( FlagPauseOnDowngrade = true + FlagFilterRTX = true ) type ForwardingStatus int @@ -1138,10 +1139,17 @@ func (f *Forwarder) Resync() { f.lastSSRC = 0 } -func (f *Forwarder) IsRtxAllowed(layer int32) bool { +func (f *Forwarder) FilterRTX(nacks []uint16) (filtered []uint16, disallowedLayers [DefaultMaxLayerSpatial + 1]bool) { + if !FlagFilterRTX { + filtered = nacks + return + } + f.lock.RLock() defer f.lock.RUnlock() + filtered = f.rtpMunger.FilterRTX(nacks) + // // Curb RTX when deficient for two cases // 1. Target layer is lower than current layer. When current hits target, a key frame should flush the decoder. @@ -1150,13 +1158,14 @@ func (f *Forwarder) IsRtxAllowed(layer int32) bool { // // Without the curb, when congestion hits, RTX rate could be so high that it further congests the channel. // - if FlagPauseOnDowngrade && - f.lastAllocation.state == VideoAllocationStateDeficient && - (f.targetLayers.spatial < f.currentLayers.spatial || layer > f.currentLayers.spatial) { - return false + for layer := int32(0); layer < DefaultMaxLayerSpatial+1; layer++ { + if f.lastAllocation.state == VideoAllocationStateDeficient && + (f.targetLayers.spatial < f.currentLayers.spatial || layer > f.currentLayers.spatial) { + disallowedLayers[layer] = true + } } - return true + return } func (f *Forwarder) GetTranslationParams(extPkt *buffer.ExtPacket, layer int32) (*TranslationParams, error) { diff --git a/pkg/sfu/rtpmunger.go b/pkg/sfu/rtpmunger.go index 938d2ef8f..1d7068425 100644 --- a/pkg/sfu/rtpmunger.go +++ b/pkg/sfu/rtpmunger.go @@ -18,6 +18,10 @@ const ( SequenceNumberOrderingDuplicate ) +const ( + RtxGateWindow = 2000 +) + type TranslationParamsRTP struct { snOrdering SequenceNumberOrdering sequenceNumber uint16 @@ -38,6 +42,9 @@ type RTPMungerParams struct { lastMarker bool missingSNs map[uint16]uint16 + + rtxGateSn uint16 + isInRtxGateRegion bool } type RTPMunger struct { @@ -155,6 +162,15 @@ func (r *RTPMunger) UpdateAndGetSnTs(extPkt *buffer.ExtPacket) (*TranslationPara r.lastTS = mungedTS r.lastMarker = extPkt.Packet.Marker + if extPkt.KeyFrame { + r.rtxGateSn = mungedSN + r.isInRtxGateRegion = true + } + + if r.isInRtxGateRegion && (mungedSN-r.rtxGateSn) > RtxGateWindow { + r.isInRtxGateRegion = false + } + return &TranslationParamsRTP{ snOrdering: ordering, sequenceNumber: mungedSN, @@ -162,6 +178,21 @@ func (r *RTPMunger) UpdateAndGetSnTs(extPkt *buffer.ExtPacket) (*TranslationPara }, nil } +func (r *RTPMunger) FilterRTX(nacks []uint16) []uint16 { + if !r.isInRtxGateRegion { + return nacks + } + + filtered := make([]uint16, 0, len(nacks)) + for _, sn := range nacks { + if (sn - r.rtxGateSn) < (1 << 15) { + filtered = append(filtered, sn) + } + } + + return filtered +} + func (r *RTPMunger) UpdateAndGetPaddingSnTs(num int, clockRate uint32, frameRate uint32, forceMarker bool) ([]SnTs, error) { tsOffset := 0 if !r.lastMarker { diff --git a/pkg/sfu/streamallocator.go b/pkg/sfu/streamallocator.go index 44c988f16..98a17c5bd 100644 --- a/pkg/sfu/streamallocator.go +++ b/pkg/sfu/streamallocator.go @@ -697,12 +697,15 @@ func (s *StreamAllocator) handleNewEstimateInProbe() { // In rare cases, the estimate gets stuck. Prevent from probe running amok // LK-TODO: Need more testing this here and ensure that probe does not cause a lot of damage // + s.params.Logger.Debugw("probe: aborting, no trend") s.abortProbe() case trend == EstimateTrendDownward: // stop immediately if estimate falls below the previously committed estimate, the probe is congesting channel more + s.params.Logger.Debugw("probe: aborting, estimate is trending downward") s.abortProbe() case s.probeEstimator.GetHighest() > s.probeGoalBps: // reached goal, stop probing + s.params.Logger.Debugw("probe: stopping, goal reached") s.stopProbe() } }