diff --git a/pkg/sfu/buffer/rtpstats_base.go b/pkg/sfu/buffer/rtpstats_base.go index f02e9241b..3fad1907c 100644 --- a/pkg/sfu/buffer/rtpstats_base.go +++ b/pkg/sfu/buffer/rtpstats_base.go @@ -89,17 +89,34 @@ type RTPDeltaInfo struct { } type snapshot struct { - startTime time.Time - extStartSN uint64 - packetsDuplicate uint64 - bytesDuplicate uint64 - headerBytesDuplicate uint64 - packetsLostOverridden uint64 - nacks uint32 - plis uint32 - firs uint32 - maxRtt uint32 - maxJitter float64 + isValid bool + + startTime time.Time + + extStartSN uint64 + bytes uint64 + headerBytes uint64 + + packetsPadding uint64 + bytesPadding uint64 + headerBytesPadding uint64 + + packetsDuplicate uint64 + bytesDuplicate uint64 + headerBytesDuplicate uint64 + + packetsOutOfOrder uint64 + + packetsLost uint64 + + frames uint32 + + nacks uint32 + plis uint32 + firs uint32 + + maxRtt uint32 + maxJitter float64 } type snInfo struct { @@ -153,8 +170,7 @@ type rtpStatsBase struct { packetsOutOfOrder uint64 - packetsLost uint64 - packetsLostOverridden uint64 + packetsLost uint64 frames uint32 @@ -189,7 +205,7 @@ type rtpStatsBase struct { srNewest *RTCPSenderReportData nextSnapshotID uint32 - snapshots map[uint32]*snapshot + snapshots []snapshot } func newRTPStatsBase(params RTPStatsParams) *rtpStatsBase { @@ -197,7 +213,7 @@ func newRTPStatsBase(params RTPStatsParams) *rtpStatsBase { params: params, logger: params.Logger, nextSnapshotID: cFirstSnapshotID, - snapshots: make(map[uint32]*snapshot), + snapshots: make([]snapshot, 2), } } @@ -273,10 +289,8 @@ func (r *rtpStatsBase) seed(from *rtpStatsBase) bool { } r.nextSnapshotID = from.nextSnapshotID - for id, ss := range from.snapshots { - ssCopy := *ss - r.snapshots[id] = &ssCopy - } + r.snapshots = make([]snapshot, cap(from.snapshots)) + copy(r.snapshots, from.snapshots) return true } @@ -295,11 +309,14 @@ func (r *rtpStatsBase) newSnapshotID(extStartSN uint64) uint32 { id := r.nextSnapshotID r.nextSnapshotID++ + if cap(r.snapshots) < int(r.nextSnapshotID) { + snapshots := make([]snapshot, r.nextSnapshotID) + copy(snapshots, r.snapshots) + r.snapshots = snapshots + } + if r.initialized { - r.snapshots[id] = &snapshot{ - startTime: time.Now(), - extStartSN: extStartSN, - } + r.snapshots[id] = r.initSnapshot(time.Now(), extStartSN) } return id } @@ -551,21 +568,25 @@ func (r *rtpStatsBase) deltaInfo(snapshotID uint32, extStartSN uint64, extHighes } } - intervalStats := r.getIntervalStats(then.extStartSN, now.extStartSN, extHighestSN) + packetsLost := uint32(now.packetsLost - then.packetsLost) + if int32(packetsLost) < 0 { + packetsLost = 0 + } return &RTPDeltaInfo{ StartTime: startTime, Duration: endTime.Sub(startTime), - Packets: uint32(packetsExpected - intervalStats.packetsPadding), - Bytes: intervalStats.bytes, - HeaderBytes: intervalStats.headerBytes, + Packets: uint32(packetsExpected - (now.packetsPadding - then.packetsPadding)), + Bytes: now.bytes - then.bytes, + HeaderBytes: now.headerBytes - then.headerBytes, PacketsDuplicate: uint32(now.packetsDuplicate - then.packetsDuplicate), BytesDuplicate: now.bytesDuplicate - then.bytesDuplicate, HeaderBytesDuplicate: now.headerBytesDuplicate - then.headerBytesDuplicate, - PacketsPadding: uint32(intervalStats.packetsPadding), - BytesPadding: intervalStats.bytesPadding, - HeaderBytesPadding: intervalStats.headerBytesPadding, - PacketsLost: uint32(intervalStats.packetsLost), - Frames: intervalStats.frames, + PacketsPadding: uint32(now.packetsPadding - then.packetsPadding), + BytesPadding: now.bytesPadding - then.bytesPadding, + HeaderBytesPadding: now.headerBytesPadding - then.headerBytesPadding, + PacketsLost: packetsLost, + PacketsOutOfOrder: uint32(now.packetsOutOfOrder - then.packetsOutOfOrder), + Frames: now.frames - then.frames, RttMax: then.maxRtt, JitterMax: then.maxJitter / float64(r.params.ClockRate) * 1e6, Nacks: now.nacks - then.nacks, @@ -894,31 +915,15 @@ func (r *rtpStatsBase) getAndResetSnapshot(snapshotID uint32, extStartSN uint64, } then := r.snapshots[snapshotID] - if then == nil { - then = &snapshot{ - startTime: r.startTime, - extStartSN: extStartSN, - } + if !then.isValid { + then = r.initSnapshot(r.startTime, extStartSN) r.snapshots[snapshotID] = then } // snapshot now - r.snapshots[snapshotID] = &snapshot{ - startTime: time.Now(), - extStartSN: extHighestSN + 1, - packetsDuplicate: r.packetsDuplicate, - bytesDuplicate: r.bytesDuplicate, - headerBytesDuplicate: r.headerBytesDuplicate, - nacks: r.nacks, - plis: r.plis, - firs: r.firs, - maxJitter: r.jitter, - maxRtt: r.rtt, - } - // make a copy so that it can be used independently - now := *r.snapshots[snapshotID] - - return then, &now + now := r.getSnapshot(time.Now(), extHighestSN+1) + r.snapshots[snapshotID] = now + return &then, &now } func (r *rtpStatsBase) getDrift(extStartTS, extHighestTS uint64) (packetDrift *livekit.RTPDrift, reportDrift *livekit.RTPDrift) { @@ -975,6 +980,38 @@ func (r *rtpStatsBase) updateGapHistogram(gap int) { } } +func (r *rtpStatsBase) initSnapshot(startTime time.Time, extStartSN uint64) snapshot { + return snapshot{ + isValid: true, + startTime: time.Now(), + extStartSN: extStartSN, + } +} + +func (r *rtpStatsBase) getSnapshot(startTime time.Time, extStartSN uint64) snapshot { + return snapshot{ + isValid: true, + startTime: time.Now(), + extStartSN: extStartSN, + bytes: r.bytes, + headerBytes: r.headerBytes, + packetsPadding: r.packetsPadding, + bytesPadding: r.bytesPadding, + headerBytesPadding: r.headerBytesPadding, + packetsDuplicate: r.packetsDuplicate, + bytesDuplicate: r.bytesDuplicate, + headerBytesDuplicate: r.headerBytesDuplicate, + packetsLost: r.packetsLost, + packetsOutOfOrder: r.packetsOutOfOrder, + frames: r.frames, + nacks: r.nacks, + plis: r.plis, + firs: r.firs, + maxRtt: r.rtt, + maxJitter: r.jitter, + } +} + // ---------------------------------- func AggregateRTPStats(statsList []*livekit.RTPStats) *livekit.RTPStats { diff --git a/pkg/sfu/buffer/rtpstats_receiver.go b/pkg/sfu/buffer/rtpstats_receiver.go index eeb8e5ef5..0d41e73d7 100644 --- a/pkg/sfu/buffer/rtpstats_receiver.go +++ b/pkg/sfu/buffer/rtpstats_receiver.go @@ -24,6 +24,10 @@ import ( "github.com/livekit/protocol/livekit" ) +const ( + cHistorySize = 2048 +) + type RTPFlowState struct { IsNotHandled bool @@ -47,6 +51,8 @@ type RTPStatsReceiver struct { sequenceNumber *utils.WrapAround[uint16, uint64] timestamp *utils.WrapAround[uint32, uint64] + + history [cHistorySize / 64]uint64 } func NewRTPStatsReceiver(params RTPStatsParams) *RTPStatsReceiver { @@ -107,10 +113,7 @@ func (r *RTPStatsReceiver) Update( // initialize snapshots if any for i := uint32(cFirstSnapshotID); i < r.nextSnapshotID; i++ { - r.snapshots[i] = &snapshot{ - startTime: r.startTime, - extStartSN: r.sequenceNumber.GetExtendedStart(), - } + r.snapshots[i] = r.initSnapshot(r.startTime, r.sequenceNumber.GetExtendedStart()) } r.logger.Debugw( @@ -170,14 +173,14 @@ func (r *RTPStatsReceiver) Update( ) } - if !r.isSnInfoLost(resSN.ExtendedVal, resSN.PreExtendedHighest) { + if !r.isLost(resSN.ExtendedVal, resSN.PreExtendedHighest) { r.bytesDuplicate += pktSize r.headerBytesDuplicate += uint64(hdrSize) r.packetsDuplicate++ flowState.IsDuplicate = true } else { r.packetsLost-- - r.setSnInfo(resSN.ExtendedVal, resSN.PreExtendedHighest, uint16(pktSize), uint16(hdrSize), uint16(payloadSize), marker, true) + r.setHistory(resSN.ExtendedVal, resSN.PreExtendedHighest) } flowState.IsOutOfOrder = true @@ -188,10 +191,10 @@ func (r *RTPStatsReceiver) Update( r.updateGapHistogram(int(gapSN)) // update missing sequence numbers - r.clearSnInfos(resSN.PreExtendedHighest+1, resSN.ExtendedVal) + r.clearHistory(resSN.PreExtendedHighest+1, resSN.ExtendedVal, resSN.PreExtendedHighest) r.packetsLost += uint64(gapSN - 1) - r.setSnInfo(resSN.ExtendedVal, resSN.PreExtendedHighest, uint16(pktSize), uint16(hdrSize), uint16(payloadSize), marker, false) + r.setHistory(resSN.ExtendedVal, resSN.PreExtendedHighest) if timestamp != uint32(resTS.PreExtendedHighest) { // update only on first packet as same timestamp could be in multiple packets. @@ -409,8 +412,10 @@ func (r *RTPStatsReceiver) GetRtcpReceptionReport(ssrc uint32, proxyFracLost uin return nil } - intervalStats := r.getIntervalStats(then.extStartSN, now.extStartSN, extHighestSN) - packetsLost := intervalStats.packetsLost + packetsLost := uint32(now.packetsLost - then.packetsLost) + if int32(packetsLost) < 0 { + packetsLost = 0 + } lossRate := float32(packetsLost) / float32(packetsExpected) fracLost := uint8(lossRate * 256.0) if proxyFracLost > fracLost { @@ -468,4 +473,62 @@ func (r *RTPStatsReceiver) ToProto() *livekit.RTPStats { ) } +func (r *RTPStatsReceiver) getOutOfOrderHistorySlot(esn uint64, ehsn uint64) (int, int) { + diff := int64(ehsn - esn) + if diff >= cHistorySize || diff < 0 { + // too old OR too new (i. e. ahead of highest) + return -1, -1 + } + + return int(esn) % len(r.history), int(esn & 63) +} + +func (r *RTPStatsReceiver) getHistorySlot(esn uint64, ehsn uint64) (int, int) { + if int64(esn-ehsn) < 0 { + return r.getOutOfOrderHistorySlot(esn, ehsn) + } + + return int(esn) % len(r.history), int(esn & 63) +} + +func (r *RTPStatsReceiver) setHistory(esn uint64, ehsn uint64) { + slot, offset := r.getHistorySlot(esn, ehsn) + if slot < 0 { + return + } + + r.history[slot] |= (1 << offset) +} + +func (r *RTPStatsReceiver) clearHistory(extStartInclusive uint64, extEndExclusive uint64, ehsn uint64) { + if extEndExclusive <= extStartInclusive { + return + } + + slot, offset := r.getHistorySlot(extStartInclusive, ehsn) + if slot < 0 { + return + } + for esn := extStartInclusive; esn != extEndExclusive; esn++ { + r.history[slot] &= ^(1 << offset) + offset++ + if offset > 63 { + offset -= 64 + slot++ + if slot >= len(r.history) { + slot -= len(r.history) + } + } + } +} + +func (r *RTPStatsReceiver) isLost(esn uint64, ehsn uint64) bool { + slot, offset := r.getHistorySlot(esn, ehsn) + if slot < 0 { + return false + } + + return r.history[slot]&(1<