From 5f76d1adccd8fa72a8109f7dcaf283266a819905 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Mon, 27 Nov 2023 23:06:53 +0530 Subject: [PATCH] Introduce `DISCONNECTED` connection quality. (#2265) * Introduce `DISCONNECTED` connection quality. Currently, this state happens when any up stream track does not send any packets in an analysis window when it is expected to send packets. This can be used by participants to know the quality of a potentially disconnected participant. Previously, it took 20 - 30 seconds for the stale timeout to kick in and disconnect the limbo participant which triggered a participant update through which other participants knew about it. Previously, `POOR` quality was also overloaded to denote that the up stream is not sending any packets. With this change, that is a separate indicator, i. e. `DISCONNECTED`. * clean up * Update deps * spelling --- go.mod | 6 +-- go.sum | 12 ++--- pkg/rtc/participant.go | 12 ++--- .../connectionquality/connectionstats_test.go | 30 +++++++++-- pkg/sfu/connectionquality/scorer.go | 53 +++++++++++-------- 5 files changed, 71 insertions(+), 42 deletions(-) diff --git a/go.mod b/go.mod index c6af6151f..52241c921 100644 --- a/go.mod +++ b/go.mod @@ -18,7 +18,7 @@ require ( github.com/jxskiss/base62 v1.1.0 github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1 github.com/livekit/mediatransportutil v0.0.0-20231017082622-43f077b4e60e - github.com/livekit/protocol v1.9.2-0.20231126171322-3f612979d8c9 + github.com/livekit/protocol v1.9.2-0.20231127172639-1dda6d8eb064 github.com/livekit/psrpc v0.5.2 github.com/mackerelio/go-osstat v0.2.4 github.com/magefile/mage v1.15.0 @@ -95,10 +95,10 @@ require ( github.com/zeebo/xxh3 v1.0.2 // indirect go.uber.org/multierr v1.11.0 // indirect go.uber.org/zap v1.26.0 // indirect - golang.org/x/crypto v0.15.0 // indirect + golang.org/x/crypto v0.16.0 // indirect golang.org/x/mod v0.14.0 // indirect golang.org/x/net v0.18.0 // indirect - golang.org/x/sys v0.14.0 // indirect + golang.org/x/sys v0.15.0 // indirect golang.org/x/text v0.14.0 // indirect golang.org/x/tools v0.15.0 // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20231120223509-83a465c0220f // indirect diff --git a/go.sum b/go.sum index e0ede4f38..caa11b8e3 100644 --- a/go.sum +++ b/go.sum @@ -125,8 +125,8 @@ github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1 h1:jm09419p0lqTkD github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1/go.mod h1:Rs3MhFwutWhGwmY1VQsygw28z5bWcnEYmS1OG9OxjOQ= github.com/livekit/mediatransportutil v0.0.0-20231017082622-43f077b4e60e h1:yNeIo7MSMUWgoLu7LkNKnBYnJBFPFH9Wq4S6h1kS44M= github.com/livekit/mediatransportutil v0.0.0-20231017082622-43f077b4e60e/go.mod h1:+WIOYwiBMive5T81V8B2wdAc2zQNRjNQiJIcPxMTILY= -github.com/livekit/protocol v1.9.2-0.20231126171322-3f612979d8c9 h1:7/S8gXZCWbtx46xHdYSBg4XqnMVpAt+YfHMTZYmWzkQ= -github.com/livekit/protocol v1.9.2-0.20231126171322-3f612979d8c9/go.mod h1:8f342d5nvfNp9YAEfJokSR+zbNFpaivgU0h6vwaYhes= +github.com/livekit/protocol v1.9.2-0.20231127172639-1dda6d8eb064 h1:sse1bDW+/I3vnuw003u+M6TjpiIER9d9KsRWXn7xtRY= +github.com/livekit/protocol v1.9.2-0.20231127172639-1dda6d8eb064/go.mod h1:8f342d5nvfNp9YAEfJokSR+zbNFpaivgU0h6vwaYhes= github.com/livekit/psrpc v0.5.2 h1:+MvG8Otm/J6MTg2MP/uuMbrkxOWsrj2hDhu/I1VIU1U= github.com/livekit/psrpc v0.5.2/go.mod h1:cQjxg1oCxYHhxxv6KJH1gSvdtCHQoRZCHgPdm5N8v2g= github.com/mackerelio/go-osstat v0.2.4 h1:qxGbdPkFo65PXOb/F/nhDKpF2nGmGaCFDLXoZjJTtUs= @@ -291,8 +291,8 @@ golang.org/x/crypto v0.8.0/go.mod h1:mRqEX+O9/h5TFCrQhkgjo2yKi0yYA+9ecGkdQoHrywE golang.org/x/crypto v0.10.0/go.mod h1:o4eNf7Ede1fv+hwOwZsTHl9EsPFO6q6ZvYR8vYfY45I= golang.org/x/crypto v0.11.0/go.mod h1:xgJhtzW8F9jGdVFWZESrid1U1bjeNy4zgy5cRr/CIio= golang.org/x/crypto v0.12.0/go.mod h1:NF0Gs7EO5K4qLn+Ylc+fih8BSTeIjAP05siRnAh98yw= -golang.org/x/crypto v0.15.0 h1:frVn1TEaCEaZcn3Tmd7Y2b5KKPaZ+I32Q2OA3kYp5TA= -golang.org/x/crypto v0.15.0/go.mod h1:4ChreQoLWfG3xLDer1WdlH5NdlQ3+mwnQq1YTKY+72g= +golang.org/x/crypto v0.16.0 h1:mMMrFzRSCF0GvB7Ne27XVtVAaXLrPmgPC7/v0tkwHaY= +golang.org/x/crypto v0.16.0/go.mod h1:gCAAfMLgwOJRpTjQ2zCCt2OcSfYMTeZVSRtQlPC7Nq4= golang.org/x/exp v0.0.0-20231110203233-9a3e6036ecaa h1:FRnLl4eNAQl8hwxVVC17teOw8kdjVDVAiFMtgUdTSRQ= golang.org/x/exp v0.0.0-20231110203233-9a3e6036ecaa/go.mod h1:zk2irFbV9DP96SEBUUAy67IdHUaZuSnrz1n472HUCLE= golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= @@ -378,8 +378,8 @@ golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.9.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.10.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.11.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.14.0 h1:Vz7Qs629MkJkGyHxUlRHizWJRG2j8fbQKjELVSNhy7Q= -golang.org/x/sys v0.14.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= +golang.org/x/sys v0.15.0 h1:h48lPFYpsTvQJZF4EKyI4aLHaev3CxivZmv7yZig9pc= +golang.org/x/sys v0.15.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8= golang.org/x/term v0.1.0/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8= diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index ca1d0c9f2..8362932b7 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -931,8 +931,7 @@ func (p *ParticipantImpl) GetConnectionQuality() *livekit.ConnectionQualityInfo numTracks++ score, quality := pt.(types.LocalMediaTrack).GetConnectionScoreAndQuality() - if quality < minQuality { - // WARNING NOTE: comparing protobuf enums directly + if utils.IsConnectionQualityLower(minQuality, quality) { minQuality = quality minScore = score } else if quality == minQuality && score < minScore { @@ -942,8 +941,7 @@ func (p *ParticipantImpl) GetConnectionQuality() *livekit.ConnectionQualityInfo p.lock.Lock() trackID := pt.ID() if prevQuality, ok := p.tracksQuality[trackID]; ok { - // WARNING NOTE: comparing protobuf enums directly - if prevQuality > quality { + if utils.IsConnectionQualityLower(prevQuality, quality) { numUpDrops++ } } @@ -958,8 +956,7 @@ func (p *ParticipantImpl) GetConnectionQuality() *livekit.ConnectionQualityInfo numTracks++ score, quality := subTrack.DownTrack().GetConnectionScoreAndQuality() - if quality < minQuality { - // WARNING NOTE: comparing protobuf enums directly + if utils.IsConnectionQualityLower(minQuality, quality) { minQuality = quality minScore = score } else if quality == minQuality && score < minScore { @@ -969,8 +966,7 @@ func (p *ParticipantImpl) GetConnectionQuality() *livekit.ConnectionQualityInfo p.lock.Lock() trackID := subTrack.ID() if prevQuality, ok := p.tracksQuality[trackID]; ok { - // WARNING NOTE: comparing protobuf enums directly - if prevQuality > quality { + if utils.IsConnectionQualityLower(prevQuality, quality) { numDownDrops++ } } diff --git a/pkg/sfu/connectionquality/connectionstats_test.go b/pkg/sfu/connectionquality/connectionstats_test.go index 73c9c649b..5af316ac2 100644 --- a/pkg/sfu/connectionquality/connectionstats_test.go +++ b/pkg/sfu/connectionquality/connectionstats_test.go @@ -259,7 +259,7 @@ func TestConnectionQuality(t *testing.T) { require.Greater(t, float32(4.6), mos) require.Equal(t, livekit.ConnectionQuality_EXCELLENT, quality) - // next update with no packets should knock quality down + // next update with no packets should knock quality down to DISCONNECTED now = now.Add(duration) trp.setStreams(map[uint32]*buffer.StreamStatsWithLayers{ 1: { @@ -273,12 +273,36 @@ func TestConnectionQuality(t *testing.T) { cs.updateScoreAt(now.Add(duration)) mos, quality = cs.GetScoreAndQuality() require.Greater(t, float32(2.1), mos) - require.Equal(t, livekit.ConnectionQuality_POOR, quality) + require.Equal(t, livekit.ConnectionQuality_DISCONNECTED, quality) - // mute/unmute to bring quality back up + // mute when DISCONNECTED should not bump up score/quality now = now.Add(duration) cs.UpdateMuteAt(true, now.Add(1*time.Second)) + mos, quality = cs.GetScoreAndQuality() + require.Greater(t, float32(2.1), mos) + require.Equal(t, livekit.ConnectionQuality_DISCONNECTED, quality) + + // unmute and send packets to bring quality back up + now = now.Add(duration) cs.UpdateMuteAt(false, now.Add(2*time.Second)) + for i := 0; i < 3; i++ { + trp.setStreams(map[uint32]*buffer.StreamStatsWithLayers{ + 1: { + RTPStats: &buffer.RTPDeltaInfo{ + StartTime: now, + Duration: duration, + Packets: 250, + PacketsLost: 0, + }, + }, + }) + cs.updateScoreAt(now.Add(duration)) + now = now.Add(duration) + } + cs.updateScoreAt(now.Add(duration)) + mos, quality = cs.GetScoreAndQuality() + require.Greater(t, float32(4.6), mos) + require.Equal(t, livekit.ConnectionQuality_EXCELLENT, quality) // with lesser number of packet (simulating DTX). // even higher loss (like 10%) should not knock down quality due to quadratic weighting of packet loss ratio diff --git a/pkg/sfu/connectionquality/scorer.go b/pkg/sfu/connectionquality/scorer.go index 340f32368..d2d1f4e5b 100644 --- a/pkg/sfu/connectionquality/scorer.go +++ b/pkg/sfu/connectionquality/scorer.go @@ -29,9 +29,10 @@ const ( MaxMOS = float32(4.5) MinMOS = float32(1.0) - maxScore = float64(100.0) - poorScore = float64(30.0) - minScore = float64(20.0) + cMaxScore = float64(100.0) + cPoorScore = float64(30.0) + cMinScore = float64(20.0) + cDisconnectedScore = cMinScore increaseFactor = float64(0.4) // slower increase, i. e. when score is recovering move up slower -> conservative decreaseFactor = float64(0.7) // faster decrease, i. e. when score is dropping move down faster -> aggressive to be responsive to quality drops @@ -107,7 +108,7 @@ func (w *windowStat) calculatePacketScore(plw float64, includeRTT bool, includeJ } lossEffect *= plw - score := maxScore - delayEffect - lossEffect + score := cMaxScore - delayEffect - lossEffect if score < 0.0 { score = 0.0 } @@ -118,7 +119,7 @@ func (w *windowStat) calculatePacketScore(plw float64, includeRTT bool, includeJ func (w *windowStat) calculateBitrateScore(expectedBitrate int64) float64 { if expectedBitrate == 0 { // unsupported mode OR all layers stopped - return maxScore + return cMaxScore } var score float64 @@ -126,9 +127,9 @@ func (w *windowStat) calculateBitrateScore(expectedBitrate int64) float64 { // using the ratio of expectedBitrate / actualBitrate // the quality inflection points are approximately // GOOD at ~2.7x, POOR at ~20.1x - score = maxScore - 20*math.Log(float64(expectedBitrate)/float64(w.bytes*8)) - if score > maxScore { - score = maxScore + score = cMaxScore - 20*math.Log(float64(expectedBitrate)/float64(w.bytes*8)) + if score > cMaxScore { + score = cMaxScore } if score < 0.0 { score = 0.0 @@ -188,7 +189,7 @@ type qualityScorer struct { func newQualityScorer(params qualityScorerParams) *qualityScorer { return &qualityScorer{ params: params, - score: maxScore, + score: cMaxScore, aggregateBitrate: utils.NewTimedAggregator[int64](utils.TimedAggregatorParams{ CapNegativeValues: true, }), @@ -219,7 +220,10 @@ func (q *qualityScorer) Start() { func (q *qualityScorer) updateMuteAtLocked(isMuted bool, at time.Time) { if isMuted { q.mutedAt = at - q.score = maxScore + // muting when DISCONNECTED should not push quality to EXCELLENT + if q.score != cDisconnectedScore { + q.score = cMaxScore + } } else { q.unmutedAt = at } @@ -264,7 +268,7 @@ func (q *qualityScorer) updateLayerMuteAtLocked(isMuted bool, at time.Time) { q.layerDistance.Reset() q.layerMutedAt = at - q.score = maxScore + q.score = cMaxScore } } else { if q.isLayerMuted() { @@ -294,7 +298,7 @@ func (q *qualityScorer) updatePauseAtLocked(isPaused bool, at time.Time) { q.layerDistance.Reset() q.pausedAt = at - q.score = poorScore + q.score = cPoorScore } } else { if q.isPaused() { @@ -353,7 +357,7 @@ func (q *qualityScorer) updateAtLocked(stat *windowStat, at time.Time) { // considered (as long as enough time has passed since unmute). // // Similarly, when paused (possibly due to congestion), score is immediately - // set to poorScore for responsiveness. The layer transision is reest. + // set to cPoorScore for responsiveness. The layer transision is reest. // On a resume, quality climbs back up using normal operation. if q.isMuted() || !q.isUnmutedEnough(at) || q.isLayerMuted() || q.isPaused() { q.lastUpdateAt = at @@ -365,11 +369,11 @@ func (q *qualityScorer) updateAtLocked(stat *windowStat, at time.Time) { var score float64 if stat.packetsExpected == 0 { reason = "dry" - score = poorScore + score = cDisconnectedScore } else { packetScore := stat.calculatePacketScore(plw, q.params.IncludeRTT, q.params.IncludeJitter) bitrateScore := stat.calculateBitrateScore(expectedBitrate) - layerScore := math.Max(math.Min(maxScore, maxScore-(expectedDistance*distanceWeight)), 0.0) + layerScore := math.Max(math.Min(cMaxScore, cMaxScore-(expectedDistance*distanceWeight)), 0.0) minScore := math.Min(packetScore, bitrateScore) minScore = math.Min(minScore, layerScore) @@ -394,22 +398,23 @@ func (q *qualityScorer) updateAtLocked(stat *windowStat, at time.Time) { } score = factor*score + (1.0-factor)*q.score } - if score < minScore { + if score < cMinScore { // lower bound to prevent score from becoming very small values due to extreme conditions. // Without a lower bound, it can get so low that it takes a long time to climb back to // better quality even under excellent conditions. - score = minScore + score = cMinScore } - // WARNING NOTE: comparing protobuf enum values directly (livekit.ConnectionQuality) - if scoreToConnectionQuality(q.score) > scoreToConnectionQuality(score) { + prevCQ := scoreToConnectionQuality(q.score) + currCQ := scoreToConnectionQuality(score) + if utils.IsConnectionQualityLower(prevCQ, currCQ) { q.params.Logger.Infow( "quality drop", "reason", reason, "prevScore", q.score, - "prevQuality", scoreToConnectionQuality(q.score), + "prevQuality", prevCQ, "prevStat", &q.stat, "score", score, - "quality", scoreToConnectionQuality(score), + "quality", currCQ, "stat", stat, "packetLossWeight", plw, "maxPPS", q.maxPPS, @@ -531,7 +536,11 @@ func scoreToConnectionQuality(score float64) livekit.ConnectionQuality { return livekit.ConnectionQuality_GOOD } - return livekit.ConnectionQuality_POOR + if score > 20.0 { + return livekit.ConnectionQuality_POOR + } + + return livekit.ConnectionQuality_DISCONNECTED } // ------------------------------------------