Push/pull for connection stats/quality scoring. (#1505)

* Push/pull for connection stats/quality scoring.

Was not happy with pure pull method missing a window because
of RTCP RR timing is slightly off for audio and using a much
larger window of data in the next update.

That also resulted in RTP stats getting some bits of code.
As that is per-packet processing, was not a good idea.

Switching to push-pull method.
For up track, it is pull, i. e. connection stats worker will pull stats.

For down track, there is a new notification about receiver report
reception. Using this to check for time to run stats. And adding a bit
of tolerance for processing window (currently set so that as long as it
is > 95% of usual processing interval). This allows two things
- for video, RTCP RR are more frequent, but we will still not process
  till enough time has passed
- for audio, RTCP RR could be once in 5 seconds or so. Can process when
  it is available rather than miss a window and use a much larger window
  later.

* uber atomic

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 {

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())
 		}
 	}
 }

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: