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/go.mod b/go.mod index d79288695..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 @@ -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 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/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/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..90dd5c91c 100644 --- a/pkg/rtc/participant.go +++ b/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) + } + } } 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) 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