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 +}