From af0b0c4734b460e0da8e9efe39c7103de7c51cc2 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Sun, 21 Apr 2024 23:35:24 +0530 Subject: [PATCH] Connection quality LOST only if RTCP is also not available. (#2670) * Connection quality LOST only if RTCP is also not available. It is possible that sender stops all layers of video due to some constraint (CPU or bandwidth). Packet reception going dry due to that should not trigger `LOST` quality. Add last received RTCP time also to distinguish the case of real `LOST` and sender stopping traffic. Some bits to watch for - With audio, RTCP reports could be more than 5 seconds apart (5 seconds is the default interval for connection quality scorer), but audio senders usually send silence packets even when there is no input. So audio completely stopping can be considered `LOST`. - With video, have to observe if all clients continue to send RTCP even if all layers are stopped. - RTCP bandwidth is not supposed to exceed the primary stream bandwidth. libwebrtc calculates that and spaces out RTCP reports accordingly. That is the reason why audio reports are that far apart. If a video stream is encoded at a very low bit rate, it could also be sending RTCP rarely. So, there is the case of LOST being indistinguishable from sender stopping all layers. But, this should be a rare case. * typo --- pkg/sfu/buffer/buffer.go | 11 ++++ pkg/sfu/buffer/rtpstats_receiver.go | 11 ++++ pkg/sfu/connectionquality/connectionstats.go | 11 ++-- .../connectionquality/connectionstats_test.go | 51 +++++++++++++++++-- pkg/sfu/connectionquality/scorer.go | 14 +++-- pkg/sfu/receiver.go | 19 +++++++ 6 files changed, 106 insertions(+), 11 deletions(-) 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)