From d07d84f99f8d641f055c563764a6be5f9cc845db Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Sat, 23 Nov 2024 10:40:10 +0530 Subject: [PATCH] Sender side snap shot clean up and logging. (#3196) * Sender side snap shot clean up and logging. Seeing cases of sender snap shot packet loss much higher the actual packets some times. Tracking a bit more to understand that better. - Rename variables to indicate what is coming from feed side clearly - Fixed an issue with wrong init of feed side loss in snapshot - Just use the loss from receiver report as it can go back (receiver would subtract on receiving out-of-order packet). - keep track sof reports in a snapshot (this is temporary for debugging/understanding it better and will be removed later) * remove check --- pkg/sfu/rtpstats/rtpstats_sender.go | 144 +++++++++++++++++----------- 1 file changed, 88 insertions(+), 56 deletions(-) diff --git a/pkg/sfu/rtpstats/rtpstats_sender.go b/pkg/sfu/rtpstats/rtpstats_sender.go index 441ece9f7..a9f573d56 100644 --- a/pkg/sfu/rtpstats/rtpstats_sender.go +++ b/pkg/sfu/rtpstats/rtpstats_sender.go @@ -53,16 +53,16 @@ type snInfo struct { // ------------------------------------------------------------------- type intervalStats struct { - packets uint64 - bytes uint64 - headerBytes uint64 - packetsPadding uint64 - bytesPadding uint64 - headerBytesPadding uint64 - packetsLost uint64 - packetsOutOfOrder uint64 - frames uint32 - packetsNotFound uint64 + packets uint64 + bytes uint64 + headerBytes uint64 + packetsPadding uint64 + bytesPadding uint64 + headerBytesPadding uint64 + packetsLostFeed uint64 + packetsOutOfOrderFeed uint64 + frames uint32 + packetsNotFoundMetadata uint64 } func (is *intervalStats) aggregate(other *intervalStats) { @@ -76,10 +76,10 @@ func (is *intervalStats) aggregate(other *intervalStats) { is.packetsPadding += other.packetsPadding is.bytesPadding += other.bytesPadding is.headerBytesPadding += other.headerBytesPadding - is.packetsLost += other.packetsLost - is.packetsOutOfOrder += other.packetsOutOfOrder + is.packetsLostFeed += other.packetsLostFeed + is.packetsOutOfOrderFeed += other.packetsOutOfOrderFeed is.frames += other.frames - is.packetsNotFound += other.packetsNotFound + is.packetsNotFoundMetadata += other.packetsNotFoundMetadata } func (is *intervalStats) MarshalLogObject(e zapcore.ObjectEncoder) error { @@ -92,10 +92,31 @@ func (is *intervalStats) MarshalLogObject(e zapcore.ObjectEncoder) error { e.AddUint64("packetsPadding", is.packetsPadding) e.AddUint64("bytesPadding", is.bytesPadding) e.AddUint64("headerBytesPadding", is.headerBytesPadding) - e.AddUint64("packetsLost", is.packetsLost) - e.AddUint64("packetsOutOfOrder", is.packetsOutOfOrder) + e.AddUint64("packetsLostFeed", is.packetsLostFeed) + e.AddUint64("packetsOutOfOrderFeed", is.packetsOutOfOrderFeed) e.AddUint32("frames", is.frames) - e.AddUint64("packetsNotFound", is.packetsNotFound) + e.AddUint64("packetsNotFoundMetadata", is.packetsNotFoundMetadata) + + return nil +} + +// ------------------------------------------------------------------- + +type wrappedReceptionReportsLogger struct { + *senderSnapshot + useSkipped bool +} + +func (w wrappedReceptionReportsLogger) MarshalLogObject(e zapcore.ObjectEncoder) error { + if w.useSkipped { + for i, rr := range w.senderSnapshot.skippedReceptionReports { + e.AddReflected(fmt.Sprintf("%d", i), rr) + } + } else { + for i, rr := range w.senderSnapshot.processedReceptionReports { + e.AddReflected(fmt.Sprintf("%d", i), rr) + } + } return nil } @@ -119,10 +140,10 @@ type senderSnapshot struct { bytesDuplicate uint64 headerBytesDuplicate uint64 - packetsOutOfOrder uint64 + packetsOutOfOrderFeed uint64 - packetsLostFeed uint64 - packetsLost uint64 + packetsLostFeed uint64 + packetsLostFromRR uint64 frames uint32 @@ -134,8 +155,10 @@ type senderSnapshot struct { maxJitterFeed float64 maxJitter float64 - extLastRRSN uint64 - intervalStats intervalStats + extLastRRSN uint64 + intervalStats intervalStats + processedReceptionReports []rtcp.ReceptionReport + skippedReceptionReports []rtcp.ReceptionReport } func (s *senderSnapshot) MarshalLogObject(e zapcore.ObjectEncoder) error { @@ -154,11 +177,20 @@ func (s *senderSnapshot) MarshalLogObject(e zapcore.ObjectEncoder) error { e.AddUint64("packetsDuplicate", s.packetsDuplicate) e.AddUint64("bytesDuplicate", s.bytesDuplicate) e.AddUint64("headerBytesDuplicate", s.headerBytesDuplicate) - e.AddUint64("packetsOutOfOrder", s.packetsOutOfOrder) + e.AddUint64("packetsOutOfOrderFeed", s.packetsOutOfOrderFeed) e.AddUint64("packetsLostFeed", s.packetsLostFeed) - e.AddUint64("packetsLost", s.packetsLost) + e.AddUint64("packetsLostFromRR", s.packetsLostFromRR) e.AddUint32("frames", s.frames) e.AddUint32("nacks", s.nacks) + e.AddUint32("plis", s.plis) + e.AddUint32("firs", s.firs) + e.AddUint32("maxRtt", s.maxRtt) + e.AddFloat64("maxJitterFeed", s.maxJitterFeed) + e.AddFloat64("maxJitter", s.maxJitter) + e.AddUint64("extLastRRSN", s.extLastRRSN) + e.AddObject("intervalStats", &s.intervalStats) + e.AddObject("processedReceptionReports", wrappedReceptionReportsLogger{s, false}) + e.AddObject("skippedReceptionReports", wrappedReceptionReportsLogger{s, true}) return nil } @@ -538,12 +570,10 @@ func (r *RTPStatsSender) UpdateFromReceiverReport(rr rtcp.ReceptionReport) (rtt } } + r.packetsLostFromRR = uint64(rr.TotalLost) 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 lossDelta < (1<<23) && rr.TotalLost < r.lastRR.TotalLost { + r.packetsLostFromRR += (1 << 24) } if isRttChanged { @@ -591,6 +621,7 @@ func (r *RTPStatsSender) UpdateFromReceiverReport(rr rtcp.ReceptionReport) (rtt "rtpStats", lockedRTPStatsSenderLogEncoder{r}, ) s.extLastRRSN = extReceivedRRSN + s.skippedReceptionReports = append(s.skippedReceptionReports, rr) continue } @@ -598,7 +629,7 @@ func (r *RTPStatsSender) UpdateFromReceiverReport(rr rtcp.ReceptionReport) (rtt is := r.getIntervalStats(s.extLastRRSN+1, extReceivedRRSN+1, r.extHighestSN) eis := &s.intervalStats eis.aggregate(&is) - if is.packetsNotFound != 0 { + if is.packetsNotFoundMetadata != 0 { timeSinceLastRR := time.Since(r.lastRRTime) if r.lastRRTime.IsZero() { timeSinceLastRR = time.Since(r.startTime) @@ -619,6 +650,7 @@ func (r *RTPStatsSender) UpdateFromReceiverReport(rr rtcp.ReceptionReport) (rtt } } s.extLastRRSN = extReceivedRRSN + s.processedReceptionReports = append(s.processedReceptionReports, rr) } r.lastRRTime = time.Now() @@ -810,7 +842,7 @@ func (r *RTPStatsSender) DeltaInfoSender(senderSnapshotID uint32) *RTPDeltaInfo return nil } - packetsLost := uint32(now.packetsLost - then.packetsLost) + packetsLost := uint32(now.packetsLostFromRR - then.packetsLostFromRR) if int32(packetsLost) < 0 { packetsLost = 0 } @@ -854,7 +886,7 @@ func (r *RTPStatsSender) DeltaInfoSender(senderSnapshotID uint32) *RTPDeltaInfo HeaderBytesPadding: now.headerBytesPadding - then.headerBytesPadding, PacketsLost: packetsLost, PacketsMissing: packetsLostFeed, - PacketsOutOfOrder: uint32(now.packetsOutOfOrder - then.packetsOutOfOrder), + PacketsOutOfOrder: uint32(now.packetsOutOfOrderFeed - then.packetsOutOfOrderFeed), Frames: now.frames - then.frames, RttMax: then.maxRtt, JitterMax: maxJitterTime, @@ -920,28 +952,28 @@ func (r *RTPStatsSender) getSenderSnapshot(startTime time.Time, s *senderSnapsho } return senderSnapshot{ - isValid: true, - startTime: startTime, - extStartSN: s.extLastRRSN + 1, - bytes: s.bytes + s.intervalStats.bytes, - headerBytes: s.headerBytes + s.intervalStats.headerBytes, - packetsPadding: s.packetsPadding + s.intervalStats.packetsPadding, - bytesPadding: s.bytesPadding + s.intervalStats.bytesPadding, - headerBytesPadding: s.headerBytesPadding + s.intervalStats.headerBytesPadding, - packetsDuplicate: r.packetsDuplicate, - bytesDuplicate: r.bytesDuplicate, - headerBytesDuplicate: r.headerBytesDuplicate, - packetsOutOfOrder: s.packetsOutOfOrder + s.intervalStats.packetsOutOfOrder, - packetsLostFeed: s.packetsLost + s.intervalStats.packetsLost, - packetsLost: r.packetsLostFromRR, - frames: s.frames + s.intervalStats.frames, - nacks: r.nacks, - plis: r.plis, - firs: r.firs, - maxRtt: r.rtt, - maxJitterFeed: r.jitter, - maxJitter: r.jitterFromRR, - extLastRRSN: s.extLastRRSN, + isValid: true, + startTime: startTime, + extStartSN: s.extLastRRSN + 1, + bytes: s.bytes + s.intervalStats.bytes, + headerBytes: s.headerBytes + s.intervalStats.headerBytes, + packetsPadding: s.packetsPadding + s.intervalStats.packetsPadding, + bytesPadding: s.bytesPadding + s.intervalStats.bytesPadding, + headerBytesPadding: s.headerBytesPadding + s.intervalStats.headerBytesPadding, + packetsDuplicate: r.packetsDuplicate, + bytesDuplicate: r.bytesDuplicate, + headerBytesDuplicate: r.headerBytesDuplicate, + packetsOutOfOrderFeed: s.packetsOutOfOrderFeed + s.intervalStats.packetsOutOfOrderFeed, + packetsLostFeed: s.packetsLostFeed + s.intervalStats.packetsLostFeed, + packetsLostFromRR: r.packetsLostFromRR, + frames: s.frames + s.intervalStats.frames, + nacks: r.nacks, + plis: r.plis, + firs: r.firs, + maxRtt: r.rtt, + maxJitterFeed: r.jitter, + maxJitter: r.jitterFromRR, + extLastRRSN: s.extLastRRSN, } } @@ -1011,14 +1043,14 @@ func (r *RTPStatsSender) getIntervalStats( processESN := func(esn uint64, ehsn uint64) { slot := r.getSnInfoOutOfOrderSlot(esn, ehsn) if slot < 0 { - intervalStats.packetsNotFound++ + intervalStats.packetsNotFoundMetadata++ return } snInfo := &r.snInfos[slot] switch { case snInfo.pktSize == 0: - intervalStats.packetsLost++ + intervalStats.packetsLostFeed++ case snInfo.flags&snInfoFlagPadding != 0: intervalStats.packetsPadding++ @@ -1030,7 +1062,7 @@ func (r *RTPStatsSender) getIntervalStats( intervalStats.bytes += uint64(snInfo.pktSize) intervalStats.headerBytes += uint64(snInfo.hdrSize) if (snInfo.flags & snInfoFlagOutOfOrder) != 0 { - intervalStats.packetsOutOfOrder++ + intervalStats.packetsOutOfOrderFeed++ } }