From b98b828618606fe4f7e3b25b3a13dd1342e39b61 Mon Sep 17 00:00:00 2001 From: cnderrauber Date: Wed, 23 Mar 2022 15:27:00 +0800 Subject: [PATCH] throttle quality come down (#556) --- pkg/config/config.go | 8 ++++++++ pkg/rtc/mediatrack.go | 2 ++ pkg/rtc/mediatrackreceiver.go | 3 +++ pkg/rtc/mediatracksubscriptions.go | 27 +++++++++++++++++++++------ pkg/rtc/participant.go | 2 ++ pkg/service/roommanager.go | 1 + 6 files changed, 37 insertions(+), 6 deletions(-) diff --git a/pkg/config/config.go b/pkg/config/config.go index 641ca8244..1b13a1aec 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -30,6 +30,7 @@ type Config struct { RTC RTCConfig `yaml:"rtc,omitempty"` Redis RedisConfig `yaml:"redis,omitempty"` Audio AudioConfig `yaml:"audio,omitempty"` + Video VideoConfig `yaml:"video,omitempty"` Room RoomConfig `yaml:"room,omitempty"` TURN TURNConfig `yaml:"turn,omitempty"` WebHook WebHookConfig `yaml:"webhook,omitempty"` @@ -112,6 +113,10 @@ type AudioConfig struct { SmoothIntervals uint32 `yaml:"smooth_intervals"` } +type VideoConfig struct { + SubscribedQualityUpdateThrottle time.Duration `yaml:"subscribed_quality_update_throttle,omitempty"` +} + type RedisConfig struct { Address string `yaml:"address"` Username string `yaml:"username"` @@ -207,6 +212,9 @@ func NewConfig(confString string, c *cli.Context) (*Config, error) { UpdateInterval: 400, SmoothIntervals: 2, }, + Video: VideoConfig{ + SubscribedQualityUpdateThrottle: 5 * time.Second, + }, Redis: RedisConfig{}, Room: RoomConfig{ AutoCreate: true, diff --git a/pkg/rtc/mediatrack.go b/pkg/rtc/mediatrack.go index 60e1be0b0..59857ec5c 100644 --- a/pkg/rtc/mediatrack.go +++ b/pkg/rtc/mediatrack.go @@ -48,6 +48,7 @@ type MediaTrackParams struct { SubscriberConfig DirectionConfig PLIThrottleConfig config.PLIThrottleConfig AudioConfig config.AudioConfig + VideoConfig config.VideoConfig Telemetry telemetry.TelemetryService Logger logger.Logger } @@ -65,6 +66,7 @@ func NewMediaTrack(params MediaTrackParams) *MediaTrack { BufferFactory: params.BufferFactory, ReceiverConfig: params.ReceiverConfig, SubscriberConfig: params.SubscriberConfig, + VideoConfig: params.VideoConfig, Telemetry: params.Telemetry, Logger: params.Logger, }) diff --git a/pkg/rtc/mediatrackreceiver.go b/pkg/rtc/mediatrackreceiver.go index f2edb47b0..1f4b38964 100644 --- a/pkg/rtc/mediatrackreceiver.go +++ b/pkg/rtc/mediatrackreceiver.go @@ -6,6 +6,7 @@ import ( "sync" "time" + "github.com/livekit/livekit-server/pkg/config" "github.com/livekit/livekit-server/pkg/rtc/types" "go.uber.org/atomic" @@ -52,6 +53,7 @@ type MediaTrackReceiverParams struct { BufferFactory *buffer.Factory ReceiverConfig ReceiverConfig SubscriberConfig DirectionConfig + VideoConfig config.VideoConfig Telemetry telemetry.TelemetryService Logger logger.Logger } @@ -66,6 +68,7 @@ func NewMediaTrackReceiver(params MediaTrackReceiverParams) *MediaTrackReceiver BufferFactory: params.BufferFactory, ReceiverConfig: params.ReceiverConfig, SubscriberConfig: params.SubscriberConfig, + VideoConfig: t.params.VideoConfig, Telemetry: params.Telemetry, Logger: params.Logger, }) diff --git a/pkg/rtc/mediatracksubscriptions.go b/pkg/rtc/mediatracksubscriptions.go index 7f54e09d7..a97563baf 100644 --- a/pkg/rtc/mediatracksubscriptions.go +++ b/pkg/rtc/mediatracksubscriptions.go @@ -6,12 +6,14 @@ import ( "sync" "time" + "github.com/bep/debounce" "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/logger" "github.com/pion/rtcp" "github.com/pion/webrtc/v3" "github.com/pion/webrtc/v3/pkg/rtcerr" + "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/buffer" @@ -37,6 +39,7 @@ type MediaTrackSubscriptions struct { maxSubscriberQuality map[livekit.ParticipantID]livekit.VideoQuality maxSubscriberNodeQuality map[livekit.NodeID]livekit.VideoQuality maxSubscribedQuality livekit.VideoQuality + maxSubscribedQualityDebounce func(func()) onSubscribedMaxQualityChange func(subscribedQualities []*livekit.SubscribedQuality, maxSubscribedQuality livekit.VideoQuality) maxQualityTimer *time.Timer } @@ -47,6 +50,7 @@ type MediaTrackSubscriptionsParams struct { BufferFactory *buffer.Factory ReceiverConfig ReceiverConfig SubscriberConfig DirectionConfig + VideoConfig config.VideoConfig Telemetry telemetry.TelemetryService @@ -55,12 +59,13 @@ type MediaTrackSubscriptionsParams struct { func NewMediaTrackSubscriptions(params MediaTrackSubscriptionsParams) *MediaTrackSubscriptions { t := &MediaTrackSubscriptions{ - params: params, - subscribedTracks: make(map[livekit.ParticipantID]types.SubscribedTrack), - pendingClose: make(map[livekit.ParticipantID]types.SubscribedTrack), - maxSubscriberQuality: make(map[livekit.ParticipantID]livekit.VideoQuality), - maxSubscriberNodeQuality: make(map[livekit.NodeID]livekit.VideoQuality), - maxSubscribedQuality: livekit.VideoQuality_LOW, + params: params, + subscribedTracks: make(map[livekit.ParticipantID]types.SubscribedTrack), + pendingClose: make(map[livekit.ParticipantID]types.SubscribedTrack), + maxSubscriberQuality: make(map[livekit.ParticipantID]livekit.VideoQuality), + maxSubscriberNodeQuality: make(map[livekit.NodeID]livekit.VideoQuality), + maxSubscribedQuality: livekit.VideoQuality_LOW, + maxSubscribedQualityDebounce: debounce.New(params.VideoConfig.SubscribedQualityUpdateThrottle), } return t @@ -506,6 +511,16 @@ func (t *MediaTrackSubscriptions) UpdateQualityChange(force bool) { return } + if (t.maxSubscribedQuality != livekit.VideoQuality_OFF && maxSubscribedQuality != livekit.VideoQuality_OFF) && + t.maxSubscribedQuality > maxSubscribedQuality && !force { + t.params.Logger.Debugw("throttle quality change", "from", t.maxSubscribedQuality, "to", maxSubscribedQuality) + t.maxQualityLock.Unlock() + t.maxSubscribedQualityDebounce(func() { + t.UpdateQualityChange(true) + }) + return + } + t.maxSubscribedQuality = maxSubscribedQuality var subscribedQualities []*livekit.SubscribedQuality diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 22a80b04b..5fba3fec2 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -51,6 +51,7 @@ type ParticipantParams struct { Config *WebRTCConfig Sink routing.MessageSink AudioConfig config.AudioConfig + VideoConfig config.VideoConfig ProtocolVersion types.ProtocolVersion Telemetry telemetry.TelemetryService PLIThrottleConfig config.PLIThrottleConfig @@ -1353,6 +1354,7 @@ func (p *ParticipantImpl) mediaTrackReceived(track *webrtc.TrackRemote, rtpRecei BufferFactory: p.params.Config.BufferFactory, ReceiverConfig: p.params.Config.Receiver, AudioConfig: p.params.AudioConfig, + VideoConfig: p.params.VideoConfig, Telemetry: p.params.Telemetry, Logger: LoggerWithTrack(p.params.Logger, livekit.TrackID(ti.Sid)), SubscriberConfig: p.params.Config.Subscriber, diff --git a/pkg/service/roommanager.go b/pkg/service/roommanager.go index 7298ed85e..ccd800b40 100644 --- a/pkg/service/roommanager.go +++ b/pkg/service/roommanager.go @@ -246,6 +246,7 @@ func (r *RoomManager) StartSession(ctx context.Context, roomName livekit.RoomNam Config: &rtcConf, Sink: responseSink, AudioConfig: r.config.Audio, + VideoConfig: r.config.Video, ProtocolVersion: pv, Telemetry: r.telemetry, PLIThrottleConfig: r.config.RTC.PLIThrottle,