diff --git a/pkg/sfu/connectionquality/connectionstats.go b/pkg/sfu/connectionquality/connectionstats.go index 98562ae1e..8608da3d6 100644 --- a/pkg/sfu/connectionquality/connectionstats.go +++ b/pkg/sfu/connectionquality/connectionstats.go @@ -21,11 +21,13 @@ const ( ) type ConnectionStatsParams struct { - UpdateInterval time.Duration - MimeType string - IsFECEnabled bool - GetDeltaStats func() map[uint32]*buffer.StreamStatsWithLayers - Logger logger.Logger + UpdateInterval time.Duration + MimeType string + IsFECEnabled bool + IsDependentRTT bool + IsDependentJitter bool + GetDeltaStats func() map[uint32]*buffer.StreamStatsWithLayers + Logger logger.Logger } type ConnectionStats struct { @@ -49,8 +51,10 @@ func NewConnectionStats(params ConnectionStatsParams) *ConnectionStats { return &ConnectionStats{ params: params, scorer: newQualityScorer(qualityScorerParams{ - PacketLossWeight: getPacketLossWeight(params.MimeType, params.IsFECEnabled), // LK-TODO: have to notify codec change? - Logger: params.Logger, + PacketLossWeight: getPacketLossWeight(params.MimeType, params.IsFECEnabled), // LK-TODO: have to notify codec change? + IsDependentRTT: params.IsDependentRTT, + IsDependentJitter: params.IsDependentJitter, + Logger: params.Logger, }), done: core.NewFuse(), } diff --git a/pkg/sfu/connectionquality/connectionstats_test.go b/pkg/sfu/connectionquality/connectionstats_test.go index 567ef485e..075d24dbc 100644 --- a/pkg/sfu/connectionquality/connectionstats_test.go +++ b/pkg/sfu/connectionquality/connectionstats_test.go @@ -11,17 +11,19 @@ import ( "github.com/stretchr/testify/require" ) -func newConnectionStats(mimeType string, isFECEnabled bool) *ConnectionStats { +func newConnectionStats(mimeType string, isFECEnabled bool, isDependentRTT bool, isDependentJitter bool) *ConnectionStats { return NewConnectionStats(ConnectionStatsParams{ - MimeType: mimeType, - IsFECEnabled: isFECEnabled, - Logger: logger.GetLogger(), + MimeType: mimeType, + IsFECEnabled: isFECEnabled, + IsDependentRTT: isDependentRTT, + IsDependentJitter: isDependentJitter, + Logger: logger.GetLogger(), }) } func TestConnectionQuality(t *testing.T) { t.Run("quality scorer state machine", func(t *testing.T) { - cs := newConnectionStats("audio/opus", false) + cs := newConnectionStats("audio/opus", false, false, false) duration := 5 * time.Second now := time.Now() @@ -344,6 +346,62 @@ func TestConnectionQuality(t *testing.T) { require.Equal(t, livekit.ConnectionQuality_GOOD, quality) }) + t.Run("quality scorer dependent rtt", func(t *testing.T) { + cs := newConnectionStats("audio/opus", false, true, false) + + duration := 5 * time.Second + now := time.Now() + cs.Start(&livekit.TrackInfo{Type: livekit.TrackType_AUDIO}, now.Add(-duration)) + cs.UpdateMute(false, now.Add(-1*time.Second)) + + // RTT does not knock quality down because it is dependent and hence not taken into account + // at 2% loss, quality should stay at EXCELLENT purely based on loss. With high RTT (700 ms) + // quality should drop to GOOD if RTT were taken into consieration + streams := map[uint32]*buffer.StreamStatsWithLayers{ + 1: &buffer.StreamStatsWithLayers{ + RTPStats: &buffer.RTPDeltaInfo{ + StartTime: now, + Duration: duration, + Packets: 250, + PacketsLost: 5, + RttMax: 700, + }, + }, + } + cs.updateScore(streams, now.Add(duration)) + mos, quality := cs.GetScoreAndQuality() + require.Greater(t, float32(4.6), mos) + require.Equal(t, livekit.ConnectionQuality_EXCELLENT, quality) + }) + + t.Run("quality scorer dependent jitter", func(t *testing.T) { + cs := newConnectionStats("audio/opus", false, false, true) + + duration := 5 * time.Second + now := time.Now() + cs.Start(&livekit.TrackInfo{Type: livekit.TrackType_AUDIO}, now.Add(-duration)) + cs.UpdateMute(false, now.Add(-1*time.Second)) + + // Jitter does not knock quality down because it is dependent and hence not taken into account + // at 2% loss, quality should stay at EXCELLENT purely based on loss. With high jitter (200 ms) + // quality should drop to GOOD if jitter were taken into consieration + streams := map[uint32]*buffer.StreamStatsWithLayers{ + 1: &buffer.StreamStatsWithLayers{ + RTPStats: &buffer.RTPDeltaInfo{ + StartTime: now, + Duration: duration, + Packets: 250, + PacketsLost: 5, + JitterMax: 200, + }, + }, + } + cs.updateScore(streams, now.Add(duration)) + mos, quality := cs.GetScoreAndQuality() + require.Greater(t, float32(4.6), mos) + require.Equal(t, livekit.ConnectionQuality_EXCELLENT, quality) + }) + t.Run("codecs - packet", func(t *testing.T) { type expectedQuality struct { packetLossPercentage float64 @@ -482,7 +540,7 @@ func TestConnectionQuality(t *testing.T) { for _, tc := range testCases { t.Run(tc.name, func(t *testing.T) { - cs := newConnectionStats(tc.mimeType, tc.isFECEnabled) + cs := newConnectionStats(tc.mimeType, tc.isFECEnabled, false, false) duration := 5 * time.Second now := time.Now() @@ -575,7 +633,7 @@ func TestConnectionQuality(t *testing.T) { for _, tc := range testCases { t.Run(tc.name, func(t *testing.T) { - cs := newConnectionStats("video/vp8", false) + cs := newConnectionStats("video/vp8", false, false, false) duration := 5 * time.Second now := time.Now() @@ -662,7 +720,7 @@ func TestConnectionQuality(t *testing.T) { for _, tc := range testCases { t.Run(tc.name, func(t *testing.T) { - cs := newConnectionStats("video/vp8", false) + cs := newConnectionStats("video/vp8", false, false, false) duration := 5 * time.Second now := time.Now() diff --git a/pkg/sfu/connectionquality/scorer.go b/pkg/sfu/connectionquality/scorer.go index 5582a7832..d89329cfe 100644 --- a/pkg/sfu/connectionquality/scorer.go +++ b/pkg/sfu/connectionquality/scorer.go @@ -38,10 +38,21 @@ type windowStat struct { jitterMax float64 } -func (w *windowStat) calculatePacketScore(plw float64) float64 { +func (w *windowStat) calculatePacketScore(plw float64, isDependentRTT bool, isDependentJitter bool) float64 { // this is based on simplified E-model based on packet loss, rtt, jitter as // outlined at https://www.pingman.com/kb/article/how-is-mos-calculated-in-pingplotter-pro-50.html. - effectiveDelay := (float64(w.rttMax) / 2.0) + ((w.jitterMax * 2.0) / 1000.0) + effectiveDelay := 0.0 + // discount the dependent factors if dependency indicated. + // for example, + // 1. in the up stream, RTT cannot be measured without RTCP-XR, it is using down stream RTT. + // 2. in the down stream, up stream jitter affects it. although jitter can be adjusted to account for up stream + // jitter, this lever can be used to discount jitter in scoring. + if !isDependentRTT { + effectiveDelay += float64(w.rttMax) / 2.0 + } + if !isDependentJitter { + effectiveDelay += (w.jitterMax * 2.0) / 1000.0 + } delayEffect := effectiveDelay / 40.0 if effectiveDelay > 160.0 { delayEffect = (effectiveDelay - 120.0) / 10.0 @@ -114,8 +125,10 @@ type layerTransition struct { } type qualityScorerParams struct { - PacketLossWeight float64 - Logger logger.Logger + PacketLossWeight float64 + IsDependentRTT bool + IsDependentJitter bool + Logger logger.Logger } type qualityScorer struct { @@ -245,7 +258,7 @@ func (q *qualityScorer) Update(stat *windowStat, at time.Time) { reason = "dry" score = poorScore } else { - packetScore := stat.calculatePacketScore(q.getPacketLossWeight(stat)) + packetScore := stat.calculatePacketScore(q.getPacketLossWeight(stat), q.params.IsDependentRTT, q.params.IsDependentJitter) bitrateScore := stat.calculateBitrateScore(expectedBitrate) layerScore := math.Max(math.Min(maxScore, maxScore-(expectedDistance*distanceWeight)), 0.0) diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index 79abe27cb..1191bf5e0 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -271,10 +271,11 @@ func NewDownTrack( }) d.connectionStats = connectionquality.NewConnectionStats(connectionquality.ConnectionStatsParams{ - MimeType: codecs[0].MimeType, // LK-TODO have to notify on codec change - IsFECEnabled: strings.EqualFold(codecs[0].MimeType, webrtc.MimeTypeOpus) && strings.Contains(strings.ToLower(codecs[0].SDPFmtpLine), "fec"), - GetDeltaStats: d.getDeltaStats, - Logger: d.logger, + MimeType: codecs[0].MimeType, // LK-TODO have to notify on codec change + IsFECEnabled: strings.EqualFold(codecs[0].MimeType, webrtc.MimeTypeOpus) && strings.Contains(strings.ToLower(codecs[0].SDPFmtpLine), "fec"), + IsDependentJitter: true, + GetDeltaStats: d.getDeltaStats, + Logger: d.logger.WithValues("direction", "down"), }) d.connectionStats.OnStatsUpdate(func(_cs *connectionquality.ConnectionStats, stat *livekit.AnalyticsStat) { if d.onStatsUpdate != nil { diff --git a/pkg/sfu/receiver.go b/pkg/sfu/receiver.go index bc5ea1705..05d7303e2 100644 --- a/pkg/sfu/receiver.go +++ b/pkg/sfu/receiver.go @@ -204,10 +204,11 @@ func NewWebRTCReceiver( }) w.connectionStats = connectionquality.NewConnectionStats(connectionquality.ConnectionStatsParams{ - MimeType: w.codec.MimeType, - IsFECEnabled: strings.EqualFold(w.codec.MimeType, webrtc.MimeTypeOpus) && strings.Contains(strings.ToLower(w.codec.SDPFmtpLine), "fec"), - GetDeltaStats: w.getDeltaStats, - Logger: w.logger, + MimeType: w.codec.MimeType, + IsFECEnabled: strings.EqualFold(w.codec.MimeType, webrtc.MimeTypeOpus) && strings.Contains(strings.ToLower(w.codec.SDPFmtpLine), "fec"), + IsDependentRTT: true, + GetDeltaStats: w.getDeltaStats, + Logger: w.logger.WithValues("direction", "up"), }) w.connectionStats.OnStatsUpdate(func(_cs *connectionquality.ConnectionStats, stat *livekit.AnalyticsStat) { if w.onStatsUpdate != nil {