throttle quality come down (#556)

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,

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,
 	})

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,
 	})

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

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,

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,