Merge pull request #37 from livekit/rtcp-throttling

Rtcp throttling

cmd/server/main.go

@@ -16,12 +16,13 @@ import (
 	"github.com/pkg/errors"
 	"github.com/urfave/cli/v2"
 
+	"github.com/livekit/protocol/auth"
+
 	"github.com/livekit/livekit-server/pkg/config"
 	"github.com/livekit/livekit-server/pkg/logger"
 	"github.com/livekit/livekit-server/pkg/routing"
 	"github.com/livekit/livekit-server/pkg/service"
 	"github.com/livekit/livekit-server/version"
-	"github.com/livekit/protocol/auth"
 )
 
 func init() {

go.mod

@@ -5,6 +5,7 @@ go 1.15
 require (
 	github.com/bep/debounce v1.2.0
 	github.com/cpuguy83/go-md2man/v2 v2.0.0 // indirect
+	github.com/frostbyte73/go-throttle v0.0.0-20210621200530-8018c891361d // indirect
 	github.com/go-logr/zapr v0.4.0
 	github.com/go-redis/redis/v8 v8.7.1
 	github.com/google/wire v0.5.0
@@ -45,3 +46,5 @@ require (
 )
 
 replace github.com/pion/webrtc/v3 => github.com/livekit/pion-webrtc/v3 v3.0.30
+
+replace github.com/pion/ion-sfu => github.com/livekit/ion-sfu v1.10.7-0.20210621205540-f74d53541f95

go.sum

@@ -89,6 +89,8 @@ github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7
 github.com/fatih/color v1.7.0/go.mod h1:Zm6kSWBoL9eyXnKyktHP6abPY2pDugNf5KwzbycvMj4=
 github.com/franela/goblin v0.0.0-20200105215937-c9ffbefa60db/go.mod h1:7dvUGVsVBjqR7JHJk0brhHOZYGmfBYOrK0ZhYMEtBr4=
 github.com/franela/goreq v0.0.0-20171204163338-bcd34c9993f8/go.mod h1:ZhphrRTfi2rbfLwlschooIH4+wKKDR4Pdxhh+TRoA20=
+github.com/frostbyte73/go-throttle v0.0.0-20210621200530-8018c891361d h1:rvSueMilKro0jF+VfxoVR42wazKPl+cUBL3rFbiBGso=
+github.com/frostbyte73/go-throttle v0.0.0-20210621200530-8018c891361d/go.mod h1:jhHJXDlUaWA2g15qvWdleCmRe0Z3noWIQQvjbTZOhGY=
 github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
 github.com/fsnotify/fsnotify v1.4.9 h1:hsms1Qyu0jgnwNXIxa+/V/PDsU6CfLf6CNO8H7IWoS4=
 github.com/fsnotify/fsnotify v1.4.9/go.mod h1:znqG4EE+3YCdAaPaxE2ZRY/06pZUdp0tY4IgpuI1SZQ=
@@ -230,10 +232,10 @@ github.com/lightstep/lightstep-tracer-common/golang/gogo v0.0.0-20190605223551-b
 github.com/lightstep/lightstep-tracer-go v0.18.1/go.mod h1:jlF1pusYV4pidLvZ+XD0UBX0ZE6WURAspgAczcDHrL4=
 github.com/lithammer/shortuuid/v3 v3.0.6 h1:pr15YQyvhiSX/qPxncFtqk+v4xLEpOZObbsY/mKrcvA=
 github.com/lithammer/shortuuid/v3 v3.0.6/go.mod h1:vMk8ke37EmiewwolSO1NLW8vP4ZaKlRuDIi8tWWmAts=
+github.com/livekit/ion-sfu v1.10.7-0.20210621205540-f74d53541f95 h1:4Vx5bjf5zD95MEsmIKennsFnHu4CzN4wfy+9JxKTX5Q=
+github.com/livekit/ion-sfu v1.10.7-0.20210621205540-f74d53541f95/go.mod h1:0Kn3ywQ/4PneeIsshuSnCNzQrbWSgRlKrgWXiGRLUp0=
 github.com/livekit/pion-webrtc/v3 v3.0.30 h1:2RomVfztLegWIePXcyPFJDiF5pbT64aIX4RtpXdDksU=
 github.com/livekit/pion-webrtc/v3 v3.0.30/go.mod h1:XFQeLYBf++bWWA0sJqh6zF1ouWluosxwTOMOoTZGaD0=
-github.com/livekit/protocol v0.5.2 h1:jYj+50gKBrtJ1q6iEVdHDruHVllA1xtOhiM1/YKqgMk=
-github.com/livekit/protocol v0.5.2/go.mod h1:wo3CGfYB7XMF8GoVJAfTARrYSP/ombi+sbLl6AYdKP0=
 github.com/livekit/protocol v0.5.3 h1:u4J6poLLX3zRFsJVGAz5X8KFv9+6dwffksMW/4EI+14=
 github.com/livekit/protocol v0.5.3/go.mod h1:wo3CGfYB7XMF8GoVJAfTARrYSP/ombi+sbLl6AYdKP0=
 github.com/lucsky/cuid v1.0.2 h1:z4XlExeoderxoPj2/dxKOyPxe9RCOu7yNq9/XWxIUMQ=
@@ -319,8 +321,6 @@ github.com/pion/ice/v2 v2.1.7 h1:FjgDfUNrVYTxQabJrkBX6ld12tvYbgzHenqPh3PJF6E=
 github.com/pion/ice/v2 v2.1.7/go.mod h1:kV4EODVD5ux2z8XncbLHIOtcXKtYXVgLVCeVqnpoeP0=
 github.com/pion/interceptor v0.0.12 h1:eC1iVneBIAQJEfaNAfDqAncJWhMDAnaXPRCJsltdokE=
 github.com/pion/interceptor v0.0.12/go.mod h1:qzeuWuD/ZXvPqOnxNcnhWfkCZ2e1kwwslicyyPnhoK4=
-github.com/pion/ion-sfu v1.10.5 h1:8VH7IwEEyAr/RiHtAyM2vSmaP0oGKGVD3sFlFCVQ6cg=
-github.com/pion/ion-sfu v1.10.5/go.mod h1:Wx6b4qGUjvSo1kGl+/fHl0ZF48g2IJOjzUFg0yCo9qY=
 github.com/pion/logging v0.2.2 h1:M9+AIj/+pxNsDfAT64+MAVgJO0rsyLnoJKCqf//DoeY=
 github.com/pion/logging v0.2.2/go.mod h1:k0/tDVsRCX2Mb2ZEmTqNa7CWsQPc+YYCB7Q+5pahoms=
 github.com/pion/mdns v0.0.5 h1:Q2oj/JB3NqfzY9xGZ1fPzZzK7sDSD8rZPOvcIQ10BCw=

pkg/config/config.go

@@ -2,6 +2,7 @@ package config
 
 import (
 	"os"
+	"time"
 
 	"github.com/mitchellh/go-homedir"
 	"github.com/urfave/cli/v2"
@@ -37,6 +38,15 @@ type RTCConfig struct {
 
 	// Max bitrate for REMB
 	MaxBitrate uint64 `yaml:"max_bitrate"`
+
+	// Throttle periods for rtcp packets
+	Throttle RTCPThrottleConfig `yaml:"rtcp_throttle"`
+}
+
+type RTCPThrottleConfig struct {
+	LowQuality  time.Duration `yaml:"low_quality"`
+	MidQuality  time.Duration `yaml:"mid_quality"`
+	HighQuality time.Duration `yaml:"high_quality"`
 }
 
 type AudioConfig struct {
@@ -80,6 +90,11 @@ func NewConfig(confString string) (*Config, error) {
 			},
 			MaxBitrate:       3 * 1024 * 1024, // 3 mbps
 			PacketBufferSize: 500,
+			Throttle: RTCPThrottleConfig{
+				LowQuality:  time.Second,
+				MidQuality:  time.Second * 2,
+				HighQuality: time.Second * 3,
+			},
 		},
 		Audio: AudioConfig{
 			ActiveLevel:    40,

pkg/rtc/mediatrack.go

@@ -270,7 +270,7 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track *webrtc.Tra
 	})
 
 	if t.receiver == nil {
-		t.receiver = sfu.NewWebRTCReceiver(receiver, track, t.params.ParticipantID)
+		t.receiver = sfu.NewWebRTCReceiver(receiver, track, t.params.ParticipantID, sfu.WithPliThrottle(0))
 		t.receiver.SetRTCPCh(t.params.RTCPChan)
 		t.receiver.OnCloseHandler(func() {
 			t.lock.Lock()

pkg/rtc/participant.go

@@ -37,6 +37,7 @@ type ParticipantParams struct {
 	AudioConfig     config.AudioConfig
 	ProtocolVersion types.ProtocolVersion
 	Stats           *RoomStatsReporter
+	ThrottleConfig  config.RTCPThrottleConfig
 }
 
 type ParticipantImpl struct {
@@ -49,6 +50,7 @@ type ParticipantImpl struct {
 	state             atomic.Value // livekit.ParticipantInfo_State
 	updateAfterActive atomic.Value // bool
 	rtcpCh            chan []rtcp.Packet
+	rtcpThrottle      *rtcpThrottle
 
 	// reliable and unreliable data channels
 	reliableDC *webrtc.DataChannel
@@ -89,6 +91,7 @@ func NewParticipant(params ParticipantParams) (*ParticipantImpl, error) {
 		params:           params,
 		id:               utils.NewGuid(utils.ParticipantPrefix),
 		rtcpCh:           make(chan []rtcp.Packet, 50),
+		rtcpThrottle:     newRtcpThrottle(params.ThrottleConfig),
 		subscribedTracks: make(map[string][]types.SubscribedTrack),
 		publishedTracks:  make(map[string]types.PublishedTrack, 0),
 		pendingTracks:    make(map[string]*livekit.TrackInfo),
@@ -398,6 +401,7 @@ func (p *ParticipantImpl) Close() error {
 	if onClose != nil {
 		onClose(p)
 	}
+	p.rtcpThrottle.close()
 	p.publisher.Close()
 	p.subscriber.Close()
 	close(p.rtcpCh)
@@ -734,8 +738,10 @@ func (p *ParticipantImpl) onMediaTrack(track *webrtc.TrackRemote, rtpReceiver *w
 		newTrack = true
 	}
 
+	ssrc := uint32(track.SSRC())
+	p.rtcpThrottle.addTrack(ssrc, track.RID())
 	if p.twcc == nil {
-		p.twcc = twcc.NewTransportWideCCResponder(uint32(track.SSRC()))
+		p.twcc = twcc.NewTransportWideCCResponder(ssrc)
 		p.twcc.OnFeedback(func(pkt rtcp.RawPacket) {
 			_ = p.publisher.pc.WriteRTCP([]rtcp.Packet{&pkt})
 		})
@@ -833,6 +839,9 @@ func (p *ParticipantImpl) handleTrackPublished(track types.PublishedTrack) {
 		if p.IsReady() && p.onTrackUpdated != nil {
 			p.onTrackUpdated(p, track)
 		}
+		if mt, ok := track.(*MediaTrack); ok {
+			p.rtcpThrottle.removeTrack(uint32(mt.ssrc))
+		}
 		track.OnClose(nil)
 	})
 
@@ -923,17 +932,37 @@ func (p *ParticipantImpl) downTracksRTCPWorker() {
 
 func (p *ParticipantImpl) rtcpSendWorker() {
 	defer Recover()
-	// read from rtcpChan
-	for pkts := range p.rtcpCh {
-		if pkts == nil {
-			return
-		}
-		// for _, pkt := range pkts {
-		//	logger.Debugw("writing RTCP", "packet", pkt)
-		// }
+
+	write := func(pkts []rtcp.Packet) {
 		if err := p.publisher.pc.WriteRTCP(pkts); err != nil {
 			logger.Errorw("could not write RTCP to participant", err,
 				"participant", p.Identity())
 		}
 	}
+
+	// read from rtcpChan
+	for pkts := range p.rtcpCh {
+		if pkts == nil {
+			return
+		}
+		fwdPkts := make([]rtcp.Packet, 0, len(pkts))
+		for _, pkt := range pkts {
+			var mediaSSRC uint32
+			switch pkt.(type) {
+			case *rtcp.PictureLossIndication:
+				mediaSSRC = pkt.(*rtcp.PictureLossIndication).MediaSSRC
+			case *rtcp.FullIntraRequest:
+				mediaSSRC = pkt.(*rtcp.FullIntraRequest).MediaSSRC
+			default:
+				fwdPkts = append(fwdPkts, pkt)
+				continue
+			}
+
+			p.rtcpThrottle.add(mediaSSRC, func() { write([]rtcp.Packet{pkt}) })
+		}
+
+		if len(fwdPkts) > 0 {
+			write(fwdPkts)
+		}
+	}
 }

pkg/rtc/participant_internal_test.go

@@ -173,6 +173,7 @@ func newParticipantForTest(identity string) *ParticipantImpl {
 		Config:          rtcConf,
 		Sink:            &routingfakes.FakeMessageSink{},
 		ProtocolVersion: 0,
+		ThrottleConfig:  conf.RTC.Throttle,
 	})
 	return p
 }

pkg/rtc/rtcpthrottle.go

@@ -0,0 +1,78 @@
+package rtc
+
+import (
+	"sync"
+	"time"
+
+	"github.com/frostbyte73/go-throttle"
+
+	"github.com/livekit/livekit-server/pkg/config"
+)
+
+type rtcpThrottle struct {
+	config    config.RTCPThrottleConfig
+	mu        sync.RWMutex
+	throttles map[uint32]func(func())
+}
+
+// github.com/pion/ion-sfu/pkg/sfu/simulcast.go
+const (
+	fullResolution    = "f"
+	halfResolution    = "h"
+	quarterResolution = "q"
+)
+
+func newRtcpThrottle(conf config.RTCPThrottleConfig) *rtcpThrottle {
+	return &rtcpThrottle{
+		config:    conf,
+		throttles: make(map[uint32]func(func())),
+	}
+}
+
+func (t *rtcpThrottle) addTrack(ssrc uint32, rid string) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+
+	var duration time.Duration
+	switch rid {
+	case fullResolution:
+		duration = t.config.HighQuality
+	case halfResolution:
+		duration = t.config.MidQuality
+	case quarterResolution:
+		duration = t.config.LowQuality
+	default:
+		duration = t.config.MidQuality
+	}
+
+	t.throttles[ssrc] = throttle.New(duration)
+}
+
+func (t *rtcpThrottle) add(ssrc uint32, f func()) {
+	t.mu.RLock()
+	defer t.mu.RUnlock()
+
+	if trackThrottle, ok := t.throttles[ssrc]; ok {
+		trackThrottle(f)
+	}
+}
+
+func (t *rtcpThrottle) removeTrack(ssrc uint32) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+
+	if trackThrottle, ok := t.throttles[ssrc]; ok {
+		trackThrottle(func() {})
+		delete(t.throttles, ssrc)
+	}
+}
+
+func (t *rtcpThrottle) close() {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+
+	for ssrc, trackThrottle := range t.throttles {
+		trackThrottle(func() {})
+		delete(t.throttles, ssrc)
+	}
+}

pkg/service/roommanager.go

@@ -261,6 +261,7 @@ func (r *RoomManager) StartSession(roomName string, pi routing.ParticipantInit,
 		AudioConfig:     r.config.Audio,
 		ProtocolVersion: pv,
 		Stats:           room.GetStatsReporter(),
+		ThrottleConfig:  r.config.RTC.Throttle,
 	})
 	if err != nil {
 		logger.Errorw("could not create participant", err)

pkg/service/wire_gen.go

@@ -1,6 +1,6 @@
 // Code generated by Wire. DO NOT EDIT.
 
-//go:generate wire
+//go:generate go run github.com/google/wire/cmd/wire
 //+build !wireinject
 
 package service