From ce3199b8b1249f4475b6b35f708b07ba018f9f83 Mon Sep 17 00:00:00 2001 From: David Colburn Date: Mon, 21 Jun 2021 14:57:26 -0700 Subject: [PATCH 1/8] throttle --- go.mod | 2 ++ go.sum | 8 ++--- pkg/rtc/mediatrack.go | 2 +- pkg/rtc/participant.go | 73 +++++++++++++++++++++++++++++++++++++----- 4 files changed, 72 insertions(+), 13 deletions(-) diff --git a/go.mod b/go.mod index d79288695..2df796374 100644 --- a/go.mod +++ b/go.mod @@ -45,3 +45,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 diff --git a/go.sum b/go.sum index 4daa3a28f..cae7c473d 100644 --- a/go.sum +++ b/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= diff --git a/pkg/rtc/mediatrack.go b/pkg/rtc/mediatrack.go index dc432a179..a73b56421 100644 --- a/pkg/rtc/mediatrack.go +++ b/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() diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 5b041ffff..689af0ce2 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -7,6 +7,7 @@ import ( "sync/atomic" "time" + "github.com/frostbyte73/go-throttle" "github.com/pion/ion-sfu/pkg/sfu" "github.com/pion/ion-sfu/pkg/twcc" "github.com/pion/rtcp" @@ -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 @@ -86,9 +88,12 @@ func NewParticipant(params ParticipantParams) (*ParticipantImpl, error) { // TODO: check to ensure params are valid, id and identity can't be empty p := &ParticipantImpl{ - params: params, - id: utils.NewGuid(utils.ParticipantPrefix), - rtcpCh: make(chan []rtcp.Packet, 50), + params: params, + id: utils.NewGuid(utils.ParticipantPrefix), + rtcpCh: make(chan []rtcp.Packet, 50), + rtcpThrottle: &rtcpThrottle{ + throttles: make(map[uint32]func(func())), + }, subscribedTracks: make(map[string][]types.SubscribedTrack), publishedTracks: make(map[string]types.PublishedTrack, 0), pendingTracks: make(map[string]*livekit.TrackInfo), @@ -714,6 +719,7 @@ func (p *ParticipantImpl) onMediaTrack(track *webrtc.TrackRemote, rtpReceiver *w p.lock.Lock() ptrack := p.publishedTracks[ti.Sid] + ssrc := uint32(track.SSRC()) var mt *MediaTrack var newTrack bool if trk, ok := ptrack.(*MediaTrack); ok { @@ -732,10 +738,11 @@ func (p *ParticipantImpl) onMediaTrack(track *webrtc.TrackRemote, rtpReceiver *w }) mt.name = ti.Name newTrack = true + p.rtcpThrottle.addTrack(ssrc) } 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 +840,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) }) @@ -928,12 +938,59 @@ func (p *ParticipantImpl) rtcpSendWorker() { if pkts == nil { return } - // for _, pkt := range pkts { - // logger.Debugw("writing RTCP", "packet", pkt) - // } - if err := p.publisher.pc.WriteRTCP(pkts); err != nil { + 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 + } + + throttlePkts := []rtcp.Packet{pkt} + p.rtcpThrottle.add(mediaSSRC, func() { + if err := p.publisher.pc.WriteRTCP(throttlePkts); err != nil { + logger.Errorw("could not write RTCP to participant", err, + "participant", p.Identity()) + } + }) + } + + if err := p.publisher.pc.WriteRTCP(fwdPkts); err != nil { logger.Errorw("could not write RTCP to participant", err, "participant", p.Identity()) } } } + +type rtcpThrottle struct { + mu sync.RWMutex + throttles map[uint32]func(func()) +} + +func (t *rtcpThrottle) addTrack(ssrc uint32) { + t.mu.Lock() + defer t.mu.Unlock() + + t.throttles[ssrc] = throttle.New(time.Millisecond * 500) +} + +func (t *rtcpThrottle) removeTrack(ssrc uint32) { + t.mu.Lock() + defer t.mu.Unlock() + + delete(t.throttles, ssrc) +} + +func (t *rtcpThrottle) add(ssrc uint32, f func()) { + t.mu.RLock() + defer t.mu.RUnlock() + + if trackThrottle := t.throttles[ssrc]; trackThrottle != nil { + trackThrottle(f) + } +} From 3748b6862e58908f2e7dc816c87b86911225b0a1 Mon Sep 17 00:00:00 2001 From: David Colburn Date: Mon, 21 Jun 2021 14:59:46 -0700 Subject: [PATCH 2/8] 0.9.15-snapshot --- version/version.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/version/version.go b/version/version.go index d3f84932f..da256244a 100644 --- a/version/version.go +++ b/version/version.go @@ -1,3 +1,3 @@ package version -const Version = "0.9.14-SNAPSHOT" +const Version = "0.9.15-SNAPSHOT" From b17731506eff4bde82084a6226c4c0933672220b Mon Sep 17 00:00:00 2001 From: David Colburn Date: Mon, 21 Jun 2021 15:06:58 -0700 Subject: [PATCH 3/8] add dep --- go.mod | 1 + 1 file changed, 1 insertion(+) diff --git a/go.mod b/go.mod index 2df796374..ce698994b 100644 --- a/go.mod +++ b/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 From e4a460f30bda42c76a711cd4d6e1cf1dc6a4b5f2 Mon Sep 17 00:00:00 2001 From: David Colburn Date: Mon, 21 Jun 2021 15:08:49 -0700 Subject: [PATCH 4/8] nil checks --- pkg/rtc/participant.go | 8 +++++--- version/version.go | 2 +- 2 files changed, 6 insertions(+), 4 deletions(-) diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 689af0ce2..355f3fbf8 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -953,9 +953,11 @@ func (p *ParticipantImpl) rtcpSendWorker() { throttlePkts := []rtcp.Packet{pkt} p.rtcpThrottle.add(mediaSSRC, func() { - if err := p.publisher.pc.WriteRTCP(throttlePkts); err != nil { - logger.Errorw("could not write RTCP to participant", err, - "participant", p.Identity()) + if p.publisher != nil && p.publisher.pc != nil { + if err := p.publisher.pc.WriteRTCP(throttlePkts); err != nil { + logger.Errorw("could not write RTCP to participant", err, + "participant", p.Identity()) + } } }) } diff --git a/version/version.go b/version/version.go index da256244a..d3f84932f 100644 --- a/version/version.go +++ b/version/version.go @@ -1,3 +1,3 @@ package version -const Version = "0.9.15-SNAPSHOT" +const Version = "0.9.14-SNAPSHOT" From fee6f426bbaab21d6f340b53c4d414cc92d20fed Mon Sep 17 00:00:00 2001 From: David Colburn Date: Mon, 21 Jun 2021 15:30:18 -0700 Subject: [PATCH 5/8] close rtcpThrottle --- pkg/rtc/participant.go | 38 +++++++++++++++++++++++++------------- 1 file changed, 25 insertions(+), 13 deletions(-) diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 355f3fbf8..1e1889a05 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -403,6 +403,7 @@ func (p *ParticipantImpl) Close() error { if onClose != nil { onClose(p) } + p.rtcpThrottle.close() p.publisher.Close() p.subscriber.Close() close(p.rtcpCh) @@ -953,11 +954,9 @@ func (p *ParticipantImpl) rtcpSendWorker() { throttlePkts := []rtcp.Packet{pkt} p.rtcpThrottle.add(mediaSSRC, func() { - if p.publisher != nil && p.publisher.pc != nil { - if err := p.publisher.pc.WriteRTCP(throttlePkts); err != nil { - logger.Errorw("could not write RTCP to participant", err, - "participant", p.Identity()) - } + if err := p.publisher.pc.WriteRTCP(throttlePkts); err != nil { + logger.Errorw("could not write RTCP to participant", err, + "participant", p.Identity()) } }) } @@ -981,18 +980,31 @@ func (t *rtcpThrottle) addTrack(ssrc uint32) { t.throttles[ssrc] = throttle.New(time.Millisecond * 500) } -func (t *rtcpThrottle) removeTrack(ssrc uint32) { - t.mu.Lock() - defer t.mu.Unlock() - - delete(t.throttles, ssrc) -} - func (t *rtcpThrottle) add(ssrc uint32, f func()) { t.mu.RLock() defer t.mu.RUnlock() - if trackThrottle := t.throttles[ssrc]; trackThrottle != nil { + 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) + } +} From 31b4148052df70c70284e488d2ad3b7b18f29c03 Mon Sep 17 00:00:00 2001 From: David Colburn Date: Mon, 21 Jun 2021 15:54:42 -0700 Subject: [PATCH 6/8] don't send empty pkts --- pkg/rtc/participant.go | 8 +++++--- pkg/service/wire_gen.go | 2 +- 2 files changed, 6 insertions(+), 4 deletions(-) diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 1e1889a05..281fcca74 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -961,9 +961,11 @@ func (p *ParticipantImpl) rtcpSendWorker() { }) } - if err := p.publisher.pc.WriteRTCP(fwdPkts); err != nil { - logger.Errorw("could not write RTCP to participant", err, - "participant", p.Identity()) + if len(fwdPkts) > 0 { + if err := p.publisher.pc.WriteRTCP(fwdPkts); err != nil { + logger.Errorw("could not write RTCP to participant", err, + "participant", p.Identity()) + } } } } diff --git a/pkg/service/wire_gen.go b/pkg/service/wire_gen.go index c7bddcffb..62d017ed7 100644 --- a/pkg/service/wire_gen.go +++ b/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 From 46ad247927b39d5b766d815255d3e7ae0fa4d8c0 Mon Sep 17 00:00:00 2001 From: David Colburn Date: Mon, 21 Jun 2021 16:16:13 -0700 Subject: [PATCH 7/8] fix addTrack --- pkg/rtc/participant.go | 25 ++++++++++++------------- 1 file changed, 12 insertions(+), 13 deletions(-) diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 281fcca74..41eab98a1 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -720,7 +720,6 @@ func (p *ParticipantImpl) onMediaTrack(track *webrtc.TrackRemote, rtpReceiver *w p.lock.Lock() ptrack := p.publishedTracks[ti.Sid] - ssrc := uint32(track.SSRC()) var mt *MediaTrack var newTrack bool if trk, ok := ptrack.(*MediaTrack); ok { @@ -739,9 +738,10 @@ func (p *ParticipantImpl) onMediaTrack(track *webrtc.TrackRemote, rtpReceiver *w }) mt.name = ti.Name newTrack = true - p.rtcpThrottle.addTrack(ssrc) } + ssrc := uint32(track.SSRC()) + p.rtcpThrottle.addTrack(ssrc) if p.twcc == nil { p.twcc = twcc.NewTransportWideCCResponder(ssrc) p.twcc.OnFeedback(func(pkt rtcp.RawPacket) { @@ -934,6 +934,14 @@ func (p *ParticipantImpl) downTracksRTCPWorker() { func (p *ParticipantImpl) rtcpSendWorker() { defer Recover() + + 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 { @@ -952,20 +960,11 @@ func (p *ParticipantImpl) rtcpSendWorker() { continue } - throttlePkts := []rtcp.Packet{pkt} - p.rtcpThrottle.add(mediaSSRC, func() { - if err := p.publisher.pc.WriteRTCP(throttlePkts); err != nil { - logger.Errorw("could not write RTCP to participant", err, - "participant", p.Identity()) - } - }) + p.rtcpThrottle.add(mediaSSRC, func() { write([]rtcp.Packet{pkt}) }) } if len(fwdPkts) > 0 { - if err := p.publisher.pc.WriteRTCP(fwdPkts); err != nil { - logger.Errorw("could not write RTCP to participant", err, - "participant", p.Identity()) - } + write(fwdPkts) } } } From eebc2f5acd3c316dd6dddd34a655f5f628d74f0e Mon Sep 17 00:00:00 2001 From: David Colburn Date: Mon, 21 Jun 2021 17:13:58 -0700 Subject: [PATCH 8/8] add throttle config --- cmd/server/main.go | 3 +- pkg/config/config.go | 15 ++++++ pkg/rtc/participant.go | 55 +++----------------- pkg/rtc/participant_internal_test.go | 1 + pkg/rtc/rtcpthrottle.go | 78 ++++++++++++++++++++++++++++ pkg/service/roommanager.go | 1 + 6 files changed, 103 insertions(+), 50 deletions(-) create mode 100644 pkg/rtc/rtcpthrottle.go diff --git a/cmd/server/main.go b/cmd/server/main.go index bb535f0b4..7089557e4 100644 --- a/cmd/server/main.go +++ b/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() { diff --git a/pkg/config/config.go b/pkg/config/config.go index ffc3fc9cf..741d4df4b 100644 --- a/pkg/config/config.go +++ b/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, diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 41eab98a1..90dd5c91c 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -7,7 +7,6 @@ import ( "sync/atomic" "time" - "github.com/frostbyte73/go-throttle" "github.com/pion/ion-sfu/pkg/sfu" "github.com/pion/ion-sfu/pkg/twcc" "github.com/pion/rtcp" @@ -38,6 +37,7 @@ type ParticipantParams struct { AudioConfig config.AudioConfig ProtocolVersion types.ProtocolVersion Stats *RoomStatsReporter + ThrottleConfig config.RTCPThrottleConfig } type ParticipantImpl struct { @@ -88,12 +88,10 @@ func NewParticipant(params ParticipantParams) (*ParticipantImpl, error) { // TODO: check to ensure params are valid, id and identity can't be empty p := &ParticipantImpl{ - params: params, - id: utils.NewGuid(utils.ParticipantPrefix), - rtcpCh: make(chan []rtcp.Packet, 50), - rtcpThrottle: &rtcpThrottle{ - throttles: make(map[uint32]func(func())), - }, + 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), @@ -741,7 +739,7 @@ func (p *ParticipantImpl) onMediaTrack(track *webrtc.TrackRemote, rtpReceiver *w } ssrc := uint32(track.SSRC()) - p.rtcpThrottle.addTrack(ssrc) + p.rtcpThrottle.addTrack(ssrc, track.RID()) if p.twcc == nil { p.twcc = twcc.NewTransportWideCCResponder(ssrc) p.twcc.OnFeedback(func(pkt rtcp.RawPacket) { @@ -968,44 +966,3 @@ func (p *ParticipantImpl) rtcpSendWorker() { } } } - -type rtcpThrottle struct { - mu sync.RWMutex - throttles map[uint32]func(func()) -} - -func (t *rtcpThrottle) addTrack(ssrc uint32) { - t.mu.Lock() - defer t.mu.Unlock() - - t.throttles[ssrc] = throttle.New(time.Millisecond * 500) -} - -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) - } -} diff --git a/pkg/rtc/participant_internal_test.go b/pkg/rtc/participant_internal_test.go index 0558b9abb..c4150acc9 100644 --- a/pkg/rtc/participant_internal_test.go +++ b/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 } diff --git a/pkg/rtc/rtcpthrottle.go b/pkg/rtc/rtcpthrottle.go new file mode 100644 index 000000000..9ca644a29 --- /dev/null +++ b/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) + } +} diff --git a/pkg/service/roommanager.go b/pkg/service/roommanager.go index 09ff82c54..efe62d22a 100644 --- a/pkg/service/roommanager.go +++ b/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)