From 5e22582c66cc09434ea68a94706c5657eaf25eed Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Sun, 6 Oct 2024 13:01:08 +0530 Subject: [PATCH] Make a lite version of sender stats to be used in relay down track. (#3069) --- pkg/sfu/rtpstats/rtpstats_base.go | 119 -------------------- pkg/sfu/rtpstats/rtpstats_base_lite.go | 97 +++++++++++++++- pkg/sfu/rtpstats/rtpstats_receiver_lite.go | 2 +- pkg/sfu/rtpstats/rtpstats_sender.go | 34 +++++- pkg/sfu/rtpstats/rtpstats_sender_lite.go | 122 +++++++++++++++++++++ 5 files changed, 248 insertions(+), 126 deletions(-) create mode 100644 pkg/sfu/rtpstats/rtpstats_sender_lite.go diff --git a/pkg/sfu/rtpstats/rtpstats_base.go b/pkg/sfu/rtpstats/rtpstats_base.go index 21bfe8b9d..19bee8d57 100644 --- a/pkg/sfu/rtpstats/rtpstats_base.go +++ b/pkg/sfu/rtpstats/rtpstats_base.go @@ -161,16 +161,6 @@ type rtpStatsBase struct { jitter float64 maxJitter float64 - nackAcks uint32 - nackMisses uint32 - nackRepeated uint32 - - plis uint32 - lastPli time.Time - - layerLockPlis uint32 - lastLayerLockPli time.Time - firs uint32 lastFir time.Time @@ -230,9 +220,6 @@ func (r *rtpStatsBase) seed(from *rtpStatsBase) bool { r.plis = from.plis r.lastPli = from.lastPli - r.layerLockPlis = from.layerLockPlis - r.lastLayerLockPli = from.lastLayerLockPli - r.firs = from.firs r.lastFir = from.lastFir @@ -267,92 +254,6 @@ func (r *rtpStatsBase) newSnapshotID(extStartSN uint64) uint32 { return id } -func (r *rtpStatsBase) UpdateNackProcessed(nackAckCount uint32, nackMissCount uint32, nackRepeatedCount uint32) { - r.lock.Lock() - defer r.lock.Unlock() - - if !r.endTime.IsZero() { - return - } - - r.nackAcks += nackAckCount - r.nackMisses += nackMissCount - r.nackRepeated += nackRepeatedCount -} - -func (r *rtpStatsBase) CheckAndUpdatePli(throttle int64, force bool) bool { - r.lock.Lock() - defer r.lock.Unlock() - - if !r.endTime.IsZero() || (!force && time.Now().UnixNano()-r.lastPli.UnixNano() < throttle) { - return false - } - r.updatePliLocked(1) - r.updatePliTimeLocked() - return true -} - -func (r *rtpStatsBase) UpdatePliAndTime(pliCount uint32) { - r.lock.Lock() - defer r.lock.Unlock() - - if !r.endTime.IsZero() { - return - } - - r.updatePliLocked(pliCount) - r.updatePliTimeLocked() -} - -func (r *rtpStatsBase) UpdatePli(pliCount uint32) { - r.lock.Lock() - defer r.lock.Unlock() - - if !r.endTime.IsZero() { - return - } - - r.updatePliLocked(pliCount) -} - -func (r *rtpStatsBase) updatePliLocked(pliCount uint32) { - r.plis += pliCount -} - -func (r *rtpStatsBase) UpdatePliTime() { - r.lock.Lock() - defer r.lock.Unlock() - - if !r.endTime.IsZero() { - return - } - - r.updatePliTimeLocked() -} - -func (r *rtpStatsBase) updatePliTimeLocked() { - r.lastPli = time.Now() -} - -func (r *rtpStatsBase) LastPli() time.Time { - r.lock.RLock() - defer r.lock.RUnlock() - - return r.lastPli -} - -func (r *rtpStatsBase) UpdateLayerLockPliAndTime(pliCount uint32) { - r.lock.Lock() - defer r.lock.Unlock() - - if !r.endTime.IsZero() { - return - } - - r.layerLockPlis += pliCount - r.lastLayerLockPli = time.Now() -} - func (r *rtpStatsBase) UpdateFir(firCount uint32) { r.lock.Lock() defer r.lock.Unlock() @@ -603,16 +504,6 @@ func (r *rtpStatsBase) marshalLogObject( e.AddFloat64("jitter", r.jitter) e.AddFloat64("maxJitter", r.maxJitter) - e.AddUint32("nackAcks", r.nackAcks) - e.AddUint32("nackMisses", r.nackMisses) - e.AddUint32("nackRepeated", r.nackRepeated) - - e.AddUint32("plis", r.plis) - e.AddTime("lastPli", r.lastPli) - - e.AddUint32("layerLockPlis", r.layerLockPlis) - e.AddTime("lastLayerLockPli", r.lastLayerLockPli) - e.AddUint32("firs", r.firs) e.AddTime("lastFir", r.lastFir) @@ -666,16 +557,6 @@ func (r *rtpStatsBase) toProto( p.JitterCurrent = jitter / float64(r.params.ClockRate) * 1e6 p.JitterMax = maxJitter / float64(r.params.ClockRate) * 1e6 - p.NackAcks = r.nackAcks - p.NackMisses = r.nackMisses - p.NackRepeated = r.nackRepeated - - p.Plis = r.plis - p.LastPli = timestamppb.New(r.lastPli) - - p.LayerLockPlis = r.layerLockPlis - p.LastLayerLockPli = timestamppb.New(r.lastLayerLockPli) - p.Firs = r.firs p.LastFir = timestamppb.New(r.lastFir) diff --git a/pkg/sfu/rtpstats/rtpstats_base_lite.go b/pkg/sfu/rtpstats/rtpstats_base_lite.go index 2bb9197dc..6763c1548 100644 --- a/pkg/sfu/rtpstats/rtpstats_base_lite.go +++ b/pkg/sfu/rtpstats/rtpstats_base_lite.go @@ -85,7 +85,13 @@ type rtpStatsBaseLite struct { gapHistogram [cGapHistogramNumBins]uint32 - nacks uint32 + nacks uint32 + nackAcks uint32 + nackMisses uint32 + nackRepeated uint32 + + plis uint32 + lastPli time.Time nextSnapshotLiteID uint32 snapshotLites []snapshotLite @@ -171,6 +177,80 @@ func (r *rtpStatsBaseLite) UpdateNack(nackCount uint32) { r.nacks += nackCount } +func (r *rtpStatsBaseLite) UpdateNackProcessed(nackAckCount uint32, nackMissCount uint32, nackRepeatedCount uint32) { + r.lock.Lock() + defer r.lock.Unlock() + + if !r.endTime.IsZero() { + return + } + + r.nackAcks += nackAckCount + r.nackMisses += nackMissCount + r.nackRepeated += nackRepeatedCount +} + +func (r *rtpStatsBaseLite) CheckAndUpdatePli(throttle int64, force bool) bool { + r.lock.Lock() + defer r.lock.Unlock() + + if !r.endTime.IsZero() || (!force && time.Now().UnixNano()-r.lastPli.UnixNano() < throttle) { + return false + } + r.updatePliLocked(1) + r.updatePliTimeLocked() + return true +} + +func (r *rtpStatsBaseLite) UpdatePliAndTime(pliCount uint32) { + r.lock.Lock() + defer r.lock.Unlock() + + if !r.endTime.IsZero() { + return + } + + r.updatePliLocked(pliCount) + r.updatePliTimeLocked() +} + +func (r *rtpStatsBaseLite) UpdatePli(pliCount uint32) { + r.lock.Lock() + defer r.lock.Unlock() + + if !r.endTime.IsZero() { + return + } + + r.updatePliLocked(pliCount) +} + +func (r *rtpStatsBaseLite) updatePliLocked(pliCount uint32) { + r.plis += pliCount +} + +func (r *rtpStatsBaseLite) UpdatePliTime() { + r.lock.Lock() + defer r.lock.Unlock() + + if !r.endTime.IsZero() { + return + } + + r.updatePliTimeLocked() +} + +func (r *rtpStatsBaseLite) updatePliTimeLocked() { + r.lastPli = time.Now() +} + +func (r *rtpStatsBaseLite) LastPli() time.Time { + r.lock.RLock() + defer r.lock.RUnlock() + + return r.lastPli +} + func (r *rtpStatsBaseLite) getPacketsSeen(extStartSN, extHighestSN uint64) uint64 { packetsExpected := getPacketsExpected(extStartSN, extHighestSN) if r.packetsLost > packetsExpected { @@ -236,8 +316,8 @@ func (r *rtpStatsBaseLite) deltaInfoLite( } func (r *rtpStatsBaseLite) marshalLogObject(e zapcore.ObjectEncoder, packetsExpected, packetsSeenMinusPadding uint64) (float64, error) { - if r == nil { - return 0, errors.New("no object") + if r == nil || !r.initialized { + return 0, errors.New("not initialized") } endTime := r.endTime @@ -289,6 +369,12 @@ func (r *rtpStatsBaseLite) marshalLogObject(e zapcore.ObjectEncoder, packetsExpe } e.AddUint32("nacks", r.nacks) + e.AddUint32("nackAcks", r.nackAcks) + e.AddUint32("nackMisses", r.nackMisses) + e.AddUint32("nackRepeated", r.nackRepeated) + + e.AddUint32("plis", r.plis) + e.AddTime("lastPli", r.lastPli) return elapsedSeconds, nil } @@ -325,6 +411,11 @@ func (r *rtpStatsBaseLite) toProto(packetsExpected, packetsSeenMinusPadding, pac PacketLossPercentage: packetLostPercentage, PacketsOutOfOrder: uint32(r.packetsOutOfOrder), Nacks: r.nacks, + NackAcks: r.nackAcks, + NackMisses: r.nackMisses, + NackRepeated: r.nackRepeated, + Plis: r.plis, + LastPli: timestamppb.New(r.lastPli), } gapsPresent := false diff --git a/pkg/sfu/rtpstats/rtpstats_receiver_lite.go b/pkg/sfu/rtpstats/rtpstats_receiver_lite.go index 93dc222ab..5f6ef29c7 100644 --- a/pkg/sfu/rtpstats/rtpstats_receiver_lite.go +++ b/pkg/sfu/rtpstats/rtpstats_receiver_lite.go @@ -99,8 +99,8 @@ func (r *RTPStatsReceiverLite) Update(packetTime int64, packetSize int, sequence return } } - gapSN := int64(resSN.ExtendedVal - resSN.PreExtendedHighest) + gapSN := int64(resSN.ExtendedVal - resSN.PreExtendedHighest) if gapSN <= 0 { // duplicate OR out-of-order r.packetsOutOfOrder++ // counting duplicate as out-of-order r.packetsLost-- diff --git a/pkg/sfu/rtpstats/rtpstats_sender.go b/pkg/sfu/rtpstats/rtpstats_sender.go index 2fa706a13..ab4799c8b 100644 --- a/pkg/sfu/rtpstats/rtpstats_sender.go +++ b/pkg/sfu/rtpstats/rtpstats_sender.go @@ -22,6 +22,7 @@ import ( "github.com/pion/rtcp" "go.uber.org/zap/zapcore" + "google.golang.org/protobuf/types/known/timestamppb" "github.com/livekit/mediatransportutil" "github.com/livekit/protocol/livekit" @@ -30,8 +31,6 @@ import ( const ( cSnInfoSize = 4096 cSnInfoMask = cSnInfoSize - 1 - - cSenderReportInitialWait = time.Second ) // ------------------------------------------------------------------- @@ -158,6 +157,9 @@ type RTPStatsSender struct { snInfos [cSnInfoSize]snInfo + layerLockPlis uint32 + lastLayerLockPli time.Time + nextSenderSnapshotID uint32 senderSnapshots []senderSnapshot @@ -204,6 +206,9 @@ func (r *RTPStatsSender) Seed(from *RTPStatsSender) { r.nextSenderSnapshotID = from.nextSenderSnapshotID r.senderSnapshots = make([]senderSnapshot, cap(from.senderSnapshots)) copy(r.senderSnapshots, from.senderSnapshots) + + r.layerLockPlis = from.layerLockPlis + r.lastLayerLockPli = from.lastLayerLockPli } func (r *RTPStatsSender) NewSnapshotId() uint32 { @@ -435,6 +440,18 @@ func (r *RTPStatsSender) Update( } } +func (r *RTPStatsSender) UpdateLayerLockPliAndTime(pliCount uint32) { + r.lock.Lock() + defer r.lock.Unlock() + + if !r.endTime.IsZero() { + return + } + + r.layerLockPlis += pliCount + r.lastLayerLockPli = time.Now() +} + func (r *RTPStatsSender) GetPacketsSeenMinusPadding() uint64 { r.lock.RLock() defer r.lock.RUnlock() @@ -812,7 +829,7 @@ func (r *RTPStatsSender) ToProto() *livekit.RTPStats { r.lock.RLock() defer r.lock.RUnlock() - return r.toProto( + p := r.toProto( getPacketsExpected(r.extStartSN, r.extHighestSN), r.getPacketsSeenMinusPadding(r.extStartSN, r.extHighestSN), r.packetsLostFromRR, @@ -821,6 +838,12 @@ func (r *RTPStatsSender) ToProto() *livekit.RTPStats { r.jitterFromRR, r.maxJitterFromRR, ) + + if p != nil { + p.LayerLockPlis = r.layerLockPlis + p.LastLayerLockPli = timestamppb.New(r.lastLayerLockPli) + } + return p } func (r *RTPStatsSender) getAndResetSenderSnapshot(senderSnapshotID uint32) (*senderSnapshot, *senderSnapshot) { @@ -1001,14 +1024,19 @@ func (r lockedRTPStatsSenderLogEncoder) MarshalLogObject(e zapcore.ObjectEncoder e.AddUint64("extStartSN", r.extStartSN) e.AddUint64("extHighestSN", r.extHighestSN) + e.AddUint64("extStartTS", r.extStartTS) e.AddUint64("extHighestTS", r.extHighestTS) + e.AddTime("lastRRTime", r.lastRRTime) e.AddReflected("lastRR", r.lastRR) e.AddUint64("extHighestSNFromRR", r.extHighestSNFromRR) e.AddUint64("packetsLostFromRR", r.packetsLostFromRR) e.AddFloat64("jitterFromRR", r.jitterFromRR) e.AddFloat64("maxJitterFromRR", r.maxJitterFromRR) + + e.AddUint32("layerLockPlis", r.layerLockPlis) + e.AddTime("lastLayerLockPli", r.lastLayerLockPli) return nil } diff --git a/pkg/sfu/rtpstats/rtpstats_sender_lite.go b/pkg/sfu/rtpstats/rtpstats_sender_lite.go new file mode 100644 index 000000000..9590ae569 --- /dev/null +++ b/pkg/sfu/rtpstats/rtpstats_sender_lite.go @@ -0,0 +1,122 @@ +// Copyright 2023 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package rtpstats + +import ( + "time" + + "github.com/livekit/protocol/livekit" + "go.uber.org/zap/zapcore" +) + +type RTPStatsSenderLite struct { + *rtpStatsBaseLite + + extStartSN uint64 + extHighestSN uint64 +} + +func NewRTPStatsSenderLite(params RTPStatsParams) *RTPStatsSenderLite { + return &RTPStatsSenderLite{ + rtpStatsBaseLite: newRTPStatsBaseLite(params), + } +} + +func (r *RTPStatsSenderLite) Update(packetTime int64, packetSize int, extSequenceNumber uint64) { + r.lock.Lock() + defer r.lock.Unlock() + + if !r.endTime.IsZero() { + return + } + + if !r.initialized { + r.initialized = true + + r.startTime = time.Now() + + r.extStartSN = extSequenceNumber + r.extHighestSN = extSequenceNumber - 1 + + r.logger.Debugw( + "rtp sender lite stream start", + "rtpStats", lockedRTPStatsSenderLiteLogEncoder{r}, + ) + } + + gapSN := int64(extSequenceNumber - r.extHighestSN) + if gapSN <= 0 { // duplicate OR out-of-order + r.packetsOutOfOrder++ // counting duplicate as out-of-order + r.packetsLost-- + } else { // in-order + r.updateGapHistogram(int(gapSN)) + r.packetsLost += uint64(gapSN - 1) + + r.extHighestSN = extSequenceNumber + } + + r.bytes += uint64(packetSize) +} + +func (r *RTPStatsSenderLite) MarshalLogObject(e zapcore.ObjectEncoder) error { + if r == nil { + return nil + } + + r.lock.RLock() + defer r.lock.RUnlock() + + return lockedRTPStatsSenderLiteLogEncoder{r}.MarshalLogObject(e) +} + +func (r *RTPStatsSenderLite) ToProto() *livekit.RTPStats { + r.lock.RLock() + defer r.lock.RUnlock() + + return r.rtpStatsBaseLite.toProto(r.extStartSN, r.extHighestSN, r.packetsLost) +} + +func (r *RTPStatsSenderLite) ExtHighestSequenceNumber() uint64 { + r.lock.RLock() + defer r.lock.RUnlock() + + return r.extHighestSN +} + +// ------------------------------------------------------------------- + +type lockedRTPStatsSenderLiteLogEncoder struct { + *RTPStatsSenderLite +} + +func (r lockedRTPStatsSenderLiteLogEncoder) MarshalLogObject(e zapcore.ObjectEncoder) error { + if r.RTPStatsSenderLite == nil { + return nil + } + + if _, err := r.rtpStatsBaseLite.marshalLogObject( + e, + getPacketsExpected(r.extStartSN, r.extHighestSN), + getPacketsExpected(r.extStartSN, r.extHighestSN), + ); err != nil { + return err + } + + e.AddUint64("extStartSN", r.extStartSN) + e.AddUint64("extHighestSN", r.extHighestSN) + return nil +} + +// -------------------------------------------------------------------