mirror of
https://github.com/livekit/livekit.git
synced 2026-08-28 05:04:10 +00:00
Make extended sequence number 64-bit. (#2003)
This commit is contained in:
Raja Subramanian
@@ -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),
|
||||
}
|
||||
|
||||
@@ -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())
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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 */
|
||||
|
||||
+78
-62
@@ -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++
|
||||
|
||||
@@ -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()
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
+45
-45
@@ -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) {
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -30,11 +30,11 @@ var (
|
||||
)
|
||||
|
||||
type rangeType interface {
|
||||
uint32
|
||||
uint32 | uint64
|
||||
}
|
||||
|
||||
type valueType interface {
|
||||
uint32
|
||||
uint32 | uint64
|
||||
}
|
||||
|
||||
type rangeVal[RT rangeType, VT valueType] struct {
|
||||
|
||||
Reference in New Issue
Block a user