More optimisation in RTPStats module. (#4298)

Was hoping for more hidden bits to optimise, but oh well.
This commit is contained in:
Raja Subramanian authored and GitHub committed 2026-02-06 21:33:41 +05:30
1 parent cefd5da940
commit bb744916fd
3 files changed
+226 -106

No files matched your search

+26 -24
View File
@@ -428,32 +428,13 @@ func (r *rtpStatsBase) maybeAdjustFirstPacketTime(
firstTime := now - samplesDuration.Nanoseconds()
adjustment = r.firstTime - firstTime
getFields := func() []any {
return []any{
"startTime", time.Unix(0, r.startTime),
"nowTime", time.Unix(0, now),
"before", time.Unix(0, r.firstTime),
"after", time.Unix(0, firstTime),
"adjustment", time.Duration(adjustment),
"firstTimeAdjustment", r.firstTimeAdjustment,
"extNowTS", extNowTS,
"extStartTS", extStartTS,
"srData", WrappedRTCPSenderReportStateLogger{srData},
"tsOffset", tsOffset,
"timeSinceReceive", timeSinceReceive,
"timeSinceFirst", timeSinceFirst,
"samplesDiff", samplesDiff,
"samplesDuration", samplesDuration,
}
}
if firstTime < r.firstTime {
if adjustment > cFirstPacketTimeAdjustThreshold {
err = errors.New("adjusting first packet time, too big, ignoring")
loggingFields = getFields()
loggingFields = r.firstPacketTimeAdjustLogFields(now, firstTime, adjustment, extNowTS, extStartTS, srData, tsOffset, timeSinceReceive, timeSinceFirst, samplesDuration, samplesDiff)
} else {
r.firstTimeAdjustment += time.Duration(adjustment)
r.logger.Debugw("adjusting first packet time", getFields()...)
r.logger.Debugw("adjusting first packet time", r.firstPacketTimeAdjustLogFields(now, firstTime, adjustment, extNowTS, extStartTS, srData, tsOffset, timeSinceReceive, timeSinceFirst, samplesDuration, samplesDiff)...)
r.firstTime = firstTime
}
}
@@ -461,6 +442,25 @@ func (r *rtpStatsBase) maybeAdjustFirstPacketTime(
return
}
func (r *rtpStatsBase) firstPacketTimeAdjustLogFields(now, firstTime int64, adjustment int64, extNowTS, extStartTS uint64, srData *livekit.RTCPSenderReportState, tsOffset uint64, timeSinceReceive, timeSinceFirst, samplesDuration time.Duration, samplesDiff int64) []any {
return []any{
"startTime", time.Unix(0, r.startTime),
"nowTime", time.Unix(0, now),
"before", time.Unix(0, r.firstTime),
"after", time.Unix(0, firstTime),
"adjustment", time.Duration(adjustment),
"firstTimeAdjustment", r.firstTimeAdjustment,
"extNowTS", extNowTS,
"extStartTS", extStartTS,
"srData", WrappedRTCPSenderReportStateLogger{srData},
"tsOffset", tsOffset,
"timeSinceReceive", timeSinceReceive,
"timeSinceFirst", timeSinceFirst,
"samplesDiff", samplesDiff,
"samplesDuration", samplesDuration,
}
}
func (r *rtpStatsBase) getPacketsSeenMinusPadding(extStartSN, extHighestSN uint64) uint64 {
packetsSeen := r.getPacketsSeen(extStartSN, extHighestSN)
if r.packetsPadding > packetsSeen {
@@ -742,12 +742,14 @@ func (r *rtpStatsBase) getDrift(extStartTS, extHighestTS uint64) (
if r.srFirst != nil && r.srNewest != nil && r.srFirst.RtpTimestamp != r.srNewest.RtpTimestamp {
rtpClockTicks := r.srNewest.RtpTimestampExt - r.srFirst.RtpTimestampExt
elapsed := mediatransportutil.NtpTime(r.srNewest.NtpTimestamp).Time().Sub(mediatransportutil.NtpTime(r.srFirst.NtpTimestamp).Time())
srFirstTime := mediatransportutil.NtpTime(r.srFirst.NtpTimestamp).Time()
srNewestTime := mediatransportutil.NtpTime(r.srNewest.NtpTimestamp).Time()
elapsed := srNewestTime.Sub(srFirstTime)
if elapsed.Seconds() > 0.0 {
driftSamples := int64(rtpClockTicks - r.rtpConverter.ToRTPExt(elapsed))
ntpReportDrift = &livekit.RTPDrift{
StartTime: timestamppb.New(mediatransportutil.NtpTime(r.srFirst.NtpTimestamp).Time()),
EndTime: timestamppb.New(mediatransportutil.NtpTime(r.srNewest.NtpTimestamp).Time()),
StartTime: timestamppb.New(srFirstTime),
EndTime: timestamppb.New(srNewestTime),
Duration: elapsed.Seconds(),
StartTimestamp: r.srFirst.RtpTimestampExt,
EndTimestamp: r.srNewest.RtpTimestampExt,
+10 -27
View File
@@ -17,6 +17,7 @@ package rtpstats
import (
"errors"
"fmt"
"strings"
"sync"
"time"
@@ -412,25 +413,18 @@ func (r *rtpStatsBaseLite) marshalLogObject(e zapcore.ObjectEncoder, packetsExpe
e.AddFloat32("packetLostPercentage", float32(r.packetsLost)/float32(packetsExpected)*100.0)
}
hasLoss := false
first := true
str := "["
var sb strings.Builder
for burst, count := range r.gapHistogram {
if count == 0 {
continue
}
hasLoss = true
if !first {
str += ", "
if sb.Len() > 0 {
sb.WriteString(", ")
}
first = false
str += fmt.Sprintf("%d:%d", burst+1, count)
sb.WriteString(fmt.Sprintf("%d:%d", burst+1, count))
}
str += "]"
if hasLoss {
e.AddString("gapHistogram", str)
if sb.Len() > 0 {
e.AddString("gapHistogram", "["+sb.String()+"]")
}
e.AddUint32("nacks", r.nacks)
@@ -486,25 +480,14 @@ func (r *rtpStatsBaseLite) toProto(packetsExpected, packetsSeenMinusPadding, pac
LastPli: timestamppb.New(time.Unix(0, r.lastPli)),
}
gapsPresent := false
for i := range len(r.gapHistogram) {
if r.gapHistogram[i] == 0 {
continue
}
gapsPresent = true
break
}
if gapsPresent {
p.GapHistogram = make(map[int32]uint32, len(r.gapHistogram))
for i := range len(r.gapHistogram) {
if r.gapHistogram[i] == 0 {
continue
}
p.GapHistogram[int32(i+1)] = r.gapHistogram[i]
if p.GapHistogram == nil {
p.GapHistogram = make(map[int32]uint32, len(r.gapHistogram))
}
p.GapHistogram[int32(i+1)] = r.gapHistogram[i]
}
return p
+190 -55
View File
@@ -21,12 +21,12 @@ import (
"time"
"github.com/pion/rtcp"
"go.uber.org/zap"
"go.uber.org/zap/zapcore"
"google.golang.org/protobuf/types/known/timestamppb"
"github.com/livekit/mediatransportutil"
"github.com/livekit/protocol/livekit"
"github.com/livekit/protocol/logger"
"github.com/livekit/protocol/utils/mono"
)
@@ -40,6 +40,9 @@ const (
snInfoFlagOutOfOrder
)
// max number of receiver reports kept per sender snapshot for logging
const cMaxProcessedReceptionReports = 64
type snInfo struct {
pktSize uint16
hdrSize uint8
@@ -272,6 +275,70 @@ func (s *senderSnapshot) maybeUpdateMaxJitter(jitter float64) {
// -------------------------------------------------------------------
type senderUpdateLoggingFields struct {
packetTime int64
extSequenceNumber uint64
extTimestamp uint64
marker bool
hdrSize int
payloadSize int
paddingSize int
gapSN int64
gapTS int64
timeSinceHighest int64
rtpStats *RTPStatsSender
}
func (sulf *senderUpdateLoggingFields) MarshalLogObject(e zapcore.ObjectEncoder) error {
if sulf != nil {
e.AddUint64("currSN", sulf.extSequenceNumber)
e.AddInt64("gapSN", sulf.gapSN)
e.AddUint64("currTS", sulf.extTimestamp)
e.AddInt64("gapTS", sulf.gapTS)
e.AddTime("packetTime", time.Unix(0, sulf.packetTime))
e.AddDuration("timeSinceHighest", time.Duration(sulf.timeSinceHighest))
e.AddBool("marker", sulf.marker)
e.AddInt("hdrSize", sulf.hdrSize)
e.AddInt("payloadSize", sulf.payloadSize)
e.AddInt("paddingSize", sulf.paddingSize)
e.AddObject("rtpStats", lockedRTPStatsSenderLogEncoder{sulf.rtpStats})
}
return nil
}
// -------------------------------------------------------------------
type senderReportLoggingFields struct {
srData *livekit.RTCPSenderReportState
publisherSRData *livekit.RTCPSenderReportState
tsOffset uint64
reportTime int64
reportTimeAdjusted int64
nowNano int64
nowRTPExt uint64
rtpStats *RTPStatsSender
}
func (srlf *senderReportLoggingFields) MarshalLogObject(e zapcore.ObjectEncoder) error {
if srlf != nil {
e.AddObject("curr", WrappedRTCPSenderReportStateLogger{srlf.srData})
e.AddObject("feed", WrappedRTCPSenderReportStateLogger{srlf.publisherSRData})
e.AddUint64("tsOffset", srlf.tsOffset)
e.AddTime("timeNow", time.Now())
e.AddInt64("reportTime", srlf.reportTime)
e.AddInt64("reportTimeAdjusted", srlf.reportTimeAdjusted)
e.AddDuration("timeSinceHighest", time.Duration(srlf.nowNano-srlf.rtpStats.highestTime))
e.AddDuration("timeSinceFirst", time.Duration(srlf.nowNano-srlf.rtpStats.firstTime))
e.AddDuration("timeSincePublisherSRAdjusted", time.Duration(srlf.nowNano-srlf.publisherSRData.AtAdjusted))
e.AddDuration("timeSincePublisherSR", time.Duration(srlf.nowNano-srlf.publisherSRData.At))
e.AddUint64("nowRTPExt", srlf.nowRTPExt)
e.AddObject("rtpStats", lockedRTPStatsSenderLogEncoder{srlf.rtpStats})
}
return nil
}
// -------------------------------------------------------------------
type rttMarker struct {
ntpTime mediatransportutil.NtpTime
sentAt time.Time
@@ -444,21 +511,6 @@ func (r *RTPStatsSender) Update(
pktSize := uint64(hdrSize + payloadSize + paddingSize)
isDuplicate := false
gapSN := int64(extSequenceNumber - r.extHighestSN)
ulgr := func() logger.UnlikelyLogger {
return r.logger.WithUnlikelyValues(
"currSN", extSequenceNumber,
"gapSN", gapSN,
"currTS", extTimestamp,
"gapTS", int64(extTimestamp-r.extHighestTS),
"packetTime", time.Unix(0, packetTime),
"timeSinceHighest", time.Duration(packetTime-r.highestTime),
"marker", marker,
"hdrSize", hdrSize,
"payloadSize", payloadSize,
"paddingSize", paddingSize,
"rtpStats", lockedRTPStatsSenderLogEncoder{r},
)
}
if gapSN <= 0 { // duplicate OR out-of-order
if payloadSize == 0 && extSequenceNumber < r.extStartSN && !r.params.IsRTX {
// do not start on a padding only packet
@@ -479,8 +531,22 @@ func (r *RTPStatsSender) Update(
r.senderSnapshots[i].maybeReinit(r.extStartSN, extSequenceNumber)
}
ulgr().Infow(
sulf := &senderUpdateLoggingFields{
packetTime: packetTime,
extSequenceNumber: extSequenceNumber,
extTimestamp: extTimestamp,
marker: marker,
hdrSize: hdrSize,
payloadSize: payloadSize,
paddingSize: paddingSize,
gapSN: gapSN,
gapTS: int64(extTimestamp - r.extHighestTS),
timeSinceHighest: packetTime - r.highestTime,
rtpStats: r,
}
r.logger.Infow(
"adjusting start sequence number",
zap.Inline(sulf),
"snAfter", extSequenceNumber,
"tsAfter", extTimestamp,
)
@@ -504,8 +570,22 @@ func (r *RTPStatsSender) Update(
if !isDuplicate && -gapSN >= cSequenceNumberLargeJumpThreshold {
r.largeJumpNegativeCount++
if (r.largeJumpNegativeCount-1)%100 == 0 {
ulgr().Warnw(
sulf := &senderUpdateLoggingFields{
packetTime: packetTime,
extSequenceNumber: extSequenceNumber,
extTimestamp: extTimestamp,
marker: marker,
hdrSize: hdrSize,
payloadSize: payloadSize,
paddingSize: paddingSize,
gapSN: gapSN,
gapTS: int64(extTimestamp - r.extHighestTS),
timeSinceHighest: packetTime - r.highestTime,
rtpStats: r,
}
r.logger.Warnw(
"large sequence number gap negative", nil,
zap.Inline(sulf),
"count", r.largeJumpNegativeCount,
)
}
@@ -514,8 +594,22 @@ func (r *RTPStatsSender) Update(
if gapSN >= cSequenceNumberLargeJumpThreshold {
r.largeJumpCount++
if (r.largeJumpCount-1)%100 == 0 {
ulgr().Warnw(
sulf := &senderUpdateLoggingFields{
packetTime: packetTime,
extSequenceNumber: extSequenceNumber,
extTimestamp: extTimestamp,
marker: marker,
hdrSize: hdrSize,
payloadSize: payloadSize,
paddingSize: paddingSize,
gapSN: gapSN,
gapTS: int64(extTimestamp - r.extHighestTS),
timeSinceHighest: packetTime - r.highestTime,
rtpStats: r,
}
r.logger.Warnw(
"large sequence number gap", nil,
zap.Inline(sulf),
"count", r.largeJumpCount,
)
}
@@ -524,8 +618,22 @@ func (r *RTPStatsSender) Update(
if extTimestamp < r.extHighestTS {
r.timeReversedCount++
if (r.timeReversedCount-1)%100 == 0 {
ulgr().Warnw(
sulf := &senderUpdateLoggingFields{
packetTime: packetTime,
extSequenceNumber: extSequenceNumber,
extTimestamp: extTimestamp,
marker: marker,
hdrSize: hdrSize,
payloadSize: payloadSize,
paddingSize: paddingSize,
gapSN: gapSN,
gapTS: int64(extTimestamp - r.extHighestTS),
timeSinceHighest: packetTime - r.highestTime,
rtpStats: r,
}
r.logger.Warnw(
"time reversed", nil,
zap.Inline(sulf),
"count", r.timeReversedCount,
)
}
@@ -544,8 +652,22 @@ func (r *RTPStatsSender) Update(
}
if extTimestamp < r.extStartTS {
ulgr().Infow(
sulf := &senderUpdateLoggingFields{
packetTime: packetTime,
extSequenceNumber: extSequenceNumber,
extTimestamp: extTimestamp,
marker: marker,
hdrSize: hdrSize,
payloadSize: payloadSize,
paddingSize: paddingSize,
gapSN: gapSN,
gapTS: int64(extTimestamp - r.extHighestTS),
timeSinceHighest: packetTime - r.highestTime,
rtpStats: r,
}
r.logger.Infow(
"adjusting start timestamp",
zap.Inline(sulf),
"snAfter", extSequenceNumber,
"tsAfter", extTimestamp,
)
@@ -628,18 +750,16 @@ func (r *RTPStatsSender) UpdateFromReceiverReport(rr rtcp.ReceptionReport) (rtt
r.lastRR = rr
}()
timeSinceLastRR := func() time.Duration {
if r.lastRRTime != 0 {
return time.Duration(nowNano - r.lastRRTime)
}
return time.Duration(nowNano - r.startTime)
timeSinceLastRR := time.Duration(nowNano - r.startTime)
if r.lastRRTime != 0 {
timeSinceLastRR = time.Duration(nowNano - r.lastRRTime)
}
extReceivedRRSN := extHighestSNFromRR + (r.extStartSN & 0xFFFF_FFFF_FFFF_0000)
if int64(r.extHighestSN-extReceivedRRSN) < 0 || int64(r.extHighestSN-extReceivedRRSN) > (1<<32) {
r.logger.Infow(
"receiver report runaway, dropping",
"timeSinceLastRR", timeSinceLastRR(),
"timeSinceLastRR", timeSinceLastRR,
"receivedRR", rr,
"extHighestSNFromRR", extHighestSNFromRR,
"extReceivedRRSN", extReceivedRRSN,
@@ -666,7 +786,7 @@ func (r *RTPStatsSender) UpdateFromReceiverReport(rr rtcp.ReceptionReport) (rtt
// RTCP RR does not change.
r.logger.Infow(
"receiver report missed rollover, adjusting",
"timeSinceLastRR", timeSinceLastRR(),
"timeSinceLastRR", timeSinceLastRR,
"receivedRR", rr,
"extHighestSNFromRR", extHighestSNFromRR,
"extReceivedRRSN", extReceivedRRSN,
@@ -679,7 +799,7 @@ func (r *RTPStatsSender) UpdateFromReceiverReport(rr rtcp.ReceptionReport) (rtt
}
r.logger.Infow(
"receiver report missed rollover, adjusted",
"timeSinceLastRR", timeSinceLastRR(),
"timeSinceLastRR", timeSinceLastRR,
"receivedRR", rr,
"extHighestSNFromRR", extHighestSNFromRR,
"extReceivedRRSN", extReceivedRRSN,
@@ -691,7 +811,7 @@ func (r *RTPStatsSender) UpdateFromReceiverReport(rr rtcp.ReceptionReport) (rtt
// if remote adjusts somehow, roll back alignment
r.logger.Infow(
"receiver report caught up rollover, adjusting",
"timeSinceLastRR", timeSinceLastRR(),
"timeSinceLastRR", timeSinceLastRR,
"receivedRR", rr,
"extHighestSNFromRR", extHighestSNFromRR,
"extReceivedRRSN", extReceivedRRSN,
@@ -704,7 +824,7 @@ func (r *RTPStatsSender) UpdateFromReceiverReport(rr rtcp.ReceptionReport) (rtt
}
r.logger.Infow(
"receiver report caught up rollover, adjusted",
"timeSinceLastRR", timeSinceLastRR(),
"timeSinceLastRR", timeSinceLastRR,
"receivedRR", rr,
"extHighestSNFromRR", extHighestSNFromRR,
"extReceivedRRSN", extReceivedRRSN,
@@ -715,7 +835,7 @@ func (r *RTPStatsSender) UpdateFromReceiverReport(rr rtcp.ReceptionReport) (rtt
if r.extHighestSNFromRR > extHighestSNFromRR {
r.logger.Debugw(
"receiver report out-of-order, dropping",
"timeSinceLastRR", timeSinceLastRR(),
"timeSinceLastRR", timeSinceLastRR,
"receivedRR", rr,
"extHighestSNFromRR", extHighestSNFromRR,
"extReceivedRRSN", extReceivedRRSN,
@@ -780,7 +900,7 @@ func (r *RTPStatsSender) UpdateFromReceiverReport(rr rtcp.ReceptionReport) (rtt
"metadata cache overflow",
"senderSnapshotID", i+cFirstSnapshotID,
"senderSnapshot", s,
"timeSinceLastRR", timeSinceLastRR(),
"timeSinceLastRR", timeSinceLastRR,
"receivedRR", rr,
"extReceivedRRSN", extReceivedRRSN,
"packetsInInterval", extReceivedRRSN-s.receiverView.extLastRRSN,
@@ -791,7 +911,13 @@ func (r *RTPStatsSender) UpdateFromReceiverReport(rr rtcp.ReceptionReport) (rtt
}
}
s.receiverView.extLastRRSN = extReceivedRRSN
s.receiverView.processedReceptionReports = append(s.receiverView.processedReceptionReports, rr)
reports := &s.receiverView.processedReceptionReports
if len(*reports) >= cMaxProcessedReceptionReports {
copy(*reports, (*reports)[1:])
(*reports)[len(*reports)-1] = rr
} else {
*reports = append(*reports, rr)
}
}
return
@@ -874,32 +1000,28 @@ func (r *RTPStatsSender) GetRtcpSenderReport(ssrc uint32, publisherSRData *livek
Octets: octetCount,
}
ulgr := func() logger.UnlikelyLogger {
return r.logger.WithUnlikelyValues(
"curr", WrappedRTCPSenderReportStateLogger{srData},
"feed", WrappedRTCPSenderReportStateLogger{publisherSRData},
"tsOffset", tsOffset,
"timeNow", mono.Now(),
"reportTime", time.Unix(0, reportTime),
"reportTimeAdjusted", time.Unix(0, reportTimeAdjusted),
"timeSinceHighest", time.Duration(nowNano-r.highestTime),
"timeSinceFirst", time.Duration(nowNano-r.firstTime),
"timeSincePublisherSRAdjusted", time.Duration(nowNano-publisherSRData.AtAdjusted),
"timeSincePublisherSR", time.Duration(nowNano-publisherSRData.At),
"nowRTPExt", nowRTPExt,
"rtpStats", lockedRTPStatsSenderLogEncoder{r},
)
}
if r.srNewest != nil && nowRTPExt >= r.srNewest.RtpTimestampExt && r.clockRate != 0 {
timeSinceLastReport := nowNTP.Time().Sub(mediatransportutil.NtpTime(r.srNewest.NtpTimestamp).Time())
nowTime := nowNTP.Time()
srNewestTime := mediatransportutil.NtpTime(r.srNewest.NtpTimestamp).Time()
timeSinceLastReport := nowTime.Sub(srNewestTime)
rtpDiffSinceLastReport := nowRTPExt - r.srNewest.RtpTimestampExt
windowClockRate := float64(rtpDiffSinceLastReport) / timeSinceLastReport.Seconds()
if timeSinceLastReport.Seconds() > 0.2 && math.Abs(float64(r.clockRate)-windowClockRate) > 0.2*float64(r.clockRate) {
r.clockSkewCount++
if (r.clockSkewCount-1)%100 == 0 {
ulgr().Infow(
srlf := &senderReportLoggingFields{
srData: srData,
publisherSRData: publisherSRData,
tsOffset: tsOffset,
reportTime: reportTime,
reportTimeAdjusted: reportTimeAdjusted,
nowNano: nowNano,
nowRTPExt: nowRTPExt,
rtpStats: r,
}
r.logger.Infow(
"sending sender report, clock skew",
zap.Inline(srlf),
"timeSinceLastReport", timeSinceLastReport,
"rtpDiffSinceLastReport", rtpDiffSinceLastReport,
"windowClockRate", windowClockRate,
@@ -912,7 +1034,20 @@ func (r *RTPStatsSender) GetRtcpSenderReport(ssrc uint32, publisherSRData *livek
if r.srNewest != nil && nowRTPExt < r.srNewest.RtpTimestampExt {
// If report being generated is behind the last report, skip it.
// Should not happen.
ulgr().Infow("sending sender report, out-of-order, skipping")
srlf := &senderReportLoggingFields{
srData: srData,
publisherSRData: publisherSRData,
tsOffset: tsOffset,
reportTime: reportTime,
reportTimeAdjusted: reportTimeAdjusted,
nowNano: nowNano,
nowRTPExt: nowRTPExt,
rtpStats: r,
}
r.logger.Infow(
"sending sender report, out-of-order, skipping",
zap.Inline(srlf),
)
return nil
}
@@ -923,7 +1058,7 @@ func (r *RTPStatsSender) GetRtcpSenderReport(ssrc uint32, publisherSRData *livek
r.rttMarker = rttMarker{
ntpTime: nowNTP,
sentAt: mono.Now(),
sentAt: time.Now(),
}
return &rtcp.SenderReport{