dandri/livekit
1
0
Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-08-28 09:24:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

Make a lite version of sender stats to be used in relay down track. (#3069)

Browse Source
This commit is contained in:
Raja Subramanian
2024-10-06 13:01:08 +05:30
committed by GitHub
parent 3261560098
commit 5e22582c66
5 changed files with 248 additions and 126 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/sfu/rtpstats/rtpstats_base.go
-119
View File
@@ -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)
pkg/sfu/rtpstats/rtpstats_base_lite.go
+94 -3
View File
@@ -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
pkg/sfu/rtpstats/rtpstats_receiver_lite.go
+1 -1
View File
@@ -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--
pkg/sfu/rtpstats/rtpstats_sender.go
+31 -3
View File
@@ -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
}
pkg/sfu/rtpstats/rtpstats_sender_lite.go
+122
View File
@@ -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
}
// -------------------------------------------------------------------