From 19b84207a16a69de3979026ead5e87e7595654e4 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Fri, 11 Aug 2023 09:43:12 +0530 Subject: [PATCH] Use 32-bit ts for first packet adjustment. (#1955) * Use 32-bit ts for first packet adjustment. Otherwise, a new subscriber on a long running sees a huge difference if the publisher side has rolled over. As this happens only in the first two minutes of a track's lifecycle it is fine to not consider rollover. * log RTP in anachronous report --- pkg/sfu/buffer/rtpstats.go | 19 +++++++++++++------ 1 file changed, 13 insertions(+), 6 deletions(-) diff --git a/pkg/sfu/buffer/rtpstats.go b/pkg/sfu/buffer/rtpstats.go index 0eed03660..f73e1676b 100644 --- a/pkg/sfu/buffer/rtpstats.go +++ b/pkg/sfu/buffer/rtpstats.go @@ -771,11 +771,11 @@ func (r *RTPStats) MaybeAdjustFirstPacketTime(srData *RTCPSenderReportData) { defer r.lock.Unlock() if srData != nil { - r.maybeAdjustFirstPacketTime(srData.RTPTimestampExt) + r.maybeAdjustFirstPacketTime(srData.RTPTimestamp) } } -func (r *RTPStats) maybeAdjustFirstPacketTime(extTS uint64) { +func (r *RTPStats) maybeAdjustFirstPacketTime(ts uint32) { if time.Since(r.startTime) > firstPacketTimeAdjustWindow { return } @@ -786,7 +786,12 @@ func (r *RTPStats) maybeAdjustFirstPacketTime(extTS uint64) { // abnormal delay (maybe due to pacing or maybe due to queuing // in some network element along the way), push back first time // to an earlier instance. - samplesDuration := time.Duration(float64(extTS-r.extStartTS) / float64(r.params.ClockRate) * float64(time.Second)) + samplesDiff := int32(ts - uint32(r.extStartTS)) + if samplesDiff < 0 { + // out-of-order, skip + return + } + samplesDuration := time.Duration(float64(samplesDiff) / float64(r.params.ClockRate) * float64(time.Second)) firstTime := time.Now().Add(-samplesDuration) if firstTime.Before(r.firstTime) { r.logger.Infow( @@ -814,8 +819,10 @@ func (r *RTPStats) SetRtcpSenderReportData(srData *RTCPSenderReportData) { if r.srNewest != nil && r.srNewest.NTPTimestamp > srData.NTPTimestamp { r.logger.Infow( "received anachronous sender report", - "current", srData.NTPTimestamp.Time(), - "last", r.srNewest.NTPTimestamp.Time(), + "currentNTP", srData.NTPTimestamp.Time(), + "currentRTP", srData.RTPTimestamp, + "lastNTP", r.srNewest.NTPTimestamp.Time(), + "lastRTP", r.srNewest.RTPTimestamp, ) return } @@ -831,7 +838,7 @@ func (r *RTPStats) SetRtcpSenderReportData(srData *RTCPSenderReportData) { srDataCopy := *srData srDataCopy.RTPTimestampExt = uint64(srDataCopy.RTPTimestamp) + cycles - r.maybeAdjustFirstPacketTime(srDataCopy.RTPTimestampExt) + r.maybeAdjustFirstPacketTime(srDataCopy.RTPTimestamp) // monitor and log RTP timestamp anomalies var ntpDiffSinceLast time.Duration