From 03ada9ba76463c5a0d1f078b692170f6207f341f Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Tue, 19 Mar 2024 11:59:24 +0530 Subject: [PATCH] Proper RTCP report past mute. (#2588) - When audio is muted, server injects silence frames which moves the time stamp forward and adjusts offset. That cannot be used against publisher side sender report. Use a pinned version. - Ignore small changes to propagation delay even while checking for sharp increase. That is spamming a lot for small changes, i.e. existing delta is 100 micro seconds or so and the new one is 300 micro seconds. Also rename to `longTerm` from `smoothed` as it is a slow varying long term estimate of propagation delay delta. And slow down that adaptation more. --- pkg/sfu/buffer/rtpstats_receiver.go | 74 +++++++++++++++-------------- pkg/sfu/forwarder.go | 4 +- pkg/sfu/rtpmunger.go | 9 +++- 3 files changed, 47 insertions(+), 40 deletions(-) diff --git a/pkg/sfu/buffer/rtpstats_receiver.go b/pkg/sfu/buffer/rtpstats_receiver.go index c49fe69d0..cbcb26794 100644 --- a/pkg/sfu/buffer/rtpstats_receiver.go +++ b/pkg/sfu/buffer/rtpstats_receiver.go @@ -48,10 +48,10 @@ const ( // To account for path changes mid-stream, if the delta of the propagation delay is consistently higher, reset. // Reset at whichever of the below happens later. // - // A smoothed version of delta of propagation delay is maintained and delta propagation delay exceeding - // a factor of the smoothed version is considered a sharp increase. That will trigger the start of the + // A long term version of delta of propagation delay is maintained and delta propagation delay exceeding + // a factor of the long term version is considered a sharp increase. That will trigger the start of the // path change condition and if it persists, propagation delay will be reset. - cPropagationDelayDeltaAdaptationFactor = float64(0.1) + cPropagationDelayDeltaAdaptationFactor = float64(0.05) cPropagationDelayDeltaHighResetNumReports = 3 cPropagationDelayDeltaHighResetWait = 10 * time.Second ) @@ -80,7 +80,7 @@ type RTPStatsReceiver struct { history *protoutils.Bitmap[uint64] propagationDelay time.Duration - smoothedDeltaPropagationDelay time.Duration + longTermDeltaPropagationDelay time.Duration propagationDelayDeltaHighCount int propagationDelayDeltaHighStartTime time.Time @@ -361,7 +361,7 @@ func (r *RTPStatsReceiver) SetRtcpSenderReportData(srData *RTCPSenderReportData) return []interface{}{ "propagationDelay", r.propagationDelay.String(), "receivedPropagationDelay", propagationDelay.String(), - "smoothedDeltaPropagationDelay", r.smoothedDeltaPropagationDelay.String(), + "longTermDeltaPropagationDelay", r.longTermDeltaPropagationDelay.String(), "receivedDeltaPropagationDelay", deltaPropagationDelay.String(), "deltaHighCount", r.propagationDelayDeltaHighCount, "sinceDeltaHighStart", time.Since(r.propagationDelayDeltaHighStartTime).String(), @@ -372,7 +372,7 @@ func (r *RTPStatsReceiver) SetRtcpSenderReportData(srData *RTCPSenderReportData) } initPropagationDelay := func(pd time.Duration) { r.propagationDelay = pd - r.smoothedDeltaPropagationDelay = 0 + r.longTermDeltaPropagationDelay = 0 r.propagationDelayDeltaHighCount = 0 r.propagationDelayDeltaHighStartTime = time.Time{} } @@ -385,39 +385,41 @@ func (r *RTPStatsReceiver) SetRtcpSenderReportData(srData *RTCPSenderReportData) r.logger.Debugw("initializing propagation delay", getPropagationFields()...) } else { deltaPropagationDelay = propagationDelay - r.propagationDelay - if r.smoothedDeltaPropagationDelay != 0 && deltaPropagationDelay > 0 && deltaPropagationDelay > r.smoothedDeltaPropagationDelay*time.Duration(cPropagationDelayDeltaThresholdMaxFactor) { - r.logger.Debugw("sharp increase in propagation delay, skipping", getPropagationFields()...) // TODO-REMOVE - r.propagationDelayDeltaHighCount++ - if r.propagationDelayDeltaHighStartTime.IsZero() { - r.propagationDelayDeltaHighStartTime = time.Now() - } - - if r.propagationDelayDeltaHighCount >= cPropagationDelayDeltaHighResetNumReports && time.Since(r.propagationDelayDeltaHighStartTime) >= cPropagationDelayDeltaHighResetWait { - r.logger.Debugw("re-initializing propagation delay", append(getPropagationFields(), "newPropagationDelay", propagationDelay.String())...) - initPropagationDelay(propagationDelay) - } - } else { - if r.smoothedDeltaPropagationDelay == 0 { - r.smoothedDeltaPropagationDelay = deltaPropagationDelay - } else { - r.smoothedDeltaPropagationDelay += time.Duration(cPropagationDelayDeltaAdaptationFactor * float64(deltaPropagationDelay-r.smoothedDeltaPropagationDelay)) - } - r.propagationDelayDeltaHighCount = 0 - r.propagationDelayDeltaHighStartTime = time.Time{} - - if deltaPropagationDelay.Abs() > cPropagationDelayDeltaThresholdMin { - factor := cPropagationDelayFallFactor - if propagationDelay > r.propagationDelay { - factor = cPropagationDelayRiseFactor + if deltaPropagationDelay.Abs() > cPropagationDelayDeltaThresholdMin { // ignore small changes + if r.longTermDeltaPropagationDelay != 0 && deltaPropagationDelay > 0 && deltaPropagationDelay > r.longTermDeltaPropagationDelay*time.Duration(cPropagationDelayDeltaThresholdMaxFactor) { + r.logger.Debugw("sharp increase in propagation delay, skipping", getPropagationFields()...) // TODO-REMOVE + r.propagationDelayDeltaHighCount++ + if r.propagationDelayDeltaHighStartTime.IsZero() { + r.propagationDelayDeltaHighStartTime = time.Now() + } + + if r.propagationDelayDeltaHighCount >= cPropagationDelayDeltaHighResetNumReports && time.Since(r.propagationDelayDeltaHighStartTime) >= cPropagationDelayDeltaHighResetWait { + r.logger.Debugw("re-initializing propagation delay", append(getPropagationFields(), "newPropagationDelay", propagationDelay.String())...) + initPropagationDelay(propagationDelay) + } + } else { + r.propagationDelayDeltaHighCount = 0 + r.propagationDelayDeltaHighStartTime = time.Time{} + + if deltaPropagationDelay.Abs() > cPropagationDelayDeltaThresholdMin { + factor := cPropagationDelayFallFactor + if propagationDelay > r.propagationDelay { + factor = cPropagationDelayRiseFactor + } + fields := append( + getPropagationFields(), + "adjustedPropagationDelay", r.propagationDelay+time.Duration(factor*float64(propagationDelay-r.propagationDelay)), + ) // TODO-REMOVE + r.logger.Debugw("adapting propagation delay", fields...) // TODO-REMOVE + r.propagationDelay += time.Duration(factor * float64(propagationDelay-r.propagationDelay)) } - fields := append( - getPropagationFields(), - "adjustedPropagationDelay", r.propagationDelay+time.Duration(factor*float64(propagationDelay-r.propagationDelay)), - ) // TODO-REMOVE - r.logger.Debugw("adapting propagation delay", fields...) // TODO-REMOVE - r.propagationDelay += time.Duration(factor * float64(propagationDelay-r.propagationDelay)) } } + if r.longTermDeltaPropagationDelay == 0 { + r.longTermDeltaPropagationDelay = deltaPropagationDelay + } else { + r.longTermDeltaPropagationDelay += time.Duration(cPropagationDelayDeltaAdaptationFactor * float64(deltaPropagationDelay-r.longTermDeltaPropagationDelay)) + } } // adjust receive time to estimated propagation delay srDataCopy.At = ntpTime.Add(r.propagationDelay) diff --git a/pkg/sfu/forwarder.go b/pkg/sfu/forwarder.go index e2bc0f1c9..de50a829b 100644 --- a/pkg/sfu/forwarder.go +++ b/pkg/sfu/forwarder.go @@ -561,10 +561,10 @@ func (f *Forwarder) GetCurrentSpatialAndTSOffset() (int32, uint64) { defer f.lock.RUnlock() if f.kind == webrtc.RTPCodecTypeAudio { - return 0, f.rtpMunger.GetTSOffset() + return 0, f.rtpMunger.GetPinnedTSOffset() } - return f.vls.GetCurrent().Spatial, f.rtpMunger.GetTSOffset() + return f.vls.GetCurrent().Spatial, f.rtpMunger.GetPinnedTSOffset() } func (f *Forwarder) isDeficientLocked() bool { diff --git a/pkg/sfu/rtpmunger.go b/pkg/sfu/rtpmunger.go index 49c918037..47287e84a 100644 --- a/pkg/sfu/rtpmunger.go +++ b/pkg/sfu/rtpmunger.go @@ -83,6 +83,7 @@ type RTPMunger struct { extLastTS uint64 extSecondLastTS uint64 tsOffset uint64 + pinnedTSOffset uint64 lastMarker bool secondLastMarker bool @@ -107,6 +108,7 @@ func (r *RTPMunger) DebugInfo() map[string]interface{} { "ExtLastTS": r.extLastTS, "ExtSecondLastTS": r.extSecondLastTS, "TSOffset": r.tsOffset, + "PinnedTSOffset": r.pinnedTSOffset, "LastMarker": r.lastMarker, "SecondLastMarker": r.secondLastMarker, } @@ -123,8 +125,8 @@ func (r *RTPMunger) GetLast() RTPMungerState { } } -func (r *RTPMunger) GetTSOffset() uint64 { - return r.tsOffset +func (r *RTPMunger) GetPinnedTSOffset() uint64 { + return r.pinnedTSOffset } func (r *RTPMunger) SeedLast(state RTPMungerState) { @@ -146,6 +148,8 @@ func (r *RTPMunger) SetLastSnTs(extPkt *buffer.ExtPacket) { r.extLastTS = extPkt.ExtTimestamp r.extSecondLastTS = extPkt.ExtTimestamp + r.tsOffset = 0 + r.pinnedTSOffset = r.tsOffset } func (r *RTPMunger) UpdateSnTsOffsets(extPkt *buffer.ExtPacket, snAdjust uint64, tsAdjust uint64) { @@ -155,6 +159,7 @@ func (r *RTPMunger) UpdateSnTsOffsets(extPkt *buffer.ExtPacket, snAdjust uint64, r.updateSnOffset() r.tsOffset = extPkt.ExtTimestamp - r.extLastTS - tsAdjust + r.pinnedTSOffset = r.tsOffset } func (r *RTPMunger) PacketDropped(extPkt *buffer.ExtPacket) {