dandri/livekit
1
0
Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-08-06 23:39:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

Log resync at Infow. (#2114)

Browse Source 
* 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
This commit is contained in:
Raja Subramanian
2023-09-30 08:42:46 +05:30
committed by GitHub
parent 39d0af4381
commit ee3a7c01bc
3 changed files with 15 additions and 99 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/sfu/buffer/rtpstats_base.go
+5 -5
View File
@@ -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
}
pkg/sfu/buffer/rtpstats_receiver.go
+6 -94
View File
@@ -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())
pkg/sfu/buffer/rtpstats_sender.go
+4
View File
@@ -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))