diff --git a/pkg/sfu/buffer/rtpstats.go b/pkg/sfu/buffer/rtpstats.go index 5c9efaf01..9a0ba637a 100644 --- a/pkg/sfu/buffer/rtpstats.go +++ b/pkg/sfu/buffer/rtpstats.go @@ -31,8 +31,6 @@ type RTPFlowState struct { } type IntervalStats struct { - earliestPktTime int64 - latestPktTime int64 packets uint32 bytes uint64 headerBytes uint64 @@ -369,7 +367,7 @@ func (r *RTPStats) Update(rtph *rtp.Header, payloadSize int, paddingSize int, pa isDuplicate = true } else { r.packetsLost-- - r.setSnInfo(rtph.SequenceNumber, packetTime, uint16(pktSize), uint16(hdrSize), uint16(payloadSize), rtph.Marker) + r.setSnInfo(rtph.SequenceNumber, uint16(pktSize), uint16(hdrSize), uint16(payloadSize), rtph.Marker) } } @@ -388,7 +386,7 @@ func (r *RTPStats) Update(rtph *rtp.Header, payloadSize int, paddingSize int, pa r.clearSnInfos(r.highestSN+1, rtph.SequenceNumber) r.packetsLost += uint32(diff - 1) - r.setSnInfo(rtph.SequenceNumber, packetTime, uint16(pktSize), uint16(hdrSize), uint16(payloadSize), rtph.Marker) + r.setSnInfo(rtph.SequenceNumber, uint16(pktSize), uint16(hdrSize), uint16(payloadSize), rtph.Marker) if rtph.SequenceNumber < r.highestSN && !first { r.cycles++ @@ -438,7 +436,7 @@ func (r *RTPStats) maybeAdjustStartSN(rtph *rtp.Header, packetTime int64, pktSiz beforeAdjust := r.extStartSN r.extStartSN = uint32(rtph.SequenceNumber) - r.setSnInfo(rtph.SequenceNumber, packetTime, uint16(pktSize), uint16(hdrSize), uint16(payloadSize), rtph.Marker) + r.setSnInfo(rtph.SequenceNumber, uint16(pktSize), uint16(hdrSize), uint16(payloadSize), rtph.Marker) for _, s := range r.snapshots { if s.extStartSN == beforeAdjust { @@ -886,9 +884,6 @@ func (r *RTPStats) DeltaInfo(snapshotId uint32) *RTPDeltaInfo { packetsMissing := uint32(0) intervalStats := r.getIntervalStats(uint16(then.extStartSN), uint16(now.extStartSN)) if r.params.IsReceiverReportDriven { - startTime = time.Unix(0, intervalStats.earliestPktTime) - endTime = time.Unix(0, intervalStats.latestPktTime) - packetsMissing = intervalStats.packetsLost packetsLost = now.packetsLostOverridden - then.packetsLostOverridden @@ -1157,7 +1152,7 @@ func (r *RTPStats) getSnInfoOutOfOrderPtr(sn uint16) int { return (r.snInfoWritePtr - int(offset) - 1) & SnInfoMask } -func (r *RTPStats) setSnInfo(sn uint16, pktTime int64, pktSize uint16, hdrSize uint16, payloadSize uint16, marker bool) { +func (r *RTPStats) setSnInfo(sn uint16, pktSize uint16, hdrSize uint16, payloadSize uint16, marker bool) { writePtr := 0 ooo := (sn - r.highestSN) > (1 << 15) if !ooo { @@ -1171,7 +1166,6 @@ func (r *RTPStats) setSnInfo(sn uint16, pktTime int64, pktSize uint16, hdrSize u } snInfo := &r.snInfos[writePtr] - snInfo.pktTime = pktTime snInfo.pktSize = pktSize snInfo.hdrSize = hdrSize snInfo.isPaddingOnly = payloadSize == 0 @@ -1181,7 +1175,6 @@ func (r *RTPStats) setSnInfo(sn uint16, pktTime int64, pktSize uint16, hdrSize u func (r *RTPStats) clearSnInfos(startInclusive uint16, endExclusive uint16) { for sn := startInclusive; sn != endExclusive; sn++ { snInfo := &r.snInfos[r.snInfoWritePtr] - snInfo.pktTime = 0 snInfo.pktSize = 0 snInfo.hdrSize = 0 snInfo.isPaddingOnly = false @@ -1229,14 +1222,6 @@ func (r *RTPStats) getIntervalStats(startInclusive uint16, endExclusive uint16) if snInfo.marker { intervalStats.frames++ } - - if intervalStats.earliestPktTime == 0 || snInfo.pktTime < intervalStats.earliestPktTime { - intervalStats.earliestPktTime = snInfo.pktTime - } - - if intervalStats.latestPktTime == 0 || snInfo.pktTime > intervalStats.latestPktTime { - intervalStats.latestPktTime = snInfo.pktTime - } } if startInclusive == endExclusive { diff --git a/pkg/sfu/connectionquality/connectionstats.go b/pkg/sfu/connectionquality/connectionstats.go index 3195b7a18..76841b2da 100644 --- a/pkg/sfu/connectionquality/connectionstats.go +++ b/pkg/sfu/connectionquality/connectionstats.go @@ -2,9 +2,11 @@ package connectionquality import ( "strings" + "sync" "time" "github.com/frostbyte73/core" + "go.uber.org/atomic" "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/logger" @@ -14,8 +16,8 @@ import ( ) const ( - UpdateInterval = 5 * time.Second - audioPacketRateThreshold = float64(25.0) + UpdateInterval = 5 * time.Second + processThreshold = 0.95 ) type ConnectionStatsParams struct { @@ -27,11 +29,15 @@ type ConnectionStatsParams struct { } type ConnectionStats struct { - params ConnectionStatsParams - trackInfo *livekit.TrackInfo + params ConnectionStatsParams + isVideo atomic.Bool onStatsUpdate func(cs *ConnectionStats, stat *livekit.AnalyticsStat) + lock sync.RWMutex + lastStatsAt time.Time + statsInProcess bool + scorer *qualityScorer done core.Fuse @@ -49,7 +55,7 @@ func NewConnectionStats(params ConnectionStatsParams) *ConnectionStats { } func (cs *ConnectionStats) Start(trackInfo *livekit.TrackInfo, at time.Time) { - cs.trackInfo = trackInfo + cs.isVideo.Store(trackInfo.Type == livekit.TrackType_VIDEO) cs.scorer.Start(at) @@ -76,6 +82,10 @@ func (cs *ConnectionStats) GetScoreAndQuality() (float32, livekit.ConnectionQual return cs.scorer.GetMOSAndQuality() } +func (cs *ConnectionStats) ReceiverReportReceived(at time.Time) { + cs.getStat(at) +} + func (cs *ConnectionStats) updateScore(streams map[uint32]*buffer.StreamStatsWithLayers, at time.Time) float32 { var stat windowStat for _, s := range streams { @@ -102,40 +112,83 @@ func (cs *ConnectionStats) updateScore(streams map[uint32]*buffer.StreamStatsWit return mos } -func (cs *ConnectionStats) getStat(at time.Time) *livekit.AnalyticsStat { +func (cs *ConnectionStats) maybeMarkInProcess() bool { + cs.lock.Lock() + defer cs.lock.Unlock() + + if cs.statsInProcess { + // already running + return false + } + + interval := cs.params.UpdateInterval + if interval == 0 { + interval = UpdateInterval + } + + if cs.lastStatsAt.IsZero() || time.Since(cs.lastStatsAt) > time.Duration(processThreshold*float64(interval)) { + cs.statsInProcess = true + return true + } + + return false +} + +func (cs *ConnectionStats) updateInProcess(isAvailable bool, at time.Time) { + cs.lock.Lock() + defer cs.lock.Unlock() + + cs.statsInProcess = false + if isAvailable { + cs.lastStatsAt = at + } +} + +func (cs *ConnectionStats) getStat(at time.Time) { if cs.params.GetDeltaStats == nil { - return nil + return + } + + if !cs.maybeMarkInProcess() { + // not yet time to process + return } streams := cs.params.GetDeltaStats() if len(streams) == 0 { - return nil + cs.updateInProcess(false, at) + return } - analyticsStreams := make([]*livekit.AnalyticsStream, 0, len(streams)) - for ssrc, stream := range streams { - as := toAnalyticsStream(ssrc, stream.RTPStats) - - // - // add video layer if either - // 1. Simulcast - even if there is only one layer per stream as it provides layer id - // 2. A stream has multiple layers - // - if cs.trackInfo.Type == livekit.TrackType_VIDEO && (len(streams) > 1 || len(stream.Layers) > 1) { - for layer, layerStats := range stream.Layers { - as.VideoLayers = append(as.VideoLayers, toAnalyticsVideoLayer(layer, layerStats)) - } - } - - analyticsStreams = append(analyticsStreams, as) - } + // stats available, update last stats time + cs.updateInProcess(true, at) score := cs.updateScore(streams, at) - return &livekit.AnalyticsStat{ - Score: score, - Streams: analyticsStreams, - Mime: cs.params.MimeType, + if cs.onStatsUpdate != nil { + analyticsStreams := make([]*livekit.AnalyticsStream, 0, len(streams)) + for ssrc, stream := range streams { + as := toAnalyticsStream(ssrc, stream.RTPStats) + + // + // add video layer if either + // 1. Simulcast - even if there is only one layer per stream as it provides layer id + // 2. A stream has multiple layers + // + if (len(streams) > 1 || len(stream.Layers) > 1) && cs.isVideo.Load() { + for layer, layerStats := range stream.Layers { + as.VideoLayers = append(as.VideoLayers, toAnalyticsVideoLayer(layer, layerStats)) + } + } + + analyticsStreams = append(analyticsStreams, as) + } + + cs.onStatsUpdate(cs, &livekit.AnalyticsStat{ + Score: score, + Streams: analyticsStreams, + Mime: cs.params.MimeType, + }) } } @@ -158,14 +211,7 @@ func (cs *ConnectionStats) updateStatsWorker() { return } - stat := cs.getStat(time.Now()) - if stat == nil { - continue - } - - if cs.onStatsUpdate != nil { - cs.onStatsUpdate(cs, stat) - } + cs.getStat(time.Now()) } } } diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index 8fdeaa589..6d2a31df1 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -1307,6 +1307,8 @@ func (d *DownTrack) handleRTCP(bytes []byte) { l(d, rr) } d.listenerLock.RUnlock() + + d.connectionStats.ReceiverReportReceived(time.Now()) } case *rtcp.TransportLayerNack: