From 318bee4f1e2b7baae58f544fb9a719fa9e2eb36b Mon Sep 17 00:00:00 2001 From: David Zhao Date: Mon, 6 Dec 2021 12:12:08 -0800 Subject: [PATCH] Improve video quality selection by using publisher feedback (#238) --- go.mod | 2 +- go.sum | 4 +- pkg/rtc/mediatrack.go | 142 +++++++---- pkg/rtc/mediatrack_test.go | 30 +++ pkg/rtc/participant.go | 10 +- pkg/rtc/plithrottle.go | 14 +- pkg/rtc/subscribedtrack.go | 41 +++- pkg/rtc/types/interfaces.go | 8 +- pkg/rtc/types/typesfakes/fake_participant.go | 39 --- .../types/typesfakes/fake_published_track.go | 44 ++++ .../types/typesfakes/fake_subscribed_track.go | 52 +++- pkg/service/roommanager.go | 226 +++++++++--------- pkg/sfu/helpers.go | 6 +- pkg/sfu/receiver.go | 4 +- 14 files changed, 380 insertions(+), 242 deletions(-) diff --git a/go.mod b/go.mod index ae5b674ac..97991c1d4 100644 --- a/go.mod +++ b/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 diff --git a/go.sum b/go.sum index e38ad5b72..d60e5abf1 100644 --- a/go.sum +++ b/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= diff --git a/pkg/rtc/mediatrack.go b/pkg/rtc/mediatrack.go index 011ec24fe..3fa510045 100644 --- a/pkg/rtc/mediatrack.go +++ b/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 { diff --git a/pkg/rtc/mediatrack_test.go b/pkg/rtc/mediatrack_test.go index bbb9a9242..3ef07f415 100644 --- a/pkg/rtc/mediatrack_test.go +++ b/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)) + }) } diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 0f5141502..0b69012b4 100644 --- a/pkg/rtc/participant.go +++ b/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( diff --git a/pkg/rtc/plithrottle.go b/pkg/rtc/plithrottle.go index db6105135..11838ba7e 100644 --- a/pkg/rtc/plithrottle.go +++ b/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 diff --git a/pkg/rtc/subscribedtrack.go b/pkg/rtc/subscribedtrack.go index 049ec684a..25f2a9227 100644 --- a/pkg/rtc/subscribedtrack.go +++ b/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() { diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index f2f6d8435..57cc27e31 100644 --- a/pkg/rtc/types/interfaces.go +++ b/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 } diff --git a/pkg/rtc/types/typesfakes/fake_participant.go b/pkg/rtc/types/typesfakes/fake_participant.go index cfc2373d3..d17a1b35a 100644 --- a/pkg/rtc/types/typesfakes/fake_participant.go +++ b/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() diff --git a/pkg/rtc/types/typesfakes/fake_published_track.go b/pkg/rtc/types/typesfakes/fake_published_track.go index 210e5ffe9..2844792da 100644 --- a/pkg/rtc/types/typesfakes/fake_published_track.go +++ b/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 diff --git a/pkg/rtc/types/typesfakes/fake_subscribed_track.go b/pkg/rtc/types/typesfakes/fake_subscribed_track.go index 7001b610a..c0b40022c 100644 --- a/pkg/rtc/types/typesfakes/fake_subscribed_track.go +++ b/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 diff --git a/pkg/service/roommanager.go b/pkg/service/roommanager.go index 813b6a3ce..a8de05532 100644 --- a/pkg/service/roommanager.go +++ b/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() diff --git a/pkg/sfu/helpers.go b/pkg/sfu/helpers.go index f00e3c239..bc67c1e2b 100644 --- a/pkg/sfu/helpers.go +++ b/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 diff --git a/pkg/sfu/receiver.go b/pkg/sfu/receiver.go index 2d686cc1e..0a08412d8 100644 --- a/pkg/sfu/receiver.go +++ b/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