From ee3a7c01bcb864aad348c4b1b6734d0b0d5688da Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Sat, 30 Sep 2023 08:42:46 +0530 Subject: [PATCH] Log resync at Infow. (#2114) * Log resync at Infow. Seeing potentially large sequence number jumps on a resync. And it seems to happen on a lot of subscribe/unsubscribe. Logging at Infow to understand better. Probably need to find a way to avoid resync. But, logging for now to check if I can catch one. * Remove resync and log large sequence number jumps --- pkg/sfu/buffer/rtpstats_base.go | 10 +-- pkg/sfu/buffer/rtpstats_receiver.go | 100 ++-------------------------- pkg/sfu/buffer/rtpstats_sender.go | 4 ++ 3 files changed, 15 insertions(+), 99 deletions(-) diff --git a/pkg/sfu/buffer/rtpstats_base.go b/pkg/sfu/buffer/rtpstats_base.go index d25341c8b..a024d18c5 100644 --- a/pkg/sfu/buffer/rtpstats_base.go +++ b/pkg/sfu/buffer/rtpstats_base.go @@ -106,11 +106,11 @@ type snapshot struct { } type RTCPSenderReportData struct { - RTPTimestamp uint32 - RTPTimestampExt uint64 - NTPTimestamp mediatransportutil.NtpTime - PacketCount uint32 - PacketCountExt uint64 + RTPTimestamp uint32 + RTPTimestampExt uint64 + NTPTimestamp mediatransportutil.NtpTime + PacketCount uint32 + // RAJA-REMOVE PacketCountExt uint64 PaddingOnlyDrops uint64 At time.Time } diff --git a/pkg/sfu/buffer/rtpstats_receiver.go b/pkg/sfu/buffer/rtpstats_receiver.go index fff35b2f5..ae42dc426 100644 --- a/pkg/sfu/buffer/rtpstats_receiver.go +++ b/pkg/sfu/buffer/rtpstats_receiver.go @@ -46,9 +46,6 @@ type RTPFlowState struct { type RTPStatsReceiver struct { *rtpStatsBase - resyncOnNextPacket bool - shouldDiscountPaddingOnlyDrops bool - sequenceNumber *utils.WrapAround[uint16, uint64] timestamp *utils.WrapAround[uint32, uint64] @@ -89,11 +86,6 @@ func (r *RTPStatsReceiver) Update( return } - if r.resyncOnNextPacket { - r.resyncOnNextPacket = false - r.resync(packetTime, sequenceNumber, timestamp) - } - var resSN utils.WrapAroundUpdateResult[uint64] var resTS utils.WrapAroundUpdateResult[uint64] if !r.initialized { @@ -136,11 +128,11 @@ func (r *RTPStatsReceiver) Update( if payloadSize == 0 { // do not start on a padding only packet if resTS.IsRestart { - r.logger.Infow("rolling back timestamp restart", "tsBefore", r.timestamp.GetExtendedStart(), "tsAfter", resTS.PreExtendedStart) + r.logger.Infow("rolling back timestamp restart", "tsAfter", r.timestamp.GetExtendedStart(), "tsBefore", resTS.PreExtendedStart) r.timestamp.RollbackRestart(resTS.PreExtendedStart) } if resSN.IsRestart { - r.logger.Infow("rolling back sequence number restart", "snBefore", r.sequenceNumber.GetExtendedStart(), "snAfter", resSN.PreExtendedStart) + r.logger.Infow("rolling back sequence number restart", "snAfter", r.sequenceNumber.GetExtendedStart(), "snBefore", resSN.PreExtendedStart) r.sequenceNumber.RollbackRestart(resSN.PreExtendedStart) return } @@ -192,6 +184,10 @@ func (r *RTPStatsReceiver) Update( flowState.ExtSequenceNumber = resSN.ExtendedVal flowState.ExtTimestamp = resTS.ExtendedVal } else { // in-order + if gapSN >= cNumSequenceNumbers { + r.logger.Warnw("large sequence number gap", nil, "prev", resSN.PreExtendedHighest, "curr", resSN.ExtendedVal, "gap", gapSN) + } + // update gap histogram r.updateGapHistogram(int(gapSN)) @@ -235,83 +231,6 @@ func (r *RTPStatsReceiver) Update( return } -func (r *RTPStatsReceiver) ResyncOnNextPacket(shouldDiscountPaddingOnlyDrops bool) { - r.lock.Lock() - defer r.lock.Unlock() - - r.resyncOnNextPacket = true - r.shouldDiscountPaddingOnlyDrops = shouldDiscountPaddingOnlyDrops -} - -func (r *RTPStatsReceiver) resync(packetTime time.Time, sn uint16, ts uint32) { - if !r.initialized { - return - } - - extHighestSN := r.sequenceNumber.GetExtendedHighest() - var newestPacketCount uint64 - var paddingOnlyDrops uint64 - var extExpectedHighestSN uint64 - var expectedHighestSN uint16 - var snCycles uint64 - - extHighestTS := r.timestamp.GetExtendedHighest() - var newestTS uint64 - var extExpectedHighestTS uint64 - var expectedHighestTS uint32 - var tsCycles uint64 - if r.srNewest != nil { - newestPacketCount = r.srNewest.PacketCountExt - paddingOnlyDrops = r.srNewest.PaddingOnlyDrops - if newestPacketCount != 0 { - extExpectedHighestSN = r.sequenceNumber.GetExtendedStart() + newestPacketCount - if r.shouldDiscountPaddingOnlyDrops { - extExpectedHighestSN -= paddingOnlyDrops - } - expectedHighestSN = uint16(extExpectedHighestSN & 0xFFFF) - snCycles = extExpectedHighestSN & 0xFFFF_FFFF_FFFF_0000 - if sn-expectedHighestSN < (1<<15) && sn < expectedHighestSN { - snCycles += (1 << 16) - } - if snCycles != 0 && expectedHighestSN-sn < (1<<15) && expectedHighestSN < sn { - snCycles -= (1 << 16) - } - } - - newestTS = r.srNewest.RTPTimestampExt - extExpectedHighestTS = newestTS - expectedHighestTS = uint32(extExpectedHighestTS & 0xFFFF_FFFF) - tsCycles = extExpectedHighestTS & 0xFFFF_FFFF_0000_0000 - if ts-expectedHighestTS < (1<<31) && ts < expectedHighestTS { - tsCycles += (1 << 32) - } - if tsCycles != 0 && expectedHighestTS-ts < (1<<31) && expectedHighestTS < ts { - tsCycles -= (1 << 32) - } - } - r.sequenceNumber.ResetHighest(snCycles + uint64(sn) - 1) - r.timestamp.ResetHighest(tsCycles + uint64(ts)) - r.highestTime = packetTime - r.logger.Debugw( - "resync", - "newestPacketCount", newestPacketCount, - "paddingOnlyDrops", paddingOnlyDrops, - "extExpectedHighestSN", extExpectedHighestSN, - "expectedHighestSN", expectedHighestSN, - "snCycles", snCycles, - "rtpSN", sn, - "beforeExtHighestSN", extHighestSN, - "afterExtHighestSN", r.sequenceNumber.GetExtendedHighest(), - "newestTS", newestTS, - "extExpectedHighestTS", extExpectedHighestTS, - "expectedHighestTS", expectedHighestTS, - "tsCycles", tsCycles, - "rtpTS", ts, - "beforeExtHighestTS", extHighestTS, - "afterExtHighestTS", r.timestamp.GetExtendedHighest(), - ) -} - func (r *RTPStatsReceiver) SetRtcpSenderReportData(srData *RTCPSenderReportData) { r.lock.Lock() defer r.lock.Unlock() @@ -335,22 +254,15 @@ func (r *RTPStatsReceiver) SetRtcpSenderReportData(srData *RTCPSenderReportData) } tsCycles := uint64(0) - pcCycles := uint64(0) if r.srNewest != nil { tsCycles = r.srNewest.RTPTimestampExt & 0xFFFF_FFFF_0000_0000 if (srData.RTPTimestamp-r.srNewest.RTPTimestamp) < (1<<31) && srData.RTPTimestamp < r.srNewest.RTPTimestamp { tsCycles += (1 << 32) } - - pcCycles = r.srNewest.PacketCountExt & 0xFFFF_FFFF_0000_0000 - if (srData.PacketCount-r.srNewest.PacketCount) < (1<<31) && srData.PacketCount < r.srNewest.PacketCount { - pcCycles += (1 << 32) - } } srDataCopy := *srData srDataCopy.RTPTimestampExt = uint64(srDataCopy.RTPTimestamp) + tsCycles - srDataCopy.PacketCountExt = uint64(srDataCopy.PacketCount) + pcCycles r.maybeAdjustFirstPacketTime(srDataCopy.RTPTimestampExt, r.timestamp.GetExtendedStart()) diff --git a/pkg/sfu/buffer/rtpstats_sender.go b/pkg/sfu/buffer/rtpstats_sender.go index 5a66c65d4..bb8ca227d 100644 --- a/pkg/sfu/buffer/rtpstats_sender.go +++ b/pkg/sfu/buffer/rtpstats_sender.go @@ -302,6 +302,10 @@ func (r *RTPStatsSender) Update( r.setSnInfo(extSequenceNumber, r.extHighestSN, uint16(pktSize), uint8(hdrSize), uint16(payloadSize), marker, true) } } else { // in-order + if gapSN >= cNumSequenceNumbers { + r.logger.Warnw("large sequence number gap", nil, "prev", r.extHighestSN, "curr", extSequenceNumber, "gap", gapSN) + } + // update gap histogram r.updateGapHistogram(int(gapSN))