From d5cc567140de8c5821c3af073676b29980a48a3f Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Thu, 21 Nov 2024 16:41:03 +0530 Subject: [PATCH] Log more details of RTP stats snap shots. (#3190) * Log more details of RTP stats snap shots. Seeing cases of loss more than 100%. Logging snap shots to understand it better. * log message * use delta to update packets lost from RR * remove cast --- pkg/sfu/downtrack.go | 18 +++++-- pkg/sfu/playoutdelay.go | 68 +++++++++++++++++++------- pkg/sfu/rtpstats/rtpstats_base.go | 66 ++++++++++++++++++++++--- pkg/sfu/rtpstats/rtpstats_base_lite.go | 54 +++++++++++++++++--- pkg/sfu/rtpstats/rtpstats_sender.go | 60 ++++++++++++++++------- 5 files changed, 215 insertions(+), 51 deletions(-) diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index 32640524b..2de3ace79 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -140,15 +140,17 @@ var ( // ------------------------------------------------------------------- type DownTrackState struct { - RTPStats *rtpstats.RTPStatsSender - DeltaStatsSenderSnapshotId uint32 - ForwarderState *livekit.RTPForwarderState + RTPStats *rtpstats.RTPStatsSender + DeltaStatsSenderSnapshotId uint32 + ForwarderState *livekit.RTPForwarderState + PlayoutDelayControllerState PlayoutDelayControllerState } func (d DownTrackState) MarshalLogObject(e zapcore.ObjectEncoder) error { e.AddObject("RTPStats", d.RTPStats) e.AddUint32("DeltaStatsSenderSnapshotId", d.DeltaStatsSenderSnapshotId) e.AddObject("ForwarderState", logger.Proto(d.ForwarderState)) + e.AddObject("PlayoutDelayControllerState", d.PlayoutDelayControllerState) return nil } @@ -1257,11 +1259,16 @@ func (d *DownTrack) MaxLayer() buffer.VideoLayer { } func (d *DownTrack) GetState() DownTrackState { - return DownTrackState{ + dts := DownTrackState{ RTPStats: d.rtpStats, DeltaStatsSenderSnapshotId: d.deltaStatsSenderSnapshotId, ForwarderState: d.forwarder.GetState(), } + + if d.playoutDelay != nil { + dts.PlayoutDelayControllerState = d.playoutDelay.GetState() + } + return dts } func (d *DownTrack) SeedState(state DownTrackState) { @@ -1271,6 +1278,9 @@ func (d *DownTrack) SeedState(state DownTrackState) { if state.RTPStats != nil { d.rtpStats.Seed(state.RTPStats) d.deltaStatsSenderSnapshotId = state.DeltaStatsSenderSnapshotId + if d.playoutDelay != nil { + d.playoutDelay.SeedState(state.PlayoutDelayControllerState) + } } d.forwarder.SeedState(state.ForwarderState) } diff --git a/pkg/sfu/playoutdelay.go b/pkg/sfu/playoutdelay.go index 573c16a6a..6f3ab3aad 100644 --- a/pkg/sfu/playoutdelay.go +++ b/pkg/sfu/playoutdelay.go @@ -22,20 +22,25 @@ import ( pd "github.com/livekit/livekit-server/pkg/sfu/rtpextension/playoutdelay" "github.com/livekit/livekit-server/pkg/sfu/rtpstats" "github.com/livekit/protocol/logger" + "go.uber.org/zap/zapcore" ) +const ( + jitterMultiToDelay = 10 + targetDelayLogThreshold = 500 + + // limit max delay change to make it smoother for a/v sync + maxDelayChangePerSec = 80 +) + +// ---------------------------------------------------- + type PlayoutDelayState int32 const ( PlayoutDelayStateChanged PlayoutDelayState = iota PlayoutDelaySending PlayoutDelayAcked - - jitterMultiToDelay = 10 - targetDelayLogThreshold = 500 - - // limit max delay change to make it smoother for a/v sync - maxDelayChangePerSec = 80 ) func (s PlayoutDelayState) String() string { @@ -50,6 +55,19 @@ func (s PlayoutDelayState) String() string { return "Unknown" } +// ---------------------------------------------------- + +type PlayoutDelayControllerState struct { + SenderSnapshotID uint32 +} + +func (p PlayoutDelayControllerState) MarshalLogObject(e zapcore.ObjectEncoder) error { + e.AddUint32("SenderSnapshotID", p.SenderSnapshotID) + return nil +} + +// ---------------------------------------------------- + type PlayoutDelayController struct { lock sync.Mutex state atomic.Int32 @@ -60,7 +78,7 @@ type PlayoutDelayController struct { sendingAtTime time.Time logger logger.Logger rtpStats *rtpstats.RTPStatsSender - snapshotID uint32 + senderSnapshotID uint32 highDelayCount atomic.Uint32 } @@ -73,24 +91,40 @@ func NewPlayoutDelayController(minDelay, maxDelay uint32, logger logger.Logger, maxDelay = pd.PlayoutDelayMaxValue } c := &PlayoutDelayController{ - currentDelay: minDelay, - minDelay: minDelay, - maxDelay: maxDelay, - logger: logger, - rtpStats: rtpStats, - snapshotID: rtpStats.NewSenderSnapshotId(), + currentDelay: minDelay, + minDelay: minDelay, + maxDelay: maxDelay, + logger: logger, + rtpStats: rtpStats, + senderSnapshotID: rtpStats.NewSenderSnapshotId(), } return c, c.createExtData() } +func (c *PlayoutDelayController) GetState() PlayoutDelayControllerState { + c.lock.Lock() + defer c.lock.Unlock() + + return PlayoutDelayControllerState{ + SenderSnapshotID: c.senderSnapshotID, + } +} + +func (c *PlayoutDelayController) SeedState(pdcs PlayoutDelayControllerState) { + c.lock.Lock() + defer c.lock.Unlock() + + c.senderSnapshotID = pdcs.SenderSnapshotID +} + func (c *PlayoutDelayController) SetJitter(jitter uint32) { - deltaInfo := c.rtpStats.DeltaInfoSender(c.snapshotID) + c.lock.Lock() + deltaInfoSender := c.rtpStats.DeltaInfoSender(c.senderSnapshotID) var nackPercent uint32 - if deltaInfo != nil && deltaInfo.Packets > 0 { - nackPercent = deltaInfo.Nacks * 100 / deltaInfo.Packets + if deltaInfoSender != nil && deltaInfoSender.Packets > 0 { + nackPercent = deltaInfoSender.Nacks * 100 / deltaInfoSender.Packets } - c.lock.Lock() targetDelay := jitter * jitterMultiToDelay if nackPercent > 60 { targetDelay += (nackPercent - 60) * 2 diff --git a/pkg/sfu/rtpstats/rtpstats_base.go b/pkg/sfu/rtpstats/rtpstats_base.go index b978d00fa..329275749 100644 --- a/pkg/sfu/rtpstats/rtpstats_base.go +++ b/pkg/sfu/rtpstats/rtpstats_base.go @@ -58,6 +58,36 @@ type RTPDeltaInfo struct { Firs uint32 } +func (r *RTPDeltaInfo) MarshalLogObject(e zapcore.ObjectEncoder) error { + if r == nil { + return nil + } + + e.AddTime("StartTime", r.StartTime) + e.AddTime("EndTime", r.EndTime) + e.AddUint32("Packets", r.Packets) + e.AddUint64("Bytes", r.Bytes) + e.AddUint64("HeaderBytes", r.HeaderBytes) + e.AddUint32("PacketsDuplicate", r.PacketsDuplicate) + e.AddUint64("BytesDuplicate", r.BytesDuplicate) + e.AddUint64("HeaderBytesDuplicate", r.HeaderBytesDuplicate) + e.AddUint32("PacketsPadding", r.PacketsPadding) + e.AddUint64("BytesPadding", r.BytesPadding) + e.AddUint64("HeaderBytesPadding", r.HeaderBytesPadding) + e.AddUint32("PacketsLost", r.PacketsLost) + e.AddUint32("PacketsMissing", r.PacketsMissing) + e.AddUint32("PacketsOutOfOrder", r.PacketsOutOfOrder) + e.AddUint32("Frames", r.Frames) + e.AddUint32("RttMax", r.RttMax) + e.AddFloat64("JitterMax", r.JitterMax) + e.AddUint32("Nacks", r.Nacks) + e.AddUint32("Plis", r.Plis) + e.AddUint32("Firs", r.Firs) + return nil +} + +// ------------------------------------------------------- + type snapshot struct { snapshotLite @@ -80,6 +110,27 @@ type snapshot struct { maxJitter float64 } +func (s *snapshot) MarshalLogObject(e zapcore.ObjectEncoder) error { + if s == nil { + return nil + } + + e.AddObject("snapshotLite", &s.snapshotLite) + e.AddUint64("headerBytes", s.headerBytes) + e.AddUint64("packetsDuplicate", s.packetsDuplicate) + e.AddUint64("bytesDuplicate", s.bytesDuplicate) + e.AddUint64("headerBytesDuplicate", s.headerBytesDuplicate) + e.AddUint64("packetsPadding", s.packetsPadding) + e.AddUint64("bytesPadding", s.bytesPadding) + e.AddUint64("headerBytesPadding", s.headerBytesPadding) + e.AddUint32("frames", s.frames) + e.AddUint32("plis", s.plis) + e.AddUint32("firs", s.firs) + e.AddUint32("maxRtt", s.maxRtt) + e.AddFloat64("maxJitter", s.maxJitter) + return nil +} + // ------------------------------------------------------------------ type wrappedRTPDriftLogger struct { @@ -399,12 +450,11 @@ func (r *rtpStatsBase) deltaInfo( } if packetsExpected > cNumSequenceNumbers { loggingFields = []interface{}{ - "startSN", then.extStartSN, - "endSN", now.extStartSN, - "packetsExpected", packetsExpected, - "startTime", startTime, - "endTime", endTime, + "snapshotID", snapshotID, + "snapshotNow", now, + "snapshotThen", then, "duration", endTime.Sub(startTime), + "packetsExpected", packetsExpected, } err = errors.New("too many packets expected in delta") return @@ -426,11 +476,13 @@ func (r *rtpStatsBase) deltaInfo( packetsPadding := now.packetsPadding - then.packetsPadding if packetsExpected < packetsPadding { loggingFields = []interface{}{ + "snapshotID", snapshotID, + "snapshotNow", now, + "snapshotThen", then, + "duration", endTime.Sub(startTime), "packetsExpected", packetsExpected, "packetsPadding", packetsPadding, "packetsLost", packetsLost, - "startSequenceNumber", then.extStartSN, - "endSequenceNumber", now.extStartSN - 1, } err = errors.New("padding packets more than expected") packetsExpected = 0 diff --git a/pkg/sfu/rtpstats/rtpstats_base_lite.go b/pkg/sfu/rtpstats/rtpstats_base_lite.go index ea6294351..cff08e078 100644 --- a/pkg/sfu/rtpstats/rtpstats_base_lite.go +++ b/pkg/sfu/rtpstats/rtpstats_base_lite.go @@ -44,6 +44,23 @@ type RTPDeltaInfoLite struct { Nacks uint32 } +func (r *RTPDeltaInfoLite) MarshalLogObject(e zapcore.ObjectEncoder) error { + if r == nil { + return nil + } + + e.AddTime("StartTime", r.StartTime) + e.AddTime("EndTime", r.EndTime) + e.AddUint32("Packets", r.Packets) + e.AddUint64("Bytes", r.Bytes) + e.AddUint32("PacketsLost", r.PacketsLost) + e.AddUint32("PacketsOutOfOrder", r.PacketsOutOfOrder) + e.AddUint32("Nacks", r.Nacks) + return nil +} + +// ------------------------------------------------------- + type snapshotLite struct { isValid bool @@ -59,6 +76,21 @@ type snapshotLite struct { nacks uint32 } +func (s *snapshotLite) MarshalLogObject(e zapcore.ObjectEncoder) error { + if s == nil { + return nil + } + + e.AddBool("isValid", s.isValid) + e.AddTime("startTime", s.startTime) + e.AddUint64("extStartSN", s.extStartSN) + e.AddUint64("bytes", s.bytes) + e.AddUint64("packetsOutOfOrder", s.packetsOutOfOrder) + e.AddUint64("packetsLost", s.packetsLost) + e.AddUint32("nacks", s.nacks) + return nil +} + // ------------------------------------------------------------------ type RTPStatsParams struct { @@ -280,17 +312,16 @@ func (r *rtpStatsBaseLite) deltaInfoLite( startTime := then.startTime endTime := now.startTime - packetsExpected := now.extStartSN - then.extStartSN + packetsExpected := uint32(now.extStartSN - then.extStartSN) if then.extStartSN > extHighestSN { packetsExpected = 0 } if packetsExpected > cNumSequenceNumbers { loggingFields = []interface{}{ - "startSN", then.extStartSN, - "endSN", now.extStartSN, + "snapshotLiteID", snapshotLiteID, + "snapshotLiteNow", now, + "snapshotLiteThen", then, "packetsExpected", packetsExpected, - "startTime", startTime, - "endTime", endTime, "duration", endTime.Sub(startTime).String(), } err = errors.New("too many packets expected in delta lite") @@ -308,11 +339,22 @@ func (r *rtpStatsBaseLite) deltaInfoLite( if int32(packetsLost) < 0 { packetsLost = 0 } + if packetsLost > packetsExpected { + loggingFields = []interface{}{ + "snapshotLiteID", snapshotLiteID, + "snapshotLiteNow", now, + "snapshotLiteThen", then, + "packetsExpected", packetsExpected, + "packetsLost", packetsLost, + "duration", endTime.Sub(startTime).String(), + } + err = errors.New("unexpected number of packets lost in delta lite") + } deltaInfoLite = &RTPDeltaInfoLite{ StartTime: startTime, EndTime: endTime, - Packets: uint32(packetsExpected), + Packets: packetsExpected, Bytes: now.bytes - then.bytes, PacketsLost: packetsLost, PacketsOutOfOrder: uint32(now.packetsOutOfOrder - then.packetsOutOfOrder), diff --git a/pkg/sfu/rtpstats/rtpstats_sender.go b/pkg/sfu/rtpstats/rtpstats_sender.go index ca9b355fb..441ece9f7 100644 --- a/pkg/sfu/rtpstats/rtpstats_sender.go +++ b/pkg/sfu/rtpstats/rtpstats_sender.go @@ -138,6 +138,30 @@ type senderSnapshot struct { intervalStats intervalStats } +func (s *senderSnapshot) MarshalLogObject(e zapcore.ObjectEncoder) error { + if s == nil { + return nil + } + + e.AddBool("isValid", s.isValid) + e.AddTime("startTime", s.startTime) + e.AddUint64("extStartSN", s.extStartSN) + e.AddUint64("bytes", s.bytes) + e.AddUint64("headerBytes", s.headerBytes) + e.AddUint64("packetsPadding", s.packetsPadding) + e.AddUint64("bytesPadding", s.bytesPadding) + e.AddUint64("headerBytesPadding", s.headerBytesPadding) + e.AddUint64("packetsDuplicate", s.packetsDuplicate) + e.AddUint64("bytesDuplicate", s.bytesDuplicate) + e.AddUint64("headerBytesDuplicate", s.headerBytesDuplicate) + e.AddUint64("packetsOutOfOrder", s.packetsOutOfOrder) + e.AddUint64("packetsLostFeed", s.packetsLostFeed) + e.AddUint64("packetsLost", s.packetsLost) + e.AddUint32("frames", s.frames) + e.AddUint32("nacks", s.nacks) + return nil +} + // ------------------------------------------------------------------- type rttMarker struct { @@ -514,9 +538,12 @@ func (r *RTPStatsSender) UpdateFromReceiverReport(rr rtcp.ReceptionReport) (rtt } } - r.packetsLostFromRR = r.packetsLostFromRR&0xFFFF_FFFF_FF00_0000 + uint64(rr.TotalLost) - if ((rr.TotalLost-r.lastRR.TotalLost)&((1<<24)-1)) < (1<<23) && rr.TotalLost < r.lastRR.TotalLost { - r.packetsLostFromRR += (1 << 24) + lossDelta := (rr.TotalLost - r.lastRR.TotalLost) & ((1 << 24) - 1) + if lossDelta > 0 && lossDelta < (1<<23) { + r.packetsLostFromRR += uint64(lossDelta) + if rr.TotalLost < r.lastRR.TotalLost { + r.packetsLostFromRR += (1 << 24) + } } if isRttChanged { @@ -563,6 +590,7 @@ func (r *RTPStatsSender) UpdateFromReceiverReport(rr rtcp.ReceptionReport) (rtt "packetsInInterval", extReceivedRRSN-s.extLastRRSN, "rtpStats", lockedRTPStatsSenderLogEncoder{r}, ) + s.extLastRRSN = extReceivedRRSN continue } @@ -768,11 +796,10 @@ func (r *RTPStatsSender) DeltaInfoSender(senderSnapshotID uint32) *RTPDeltaInfo if packetsExpected > cNumSequenceNumbers { r.logger.Warnw( "too many packets expected in delta (sender)", nil, - "startSN", then.extStartSN, - "endSN", now.extStartSN, + "senderSnapshotID", senderSnapshotID, + "senderSnapshotNow", now, + "senderSnapshotThen", then, "packetsExpected", packetsExpected, - "startTime", startTime, - "endTime", endTime, "duration", endTime.Sub(startTime), "rtpStats", lockedRTPStatsSenderLogEncoder{r}, ) @@ -793,15 +820,14 @@ func (r *RTPStatsSender) DeltaInfoSender(senderSnapshotID uint32) *RTPDeltaInfo } if packetsLost > packetsExpected { r.logger.Warnw( - "unexpected number of packets lost", - fmt.Errorf( - "start: %d, end: %d, expected: %d, lost: report: %d, feed: %d", - then.extStartSN, - now.extStartSN, - packetsExpected, - packetsLost, - packetsLostFeed, - ), + "unexpected number of packets lost", nil, + "senderSnapshotID", senderSnapshotID, + "senderSnapshotNow", now, + "senderSnapshotThen", then, + "packetsExpected", packetsExpected, + "packetsLost", packetsLost, + "packetsLostFeed", packetsLostFeed, + "duration", endTime.Sub(startTime), "rtpStats", lockedRTPStatsSenderLogEncoder{r}, ) packetsLost = packetsExpected @@ -906,7 +932,7 @@ func (r *RTPStatsSender) getSenderSnapshot(startTime time.Time, s *senderSnapsho bytesDuplicate: r.bytesDuplicate, headerBytesDuplicate: r.headerBytesDuplicate, packetsOutOfOrder: s.packetsOutOfOrder + s.intervalStats.packetsOutOfOrder, - packetsLostFeed: r.packetsLost, + packetsLostFeed: s.packetsLost + s.intervalStats.packetsLost, packetsLost: r.packetsLostFromRR, frames: s.frames + s.intervalStats.frames, nacks: r.nacks,