Pacer interface to send packets

go.mod

@@ -3,6 +3,7 @@ module github.com/livekit/livekit-server
 go 1.18
 
 require (
+	github.com/MicahParks/peakdetect v0.0.6
 	github.com/bep/debounce v1.2.1
 	github.com/d5/tengo/v2 v2.16.0
 	github.com/dustin/go-humanize v1.0.1

go.sum

@@ -1,3 +1,5 @@
+github.com/MicahParks/peakdetect v0.0.6 h1:OzupTN1rWcjSCsDZrg+4w8D4FbDrR5WO4ItQOwCTwvk=
+github.com/MicahParks/peakdetect v0.0.6/go.mod h1:NcgqFDGbT9/nVTFdZKY8iLS8n1BSd9eKp+rbQ73A118=
 github.com/benbjohnson/clock v1.1.0 h1:Q92kusRqC1XV2MjkWETPvjJVqKetz1OzxZB7mHJLju8=
 github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
 github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=

pkg/config/config.go

@@ -114,6 +114,7 @@ type CongestionControlConfig struct {
 	UseSendSideBWE     bool                       `yaml:"send_side_bandwidth_estimation,omitempty"`
 	ProbeMode          CongestionControlProbeMode `yaml:"padding_mode,omitempty"`
 	MinChannelCapacity int64                      `yaml:"min_channel_capacity,omitempty"`
+	UseTWCC            bool                       `yaml:"use_twcc"`
 }
 
 type AudioConfig struct {
@@ -272,6 +273,7 @@ func NewConfig(confString string, strictMode bool, c *cli.Context, baseFlags []c
 				Enabled:    true,
 				AllowPause: false,
 				ProbeMode:  CongestionControlProbeModePadding,
+				UseTWCC:    true,
 			},
 		},
 		Audio: AudioConfig{

pkg/rtc/config.go

@@ -99,7 +99,7 @@ func NewWebRTCConfig(conf *config.Config) (*WebRTCConfig, error) {
 			},
 		},
 	}
-	if rtcConf.CongestionControl.UseSendSideBWE {
+	if rtcConf.CongestionControl.UseSendSideBWE || rtcConf.CongestionControl.UseTWCC {
 		subscriberConfig.RTPHeaderExtension.Video = append(subscriberConfig.RTPHeaderExtension.Video, sdp.TransportCCURI)
 		subscriberConfig.RTCPFeedback.Video = append(subscriberConfig.RTCPFeedback.Video, webrtc.RTCPFeedback{Type: webrtc.TypeRTCPFBTransportCC})
 	} else {

pkg/rtc/mediatracksubscriptions.go

@@ -105,6 +105,7 @@ func (t *MediaTrackSubscriptions) AddSubscriber(sub types.LocalParticipant, wr *
 		subscriberID,
 		t.params.ReceiverConfig.PacketBufferSize,
 		sub.GetAllowTimestampAdjustment(),
+		sub.GetPacer(),
 		LoggerWithTrack(sub.GetLogger(), trackID, t.params.IsRelayed),
 	)
 	if err != nil {

pkg/rtc/participant.go

@@ -24,6 +24,7 @@ import (
 	"github.com/livekit/livekit-server/pkg/sfu"
 	"github.com/livekit/livekit-server/pkg/sfu/buffer"
 	"github.com/livekit/livekit-server/pkg/sfu/connectionquality"
+	"github.com/livekit/livekit-server/pkg/sfu/pacer"
 	"github.com/livekit/livekit-server/pkg/sfu/streamallocator"
 	"github.com/livekit/livekit-server/pkg/telemetry"
 	"github.com/livekit/livekit-server/pkg/telemetry/prometheus"
@@ -237,6 +238,10 @@ func (p *ParticipantImpl) GetAllowTimestampAdjustment() bool {
 	return p.params.AllowTimestampAdjustment
 }
 
+func (p *ParticipantImpl) GetPacer() pacer.Pacer {
+	return p.TransportManager.GetSubscriberPacer()
+}
+
 func (p *ParticipantImpl) ID() livekit.ParticipantID {
 	return p.params.SID
 }

pkg/rtc/transport.go

@@ -27,6 +27,8 @@ import (
 
 	"github.com/livekit/livekit-server/pkg/config"
 	"github.com/livekit/livekit-server/pkg/rtc/types"
+	"github.com/livekit/livekit-server/pkg/sfu/pacer"
+	"github.com/livekit/livekit-server/pkg/sfu/sendsidebwe"
 	"github.com/livekit/livekit-server/pkg/sfu/streamallocator"
 	"github.com/livekit/livekit-server/pkg/telemetry"
 	"github.com/livekit/livekit-server/pkg/telemetry/prometheus"
@@ -185,6 +187,10 @@ type PCTransport struct {
 	// stream allocator for subscriber PC
 	streamAllocator *streamallocator.StreamAllocator
 
+	// only for subscriber PC
+	sendSideBWE *sendsidebwe.SendSideBWE
+	pacer       pacer.Pacer
+
 	previousAnswer *webrtc.SessionDescription
 	// track id -> description map in previous offer sdp
 	previousTrackDescription map[string]*trackDescription
@@ -232,9 +238,7 @@ type TransportParams struct {
 }
 
 func newPeerConnection(params TransportParams, onBandwidthEstimator func(estimator cc.BandwidthEstimator)) (*webrtc.PeerConnection, *webrtc.MediaEngine, error) {
-	directionConfig := params.DirectionConfig
-
-	me, err := createMediaEngine(params.EnabledCodecs, directionConfig)
+	me, err := createMediaEngine(params.EnabledCodecs, params.DirectionConfig)
 	if err != nil {
 		return nil, nil, err
 	}
@@ -305,21 +309,7 @@ func newPeerConnection(params TransportParams, onBandwidthEstimator func(estimat
 
 	ir := &interceptor.Registry{}
 	if params.IsSendSide {
-		isSendSideBWE := false
-		for _, ext := range directionConfig.RTPHeaderExtension.Video {
-			if ext == sdp.TransportCCURI {
-				isSendSideBWE = true
-				break
-			}
-		}
-		for _, ext := range directionConfig.RTPHeaderExtension.Audio {
-			if ext == sdp.TransportCCURI {
-				isSendSideBWE = true
-				break
-			}
-		}
-
-		if isSendSideBWE {
+		if params.CongestionControlConfig.UseSendSideBWE && !params.CongestionControlConfig.UseTWCC {
 			gf, err := cc.NewInterceptor(func() (cc.BandwidthEstimator, error) {
 				return gcc.NewSendSideBWE(
 					gcc.SendSideBWEInitialBitrate(1*1000*1000),
@@ -376,6 +366,15 @@ func NewPCTransport(params TransportParams) (*PCTransport, error) {
 			Logger: params.Logger,
 		})
 		t.streamAllocator.Start()
+
+		if params.CongestionControlConfig.UseTWCC {
+			t.sendSideBWE = sendsidebwe.NewSendSideBWE(params.Logger)
+			t.pacer = pacer.NewNoQueue(params.Logger, t.sendSideBWE)
+
+			t.streamAllocator.SetSendSideBWE(t.sendSideBWE)
+		} else {
+			t.pacer = pacer.NewPassThrough(params.Logger, nil)
+		}
 	}
 
 	if err := t.createPeerConnection(); err != nil {
@@ -414,6 +413,10 @@ func (t *PCTransport) createPeerConnection() error {
 	return nil
 }
 
+func (t *PCTransport) GetPacer() pacer.Pacer {
+	return t.pacer
+}
+
 func (t *PCTransport) SetSignalingRTT(rtt uint32) {
 	t.signalingRTT.Store(rtt)
 }
@@ -898,6 +901,12 @@ func (t *PCTransport) Close() {
 	if t.streamAllocator != nil {
 		t.streamAllocator.Stop()
 	}
+	if t.pacer != nil {
+		t.pacer.Stop()
+	}
+	if t.sendSideBWE != nil {
+		t.sendSideBWE.Stop()
+	}
 
 	_ = t.pc.Close()
 

pkg/rtc/transportmanager.go

@@ -17,6 +17,7 @@ import (
 	"github.com/livekit/livekit-server/pkg/config"
 	"github.com/livekit/livekit-server/pkg/rtc/types"
 	"github.com/livekit/livekit-server/pkg/sfu"
+	"github.com/livekit/livekit-server/pkg/sfu/pacer"
 	"github.com/livekit/livekit-server/pkg/sfu/streamallocator"
 	"github.com/livekit/livekit-server/pkg/telemetry"
 	"github.com/livekit/protocol/livekit"
@@ -283,6 +284,10 @@ func (t *TransportManager) WriteSubscriberRTCP(pkts []rtcp.Packet) error {
 	return t.subscriber.WriteRTCP(pkts)
 }
 
+func (t *TransportManager) GetSubscriberPacer() pacer.Pacer {
+	return t.subscriber.GetPacer()
+}
+
 func (t *TransportManager) OnPrimaryTransportInitialConnected(f func()) {
 	t.onPrimaryTransportInitialConnected = f
 }

pkg/rtc/types/interfaces.go

@@ -15,6 +15,7 @@ import (
 	"github.com/livekit/livekit-server/pkg/routing"
 	"github.com/livekit/livekit-server/pkg/sfu"
 	"github.com/livekit/livekit-server/pkg/sfu/buffer"
+	"github.com/livekit/livekit-server/pkg/sfu/pacer"
 )
 
 //go:generate go run github.com/maxbrunsfeld/counterfeiter/v6 -generate
@@ -343,6 +344,8 @@ type LocalParticipant interface {
 	SetSubscriberChannelCapacity(channelCapacity int64)
 
 	GetAllowTimestampAdjustment() bool
+
+	GetPacer() pacer.Pacer
 }
 
 // Room is a container of participants, and can provide room-level actions