From faa66d11384eb9a2ba6471025c7219025ee2ea96 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Mon, 15 Jul 2024 11:27:31 +0530 Subject: [PATCH] Make sender report pass through an option. (#2861) Enabled by default. Also, tweak the long term propagation delay a bit. The first propagation delay itself was too high and the long term initialized with a high value. Prevent that and also ensure large negtaives do not have an effect by using a lower bound of 0. Lower bound of 0 is okay as the main purpose is to track sustained high positive values. --- pkg/rtc/mediatracksubscriptions.go | 25 +++---- pkg/rtc/participant.go | 67 ++++++++++--------- pkg/rtc/types/interfaces.go | 2 + .../typesfakes/fake_local_participant.go | 65 ++++++++++++++++++ pkg/sfu/buffer/rtpstats_base.go | 6 +- pkg/sfu/buffer/rtpstats_receiver.go | 16 +++-- pkg/sfu/buffer/rtpstats_sender.go | 4 +- pkg/sfu/downtrack.go | 29 ++++---- 8 files changed, 144 insertions(+), 70 deletions(-) diff --git a/pkg/rtc/mediatracksubscriptions.go b/pkg/rtc/mediatracksubscriptions.go index f49faeb68..acd85037d 100644 --- a/pkg/rtc/mediatracksubscriptions.go +++ b/pkg/rtc/mediatracksubscriptions.go @@ -129,18 +129,19 @@ func (t *MediaTrackSubscriptions) AddSubscriber(sub types.LocalParticipant, wr * } downTrack, err := sfu.NewDownTrack(sfu.DowntrackParams{ - Codecs: codecs, - Source: t.params.MediaTrack.Source(), - Receiver: wr, - BufferFactory: sub.GetBufferFactory(), - SubID: subscriberID, - StreamID: streamID, - MaxTrack: maxTrack, - PlayoutDelayLimit: sub.GetPlayoutDelayConfig(), - Pacer: sub.GetPacer(), - Trailer: trailer, - Logger: LoggerWithTrack(sub.GetLogger().WithComponent(sutils.ComponentSub), trackID, t.params.IsRelayed), - RTCPWriter: sub.WriteSubscriberRTCP, + Codecs: codecs, + Source: t.params.MediaTrack.Source(), + Receiver: wr, + BufferFactory: sub.GetBufferFactory(), + SubID: subscriberID, + StreamID: streamID, + MaxTrack: maxTrack, + PlayoutDelayLimit: sub.GetPlayoutDelayConfig(), + Pacer: sub.GetPacer(), + Trailer: trailer, + Logger: LoggerWithTrack(sub.GetLogger().WithComponent(sutils.ComponentSub), trackID, t.params.IsRelayed), + RTCPWriter: sub.WriteSubscriberRTCP, + DisableSenderReportPassThrough: sub.GetDisableSenderReportPassThrough(), }) if err != nil { return nil, err diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index e8248d0b0..86ce9c3ef 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -112,37 +112,38 @@ type ParticipantParams struct { PLIThrottleConfig config.PLIThrottleConfig CongestionControlConfig config.CongestionControlConfig // codecs that are enabled for this room - PublishEnabledCodecs []*livekit.Codec - SubscribeEnabledCodecs []*livekit.Codec - Logger logger.Logger - SimTracks map[uint32]SimulcastTrackInfo - Grants *auth.ClaimGrants - InitialVersion uint32 - ClientConf *livekit.ClientConfiguration - ClientInfo ClientInfo - Region string - Migration bool - AdaptiveStream bool - AllowTCPFallback bool - TCPFallbackRTTThreshold int - AllowUDPUnstableFallback bool - TURNSEnabled bool - GetParticipantInfo func(pID livekit.ParticipantID) *livekit.ParticipantInfo - GetRegionSettings func(ip string) *livekit.RegionSettings - DisableSupervisor bool - ReconnectOnPublicationError bool - ReconnectOnSubscriptionError bool - ReconnectOnDataChannelError bool - DataChannelMaxBufferedAmount uint64 - VersionGenerator utils.TimedVersionGenerator - TrackResolver types.MediaTrackResolver - DisableDynacast bool - SubscriberAllowPause bool - SubscriptionLimitAudio int32 - SubscriptionLimitVideo int32 - PlayoutDelay *livekit.PlayoutDelay - SyncStreams bool - ForwardStats *sfu.ForwardStats + PublishEnabledCodecs []*livekit.Codec + SubscribeEnabledCodecs []*livekit.Codec + Logger logger.Logger + SimTracks map[uint32]SimulcastTrackInfo + Grants *auth.ClaimGrants + InitialVersion uint32 + ClientConf *livekit.ClientConfiguration + ClientInfo ClientInfo + Region string + Migration bool + AdaptiveStream bool + AllowTCPFallback bool + TCPFallbackRTTThreshold int + AllowUDPUnstableFallback bool + TURNSEnabled bool + GetParticipantInfo func(pID livekit.ParticipantID) *livekit.ParticipantInfo + GetRegionSettings func(ip string) *livekit.RegionSettings + DisableSupervisor bool + ReconnectOnPublicationError bool + ReconnectOnSubscriptionError bool + ReconnectOnDataChannelError bool + DataChannelMaxBufferedAmount uint64 + VersionGenerator utils.TimedVersionGenerator + TrackResolver types.MediaTrackResolver + DisableDynacast bool + SubscriberAllowPause bool + SubscriptionLimitAudio int32 + SubscriptionLimitVideo int32 + PlayoutDelay *livekit.PlayoutDelay + SyncStreams bool + ForwardStats *sfu.ForwardStats + DisableSenderReportPassThrough bool } type ParticipantImpl struct { @@ -320,6 +321,10 @@ func (p *ParticipantImpl) GetPacer() pacer.Pacer { return p.TransportManager.GetSubscriberPacer() } +func (p *ParticipantImpl) GetDisableSenderReportPassThrough() bool { + return p.params.DisableSenderReportPassThrough +} + func (p *ParticipantImpl) ID() livekit.ParticipantID { return p.params.SID } diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index 04568f1e0..b275e80f5 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -427,6 +427,8 @@ type LocalParticipant interface { SetSubscriberChannelCapacity(channelCapacity int64) GetPacer() pacer.Pacer + + GetDisableSenderReportPassThrough() bool } // Room is a container of participants, and can provide room-level actions diff --git a/pkg/rtc/types/typesfakes/fake_local_participant.go b/pkg/rtc/types/typesfakes/fake_local_participant.go index 233c013e3..a21c7c04d 100644 --- a/pkg/rtc/types/typesfakes/fake_local_participant.go +++ b/pkg/rtc/types/typesfakes/fake_local_participant.go @@ -266,6 +266,16 @@ type FakeLocalParticipant struct { getConnectionQualityReturnsOnCall map[int]struct { result1 *livekit.ConnectionQualityInfo } + GetDisableSenderReportPassThroughStub func() bool + getDisableSenderReportPassThroughMutex sync.RWMutex + getDisableSenderReportPassThroughArgsForCall []struct { + } + getDisableSenderReportPassThroughReturns struct { + result1 bool + } + getDisableSenderReportPassThroughReturnsOnCall map[int]struct { + result1 bool + } GetICEConnectionDetailsStub func() []*types.ICEConnectionDetails getICEConnectionDetailsMutex sync.RWMutex getICEConnectionDetailsArgsForCall []struct { @@ -2306,6 +2316,59 @@ func (fake *FakeLocalParticipant) GetConnectionQualityReturnsOnCall(i int, resul }{result1} } +func (fake *FakeLocalParticipant) GetDisableSenderReportPassThrough() bool { + fake.getDisableSenderReportPassThroughMutex.Lock() + ret, specificReturn := fake.getDisableSenderReportPassThroughReturnsOnCall[len(fake.getDisableSenderReportPassThroughArgsForCall)] + fake.getDisableSenderReportPassThroughArgsForCall = append(fake.getDisableSenderReportPassThroughArgsForCall, struct { + }{}) + stub := fake.GetDisableSenderReportPassThroughStub + fakeReturns := fake.getDisableSenderReportPassThroughReturns + fake.recordInvocation("GetDisableSenderReportPassThrough", []interface{}{}) + fake.getDisableSenderReportPassThroughMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeLocalParticipant) GetDisableSenderReportPassThroughCallCount() int { + fake.getDisableSenderReportPassThroughMutex.RLock() + defer fake.getDisableSenderReportPassThroughMutex.RUnlock() + return len(fake.getDisableSenderReportPassThroughArgsForCall) +} + +func (fake *FakeLocalParticipant) GetDisableSenderReportPassThroughCalls(stub func() bool) { + fake.getDisableSenderReportPassThroughMutex.Lock() + defer fake.getDisableSenderReportPassThroughMutex.Unlock() + fake.GetDisableSenderReportPassThroughStub = stub +} + +func (fake *FakeLocalParticipant) GetDisableSenderReportPassThroughReturns(result1 bool) { + fake.getDisableSenderReportPassThroughMutex.Lock() + defer fake.getDisableSenderReportPassThroughMutex.Unlock() + fake.GetDisableSenderReportPassThroughStub = nil + fake.getDisableSenderReportPassThroughReturns = struct { + result1 bool + }{result1} +} + +func (fake *FakeLocalParticipant) GetDisableSenderReportPassThroughReturnsOnCall(i int, result1 bool) { + fake.getDisableSenderReportPassThroughMutex.Lock() + defer fake.getDisableSenderReportPassThroughMutex.Unlock() + fake.GetDisableSenderReportPassThroughStub = nil + if fake.getDisableSenderReportPassThroughReturnsOnCall == nil { + fake.getDisableSenderReportPassThroughReturnsOnCall = make(map[int]struct { + result1 bool + }) + } + fake.getDisableSenderReportPassThroughReturnsOnCall[i] = struct { + result1 bool + }{result1} +} + func (fake *FakeLocalParticipant) GetICEConnectionDetails() []*types.ICEConnectionDetails { fake.getICEConnectionDetailsMutex.Lock() ret, specificReturn := fake.getICEConnectionDetailsReturnsOnCall[len(fake.getICEConnectionDetailsArgsForCall)] @@ -6646,6 +6709,8 @@ func (fake *FakeLocalParticipant) Invocations() map[string][][]interface{} { defer fake.getClientInfoMutex.RUnlock() fake.getConnectionQualityMutex.RLock() defer fake.getConnectionQualityMutex.RUnlock() + fake.getDisableSenderReportPassThroughMutex.RLock() + defer fake.getDisableSenderReportPassThroughMutex.RUnlock() fake.getICEConnectionDetailsMutex.RLock() defer fake.getICEConnectionDetailsMutex.RUnlock() fake.getLoggerMutex.RLock() diff --git a/pkg/sfu/buffer/rtpstats_base.go b/pkg/sfu/buffer/rtpstats_base.go index 553c0cbd8..0eb54ee97 100644 --- a/pkg/sfu/buffer/rtpstats_base.go +++ b/pkg/sfu/buffer/rtpstats_base.go @@ -37,8 +37,6 @@ const ( cFirstPacketTimeAdjustWindow = 2 * time.Minute cFirstPacketTimeAdjustThreshold = 15 * 1e9 - cPassthroughNTPTimestamp = true - cSequenceNumberLargeJumpThreshold = 100 ) @@ -122,8 +120,8 @@ type RTCPSenderReportData struct { AtAdjusted time.Time } -func (r *RTCPSenderReportData) PropagationDelay() time.Duration { - if cPassthroughNTPTimestamp { +func (r *RTCPSenderReportData) PropagationDelay(passThrough bool) time.Duration { + if passThrough { return 0 } diff --git a/pkg/sfu/buffer/rtpstats_receiver.go b/pkg/sfu/buffer/rtpstats_receiver.go index 545c9024d..11f2d4a34 100644 --- a/pkg/sfu/buffer/rtpstats_receiver.go +++ b/pkg/sfu/buffer/rtpstats_receiver.go @@ -260,7 +260,7 @@ func (r *RTPStatsReceiver) Update( r.largeJumpCount++ if (r.largeJumpCount-1)%100 == 0 { r.logger.Warnw( - "large sequence number gap OR time reversed", nil, + "large sequence number gap", nil, append(getLoggingFields(), "count", r.largeJumpCount)..., ) } @@ -492,7 +492,6 @@ func (r *RTPStatsReceiver) updatePropagationDelayAndRecordSenderReport(srData *R deltaPropagationDelay = propagationDelay - r.propagationDelay if deltaPropagationDelay > cPropagationDelayDeltaThresholdMin { // ignore small changes for path change consideration if r.longTermDeltaPropagationDelay != 0 && - deltaPropagationDelay > 0 && deltaPropagationDelay > r.longTermDeltaPropagationDelay*time.Duration(cPropagationDelayDeltaThresholdMaxFactor) { r.logger.Debugw("sharp increase in propagation delay", getPropagationFields()...) r.propagationDelayDeltaHighCount++ @@ -522,10 +521,10 @@ func (r *RTPStatsReceiver) updatePropagationDelayAndRecordSenderReport(srData *R r.propagationDelay += time.Duration(factor * float64(propagationDelay-r.propagationDelay)) } - if r.longTermDeltaPropagationDelay == 0 { - r.longTermDeltaPropagationDelay = deltaPropagationDelay - } else { - if deltaPropagationDelay < cPropagationDelayDeltaLongTermAdaptationThreshold { + if deltaPropagationDelay < cPropagationDelayDeltaLongTermAdaptationThreshold { + if r.longTermDeltaPropagationDelay == 0 { + r.longTermDeltaPropagationDelay = deltaPropagationDelay + } else { // do not adapt to large +ve spikes, can happen when channel is congested and reports are delivered very late // if the spike is in fact a path change, it will persist and handled by path change detection above sinceLastReport := srData.NTPTimestamp.Time().Sub(r.srNewest.NTPTimestamp.Time()) @@ -533,6 +532,9 @@ func (r *RTPStatsReceiver) updatePropagationDelayAndRecordSenderReport(srData *R r.longTermDeltaPropagationDelay += time.Duration(adaptationFactor * float64(deltaPropagationDelay-r.longTermDeltaPropagationDelay)) } } + if r.longTermDeltaPropagationDelay < 0 { + r.longTermDeltaPropagationDelay = 0 + } } // adjust receive time to estimated propagation delay srData.AtAdjusted = ntpTime.Add(r.propagationDelay) @@ -563,8 +565,8 @@ func (r *RTPStatsReceiver) SetRtcpSenderReportData(srData *RTCPSenderReportData) return false } - r.checkRTPClockSkewForSenderReport(srDataExt) r.updatePropagationDelayAndRecordSenderReport(srDataExt) + r.checkRTPClockSkewForSenderReport(srDataExt) r.checkRTPClockSkewAgainstMediaPathForSenderReport(srDataExt) if err, loggingFields := r.maybeAdjustFirstPacketTime(r.srNewest, 0, r.timestamp.GetExtendedStart()); err != nil { diff --git a/pkg/sfu/buffer/rtpstats_sender.go b/pkg/sfu/buffer/rtpstats_sender.go index 86dbf267d..78899f0c6 100644 --- a/pkg/sfu/buffer/rtpstats_sender.go +++ b/pkg/sfu/buffer/rtpstats_sender.go @@ -606,7 +606,7 @@ func (r *RTPStatsSender) GetExpectedRTPTimestamp(at time.Time) (expectedTSExt ui return } -func (r *RTPStatsSender) GetRtcpSenderReport(ssrc uint32, publisherSRData *RTCPSenderReportData, tsOffset uint64) *rtcp.SenderReport { +func (r *RTPStatsSender) GetRtcpSenderReport(ssrc uint32, publisherSRData *RTCPSenderReportData, tsOffset uint64, passThrough bool) *rtcp.SenderReport { r.lock.Lock() defer r.lock.Unlock() @@ -620,7 +620,7 @@ func (r *RTPStatsSender) GetRtcpSenderReport(ssrc uint32, publisherSRData *RTCPS nowNTP mediatransportutil.NtpTime nowRTPExt uint64 ) - if cPassthroughNTPTimestamp { + if passThrough { nowNTP = publisherSRData.NTPTimestamp nowRTPExt = publisherSRData.RTPTimestampExt - tsOffset } else { diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index b68a60457..9001805c6 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -202,18 +202,19 @@ type DownTrackStreamAllocatorListener interface { type ReceiverReportListener func(dt *DownTrack, report *rtcp.ReceiverReport) type DowntrackParams struct { - Codecs []webrtc.RTPCodecParameters - Source livekit.TrackSource - Receiver TrackReceiver - BufferFactory *buffer.Factory - SubID livekit.ParticipantID - StreamID string - MaxTrack int - PlayoutDelayLimit *livekit.PlayoutDelay - Pacer pacer.Pacer - Logger logger.Logger - Trailer []byte - RTCPWriter func([]rtcp.Packet) error + Codecs []webrtc.RTPCodecParameters + Source livekit.TrackSource + Receiver TrackReceiver + BufferFactory *buffer.Factory + SubID livekit.ParticipantID + StreamID string + MaxTrack int + PlayoutDelayLimit *livekit.PlayoutDelay + Pacer pacer.Pacer + Logger logger.Logger + Trailer []byte + RTCPWriter func([]rtcp.Packet) error + DisableSenderReportPassThrough bool } // DownTrack implements TrackLocal, is the track used to write packets @@ -813,7 +814,7 @@ func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) error { _, _, refSenderReport := d.forwarder.GetSenderReportParams() if refSenderReport != nil { actExtCopy := *extPkt.AbsCaptureTimeExt - if err = actExtCopy.Rewrite(refSenderReport.PropagationDelay()); err == nil { + if err = actExtCopy.Rewrite(refSenderReport.PropagationDelay(!d.params.DisableSenderReportPassThrough)); err == nil { actBytes, err = actExtCopy.Marshal() if err == nil { extensions = append( @@ -1404,7 +1405,7 @@ func (d *DownTrack) CreateSenderReport() *rtcp.SenderReport { } _, tsOffset, refSenderReport := d.forwarder.GetSenderReportParams() - return d.rtpStats.GetRtcpSenderReport(d.ssrc, refSenderReport, tsOffset) + return d.rtpStats.GetRtcpSenderReport(d.ssrc, refSenderReport, tsOffset, !d.params.DisableSenderReportPassThrough) } func (d *DownTrack) writeBlankFrameRTP(duration float32, generation uint32) chan struct{} {