Improve video quality selection by using publisher feedback (#238)

go.mod

@@ -15,7 +15,7 @@ require (
 	github.com/gorilla/websocket v1.4.2
 	github.com/hashicorp/golang-lru v0.5.4
 	github.com/jxskiss/base62 v0.0.0-20191017122030-4f11678b909b
-	github.com/livekit/protocol v0.10.4
+	github.com/livekit/protocol v0.10.5
 	github.com/magefile/mage v1.11.0
 	github.com/maxbrunsfeld/counterfeiter/v6 v6.3.0
 	github.com/mitchellh/go-homedir v1.1.0

go.sum

@@ -132,8 +132,8 @@ github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
 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/protocol v0.10.4 h1:pn4Du5o8EfDpjHPhkjez8qHRYzNNy6ZP2c0T87cQKc0=
-github.com/livekit/protocol v0.10.4/go.mod h1:YoHW9YbWbPnuVsgwBB4hAINKT+V68jmfh9zXBSSn6Wg=
+github.com/livekit/protocol v0.10.5 h1:cFGhMH9dPckX7zBfmZNtHq2B+gocYs7PV9BkQU1Wi1I=
+github.com/livekit/protocol v0.10.5/go.mod h1:YoHW9YbWbPnuVsgwBB4hAINKT+V68jmfh9zXBSSn6Wg=
 github.com/magefile/mage v1.11.0 h1:C/55Ywp9BpgVVclD3lRnSYCwXTYxmSppIgLeDYlNuls=
 github.com/magefile/mage v1.11.0/go.mod h1:z5UZb/iS3GoOSn0JgWuiw7dxlurVYTu+/jHXqQg881A=
 github.com/mattn/go-runewidth v0.0.9 h1:Lm995f3rfxdpd6TSmuVCHVb/QhupuXlYr8sCI/QdE+0=

pkg/rtc/mediatrack.go

@@ -3,6 +3,7 @@ package rtc
 import (
 	"context"
 	"errors"
+	"sort"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -30,7 +31,8 @@ var (
 )
 
 const (
-	lostUpdateDelta = time.Second
+	lostUpdateDelta         = time.Second
+	layerSelectionTolerance = 0.8
 )
 
 // MediaTrack represents a WebRTC track that needs to be forwarded
@@ -47,12 +49,13 @@ type MediaTrack struct {
 
 	// channel to send RTCP packets to the source
 	lock sync.RWMutex
-	// map of target participantId -> *SubscribedTrack
-	subscribedTracks map[string]*SubscribedTrack
+	// map of target participantId -> types.SubscribedTrack
+	subscribedTracks sync.Map
 	twcc             *twcc.Responder
 	audioLevel       *AudioLevel
 	receiver         sfu.Receiver
 	lastPLI          time.Time
+	layerDimensions  sync.Map // quality => *livekit.VideoLayer
 
 	// track audio fraction lost
 	fracLostLock      sync.Mutex
@@ -81,16 +84,19 @@ type MediaTrackParams struct {
 
 func NewMediaTrack(track *webrtc.TrackRemote, params MediaTrackParams) *MediaTrack {
 	t := &MediaTrack{
-		params:           params,
-		ssrc:             track.SSRC(),
-		streamID:         track.StreamID(),
-		codec:            track.Codec(),
-		subscribedTracks: make(map[string]*SubscribedTrack),
+		params:   params,
+		ssrc:     track.SSRC(),
+		streamID: track.StreamID(),
+		codec:    track.Codec(),
 	}
 
 	if params.TrackInfo.Muted {
 		t.SetMuted(true)
 	}
+
+	if params.TrackInfo != nil && t.Kind() == livekit.TrackType_VIDEO {
+		t.UpdateVideoLayers(params.TrackInfo.Layers)
+	}
 	return t
 }
 
@@ -128,11 +134,15 @@ func (t *MediaTrack) SetMuted(muted bool) {
 	if t.receiver != nil {
 		t.receiver.SetUpTrackPaused(muted)
 	}
-	// mute all subscribed tracks
-	for _, st := range t.subscribedTracks {
-		st.SetPublisherMuted(muted)
-	}
 	t.lock.RUnlock()
+
+	// mute all subscribed tracks
+	t.subscribedTracks.Range(func(_, value interface{}) bool {
+		if st, ok := value.(types.SubscribedTrack); ok {
+			st.SetPublisherMuted(muted)
+		}
+		return true
+	})
 }
 
 func (t *MediaTrack) AddOnClose(f func()) {
@@ -143,9 +153,8 @@ func (t *MediaTrack) AddOnClose(f func()) {
 }
 
 func (t *MediaTrack) IsSubscriber(subId string) bool {
-	t.lock.RLock()
-	defer t.lock.RUnlock()
-	return t.subscribedTracks[subId] != nil
+	_, ok := t.subscribedTracks.Load(subId)
+	return ok
 }
 
 func (t *MediaTrack) PublishLossPercentage() uint32 {
@@ -160,10 +169,9 @@ func (t *MediaTrack) AddSubscriber(sub types.Participant) error {
 
 	t.lock.Lock()
 	defer t.lock.Unlock()
-	existingSt := t.subscribedTracks[sub.ID()]
 
 	// don't subscribe to the same track multiple times
-	if existingSt != nil {
+	if _, ok := t.subscribedTracks.Load(sub.ID()); ok {
 		return nil
 	}
 
@@ -191,7 +199,7 @@ func (t *MediaTrack) AddSubscriber(sub types.Participant) error {
 	if err != nil {
 		return err
 	}
-	subTrack := NewSubscribedTrack(t.params.ParticipantIdentity, downTrack)
+	subTrack := NewSubscribedTrack(t, t.params.ParticipantIdentity, downTrack)
 
 	var transceiver *webrtc.RTPTransceiver
 	var sender *webrtc.RTPSender
@@ -249,10 +257,7 @@ func (t *MediaTrack) AddSubscriber(sub types.Participant) error {
 
 	downTrack.OnCloseHandler(func() {
 		go func() {
-			t.lock.Lock()
-			delete(t.subscribedTracks, sub.ID())
-			t.lock.Unlock()
-
+			t.subscribedTracks.Delete(sub.ID())
 			t.params.Telemetry.TrackUnsubscribed(context.Background(), sub.ID(), t.ToProto())
 
 			// ignore if the subscribing sub is not connected
@@ -293,7 +298,7 @@ func (t *MediaTrack) AddSubscriber(sub types.Participant) error {
 		downTrack.AddReceiverReportListener(t.handleMaxLossFeedback)
 	}
 
-	t.subscribedTracks[sub.ID()] = subTrack
+	t.subscribedTracks.Store(sub.ID(), subTrack)
 	subTrack.SetPublisherMuted(t.IsMuted())
 
 	t.receiver.AddDownTrack(downTrack)
@@ -406,10 +411,8 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track *webrtc.Tra
 // RemoveSubscriber removes participant from subscription
 // stop all forwarders to the client
 func (t *MediaTrack) RemoveSubscriber(participantId string) {
-	t.lock.RLock()
-	defer t.lock.RUnlock()
-
-	if subTrack := t.subscribedTracks[participantId]; subTrack != nil {
+	subTrack := t.getSubscribedTrack(participantId)
+	if subTrack != nil {
 		go subTrack.DownTrack().Close()
 	}
 }
@@ -418,21 +421,46 @@ func (t *MediaTrack) RemoveAllSubscribers() {
 	t.params.Logger.Debugw("removing all subscribers", "track", t.ID())
 	t.lock.Lock()
 	defer t.lock.Unlock()
-	for _, subTrack := range t.subscribedTracks {
-		go subTrack.DownTrack().Close()
-	}
-	t.subscribedTracks = make(map[string]*SubscribedTrack)
+	t.subscribedTracks.Range(func(_, val interface{}) bool {
+		if subTrack, ok := val.(types.SubscribedTrack); ok {
+			go subTrack.DownTrack().Close()
+		}
+		return true
+	})
+	t.subscribedTracks = sync.Map{}
 }
 
 func (t *MediaTrack) ToProto() *livekit.TrackInfo {
 	info := t.params.TrackInfo
 	info.Muted = t.IsMuted()
 	info.Simulcast = t.simulcasted.Get()
+	layers := make([]*livekit.VideoLayer, 0)
+	t.layerDimensions.Range(func(_, val interface{}) bool {
+		if layer, ok := val.(*livekit.VideoLayer); ok {
+			layers = append(layers, layer)
+		}
+		return true
+	})
+	info.Layers = layers
+
 	return info
 }
 
+func (t *MediaTrack) UpdateVideoLayers(layers []*livekit.VideoLayer) {
+	for _, layer := range layers {
+		t.layerDimensions.Store(layer.Quality, layer)
+	}
+	t.subscribedTracks.Range(func(_, val interface{}) bool {
+		if st, ok := val.(types.SubscribedTrack); ok {
+			st.UpdateVideoLayer()
+		}
+		return true
+	})
+	// TODO: this might need to trigger a participant update for clients to pick up dimension change
+}
+
 // GetQualityForDimension finds the closest quality to use for desired dimensions
-// affords a 10% tolerance on dimension
+// affords a 20% tolerance on dimension
 func (t *MediaTrack) GetQualityForDimension(width, height uint32) livekit.VideoQuality {
 	quality := livekit.VideoQuality_HIGH
 	if t.Kind() == livekit.TrackType_AUDIO || t.params.TrackInfo.Height == 0 {
@@ -446,11 +474,26 @@ func (t *MediaTrack) GetQualityForDimension(width, height uint32) livekit.VideoQ
 		requestedSize = width
 	}
 
-	// representing qualities low - high
+	// default sizes representing qualities low - high
 	layerSizes := []uint32{180, 360, origSize}
+	var providedSizes []uint32
+	t.layerDimensions.Range(func(_, val interface{}) bool {
+		if layer, ok := val.(*livekit.VideoLayer); ok {
+			providedSizes = append(providedSizes, layer.Height)
+		}
+		return true
+	})
+	if len(providedSizes) > 0 {
+		layerSizes = providedSizes
+		// comparing height always
+		requestedSize = height
+		sort.Slice(layerSizes, func(i, j int) bool {
+			return layerSizes[i] < layerSizes[j]
+		})
+	}
 
 	// finds the lowest layer that could satisfy client demands
-	requestedSize = uint32(float32(requestedSize) * 0.9)
+	requestedSize = uint32(float32(requestedSize) * layerSelectionTolerance)
 	for i, s := range layerSizes {
 		quality = livekit.VideoQuality(i)
 		if s >= requestedSize {
@@ -461,15 +504,21 @@ func (t *MediaTrack) GetQualityForDimension(width, height uint32) livekit.VideoQ
 	return quality
 }
 
+func (t *MediaTrack) getSubscribedTrack(id string) types.SubscribedTrack {
+	if val, ok := t.subscribedTracks.Load(id); ok {
+		if st, ok := val.(types.SubscribedTrack); ok {
+			return st
+		}
+	}
+	return nil
+}
+
 // TODO: send for all downtracks from the source participant
 // https://tools.ietf.org/html/rfc7941
 func (t *MediaTrack) sendDownTrackBindingReports(sub types.Participant) {
 	var sd []rtcp.SourceDescriptionChunk
 
-	t.lock.RLock()
-	subTrack := t.subscribedTracks[sub.ID()]
-	t.lock.RUnlock()
-
+	subTrack := t.getSubscribedTrack(sub.ID())
 	if subTrack == nil {
 		return
 	}
@@ -575,14 +624,15 @@ func (t *MediaTrack) DebugInfo() map[string]interface{} {
 	}
 
 	subscribedTrackInfo := make([]map[string]interface{}, 0)
-	t.lock.RLock()
-	for _, track := range t.subscribedTracks {
-		dt := track.dt.DebugInfo()
-		dt["PubMuted"] = track.pubMuted.Get()
-		dt["SubMuted"] = track.subMuted.Get()
-		subscribedTrackInfo = append(subscribedTrackInfo, dt)
-	}
-	t.lock.RUnlock()
+	t.subscribedTracks.Range(func(_, val interface{}) bool {
+		if track, ok := val.(*SubscribedTrack); ok {
+			dt := track.dt.DebugInfo()
+			dt["PubMuted"] = track.pubMuted.Get()
+			dt["SubMuted"] = track.subMuted.Get()
+			subscribedTrackInfo = append(subscribedTrackInfo, dt)
+		}
+		return true
+	})
 	info["DownTracks"] = subscribedTrackInfo
 
 	if t.receiver != nil {

pkg/rtc/mediatrack_test.go

@@ -69,4 +69,34 @@ func TestGetQualityForDimension(t *testing.T) {
 		require.Equal(t, livekit.VideoQuality_MEDIUM, mt.GetQualityForDimension(400, 700))
 		require.Equal(t, livekit.VideoQuality_HIGH, mt.GetQualityForDimension(600, 900))
 	})
+
+	t.Run("layers provided", func(t *testing.T) {
+		mt := NewMediaTrack(&webrtc.TrackRemote{}, MediaTrackParams{TrackInfo: &livekit.TrackInfo{
+			Type:   livekit.TrackType_VIDEO,
+			Width:  1080,
+			Height: 720,
+			Layers: []*livekit.VideoLayer{
+				{
+					Quality: livekit.VideoQuality_LOW,
+					Width:   480,
+					Height:  270,
+				},
+				{
+					Quality: livekit.VideoQuality_MEDIUM,
+					Width:   960,
+					Height:  540,
+				},
+				{
+					Quality: livekit.VideoQuality_HIGH,
+					Width:   1080,
+					Height:  720,
+				},
+			},
+		}})
+
+		require.Equal(t, livekit.VideoQuality_LOW, mt.GetQualityForDimension(120, 120))
+		require.Equal(t, livekit.VideoQuality_LOW, mt.GetQualityForDimension(300, 300))
+		require.Equal(t, livekit.VideoQuality_MEDIUM, mt.GetQualityForDimension(800, 500))
+		require.Equal(t, livekit.VideoQuality_HIGH, mt.GetQualityForDimension(1000, 700))
+	})
 }

pkg/rtc/participant.go

@@ -365,6 +365,7 @@ func (p *ParticipantImpl) AddTrack(req *livekit.AddTrackRequest) {
 		Muted:      req.Muted,
 		DisableDtx: req.DisableDtx,
 		Source:     req.Source,
+		Layers:     req.Layers,
 	}
 	p.pendingTracks[req.Cid] = ti
 
@@ -498,15 +499,6 @@ func (p *ParticipantImpl) AddSubscriber(op types.Participant) (int, error) {
 	return n, nil
 }
 
-func (p *ParticipantImpl) RemoveSubscriber(participantId string) {
-	p.lock.RLock()
-	defer p.lock.RUnlock()
-
-	for _, track := range p.publishedTracks {
-		track.RemoveSubscriber(participantId)
-	}
-}
-
 // signal connection methods
 
 func (p *ParticipantImpl) SendJoinResponse(

pkg/rtc/plithrottle.go

@@ -5,6 +5,7 @@ import (
 	"time"
 
 	"github.com/livekit/livekit-server/pkg/config"
+	"github.com/livekit/livekit-server/pkg/sfu"
 )
 
 type pliThrottle struct {
@@ -14,13 +15,6 @@ type pliThrottle struct {
 	lastSent map[uint32]int64
 }
 
-// github.com/livekit/livekit-server/pkg/sfu/simulcast.go
-const (
-	fullResolution    = "f"
-	halfResolution    = "h"
-	quarterResolution = "q"
-)
-
 func newPLIThrottle(conf config.PLIThrottleConfig) *pliThrottle {
 	return &pliThrottle{
 		config:   conf,
@@ -35,11 +29,11 @@ func (t *pliThrottle) addTrack(ssrc uint32, rid string) {
 
 	var duration time.Duration
 	switch rid {
-	case fullResolution:
+	case sfu.FullResolution:
 		duration = t.config.HighQuality
-	case halfResolution:
+	case sfu.HalfResolution:
 		duration = t.config.MidQuality
-	case quarterResolution:
+	case sfu.QuarterResolution:
 		duration = t.config.LowQuality
 	default:
 		duration = t.config.MidQuality

pkg/rtc/subscribedtrack.go

@@ -1,9 +1,11 @@
 package rtc
 
 import (
+	"sync/atomic"
 	"time"
 
 	"github.com/bep/debounce"
+	"github.com/livekit/livekit-server/pkg/rtc/types"
 	"github.com/livekit/livekit-server/pkg/sfu"
 	livekit "github.com/livekit/protocol/proto"
 	"github.com/livekit/protocol/utils"
@@ -15,16 +17,19 @@ const (
 )
 
 type SubscribedTrack struct {
+	publishedTrack    types.PublishedTrack
 	dt                *sfu.DownTrack
 	publisherIdentity string
 	subMuted          utils.AtomicFlag
 	pubMuted          utils.AtomicFlag
+	settings          atomic.Value // *livekit.UpdateTrackSettings
 
 	debouncer func(func())
 }
 
-func NewSubscribedTrack(publisherIdentity string, dt *sfu.DownTrack) *SubscribedTrack {
+func NewSubscribedTrack(publishedTrack types.PublishedTrack, publisherIdentity string, dt *sfu.DownTrack) *SubscribedTrack {
 	return &SubscribedTrack{
+		publishedTrack:    publishedTrack,
 		publisherIdentity: publisherIdentity,
 		dt:                dt,
 		debouncer:         debounce.New(subscriptionDebounceInterval),
@@ -57,14 +62,32 @@ func (t *SubscribedTrack) SetPublisherMuted(muted bool) {
 	t.updateDownTrackMute()
 }
 
-func (t *SubscribedTrack) UpdateSubscriberSettings(enabled bool, quality livekit.VideoQuality) {
-	t.debouncer(func() {
-		t.subMuted.TrySet(!enabled)
-		t.updateDownTrackMute()
-		if enabled && t.dt.Kind() == webrtc.RTPCodecTypeVideo {
-			t.dt.SetMaxSpatialLayer(spatialLayerForQuality(quality))
-		}
-	})
+func (t *SubscribedTrack) UpdateSubscriberSettings(settings *livekit.UpdateTrackSettings) {
+	visibilityChanged := t.subMuted.TrySet(settings.Disabled)
+	t.settings.Store(settings)
+	// avoid frequent changes to mute & video layers, unless it became visible
+	if visibilityChanged && !settings.Disabled {
+		t.UpdateVideoLayer()
+	} else {
+		t.debouncer(t.UpdateVideoLayer)
+	}
+}
+
+func (t *SubscribedTrack) UpdateVideoLayer() {
+	t.updateDownTrackMute()
+	if t.subMuted.Get() || t.dt.Kind() != webrtc.RTPCodecTypeVideo {
+		return
+	}
+	settings, ok := t.settings.Load().(*livekit.UpdateTrackSettings)
+	if !ok {
+		return
+	}
+
+	quality := settings.Quality
+	if settings.Width > 0 {
+		quality = t.publishedTrack.GetQualityForDimension(settings.Width, settings.Height)
+	}
+	t.dt.SetMaxSpatialLayer(spatialLayerForQuality(quality))
 }
 
 func (t *SubscribedTrack) updateDownTrackMute() {

pkg/rtc/types/interfaces.go

@@ -47,7 +47,6 @@ type Participant interface {
 	HandleAnswer(sdp webrtc.SessionDescription) error
 	AddICECandidate(candidate webrtc.ICECandidateInit, target livekit.SignalTarget) error
 	AddSubscriber(op Participant) (int, error)
-	RemoveSubscriber(peerId string)
 	SendJoinResponse(info *livekit.Room, otherParticipants []*livekit.ParticipantInfo, iceServers []*livekit.ICEServer) error
 	SendParticipantUpdate(participants []*livekit.ParticipantInfo, updatedAt time.Time) error
 	SendSpeakerUpdate(speakers []*livekit.SpeakerInfo) error
@@ -104,6 +103,9 @@ type PublishedTrack interface {
 	Name() string
 	IsMuted() bool
 	SetMuted(muted bool)
+	UpdateVideoLayers(layers []*livekit.VideoLayer)
+
+	// subscribers
 	AddSubscriber(participant Participant) error
 	RemoveSubscriber(participantId string)
 	IsSubscriber(subId string) bool
@@ -127,7 +129,9 @@ type SubscribedTrack interface {
 	DownTrack() *sfu.DownTrack
 	IsMuted() bool
 	SetPublisherMuted(muted bool)
-	UpdateSubscriberSettings(enabled bool, quality livekit.VideoQuality)
+	UpdateSubscriberSettings(settings *livekit.UpdateTrackSettings)
+	// selects appropriate video layer according to subscriber preferences
+	UpdateVideoLayer()
 	SubscribeLossPercentage() uint32
 }
 

pkg/rtc/types/typesfakes/fake_participant.go

@@ -346,11 +346,6 @@ type FakeParticipant struct {
 	removeSubscribedTrackArgsForCall []struct {
 		arg1 types.SubscribedTrack
 	}
-	RemoveSubscriberStub        func(string)
-	removeSubscriberMutex       sync.RWMutex
-	removeSubscriberArgsForCall []struct {
-		arg1 string
-	}
 	SendConnectionQualityUpdateStub        func(*livekit.ConnectionQualityUpdate) error
 	sendConnectionQualityUpdateMutex       sync.RWMutex
 	sendConnectionQualityUpdateArgsForCall []struct {
@@ -2309,38 +2304,6 @@ func (fake *FakeParticipant) RemoveSubscribedTrackArgsForCall(i int) types.Subsc
 	return argsForCall.arg1
 }
 
-func (fake *FakeParticipant) RemoveSubscriber(arg1 string) {
-	fake.removeSubscriberMutex.Lock()
-	fake.removeSubscriberArgsForCall = append(fake.removeSubscriberArgsForCall, struct {
-		arg1 string
-	}{arg1})
-	stub := fake.RemoveSubscriberStub
-	fake.recordInvocation("RemoveSubscriber", []interface{}{arg1})
-	fake.removeSubscriberMutex.Unlock()
-	if stub != nil {
-		fake.RemoveSubscriberStub(arg1)
-	}
-}
-
-func (fake *FakeParticipant) RemoveSubscriberCallCount() int {
-	fake.removeSubscriberMutex.RLock()
-	defer fake.removeSubscriberMutex.RUnlock()
-	return len(fake.removeSubscriberArgsForCall)
-}
-
-func (fake *FakeParticipant) RemoveSubscriberCalls(stub func(string)) {
-	fake.removeSubscriberMutex.Lock()
-	defer fake.removeSubscriberMutex.Unlock()
-	fake.RemoveSubscriberStub = stub
-}
-
-func (fake *FakeParticipant) RemoveSubscriberArgsForCall(i int) string {
-	fake.removeSubscriberMutex.RLock()
-	defer fake.removeSubscriberMutex.RUnlock()
-	argsForCall := fake.removeSubscriberArgsForCall[i]
-	return argsForCall.arg1
-}
-
 func (fake *FakeParticipant) SendConnectionQualityUpdate(arg1 *livekit.ConnectionQualityUpdate) error {
 	fake.sendConnectionQualityUpdateMutex.Lock()
 	ret, specificReturn := fake.sendConnectionQualityUpdateReturnsOnCall[len(fake.sendConnectionQualityUpdateArgsForCall)]
@@ -3226,8 +3189,6 @@ func (fake *FakeParticipant) Invocations() map[string][][]interface{} {
 	defer fake.rTCPChanMutex.RUnlock()
 	fake.removeSubscribedTrackMutex.RLock()
 	defer fake.removeSubscribedTrackMutex.RUnlock()
-	fake.removeSubscriberMutex.RLock()
-	defer fake.removeSubscriberMutex.RUnlock()
 	fake.sendConnectionQualityUpdateMutex.RLock()
 	defer fake.sendConnectionQualityUpdateMutex.RUnlock()
 	fake.sendDataPacketMutex.RLock()

pkg/rtc/types/typesfakes/fake_published_track.go

@@ -169,6 +169,11 @@ type FakePublishedTrack struct {
 	toProtoReturnsOnCall map[int]struct {
 		result1 *livekit.TrackInfo
 	}
+	UpdateVideoLayersStub        func([]*livekit.VideoLayer)
+	updateVideoLayersMutex       sync.RWMutex
+	updateVideoLayersArgsForCall []struct {
+		arg1 []*livekit.VideoLayer
+	}
 	invocations      map[string][][]interface{}
 	invocationsMutex sync.RWMutex
 }
@@ -1034,6 +1039,43 @@ func (fake *FakePublishedTrack) ToProtoReturnsOnCall(i int, result1 *livekit.Tra
 	}{result1}
 }
 
+func (fake *FakePublishedTrack) UpdateVideoLayers(arg1 []*livekit.VideoLayer) {
+	var arg1Copy []*livekit.VideoLayer
+	if arg1 != nil {
+		arg1Copy = make([]*livekit.VideoLayer, len(arg1))
+		copy(arg1Copy, arg1)
+	}
+	fake.updateVideoLayersMutex.Lock()
+	fake.updateVideoLayersArgsForCall = append(fake.updateVideoLayersArgsForCall, struct {
+		arg1 []*livekit.VideoLayer
+	}{arg1Copy})
+	stub := fake.UpdateVideoLayersStub
+	fake.recordInvocation("UpdateVideoLayers", []interface{}{arg1Copy})
+	fake.updateVideoLayersMutex.Unlock()
+	if stub != nil {
+		fake.UpdateVideoLayersStub(arg1)
+	}
+}
+
+func (fake *FakePublishedTrack) UpdateVideoLayersCallCount() int {
+	fake.updateVideoLayersMutex.RLock()
+	defer fake.updateVideoLayersMutex.RUnlock()
+	return len(fake.updateVideoLayersArgsForCall)
+}
+
+func (fake *FakePublishedTrack) UpdateVideoLayersCalls(stub func([]*livekit.VideoLayer)) {
+	fake.updateVideoLayersMutex.Lock()
+	defer fake.updateVideoLayersMutex.Unlock()
+	fake.UpdateVideoLayersStub = stub
+}
+
+func (fake *FakePublishedTrack) UpdateVideoLayersArgsForCall(i int) []*livekit.VideoLayer {
+	fake.updateVideoLayersMutex.RLock()
+	defer fake.updateVideoLayersMutex.RUnlock()
+	argsForCall := fake.updateVideoLayersArgsForCall[i]
+	return argsForCall.arg1
+}
+
 func (fake *FakePublishedTrack) Invocations() map[string][][]interface{} {
 	fake.invocationsMutex.RLock()
 	defer fake.invocationsMutex.RUnlock()
@@ -1073,6 +1115,8 @@ func (fake *FakePublishedTrack) Invocations() map[string][][]interface{} {
 	defer fake.startMutex.RUnlock()
 	fake.toProtoMutex.RLock()
 	defer fake.toProtoMutex.RUnlock()
+	fake.updateVideoLayersMutex.RLock()
+	defer fake.updateVideoLayersMutex.RUnlock()
 	copiedInvocations := map[string][][]interface{}{}
 	for key, value := range fake.invocations {
 		copiedInvocations[key] = value

pkg/rtc/types/typesfakes/fake_subscribed_track.go

@@ -65,11 +65,14 @@ type FakeSubscribedTrack struct {
 	subscribeLossPercentageReturnsOnCall map[int]struct {
 		result1 uint32
 	}
-	UpdateSubscriberSettingsStub        func(bool, livekit.VideoQuality)
+	UpdateSubscriberSettingsStub        func(*livekit.UpdateTrackSettings)
 	updateSubscriberSettingsMutex       sync.RWMutex
 	updateSubscriberSettingsArgsForCall []struct {
-		arg1 bool
-		arg2 livekit.VideoQuality
+		arg1 *livekit.UpdateTrackSettings
+	}
+	UpdateVideoLayerStub        func()
+	updateVideoLayerMutex       sync.RWMutex
+	updateVideoLayerArgsForCall []struct {
 	}
 	invocations      map[string][][]interface{}
 	invocationsMutex sync.RWMutex
@@ -372,17 +375,16 @@ func (fake *FakeSubscribedTrack) SubscribeLossPercentageReturnsOnCall(i int, res
 	}{result1}
 }
 
-func (fake *FakeSubscribedTrack) UpdateSubscriberSettings(arg1 bool, arg2 livekit.VideoQuality) {
+func (fake *FakeSubscribedTrack) UpdateSubscriberSettings(arg1 *livekit.UpdateTrackSettings) {
 	fake.updateSubscriberSettingsMutex.Lock()
 	fake.updateSubscriberSettingsArgsForCall = append(fake.updateSubscriberSettingsArgsForCall, struct {
-		arg1 bool
-		arg2 livekit.VideoQuality
-	}{arg1, arg2})
+		arg1 *livekit.UpdateTrackSettings
+	}{arg1})
 	stub := fake.UpdateSubscriberSettingsStub
-	fake.recordInvocation("UpdateSubscriberSettings", []interface{}{arg1, arg2})
+	fake.recordInvocation("UpdateSubscriberSettings", []interface{}{arg1})
 	fake.updateSubscriberSettingsMutex.Unlock()
 	if stub != nil {
-		fake.UpdateSubscriberSettingsStub(arg1, arg2)
+		fake.UpdateSubscriberSettingsStub(arg1)
 	}
 }
 
@@ -392,17 +394,41 @@ func (fake *FakeSubscribedTrack) UpdateSubscriberSettingsCallCount() int {
 	return len(fake.updateSubscriberSettingsArgsForCall)
 }
 
-func (fake *FakeSubscribedTrack) UpdateSubscriberSettingsCalls(stub func(bool, livekit.VideoQuality)) {
+func (fake *FakeSubscribedTrack) UpdateSubscriberSettingsCalls(stub func(*livekit.UpdateTrackSettings)) {
 	fake.updateSubscriberSettingsMutex.Lock()
 	defer fake.updateSubscriberSettingsMutex.Unlock()
 	fake.UpdateSubscriberSettingsStub = stub
 }
 
-func (fake *FakeSubscribedTrack) UpdateSubscriberSettingsArgsForCall(i int) (bool, livekit.VideoQuality) {
+func (fake *FakeSubscribedTrack) UpdateSubscriberSettingsArgsForCall(i int) *livekit.UpdateTrackSettings {
 	fake.updateSubscriberSettingsMutex.RLock()
 	defer fake.updateSubscriberSettingsMutex.RUnlock()
 	argsForCall := fake.updateSubscriberSettingsArgsForCall[i]
-	return argsForCall.arg1, argsForCall.arg2
+	return argsForCall.arg1
+}
+
+func (fake *FakeSubscribedTrack) UpdateVideoLayer() {
+	fake.updateVideoLayerMutex.Lock()
+	fake.updateVideoLayerArgsForCall = append(fake.updateVideoLayerArgsForCall, struct {
+	}{})
+	stub := fake.UpdateVideoLayerStub
+	fake.recordInvocation("UpdateVideoLayer", []interface{}{})
+	fake.updateVideoLayerMutex.Unlock()
+	if stub != nil {
+		fake.UpdateVideoLayerStub()
+	}
+}
+
+func (fake *FakeSubscribedTrack) UpdateVideoLayerCallCount() int {
+	fake.updateVideoLayerMutex.RLock()
+	defer fake.updateVideoLayerMutex.RUnlock()
+	return len(fake.updateVideoLayerArgsForCall)
+}
+
+func (fake *FakeSubscribedTrack) UpdateVideoLayerCalls(stub func()) {
+	fake.updateVideoLayerMutex.Lock()
+	defer fake.updateVideoLayerMutex.Unlock()
+	fake.UpdateVideoLayerStub = stub
 }
 
 func (fake *FakeSubscribedTrack) Invocations() map[string][][]interface{} {
@@ -422,6 +448,8 @@ func (fake *FakeSubscribedTrack) Invocations() map[string][][]interface{} {
 	defer fake.subscribeLossPercentageMutex.RUnlock()
 	fake.updateSubscriberSettingsMutex.RLock()
 	defer fake.updateSubscriberSettingsMutex.RUnlock()
+	fake.updateVideoLayerMutex.RLock()
+	defer fake.updateVideoLayerMutex.RUnlock()
 	copiedInvocations := map[string][][]interface{}{}
 	for key, value := range fake.invocations {
 		copiedInvocations[key] = value

pkg/service/roommanager.go

@@ -361,118 +361,130 @@ func (r *RoomManager) rtcSessionWorker(room *rtc.Room, participant types.Partici
 			}
 
 			req := obj.(*livekit.SignalRequest)
-
-			switch msg := req.Message.(type) {
-			case *livekit.SignalRequest_Offer:
-				_, err := participant.HandleOffer(rtc.FromProtoSessionDescription(msg.Offer))
-				if err != nil {
-					logger.Errorw("could not handle offer", err,
-						"room", room.Room.Name,
-						"participant", participant.Identity(),
-						"pID", participant.ID(),
-					)
-					return
-				}
-			case *livekit.SignalRequest_AddTrack:
-				logger.Debugw("add track request",
-					"room", room.Room.Name,
-					"participant", participant.Identity(),
-					"pID", participant.ID(),
-					"track", msg.AddTrack.Cid)
-				participant.AddTrack(msg.AddTrack)
-			case *livekit.SignalRequest_Answer:
-				sd := rtc.FromProtoSessionDescription(msg.Answer)
-				if err := participant.HandleAnswer(sd); err != nil {
-					logger.Errorw("could not handle answer", err,
-						"room", room.Room.Name,
-						"participant", participant.Identity(),
-						"pID", participant.ID(),
-					)
-				}
-			case *livekit.SignalRequest_Trickle:
-				candidateInit, err := rtc.FromProtoTrickle(msg.Trickle)
-				if err != nil {
-					logger.Errorw("could not decode trickle", err,
-						"room", room.Room.Name,
-						"participant", participant.Identity(),
-						"pID", participant.ID(),
-					)
-					break
-				}
-				// logger.Debugw("adding peer candidate", "participant", participant.Identity())
-				if err := participant.AddICECandidate(candidateInit, msg.Trickle.Target); err != nil {
-					logger.Errorw("could not handle trickle", err,
-						"room", room.Room.Name,
-						"participant", participant.Identity(),
-						"pID", participant.ID(),
-					)
-				}
-			case *livekit.SignalRequest_Mute:
-				participant.SetTrackMuted(msg.Mute.Sid, msg.Mute.Muted, false)
-			case *livekit.SignalRequest_Subscription:
-				if err := room.UpdateSubscriptions(participant, msg.Subscription.TrackSids, msg.Subscription.Subscribe); err != nil {
-					logger.Warnw("could not update subscription", err,
-						"room", room.Room.Name,
-						"participant", participant.Identity(),
-						"pID", participant.ID(),
-						"tracks", msg.Subscription.TrackSids,
-						"subscribe", msg.Subscription.Subscribe)
-				}
-			case *livekit.SignalRequest_TrackSetting:
-				for _, sid := range msg.TrackSetting.TrackSids {
-					subTrack := participant.GetSubscribedTrack(sid)
-					if subTrack == nil {
-						logger.Warnw("unable to find SubscribedTrack", nil,
-							"room", room.Room.Name,
-							"participant", participant.Identity(),
-							"pID", participant.ID(),
-							"track", sid)
-						continue
-					}
-
-					// find the source PublishedTrack
-					publisher := room.GetParticipant(subTrack.PublisherIdentity())
-					if publisher == nil {
-						logger.Warnw("unable to find publisher of SubscribedTrack", nil,
-							"room", room.Room.Name,
-							"participant", participant.Identity(),
-							"pID", participant.ID(),
-							"publisher", subTrack.PublisherIdentity(),
-							"track", sid)
-						continue
-					}
-
-					pubTrack := publisher.GetPublishedTrack(sid)
-					if pubTrack == nil {
-						logger.Warnw("unable to find PublishedTrack", nil,
-							"room", room.Room.Name,
-							"participant", publisher.Identity(),
-							"pID", publisher.ID(),
-							"track", sid)
-						continue
-					}
-					if msg.TrackSetting.Width > 0 {
-						msg.TrackSetting.Quality = pubTrack.GetQualityForDimension(msg.TrackSetting.Width, msg.TrackSetting.Height)
-					}
-
-					// find quality for published track
-					logger.Debugw("updating track settings",
-						"room", room.Room.Name,
-						"participant", participant.Identity(),
-						"pID", participant.ID(),
-						"settings", msg.TrackSetting)
-					subTrack.UpdateSubscriberSettings(
-						!msg.TrackSetting.Disabled,
-						msg.TrackSetting.Quality,
-					)
-				}
-			case *livekit.SignalRequest_Leave:
-				_ = participant.Close()
+			if err := r.handleSignalRequest(room, participant, req); err != nil {
+				// more specific errors are already logged
+				// treat errors returned as fatal
+				return
 			}
 		}
 	}
 }
 
+func (r *RoomManager) handleSignalRequest(room *rtc.Room, participant types.Participant, req *livekit.SignalRequest) error {
+	switch msg := req.Message.(type) {
+	case *livekit.SignalRequest_Offer:
+		_, err := participant.HandleOffer(rtc.FromProtoSessionDescription(msg.Offer))
+		if err != nil {
+			logger.Errorw("could not handle offer", err,
+				"room", room.Room.Name,
+				"participant", participant.Identity(),
+				"pID", participant.ID(),
+			)
+			return err
+		}
+	case *livekit.SignalRequest_AddTrack:
+		logger.Debugw("add track request",
+			"room", room.Room.Name,
+			"participant", participant.Identity(),
+			"pID", participant.ID(),
+			"track", msg.AddTrack.Cid)
+		participant.AddTrack(msg.AddTrack)
+	case *livekit.SignalRequest_Answer:
+		sd := rtc.FromProtoSessionDescription(msg.Answer)
+		if err := participant.HandleAnswer(sd); err != nil {
+			logger.Errorw("could not handle answer", err,
+				"room", room.Room.Name,
+				"participant", participant.Identity(),
+				"pID", participant.ID(),
+			)
+			// connection cannot be successful if we can't answer
+			return err
+		}
+	case *livekit.SignalRequest_Trickle:
+		candidateInit, err := rtc.FromProtoTrickle(msg.Trickle)
+		if err != nil {
+			logger.Warnw("could not decode trickle", err,
+				"room", room.Room.Name,
+				"participant", participant.Identity(),
+				"pID", participant.ID(),
+			)
+			return nil
+		}
+		// logger.Debugw("adding peer candidate", "participant", participant.Identity())
+		if err := participant.AddICECandidate(candidateInit, msg.Trickle.Target); err != nil {
+			logger.Warnw("could not handle trickle", err,
+				"room", room.Room.Name,
+				"participant", participant.Identity(),
+				"pID", participant.ID(),
+			)
+		}
+	case *livekit.SignalRequest_Mute:
+		participant.SetTrackMuted(msg.Mute.Sid, msg.Mute.Muted, false)
+	case *livekit.SignalRequest_Subscription:
+		if err := room.UpdateSubscriptions(participant, msg.Subscription.TrackSids, msg.Subscription.Subscribe); err != nil {
+			logger.Warnw("could not update subscription", err,
+				"room", room.Room.Name,
+				"participant", participant.Identity(),
+				"pID", participant.ID(),
+				"tracks", msg.Subscription.TrackSids,
+				"subscribe", msg.Subscription.Subscribe)
+		}
+	case *livekit.SignalRequest_TrackSetting:
+		for _, sid := range msg.TrackSetting.TrackSids {
+			subTrack := participant.GetSubscribedTrack(sid)
+			if subTrack == nil {
+				logger.Warnw("unable to find SubscribedTrack", nil,
+					"room", room.Room.Name,
+					"participant", participant.Identity(),
+					"pID", participant.ID(),
+					"track", sid)
+				continue
+			}
+
+			// find the source PublishedTrack
+			publisher := room.GetParticipant(subTrack.PublisherIdentity())
+			if publisher == nil {
+				logger.Warnw("unable to find publisher of SubscribedTrack", nil,
+					"room", room.Room.Name,
+					"participant", participant.Identity(),
+					"pID", participant.ID(),
+					"publisher", subTrack.PublisherIdentity(),
+					"track", sid)
+				continue
+			}
+
+			pubTrack := publisher.GetPublishedTrack(sid)
+			if pubTrack == nil {
+				logger.Warnw("unable to find PublishedTrack", nil,
+					"room", room.Room.Name,
+					"participant", publisher.Identity(),
+					"pID", publisher.ID(),
+					"track", sid)
+				continue
+			}
+
+			// find quality for published track
+			logger.Debugw("updating track settings",
+				"room", room.Room.Name,
+				"participant", participant.Identity(),
+				"pID", participant.ID(),
+				"settings", msg.TrackSetting)
+			subTrack.UpdateSubscriberSettings(msg.TrackSetting)
+		}
+	case *livekit.SignalRequest_UpdateLayers:
+		track := participant.GetPublishedTrack(msg.UpdateLayers.TrackSid)
+		if track == nil {
+			logger.Warnw("could not find published track", nil,
+				"track", msg.UpdateLayers.TrackSid)
+			return nil
+		}
+		track.UpdateVideoLayers(msg.UpdateLayers.Layers)
+	case *livekit.SignalRequest_Leave:
+		_ = participant.Close()
+	}
+	return nil
+}
+
 // handles RTC messages resulted from Room API calls
 func (r *RoomManager) handleRTCMessage(ctx context.Context, roomName, identity string, msg *livekit.RTCNodeMessage) {
 	r.lock.RLock()

pkg/sfu/helpers.go

@@ -15,9 +15,9 @@ var (
 type ntpTime uint64
 
 const (
-	quarterResolution = "q"
-	halfResolution    = "h"
-	fullResolution    = "f"
+	QuarterResolution = "q"
+	HalfResolution    = "h"
+	FullResolution    = "f"
 )
 
 // Do a fuzzy find for a codec in the list of codecs

pkg/sfu/receiver.go

@@ -177,9 +177,9 @@ func (w *WebRTCReceiver) AddUpTrack(track *webrtc.TrackRemote, buff *buffer.Buff
 
 	var layer int32
 	switch track.RID() {
-	case fullResolution:
+	case FullResolution:
 		layer = 2
-	case halfResolution:
+	case HalfResolution:
 		layer = 1
 	default:
 		layer = 0