From a48660fa7738f52c12b6d2d0adf48e3ad67dadf1 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Sun, 27 Aug 2023 21:26:31 +0530 Subject: [PATCH] Make extended sequence number 64-bit. (#2003) --- pkg/sfu/buffer/buffer.go | 6 +- pkg/sfu/buffer/buffer_test.go | 2 +- pkg/sfu/buffer/helpers.go | 2 +- pkg/sfu/buffer/rtpstats.go | 140 ++++++++++++++++++-------------- pkg/sfu/buffer/rtpstats_test.go | 22 ++--- pkg/sfu/downtrack.go | 2 +- pkg/sfu/rtpmunger.go | 16 ++-- pkg/sfu/rtpmunger_test.go | 90 ++++++++++---------- pkg/sfu/testutils/data.go | 2 +- pkg/sfu/utils/rangemap.go | 4 +- 10 files changed, 151 insertions(+), 135 deletions(-) diff --git a/pkg/sfu/buffer/buffer.go b/pkg/sfu/buffer/buffer.go index 73704e883..939f5c6d8 100644 --- a/pkg/sfu/buffer/buffer.go +++ b/pkg/sfu/buffer/buffer.go @@ -54,7 +54,7 @@ type pendingPacket struct { type ExtPacket struct { VideoLayer Arrival time.Time - ExtSequenceNumber uint32 + ExtSequenceNumber uint64 ExtTimestamp uint64 Packet *rtp.Packet Payload interface{} @@ -83,7 +83,7 @@ type Buffer struct { closed atomic.Bool mime string - snRangeMap *utils.RangeMap[uint32, uint32] + snRangeMap *utils.RangeMap[uint64, uint64] latestTSForAudioLevelInitialized bool latestTSForAudioLevel uint32 @@ -128,7 +128,7 @@ func NewBuffer(ssrc uint32, vp, ap *sync.Pool) *Buffer { mediaSSRC: ssrc, videoPool: vp, audioPool: ap, - snRangeMap: utils.NewRangeMap[uint32, uint32](100), + snRangeMap: utils.NewRangeMap[uint64, uint64](100), pliThrottle: int64(500 * time.Millisecond), logger: l.WithComponent(sutils.ComponentPub).WithComponent(sutils.ComponentSFU), } diff --git a/pkg/sfu/buffer/buffer_test.go b/pkg/sfu/buffer/buffer_test.go index 7f3edf125..7f685186c 100644 --- a/pkg/sfu/buffer/buffer_test.go +++ b/pkg/sfu/buffer/buffer_test.go @@ -213,7 +213,7 @@ func TestNewBuffer(t *testing.T) { _, _ = buff.Write(buf) } require.Equal(t, uint16(2), buff.rtpStats.sequenceNumber.GetHighest()) - require.Equal(t, uint32(65536+2), buff.rtpStats.sequenceNumber.GetExtendedHighest()) + require.Equal(t, uint64(65536+2), buff.rtpStats.sequenceNumber.GetExtendedHighest()) }) } } diff --git a/pkg/sfu/buffer/helpers.go b/pkg/sfu/buffer/helpers.go index 274b73e47..5be302959 100644 --- a/pkg/sfu/buffer/helpers.go +++ b/pkg/sfu/buffer/helpers.go @@ -52,7 +52,7 @@ type VP8 struct { I bool M bool - PictureID uint16 /* 8 or 16 bits, picture ID */ + PictureID uint16 /* 7 or 15 bits, picture ID */ L bool TL0PICIDX uint8 /* 8 bits temporal level zero index */ diff --git a/pkg/sfu/buffer/rtpstats.go b/pkg/sfu/buffer/rtpstats.go index 4c17c93e4..3687a0ac9 100644 --- a/pkg/sfu/buffer/rtpstats.go +++ b/pkg/sfu/buffer/rtpstats.go @@ -66,24 +66,24 @@ func (d driftResult) String() string { type RTPFlowState struct { HasLoss bool - LossStartInclusive uint32 - LossEndExclusive uint32 + LossStartInclusive uint64 + LossEndExclusive uint64 IsOutOfOrder bool - ExtSequenceNumber uint32 + ExtSequenceNumber uint64 ExtTimestamp uint64 } type IntervalStats struct { - packets uint32 + packets uint64 bytes uint64 headerBytes uint64 - packetsPadding uint32 + packetsPadding uint64 bytesPadding uint64 headerBytesPadding uint64 - packetsLost uint32 - packetsOutOfOrder uint32 + packetsLost uint64 + packetsOutOfOrder uint64 frames uint32 } @@ -112,12 +112,12 @@ type RTPDeltaInfo struct { type Snapshot struct { startTime time.Time - extStartSN uint32 - extStartSNOverridden uint32 - packetsDuplicate uint32 + extStartSN uint64 + extStartSNOverridden uint64 + packetsDuplicate uint64 bytesDuplicate uint64 headerBytesDuplicate uint64 - packetsLostOverridden uint32 + packetsLostOverridden uint64 nacks uint32 plis uint32 firs uint32 @@ -158,9 +158,9 @@ type RTPStats struct { startTime time.Time endTime time.Time - sequenceNumber *utils.WrapAround[uint16, uint32] + sequenceNumber *utils.WrapAround[uint16, uint64] - extHighestSNOverridden uint32 + extHighestSNOverridden uint64 lastRRTime time.Time lastRR rtcp.ReceptionReport @@ -178,13 +178,13 @@ type RTPStats struct { headerBytesDuplicate uint64 bytesPadding uint64 headerBytesPadding uint64 - packetsDuplicate uint32 - packetsPadding uint32 + packetsDuplicate uint64 + packetsPadding uint64 - packetsOutOfOrder uint32 + packetsOutOfOrder uint64 - packetsLost uint32 - packetsLostOverridden uint32 + packetsLost uint64 + packetsLostOverridden uint64 frames uint32 @@ -229,7 +229,7 @@ func NewRTPStats(params RTPStatsParams) *RTPStats { return &RTPStats{ params: params, logger: params.Logger, - sequenceNumber: utils.NewWrapAround[uint16, uint32](), + sequenceNumber: utils.NewWrapAround[uint16, uint64](), timestamp: utils.NewWrapAround[uint32, uint64](), nextSnapshotId: FirstSnapshotId, snapshots: make(map[uint32]*Snapshot), @@ -374,7 +374,7 @@ func (r *RTPStats) Update(rtph *rtp.Header, payloadSize int, paddingSize int, pa return } - var resSN utils.WrapAroundUpdateResult[uint32] + var resSN utils.WrapAroundUpdateResult[uint64] var resTS utils.WrapAroundUpdateResult[uint64] if !r.initialized { if payloadSize == 0 { @@ -417,8 +417,8 @@ func (r *RTPStats) Update(rtph *rtp.Header, payloadSize int, paddingSize int, pa hdrSize := uint64(rtph.MarshalSize()) pktSize := hdrSize + uint64(payloadSize+paddingSize) isDuplicate := false - gapSN := resSN.ExtendedVal - resSN.PreExtendedHighest - if gapSN == 0 || gapSN > (1<<31) { // duplicate OR out-of-order + gapSN := int64(resSN.ExtendedVal - resSN.PreExtendedHighest) + if gapSN <= 0 { // duplicate OR out-of-order if payloadSize == 0 { // do not start on a padding only packet if resTS.IsRestart { @@ -480,7 +480,7 @@ func (r *RTPStats) Update(rtph *rtp.Header, payloadSize int, paddingSize int, pa // update missing sequence numbers r.clearSnInfos(resSN.PreExtendedHighest+1, resSN.ExtendedVal) - r.packetsLost += gapSN - 1 + r.packetsLost += uint64(gapSN - 1) r.setSnInfo(resSN.ExtendedVal, resSN.PreExtendedHighest, uint16(pktSize), uint16(hdrSize), uint16(payloadSize), rtph.Marker, false) @@ -518,7 +518,7 @@ func (r *RTPStats) Update(rtph *rtp.Header, payloadSize int, paddingSize int, pa return } -func (r *RTPStats) Resync(esn uint32, ets uint64, at time.Time) { +func (r *RTPStats) Resync(esn uint64, ets uint64, at time.Time) { r.lock.Lock() defer r.lock.Unlock() @@ -530,18 +530,18 @@ func (r *RTPStats) Resync(esn uint32, ets uint64, at time.Time) { r.highestTime = at } -func (r *RTPStats) getPacketsExpected() uint32 { +func (r *RTPStats) getPacketsExpected() uint64 { return r.sequenceNumber.GetExtendedHighest() - r.sequenceNumber.GetExtendedStart() + 1 } -func (r *RTPStats) GetTotalPacketsPrimary() uint32 { +func (r *RTPStats) GetTotalPacketsPrimary() uint64 { r.lock.RLock() defer r.lock.RUnlock() return r.getTotalPacketsPrimary() } -func (r *RTPStats) getTotalPacketsPrimary() uint32 { +func (r *RTPStats) getTotalPacketsPrimary() uint64 { packetsExpected := r.getPacketsExpected() if r.packetsLost > packetsExpected { // should not happen @@ -560,7 +560,19 @@ func (r *RTPStats) UpdateFromReceiverReport(rr rtcp.ReceptionReport) (rtt uint32 r.lock.Lock() defer r.lock.Unlock() - if !r.initialized || !r.endTime.IsZero() || !r.params.IsReceiverReportDriven || rr.LastSequenceNumber < r.sequenceNumber.GetExtendedHighest() { + if !r.initialized || !r.endTime.IsZero() || !r.params.IsReceiverReportDriven || uint64(rr.LastSequenceNumber) < r.sequenceNumber.GetExtendedHighest() { + // it is possible that the `LastSequenceNumber` in the receiver report is before the starting + // sequence number when dummy packets are used to trigger Pion's OnTrack path. + return + } + + extHighestSNOverridden := r.extHighestSNOverridden&0xFFFF_FFFF_0000_0000 + uint64(rr.LastSequenceNumber) + if !r.lastRRTime.IsZero() { + if (rr.LastSequenceNumber-r.lastRR.LastSequenceNumber) < (1<<31) && rr.LastSequenceNumber < r.lastRR.LastSequenceNumber { + extHighestSNOverridden += (1 << 32) + } + } + if extHighestSNOverridden < r.sequenceNumber.GetExtendedHighest() { // it is possible that the `LastSequenceNumber` in the receiver report is before the starting // sequence number when dummy packets are used to trigger Pion's OnTrack path. return @@ -578,9 +590,14 @@ func (r *RTPStats) UpdateFromReceiverReport(rr rtcp.ReceptionReport) (rtt uint32 } } - if r.lastRRTime.IsZero() || r.extHighestSNOverridden <= rr.LastSequenceNumber { - r.extHighestSNOverridden = rr.LastSequenceNumber - r.packetsLostOverridden = rr.TotalLost + if r.lastRRTime.IsZero() || r.extHighestSNOverridden <= extHighestSNOverridden { + r.extHighestSNOverridden = extHighestSNOverridden + + packetsLostOverridden := r.packetsLostOverridden&0xFFFF_FFFF_0000_0000 + uint64(rr.TotalLost) + if (rr.TotalLost-r.lastRR.TotalLost) < (1<<31) && rr.TotalLost < r.lastRR.TotalLost { + packetsLostOverridden += (1 << 32) + } + r.packetsLostOverridden = packetsLostOverridden if isRttChanged { r.rtt = rtt @@ -856,7 +873,7 @@ func (r *RTPStats) SetRtcpSenderReportData(srData *RTCPSenderReportData) { cycles := uint64(0) if r.srNewest != nil { - cycles = r.srNewest.RTPTimestampExt & 0xFF_FF_FF_FF_00_00_00_00 + cycles = r.srNewest.RTPTimestampExt & 0xFFFF_FFFF_0000_0000 if (srData.RTPTimestamp-r.srNewest.RTPTimestamp) < (1<<31) && srData.RTPTimestamp < r.srNewest.RTPTimestamp { cycles += (1 << 32) } @@ -1070,7 +1087,7 @@ func (r *RTPStats) GetRtcpSenderReport(ssrc uint32, calculatedClockRate uint32) SSRC: ssrc, NTPTime: uint64(nowNTP), RTPTime: nowRTP, - PacketCount: r.getTotalPacketsPrimary() + r.packetsDuplicate + r.packetsPadding, + PacketCount: uint32(r.getTotalPacketsPrimary() + r.packetsDuplicate + r.packetsPadding), OctetCount: uint32(r.bytes + r.bytesDuplicate + r.bytesPadding), } } @@ -1121,8 +1138,8 @@ func (r *RTPStats) SnapshotRtcpReceptionReport(ssrc uint32, proxyFracLost uint8, return &rtcp.ReceptionReport{ SSRC: ssrc, FractionLost: fracLost, - TotalLost: r.packetsLost, - LastSequenceNumber: now.extStartSN, + TotalLost: uint32(r.packetsLost), + LastSequenceNumber: uint32(now.extStartSN), Jitter: uint32(r.jitter), LastSenderReport: lastSR, Delay: dlsr, @@ -1163,16 +1180,16 @@ func (r *RTPStats) DeltaInfo(snapshotId uint32) *RTPDeltaInfo { return &RTPDeltaInfo{ StartTime: startTime, Duration: endTime.Sub(startTime), - Packets: packetsExpected - intervalStats.packetsPadding, + Packets: uint32(packetsExpected - intervalStats.packetsPadding), Bytes: intervalStats.bytes, HeaderBytes: intervalStats.headerBytes, - PacketsDuplicate: now.packetsDuplicate - then.packetsDuplicate, + PacketsDuplicate: uint32(now.packetsDuplicate - then.packetsDuplicate), BytesDuplicate: now.bytesDuplicate - then.bytesDuplicate, HeaderBytesDuplicate: now.headerBytesDuplicate - then.headerBytesDuplicate, - PacketsPadding: intervalStats.packetsPadding, + PacketsPadding: uint32(intervalStats.packetsPadding), BytesPadding: intervalStats.bytesPadding, HeaderBytesPadding: intervalStats.headerBytesPadding, - PacketsLost: intervalStats.packetsLost, + PacketsLost: uint32(intervalStats.packetsLost), Frames: intervalStats.frames, RttMax: then.maxRtt, JitterMax: then.maxJitter / float64(r.params.ClockRate) * 1e6, @@ -1245,18 +1262,18 @@ func (r *RTPStats) DeltaInfoOverridden(snapshotId uint32) *RTPDeltaInfo { return &RTPDeltaInfo{ StartTime: startTime, Duration: endTime.Sub(startTime), - Packets: packetsExpected - intervalStats.packetsPadding, + Packets: uint32(packetsExpected - intervalStats.packetsPadding), Bytes: intervalStats.bytes, HeaderBytes: intervalStats.headerBytes, - PacketsDuplicate: now.packetsDuplicate - then.packetsDuplicate, + PacketsDuplicate: uint32(now.packetsDuplicate - then.packetsDuplicate), BytesDuplicate: now.bytesDuplicate - then.bytesDuplicate, HeaderBytesDuplicate: now.headerBytesDuplicate - then.headerBytesDuplicate, - PacketsPadding: intervalStats.packetsPadding, + PacketsPadding: uint32(intervalStats.packetsPadding), BytesPadding: intervalStats.bytesPadding, HeaderBytesPadding: intervalStats.headerBytesPadding, - PacketsLost: packetsLost, - PacketsMissing: intervalStats.packetsLost, - PacketsOutOfOrder: intervalStats.packetsOutOfOrder, + PacketsLost: uint32(packetsLost), + PacketsMissing: uint32(intervalStats.packetsLost), + PacketsOutOfOrder: uint32(intervalStats.packetsOutOfOrder), Frames: intervalStats.frames, RttMax: then.maxRtt, JitterMax: maxJitterTime, @@ -1392,25 +1409,25 @@ func (r *RTPStats) ToProto() *livekit.RTPStats { StartTime: timestamppb.New(r.startTime), EndTime: timestamppb.New(endTime), Duration: elapsed, - Packets: packets, + Packets: uint32(packets), PacketRate: packetRate, Bytes: r.bytes, HeaderBytes: r.headerBytes, Bitrate: bitrate, - PacketsLost: packetsLost, + PacketsLost: uint32(packetsLost), PacketLossRate: packetLostRate, PacketLossPercentage: packetLostPercentage, - PacketsDuplicate: r.packetsDuplicate, + PacketsDuplicate: uint32(r.packetsDuplicate), PacketDuplicateRate: packetDuplicateRate, BytesDuplicate: r.bytesDuplicate, HeaderBytesDuplicate: r.headerBytesDuplicate, BitrateDuplicate: bitrateDuplicate, - PacketsPadding: r.packetsPadding, + PacketsPadding: uint32(r.packetsPadding), PacketPaddingRate: packetPaddingRate, BytesPadding: r.bytesPadding, HeaderBytesPadding: r.headerBytesPadding, BitratePadding: bitratePadding, - PacketsOutOfOrder: r.packetsOutOfOrder, + PacketsOutOfOrder: uint32(r.packetsOutOfOrder), Frames: r.frames, FrameRate: frameRate, KeyFrames: r.keyFrames, @@ -1457,7 +1474,7 @@ func (r *RTPStats) ToProto() *livekit.RTPStats { return p } -func (r *RTPStats) getExtHighestSNAdjusted() uint32 { +func (r *RTPStats) getExtHighestSNAdjusted() uint64 { if r.params.IsReceiverReportDriven && !r.lastRRTime.IsZero() { return r.extHighestSNOverridden } @@ -1465,7 +1482,7 @@ func (r *RTPStats) getExtHighestSNAdjusted() uint32 { return r.sequenceNumber.GetExtendedHighest() } -func (r *RTPStats) getPacketsLost() uint32 { +func (r *RTPStats) getPacketsLost() uint64 { if r.params.IsReceiverReportDriven && !r.lastRRTime.IsZero() { return r.packetsLostOverridden } @@ -1473,13 +1490,12 @@ func (r *RTPStats) getPacketsLost() uint32 { return r.packetsLost } -func (r *RTPStats) getSnInfoOutOfOrderPtr(esn uint32, ehsn uint32) int { - offset := esn - ehsn - if offset > 0 && offset < (1<<31) { +func (r *RTPStats) getSnInfoOutOfOrderPtr(esn uint64, ehsn uint64) int { + if int64(esn-ehsn) > 0 { return -1 // in-order, not expected, maybe too new } - offset = ehsn - esn + offset := ehsn - esn if int(offset) >= SnInfoSize { // too old, ignore return -1 @@ -1488,9 +1504,9 @@ func (r *RTPStats) getSnInfoOutOfOrderPtr(esn uint32, ehsn uint32) int { return (r.snInfoWritePtr - int(offset) - 1) & SnInfoMask } -func (r *RTPStats) setSnInfo(esn uint32, ehsn uint32, pktSize uint16, hdrSize uint16, payloadSize uint16, marker bool, isOutOfOrder bool) { +func (r *RTPStats) setSnInfo(esn uint64, ehsn uint64, pktSize uint16, hdrSize uint16, payloadSize uint16, marker bool, isOutOfOrder bool) { writePtr := 0 - ooo := (esn - ehsn) > (1 << 31) + ooo := int64(esn-ehsn) < 0 if !ooo { writePtr = r.snInfoWritePtr r.snInfoWritePtr = (writePtr + 1) & SnInfoMask @@ -1509,7 +1525,7 @@ func (r *RTPStats) setSnInfo(esn uint32, ehsn uint32, pktSize uint16, hdrSize ui snInfo.isOutOfOrder = isOutOfOrder } -func (r *RTPStats) clearSnInfos(extStartInclusive uint32, extEndExclusive uint32) { +func (r *RTPStats) clearSnInfos(extStartInclusive uint64, extEndExclusive uint64) { for esn := extStartInclusive; esn != extEndExclusive; esn++ { snInfo := &r.snInfos[r.snInfoWritePtr] snInfo.pktSize = 0 @@ -1521,7 +1537,7 @@ func (r *RTPStats) clearSnInfos(extStartInclusive uint32, extEndExclusive uint32 } } -func (r *RTPStats) isSnInfoLost(esn uint32, ehsn uint32) bool { +func (r *RTPStats) isSnInfoLost(esn uint64, ehsn uint64) bool { readPtr := r.getSnInfoOutOfOrderPtr(esn, ehsn) if readPtr < 0 { return false @@ -1531,9 +1547,9 @@ func (r *RTPStats) isSnInfoLost(esn uint32, ehsn uint32) bool { return snInfo.pktSize == 0 } -func (r *RTPStats) getIntervalStats(extStartInclusive uint32, extEndExclusive uint32) (intervalStats IntervalStats) { +func (r *RTPStats) getIntervalStats(extStartInclusive uint64, extEndExclusive uint64) (intervalStats IntervalStats) { packetsNotFound := uint32(0) - processESN := func(esn uint32, ehsn uint32) { + processESN := func(esn uint64, ehsn uint64) { readPtr := r.getSnInfoOutOfOrderPtr(esn, ehsn) if readPtr < 0 { packetsNotFound++ diff --git a/pkg/sfu/buffer/rtpstats_test.go b/pkg/sfu/buffer/rtpstats_test.go index 1a803578d..e59a21de2 100644 --- a/pkg/sfu/buffer/rtpstats_test.go +++ b/pkg/sfu/buffer/rtpstats_test.go @@ -114,8 +114,8 @@ func TestRTPStats_Update(t *testing.T) { require.Equal(t, sequenceNumber, uint16(r.sequenceNumber.GetExtendedHighest())) require.Equal(t, timestamp, r.timestamp.GetHighest()) require.Equal(t, timestamp, uint32(r.timestamp.GetExtendedHighest())) - require.Equal(t, uint32(1), r.packetsOutOfOrder) - require.Equal(t, uint32(0), r.packetsDuplicate) + require.Equal(t, uint64(1), r.packetsOutOfOrder) + require.Equal(t, uint64(0), r.packetsDuplicate) // duplicate packet = getPacket(sequenceNumber-10, timestamp-30000, 1000) @@ -125,8 +125,8 @@ func TestRTPStats_Update(t *testing.T) { require.Equal(t, sequenceNumber, uint16(r.sequenceNumber.GetExtendedHighest())) require.Equal(t, timestamp, r.timestamp.GetHighest()) require.Equal(t, timestamp, uint32(r.timestamp.GetExtendedHighest())) - require.Equal(t, uint32(2), r.packetsOutOfOrder) - require.Equal(t, uint32(1), r.packetsDuplicate) + require.Equal(t, uint64(2), r.packetsOutOfOrder) + require.Equal(t, uint64(1), r.packetsDuplicate) // loss sequenceNumber += 10 @@ -134,9 +134,9 @@ func TestRTPStats_Update(t *testing.T) { packet = getPacket(sequenceNumber, timestamp, 1000) flowState = r.Update(&packet.Header, len(packet.Payload), 0, time.Now()) require.True(t, flowState.HasLoss) - require.Equal(t, uint32(sequenceNumber-9), flowState.LossStartInclusive) - require.Equal(t, uint32(sequenceNumber), flowState.LossEndExclusive) - require.Equal(t, uint32(17), r.packetsLost) + require.Equal(t, uint64(sequenceNumber-9), flowState.LossStartInclusive) + require.Equal(t, uint64(sequenceNumber), flowState.LossEndExclusive) + require.Equal(t, uint64(17), r.packetsLost) // out-of-order should decrement number of lost packets packet = getPacket(sequenceNumber-15, timestamp-45000, 1000) @@ -146,11 +146,11 @@ func TestRTPStats_Update(t *testing.T) { require.Equal(t, sequenceNumber, uint16(r.sequenceNumber.GetExtendedHighest())) require.Equal(t, timestamp, r.timestamp.GetHighest()) require.Equal(t, timestamp, uint32(r.timestamp.GetExtendedHighest())) - require.Equal(t, uint32(3), r.packetsOutOfOrder) - require.Equal(t, uint32(1), r.packetsDuplicate) - require.Equal(t, uint32(16), r.packetsLost) + require.Equal(t, uint64(3), r.packetsOutOfOrder) + require.Equal(t, uint64(1), r.packetsDuplicate) + require.Equal(t, uint64(16), r.packetsLost) intervalStats := r.getIntervalStats(r.sequenceNumber.GetExtendedStart(), r.sequenceNumber.GetExtendedHighest()+1) - require.Equal(t, uint32(16), intervalStats.packetsLost) + require.Equal(t, uint64(16), intervalStats.packetsLost) r.Stop() } diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index 6cc495e90..707640bd2 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -1698,7 +1698,7 @@ func (d *DownTrack) getDeltaStatsOverridden() map[uint32]*buffer.StreamStatsWith } func (d *DownTrack) GetNackStats() (totalPackets uint32, totalRepeatedNACKs uint32) { - totalPackets = d.rtpStats.GetTotalPacketsPrimary() + totalPackets = uint32(d.rtpStats.GetTotalPacketsPrimary()) totalRepeatedNACKs = d.totalRepeatedNACKs.Load() return } diff --git a/pkg/sfu/rtpmunger.go b/pkg/sfu/rtpmunger.go index 3f26ddfe7..e2c535329 100644 --- a/pkg/sfu/rtpmunger.go +++ b/pkg/sfu/rtpmunger.go @@ -53,7 +53,7 @@ type SnTs struct { // ---------------------------------------------------------------------- type RTPMungerState struct { - ExtLastSN uint32 + ExtLastSN uint64 ExtLastTS uint64 } @@ -66,22 +66,22 @@ func (r RTPMungerState) String() string { type RTPMunger struct { logger logger.Logger - extHighestIncomingSN uint32 - snRangeMap *utils.RangeMap[uint32, uint32] + extHighestIncomingSN uint64 + snRangeMap *utils.RangeMap[uint64, uint64] - extLastSN uint32 + extLastSN uint64 extLastTS uint64 tsOffset uint64 lastMarker bool - extRtxGateSn uint32 + extRtxGateSn uint64 isInRtxGateRegion bool } func NewRTPMunger(logger logger.Logger) *RTPMunger { return &RTPMunger{ logger: logger, - snRangeMap: utils.NewRangeMap[uint32, uint32](100), + snRangeMap: utils.NewRangeMap[uint64, uint64](100), } } @@ -115,7 +115,7 @@ func (r *RTPMunger) SetLastSnTs(extPkt *buffer.ExtPacket) { r.extLastTS = extPkt.ExtTimestamp } -func (r *RTPMunger) UpdateSnTsOffsets(extPkt *buffer.ExtPacket, snAdjust uint32, tsAdjust uint64) { +func (r *RTPMunger) UpdateSnTsOffsets(extPkt *buffer.ExtPacket, snAdjust uint64, tsAdjust uint64) { r.extHighestIncomingSN = extPkt.ExtSequenceNumber - 1 r.snRangeMap.ClearAndResetValue(extPkt.ExtSequenceNumber - r.extLastSN - snAdjust) r.tsOffset = extPkt.ExtTimestamp - r.extLastTS - tsAdjust @@ -260,7 +260,7 @@ func (r *RTPMunger) UpdateAndGetPaddingSnTs(num int, clockRate uint32, frameRate } r.extLastSN = extLastSN - r.snRangeMap.DecValue(uint32(num)) + r.snRangeMap.DecValue(uint64(num)) r.tsOffset -= extLastTS - r.extLastTS r.extLastTS = extLastTS diff --git a/pkg/sfu/rtpmunger_test.go b/pkg/sfu/rtpmunger_test.go index 4119fd2c5..2ade34e7b 100644 --- a/pkg/sfu/rtpmunger_test.go +++ b/pkg/sfu/rtpmunger_test.go @@ -41,12 +41,12 @@ func TestSetLastSnTs(t *testing.T) { require.NotNil(t, extPkt) r.SetLastSnTs(extPkt) - require.Equal(t, uint32(23332), r.extHighestIncomingSN) - require.Equal(t, uint32(23333), r.extLastSN) + require.Equal(t, uint64(23332), r.extHighestIncomingSN) + require.Equal(t, uint64(23333), r.extLastSN) require.Equal(t, uint64(0xabcdef), r.extLastTS) snOffset, err := r.snRangeMap.GetValue(r.extHighestIncomingSN) require.NoError(t, err) - require.Equal(t, uint32(0), snOffset) + require.Equal(t, uint64(0), snOffset) require.Equal(t, uint64(0), r.tsOffset) } @@ -68,12 +68,12 @@ func TestUpdateSnTsOffsets(t *testing.T) { } extPkt, _ = testutils.GetTestExtPacket(params) r.UpdateSnTsOffsets(extPkt, 1, 1) - require.Equal(t, uint32(33332), r.extHighestIncomingSN) - require.Equal(t, uint32(23333), r.extLastSN) + require.Equal(t, uint64(33332), r.extHighestIncomingSN) + require.Equal(t, uint64(23333), r.extLastSN) require.Equal(t, uint64(0xabcdef), r.extLastTS) snOffset, err := r.snRangeMap.GetValue(r.extHighestIncomingSN) require.NoError(t, err) - require.Equal(t, uint32(9999), snOffset) + require.Equal(t, uint64(9999), snOffset) require.Equal(t, uint64(0xffff_ffff_ffff_ffff), r.tsOffset) } @@ -88,12 +88,12 @@ func TestPacketDropped(t *testing.T) { } extPkt, _ := testutils.GetTestExtPacket(params) r.SetLastSnTs(extPkt) - require.Equal(t, uint32(23332), r.extHighestIncomingSN) - require.Equal(t, uint32(23333), r.extLastSN) + require.Equal(t, uint64(23332), r.extHighestIncomingSN) + require.Equal(t, uint64(23333), r.extLastSN) require.Equal(t, uint64(0xabcdef), r.extLastTS) snOffset, err := r.snRangeMap.GetValue(r.extHighestIncomingSN) require.NoError(t, err) - require.Equal(t, uint32(0), snOffset) + require.Equal(t, uint64(0), snOffset) require.Equal(t, uint64(0), r.tsOffset) r.UpdateAndGetSnTs(extPkt) // update sequence number offset @@ -106,11 +106,11 @@ func TestPacketDropped(t *testing.T) { } extPkt, _ = testutils.GetTestExtPacket(params) r.PacketDropped(extPkt) - require.Equal(t, uint32(23333), r.extHighestIncomingSN) - require.Equal(t, uint32(23333), r.extLastSN) + require.Equal(t, uint64(23333), r.extHighestIncomingSN) + require.Equal(t, uint64(23333), r.extLastSN) snOffset, err = r.snRangeMap.GetValue(r.extHighestIncomingSN) require.NoError(t, err) - require.Equal(t, uint32(0), snOffset) + require.Equal(t, uint64(0), snOffset) // drop a head packet and check offset increases params = &testutils.TestExtPacketParams{ @@ -124,10 +124,10 @@ func TestPacketDropped(t *testing.T) { r.UpdateAndGetSnTs(extPkt) // update sequence number offset r.PacketDropped(extPkt) - require.Equal(t, uint32(44443), r.extLastSN) + require.Equal(t, uint64(44443), r.extLastSN) snOffset, err = r.snRangeMap.GetValue(r.extHighestIncomingSN) require.NoError(t, err) - require.Equal(t, uint32(1), snOffset) + require.Equal(t, uint64(1), snOffset) params = &testutils.TestExtPacketParams{ SequenceNumber: 44445, @@ -138,10 +138,10 @@ func TestPacketDropped(t *testing.T) { extPkt, _ = testutils.GetTestExtPacket(params) r.UpdateAndGetSnTs(extPkt) // update sequence number offset - require.Equal(t, r.extLastSN, uint32(44444)) + require.Equal(t, r.extLastSN, uint64(44444)) snOffset, err = r.snRangeMap.GetValue(r.extHighestIncomingSN) require.NoError(t, err) - require.Equal(t, uint32(1), snOffset) + require.Equal(t, uint64(1), snOffset) } func TestOutOfOrderSequenceNumber(t *testing.T) { @@ -243,11 +243,11 @@ func TestPaddingOnlyPacket(t *testing.T) { require.Error(t, err) require.ErrorIs(t, err, ErrPaddingOnlyPacket) require.Equal(t, tpExpected, *tp) - require.Equal(t, uint32(23333), r.extHighestIncomingSN) - require.Equal(t, uint32(23333), r.extLastSN) + require.Equal(t, uint64(23333), r.extHighestIncomingSN) + require.Equal(t, uint64(23333), r.extLastSN) snOffset, err := r.snRangeMap.GetValue(r.extHighestIncomingSN) require.NoError(t, err) - require.Equal(t, uint32(1), snOffset) + require.Equal(t, uint64(1), snOffset) // padding only packet with a gap should not report an error params = &testutils.TestExtPacketParams{ @@ -266,11 +266,11 @@ func TestPaddingOnlyPacket(t *testing.T) { tp, err = r.UpdateAndGetSnTs(extPkt) require.NoError(t, err) require.Equal(t, tpExpected, *tp) - require.Equal(t, uint32(23335), r.extHighestIncomingSN) - require.Equal(t, uint32(23334), r.extLastSN) + require.Equal(t, uint64(23335), r.extHighestIncomingSN) + require.Equal(t, uint64(23334), r.extLastSN) snOffset, err = r.snRangeMap.GetValue(r.extHighestIncomingSN) require.NoError(t, err) - require.Equal(t, uint32(1), snOffset) + require.Equal(t, uint64(1), snOffset) } func TestGapInSequenceNumber(t *testing.T) { @@ -307,19 +307,19 @@ func TestGapInSequenceNumber(t *testing.T) { tp, err := r.UpdateAndGetSnTs(extPkt) require.NoError(t, err) require.Equal(t, tpExpected, *tp) - require.Equal(t, uint32(65536+1), r.extHighestIncomingSN) - require.Equal(t, uint32(65536+1), r.extLastSN) + require.Equal(t, uint64(65536+1), r.extHighestIncomingSN) + require.Equal(t, uint64(65536+1), r.extLastSN) snOffset, err := r.snRangeMap.GetValue(r.extHighestIncomingSN) require.NoError(t, err) - require.Equal(t, uint32(0), snOffset) + require.Equal(t, uint64(0), snOffset) // ensure missing sequence numbers got recorded in cache // last received, three missing in between and current received should all be in cache - for i := uint32(65534); i != 65536+1; i++ { + for i := uint64(65534); i != 65536+1; i++ { offset, err := r.snRangeMap.GetValue(i) require.NoError(t, err) - require.Equal(t, uint32(0), offset) + require.Equal(t, uint64(0), offset) } // a padding only packet should be dropped @@ -338,11 +338,11 @@ func TestGapInSequenceNumber(t *testing.T) { tp, err = r.UpdateAndGetSnTs(extPkt) require.ErrorIs(t, err, ErrPaddingOnlyPacket) require.Equal(t, tpExpected, *tp) - require.Equal(t, uint32(65536+2), r.extHighestIncomingSN) - require.Equal(t, uint32(65536+1), r.extLastSN) + require.Equal(t, uint64(65536+2), r.extHighestIncomingSN) + require.Equal(t, uint64(65536+1), r.extLastSN) snOffset, err = r.snRangeMap.GetValue(r.extHighestIncomingSN) require.NoError(t, err) - require.Equal(t, uint32(1), snOffset) + require.Equal(t, uint64(1), snOffset) // a packet with a gap should be adding to missing cache params = &testutils.TestExtPacketParams{ @@ -363,11 +363,11 @@ func TestGapInSequenceNumber(t *testing.T) { tp, err = r.UpdateAndGetSnTs(extPkt) require.NoError(t, err) require.Equal(t, tpExpected, *tp) - require.Equal(t, uint32(65536+4), r.extHighestIncomingSN) - require.Equal(t, uint32(65536+3), r.extLastSN) + require.Equal(t, uint64(65536+4), r.extHighestIncomingSN) + require.Equal(t, uint64(65536+3), r.extLastSN) snOffset, err = r.snRangeMap.GetValue(r.extHighestIncomingSN) require.NoError(t, err) - require.Equal(t, uint32(1), snOffset) + require.Equal(t, uint64(1), snOffset) // another contiguous padding only packet should be dropped params = &testutils.TestExtPacketParams{ @@ -385,11 +385,11 @@ func TestGapInSequenceNumber(t *testing.T) { tp, err = r.UpdateAndGetSnTs(extPkt) require.ErrorIs(t, err, ErrPaddingOnlyPacket) require.Equal(t, tpExpected, *tp) - require.Equal(t, uint32(65536+5), r.extHighestIncomingSN) - require.Equal(t, uint32(65536+3), r.extLastSN) + require.Equal(t, uint64(65536+5), r.extHighestIncomingSN) + require.Equal(t, uint64(65536+3), r.extLastSN) snOffset, err = r.snRangeMap.GetValue(r.extHighestIncomingSN) require.NoError(t, err) - require.Equal(t, uint32(2), snOffset) + require.Equal(t, uint64(2), snOffset) // a packet with a gap should be adding to missing cache params = &testutils.TestExtPacketParams{ @@ -410,11 +410,11 @@ func TestGapInSequenceNumber(t *testing.T) { tp, err = r.UpdateAndGetSnTs(extPkt) require.NoError(t, err) require.Equal(t, tpExpected, *tp) - require.Equal(t, uint32(65536+7), r.extHighestIncomingSN) - require.Equal(t, uint32(65536+5), r.extLastSN) + require.Equal(t, uint64(65536+7), r.extHighestIncomingSN) + require.Equal(t, uint64(65536+5), r.extLastSN) snOffset, err = r.snRangeMap.GetValue(r.extHighestIncomingSN) require.NoError(t, err) - require.Equal(t, uint32(2), snOffset) + require.Equal(t, uint64(2), snOffset) // check the missing packets params = &testutils.TestExtPacketParams{ @@ -434,11 +434,11 @@ func TestGapInSequenceNumber(t *testing.T) { tp, err = r.UpdateAndGetSnTs(extPkt) require.NoError(t, err) require.Equal(t, tpExpected, *tp) - require.Equal(t, uint32(65536+7), r.extHighestIncomingSN) - require.Equal(t, uint32(65536+5), r.extLastSN) + require.Equal(t, uint64(65536+7), r.extHighestIncomingSN) + require.Equal(t, uint64(65536+5), r.extLastSN) snOffset, err = r.snRangeMap.GetValue(r.extHighestIncomingSN) require.NoError(t, err) - require.Equal(t, uint32(2), snOffset) + require.Equal(t, uint64(2), snOffset) params = &testutils.TestExtPacketParams{ SequenceNumber: 3, @@ -457,11 +457,11 @@ func TestGapInSequenceNumber(t *testing.T) { tp, err = r.UpdateAndGetSnTs(extPkt) require.NoError(t, err) require.Equal(t, tpExpected, *tp) - require.Equal(t, uint32(65536+7), r.extHighestIncomingSN) - require.Equal(t, uint32(65536+5), r.extLastSN) + require.Equal(t, uint64(65536+7), r.extHighestIncomingSN) + require.Equal(t, uint64(65536+5), r.extLastSN) snOffset, err = r.snRangeMap.GetValue(r.extHighestIncomingSN) require.NoError(t, err) - require.Equal(t, uint32(2), snOffset) + require.Equal(t, uint64(2), snOffset) } func TestUpdateAndGetPaddingSnTs(t *testing.T) { diff --git a/pkg/sfu/testutils/data.go b/pkg/sfu/testutils/data.go index 2f0096c63..1bebc420b 100644 --- a/pkg/sfu/testutils/data.go +++ b/pkg/sfu/testutils/data.go @@ -64,7 +64,7 @@ func GetTestExtPacket(params *TestExtPacketParams) (*buffer.ExtPacket, error) { ep := &buffer.ExtPacket{ VideoLayer: params.VideoLayer, - ExtSequenceNumber: uint32(params.SNCycles<<16) + uint32(params.SequenceNumber), + ExtSequenceNumber: uint64(params.SNCycles<<16) + uint64(params.SequenceNumber), ExtTimestamp: uint64(params.TSCycles<<32) + uint64(params.Timestamp), Arrival: params.ArrivalTime, Packet: &packet, diff --git a/pkg/sfu/utils/rangemap.go b/pkg/sfu/utils/rangemap.go index 3654791f0..acaa4a320 100644 --- a/pkg/sfu/utils/rangemap.go +++ b/pkg/sfu/utils/rangemap.go @@ -30,11 +30,11 @@ var ( ) type rangeType interface { - uint32 + uint32 | uint64 } type valueType interface { - uint32 + uint32 | uint64 } type rangeVal[RT rangeType, VT valueType] struct {