diff --git a/pkg/sfu/buffer/buffer.go b/pkg/sfu/buffer/buffer.go index 04e468d7e..85a9f07eb 100644 --- a/pkg/sfu/buffer/buffer.go +++ b/pkg/sfu/buffer/buffer.go @@ -976,6 +976,17 @@ func (b *Buffer) GetDeltaStats() *StreamStatsWithLayers { } } +func (b *Buffer) GetLastSenderReportTime() time.Time { + b.RLock() + defer b.RUnlock() + + if b.rtpStats == nil { + return time.Time{} + } + + return b.rtpStats.LastSenderReportTime() +} + func (b *Buffer) GetAudioLevel() (float64, bool) { b.RLock() defer b.RUnlock() diff --git a/pkg/sfu/buffer/rtpstats_receiver.go b/pkg/sfu/buffer/rtpstats_receiver.go index 3fc6c19a1..6fa9969af 100644 --- a/pkg/sfu/buffer/rtpstats_receiver.go +++ b/pkg/sfu/buffer/rtpstats_receiver.go @@ -458,6 +458,17 @@ func (r *RTPStatsReceiver) GetRtcpSenderReportData() *RTCPSenderReportData { return &srNewestCopy } +func (r *RTPStatsReceiver) LastSenderReportTime() time.Time { + r.lock.RLock() + defer r.lock.RUnlock() + + if r.srNewest != nil { + return r.srNewest.At + } + + return time.Time{} +} + func (r *RTPStatsReceiver) GetRtcpReceptionReport(ssrc uint32, proxyFracLost uint8, snapshotID uint32) *rtcp.ReceptionReport { r.lock.Lock() defer r.lock.Unlock() diff --git a/pkg/sfu/connectionquality/connectionstats.go b/pkg/sfu/connectionquality/connectionstats.go index 8f6035a06..f124edb5c 100644 --- a/pkg/sfu/connectionquality/connectionstats.go +++ b/pkg/sfu/connectionquality/connectionstats.go @@ -37,6 +37,7 @@ const ( type ConnectionStatsReceiverProvider interface { GetDeltaStats() map[uint32]*buffer.StreamStatsWithLayers + GetLastSenderReportTime() time.Time } type ConnectionStatsSenderProvider interface { @@ -202,7 +203,7 @@ func (cs *ConnectionStats) GetScoreAndQuality() (float32, livekit.ConnectionQual return cs.scorer.GetMOSAndQuality() } -func (cs *ConnectionStats) updateScoreWithAggregate(agg *buffer.RTPDeltaInfo, at time.Time) float32 { +func (cs *ConnectionStats) updateScoreWithAggregate(agg *buffer.RTPDeltaInfo, lastRTCPAt time.Time, at time.Time) float32 { var stat windowStat if agg != nil { stat.startedAt = agg.StartTime @@ -214,6 +215,8 @@ func (cs *ConnectionStats) updateScoreWithAggregate(agg *buffer.RTPDeltaInfo, at stat.bytes = agg.Bytes - agg.HeaderBytes // only use media payload size stat.rttMax = agg.RttMax stat.jitterMax = agg.JitterMax + + stat.lastRTCPAt = lastRTCPAt } if at.IsZero() { cs.scorer.Update(&stat) @@ -246,7 +249,7 @@ func (cs *ConnectionStats) updateScoreFromReceiverReport(at time.Time) (float32, } if time.Since(marker) > noReceiverReportTooLongThreshold { // have not received receiver report for a long time when streaming, run with nil stat - return cs.updateScoreWithAggregate(nil, at), nil + return cs.updateScoreWithAggregate(nil, time.Time{}, at), nil } // wait for receiver report, return current score @@ -266,7 +269,7 @@ func (cs *ConnectionStats) updateScoreFromReceiverReport(at time.Time) (float32, if streamingStartedAt.After(agg.StartTime) { agg.StartTime = streamingStartedAt } - return cs.updateScoreWithAggregate(agg, at), streams + return cs.updateScoreWithAggregate(agg, time.Time{}, at), streams } func (cs *ConnectionStats) updateScoreAt(at time.Time) (float32, map[uint32]*buffer.StreamStatsWithLayers) { @@ -290,7 +293,7 @@ func (cs *ConnectionStats) updateScoreAt(at time.Time) (float32, map[uint32]*buf deltaInfoList = append(deltaInfoList, s.RTPStats) } agg := buffer.AggregateRTPDeltaInfo(deltaInfoList) - return cs.updateScoreWithAggregate(agg, at), streams + return cs.updateScoreWithAggregate(agg, cs.params.ReceiverProvider.GetLastSenderReportTime(), at), streams } func (cs *ConnectionStats) updateStreamingStart(at time.Time) time.Time { diff --git a/pkg/sfu/connectionquality/connectionstats_test.go b/pkg/sfu/connectionquality/connectionstats_test.go index 0feedac7f..78e217130 100644 --- a/pkg/sfu/connectionquality/connectionstats_test.go +++ b/pkg/sfu/connectionquality/connectionstats_test.go @@ -29,7 +29,8 @@ import ( // ----------------------------------------------- type testReceiverProvider struct { - streams map[uint32]*buffer.StreamStatsWithLayers + streams map[uint32]*buffer.StreamStatsWithLayers + lastSenderReportTime time.Time } func newTestReceiverProvider() *testReceiverProvider { @@ -44,6 +45,14 @@ func (trp *testReceiverProvider) GetDeltaStats() map[uint32]*buffer.StreamStatsW return trp.streams } +func (trp *testReceiverProvider) setLastSenderReportTime(at time.Time) { + trp.lastSenderReportTime = at +} + +func (trp *testReceiverProvider) GetLastSenderReportTime() time.Time { + return trp.lastSenderReportTime +} + // ----------------------------------------------- func TestConnectionQuality(t *testing.T) { @@ -232,7 +241,7 @@ func TestConnectionQuality(t *testing.T) { require.Greater(t, float32(4.6), mos) require.Equal(t, livekit.ConnectionQuality_EXCELLENT, quality) - // unmute at time so that next window does not satisfy the unmute time threshold. + // unmute at specific time to ensure next window does not satisfy the unmute time threshold. // that means even if the next update has 0 packets, it should hold state and stay at EXCELLENT quality cs.UpdateMuteAt(false, now.Add(3*time.Second)) @@ -250,7 +259,8 @@ 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 to LOST + // next update with no packets, + // but last RTCP is not set, should knock quality down to POOR now = now.Add(duration) trp.setStreams(map[uint32]*buffer.StreamStatsWithLayers{ 1: { @@ -264,13 +274,46 @@ 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) + + // another dry spell, but last RTCP is not stale, should keep quality at POOR + now = now.Add(duration) + trp.setLastSenderReportTime(now.Add(time.Second)) + trp.setStreams(map[uint32]*buffer.StreamStatsWithLayers{ + 1: { + RTPStats: &buffer.RTPDeltaInfo{ + StartTime: now, + EndTime: now.Add(duration), + Packets: 0, + }, + }, + }) + cs.updateScoreAt(now.Add(duration)) + mos, quality = cs.GetScoreAndQuality() + require.Greater(t, float32(2.1), mos) + require.Equal(t, livekit.ConnectionQuality_POOR, quality) + + // yet another dry spell, but last RTCP is stale, should knock down quality at LOST + now = now.Add(duration) + trp.setStreams(map[uint32]*buffer.StreamStatsWithLayers{ + 1: { + RTPStats: &buffer.RTPDeltaInfo{ + StartTime: now, + EndTime: now.Add(duration), + Packets: 0, + }, + }, + }) + cs.updateScoreAt(now.Add(duration)) + mos, quality = cs.GetScoreAndQuality() + require.Greater(t, float32(1.3), mos) require.Equal(t, livekit.ConnectionQuality_LOST, quality) // mute when LOST 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.Greater(t, float32(1.3), mos) require.Equal(t, livekit.ConnectionQuality_LOST, quality) // unmute and send packets to bring quality back up diff --git a/pkg/sfu/connectionquality/scorer.go b/pkg/sfu/connectionquality/scorer.go index 61019d013..facef770a 100644 --- a/pkg/sfu/connectionquality/scorer.go +++ b/pkg/sfu/connectionquality/scorer.go @@ -61,6 +61,7 @@ type windowStat struct { bytes uint64 rttMax uint32 jitterMax float64 + lastRTCPAt time.Time } func (w *windowStat) calculatePacketScore(plw float64, includeRTT bool, includeJitter bool) float64 { @@ -147,7 +148,7 @@ func (w *windowStat) calculateBitrateScore(expectedBitrate int64, isEnabled bool } func (w *windowStat) String() string { - return fmt.Sprintf("start: %+v, dur: %+v, pe: %d, pl: %d, pm: %d, pooo: %d, b: %d, rtt: %d, jitter: %0.2f", + return fmt.Sprintf("start: %+v, dur: %+v, pe: %d, pl: %d, pm: %d, pooo: %d, b: %d, rtt: %d, jitter: %0.2f, lastRTCP: %+v", w.startedAt, w.duration, w.packetsExpected, @@ -157,6 +158,7 @@ func (w *windowStat) String() string { w.bytes, w.rttMax, w.jitterMax, + w.lastRTCPAt, ) } @@ -174,6 +176,7 @@ func (w *windowStat) MarshalLogObject(e zapcore.ObjectEncoder) error { e.AddUint64("bytes", w.bytes) e.AddUint32("rttMax", w.rttMax) e.AddFloat64("jitterMax", w.jitterMax) + e.AddTime("lastRTCPAt", w.lastRTCPAt) return nil } @@ -393,8 +396,13 @@ func (q *qualityScorer) updateAtLocked(stat *windowStat, at time.Time) { reason := "none" var score float64 if stat.packetsExpected == 0 { - reason = "dry" - score = qualityTransitionScore[livekit.ConnectionQuality_LOST] + if !stat.lastRTCPAt.IsZero() && at.Sub(stat.lastRTCPAt) > stat.duration { + reason = "dry" + score = qualityTransitionScore[livekit.ConnectionQuality_LOST] + } else { + reason = "rtcp" + score = qualityTransitionScore[livekit.ConnectionQuality_POOR] + } } else { packetScore := stat.calculatePacketScore(plw, q.params.IncludeRTT, q.params.IncludeJitter) bitrateScore := stat.calculateBitrateScore(expectedBitrate, q.params.EnableBitrateScore) diff --git a/pkg/sfu/receiver.go b/pkg/sfu/receiver.go index 78d49fb2b..0cb210273 100644 --- a/pkg/sfu/receiver.go +++ b/pkg/sfu/receiver.go @@ -647,6 +647,25 @@ func (w *WebRTCReceiver) GetDeltaStats() map[uint32]*buffer.StreamStatsWithLayer return deltaStats } +func (w *WebRTCReceiver) GetLastSenderReportTime() time.Time { + w.bufferMu.RLock() + defer w.bufferMu.RUnlock() + + latestSRTime := time.Time{} + for _, buff := range w.buffers { + if buff == nil { + continue + } + + srAt := buff.GetLastSenderReportTime() + if srAt.After(latestSRTime) { + latestSRTime = srAt + } + } + + return latestSRTime +} + func (w *WebRTCReceiver) forwardRTP(layer int32) { pktBuf := make([]byte, bucket.MaxPktSize) tracker := w.streamTrackerManager.GetTracker(layer)