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

@@ -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
 		}
 

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) {

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 {

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