From f7291fdaa8289d775d7343480cff49d0578fe346 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Sat, 30 Aug 2025 19:22:14 +0530 Subject: [PATCH] Do not send both asb-send-time and twcc. (#3890) * Do not send both asb-send-time and twcc. In single peer connection mode, both extensions are set on the media engine and both would be negotiated. Unfortunately, pion/webrtc does not yet support RTPSender.SetParameters() which would allow setting specific header extensions for the sender. So, check for TWCC enabled and use it. If not, do abs-send-time if that is enabled. * check BWE type * comment --- pkg/rtc/mediaengine.go | 1 - pkg/rtc/transport.go | 3 ++- pkg/sfu/bwe/bwe.go | 25 ++++++++++++++++++++++ pkg/sfu/bwe/remotebwe/remote_bwe.go | 6 ++++++ pkg/sfu/bwe/sendsidebwe/send_side_bwe.go | 6 ++++++ pkg/sfu/downtrack.go | 10 +++++++-- pkg/sfu/streamallocator/streamallocator.go | 4 ++++ 7 files changed, 51 insertions(+), 4 deletions(-) diff --git a/pkg/rtc/mediaengine.go b/pkg/rtc/mediaengine.go index 6b24dac6e..6a61eba4c 100644 --- a/pkg/rtc/mediaengine.go +++ b/pkg/rtc/mediaengine.go @@ -286,7 +286,6 @@ func filterCodecs( for _, enabledCodec := range enabledCodecs { if mime.NormalizeMimeType(enabledCodec.Mime) == mime.NormalizeMimeType(c.RTPCodecCapability.MimeType) { - // SINGLE-PEER-CONNECTION-TOOD: remove `nack` for RED? if mime.IsMimeTypeStringVideo(c.RTPCodecCapability.MimeType) { c.RTPCodecCapability.RTCPFeedback = rtcpFeedbackConfig.Video } else { diff --git a/pkg/rtc/transport.go b/pkg/rtc/transport.go index 6b4e56acc..195003bb8 100644 --- a/pkg/rtc/transport.go +++ b/pkg/rtc/transport.go @@ -424,7 +424,8 @@ func newPeerConnection(params TransportParams, onBandwidthEstimator func(estimat } } } - } else { + } + if !params.IsOfferer { // sfu only use interceptor to send XR but don't read response from it (use buffer instead), // so use a empty callback here ir.Add(lkinterceptor.NewRTTFromXRFactory(func(rtt uint32) {})) diff --git a/pkg/sfu/bwe/bwe.go b/pkg/sfu/bwe/bwe.go index 5c806df04..ad61a700a 100644 --- a/pkg/sfu/bwe/bwe.go +++ b/pkg/sfu/bwe/bwe.go @@ -31,6 +31,29 @@ const ( // ------------------------------------------------ +type BWEType int + +const ( + BWETypeNone BWEType = iota + BWETypeRemote + BWETypeSendSide +) + +func (b BWEType) String() string { + switch b { + case BWETypeNone: + return "NONE" + case BWETypeRemote: + return "REMOTE" + case BWETypeSendSide: + return "SEND_SIDE" + default: + return fmt.Sprintf("%d", int(b)) + } +} + +// ------------------------------------------------ + type CongestionState int const ( @@ -55,6 +78,8 @@ func (c CongestionState) String() string { // ------------------------------------------------ type BWE interface { + Type() BWEType + SetBWEListener(bweListner BWEListener) Reset() diff --git a/pkg/sfu/bwe/remotebwe/remote_bwe.go b/pkg/sfu/bwe/remotebwe/remote_bwe.go index 30a9f70d1..f6eb4bacb 100644 --- a/pkg/sfu/bwe/remotebwe/remote_bwe.go +++ b/pkg/sfu/bwe/remotebwe/remote_bwe.go @@ -24,6 +24,8 @@ import ( "github.com/livekit/protocol/utils/mono" ) +var _ bwe.BWE = (*RemoteBWE)(nil) + // --------------------------------------------------------------------------- type RemoteBWEConfig struct { @@ -81,6 +83,10 @@ func NewRemoteBWE(params RemoteBWEParams) *RemoteBWE { return r } +func (r *RemoteBWE) Type() bwe.BWEType { + return bwe.BWETypeRemote +} + func (r *RemoteBWE) SetBWEListener(bweListener bwe.BWEListener) { r.lock.Lock() defer r.lock.Unlock() diff --git a/pkg/sfu/bwe/sendsidebwe/send_side_bwe.go b/pkg/sfu/bwe/sendsidebwe/send_side_bwe.go index 1e3cc7444..c31be1e7a 100644 --- a/pkg/sfu/bwe/sendsidebwe/send_side_bwe.go +++ b/pkg/sfu/bwe/sendsidebwe/send_side_bwe.go @@ -23,6 +23,8 @@ import ( "github.com/pion/rtcp" ) +var _ bwe.BWE = (*SendSideBWE)(nil) + // // Based on a simplified/modified version of JitterPath paper // (https://homepage.iis.sinica.edu.tw/papers/lcs/2114-F.pdf) @@ -88,6 +90,10 @@ func NewSendSideBWE(params SendSideBWEParams) *SendSideBWE { } } +func (r *SendSideBWE) Type() bwe.BWEType { + return bwe.BWETypeSendSide +} + func (s *SendSideBWE) SetBWEListener(bweListener bwe.BWEListener) { s.congestionDetector.SetBWEListener(bweListener) } diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index 1bf366038..2a3ff85d7 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -38,6 +38,7 @@ import ( "github.com/livekit/protocol/utils/mono" "github.com/livekit/livekit-server/pkg/sfu/buffer" + "github.com/livekit/livekit-server/pkg/sfu/bwe" "github.com/livekit/livekit-server/pkg/sfu/ccutils" "github.com/livekit/livekit-server/pkg/sfu/connectionquality" "github.com/livekit/livekit-server/pkg/sfu/mime" @@ -192,6 +193,9 @@ type DownTrackStreamAllocatorListener interface { // check if track should participate in BWE IsBWEEnabled(dt *DownTrack) bool + // get the BWE type in use + BWEType() bwe.BWEType + // check if subscription mute can be applied IsSubscribeMutable(dt *DownTrack) bool } @@ -793,13 +797,15 @@ func (d *DownTrack) SetReceiver(r TrackReceiver) { // Sets RTP header extensions for this track func (d *DownTrack) SetRTPHeaderExtensions(rtpHeaderExtensions []webrtc.RTPHeaderExtensionParameter) { isBWEEnabled := true + bweType := bwe.BWETypeNone if sal := d.getStreamAllocatorListener(); sal != nil { isBWEEnabled = sal.IsBWEEnabled(d) + bweType = sal.BWEType() } for _, ext := range rtpHeaderExtensions { switch ext.URI { case sdp.ABSSendTimeURI: - if isBWEEnabled { + if isBWEEnabled && bweType == bwe.BWETypeRemote { d.absSendTimeExtID = ext.ID } else { d.absSendTimeExtID = 0 @@ -809,7 +815,7 @@ func (d *DownTrack) SetRTPHeaderExtensions(rtpHeaderExtensions []webrtc.RTPHeade case pd.PlayoutDelayURI: d.playoutDelayExtID = ext.ID case sdp.TransportCCURI: - if isBWEEnabled { + if isBWEEnabled && bweType == bwe.BWETypeSendSide { d.transportWideExtID = ext.ID } else { d.transportWideExtID = 0 diff --git a/pkg/sfu/streamallocator/streamallocator.go b/pkg/sfu/streamallocator/streamallocator.go index 4c543f201..3cc67a478 100644 --- a/pkg/sfu/streamallocator/streamallocator.go +++ b/pkg/sfu/streamallocator/streamallocator.go @@ -557,6 +557,10 @@ func (s *StreamAllocator) IsBWEEnabled(downTrack *sfu.DownTrack) bool { return true } +func (s *StreamAllocator) BWEType() bwe.BWEType { + return s.params.BWE.Type() +} + // called to check if track subscription mute can be applied func (s *StreamAllocator) IsSubscribeMutable(downTrack *sfu.DownTrack) bool { s.videoTracksMu.Lock()