Refactoring connection stats (#384)

pkg/rtc/mediatrack.go

@@ -21,7 +21,6 @@ import (
 )
 
 const (
-	upLostUpdateDelta               = time.Second
 	connectionQualityUpdateInterval = 5 * time.Second
 )
 
@@ -37,18 +36,12 @@ type MediaTrack struct {
 	audioLevelMu sync.RWMutex
 	audioLevel   *AudioLevel
 
-	statsLock         sync.Mutex
-	currentUpFracLost uint32
-	maxUpFracLost     uint8
-	maxUpFracLostTs   time.Time
-	connectionStats   *connectionquality.ConnectionStats
+	connectionStats *connectionquality.ConnectionStats
 
 	done chan struct{}
 
 	*MediaTrackReceiver
 
-	onMediaLossUpdate func(trackID livekit.TrackID, fractionalLoss uint32)
-
 	lock sync.RWMutex
 }
 
@@ -70,9 +63,7 @@ type MediaTrackParams struct {
 
 func NewMediaTrack(track *webrtc.TrackRemote, params MediaTrackParams) *MediaTrack {
 	t := &MediaTrack{
-		params:          params,
-		connectionStats: connectionquality.NewConnectionStats(),
-		done:            make(chan struct{}),
+		params: params,
 	}
 
 	t.MediaTrackReceiver = NewMediaTrackReceiver(MediaTrackReceiverParams{
@@ -106,13 +97,30 @@ func NewMediaTrack(track *webrtc.TrackRemote, params MediaTrackParams) *MediaTra
 		}
 	})
 
-	// on close signal via closing channel to workers
-	t.AddOnClose(t.closeChan)
-	return t
-}
+	t.connectionStats = connectionquality.NewConnectionStats(connectionquality.ConnectionStatsParams{
+		UpdateInterval: connectionQualityUpdateInterval,
+		CodecType:      track.Kind(),
+		GetTotalBytes: func() uint64 {
+			receiver := t.Receiver()
+			if receiver != nil {
+				return receiver.(*sfu.WebRTCReceiver).GetTotalBytes()
+			}
 
-func (t *MediaTrack) OnMediaLossUpdate(f func(trackID livekit.TrackID, fractionalLoss uint32)) {
-	t.onMediaLossUpdate = f
+			return 0
+		},
+		GetIsReducedQuality: func() bool {
+			publishing, expected := t.getNumUpTracks()
+			return publishing < expected
+		},
+	})
+	t.connectionStats.OnStatsUpdate(func(_cs *connectionquality.ConnectionStats, stat *livekit.AnalyticsStat) {
+		t.params.Telemetry.TrackStats(livekit.StreamType_UPSTREAM, t.PublisherID(), t.ID(), stat)
+	})
+
+	t.AddOnClose(func() {
+		t.connectionStats.Close()
+	})
+	return t
 }
 
 func (t *MediaTrack) SignalCid() string {
@@ -138,10 +146,6 @@ func (t *MediaTrack) ToProto() *livekit.TrackInfo {
 	return info
 }
 
-func (t *MediaTrack) publishLossPercentage() uint32 {
-	return FixedPointToPercent(uint8(atomic.LoadUint32(&t.currentUpFracLost)))
-}
-
 func (t *MediaTrack) getNumUpTracks() (uint32, uint32) {
 	numExpected := atomic.LoadUint32(&t.numUpTracks)
 
@@ -224,7 +228,8 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track *webrtc.Tra
 		t.buffer = buff
 
 		t.MediaTrackReceiver.SetupReceiver(wr)
-		go t.updateStats()
+
+		t.connectionStats.Start()
 	}
 	t.lock.Unlock()
 
@@ -266,80 +271,7 @@ func (t *MediaTrack) GetAudioLevel() (level uint8, active bool) {
 }
 
 func (t *MediaTrack) handlePublisherFeedback(packets []rtcp.Packet) {
-	var maxLost uint8
-	var hasReport bool
-	var delay uint32
-	var jitter uint32
-	var totalLost uint32
-	var maxSeqNum uint32
-	var nackCount int32
-	var pliCount int32
-	var firCount int32
-
-	for _, p := range packets {
-		switch pkt := p.(type) {
-		// sfu.Buffer generates ReceiverReports for the publisher
-		case *rtcp.ReceiverReport:
-			for _, rr := range pkt.Reports {
-				if rr.FractionLost > maxLost {
-					maxLost = rr.FractionLost
-				}
-				if rr.Delay > delay {
-					delay = rr.Delay
-				}
-				if rr.Jitter > jitter {
-					jitter = rr.Jitter
-				}
-				if rr.LastSequenceNumber > maxSeqNum {
-					maxSeqNum = rr.LastSequenceNumber
-				}
-				totalLost = rr.TotalLost
-			}
-			hasReport = true
-		case *rtcp.TransportLayerNack:
-			nackCount++
-			hasReport = true
-		case *rtcp.PictureLossIndication:
-			pliCount++
-			hasReport = true
-		case *rtcp.FullIntraRequest:
-			firCount++
-			hasReport = true
-		}
-	}
-
-	if hasReport {
-		t.statsLock.Lock()
-		if maxLost > t.maxUpFracLost {
-			t.maxUpFracLost = maxLost
-		}
-
-		now := time.Now()
-		if now.Sub(t.maxUpFracLostTs) > upLostUpdateDelta {
-			atomic.StoreUint32(&t.currentUpFracLost, uint32(t.maxUpFracLost))
-			t.maxUpFracLost = 0
-			t.maxUpFracLostTs = now
-		}
-		t.statsLock.Unlock()
-
-		// update feedback stats
-		t.connectionStats.Lock.Lock()
-		current := t.connectionStats
-		if jitter > current.Jitter {
-			current.Jitter = jitter
-		}
-		if delay > current.Delay {
-			current.Delay = delay
-		}
-		if maxSeqNum > current.LastSeqNum {
-			current.LastSeqNum = maxSeqNum
-		}
-		current.PacketsLost = totalLost
-		current.NackCount += nackCount
-		current.PliCount += pliCount
-		current.FirCount += firCount
-		t.connectionStats.Lock.Unlock()
-	}
+	t.connectionStats.RTCPFeedback(packets, 0)
 
 	// also look for sender reports
 	// feedback for the source RTCP
@@ -347,60 +279,5 @@ func (t *MediaTrack) handlePublisherFeedback(packets []rtcp.Packet) {
 }
 
 func (t *MediaTrack) GetConnectionScore() float64 {
-	t.connectionStats.Lock.Lock()
-	defer t.connectionStats.Lock.Unlock()
-	return t.connectionStats.Score
-}
-
-func (t *MediaTrack) closeChan() {
-	close(t.done)
-}
-
-func (t *MediaTrack) updateStats() {
-
-	for {
-		select {
-		case <-t.done:
-			return
-		case <-time.After(connectionQualityUpdateInterval):
-			t.connectionStats.Lock.Lock()
-
-			//we are accessing stats after receiver has buffer assigned
-			stats := t.buffer.GetStats()
-
-			delta := t.connectionStats.UpdateStats(stats.TotalByte)
-			if t.Kind() == livekit.TrackType_AUDIO {
-				t.connectionStats.Score = connectionquality.AudioConnectionScore(delta, t.connectionStats.Jitter)
-			} else {
-				t.connectionStats.Score = t.calculateVideoScore()
-			}
-			stat := &livekit.AnalyticsStat{
-				Jitter:          float64(t.connectionStats.Jitter),
-				TotalPackets:    uint64(t.connectionStats.TotalPackets),
-				PacketLost:      uint64(t.connectionStats.PacketsLost),
-				Delay:           uint64(t.connectionStats.Delay),
-				TotalBytes:      t.connectionStats.TotalBytes,
-				NackCount:       t.connectionStats.NackCount,
-				PliCount:        t.connectionStats.PliCount,
-				FirCount:        t.connectionStats.FirCount,
-				ConnectionScore: float32(t.connectionStats.Score),
-			}
-			t.params.Telemetry.TrackStats(livekit.StreamType_UPSTREAM, t.PublisherID(), t.ID(), stat)
-			t.connectionStats.Lock.Unlock()
-		}
-	}
-}
-
-func (t *MediaTrack) calculateVideoScore() float64 {
-	var reducedQuality bool
-	publishing, expected := t.getNumUpTracks()
-	if publishing < expected {
-		reducedQuality = true
-	}
-
-	loss := t.publishLossPercentage()
-	if expected == 0 {
-		loss = 0
-	}
-	return connectionquality.VideoConnectionScore(loss, reducedQuality)
+	return t.connectionStats.GetScore()
 }

pkg/rtc/participant.go

@@ -734,21 +734,21 @@ func (p *ParticipantImpl) GetAudioLevel() (level uint8, active bool) {
 
 func (p *ParticipantImpl) GetConnectionQuality() *livekit.ConnectionQualityInfo {
 	// avg loss across all tracks, weigh published the same as subscribed
-	scores, numTracks := p.getPublisherConnectionQuality()
+	totalScore, numTracks := p.getPublisherConnectionQuality()
 
 	p.lock.RLock()
 	for _, subTrack := range p.subscribedTracks {
 		if subTrack.IsMuted() || subTrack.MediaTrack().IsMuted() {
 			continue
 		}
-		scores += subTrack.DownTrack().GetConnectionScore()
+		totalScore += subTrack.DownTrack().GetConnectionScore()
 		numTracks++
 	}
 	p.lock.RUnlock()
 
 	avgScore := 5.0
 	if numTracks > 0 {
-		avgScore = scores / float64(numTracks)
+		avgScore = totalScore / float64(numTracks)
 	}
 
 	rating := connectionquality.Score2Rating(avgScore)
@@ -1332,12 +1332,12 @@ func (p *ParticipantImpl) setTrackMuted(trackID livekit.TrackID, muted bool) {
 	}
 }
 
-func (p *ParticipantImpl) getPublisherConnectionQuality() (scores float64, numTracks int) {
+func (p *ParticipantImpl) getPublisherConnectionQuality() (totalScore float64, numTracks int) {
 	for _, pt := range p.GetPublishedTracks() {
 		if pt.IsMuted() {
 			continue
 		}
-		scores += pt.(types.LocalMediaTrack).GetConnectionScore()
+		totalScore += pt.(types.LocalMediaTrack).GetConnectionScore()
 		numTracks++
 	}
 

pkg/rtc/participant_internal_test.go

@@ -245,7 +245,7 @@ func TestMuteSetting(t *testing.T) {
 func TestConnectionQuality(t *testing.T) {
 
 	// loss based score is currently a publisher method.
-	videoScore := func(loss, numPublishing, numRegistered uint32) float64 {
+	videoScore := func(loss float64, numPublishing, numRegistered uint32) float64 {
 		var reducedQuality bool
 		if numRegistered > 0 && numPublishing != numRegistered {
 			reducedQuality = true
@@ -253,7 +253,7 @@ func TestConnectionQuality(t *testing.T) {
 		return connectionquality.VideoConnectionScore(loss, reducedQuality)
 	}
 
-	testPublishedVideoTrack := func(loss, numPublishing, numRegistered uint32) *typesfakes.FakeLocalMediaTrack {
+	testPublishedVideoTrack := func(loss float64, numPublishing, numRegistered uint32) *typesfakes.FakeLocalMediaTrack {
 		tr := &typesfakes.FakeLocalMediaTrack{}
 		score := videoScore(loss, numPublishing, numRegistered)
 		t.Log("video score: ", score)
@@ -264,22 +264,13 @@ func TestConnectionQuality(t *testing.T) {
 	testPublishedAudioTrack := func(totalPackets, packetsLost uint32) *typesfakes.FakeLocalMediaTrack {
 		tr := &typesfakes.FakeLocalMediaTrack{}
 
-		stat := connectionquality.ConnectionStat{
-			PacketsLost:  packetsLost,
-			TotalPackets: totalPackets,
-			LastSeqNum:   0,
+		pctLoss := float64(0.0)
+		if totalPackets > 0 {
+			pctLoss = (float64(packetsLost) / float64(totalPackets)) * 100.0
 		}
-		stats := &connectionquality.ConnectionStats{
-			ConnectionStat: stat,
-			Prev:           &connectionquality.ConnectionStat{},
-		}
-		stats.Score = connectionquality.AudioConnectionScore(connectionquality.ConnectionStat{
-			TotalPackets: stat.TotalPackets,
-			PacketsLost:  stat.PacketsLost,
-			TotalBytes:   stat.TotalBytes,
-		}, 0)
-		t.Log("audio score: ", stats.Score)
-		tr.GetConnectionScoreReturns(stats.Score)
+		score := connectionquality.AudioConnectionScore(pctLoss, 0)
+		t.Log("audio score: ", score)
+		tr.GetConnectionScoreReturns(score)
 		return tr
 	}
 

pkg/rtc/utils.go

@@ -45,11 +45,6 @@ func UnpackDataTrackLabel(packed string) (peerID livekit.ParticipantID, trackID
 	return
 }
 
-// converts a fixed point number to the number part of %
-func FixedPointToPercent(frac uint8) uint32 {
-	return (uint32(frac) * 100) >> 8
-}
-
 func ToProtoParticipants(participants []types.LocalParticipant) []*livekit.ParticipantInfo {
 	infos := make([]*livekit.ParticipantInfo, 0, len(participants))
 	for _, op := range participants {

pkg/sfu/buffer/buffer.go

@@ -101,7 +101,7 @@ type Stats struct {
 	LostRate     float32
 	PacketCount  uint32  // Number of packets received from this source.
 	Jitter       float64 // An estimate of the statistical variance of the RTP data packet inter-arrival time.
-	TotalByte    uint64
+	TotalBytes   uint64
 }
 
 // BufferOptions provides configuration options for the buffer
@@ -313,7 +313,7 @@ func (b *Buffer) calc(pkt []byte, arrivalTime int64) {
 		}
 	}
 
-	b.stats.TotalByte += uint64(len(pkt))
+	b.stats.TotalBytes += uint64(len(pkt))
 	b.stats.PacketCount++
 
 	ep := ExtPacket{
@@ -458,7 +458,7 @@ func (b *Buffer) buildREMBPacket() *rtcp.ReceiverEstimatedMaximumBitrate {
 	if br < 100000 {
 		br = 100000
 	}
-	b.stats.TotalByte = 0
+	b.stats.TotalBytes = 0
 
 	return &rtcp.ReceiverEstimatedMaximumBitrate{
 		Bitrate: float32(br),
@@ -481,9 +481,9 @@ func (b *Buffer) buildReceptionReport() rtcp.ReceptionReport {
 
 	lostInterval := expectedInterval - receivedInterval
 
-	b.stats.LostRate = float32(lostInterval) / float32(expectedInterval)
 	var fracLost uint8
-	if expectedInterval != 0 && lostInterval > 0 {
+	if expectedInterval != 0 {
+		b.stats.LostRate = float32(lostInterval) / float32(expectedInterval)
 		fracLost = uint8((lostInterval << 8) / expectedInterval)
 	}
 	if b.lastFractionLostToReport > fracLost {

pkg/sfu/connectionquality/connectionstats.go

@@ -1,57 +1,205 @@
 package connectionquality
 
-import "sync"
+import (
+	"sync"
+	"time"
 
-type ConnectionStat struct {
-	PacketsLost  uint32
-	TotalPackets uint32
-	LastSeqNum   uint32
-	TotalBytes   uint64
+	"github.com/livekit/protocol/livekit"
+	"github.com/livekit/protocol/utils"
+	"github.com/pion/rtcp"
+	"github.com/pion/webrtc/v3"
+)
+
+type Snapshot struct {
+	Initialized      bool
+	TotalPacketsLost uint32
+	HighestSeqNum    uint32
+}
+
+type ConnectionStatsParams struct {
+	UpdateInterval      time.Duration
+	CodecType           webrtc.RTPCodecType
+	GetTotalBytes       func() uint64
+	GetIsReducedQuality func() bool
 }
 
 type ConnectionStats struct {
-	Lock sync.Mutex
-	ConnectionStat
-	Prev      *ConnectionStat
-	Delay     uint32
-	Jitter    uint32
-	NackCount int32
-	PliCount  int32
-	FirCount  int32
-	Score     float64
+	lock sync.RWMutex
+
+	params ConnectionStatsParams
+
+	baseSeqNumInitialized bool
+	baseSeqNum            uint32
+	highestSeqNum         uint32
+	totalPacketsLost      uint32
+	totalBytes            uint64
+
+	snapshot Snapshot
+
+	maxDelay  uint32
+	maxJitter uint32
+
+	nackCount int32
+	pliCount  int32
+	firCount  int32
+
+	score float64
+
+	onStatsUpdate func(cs *ConnectionStats, stat *livekit.AnalyticsStat)
+
+	done     chan struct{}
+	isClosed utils.AtomicFlag
 }
 
-func NewConnectionStats() *ConnectionStats {
-	return &ConnectionStats{Prev: &ConnectionStat{}, Score: 4.0}
+func NewConnectionStats(params ConnectionStatsParams) *ConnectionStats {
+	return &ConnectionStats{
+		params: params,
+		score:  4.0,
+		done:   make(chan struct{}),
+	}
 }
 
-func getTotalPackets(curSN, prevSN uint32) uint32 {
-	delta := curSN - prevSN
-	// lower 16 bits is the counter
-	counter := uint16(delta)
-	// upper 16 bits contains the cycles/wrap
-	cycles := uint16(delta >> 16)
-	increment := (uint32(cycles) * (1 << 16)) + uint32(counter)
-
-	return increment
+func (cs *ConnectionStats) Start() {
+	go cs.updateStats()
 }
 
-func (cs *ConnectionStats) UpdateStats(totalBytes uint64) ConnectionStat {
-	// update feedback stats
-	previous := cs.Prev
+func (cs *ConnectionStats) Close() {
+	if !cs.isClosed.TrySet(true) {
+		return
+	}
 
-	// Update TotalPackets from SeqNum here
-	cs.TotalPackets += getTotalPackets(cs.LastSeqNum, previous.LastSeqNum)
-	cs.TotalBytes = totalBytes
-
-	var delta ConnectionStat
-
-	delta.TotalPackets = cs.TotalPackets - previous.TotalPackets
-	delta.PacketsLost = cs.PacketsLost - previous.PacketsLost
-	delta.TotalBytes = cs.TotalBytes - previous.TotalBytes
-
-	// store previous stats
-	cs.Prev = &ConnectionStat{TotalPackets: cs.TotalPackets, PacketsLost: cs.PacketsLost, TotalBytes: cs.TotalBytes, LastSeqNum: cs.LastSeqNum}
-
-	return delta
+	close(cs.done)
+}
+
+func (cs *ConnectionStats) OnStatsUpdate(fn func(cs *ConnectionStats, stat *livekit.AnalyticsStat)) {
+	cs.onStatsUpdate = fn
+}
+
+func (cs *ConnectionStats) RTCPFeedback(packets []rtcp.Packet, expectedSSRC uint32) {
+	if cs.isClosed.Get() {
+		return
+	}
+
+	cs.lock.Lock()
+	defer cs.lock.Unlock()
+
+	for _, p := range packets {
+		switch pkt := p.(type) {
+		case *rtcp.ReceiverReport:
+			for _, r := range pkt.Reports {
+				if expectedSSRC != 0 && r.SSRC != expectedSSRC {
+					continue
+				}
+
+				if r.Delay > cs.maxDelay {
+					cs.maxDelay = r.Delay
+				}
+
+				if r.Jitter > cs.maxJitter {
+					cs.maxJitter = r.Jitter
+				}
+
+				if !cs.baseSeqNumInitialized {
+					cs.baseSeqNumInitialized = true
+					cs.baseSeqNum = r.LastSequenceNumber
+					cs.highestSeqNum = r.LastSequenceNumber
+					cs.totalPacketsLost = r.TotalLost
+				}
+
+				if r.LastSequenceNumber > cs.highestSeqNum {
+					cs.highestSeqNum = r.LastSequenceNumber
+					cs.totalPacketsLost = r.TotalLost
+				}
+			}
+
+		case *rtcp.TransportLayerNack:
+			nackCount := 0
+			for _, pair := range pkt.Nacks {
+				nackCount += len(pair.PacketList())
+			}
+			cs.nackCount += int32(nackCount)
+
+		case *rtcp.PictureLossIndication:
+			cs.pliCount += 1
+
+		case *rtcp.FullIntraRequest:
+			cs.firCount += 1
+		}
+	}
+}
+
+func (cs *ConnectionStats) GetScore() float64 {
+	cs.lock.RLock()
+	defer cs.lock.RUnlock()
+
+	return cs.score
+}
+
+func (cs *ConnectionStats) updateAndGetPercentageLoss() float64 {
+	if cs.params.GetTotalBytes != nil {
+		cs.totalBytes = cs.params.GetTotalBytes()
+	}
+
+	if !cs.snapshot.Initialized {
+		cs.snapshot.Initialized = true
+		cs.snapshot.HighestSeqNum = cs.highestSeqNum
+		cs.snapshot.TotalPacketsLost = cs.totalPacketsLost
+	}
+
+	packetsLostInInterval := cs.totalPacketsLost - cs.snapshot.TotalPacketsLost
+	expectedPacketsInInterval := cs.highestSeqNum - cs.snapshot.HighestSeqNum
+	percentageLoss := float64(0.0)
+	if expectedPacketsInInterval > 0 {
+		percentageLoss = (float64(packetsLostInInterval) / float64(expectedPacketsInInterval)) * 100
+	}
+
+	cs.snapshot.HighestSeqNum = cs.highestSeqNum
+	cs.snapshot.TotalPacketsLost = cs.totalPacketsLost
+
+	return percentageLoss
+}
+
+func (cs *ConnectionStats) updateStats() {
+	tk := time.NewTicker(cs.params.UpdateInterval)
+	for {
+		select {
+		case <-cs.done:
+			return
+
+		case <-tk.C:
+			cs.lock.Lock()
+			pctLoss := cs.updateAndGetPercentageLoss()
+			if cs.params.CodecType == webrtc.RTPCodecTypeAudio {
+				cs.score = AudioConnectionScore(pctLoss, cs.maxJitter)
+			} else {
+				isReducedQuality := false
+				if cs.params.GetIsReducedQuality != nil {
+					isReducedQuality = cs.params.GetIsReducedQuality()
+				}
+				cs.score = VideoConnectionScore(pctLoss, isReducedQuality)
+			}
+
+			totalPacketsReceived := uint32(0)
+			if cs.baseSeqNumInitialized {
+				totalPacketsReceived = cs.highestSeqNum - cs.baseSeqNum
+			}
+
+			stat := &livekit.AnalyticsStat{
+				TotalPackets:    uint64(totalPacketsReceived),
+				PacketLost:      uint64(cs.totalPacketsLost),
+				TotalBytes:      cs.totalBytes,
+				Delay:           uint64(cs.maxDelay),
+				Jitter:          float64(cs.maxJitter),
+				NackCount:       cs.nackCount,
+				PliCount:        cs.pliCount,
+				FirCount:        cs.firCount,
+				ConnectionScore: float32(cs.score),
+			}
+			cs.lock.Unlock()
+
+			if cs.onStatsUpdate != nil {
+				cs.onStatsUpdate(cs, stat)
+			}
+		}
+	}
 }

pkg/sfu/connectionquality/mos.go

@@ -22,16 +22,8 @@ func Score2Rating(score float64) livekit.ConnectionQuality {
 	return livekit.ConnectionQuality_POOR
 }
 
-func mosAudioEmodel(delta ConnectionStat, jitter uint32) float64 {
-
-	// find percentage of lost packets in this window
-	if delta.TotalPackets == 0 {
-		return 0.0
-	}
-
-	percentageLost := (float64(delta.PacketsLost) / float64(delta.TotalPackets)) * 100
-
-	rx := 93.2 - percentageLost
+func mosAudioEmodel(pctLoss float64, jitter uint32) float64 {
+	rx := 93.2 - pctLoss
 	ry := 0.18*rx*rx - 27.9*rx + 1126.62
 
 	// Jitter is in MicroSecs (1/1e6) units. Convert it to MilliSecs
@@ -55,27 +47,27 @@ func mosAudioEmodel(delta ConnectionStat, jitter uint32) float64 {
 	return score
 }
 
-func loss2Score(loss uint32, reducedQuality bool) float64 {
+func loss2Score(pctLoss float64, reducedQuality bool) float64 {
 	// No Loss, excellent
-	if loss == 0 && !reducedQuality {
+	if pctLoss == 0.0 && !reducedQuality {
 		return 5
 	}
 	// default when loss is minimal, but reducedQuality
 	score := 3.5
 	// loss is bad
-	if loss >= 4 {
+	if pctLoss >= 4.0 {
 		score = 2.0
-	} else if loss <= 2 && !reducedQuality {
+	} else if pctLoss <= 2.0 && !reducedQuality {
 		// loss is acceptable and at reduced quality
 		score = 4.5
 	}
 	return score
 }
 
-func AudioConnectionScore(delta ConnectionStat, jitter uint32) float64 {
-	return mosAudioEmodel(delta, jitter)
+func AudioConnectionScore(pctLoss float64, jitter uint32) float64 {
+	return mosAudioEmodel(pctLoss, jitter)
 }
 
-func VideoConnectionScore(pctLoss uint32, reducedQuality bool) float64 {
+func VideoConnectionScore(pctLoss float64, reducedQuality bool) float64 {
 	return loss2Score(pctLoss, reducedQuality)
 }

pkg/sfu/downtrack.go

@@ -100,13 +100,11 @@ type DownTrack struct {
 	closeOnce               sync.Once
 
 	// Report helpers
-	primaryStats    atomic.Value // contains *PacketStats
-	rtxStats        atomic.Value // contains *PacketStats
-	paddingStats    atomic.Value // contains *PacketStats
-	connectionStats *connectionquality.ConnectionStats
-	done            chan struct{}
+	primaryStats atomic.Value // contains *PacketStats
+	rtxStats     atomic.Value // contains *PacketStats
+	paddingStats atomic.Value // contains *PacketStats
 
-	lossFraction atomicUint8
+	connectionStats *connectionquality.ConnectionStats
 
 	// Debug info
 	lastPli     atomicInt64
@@ -114,7 +112,6 @@ type DownTrack struct {
 	pktsDropped atomicUint32
 
 	// RTCP callbacks
-	onRTCP                func([]rtcp.Packet)
 	onREMB                func(dt *DownTrack, remb *rtcp.ReceiverEstimatedMaximumBitrate)
 	onTransportCCFeedback func(dt *DownTrack, cc *rtcp.TransportLayerCC)
 
@@ -150,25 +147,39 @@ func NewDownTrack(c webrtc.RTPCodecCapability, r TrackReceiver, bf *buffer.Facto
 	}
 
 	d := &DownTrack{
-		id:              r.TrackID(),
-		peerID:          peerID,
-		maxTrack:        mt,
-		streamID:        r.StreamID(),
-		bufferFactory:   bf,
-		receiver:        r,
-		codec:           c,
-		kind:            kind,
-		forwarder:       NewForwarder(c, kind),
-		connectionStats: connectionquality.NewConnectionStats(),
-		done:            make(chan struct{}),
+		id:            r.TrackID(),
+		peerID:        peerID,
+		maxTrack:      mt,
+		streamID:      r.StreamID(),
+		bufferFactory: bf,
+		receiver:      r,
+		codec:         c,
+		kind:          kind,
+		forwarder:     NewForwarder(c, kind),
 	}
 
+	d.connectionStats = connectionquality.NewConnectionStats(connectionquality.ConnectionStatsParams{
+		UpdateInterval: connectionQualityUpdateInterval,
+		CodecType:      kind,
+		GetTotalBytes: func() uint64 {
+			octets, _ := d.getSRStats()
+			return octets
+		},
+		GetIsReducedQuality: func() bool {
+			return d.GetForwardingStatus() != ForwardingStatusOptimal
+		},
+	})
+	d.connectionStats.OnStatsUpdate(func(_cs *connectionquality.ConnectionStats, stat *livekit.AnalyticsStat) {
+		if d.onStatsUpdate != nil {
+			d.onStatsUpdate(d, stat)
+		}
+	})
+	d.connectionStats.Start()
+
 	d.primaryStats.Store(new(PacketStats))
 	d.rtxStats.Store(new(PacketStats))
 	d.paddingStats.Store(new(PacketStats))
 
-	go d.updateStats()
-
 	return d, nil
 }
 
@@ -423,10 +434,6 @@ func (d *DownTrack) Mute(val bool) {
 		return
 	}
 
-	if val {
-		d.lossFraction.set(0)
-	}
-
 	if d.onSubscriptionChanged != nil {
 		d.onSubscriptionChanged(d)
 	}
@@ -457,11 +464,11 @@ func (d *DownTrack) CloseWithFlush(flush bool) {
 		Logger.V(1).Info("Closing sender", "peer_id", d.peerID, "kind", d.kind)
 		d.receiver.DeleteDownTrack(d.peerID)
 
+		d.connectionStats.Close()
+
 		if d.onCloseHandler != nil {
 			d.onCloseHandler()
 		}
-
-		close(d.done)
 	})
 }
 
@@ -512,10 +519,6 @@ func (d *DownTrack) OnBind(fn func()) {
 	d.onBind = fn
 }
 
-func (d *DownTrack) OnRTCP(fn func([]rtcp.Packet)) {
-	d.onRTCP = fn
-}
-
 func (d *DownTrack) OnREMB(fn func(dt *DownTrack, remb *rtcp.ReceiverEstimatedMaximumBitrate)) {
 	d.onREMB = fn
 }
@@ -524,13 +527,10 @@ func (d *DownTrack) OnTransportCCFeedback(fn func(dt *DownTrack, cc *rtcp.Transp
 	d.onTransportCCFeedback = fn
 }
 
-func (d *DownTrack) CurrentMaxLossFraction() uint8 {
-	return d.lossFraction.get()
-}
-
 func (d *DownTrack) AddReceiverReportListener(listener ReceiverReportListener) {
 	d.listenerLock.Lock()
 	defer d.listenerLock.Unlock()
+
 	d.receiverReportListeners = append(d.receiverReportListeners, listener)
 }
 
@@ -793,10 +793,6 @@ func (d *DownTrack) handleRTCP(bytes []byte) {
 		return
 	}
 
-	if d.onRTCP != nil {
-		d.onRTCP(pkts)
-	}
-
 	pliOnce := true
 	sendPliOnce := func() {
 		if pliOnce {
@@ -809,30 +805,19 @@ func (d *DownTrack) handleRTCP(bytes []byte) {
 		}
 	}
 
-	var hasReport bool
-	var delay uint32
-	var jitter uint32
-	var totalLost uint32
-	var maxSeqNum uint32
-	var nackCount int32
-	var pliCount int32
-	var firCount int32
-
-	maxRatePacketLoss := uint8(0)
 	for _, pkt := range pkts {
 		switch p := pkt.(type) {
 		case *rtcp.PictureLossIndication:
 			sendPliOnce()
-			pliCount++
-			hasReport = true
+
 		case *rtcp.FullIntraRequest:
 			sendPliOnce()
-			firCount++
-			hasReport = true
+
 		case *rtcp.ReceiverEstimatedMaximumBitrate:
 			if d.onREMB != nil {
 				d.onREMB(d, p)
 			}
+
 		case *rtcp.ReceiverReport:
 			// create new receiver report w/ only valid reception reports
 			rr := &rtcp.ReceiverReport{
@@ -844,23 +829,7 @@ func (d *DownTrack) handleRTCP(bytes []byte) {
 					continue
 				}
 				rr.Reports = append(rr.Reports, r)
-				if maxRatePacketLoss == 0 || maxRatePacketLoss < r.FractionLost {
-					maxRatePacketLoss = r.FractionLost
-				}
-				if r.Delay > delay {
-					delay = r.Delay
-				}
-				if r.Jitter > jitter {
-					jitter = r.Jitter
-				}
-				if r.LastSequenceNumber > maxSeqNum {
-					maxSeqNum = r.LastSequenceNumber
-				}
-				totalLost = r.TotalLost
-
-				hasReport = true
 			}
-			d.lossFraction.set(maxRatePacketLoss)
 			if len(rr.Reports) > 0 {
 				d.listenerLock.RLock()
 				for _, l := range d.receiverReportListeners {
@@ -875,33 +844,15 @@ func (d *DownTrack) handleRTCP(bytes []byte) {
 				nackedPackets = append(nackedPackets, d.sequencer.getSeqNoPairs(pair.PacketList())...)
 			}
 			go d.retransmitPackets(nackedPackets)
-			nackCount += int32(len(nackedPackets))
-			hasReport = true
+
 		case *rtcp.TransportLayerCC:
 			if p.MediaSSRC == d.ssrc && d.onTransportCCFeedback != nil {
 				d.onTransportCCFeedback(d, p)
 			}
 		}
 	}
-	if hasReport {
-		d.connectionStats.Lock.Lock()
-		current := d.connectionStats
-		// update feedback stats
-		if jitter > current.Jitter {
-			current.Jitter = jitter
-		}
-		if delay > current.Delay {
-			current.Delay = delay
-		}
-		if maxSeqNum > current.LastSeqNum {
-			current.LastSeqNum = maxSeqNum
-		}
-		current.PacketsLost = totalLost
-		current.NackCount += nackCount
-		current.PliCount += pliCount
-		current.FirCount += firCount
-		d.connectionStats.Lock.Unlock()
-	}
+
+	d.connectionStats.RTCPFeedback(pkts, d.ssrc)
 }
 
 func (d *DownTrack) retransmitPackets(nackedPackets []packetMeta) {
@@ -1081,49 +1032,5 @@ func (d *DownTrack) DebugInfo() map[string]interface{} {
 }
 
 func (d *DownTrack) GetConnectionScore() float64 {
-	d.connectionStats.Lock.Lock()
-	defer d.connectionStats.Lock.Unlock()
-	return d.connectionStats.Score
-}
-
-func (d *DownTrack) updateStats() {
-	for {
-		select {
-		case <-d.done:
-			return
-		case <-time.After(connectionQualityUpdateInterval):
-			d.connectionStats.Lock.Lock()
-			totalBytes, _ := d.getSRStats()
-			delta := d.connectionStats.UpdateStats(totalBytes)
-			if d.Kind() == webrtc.RTPCodecTypeAudio {
-				d.connectionStats.Score = connectionquality.AudioConnectionScore(delta, d.connectionStats.Jitter)
-			} else {
-				var reducedQuality bool
-				if d.GetForwardingStatus() != ForwardingStatusOptimal {
-					reducedQuality = true
-				}
-				d.connectionStats.Score = connectionquality.VideoConnectionScore(FixedPointToPercent(d.CurrentMaxLossFraction()), reducedQuality)
-			}
-			stat := &livekit.AnalyticsStat{
-				Jitter:          float64(d.connectionStats.Jitter),
-				TotalPackets:    uint64(d.connectionStats.TotalPackets),
-				PacketLost:      uint64(d.connectionStats.PacketsLost),
-				Delay:           uint64(d.connectionStats.Delay),
-				TotalBytes:      d.connectionStats.TotalBytes,
-				NackCount:       d.connectionStats.NackCount,
-				PliCount:        d.connectionStats.PliCount,
-				FirCount:        d.connectionStats.FirCount,
-				ConnectionScore: float32(d.connectionStats.Score),
-			}
-			if d.onStatsUpdate != nil {
-				d.onStatsUpdate(d, stat)
-			}
-			d.connectionStats.Lock.Unlock()
-		}
-	}
-}
-
-// converts a fixed point number to the number part of %
-func FixedPointToPercent(frac uint8) uint32 {
-	return (uint32(frac) * 100) >> 8
+	return d.connectionStats.GetScore()
 }

pkg/sfu/receiver.go

@@ -463,6 +463,21 @@ func (w *WebRTCReceiver) ReadRTP(buf []byte, layer uint8, sn uint16) (int, error
 	return buff.GetPacket(buf, sn)
 }
 
+func (w *WebRTCReceiver) GetTotalBytes() uint64 {
+	w.bufferMu.RLock()
+	defer w.bufferMu.RUnlock()
+
+	totalBytes := uint64(0)
+	for _, buffer := range w.buffers {
+		if buffer != nil {
+			stats := buffer.GetStats()
+			totalBytes += stats.TotalBytes
+		}
+	}
+
+	return totalBytes
+}
+
 func (w *WebRTCReceiver) forwardRTP(layer int32) {
 	w.upTrackMu.RLock()
 	tracker := w.trackers[layer]