From fa01297d96624a6773336d61b8706778db65cb2f Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Wed, 25 Oct 2023 23:12:14 +0530 Subject: [PATCH] Slight sequencer tweaks. (#2184) The buffer is not for padding packets. So, calculate adjusted sequence numbers before comparing against size. Also, it is possible that invalidated slot is accessed due to not being able to exclude padding range. This was causing time stamp reset to 0. Will remove the error log after this goes out and the condition does not show up for a few days. --- pkg/sfu/buffer/rtpstats_receiver.go | 4 +- pkg/sfu/buffer/rtpstats_sender.go | 8 +- pkg/sfu/sequencer.go | 114 ++++++++++++++++++++-------- 3 files changed, 89 insertions(+), 37 deletions(-) diff --git a/pkg/sfu/buffer/rtpstats_receiver.go b/pkg/sfu/buffer/rtpstats_receiver.go index a5d102c61..3869ebae8 100644 --- a/pkg/sfu/buffer/rtpstats_receiver.go +++ b/pkg/sfu/buffer/rtpstats_receiver.go @@ -154,7 +154,7 @@ func (r *RTPStatsReceiver) Update( return } } - if -gapSN >= cNumSequenceNumbers { + if -gapSN >= cNumSequenceNumbers/2 { r.logger.Warnw( "large sequence number gap negative", nil, "extStartSN", r.sequenceNumber.GetExtendedStart(), @@ -226,7 +226,7 @@ func (r *RTPStatsReceiver) Update( flowState.ExtSequenceNumber = resSN.ExtendedVal flowState.ExtTimestamp = resTS.ExtendedVal } else { // in-order - if gapSN >= cNumSequenceNumbers { + if gapSN >= cNumSequenceNumbers/2 { r.logger.Warnw( "large sequence number gap", nil, "extStartSN", r.sequenceNumber.GetExtendedStart(), diff --git a/pkg/sfu/buffer/rtpstats_sender.go b/pkg/sfu/buffer/rtpstats_sender.go index 8efb6cef2..ff262c11c 100644 --- a/pkg/sfu/buffer/rtpstats_sender.go +++ b/pkg/sfu/buffer/rtpstats_sender.go @@ -286,7 +286,7 @@ func (r *RTPStatsSender) Update( // do not start on a padding only packet return } - if -gapSN >= cNumSequenceNumbers { + if -gapSN >= cNumSequenceNumbers/2 { r.logger.Warnw( "large sequence number gap negative", nil, "extStartSN", r.extStartSN, @@ -352,7 +352,7 @@ func (r *RTPStatsSender) Update( r.setSnInfo(extSequenceNumber, r.extHighestSN, uint16(pktSize), uint8(hdrSize), uint16(payloadSize), marker, true) } } else { // in-order - if gapSN >= cNumSequenceNumbers { + if gapSN >= cNumSequenceNumbers/2 { r.logger.Warnw( "large sequence number gap", nil, "extStartSN", r.extStartSN, @@ -399,8 +399,12 @@ func (r *RTPStatsSender) Update( "invalid start timestamp", nil, "snBefore", r.extStartSN, "snAfter", extSequenceNumber, + "snHighest", r.extHighestSN, "tsBefore", r.extStartTS, "tsAfter", extTimestamp, + "tsHighest", r.extHighestTS, + "firstTime", r.firstTime.String(), + "startTime", r.startTime.String(), ) } if extTimestamp != 0 { diff --git a/pkg/sfu/sequencer.go b/pkg/sfu/sequencer.go index ee6ed8bdb..4e26a8735 100644 --- a/pkg/sfu/sequencer.go +++ b/pkg/sfu/sequencer.go @@ -84,6 +84,7 @@ type sequencer struct { size int startTime int64 initialized bool + extStartSN uint64 extHighestSN uint64 snOffset uint64 extHighestTS uint64 @@ -133,46 +134,61 @@ func (s *sequencer) push( if !s.initialized { s.initialized = true - s.extHighestSN = extModifiedSN - 1 + s.extStartSN = extModifiedSN + s.extHighestSN = extModifiedSN s.extHighestTS = extModifiedTS s.updateSNOffset() } - snOffset := s.snOffset - diff := int64(extModifiedSN - s.extHighestSN) - if diff >= 0 { - s.extHighestSN = extModifiedSN - } else { - if diff < -int64(s.size) { - s.logger.Warnw( - "old packet, cannot be sequenced", nil, - "extHighestSN", s.extHighestSN, - "extIncomingSN", extIncomingSN, - "extModifiedSN", extModifiedSN, - ) - return - } + if extModifiedSN < s.extStartSN { + // old packet, should not happen + return + } + extHighestSNAdjusted := s.extHighestSN - s.snOffset + extModifiedSNAdjusted := extModifiedSN - s.snOffset + if extModifiedSN < s.extHighestSN { if s.snRangeMap != nil { - var err error - snOffset, err = s.snRangeMap.GetValue(extModifiedSN) + snOffset, err := s.snRangeMap.GetValue(extModifiedSN) if err != nil { s.logger.Errorw( "could not get sequence number offset", err, + "extStartSN", s.extStartSN, "extHighestSN", s.extHighestSN, "extIncomingSN", extIncomingSN, "extModifiedSN", extModifiedSN, + "snOffset", s.snOffset, ) return } + + extModifiedSNAdjusted = extModifiedSN - snOffset + } + } + + if int64(extModifiedSNAdjusted-extHighestSNAdjusted) <= -int64(s.size) { + s.logger.Warnw( + "old packet, cannot be sequenced", nil, + "extHighestSN", s.extHighestSN, + "extIncomingSN", extIncomingSN, + "extModifiedSN", extModifiedSN, + ) + return + } + + // invalidate missing sequence numbers + if extModifiedSNAdjusted > extHighestSNAdjusted { + numInvalidated := 0 + for esn := extHighestSNAdjusted + 1; esn != extModifiedSNAdjusted; esn++ { + s.invalidateSlot(int(esn % uint64(s.size))) + numInvalidated++ + if numInvalidated >= s.size { + break + } } } - if int64(extModifiedTS-s.extHighestTS) >= 0 { - s.extHighestTS = extModifiedTS - } - - slot := (extModifiedSN - snOffset) % uint64(s.size) + slot := extModifiedSNAdjusted % uint64(s.size) s.meta[slot] = packetMeta{ sourceSeqNo: uint16(extIncomingSN), targetSeqNo: uint16(extModifiedSN), @@ -183,6 +199,13 @@ func (s *sequencer) push( ddBytes: append([]byte{}, ddBytes...), lastNack: s.getRefTime(packetTime), // delay retransmissions after the original transmission } + + if extModifiedSN > s.extHighestSN { + s.extHighestSN = extModifiedSN + } + if extModifiedTS > s.extHighestTS { + s.extHighestTS = extModifiedTS + } } func (s *sequencer) pushPadding(extStartSNInclusive uint64, extEndSNInclusive uint64) { @@ -223,10 +246,7 @@ func (s *sequencer) pushPadding(extStartSNInclusive uint64, extEndSNInclusive ui } slot := (sn - snOffset) % uint64(s.size) - s.meta[slot] = packetMeta{ - sourceSeqNo: 0, - targetSeqNo: 0, - } + s.invalidateSlot(int(slot)) } return } @@ -244,6 +264,10 @@ func (s *sequencer) getExtPacketMetas(seqNo []uint16) []extPacketMeta { s.Lock() defer s.Unlock() + if !s.initialized { + return nil + } + snOffset := uint64(0) var err error extPacketMetas := make([]extPacketMeta, 0, len(seqNo)) @@ -263,11 +287,6 @@ func (s *sequencer) getExtPacketMetas(seqNo []uint16) []extPacketMeta { extSN -= (1 << 16) } - if s.extHighestSN-extSN >= uint64(s.size) { - // too old - continue - } - if s.snRangeMap != nil { snOffset, err = s.snRangeMap.GetValue(extSN) if err != nil { @@ -276,9 +295,17 @@ func (s *sequencer) getExtPacketMetas(seqNo []uint16) []extPacketMeta { } } - slot := (extSN - snOffset) % uint64(s.size) + extSNAdjusted := extSN - snOffset + extHighestSNAdjusted := s.extHighestSN - s.snOffset + if extHighestSNAdjusted-extSNAdjusted >= uint64(s.size) { + // too old + continue + } + + slot := extSNAdjusted % uint64(s.size) meta := &s.meta[slot] - if meta.targetSeqNo != sn { + if meta.targetSeqNo != sn || s.isInvalidSlot(int(slot)) { + // invalid slot access could happen if padding packets exclusion range could not be recorded continue } @@ -320,3 +347,24 @@ func (s *sequencer) updateSNOffset() { } s.snOffset = snOffset } + +func (s *sequencer) invalidateSlot(slot int) { + if slot >= len(s.meta) { + return + } + + s.meta[slot] = packetMeta{ + sourceSeqNo: 0, + targetSeqNo: 0, + lastNack: 0, + } +} + +func (s *sequencer) isInvalidSlot(slot int) bool { + if slot >= len(s.meta) { + return true + } + + meta := &s.meta[slot] + return meta.sourceSeqNo == 0 && meta.targetSeqNo == 0 && meta.lastNack == 0 +}