From 5b57522c05ecc468cb8535c825f5e0dc77814e2a Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Sat, 29 Jan 2022 00:55:00 +0530 Subject: [PATCH] Refactoring connection stats (#384) --- pkg/rtc/mediatrack.go | 181 +++----------- pkg/rtc/participant.go | 10 +- pkg/rtc/participant_internal_test.go | 25 +- pkg/rtc/utils.go | 5 - pkg/sfu/buffer/buffer.go | 10 +- pkg/sfu/connectionquality/connectionstats.go | 234 +++++++++++++++---- pkg/sfu/connectionquality/mos.go | 26 +-- pkg/sfu/downtrack.go | 175 ++++---------- pkg/sfu/receiver.go | 15 ++ 9 files changed, 303 insertions(+), 378 deletions(-) diff --git a/pkg/rtc/mediatrack.go b/pkg/rtc/mediatrack.go index d2fff6d91..68da84e97 100644 --- a/pkg/rtc/mediatrack.go +++ b/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() } diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 2186228fa..6b0617209 100644 --- a/pkg/rtc/participant.go +++ b/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++ } diff --git a/pkg/rtc/participant_internal_test.go b/pkg/rtc/participant_internal_test.go index 9e6430d50..58d260a1b 100644 --- a/pkg/rtc/participant_internal_test.go +++ b/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 } diff --git a/pkg/rtc/utils.go b/pkg/rtc/utils.go index 1f4f7b1f8..628e0c076 100644 --- a/pkg/rtc/utils.go +++ b/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 { diff --git a/pkg/sfu/buffer/buffer.go b/pkg/sfu/buffer/buffer.go index 714132bbc..cebaf47ee 100644 --- a/pkg/sfu/buffer/buffer.go +++ b/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 { diff --git a/pkg/sfu/connectionquality/connectionstats.go b/pkg/sfu/connectionquality/connectionstats.go index 396c6bbac..8f9d55e91 100644 --- a/pkg/sfu/connectionquality/connectionstats.go +++ b/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) + } + } + } } diff --git a/pkg/sfu/connectionquality/mos.go b/pkg/sfu/connectionquality/mos.go index a231d4817..0fcbf06c4 100644 --- a/pkg/sfu/connectionquality/mos.go +++ b/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) } diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index 600ab8db3..b0b5dc799 100644 --- a/pkg/sfu/downtrack.go +++ b/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() } diff --git a/pkg/sfu/receiver.go b/pkg/sfu/receiver.go index d85f03fe2..0a0fbcbe5 100644 --- a/pkg/sfu/receiver.go +++ b/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]