From 044f6cec408fa35f02d000df01b573455fe9991f Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Fri, 15 Sep 2023 21:39:03 +0530 Subject: [PATCH] Reduce packet meta data cache - part 1 (#2073) * Reduce packet meta data cache - part 1 Packet meta data cache takes a good amount of space. That cache is 8K entries deep and each entry is 8 bytes. So, that takes 64KB per RTP stream. It is mostly needed for down stream to line up with receiver reports. So, removing cache from up stream (RTPStatsReceiver) as part 1. Will look at optimising the down stream in part 2. * Remove caching from RTPStatsReceiver * clean up a bit more * maintain history and fix test --- pkg/sfu/buffer/rtpstats_base.go | 143 ++++++++++++++--------- pkg/sfu/buffer/rtpstats_receiver.go | 83 +++++++++++-- pkg/sfu/buffer/rtpstats_receiver_test.go | 65 ++--------- pkg/sfu/buffer/rtpstats_sender.go | 61 ++++------ 4 files changed, 196 insertions(+), 156 deletions(-) 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<