dandri/livekit
1
0 Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-10-22 11:02:27 +00:00
Code Issues Packages Projects Releases Wiki Activity

Introduce DISCONNECTED connection quality. (#2265)

Browse Source 
* 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
This commit is contained in:
Raja Subramanian authored and GitHub committed 2023-11-27 23:06:53 +05:30
1 parent 53542b09a0
commit 5f76d1adcc
5 files changed
+71 -42
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

go.mod
+3 -3
View File
@@ -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
go.sum
+6 -6
View File
@@ -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=
pkg/rtc/participant.go
+4 -8
View File
@@ -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++
}
}
pkg/sfu/connectionquality/connectionstats_test.go
+27 -3
View File
@@ -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
pkg/sfu/connectionquality/scorer.go
+31 -22
View File
@@ -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
}
// ------------------------------------------