dandri/livekit
1
0
Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-07-31 18:09:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

Sender side snap shot clean up and logging. (#3196)

Browse Source 
* 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
This commit is contained in:
Raja Subramanian
2024-11-23 10:40:10 +05:30
committed by GitHub
parent 29c7906250
commit d07d84f99f
1 changed files with 88 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/sfu/rtpstats/rtpstats_sender.go
+88 -56
View File
@@ -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++
}
}