diff --git a/pkg/rtc/mediatrack.go b/pkg/rtc/mediatrack.go index b7795a93d..f15a45c9f 100644 --- a/pkg/rtc/mediatrack.go +++ b/pkg/rtc/mediatrack.go @@ -3,6 +3,7 @@ package rtc import ( "context" "errors" + "github.com/livekit/livekit-server/pkg/sfu/connectionquality" "sort" "sync" "sync/atomic" @@ -31,8 +32,9 @@ var ( ) const ( - lostUpdateDelta = time.Second - layerSelectionTolerance = 0.8 + lostUpdateDelta = time.Second + connectionQualityUpdateInterval = 5 * time.Second + layerSelectionTolerance = 0.8 ) // MediaTrack represents a WebRTC track that needs to be forwarded @@ -58,13 +60,15 @@ type MediaTrack struct { layerDimensions sync.Map // quality => *livekit.VideoLayer // track audio fraction lost - fracLostLock sync.Mutex + statsLock sync.Mutex maxDownFracLost uint8 maxDownFracLostTs time.Time currentUpFracLost uint32 maxUpFracLost uint8 maxUpFracLostTs time.Time + connectionStats *connectionquality.ConnectionStats + done chan struct{} // quality level enable/disable maxQualityLock sync.Mutex maxSubscriberQuality map[string]livekit.VideoQuality @@ -95,6 +99,8 @@ func NewMediaTrack(track *webrtc.TrackRemote, params MediaTrackParams) *MediaTra ssrc: track.SSRC(), streamID: track.StreamID(), codec: track.Codec(), + connectionStats: connectionquality.NewConnectionStats(), + done: make(chan struct{}), maxSubscriberQuality: make(map[string]livekit.VideoQuality), } @@ -106,6 +112,10 @@ func NewMediaTrack(track *webrtc.TrackRemote, params MediaTrackParams) *MediaTra t.UpdateVideoLayers(params.TrackInfo.Layers) // LK-TODO: maybe use this or simulcast flag in TrackInfo to set simulcasted here } + // on close signal via closing channel to workers + t.AddOnClose(t.closeChan) + go t.updateStats() + return t } @@ -576,6 +586,11 @@ func (t *MediaTrack) sendDownTrackBindingReports(sub types.Participant) { 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 + for _, p := range packets { switch pkt := p.(type) { // sfu.Buffer generates ReceiverReports for the publisher @@ -584,13 +599,26 @@ func (t *MediaTrack) handlePublisherFeedback(packets []rtcp.Packet) { 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 } } } if hasReport { - t.fracLostLock.Lock() + t.statsLock.Lock() if maxLost > t.maxUpFracLost { t.maxUpFracLost = maxLost } @@ -601,7 +629,19 @@ func (t *MediaTrack) handlePublisherFeedback(packets []rtcp.Packet) { t.maxUpFracLost = 0 t.maxUpFracLostTs = now } - t.fracLostLock.Unlock() + // update feedback stats + current := t.connectionStats.Curr + if jitter > current.Jitter { + current.Jitter = jitter + } + if delay > current.Delay { + current.Delay = delay + } + if maxSeqNum > current.LastSeqNum { + current.LastSeqNum = maxSeqNum + } + current.PacketsLost = totalLost + t.statsLock.Unlock() } // also look for sender reports @@ -610,12 +650,13 @@ func (t *MediaTrack) handlePublisherFeedback(packets []rtcp.Packet) { } // handles max loss for audio packets -func (t *MediaTrack) handleMaxLossFeedback(_ *sfu.DownTrack, report *rtcp.ReceiverReport) { +func (t *MediaTrack) handleMaxLossFeedback(_ *sfu.DownTrack, + report *rtcp.ReceiverReport) { var ( shouldUpdate bool maxLost uint8 ) - t.fracLostLock.Lock() + t.statsLock.Lock() for _, rr := range report.Reports { if t.maxDownFracLost < rr.FractionLost { t.maxDownFracLost = rr.FractionLost @@ -629,7 +670,7 @@ func (t *MediaTrack) handleMaxLossFeedback(_ *sfu.DownTrack, report *rtcp.Receiv t.maxDownFracLost = 0 t.maxDownFracLostTs = now } - t.fracLostLock.Unlock() + t.statsLock.Unlock() if shouldUpdate && t.buffer != nil { // ok to access buffer since receivers are added before subscribers @@ -671,6 +712,42 @@ func (t *MediaTrack) Receiver() sfu.TrackReceiver { return t.receiver } +func (t *MediaTrack) GetConnectionScore() float64 { + t.statsLock.Lock() + defer t.statsLock.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.statsLock.Lock() + if t.Kind() == livekit.TrackType_AUDIO { + t.connectionStats.CalculateAudioScore() + } else { + t.calculateVideoScore() + } + t.statsLock.Unlock() + } + } +} + +func (t *MediaTrack) calculateVideoScore() { + var reducedQuality bool + publishing, registered := t.NumUpTracks() + if registered > 0 && publishing != registered { + reducedQuality = true + } + t.connectionStats.Score = connectionquality.Loss2Score(t.PublishLossPercentage(), reducedQuality) +} + func (t *MediaTrack) OnSubscribedMaxQualityChange(f func(trackSid string, subscribedQualities []*livekit.SubscribedQuality) error) { t.onSubscribedMaxQualityChange = f } diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 201e52896..63cdd603f 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -2,6 +2,7 @@ package rtc import ( "fmt" + "github.com/livekit/livekit-server/pkg/sfu/connectionquality" "io" "strings" "sync" @@ -665,50 +666,40 @@ func (p *ParticipantImpl) GetAudioLevel() (level uint8, active bool) { return } -func (p *ParticipantImpl) GetConnectionQuality() livekit.ConnectionQuality { +func (p *ParticipantImpl) GetConnectionQuality() *livekit.ConnectionQualityInfo { // avg loss across all tracks, weigh published the same as subscribed - var pubLoss, subLoss uint32 - var reducedQualityPub bool - var reducedQualitySub bool + var scores float64 + var numTracks int p.lock.RLock() defer p.lock.RUnlock() for _, pubTrack := range p.publishedTracks { if pubTrack.IsMuted() { continue } - pubLoss += pubTrack.PublishLossPercentage() - publishing, registered := pubTrack.NumUpTracks() - if registered > 0 && publishing != registered { - reducedQualityPub = true - } - } - numTracks := uint32(len(p.publishedTracks)) - if numTracks > 0 { - pubLoss /= numTracks + scores += pubTrack.GetConnectionScore() + numTracks++ } for _, subTrack := range p.subscribedTracks { if subTrack.IsMuted() { continue } - if subTrack.DownTrack().GetForwardingStatus() != sfu.ForwardingStatusOptimal { - reducedQualitySub = true - } - subLoss += subTrack.SubscribeLossPercentage() + scores += subTrack.DownTrack().GetConnectionScore() + numTracks++ } - numTracks = uint32(len(p.subscribedTracks)) + + var avgScore float64 if numTracks > 0 { - subLoss /= numTracks + avgScore = scores / float64(numTracks) } - avgLoss := (pubLoss + subLoss) / 2 - if avgLoss >= 4 { - return livekit.ConnectionQuality_POOR - } else if avgLoss <= 2 && !reducedQualityPub && !reducedQualitySub { - return livekit.ConnectionQuality_EXCELLENT - } + rating := connectionquality.Score2Rating(avgScore) - return livekit.ConnectionQuality_GOOD + return &livekit.ConnectionQualityInfo{ + ParticipantSid: p.ID(), + Quality: rating, + Score: float32(avgScore), + } } func (p *ParticipantImpl) IsSubscribedTo(identity string) bool { diff --git a/pkg/rtc/participant_internal_test.go b/pkg/rtc/participant_internal_test.go index 3131c1415..8dbaf2dad 100644 --- a/pkg/rtc/participant_internal_test.go +++ b/pkg/rtc/participant_internal_test.go @@ -1,6 +1,7 @@ package rtc import ( + "github.com/livekit/livekit-server/pkg/sfu/connectionquality" "testing" "time" @@ -245,11 +246,40 @@ func TestMuteSetting(t *testing.T) { } func TestConnectionQuality(t *testing.T) { - testPublishedTrack := func(loss, numPublishing, numRegistered uint32) *typesfakes.FakePublishedTrack { - t := &typesfakes.FakePublishedTrack{} - t.PublishLossPercentageReturns(loss) - t.NumUpTracksReturns(numPublishing, numRegistered) - return t + + // loss based score is currently a publisher method. + videoScore := func(loss, numPublishing, numRegistered uint32) float64 { + var reducedQuality bool + if numRegistered > 0 && numPublishing != numRegistered { + reducedQuality = true + } + return connectionquality.Loss2Score(loss, reducedQuality) + } + + testPublishedVideoTrack := func(loss, numPublishing, numRegistered uint32) *typesfakes.FakePublishedTrack { + tr := &typesfakes.FakePublishedTrack{} + score := videoScore(loss, numPublishing, numRegistered) + t.Log("video score: ", score) + tr.GetConnectionScoreReturns(score) + return tr + } + + testPublishedAudioTrack := func(totalPackets, packetsLost uint32) *typesfakes.FakePublishedTrack { + tr := &typesfakes.FakePublishedTrack{} + + stat := &connectionquality.ConnectionStat{ + PacketsLost: packetsLost, + TotalPackets: totalPackets, + LastSeqNum: 0, + } + stats := &connectionquality.ConnectionStats{ + Curr: stat, + Prev: &connectionquality.ConnectionStat{}, + } + stats.CalculateAudioScore() + t.Log("audio score: ", stats.Score) + tr.GetConnectionScoreReturns(stats.Score) + return tr } // TODO: this test is rather limited since we cannot mock DownTrack's Target & Max spatial layers @@ -257,18 +287,32 @@ func TestConnectionQuality(t *testing.T) { t.Run("smooth sailing", func(t *testing.T) { p := newParticipantForTest("test") - p.publishedTracks["video"] = testPublishedTrack(1, 3, 3) - p.publishedTracks["audio"] = testPublishedTrack(0, 1, 1) + p.publishedTracks["video"] = testPublishedVideoTrack(2, 3, 3) + p.publishedTracks["audio"] = testPublishedAudioTrack(1000, 0) - require.Equal(t, livekit.ConnectionQuality_EXCELLENT, p.GetConnectionQuality()) + require.Equal(t, livekit.ConnectionQuality_EXCELLENT, p.GetConnectionQuality().GetQuality()) }) t.Run("reduced publishing", func(t *testing.T) { p := newParticipantForTest("test") - p.publishedTracks["video"] = testPublishedTrack(3, 2, 3) - p.publishedTracks["audio"] = testPublishedTrack(3, 1, 1) + p.publishedTracks["video"] = testPublishedVideoTrack(3, 2, 3) + p.publishedTracks["audio"] = testPublishedAudioTrack(1000, 100) - require.Equal(t, livekit.ConnectionQuality_GOOD, p.GetConnectionQuality()) + require.Equal(t, livekit.ConnectionQuality_GOOD, p.GetConnectionQuality().GetQuality()) + }) + + t.Run("audio smooth publishing", func(t *testing.T) { + p := newParticipantForTest("test") + p.publishedTracks["audio"] = testPublishedAudioTrack(1000, 10) + + require.Equal(t, livekit.ConnectionQuality_EXCELLENT, p.GetConnectionQuality().GetQuality()) + }) + + t.Run("audio reduced publishing", func(t *testing.T) { + p := newParticipantForTest("test") + p.publishedTracks["audio"] = testPublishedAudioTrack(1000, 100) + + require.Equal(t, livekit.ConnectionQuality_GOOD, p.GetConnectionQuality().GetQuality()) }) } diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index da77cca15..ccb977f7a 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -735,10 +735,7 @@ func (r *Room) connectionQualityWorker() { connectionInfos := make(map[string]*livekit.ConnectionQualityInfo, len(participants)) for _, p := range participants { - connectionInfos[p.Identity()] = &livekit.ConnectionQualityInfo{ - ParticipantSid: p.ID(), - Quality: p.GetConnectionQuality(), - } + connectionInfos[p.Identity()] = p.GetConnectionQuality() } for _, op := range participants { diff --git a/pkg/rtc/subscribedtrack.go b/pkg/rtc/subscribedtrack.go index f6dfdd3dd..969487b3c 100644 --- a/pkg/rtc/subscribedtrack.go +++ b/pkg/rtc/subscribedtrack.go @@ -67,10 +67,6 @@ func (t *SubscribedTrack) PublishedTrack() types.MediaTrack { return t.publishedTrack } -func (t *SubscribedTrack) SubscribeLossPercentage() uint32 { - return FixedPointToPercent(t.DownTrack().CurrentMaxLossFraction()) -} - // has subscriber indicated it wants to mute this track func (t *SubscribedTrack) IsMuted() bool { return t.subMuted.Get() diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index 6b70523a1..e42c595c5 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -55,7 +55,7 @@ type Participant interface { SendConnectionQualityUpdate(update *livekit.ConnectionQualityUpdate) error SetTrackMuted(trackId string, muted bool, fromAdmin bool) GetAudioLevel() (level uint8, active bool) - GetConnectionQuality() livekit.ConnectionQuality + GetConnectionQuality() *livekit.ConnectionQualityInfo IsSubscribedTo(identity string) bool // returns list of participant identities that the current participant is subscribed to GetSubscribedParticipants() []string @@ -117,6 +117,7 @@ type MediaTrack interface { // returns quality information that's appropriate for width & height GetQualityForDimension(width, height uint32) livekit.VideoQuality + GetConnectionScore() float64 NotifySubscriberMute(subscriberID string) NotifySubscriberMaxQuality(subscriberID string, quality livekit.VideoQuality) OnSubscribedMaxQualityChange(f func(trackSid string, subscribedQualities []*livekit.SubscribedQuality) error) @@ -153,7 +154,6 @@ type SubscribedTrack interface { UpdateSubscriberSettings(settings *livekit.UpdateTrackSettings) // selects appropriate video layer according to subscriber preferences UpdateVideoLayer() - SubscribeLossPercentage() uint32 } // interface for properties of webrtc.TrackRemote diff --git a/pkg/rtc/types/typesfakes/fake_media_track.go b/pkg/rtc/types/typesfakes/fake_media_track.go index 4f41fba30..d01efa709 100644 --- a/pkg/rtc/types/typesfakes/fake_media_track.go +++ b/pkg/rtc/types/typesfakes/fake_media_track.go @@ -20,6 +20,16 @@ type FakeMediaTrack struct { addSubscriberReturnsOnCall map[int]struct { result1 error } + GetConnectionScoreStub func() float64 + getConnectionScoreMutex sync.RWMutex + getConnectionScoreArgsForCall []struct { + } + getConnectionScoreReturns struct { + result1 float64 + } + getConnectionScoreReturnsOnCall map[int]struct { + result1 float64 + } GetQualityForDimensionStub func(uint32, uint32) livekit.VideoQuality getQualityForDimensionMutex sync.RWMutex getQualityForDimensionArgsForCall []struct { @@ -203,6 +213,59 @@ func (fake *FakeMediaTrack) AddSubscriberReturnsOnCall(i int, result1 error) { }{result1} } +func (fake *FakeMediaTrack) GetConnectionScore() float64 { + fake.getConnectionScoreMutex.Lock() + ret, specificReturn := fake.getConnectionScoreReturnsOnCall[len(fake.getConnectionScoreArgsForCall)] + fake.getConnectionScoreArgsForCall = append(fake.getConnectionScoreArgsForCall, struct { + }{}) + stub := fake.GetConnectionScoreStub + fakeReturns := fake.getConnectionScoreReturns + fake.recordInvocation("GetConnectionScore", []interface{}{}) + fake.getConnectionScoreMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeMediaTrack) GetConnectionScoreCallCount() int { + fake.getConnectionScoreMutex.RLock() + defer fake.getConnectionScoreMutex.RUnlock() + return len(fake.getConnectionScoreArgsForCall) +} + +func (fake *FakeMediaTrack) GetConnectionScoreCalls(stub func() float64) { + fake.getConnectionScoreMutex.Lock() + defer fake.getConnectionScoreMutex.Unlock() + fake.GetConnectionScoreStub = stub +} + +func (fake *FakeMediaTrack) GetConnectionScoreReturns(result1 float64) { + fake.getConnectionScoreMutex.Lock() + defer fake.getConnectionScoreMutex.Unlock() + fake.GetConnectionScoreStub = nil + fake.getConnectionScoreReturns = struct { + result1 float64 + }{result1} +} + +func (fake *FakeMediaTrack) GetConnectionScoreReturnsOnCall(i int, result1 float64) { + fake.getConnectionScoreMutex.Lock() + defer fake.getConnectionScoreMutex.Unlock() + fake.GetConnectionScoreStub = nil + if fake.getConnectionScoreReturnsOnCall == nil { + fake.getConnectionScoreReturnsOnCall = make(map[int]struct { + result1 float64 + }) + } + fake.getConnectionScoreReturnsOnCall[i] = struct { + result1 float64 + }{result1} +} + func (fake *FakeMediaTrack) GetQualityForDimension(arg1 uint32, arg2 uint32) livekit.VideoQuality { fake.getQualityForDimensionMutex.Lock() ret, specificReturn := fake.getQualityForDimensionReturnsOnCall[len(fake.getQualityForDimensionArgsForCall)] @@ -871,6 +934,8 @@ func (fake *FakeMediaTrack) Invocations() map[string][][]interface{} { defer fake.invocationsMutex.RUnlock() fake.addSubscriberMutex.RLock() defer fake.addSubscriberMutex.RUnlock() + fake.getConnectionScoreMutex.RLock() + defer fake.getConnectionScoreMutex.RUnlock() fake.getQualityForDimensionMutex.RLock() defer fake.getQualityForDimensionMutex.RUnlock() fake.iDMutex.RLock() diff --git a/pkg/rtc/types/typesfakes/fake_participant.go b/pkg/rtc/types/typesfakes/fake_participant.go index b9bdccb2b..f19484918 100644 --- a/pkg/rtc/types/typesfakes/fake_participant.go +++ b/pkg/rtc/types/typesfakes/fake_participant.go @@ -120,15 +120,15 @@ type FakeParticipant struct { result1 uint8 result2 bool } - GetConnectionQualityStub func() livekit.ConnectionQuality + GetConnectionQualityStub func() *livekit.ConnectionQualityInfo getConnectionQualityMutex sync.RWMutex getConnectionQualityArgsForCall []struct { } getConnectionQualityReturns struct { - result1 livekit.ConnectionQuality + result1 *livekit.ConnectionQualityInfo } getConnectionQualityReturnsOnCall map[int]struct { - result1 livekit.ConnectionQuality + result1 *livekit.ConnectionQualityInfo } GetPublishedTrackStub func(string) types.PublishedTrack getPublishedTrackMutex sync.RWMutex @@ -1059,7 +1059,7 @@ func (fake *FakeParticipant) GetAudioLevelReturnsOnCall(i int, result1 uint8, re }{result1, result2} } -func (fake *FakeParticipant) GetConnectionQuality() livekit.ConnectionQuality { +func (fake *FakeParticipant) GetConnectionQuality() *livekit.ConnectionQualityInfo { fake.getConnectionQualityMutex.Lock() ret, specificReturn := fake.getConnectionQualityReturnsOnCall[len(fake.getConnectionQualityArgsForCall)] fake.getConnectionQualityArgsForCall = append(fake.getConnectionQualityArgsForCall, struct { @@ -1083,32 +1083,32 @@ func (fake *FakeParticipant) GetConnectionQualityCallCount() int { return len(fake.getConnectionQualityArgsForCall) } -func (fake *FakeParticipant) GetConnectionQualityCalls(stub func() livekit.ConnectionQuality) { +func (fake *FakeParticipant) GetConnectionQualityCalls(stub func() *livekit.ConnectionQualityInfo) { fake.getConnectionQualityMutex.Lock() defer fake.getConnectionQualityMutex.Unlock() fake.GetConnectionQualityStub = stub } -func (fake *FakeParticipant) GetConnectionQualityReturns(result1 livekit.ConnectionQuality) { +func (fake *FakeParticipant) GetConnectionQualityReturns(result1 *livekit.ConnectionQualityInfo) { fake.getConnectionQualityMutex.Lock() defer fake.getConnectionQualityMutex.Unlock() fake.GetConnectionQualityStub = nil fake.getConnectionQualityReturns = struct { - result1 livekit.ConnectionQuality + result1 *livekit.ConnectionQualityInfo }{result1} } -func (fake *FakeParticipant) GetConnectionQualityReturnsOnCall(i int, result1 livekit.ConnectionQuality) { +func (fake *FakeParticipant) GetConnectionQualityReturnsOnCall(i int, result1 *livekit.ConnectionQualityInfo) { fake.getConnectionQualityMutex.Lock() defer fake.getConnectionQualityMutex.Unlock() fake.GetConnectionQualityStub = nil if fake.getConnectionQualityReturnsOnCall == nil { fake.getConnectionQualityReturnsOnCall = make(map[int]struct { - result1 livekit.ConnectionQuality + result1 *livekit.ConnectionQualityInfo }) } fake.getConnectionQualityReturnsOnCall[i] = struct { - result1 livekit.ConnectionQuality + result1 *livekit.ConnectionQualityInfo }{result1} } diff --git a/pkg/rtc/types/typesfakes/fake_published_track.go b/pkg/rtc/types/typesfakes/fake_published_track.go index 414ed06d2..e8c58b7fa 100644 --- a/pkg/rtc/types/typesfakes/fake_published_track.go +++ b/pkg/rtc/types/typesfakes/fake_published_track.go @@ -26,6 +26,16 @@ type FakePublishedTrack struct { addSubscriberReturnsOnCall map[int]struct { result1 error } + GetConnectionScoreStub func() float64 + getConnectionScoreMutex sync.RWMutex + getConnectionScoreArgsForCall []struct { + } + getConnectionScoreReturns struct { + result1 float64 + } + getConnectionScoreReturnsOnCall map[int]struct { + result1 float64 + } GetQualityForDimensionStub func(uint32, uint32) livekit.VideoQuality getQualityForDimensionMutex sync.RWMutex getQualityForDimensionArgsForCall []struct { @@ -303,6 +313,59 @@ func (fake *FakePublishedTrack) AddSubscriberReturnsOnCall(i int, result1 error) }{result1} } +func (fake *FakePublishedTrack) GetConnectionScore() float64 { + fake.getConnectionScoreMutex.Lock() + ret, specificReturn := fake.getConnectionScoreReturnsOnCall[len(fake.getConnectionScoreArgsForCall)] + fake.getConnectionScoreArgsForCall = append(fake.getConnectionScoreArgsForCall, struct { + }{}) + stub := fake.GetConnectionScoreStub + fakeReturns := fake.getConnectionScoreReturns + fake.recordInvocation("GetConnectionScore", []interface{}{}) + fake.getConnectionScoreMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakePublishedTrack) GetConnectionScoreCallCount() int { + fake.getConnectionScoreMutex.RLock() + defer fake.getConnectionScoreMutex.RUnlock() + return len(fake.getConnectionScoreArgsForCall) +} + +func (fake *FakePublishedTrack) GetConnectionScoreCalls(stub func() float64) { + fake.getConnectionScoreMutex.Lock() + defer fake.getConnectionScoreMutex.Unlock() + fake.GetConnectionScoreStub = stub +} + +func (fake *FakePublishedTrack) GetConnectionScoreReturns(result1 float64) { + fake.getConnectionScoreMutex.Lock() + defer fake.getConnectionScoreMutex.Unlock() + fake.GetConnectionScoreStub = nil + fake.getConnectionScoreReturns = struct { + result1 float64 + }{result1} +} + +func (fake *FakePublishedTrack) GetConnectionScoreReturnsOnCall(i int, result1 float64) { + fake.getConnectionScoreMutex.Lock() + defer fake.getConnectionScoreMutex.Unlock() + fake.GetConnectionScoreStub = nil + if fake.getConnectionScoreReturnsOnCall == nil { + fake.getConnectionScoreReturnsOnCall = make(map[int]struct { + result1 float64 + }) + } + fake.getConnectionScoreReturnsOnCall[i] = struct { + result1 float64 + }{result1} +} + func (fake *FakePublishedTrack) GetQualityForDimension(arg1 uint32, arg2 uint32) livekit.VideoQuality { fake.getQualityForDimensionMutex.Lock() ret, specificReturn := fake.getQualityForDimensionReturnsOnCall[len(fake.getQualityForDimensionArgsForCall)] @@ -1294,6 +1357,8 @@ func (fake *FakePublishedTrack) Invocations() map[string][][]interface{} { defer fake.addOnCloseMutex.RUnlock() fake.addSubscriberMutex.RLock() defer fake.addSubscriberMutex.RUnlock() + fake.getConnectionScoreMutex.RLock() + defer fake.getConnectionScoreMutex.RUnlock() fake.getQualityForDimensionMutex.RLock() defer fake.getQualityForDimensionMutex.RUnlock() fake.iDMutex.RLock() diff --git a/pkg/rtc/types/typesfakes/fake_subscribed_track.go b/pkg/rtc/types/typesfakes/fake_subscribed_track.go index 051662e98..402074e93 100644 --- a/pkg/rtc/types/typesfakes/fake_subscribed_track.go +++ b/pkg/rtc/types/typesfakes/fake_subscribed_track.go @@ -70,16 +70,6 @@ type FakeSubscribedTrack struct { setPublisherMutedArgsForCall []struct { arg1 bool } - SubscribeLossPercentageStub func() uint32 - subscribeLossPercentageMutex sync.RWMutex - subscribeLossPercentageArgsForCall []struct { - } - subscribeLossPercentageReturns struct { - result1 uint32 - } - subscribeLossPercentageReturnsOnCall map[int]struct { - result1 uint32 - } UpdateSubscriberSettingsStub func(*livekit.UpdateTrackSettings) updateSubscriberSettingsMutex sync.RWMutex updateSubscriberSettingsArgsForCall []struct { @@ -422,59 +412,6 @@ func (fake *FakeSubscribedTrack) SetPublisherMutedArgsForCall(i int) bool { return argsForCall.arg1 } -func (fake *FakeSubscribedTrack) SubscribeLossPercentage() uint32 { - fake.subscribeLossPercentageMutex.Lock() - ret, specificReturn := fake.subscribeLossPercentageReturnsOnCall[len(fake.subscribeLossPercentageArgsForCall)] - fake.subscribeLossPercentageArgsForCall = append(fake.subscribeLossPercentageArgsForCall, struct { - }{}) - stub := fake.SubscribeLossPercentageStub - fakeReturns := fake.subscribeLossPercentageReturns - fake.recordInvocation("SubscribeLossPercentage", []interface{}{}) - fake.subscribeLossPercentageMutex.Unlock() - if stub != nil { - return stub() - } - if specificReturn { - return ret.result1 - } - return fakeReturns.result1 -} - -func (fake *FakeSubscribedTrack) SubscribeLossPercentageCallCount() int { - fake.subscribeLossPercentageMutex.RLock() - defer fake.subscribeLossPercentageMutex.RUnlock() - return len(fake.subscribeLossPercentageArgsForCall) -} - -func (fake *FakeSubscribedTrack) SubscribeLossPercentageCalls(stub func() uint32) { - fake.subscribeLossPercentageMutex.Lock() - defer fake.subscribeLossPercentageMutex.Unlock() - fake.SubscribeLossPercentageStub = stub -} - -func (fake *FakeSubscribedTrack) SubscribeLossPercentageReturns(result1 uint32) { - fake.subscribeLossPercentageMutex.Lock() - defer fake.subscribeLossPercentageMutex.Unlock() - fake.SubscribeLossPercentageStub = nil - fake.subscribeLossPercentageReturns = struct { - result1 uint32 - }{result1} -} - -func (fake *FakeSubscribedTrack) SubscribeLossPercentageReturnsOnCall(i int, result1 uint32) { - fake.subscribeLossPercentageMutex.Lock() - defer fake.subscribeLossPercentageMutex.Unlock() - fake.SubscribeLossPercentageStub = nil - if fake.subscribeLossPercentageReturnsOnCall == nil { - fake.subscribeLossPercentageReturnsOnCall = make(map[int]struct { - result1 uint32 - }) - } - fake.subscribeLossPercentageReturnsOnCall[i] = struct { - result1 uint32 - }{result1} -} - func (fake *FakeSubscribedTrack) UpdateSubscriberSettings(arg1 *livekit.UpdateTrackSettings) { fake.updateSubscriberSettingsMutex.Lock() fake.updateSubscriberSettingsArgsForCall = append(fake.updateSubscriberSettingsArgsForCall, struct { @@ -548,8 +485,6 @@ func (fake *FakeSubscribedTrack) Invocations() map[string][][]interface{} { defer fake.publisherIdentityMutex.RUnlock() fake.setPublisherMutedMutex.RLock() defer fake.setPublisherMutedMutex.RUnlock() - fake.subscribeLossPercentageMutex.RLock() - defer fake.subscribeLossPercentageMutex.RUnlock() fake.updateSubscriberSettingsMutex.RLock() defer fake.updateSubscriberSettingsMutex.RUnlock() fake.updateVideoLayerMutex.RLock() diff --git a/pkg/sfu/connectionquality/connectionstats.go b/pkg/sfu/connectionquality/connectionstats.go new file mode 100644 index 000000000..e2e36fd8c --- /dev/null +++ b/pkg/sfu/connectionquality/connectionstats.go @@ -0,0 +1,50 @@ +package connectionquality + +import ( + "github.com/livekit/protocol/livekit" +) + +type ConnectionStat struct { + PacketsLost uint32 + Delay uint32 + Jitter uint32 + TotalPackets uint32 + LastSeqNum uint32 +} + +type ConnectionStats struct { + Curr *ConnectionStat + Prev *ConnectionStat + Score float64 +} + +func NewConnectionStats() *ConnectionStats { + return &ConnectionStats{Curr: &ConnectionStat{}, Prev: &ConnectionStat{}, Score: 4.0} +} + +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) CalculateAudioScore() float64 { + // update feedback stats + current := cs.Curr + previous := cs.Prev + + // Update TotalPackets from SeqNum here + current.TotalPackets += getTotalPackets(current.LastSeqNum, previous.LastSeqNum) + cs.Score = ConnectionScore(current, previous, livekit.TrackType_AUDIO) + + // store previous stats + cs.Prev = current + cs.Curr = &ConnectionStat{TotalPackets: cs.Prev.TotalPackets, PacketsLost: cs.Prev.PacketsLost} + + return cs.Score +} diff --git a/pkg/sfu/connectionquality/mos.go b/pkg/sfu/connectionquality/mos.go new file mode 100644 index 000000000..ed42b7aa9 --- /dev/null +++ b/pkg/sfu/connectionquality/mos.go @@ -0,0 +1,84 @@ +package connectionquality + +import ( + "github.com/livekit/protocol/livekit" +) + +// MOS score calculation is based on webrtc-stats +// available @ https://github.com/oanguenot/webrtc-stats + +const ( + rtt = 70 +) + +func Score2Rating(score float64) livekit.ConnectionQuality { + if score > 3.9 { + return livekit.ConnectionQuality_EXCELLENT + } + + if score > 2.5 { + return livekit.ConnectionQuality_GOOD + } + return livekit.ConnectionQuality_POOR +} + +func mosAudioEmodel(cur, prev *ConnectionStat) float64 { + + if cur == nil { + return 0.0 + } + + // find percentage of lost packets in this window + deltaTotalPackets := cur.TotalPackets - prev.TotalPackets + if deltaTotalPackets == 0 { + return 0.0 + } + + deltaTotalLostPackets := cur.PacketsLost - prev.PacketsLost + percentageLost := (float64(deltaTotalLostPackets) / float64(deltaTotalPackets)) * 100 + + rx := 93.2 - percentageLost + ry := 0.18*rx*rx - 27.9*rx + 1126.62 + + d := float64(rtt + cur.Jitter) + h := d - 177.3 + if h < 0 { + h = 0 + } else { + h = 1 + } + id := 0.024*d + 0.11*(d-177.3)*h + r := ry - (id) + if r < 0 { + return 1 + } + if r > 100 { + return 4.5 + } + score := 1 + (0.035 * r) + (7.0/1000000)*r*(r-60)*(100-r) + + return score +} + +func Loss2Score(loss uint32, reducedQuality bool) float64 { + // No Loss, excellent + if loss == 0 && !reducedQuality { + return 5 + } + score := 3.0 + // loss is bad + if loss >= 4 { + score = 2.0 + } else if loss <= 2 && !reducedQuality { + // loss is acceptable and at reduced quality + score = 4.5 + } + return score +} + +func ConnectionScore(cur, prev *ConnectionStat, kind livekit.TrackType) float64 { + if kind == livekit.TrackType_AUDIO { + return mosAudioEmodel(cur, prev) + } + return 0 +} diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index d4da3f577..0e8ce6cb8 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -4,6 +4,7 @@ import ( "encoding/binary" "errors" "fmt" + "github.com/livekit/livekit-server/pkg/sfu/connectionquality" "io" "strings" "sync" @@ -19,6 +20,10 @@ import ( "github.com/livekit/livekit-server/pkg/sfu/buffer" ) +const ( + connectionQualityUpdateInterval = 5 * time.Second +) + // TrackSender defines an interface send media to remote peer type TrackSender interface { UptrackLayersChange(availableLayers []uint16) @@ -94,9 +99,12 @@ type DownTrack struct { closeOnce sync.Once // Report helpers - primaryStats atomic.Value // contains *PacketStats - rtxStats atomic.Value // contains *PacketStats - paddingStats atomic.Value // contains *PacketStats + primaryStats atomic.Value // contains *PacketStats + rtxStats atomic.Value // contains *PacketStats + paddingStats atomic.Value // contains *PacketStats + statsLock sync.Mutex + connectionStats *connectionquality.ConnectionStats + done chan struct{} lossFraction atomicUint8 @@ -138,21 +146,25 @@ 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), + 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{}), } d.primaryStats.Store(new(PacketStats)) d.rtxStats.Store(new(PacketStats)) d.paddingStats.Store(new(PacketStats)) + go d.updateStats() + return d, nil } @@ -426,6 +438,7 @@ func (d *DownTrack) Close() { if d.onCloseHandler != nil { d.onCloseHandler() } + close(d.done) }) } @@ -764,6 +777,12 @@ func (d *DownTrack) handleRTCP(bytes []byte) { } } + var hasReport bool + var delay uint32 + var jitter uint32 + var totalLost uint32 + var maxSeqNum uint32 + maxRatePacketLoss := uint8(0) for _, pkt := range pkts { switch p := pkt.(type) { @@ -789,6 +808,18 @@ func (d *DownTrack) handleRTCP(bytes []byte) { 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 { @@ -798,6 +829,22 @@ func (d *DownTrack) handleRTCP(bytes []byte) { } d.listenerLock.RUnlock() } + if hasReport { + d.statsLock.Lock() + // update feedback stats + current := d.connectionStats.Curr + if jitter > current.Jitter { + current.Jitter = jitter + } + if delay > current.Delay { + current.Delay = delay + } + if maxSeqNum > current.LastSeqNum { + current.LastSeqNum = maxSeqNum + } + current.PacketsLost = totalLost + d.statsLock.Unlock() + } case *rtcp.TransportLayerNack: var nackedPackets []packetMeta for _, pair := range p.Nacks { @@ -983,3 +1030,40 @@ func (d *DownTrack) DebugInfo() map[string]interface{} { "Stats": stats, } } + +func (d *DownTrack) GetConnectionScore() float64 { + d.statsLock.Lock() + defer d.statsLock.Unlock() + return d.connectionStats.Score +} + +func (d *DownTrack) updateStats() { + for { + select { + case <-d.done: + return + case <-time.After(connectionQualityUpdateInterval): + d.statsLock.Lock() + if d.Kind() == webrtc.RTPCodecTypeAudio { + d.connectionStats.CalculateAudioScore() + } else { + d.calculateVideoScore() + } + d.statsLock.Unlock() + } + } +} + +// converts a fixed point number to the number part of % +func FixedPointToPercent(frac uint8) uint32 { + return (uint32(frac) * 100) >> 8 +} + +func (d *DownTrack) calculateVideoScore() { + var reducedQuality bool + if d.GetForwardingStatus() != ForwardingStatusOptimal { + reducedQuality = true + } + d.connectionStats.Score = connectionquality.Loss2Score(FixedPointToPercent(d.CurrentMaxLossFraction()), reducedQuality) + return +}