dandri/livekit
1
0
Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-08-03 15:29:39 +00:00
Code Issues Packages Projects Releases Wiki Activity

Reduce packet meta data cache - part 1 (#2073)

Browse Source 
* 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
This commit is contained in:
Raja Subramanian
2023-09-15 21:39:03 +05:30
committed by GitHub
parent 7ac0abdf1b
commit 044f6cec40
4 changed files with 196 additions and 156 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/sfu/buffer/rtpstats_base.go
+90 -53
View File
@@ -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 {
pkg/sfu/buffer/rtpstats_receiver.go
+73 -10
View File
@@ -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<<offset) == 0
}
// ----------------------------------
pkg/sfu/buffer/rtpstats_receiver_test.go
+8 -57
View File
@@ -205,14 +205,8 @@ func Test_RTPStatsReceiver_Update(t *testing.T) {
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,
r.sequenceNumber.GetExtendedHighest(),
)
require.Equal(t, uint64(16), intervalStats.packetsLost)
// test sequence number cache
// test sequence number history
// with a gap
sequenceNumber += 2
timestamp += 6000
@@ -230,14 +224,7 @@ func Test_RTPStatsReceiver_Update(t *testing.T) {
require.Equal(t, uint64(sequenceNumber-1), flowState.LossStartInclusive)
require.Equal(t, uint64(sequenceNumber), flowState.LossEndExclusive)
require.Equal(t, uint64(17), r.packetsLost)
expectedSnInfo := snInfo{
hdrSize: 12,
pktSize: 1012,
isPaddingOnly: false,
marker: false,
isOutOfOrder: false,
}
require.Equal(t, expectedSnInfo, r.snInfos[sequenceNumber&cSnInfoMask])
require.True(t, r.isLost(uint64(sequenceNumber)-1, r.sequenceNumber.GetExtendedHighest()))
// out-of-order
sequenceNumber--
@@ -254,23 +241,8 @@ func Test_RTPStatsReceiver_Update(t *testing.T) {
)
require.False(t, flowState.HasLoss)
require.Equal(t, uint64(16), r.packetsLost)
expectedSnInfo = snInfo{
hdrSize: 12,
pktSize: 1011,
isPaddingOnly: false,
marker: false,
isOutOfOrder: true,
}
require.Equal(t, expectedSnInfo, r.snInfos[sequenceNumber&cSnInfoMask])
// check that last one is still fine
expectedSnInfo = snInfo{
hdrSize: 12,
pktSize: 1012,
isPaddingOnly: false,
marker: false,
isOutOfOrder: false,
}
require.Equal(t, expectedSnInfo, r.snInfos[(sequenceNumber+1)&cSnInfoMask])
require.Equal(t, uint64(4), r.packetsOutOfOrder)
require.False(t, r.isLost(uint64(sequenceNumber), r.sequenceNumber.GetExtendedHighest()))
// padding only
sequenceNumber += 2
@@ -286,31 +258,10 @@ func Test_RTPStatsReceiver_Update(t *testing.T) {
)
require.False(t, flowState.HasLoss)
require.Equal(t, uint64(16), r.packetsLost)
expectedSnInfo = snInfo{
hdrSize: 12,
pktSize: 37,
isPaddingOnly: true,
marker: false,
isOutOfOrder: false,
}
require.Equal(t, expectedSnInfo, r.snInfos[sequenceNumber&cSnInfoMask])
// check that last two are still fine
expectedSnInfo = snInfo{
hdrSize: 12,
pktSize: 1011,
isPaddingOnly: false,
marker: false,
isOutOfOrder: true,
}
require.Equal(t, expectedSnInfo, r.snInfos[(sequenceNumber-2)&cSnInfoMask])
expectedSnInfo = snInfo{
hdrSize: 12,
pktSize: 1012,
isPaddingOnly: false,
marker: false,
isOutOfOrder: false,
}
require.Equal(t, expectedSnInfo, r.snInfos[(sequenceNumber-1)&cSnInfoMask])
require.Equal(t, uint64(4), r.packetsOutOfOrder)
require.False(t, r.isLost(uint64(sequenceNumber), r.sequenceNumber.GetExtendedHighest()))
require.False(t, r.isLost(uint64(sequenceNumber)-1, r.sequenceNumber.GetExtendedHighest()))
require.False(t, r.isLost(uint64(sequenceNumber)-2, r.sequenceNumber.GetExtendedHighest()))
r.Stop()
}
pkg/sfu/buffer/rtpstats_sender.go
+25 -36
View File
@@ -51,14 +51,14 @@ type RTPStatsSender struct {
maxJitterFromRR float64
nextSenderSnapshotID uint32
senderSnapshots map[uint32]*senderSnapshot
senderSnapshots []senderSnapshot
}
func NewRTPStatsSender(params RTPStatsParams) *RTPStatsSender {
return &RTPStatsSender{
rtpStatsBase: newRTPStatsBase(params),
nextSenderSnapshotID: cFirstSnapshotID,
senderSnapshots: make(map[uint32]*senderSnapshot),
senderSnapshots: make([]senderSnapshot, 2),
}
}
@@ -86,10 +86,8 @@ func (r *RTPStatsSender) Seed(from *RTPStatsSender) {
r.maxJitterFromRR = from.maxJitterFromRR
r.nextSenderSnapshotID = from.nextSenderSnapshotID
for id, ss := range from.senderSnapshots {
ssCopy := *ss
r.senderSnapshots[id] = &ssCopy
}
r.senderSnapshots = make([]senderSnapshot, cap(from.senderSnapshots))
copy(r.senderSnapshots, from.senderSnapshots)
}
func (r *RTPStatsSender) NewSnapshotId() uint32 {
@@ -104,12 +102,17 @@ func (r *RTPStatsSender) NewSenderSnapshotId() uint32 {
defer r.lock.Unlock()
id := r.nextSenderSnapshotID
r.nextSenderSnapshotID++
if cap(r.senderSnapshots) < int(r.nextSenderSnapshotID) {
senderSnapshots := make([]senderSnapshot, r.nextSenderSnapshotID)
copy(senderSnapshots, r.senderSnapshots)
r.senderSnapshots = senderSnapshots
}
if r.initialized {
r.senderSnapshots[id] = &senderSnapshot{
snapshot: snapshot{
startTime: time.Now(),
extStartSN: r.extStartSN,
},
r.senderSnapshots[id] = senderSnapshot{
snapshot: r.initSnapshot(time.Now(), r.extStartSN),
extStartSNFromRR: r.extStartSN,
}
}
@@ -153,14 +156,16 @@ func (r *RTPStatsSender) Update(
// initialize snapshots if any
for i := uint32(cFirstSnapshotID); i < r.nextSnapshotID; i++ {
r.snapshots[i] = &snapshot{
r.snapshots[i] = snapshot{
isValid: true,
startTime: r.startTime,
extStartSN: r.extStartSN,
}
}
for i := uint32(cFirstSnapshotID); i < r.nextSenderSnapshotID; i++ {
r.senderSnapshots[i] = &senderSnapshot{
r.senderSnapshots[i] = senderSnapshot{
snapshot: snapshot{
isValid: true,
startTime: r.startTime,
extStartSN: r.extStartSN,
},
@@ -580,39 +585,23 @@ func (r *RTPStatsSender) getAndResetSenderSnapshot(senderSnapshotID uint32) (*se
}
then := r.senderSnapshots[senderSnapshotID]
if then == nil {
then = &senderSnapshot{
snapshot: snapshot{
startTime: r.startTime,
extStartSN: r.extStartSN,
},
if !then.isValid {
then = senderSnapshot{
snapshot: r.initSnapshot(r.startTime, r.extStartSN),
extStartSNFromRR: r.extStartSN,
}
r.senderSnapshots[senderSnapshotID] = then
}
// snapshot now
r.senderSnapshots[senderSnapshotID] = &senderSnapshot{
snapshot: snapshot{
startTime: r.lastRRTime,
extStartSN: r.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,
},
now := senderSnapshot{
snapshot: r.getSnapshot(r.lastRRTime, r.extHighestSN+1),
extStartSNFromRR: r.extHighestSNFromRR + (r.extStartSN & 0xFFFF_FFFF_FFFF_0000) + 1,
packetsLostFromRR: r.packetsLostFromRR,
maxJitterFromRR: r.jitterFromRR,
}
// make a copy so that it can be used independently
now := *r.senderSnapshots[senderSnapshotID]
return then, &now
r.senderSnapshots[senderSnapshotID] = now
return &then, &now
}
// -------------------------------------------------------------------