From e54ce4f674d30e990fd4fd43b25562f0e6ec6896 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Tue, 14 Dec 2021 12:48:09 +0530 Subject: [PATCH] Stream Allocator Try 3 (#257) * Stream Allocator Try 3 Making an intermediate PR to do - Special treatment for screen share tracks - When allocating all tracks, o try to stream all tracks by starting with the lowest layer o multi-pass across tracks to get a more even distribution Not yet done: ------------- In deficient state, o Allocate a specific track on a change o Steal from other tracks * Correct sense of managed track * have to range to copy * generate * fix VideoLayers compare * Use t.simulcasted --- pkg/routing/interfaces.go | 2 +- pkg/routing/localrouter.go | 2 +- pkg/routing/redisrouter.go | 2 +- pkg/rtc/mediatrack.go | 10 + pkg/rtc/subscribedtrack.go | 4 + pkg/rtc/transport.go | 4 +- pkg/rtc/types/interfaces.go | 3 + pkg/rtc/types/typesfakes/fake_media_track.go | 130 ++++ .../types/typesfakes/fake_published_track.go | 130 ++++ pkg/rtc/types/typesfakes/fake_room.go | 74 -- .../types/typesfakes/fake_subscribed_track.go | 65 ++ pkg/service/recordingservice.go | 2 +- pkg/service/roomallocator.go | 2 +- pkg/service/server.go | 2 +- pkg/service/utils.go | 2 +- pkg/service/wsprotocol.go | 2 +- pkg/sfu/downtrack.go | 53 +- pkg/sfu/forwarder.go | 728 ++++++++++-------- pkg/sfu/forwarder_test.go | 271 ++++--- pkg/sfu/receiver.go | 38 +- pkg/sfu/streamallocator.go | 490 ++++-------- pkg/telemetry/analytics.go | 2 +- test/integration_helpers.go | 2 +- test/scenarios.go | 2 +- test/webhook_test.go | 2 +- 25 files changed, 1168 insertions(+), 856 deletions(-) diff --git a/pkg/routing/interfaces.go b/pkg/routing/interfaces.go index 3ba9e8fa1..b7b0c83da 100644 --- a/pkg/routing/interfaces.go +++ b/pkg/routing/interfaces.go @@ -4,8 +4,8 @@ import ( "context" "github.com/go-redis/redis/v8" - "github.com/livekit/protocol/logger" livekit "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/logger" "google.golang.org/protobuf/proto" "github.com/livekit/livekit-server/pkg/config" diff --git a/pkg/routing/localrouter.go b/pkg/routing/localrouter.go index 958c0324d..12d316c3a 100644 --- a/pkg/routing/localrouter.go +++ b/pkg/routing/localrouter.go @@ -5,8 +5,8 @@ import ( "sync" "time" - "github.com/livekit/protocol/logger" livekit "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/logger" "github.com/livekit/protocol/utils" "google.golang.org/protobuf/proto" ) diff --git a/pkg/routing/redisrouter.go b/pkg/routing/redisrouter.go index 1f91372ab..8be8258e4 100644 --- a/pkg/routing/redisrouter.go +++ b/pkg/routing/redisrouter.go @@ -5,8 +5,8 @@ import ( "time" "github.com/go-redis/redis/v8" - "github.com/livekit/protocol/logger" livekit "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/logger" "github.com/livekit/protocol/utils" "github.com/pkg/errors" "google.golang.org/protobuf/proto" diff --git a/pkg/rtc/mediatrack.go b/pkg/rtc/mediatrack.go index 578d505f5..8e0597819 100644 --- a/pkg/rtc/mediatrack.go +++ b/pkg/rtc/mediatrack.go @@ -96,6 +96,7 @@ func NewMediaTrack(track *webrtc.TrackRemote, params MediaTrackParams) *MediaTra if params.TrackInfo != nil && t.Kind() == livekit.TrackType_VIDEO { t.UpdateVideoLayers(params.TrackInfo.Layers) + // LK-TODO: maybe use this or simulcast flag in TrackInfo to set simulcasted here } return t } @@ -116,6 +117,14 @@ func (t *MediaTrack) Kind() livekit.TrackType { return t.params.TrackInfo.Type } +func (t *MediaTrack) Source() livekit.TrackSource { + return t.params.TrackInfo.Source +} + +func (t *MediaTrack) IsSimulcast() bool { + return t.simulcasted.Get() +} + func (t *MediaTrack) Name() string { return t.params.TrackInfo.Name } @@ -391,6 +400,7 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track *webrtc.Tra t.params.Telemetry.AddUpTrack(t.params.ParticipantID, buff) atomic.AddUint32(&t.numUpTracks, 1) + // LK-TODO: can remove this completely when VideoLayers protocol becomes the default as it has info from client or if we decide to use TrackInfo.Simulcast if atomic.LoadUint32(&t.numUpTracks) > 1 || track.RID() != "" { // cannot only rely on numUpTracks since we fire metadata events immediately after the first layer t.simulcasted.TrySet(true) diff --git a/pkg/rtc/subscribedtrack.go b/pkg/rtc/subscribedtrack.go index a548dbd78..b3b463dea 100644 --- a/pkg/rtc/subscribedtrack.go +++ b/pkg/rtc/subscribedtrack.go @@ -48,6 +48,10 @@ func (t *SubscribedTrack) DownTrack() *sfu.DownTrack { return t.dt } +func (t *SubscribedTrack) PublishedTrack() types.MediaTrack { + return t.publishedTrack +} + func (t *SubscribedTrack) SubscribeLossPercentage() uint32 { return FixedPointToPercent(t.DownTrack().CurrentMaxLossFraction()) } diff --git a/pkg/rtc/transport.go b/pkg/rtc/transport.go index 439cd7503..ea4c00747 100644 --- a/pkg/rtc/transport.go +++ b/pkg/rtc/transport.go @@ -268,7 +268,9 @@ func (t *PCTransport) AddTrack(subTrack types.SubscribedTrack) { return } - t.streamAllocator.AddTrack(subTrack.DownTrack()) + source := subTrack.PublishedTrack().Source() + isManaged := (source != livekit.TrackSource_SCREEN_SHARE && source != livekit.TrackSource_SCREEN_SHARE_AUDIO) || subTrack.PublishedTrack().IsSimulcast() + t.streamAllocator.AddTrack(subTrack.DownTrack(), isManaged) } func (t *PCTransport) RemoveTrack(subTrack types.SubscribedTrack) { diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index ce6b1ea77..74f827b60 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -106,6 +106,8 @@ type MediaTrack interface { IsMuted() bool SetMuted(muted bool) UpdateVideoLayers(layers []*livekit.VideoLayer) + Source() livekit.TrackSource + IsSimulcast() bool // subscribers AddSubscriber(participant Participant) error @@ -140,6 +142,7 @@ type SubscribedTrack interface { ID() string PublisherIdentity() string DownTrack() *sfu.DownTrack + PublishedTrack() MediaTrack IsMuted() bool SetPublisherMuted(muted bool) UpdateSubscriberSettings(settings *livekit.UpdateTrackSettings) diff --git a/pkg/rtc/types/typesfakes/fake_media_track.go b/pkg/rtc/types/typesfakes/fake_media_track.go index fa5cdf309..2b44870f1 100644 --- a/pkg/rtc/types/typesfakes/fake_media_track.go +++ b/pkg/rtc/types/typesfakes/fake_media_track.go @@ -52,6 +52,16 @@ type FakeMediaTrack struct { isMutedReturnsOnCall map[int]struct { result1 bool } + IsSimulcastStub func() bool + isSimulcastMutex sync.RWMutex + isSimulcastArgsForCall []struct { + } + isSimulcastReturns struct { + result1 bool + } + isSimulcastReturnsOnCall map[int]struct { + result1 bool + } IsSubscriberStub func(string) bool isSubscriberMutex sync.RWMutex isSubscriberArgsForCall []struct { @@ -97,6 +107,16 @@ type FakeMediaTrack struct { setMutedArgsForCall []struct { arg1 bool } + SourceStub func() livekit.TrackSource + sourceMutex sync.RWMutex + sourceArgsForCall []struct { + } + sourceReturns struct { + result1 livekit.TrackSource + } + sourceReturnsOnCall map[int]struct { + result1 livekit.TrackSource + } UpdateVideoLayersStub func([]*livekit.VideoLayer) updateVideoLayersMutex sync.RWMutex updateVideoLayersArgsForCall []struct { @@ -335,6 +355,59 @@ func (fake *FakeMediaTrack) IsMutedReturnsOnCall(i int, result1 bool) { }{result1} } +func (fake *FakeMediaTrack) IsSimulcast() bool { + fake.isSimulcastMutex.Lock() + ret, specificReturn := fake.isSimulcastReturnsOnCall[len(fake.isSimulcastArgsForCall)] + fake.isSimulcastArgsForCall = append(fake.isSimulcastArgsForCall, struct { + }{}) + stub := fake.IsSimulcastStub + fakeReturns := fake.isSimulcastReturns + fake.recordInvocation("IsSimulcast", []interface{}{}) + fake.isSimulcastMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeMediaTrack) IsSimulcastCallCount() int { + fake.isSimulcastMutex.RLock() + defer fake.isSimulcastMutex.RUnlock() + return len(fake.isSimulcastArgsForCall) +} + +func (fake *FakeMediaTrack) IsSimulcastCalls(stub func() bool) { + fake.isSimulcastMutex.Lock() + defer fake.isSimulcastMutex.Unlock() + fake.IsSimulcastStub = stub +} + +func (fake *FakeMediaTrack) IsSimulcastReturns(result1 bool) { + fake.isSimulcastMutex.Lock() + defer fake.isSimulcastMutex.Unlock() + fake.IsSimulcastStub = nil + fake.isSimulcastReturns = struct { + result1 bool + }{result1} +} + +func (fake *FakeMediaTrack) IsSimulcastReturnsOnCall(i int, result1 bool) { + fake.isSimulcastMutex.Lock() + defer fake.isSimulcastMutex.Unlock() + fake.IsSimulcastStub = nil + if fake.isSimulcastReturnsOnCall == nil { + fake.isSimulcastReturnsOnCall = make(map[int]struct { + result1 bool + }) + } + fake.isSimulcastReturnsOnCall[i] = struct { + result1 bool + }{result1} +} + func (fake *FakeMediaTrack) IsSubscriber(arg1 string) bool { fake.isSubscriberMutex.Lock() ret, specificReturn := fake.isSubscriberReturnsOnCall[len(fake.isSubscriberArgsForCall)] @@ -590,6 +663,59 @@ func (fake *FakeMediaTrack) SetMutedArgsForCall(i int) bool { return argsForCall.arg1 } +func (fake *FakeMediaTrack) Source() livekit.TrackSource { + fake.sourceMutex.Lock() + ret, specificReturn := fake.sourceReturnsOnCall[len(fake.sourceArgsForCall)] + fake.sourceArgsForCall = append(fake.sourceArgsForCall, struct { + }{}) + stub := fake.SourceStub + fakeReturns := fake.sourceReturns + fake.recordInvocation("Source", []interface{}{}) + fake.sourceMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeMediaTrack) SourceCallCount() int { + fake.sourceMutex.RLock() + defer fake.sourceMutex.RUnlock() + return len(fake.sourceArgsForCall) +} + +func (fake *FakeMediaTrack) SourceCalls(stub func() livekit.TrackSource) { + fake.sourceMutex.Lock() + defer fake.sourceMutex.Unlock() + fake.SourceStub = stub +} + +func (fake *FakeMediaTrack) SourceReturns(result1 livekit.TrackSource) { + fake.sourceMutex.Lock() + defer fake.sourceMutex.Unlock() + fake.SourceStub = nil + fake.sourceReturns = struct { + result1 livekit.TrackSource + }{result1} +} + +func (fake *FakeMediaTrack) SourceReturnsOnCall(i int, result1 livekit.TrackSource) { + fake.sourceMutex.Lock() + defer fake.sourceMutex.Unlock() + fake.SourceStub = nil + if fake.sourceReturnsOnCall == nil { + fake.sourceReturnsOnCall = make(map[int]struct { + result1 livekit.TrackSource + }) + } + fake.sourceReturnsOnCall[i] = struct { + result1 livekit.TrackSource + }{result1} +} + func (fake *FakeMediaTrack) UpdateVideoLayers(arg1 []*livekit.VideoLayer) { var arg1Copy []*livekit.VideoLayer if arg1 != nil { @@ -638,6 +764,8 @@ func (fake *FakeMediaTrack) Invocations() map[string][][]interface{} { defer fake.iDMutex.RUnlock() fake.isMutedMutex.RLock() defer fake.isMutedMutex.RUnlock() + fake.isSimulcastMutex.RLock() + defer fake.isSimulcastMutex.RUnlock() fake.isSubscriberMutex.RLock() defer fake.isSubscriberMutex.RUnlock() fake.kindMutex.RLock() @@ -650,6 +778,8 @@ func (fake *FakeMediaTrack) Invocations() map[string][][]interface{} { defer fake.removeSubscriberMutex.RUnlock() fake.setMutedMutex.RLock() defer fake.setMutedMutex.RUnlock() + fake.sourceMutex.RLock() + defer fake.sourceMutex.RUnlock() fake.updateVideoLayersMutex.RLock() defer fake.updateVideoLayersMutex.RUnlock() copiedInvocations := map[string][][]interface{}{} diff --git a/pkg/rtc/types/typesfakes/fake_published_track.go b/pkg/rtc/types/typesfakes/fake_published_track.go index 1ec9d64e0..1b975c905 100644 --- a/pkg/rtc/types/typesfakes/fake_published_track.go +++ b/pkg/rtc/types/typesfakes/fake_published_track.go @@ -58,6 +58,16 @@ type FakePublishedTrack struct { isMutedReturnsOnCall map[int]struct { result1 bool } + IsSimulcastStub func() bool + isSimulcastMutex sync.RWMutex + isSimulcastArgsForCall []struct { + } + isSimulcastReturns struct { + result1 bool + } + isSimulcastReturnsOnCall map[int]struct { + result1 bool + } IsSubscriberStub func(string) bool isSubscriberMutex sync.RWMutex isSubscriberArgsForCall []struct { @@ -155,6 +165,16 @@ type FakePublishedTrack struct { signalCidReturnsOnCall map[int]struct { result1 string } + SourceStub func() livekit.TrackSource + sourceMutex sync.RWMutex + sourceArgsForCall []struct { + } + sourceReturns struct { + result1 livekit.TrackSource + } + sourceReturnsOnCall map[int]struct { + result1 livekit.TrackSource + } ToProtoStub func() *livekit.TrackInfo toProtoMutex sync.RWMutex toProtoArgsForCall []struct { @@ -435,6 +455,59 @@ func (fake *FakePublishedTrack) IsMutedReturnsOnCall(i int, result1 bool) { }{result1} } +func (fake *FakePublishedTrack) IsSimulcast() bool { + fake.isSimulcastMutex.Lock() + ret, specificReturn := fake.isSimulcastReturnsOnCall[len(fake.isSimulcastArgsForCall)] + fake.isSimulcastArgsForCall = append(fake.isSimulcastArgsForCall, struct { + }{}) + stub := fake.IsSimulcastStub + fakeReturns := fake.isSimulcastReturns + fake.recordInvocation("IsSimulcast", []interface{}{}) + fake.isSimulcastMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakePublishedTrack) IsSimulcastCallCount() int { + fake.isSimulcastMutex.RLock() + defer fake.isSimulcastMutex.RUnlock() + return len(fake.isSimulcastArgsForCall) +} + +func (fake *FakePublishedTrack) IsSimulcastCalls(stub func() bool) { + fake.isSimulcastMutex.Lock() + defer fake.isSimulcastMutex.Unlock() + fake.IsSimulcastStub = stub +} + +func (fake *FakePublishedTrack) IsSimulcastReturns(result1 bool) { + fake.isSimulcastMutex.Lock() + defer fake.isSimulcastMutex.Unlock() + fake.IsSimulcastStub = nil + fake.isSimulcastReturns = struct { + result1 bool + }{result1} +} + +func (fake *FakePublishedTrack) IsSimulcastReturnsOnCall(i int, result1 bool) { + fake.isSimulcastMutex.Lock() + defer fake.isSimulcastMutex.Unlock() + fake.IsSimulcastStub = nil + if fake.isSimulcastReturnsOnCall == nil { + fake.isSimulcastReturnsOnCall = make(map[int]struct { + result1 bool + }) + } + fake.isSimulcastReturnsOnCall[i] = struct { + result1 bool + }{result1} +} + func (fake *FakePublishedTrack) IsSubscriber(arg1 string) bool { fake.isSubscriberMutex.Lock() ret, specificReturn := fake.isSubscriberReturnsOnCall[len(fake.isSubscriberArgsForCall)] @@ -958,6 +1031,59 @@ func (fake *FakePublishedTrack) SignalCidReturnsOnCall(i int, result1 string) { }{result1} } +func (fake *FakePublishedTrack) Source() livekit.TrackSource { + fake.sourceMutex.Lock() + ret, specificReturn := fake.sourceReturnsOnCall[len(fake.sourceArgsForCall)] + fake.sourceArgsForCall = append(fake.sourceArgsForCall, struct { + }{}) + stub := fake.SourceStub + fakeReturns := fake.sourceReturns + fake.recordInvocation("Source", []interface{}{}) + fake.sourceMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakePublishedTrack) SourceCallCount() int { + fake.sourceMutex.RLock() + defer fake.sourceMutex.RUnlock() + return len(fake.sourceArgsForCall) +} + +func (fake *FakePublishedTrack) SourceCalls(stub func() livekit.TrackSource) { + fake.sourceMutex.Lock() + defer fake.sourceMutex.Unlock() + fake.SourceStub = stub +} + +func (fake *FakePublishedTrack) SourceReturns(result1 livekit.TrackSource) { + fake.sourceMutex.Lock() + defer fake.sourceMutex.Unlock() + fake.SourceStub = nil + fake.sourceReturns = struct { + result1 livekit.TrackSource + }{result1} +} + +func (fake *FakePublishedTrack) SourceReturnsOnCall(i int, result1 livekit.TrackSource) { + fake.sourceMutex.Lock() + defer fake.sourceMutex.Unlock() + fake.SourceStub = nil + if fake.sourceReturnsOnCall == nil { + fake.sourceReturnsOnCall = make(map[int]struct { + result1 livekit.TrackSource + }) + } + fake.sourceReturnsOnCall[i] = struct { + result1 livekit.TrackSource + }{result1} +} + func (fake *FakePublishedTrack) ToProto() *livekit.TrackInfo { fake.toProtoMutex.Lock() ret, specificReturn := fake.toProtoReturnsOnCall[len(fake.toProtoArgsForCall)] @@ -1061,6 +1187,8 @@ func (fake *FakePublishedTrack) Invocations() map[string][][]interface{} { defer fake.iDMutex.RUnlock() fake.isMutedMutex.RLock() defer fake.isMutedMutex.RUnlock() + fake.isSimulcastMutex.RLock() + defer fake.isSimulcastMutex.RUnlock() fake.isSubscriberMutex.RLock() defer fake.isSubscriberMutex.RUnlock() fake.kindMutex.RLock() @@ -1083,6 +1211,8 @@ func (fake *FakePublishedTrack) Invocations() map[string][][]interface{} { defer fake.setMutedMutex.RUnlock() fake.signalCidMutex.RLock() defer fake.signalCidMutex.RUnlock() + fake.sourceMutex.RLock() + defer fake.sourceMutex.RUnlock() fake.toProtoMutex.RLock() defer fake.toProtoMutex.RUnlock() fake.updateVideoLayersMutex.RLock() diff --git a/pkg/rtc/types/typesfakes/fake_room.go b/pkg/rtc/types/typesfakes/fake_room.go index 78172f3ce..46cc27c64 100644 --- a/pkg/rtc/types/typesfakes/fake_room.go +++ b/pkg/rtc/types/typesfakes/fake_room.go @@ -8,17 +8,6 @@ import ( ) type FakeRoom struct { - GetParticipantStub func(string) types.Participant - getParticipantMutex sync.RWMutex - getParticipantArgsForCall []struct { - arg1 string - } - getParticipantReturns struct { - result1 types.Participant - } - getParticipantReturnsOnCall map[int]struct { - result1 types.Participant - } NameStub func() string nameMutex sync.RWMutex nameArgsForCall []struct { @@ -46,67 +35,6 @@ type FakeRoom struct { invocationsMutex sync.RWMutex } -func (fake *FakeRoom) GetParticipant(arg1 string) types.Participant { - fake.getParticipantMutex.Lock() - ret, specificReturn := fake.getParticipantReturnsOnCall[len(fake.getParticipantArgsForCall)] - fake.getParticipantArgsForCall = append(fake.getParticipantArgsForCall, struct { - arg1 string - }{arg1}) - stub := fake.GetParticipantStub - fakeReturns := fake.getParticipantReturns - fake.recordInvocation("GetParticipant", []interface{}{arg1}) - fake.getParticipantMutex.Unlock() - if stub != nil { - return stub(arg1) - } - if specificReturn { - return ret.result1 - } - return fakeReturns.result1 -} - -func (fake *FakeRoom) GetParticipantCallCount() int { - fake.getParticipantMutex.RLock() - defer fake.getParticipantMutex.RUnlock() - return len(fake.getParticipantArgsForCall) -} - -func (fake *FakeRoom) GetParticipantCalls(stub func(string) types.Participant) { - fake.getParticipantMutex.Lock() - defer fake.getParticipantMutex.Unlock() - fake.GetParticipantStub = stub -} - -func (fake *FakeRoom) GetParticipantArgsForCall(i int) string { - fake.getParticipantMutex.RLock() - defer fake.getParticipantMutex.RUnlock() - argsForCall := fake.getParticipantArgsForCall[i] - return argsForCall.arg1 -} - -func (fake *FakeRoom) GetParticipantReturns(result1 types.Participant) { - fake.getParticipantMutex.Lock() - defer fake.getParticipantMutex.Unlock() - fake.GetParticipantStub = nil - fake.getParticipantReturns = struct { - result1 types.Participant - }{result1} -} - -func (fake *FakeRoom) GetParticipantReturnsOnCall(i int, result1 types.Participant) { - fake.getParticipantMutex.Lock() - defer fake.getParticipantMutex.Unlock() - fake.GetParticipantStub = nil - if fake.getParticipantReturnsOnCall == nil { - fake.getParticipantReturnsOnCall = make(map[int]struct { - result1 types.Participant - }) - } - fake.getParticipantReturnsOnCall[i] = struct { - result1 types.Participant - }{result1} -} - func (fake *FakeRoom) Name() string { fake.nameMutex.Lock() ret, specificReturn := fake.nameReturnsOnCall[len(fake.nameArgsForCall)] @@ -231,8 +159,6 @@ func (fake *FakeRoom) UpdateSubscriptionsReturnsOnCall(i int, result1 error) { func (fake *FakeRoom) Invocations() map[string][][]interface{} { fake.invocationsMutex.RLock() defer fake.invocationsMutex.RUnlock() - fake.getParticipantMutex.RLock() - defer fake.getParticipantMutex.RUnlock() fake.nameMutex.RLock() defer fake.nameMutex.RUnlock() fake.updateSubscriptionsMutex.RLock() diff --git a/pkg/rtc/types/typesfakes/fake_subscribed_track.go b/pkg/rtc/types/typesfakes/fake_subscribed_track.go index d2127c17a..b3b834497 100644 --- a/pkg/rtc/types/typesfakes/fake_subscribed_track.go +++ b/pkg/rtc/types/typesfakes/fake_subscribed_track.go @@ -40,6 +40,16 @@ type FakeSubscribedTrack struct { isMutedReturnsOnCall map[int]struct { result1 bool } + PublishedTrackStub func() types.MediaTrack + publishedTrackMutex sync.RWMutex + publishedTrackArgsForCall []struct { + } + publishedTrackReturns struct { + result1 types.MediaTrack + } + publishedTrackReturnsOnCall map[int]struct { + result1 types.MediaTrack + } PublisherIdentityStub func() string publisherIdentityMutex sync.RWMutex publisherIdentityArgsForCall []struct { @@ -237,6 +247,59 @@ func (fake *FakeSubscribedTrack) IsMutedReturnsOnCall(i int, result1 bool) { }{result1} } +func (fake *FakeSubscribedTrack) PublishedTrack() types.MediaTrack { + fake.publishedTrackMutex.Lock() + ret, specificReturn := fake.publishedTrackReturnsOnCall[len(fake.publishedTrackArgsForCall)] + fake.publishedTrackArgsForCall = append(fake.publishedTrackArgsForCall, struct { + }{}) + stub := fake.PublishedTrackStub + fakeReturns := fake.publishedTrackReturns + fake.recordInvocation("PublishedTrack", []interface{}{}) + fake.publishedTrackMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeSubscribedTrack) PublishedTrackCallCount() int { + fake.publishedTrackMutex.RLock() + defer fake.publishedTrackMutex.RUnlock() + return len(fake.publishedTrackArgsForCall) +} + +func (fake *FakeSubscribedTrack) PublishedTrackCalls(stub func() types.MediaTrack) { + fake.publishedTrackMutex.Lock() + defer fake.publishedTrackMutex.Unlock() + fake.PublishedTrackStub = stub +} + +func (fake *FakeSubscribedTrack) PublishedTrackReturns(result1 types.MediaTrack) { + fake.publishedTrackMutex.Lock() + defer fake.publishedTrackMutex.Unlock() + fake.PublishedTrackStub = nil + fake.publishedTrackReturns = struct { + result1 types.MediaTrack + }{result1} +} + +func (fake *FakeSubscribedTrack) PublishedTrackReturnsOnCall(i int, result1 types.MediaTrack) { + fake.publishedTrackMutex.Lock() + defer fake.publishedTrackMutex.Unlock() + fake.PublishedTrackStub = nil + if fake.publishedTrackReturnsOnCall == nil { + fake.publishedTrackReturnsOnCall = make(map[int]struct { + result1 types.MediaTrack + }) + } + fake.publishedTrackReturnsOnCall[i] = struct { + result1 types.MediaTrack + }{result1} +} + func (fake *FakeSubscribedTrack) PublisherIdentity() string { fake.publisherIdentityMutex.Lock() ret, specificReturn := fake.publisherIdentityReturnsOnCall[len(fake.publisherIdentityArgsForCall)] @@ -440,6 +503,8 @@ func (fake *FakeSubscribedTrack) Invocations() map[string][][]interface{} { defer fake.iDMutex.RUnlock() fake.isMutedMutex.RLock() defer fake.isMutedMutex.RUnlock() + fake.publishedTrackMutex.RLock() + defer fake.publishedTrackMutex.RUnlock() fake.publisherIdentityMutex.RLock() defer fake.publisherIdentityMutex.RUnlock() fake.setPublisherMutedMutex.RLock() diff --git a/pkg/service/recordingservice.go b/pkg/service/recordingservice.go index 5b9aef901..7712098bf 100644 --- a/pkg/service/recordingservice.go +++ b/pkg/service/recordingservice.go @@ -4,8 +4,8 @@ import ( "context" "errors" - "github.com/livekit/protocol/logger" livekit "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/logger" "github.com/livekit/protocol/recording" "github.com/livekit/protocol/utils" "google.golang.org/protobuf/proto" diff --git a/pkg/service/roomallocator.go b/pkg/service/roomallocator.go index 8054dd27b..93c2f452f 100644 --- a/pkg/service/roomallocator.go +++ b/pkg/service/roomallocator.go @@ -4,8 +4,8 @@ import ( "context" "time" - "github.com/livekit/protocol/logger" livekit "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/logger" "github.com/livekit/protocol/utils" "github.com/livekit/livekit-server/pkg/config" diff --git a/pkg/service/server.go b/pkg/service/server.go index 3c4c23aaa..de9341d0b 100644 --- a/pkg/service/server.go +++ b/pkg/service/server.go @@ -11,8 +11,8 @@ import ( "time" "github.com/livekit/protocol/auth" - "github.com/livekit/protocol/logger" livekit "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/logger" "github.com/livekit/protocol/utils" "github.com/pion/turn/v2" "github.com/prometheus/client_golang/prometheus/promhttp" diff --git a/pkg/service/utils.go b/pkg/service/utils.go index d8ff0f91c..cb83508ac 100644 --- a/pkg/service/utils.go +++ b/pkg/service/utils.go @@ -5,8 +5,8 @@ import ( "regexp" "github.com/livekit/protocol/auth" - "github.com/livekit/protocol/logger" livekit "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/logger" ) func handleError(w http.ResponseWriter, status int, msg string) { diff --git a/pkg/service/wsprotocol.go b/pkg/service/wsprotocol.go index 6af4f4af6..65e536364 100644 --- a/pkg/service/wsprotocol.go +++ b/pkg/service/wsprotocol.go @@ -5,8 +5,8 @@ import ( "time" "github.com/gorilla/websocket" - "github.com/livekit/protocol/logger" livekit "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/logger" "google.golang.org/protobuf/encoding/protojson" "google.golang.org/protobuf/proto" diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index 309f3b2cf..6e0dc6d5e 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -26,7 +26,6 @@ type TrackSender interface { Close() // ID is the globally unique identifier for this Track. ID() string - SetTrackType(isSimulcast bool) Codec() webrtc.RTPCodecCapability PeerID() string } @@ -127,7 +126,7 @@ type DownTrack struct { onSubscriptionChanged func(dt *DownTrack) // max layer change callback - onSubscribedLayersChanged func(dt *DownTrack, layers VideoLayers, layerPref LayerPreference) + onSubscribedLayersChanged func(dt *DownTrack, layers VideoLayers) // packet sent callback onPacketSent []func(dt *DownTrack, size int) @@ -161,15 +160,13 @@ func NewDownTrack(c webrtc.RTPCodecCapability, r TrackReceiver, bf *buffer.Facto d.rtxStats.Store(new(PacketStats)) d.paddingStats.Store(new(PacketStats)) - return d, nil -} - -func (d *DownTrack) SetTrackType(isSimulcast bool) { - if isSimulcast { + if r.IsSimulcast() { d.trackType = SimulcastDownTrack } else { d.trackType = SimpleDownTrack } + + return d, nil } // Bind is called by the PeerConnection after negotiation is complete @@ -449,24 +446,24 @@ func (d *DownTrack) Close() { } func (d *DownTrack) SetMaxSpatialLayer(spatialLayer int32) { - changed, maxLayers, layerPref := d.forwarder.SetMaxSpatialLayer(spatialLayer) + changed, maxLayers := d.forwarder.SetMaxSpatialLayer(spatialLayer) if !changed { return } if d.onSubscribedLayersChanged != nil { - d.onSubscribedLayersChanged(d, maxLayers, layerPref) + d.onSubscribedLayersChanged(d, maxLayers) } } func (d *DownTrack) SetMaxTemporalLayer(temporalLayer int32) { - changed, maxLayers, layerPref := d.forwarder.SetMaxTemporalLayer(temporalLayer) + changed, maxLayers := d.forwarder.SetMaxTemporalLayer(temporalLayer) if !changed { return } if d.onSubscribedLayersChanged != nil { - d.onSubscribedLayersChanged(d, maxLayers, layerPref) + d.onSubscribedLayersChanged(d, maxLayers) } } @@ -474,10 +471,6 @@ func (d *DownTrack) MaxLayers() VideoLayers { return d.forwarder.MaxLayers() } -func (d *DownTrack) GetLayerPreference() LayerPreference { - return d.forwarder.GetLayerPreference() -} - func (d *DownTrack) GetForwardingStatus() ForwardingStatus { return d.forwarder.GetForwardingStatus() } @@ -525,7 +518,7 @@ func (d *DownTrack) OnSubscriptionChanged(fn func(dt *DownTrack)) { d.onSubscriptionChanged = fn } -func (d *DownTrack) OnSubscribedLayersChanged(fn func(dt *DownTrack, layers VideoLayers, layerPref LayerPreference)) { +func (d *DownTrack) OnSubscribedLayersChanged(fn func(dt *DownTrack, layers VideoLayers)) { d.onSubscribedLayersChanged = fn } @@ -533,12 +526,32 @@ func (d *DownTrack) OnPacketSent(fn func(dt *DownTrack, size int)) { d.onPacketSent = append(d.onPacketSent, fn) } +func (d *DownTrack) IsDeficient() bool { + return d.forwarder.IsDeficient() +} + +func (d *DownTrack) BandwidthRequested() int64 { + return d.forwarder.BandwidthRequested() +} + +func (d *DownTrack) DistanceToDesired() int32 { + return d.forwarder.DistanceToDesired() +} + func (d *DownTrack) Allocate(availableChannelCapacity int64) VideoAllocation { return d.forwarder.Allocate(availableChannelCapacity, d.receiver.GetBitrateTemporalCumulative()) } -func (d *DownTrack) TryAllocate(additionalChannelCapacity int64) VideoAllocation { - return d.forwarder.TryAllocate(additionalChannelCapacity, d.receiver.GetBitrateTemporalCumulative()) +func (d *DownTrack) ProvisionalAllocatePrepare() { + d.forwarder.ProvisionalAllocatePrepare(d.receiver.GetBitrateTemporalCumulative()) +} + +func (d *DownTrack) ProvisionalAllocate(availableChannelCapacity int64, layers VideoLayers) int64 { + return d.forwarder.ProvisionalAllocate(availableChannelCapacity, layers) +} + +func (d *DownTrack) ProvisionalAllocateCommit() VideoAllocation { + return d.forwarder.ProvisionalAllocateCommit() } func (d *DownTrack) FinalizeAllocate() VideoAllocation { @@ -549,8 +562,8 @@ func (d *DownTrack) AllocateNextHigher() (VideoAllocation, bool) { return d.forwarder.AllocateNextHigher(d.receiver.GetBitrateTemporalCumulative()) } -func (d *DownTrack) LastAllocation() VideoAllocation { - return d.forwarder.LastAllocation() +func (d *DownTrack) Pause() VideoAllocation { + return d.forwarder.Pause(d.receiver.GetBitrateTemporalCumulative()) } func (d *DownTrack) CreateSourceDescriptionChunks() []rtcp.SourceDescriptionChunk { diff --git a/pkg/sfu/forwarder.go b/pkg/sfu/forwarder.go index 8965a0d1b..a56ec6148 100644 --- a/pkg/sfu/forwarder.go +++ b/pkg/sfu/forwarder.go @@ -33,14 +33,6 @@ const ( LayerDirectionHighToLow ) -type LayerPreference int - -const ( - LayerPreferenceNone LayerPreference = iota - LayerPreferenceSpatial - LayerPreferenceTemporal -) - type VideoStreamingChange int const ( @@ -98,8 +90,9 @@ type VideoAllocation struct { bandwidthRequested int64 bandwidthDelta int64 availableLayers []uint16 - bitrates [3][4]int64 + bitrates Bitrates targetLayers VideoLayers + distanceToDesired int32 } func (v VideoAllocation) String() string { @@ -113,6 +106,24 @@ var ( } ) +type VideoAllocationProvisional struct { + layers VideoLayers + muted bool + bitrates Bitrates +} + +const ( + SpatialTransitionCost = 10 + TemporalTransitionCost = 0 +) + +type VideoAllocationMove struct { + layers VideoLayers + deltaBitrate int64 + transitionCost int32 + qualityCost int32 +} + type TranslationParams struct { shouldDrop bool isDroppingRelevant bool @@ -130,6 +141,10 @@ func (v VideoLayers) String() string { return fmt.Sprintf("VideoLayers{s: %d, t: %d}", v.spatial, v.temporal) } +func (v VideoLayers) GreaterThan(v2 VideoLayers) bool { + return v.spatial > v2.spatial || (v.spatial == v2.spatial && v.temporal > v2.temporal) +} + const ( DefaultMaxLayerSpatial = int32(2) DefaultMaxLayerTemporal = int32(3) @@ -157,8 +172,6 @@ type Forwarder struct { codec webrtc.RTPCodecCapability kind webrtc.RTPCodecType - layerPref LayerPreference - muted bool started bool @@ -169,6 +182,8 @@ type Forwarder struct { currentLayers VideoLayers targetLayers VideoLayers + provisional *VideoAllocationProvisional + lastAllocation VideoAllocation availableLayers []uint16 @@ -182,8 +197,6 @@ func NewForwarder(codec webrtc.RTPCodecCapability, kind webrtc.RTPCodecType) *Fo codec: codec, kind: kind, - layerPref: LayerPreferenceSpatial, - // start off with nothing, let streamallocator set things currentLayers: InvalidLayers, targetLayers: InvalidLayers, @@ -225,30 +238,30 @@ func (f *Forwarder) Muted() bool { return f.muted } -func (f *Forwarder) SetMaxSpatialLayer(spatialLayer int32) (bool, VideoLayers, LayerPreference) { +func (f *Forwarder) SetMaxSpatialLayer(spatialLayer int32) (bool, VideoLayers) { f.lock.Lock() defer f.lock.Unlock() if f.kind == webrtc.RTPCodecTypeAudio || spatialLayer == f.maxLayers.spatial { - return false, InvalidLayers, LayerPreferenceNone + return false, InvalidLayers } f.maxLayers.spatial = spatialLayer - return true, f.maxLayers, f.layerPref + return true, f.maxLayers } -func (f *Forwarder) SetMaxTemporalLayer(temporalLayer int32) (bool, VideoLayers, LayerPreference) { +func (f *Forwarder) SetMaxTemporalLayer(temporalLayer int32) (bool, VideoLayers) { f.lock.Lock() defer f.lock.Unlock() if f.kind == webrtc.RTPCodecTypeAudio || temporalLayer == f.maxLayers.temporal { - return false, InvalidLayers, LayerPreferenceNone + return false, InvalidLayers } f.maxLayers.temporal = temporalLayer - return true, f.maxLayers, f.layerPref + return true, f.maxLayers } func (f *Forwarder) MaxLayers() VideoLayers { @@ -258,13 +271,6 @@ func (f *Forwarder) MaxLayers() VideoLayers { return f.maxLayers } -func (f *Forwarder) GetLayerPreference() LayerPreference { - f.lock.RLock() - defer f.lock.RUnlock() - - return f.layerPref -} - func (f *Forwarder) CurrentLayers() VideoLayers { f.lock.RLock() defer f.lock.RUnlock() @@ -301,12 +307,7 @@ func (f *Forwarder) UptrackLayersChange(availableLayers []uint16) { f.availableLayers = availableLayers } -func (f *Forwarder) disable() { - f.currentLayers = InvalidLayers - f.targetLayers = InvalidLayers -} - -func (f *Forwarder) getOptimalBandwidthNeeded(brs [3][4]int64) int64 { +func (f *Forwarder) getOptimalBandwidthNeeded(brs Bitrates) int64 { for i := f.maxLayers.spatial; i >= 0; i-- { for j := f.maxLayers.temporal; j >= 0; j-- { if brs[i][j] == 0 { @@ -320,255 +321,273 @@ func (f *Forwarder) getOptimalBandwidthNeeded(brs [3][4]int64) int64 { return 0 } -func (f *Forwarder) findBestLayers( - minLayers VideoLayers, - maxLayers VideoLayers, - brs [3][4]int64, - optimalBandwidthNeeded int64, - direction LayerDirection, - preference LayerPreference, - availableChannelCapacity int64, - canPause bool, -) VideoAllocation { - targetLayers := InvalidLayers +func (f *Forwarder) getDistanceToDesired(brs Bitrates, targetLayers VideoLayers) int32 { + if f.muted { + return 0 + } - switch direction { - case LayerDirectionLowToHigh: - switch preference { - case LayerPreferenceSpatial: - for i := minLayers.spatial; i <= maxLayers.spatial; i++ { - for j := minLayers.temporal; j <= maxLayers.temporal; j++ { - if brs[i][j] != 0 && brs[i][j] <= availableChannelCapacity { - targetLayers = VideoLayers{ - spatial: i, - temporal: j, - } - break - } - } - if targetLayers != InvalidLayers { - break - } + distance := int32(0) + for s := f.maxLayers.spatial; s >= 0; s-- { + found := false + for t := f.maxLayers.temporal; t >= 0; t-- { + if brs[s][t] == 0 { + continue } - case LayerPreferenceTemporal: - for i := minLayers.temporal; i <= maxLayers.temporal; i++ { - for j := minLayers.spatial; j <= maxLayers.spatial; j++ { - if brs[j][i] != 0 && brs[j][i] <= availableChannelCapacity { - targetLayers = VideoLayers{ - spatial: j, - temporal: i, - } - break - } - } - if targetLayers != InvalidLayers { - break - } + if s == targetLayers.spatial && t == targetLayers.temporal { + found = true + break } + + distance++ } - case LayerDirectionHighToLow: - switch preference { - case LayerPreferenceSpatial: - for i := maxLayers.spatial; i >= minLayers.spatial; i-- { - for j := maxLayers.temporal; j >= minLayers.temporal; j-- { - if brs[i][j] != 0 && brs[i][j] <= availableChannelCapacity { - targetLayers = VideoLayers{ - spatial: i, - temporal: j, - } - break - } - } - if targetLayers != InvalidLayers { - break - } - } - case LayerPreferenceTemporal: - for i := maxLayers.temporal; i >= minLayers.temporal; i-- { - for j := maxLayers.spatial; j >= minLayers.spatial; j-- { - if brs[j][i] != 0 && brs[j][i] <= availableChannelCapacity { - targetLayers = VideoLayers{ - spatial: j, - temporal: i, - } - break - } - } - if targetLayers != InvalidLayers { - break - } - } + + if found { + break } } - if targetLayers == InvalidLayers && !canPause { - // - // Do not pause if preserving even if allocation does not fit in available channel capacity. - // - // Note that currently streamed layers could have a different bitrate compared to - // when the allocation was done. But not updating to avoid any unnecessary perturbation - // in the allocation system. Let channel changes happen and update state as needed via - // a fresh allocation. - // + return distance +} + +func (f *Forwarder) IsDeficient() bool { + f.lock.RLock() + defer f.lock.RUnlock() + + return f.lastAllocation.state == VideoAllocationStateDeficient +} + +func (f *Forwarder) BandwidthRequested() int64 { + f.lock.RLock() + defer f.lock.RUnlock() + + return f.lastAllocation.bandwidthRequested +} + +func (f *Forwarder) DistanceToDesired() int32 { + f.lock.RLock() + defer f.lock.RUnlock() + + return f.lastAllocation.distanceToDesired +} + +func (f *Forwarder) Allocate(availableChannelCapacity int64, brs Bitrates) VideoAllocation { + f.lock.Lock() + defer f.lock.Unlock() + + if f.kind == webrtc.RTPCodecTypeAudio { return f.lastAllocation } - var allocation VideoAllocation - - // change in streaming state? - switch { - case f.targetLayers != InvalidLayers && targetLayers == InvalidLayers: - allocation.change = VideoStreamingChangePausing - case f.targetLayers == InvalidLayers && targetLayers != InvalidLayers: - allocation.change = VideoStreamingChangeResuming - } - - // how much bandwidth is needed and delta from previous allocation - if targetLayers == InvalidLayers { - allocation.bandwidthRequested = 0 - } else { - allocation.bandwidthRequested = brs[targetLayers.spatial][targetLayers.temporal] - } - allocation.bandwidthDelta = allocation.bandwidthRequested - f.lastAllocation.bandwidthRequested - - // state of allocation - if allocation.bandwidthRequested == optimalBandwidthNeeded { - allocation.state = VideoAllocationStateOptimal - } else { - allocation.state = VideoAllocationStateDeficient - } - - allocation.availableLayers = f.availableLayers - allocation.bitrates = brs - allocation.targetLayers = targetLayers - - return allocation -} - -func (f *Forwarder) allocate(availableChannelCapacity int64, canPause bool, brs [3][4]int64) { - // should never get called on audio tracks, just for safety - if f.kind == webrtc.RTPCodecTypeAudio { - return - } - - if f.muted { - f.lastAllocation = VideoAllocation{ - state: VideoAllocationStateMuted, - change: VideoStreamingChangeNone, - bandwidthRequested: 0, - bandwidthDelta: 0 - f.lastAllocation.bandwidthRequested, - availableLayers: f.availableLayers, - bitrates: brs, - targetLayers: f.targetLayers, - } - return - } - optimalBandwidthNeeded := f.getOptimalBandwidthNeeded(brs) - if optimalBandwidthNeeded == 0 { + + state := VideoAllocationStateNone + change := VideoStreamingChangeNone + bandwidthRequested := int64(0) + targetLayers := InvalidLayers + + switch { + case f.muted: + state = VideoAllocationStateMuted + case optimalBandwidthNeeded == 0: if len(f.availableLayers) == 0 { // feed is dry - f.lastAllocation = VideoAllocation{ - state: VideoAllocationStateFeedDry, - change: VideoStreamingChangeNone, - bandwidthRequested: 0, - bandwidthDelta: 0 - f.lastAllocation.bandwidthRequested, - availableLayers: f.availableLayers, - bitrates: brs, - targetLayers: f.targetLayers, - } - return - } - - // feed bitrate is not yet calculated - if availableChannelCapacity == ChannelCapacityInfinity { - // - // Channel capacity allows a free pass. - // So, resume with the highest layer available <= max subscribed layer - // If already resumed, move allocation to the highest available layer <= max subscribed layer - // - change := VideoStreamingChangeNone - if f.targetLayers == InvalidLayers { - change = VideoStreamingChangeResuming - } - - f.targetLayers.spatial = int32(f.availableLayers[len(f.availableLayers)-1]) - if f.targetLayers.spatial > f.maxLayers.spatial { - f.targetLayers.spatial = f.maxLayers.spatial - } - - f.targetLayers.temporal = int32(math.Max(0, float64(f.maxLayers.temporal))) - - f.lastAllocation = VideoAllocation{ - state: VideoAllocationStateAwaitingMeasurement, - change: change, - bandwidthRequested: 0, // unavailable yet - bandwidthDelta: 0 - f.lastAllocation.bandwidthRequested, - availableLayers: f.availableLayers, - bitrates: brs, - targetLayers: f.targetLayers, - } + state = VideoAllocationStateFeedDry } else { - // if not optimistically started, nothing else to do - if f.targetLayers == InvalidLayers { - f.lastAllocation = VideoAllocation{ - state: VideoAllocationStateDeficient, - change: VideoStreamingChangeNone, - bandwidthRequested: 0, // unavailable yet - bandwidthDelta: 0 - f.lastAllocation.bandwidthRequested, - availableLayers: f.availableLayers, - bitrates: brs, - targetLayers: f.targetLayers, - } - } else if canPause { - // disable it as it is not known how big this stream is - // and if it will fit in the available channel capacity - f.disable() + // feed bitrate is not yet calculated + state = VideoAllocationStateAwaitingMeasurement - f.lastAllocation = VideoAllocation{ - state: VideoAllocationStateDeficient, - change: VideoStreamingChangePausing, - bandwidthRequested: 0, // unavailable yet - bandwidthDelta: 0 - f.lastAllocation.bandwidthRequested, - availableLayers: f.availableLayers, - bitrates: brs, - targetLayers: f.targetLayers, + if availableChannelCapacity == ChannelCapacityInfinity { + // + // Channel capacity allows a free pass. + // So, resume with the highest layer available <= max subscribed layer + // If already resumed, move allocation to the highest available layer <= max subscribed layer + // + targetLayers.spatial = int32(math.Min(float64(f.maxLayers.spatial), float64(f.availableLayers[len(f.availableLayers)-1]))) + targetLayers.temporal = int32(math.Max(0, float64(f.maxLayers.temporal))) + + if f.targetLayers == InvalidLayers { + change = VideoStreamingChangeResuming + } + } else { + // disable forwarding as it is not known how big this stream is + // and if it will fit in the available channel capacity + f.currentLayers = InvalidLayers + + state = VideoAllocationStateDeficient + + if f.targetLayers != InvalidLayers { + change = VideoStreamingChangePausing } } } - return + default: + // allocate best layer that fits + for s := f.maxLayers.spatial; s >= 0; s-- { + for t := f.maxLayers.temporal; t >= 0; t-- { + if brs[s][t] == 0 { + continue + } + + if brs[s][t] <= availableChannelCapacity { + targetLayers = VideoLayers{ + spatial: s, + temporal: t, + } + + bandwidthRequested = brs[s][t] + if bandwidthRequested == optimalBandwidthNeeded { + state = VideoAllocationStateOptimal + } else { + state = VideoAllocationStateDeficient + } + + if f.targetLayers == InvalidLayers { + change = VideoStreamingChangeResuming + } + break + } + } + if bandwidthRequested != 0 { + break + } + } + + if bandwidthRequested == 0 { + state = VideoAllocationStateDeficient + + if f.targetLayers != InvalidLayers { + change = VideoStreamingChangePausing + } + } } - f.lastAllocation = f.findBestLayers( - MinLayers, - f.maxLayers, - brs, - optimalBandwidthNeeded, - LayerDirectionHighToLow, - f.layerPref, - availableChannelCapacity, - canPause, - ) + f.lastAllocation = VideoAllocation{ + state: state, + change: change, + bandwidthRequested: bandwidthRequested, + bandwidthDelta: bandwidthRequested - f.lastAllocation.bandwidthRequested, + availableLayers: f.availableLayers, + bitrates: brs, + targetLayers: targetLayers, + distanceToDesired: f.getDistanceToDesired(brs, targetLayers), + } f.targetLayers = f.lastAllocation.targetLayers -} + if f.targetLayers == InvalidLayers { + f.currentLayers = InvalidLayers + } -func (f *Forwarder) Allocate(availableChannelCapacity int64, brs [3][4]int64) VideoAllocation { - f.lock.Lock() - defer f.lock.Unlock() - - f.allocate(availableChannelCapacity, true, brs) return f.lastAllocation } -func (f *Forwarder) TryAllocate(additionalChannelCapacity int64, brs [3][4]int64) VideoAllocation { +func (f *Forwarder) ProvisionalAllocatePrepare(bitrates Bitrates) { f.lock.Lock() defer f.lock.Unlock() - f.allocate(f.lastAllocation.bandwidthRequested+additionalChannelCapacity, false, brs) + f.provisional = &VideoAllocationProvisional{ + layers: InvalidLayers, + muted: f.muted, + bitrates: bitrates, + } +} + +func (f *Forwarder) ProvisionalAllocate(availableChannelCapacity int64, layers VideoLayers) int64 { + f.lock.Lock() + defer f.lock.Unlock() + + if f.provisional.muted { + return 0 + } + + if layers.GreaterThan(f.maxLayers) { + return 0 + } + + requiredBitrate := f.provisional.bitrates[layers.spatial][layers.temporal] + if requiredBitrate == 0 { + return 0 + } + + alreadyAllocatedBitrate := int64(0) + if f.provisional.layers != InvalidLayers { + alreadyAllocatedBitrate = f.provisional.bitrates[f.provisional.layers.spatial][f.provisional.layers.temporal] + } + + if requiredBitrate <= (availableChannelCapacity + alreadyAllocatedBitrate) { + f.provisional.layers = layers + return requiredBitrate + } + + return 0 +} + +func (f *Forwarder) ProvisionalAllocateCommit() VideoAllocation { + f.lock.Lock() + defer f.lock.Unlock() + + optimalBandwidthNeeded := f.getOptimalBandwidthNeeded(f.provisional.bitrates) + + state := VideoAllocationStateNone + change := VideoStreamingChangeNone + bandwidthRequested := int64(0) + + switch { + case f.muted: + state = VideoAllocationStateMuted + case optimalBandwidthNeeded == 0: + if len(f.availableLayers) == 0 { + // feed is dry + state = VideoAllocationStateFeedDry + } else { + // feed bitrate is not yet calculated + state = VideoAllocationStateDeficient + + // disable forwarding as it is not known how big this stream is + // and if it will fit in the available channel capacity + + if f.targetLayers != InvalidLayers { + change = VideoStreamingChangePausing + } + } + case f.provisional.layers == InvalidLayers: + state = VideoAllocationStateDeficient + + if f.targetLayers != InvalidLayers { + change = VideoStreamingChangePausing + } + default: + bandwidthRequested = f.provisional.bitrates[f.provisional.layers.spatial][f.provisional.layers.temporal] + if bandwidthRequested == optimalBandwidthNeeded { + state = VideoAllocationStateOptimal + } else { + state = VideoAllocationStateDeficient + } + + if f.targetLayers == InvalidLayers { + change = VideoStreamingChangeResuming + } + } + + f.lastAllocation = VideoAllocation{ + state: state, + change: change, + bandwidthRequested: bandwidthRequested, + bandwidthDelta: bandwidthRequested - f.lastAllocation.bandwidthRequested, + availableLayers: f.availableLayers, + bitrates: f.provisional.bitrates, + targetLayers: f.provisional.layers, + distanceToDesired: f.getDistanceToDesired(f.provisional.bitrates, f.provisional.layers), + } + f.targetLayers = f.lastAllocation.targetLayers + if f.targetLayers == InvalidLayers { + f.currentLayers = InvalidLayers + } + + f.provisional = nil + return f.lastAllocation } -func (f *Forwarder) FinalizeAllocate(brs [3][4]int64) VideoAllocation { +func (f *Forwarder) FinalizeAllocate(brs Bitrates) VideoAllocation { f.lock.Lock() defer f.lock.Unlock() @@ -587,7 +606,12 @@ func (f *Forwarder) FinalizeAllocate(brs [3][4]int64) VideoAllocation { bandwidthDelta: 0 - f.lastAllocation.bandwidthRequested, availableLayers: f.availableLayers, bitrates: brs, - targetLayers: f.targetLayers, + targetLayers: InvalidLayers, + distanceToDesired: f.getDistanceToDesired(brs, InvalidLayers), + } + f.targetLayers = f.lastAllocation.targetLayers + if f.targetLayers == InvalidLayers { + f.currentLayers = InvalidLayers } } @@ -595,21 +619,44 @@ func (f *Forwarder) FinalizeAllocate(brs [3][4]int64) VideoAllocation { return f.lastAllocation } - f.lastAllocation = f.findBestLayers( - MinLayers, - f.maxLayers, - brs, - optimalBandwidthNeeded, - LayerDirectionHighToLow, - f.layerPref, - ChannelCapacityInfinity, - false, - ) - f.targetLayers = f.lastAllocation.targetLayers + // finalize using optimal layer + for s := f.maxLayers.spatial; s >= 0; s-- { + for t := f.maxLayers.temporal; t >= 0; t-- { + bandwidthRequested := brs[s][t] + if bandwidthRequested == 0 { + continue + } + + state := VideoAllocationStateOptimal + if bandwidthRequested != optimalBandwidthNeeded { + state = VideoAllocationStateDeficient + } + + change := VideoStreamingChangeNone + if f.targetLayers == InvalidLayers { + change = VideoStreamingChangeResuming + } + + targetLayers := VideoLayers{spatial: s, temporal: t} + f.lastAllocation = VideoAllocation{ + state: state, + change: change, + bandwidthRequested: bandwidthRequested, + bandwidthDelta: bandwidthRequested - f.lastAllocation.bandwidthRequested, + availableLayers: f.availableLayers, + bitrates: brs, + targetLayers: targetLayers, + distanceToDesired: f.getDistanceToDesired(brs, targetLayers), + } + f.targetLayers = f.lastAllocation.targetLayers + return f.lastAllocation + } + } + return f.lastAllocation } -func (f *Forwarder) AllocateNextHigher(brs [3][4]int64) (VideoAllocation, bool) { +func (f *Forwarder) AllocateNextHigher(brs Bitrates) (VideoAllocation, bool) { f.lock.Lock() defer f.lock.Unlock() @@ -635,77 +682,116 @@ func (f *Forwarder) AllocateNextHigher(brs [3][4]int64) (VideoAllocation, bool) // try moving temporal layer up in currently streaming spatial layer if f.targetLayers != InvalidLayers { - minLayers := VideoLayers{ - spatial: f.targetLayers.spatial, - temporal: f.targetLayers.temporal + 1, - } - maxLayers := VideoLayers{ - spatial: f.targetLayers.spatial, - temporal: f.maxLayers.temporal, - } - allocation := f.findBestLayers( - minLayers, - maxLayers, - brs, - optimalBandwidthNeeded, - LayerDirectionLowToHigh, - f.layerPref, - ChannelCapacityInfinity, - false, - ) - if allocation.targetLayers != f.targetLayers { - f.lastAllocation = allocation - f.targetLayers = allocation.targetLayers + for t := f.targetLayers.temporal + 1; t <= f.maxLayers.temporal; t++ { + bandwidthRequested := brs[f.targetLayers.spatial][t] + if bandwidthRequested == 0 { + continue + } + + state := VideoAllocationStateOptimal + if bandwidthRequested != optimalBandwidthNeeded { + state = VideoAllocationStateDeficient + } + + targetLayers := VideoLayers{spatial: f.targetLayers.spatial, temporal: t} + f.lastAllocation = VideoAllocation{ + state: state, + change: VideoStreamingChangeNone, + bandwidthRequested: bandwidthRequested, + bandwidthDelta: bandwidthRequested - f.lastAllocation.bandwidthRequested, + availableLayers: f.availableLayers, + bitrates: brs, + targetLayers: targetLayers, + distanceToDesired: f.getDistanceToDesired(brs, targetLayers), + } + f.targetLayers = f.lastAllocation.targetLayers return f.lastAllocation, true } } // try moving spatial layer up if temporal layer move up is not available - minLayers := VideoLayers{ - spatial: f.targetLayers.spatial + 1, - temporal: 0, - } - maxLayers := VideoLayers{ - spatial: f.maxLayers.spatial, - temporal: f.maxLayers.temporal, - } - allocation := f.findBestLayers( - minLayers, - maxLayers, - brs, - optimalBandwidthNeeded, - LayerDirectionLowToHigh, - f.layerPref, - ChannelCapacityInfinity, - false, - ) - if allocation.targetLayers != f.targetLayers { - f.lastAllocation = allocation - f.targetLayers = allocation.targetLayers - return f.lastAllocation, true + for s := f.targetLayers.spatial + 1; s <= f.maxLayers.spatial; s++ { + for t := int32(0); t <= f.maxLayers.temporal; t++ { + bandwidthRequested := brs[s][t] + if bandwidthRequested == 0 { + continue + } + + state := VideoAllocationStateOptimal + if bandwidthRequested != optimalBandwidthNeeded { + state = VideoAllocationStateDeficient + } + + change := VideoStreamingChangeNone + if f.targetLayers == InvalidLayers { + change = VideoStreamingChangeResuming + } + + targetLayers := VideoLayers{spatial: s, temporal: t} + f.lastAllocation = VideoAllocation{ + state: state, + change: change, + bandwidthRequested: bandwidthRequested, + bandwidthDelta: bandwidthRequested - f.lastAllocation.bandwidthRequested, + availableLayers: f.availableLayers, + bitrates: brs, + targetLayers: targetLayers, + distanceToDesired: f.getDistanceToDesired(brs, targetLayers), + } + f.targetLayers = f.lastAllocation.targetLayers + return f.lastAllocation, true + } } return f.lastAllocation, false } -/* -func (f *Forwarder) AllocationState() VideoAllocationState { - f.lock.RLock() - defer f.lock.RUnlock() +func (f *Forwarder) Pause(brs Bitrates) VideoAllocation { + f.lock.Lock() + defer f.lock.Unlock() - return f.lastAllocationState -} + optimalBandwidthNeeded := f.getOptimalBandwidthNeeded(brs) -func (f *Forwarder) AllocationBandwidth() int64 { - f.lock.RLock() - defer f.lock.RUnlock() + state := VideoAllocationStateNone + change := VideoStreamingChangeNone - return f.lastAllocationRequestBps -} -*/ -func (f *Forwarder) LastAllocation() VideoAllocation { - f.lock.RLock() - defer f.lock.RUnlock() + switch { + case f.muted: + state = VideoAllocationStateMuted + case optimalBandwidthNeeded == 0: + if len(f.availableLayers) == 0 { + // feed is dry + state = VideoAllocationStateFeedDry + } else { + // feed bitrate is not yet calculated + state = VideoAllocationStateDeficient + + if f.targetLayers != InvalidLayers { + change = VideoStreamingChangePausing + } + } + default: + state = VideoAllocationStateDeficient + + if f.targetLayers != InvalidLayers { + change = VideoStreamingChangePausing + } + } + + f.lastAllocation = VideoAllocation{ + state: state, + change: change, + bandwidthRequested: 0, + bandwidthDelta: 0 - f.lastAllocation.bandwidthRequested, + availableLayers: f.availableLayers, + bitrates: brs, + targetLayers: InvalidLayers, + distanceToDesired: f.getDistanceToDesired(brs, InvalidLayers), + } + f.targetLayers = f.lastAllocation.targetLayers + if f.targetLayers == InvalidLayers { + f.currentLayers = InvalidLayers + } return f.lastAllocation } diff --git a/pkg/sfu/forwarder_test.go b/pkg/sfu/forwarder_test.go index e18964af2..55577e40c 100644 --- a/pkg/sfu/forwarder_test.go +++ b/pkg/sfu/forwarder_test.go @@ -12,6 +12,11 @@ import ( "github.com/stretchr/testify/require" ) +func disable(f *Forwarder) { + f.currentLayers = InvalidLayers + f.targetLayers = InvalidLayers +} + func TestForwarderMute(t *testing.T) { f := NewForwarder(testutils.TestOpusCodec, webrtc.RTPCodecTypeAudio) require.False(t, f.Muted()) @@ -31,11 +36,11 @@ func TestForwarderLayersAudio(t *testing.T) { require.Equal(t, InvalidLayers, f.CurrentLayers()) require.Equal(t, InvalidLayers, f.TargetLayers()) - changed, layers, _ := f.SetMaxSpatialLayer(1) + changed, layers := f.SetMaxSpatialLayer(1) require.False(t, changed) require.Equal(t, InvalidLayers, layers) - changed, layers, _ = f.SetMaxTemporalLayer(1) + changed, layers = f.SetMaxTemporalLayer(1) require.False(t, changed) require.Equal(t, InvalidLayers, layers) @@ -55,11 +60,11 @@ func TestForwarderLayersVideo(t *testing.T) { require.Equal(t, InvalidLayers, f.CurrentLayers()) require.Equal(t, InvalidLayers, f.TargetLayers()) - changed, layers, _ := f.SetMaxSpatialLayer(DefaultMaxLayerSpatial) + changed, layers := f.SetMaxSpatialLayer(DefaultMaxLayerSpatial) require.False(t, changed) require.Equal(t, InvalidLayers, layers) - changed, layers, _ = f.SetMaxSpatialLayer(DefaultMaxLayerSpatial - 1) + changed, layers = f.SetMaxSpatialLayer(DefaultMaxLayerSpatial - 1) require.True(t, changed) expectedLayers = VideoLayers{ spatial: DefaultMaxLayerSpatial - 1, @@ -68,11 +73,11 @@ func TestForwarderLayersVideo(t *testing.T) { require.Equal(t, expectedLayers, layers) require.Equal(t, expectedLayers, f.MaxLayers()) - changed, layers, _ = f.SetMaxTemporalLayer(DefaultMaxLayerTemporal) + changed, layers = f.SetMaxTemporalLayer(DefaultMaxLayerTemporal) require.False(t, changed) require.Equal(t, InvalidLayers, layers) - changed, layers, _ = f.SetMaxTemporalLayer(DefaultMaxLayerTemporal - 1) + changed, layers = f.SetMaxTemporalLayer(DefaultMaxLayerTemporal - 1) require.True(t, changed) expectedLayers = VideoLayers{ spatial: DefaultMaxLayerSpatial - 1, @@ -115,8 +120,8 @@ func TestForwarderUptrackLayersChange(t *testing.T) { func TestForwarderAllocate(t *testing.T) { f := NewForwarder(testutils.TestVP8Codec, webrtc.RTPCodecTypeVideo) - emptyBitrates := [DefaultMaxLayerSpatial + 1][DefaultMaxLayerTemporal + 1]int64{} - bitrates := [DefaultMaxLayerSpatial + 1][DefaultMaxLayerTemporal + 1]int64{ + emptyBitrates := Bitrates{} + bitrates := Bitrates{ {2, 3, 0, 0}, {4, 0, 0, 5}, {0, 7, 0, 0}, @@ -124,15 +129,16 @@ func TestForwarderAllocate(t *testing.T) { // muted should not consume any bandwidth f.Mute(true) - f.disable() + disable(f) expectedResult := VideoAllocation{ - change: VideoStreamingChangeNone, state: VideoAllocationStateMuted, + change: VideoStreamingChangeNone, bandwidthRequested: 0, bandwidthDelta: 0, availableLayers: nil, bitrates: bitrates, targetLayers: InvalidLayers, + distanceToDesired: 0, } result := f.Allocate(ChannelCapacityInfinity, bitrates) require.Equal(t, expectedResult, result) @@ -141,15 +147,16 @@ func TestForwarderAllocate(t *testing.T) { // feed dry state f.Mute(false) f.lastAllocation.state = VideoAllocationStateNone - f.disable() + disable(f) expectedResult = VideoAllocation{ - change: VideoStreamingChangeNone, state: VideoAllocationStateFeedDry, + change: VideoStreamingChangeNone, bandwidthRequested: 0, bandwidthDelta: 0, availableLayers: nil, bitrates: emptyBitrates, targetLayers: InvalidLayers, + distanceToDesired: 0, } result = f.Allocate(ChannelCapacityInfinity, emptyBitrates) require.Equal(t, expectedResult, result) @@ -157,20 +164,21 @@ func TestForwarderAllocate(t *testing.T) { // awaiting measurement, i. e. bitrates are not available, but layers available f.lastAllocation.state = VideoAllocationStateNone - f.disable() + disable(f) f.UptrackLayersChange([]uint16{0}) expectedTargetLayers := VideoLayers{ spatial: 0, temporal: DefaultMaxLayerTemporal, } expectedResult = VideoAllocation{ - change: VideoStreamingChangeResuming, state: VideoAllocationStateAwaitingMeasurement, + change: VideoStreamingChangeResuming, bandwidthRequested: 0, bandwidthDelta: 0, availableLayers: []uint16{0}, bitrates: emptyBitrates, targetLayers: expectedTargetLayers, + distanceToDesired: 0, } result = f.Allocate(ChannelCapacityInfinity, emptyBitrates) require.Equal(t, expectedResult, result) @@ -180,13 +188,14 @@ func TestForwarderAllocate(t *testing.T) { // while awaiting measurement, less than infinite channel capacity should pause the stream expectedResult = VideoAllocation{ - change: VideoStreamingChangePausing, state: VideoAllocationStateDeficient, + change: VideoStreamingChangePausing, bandwidthRequested: 0, bandwidthDelta: 0, availableLayers: []uint16{0}, bitrates: emptyBitrates, targetLayers: InvalidLayers, + distanceToDesired: 0, } result = f.Allocate(ChannelCapacityInfinity-1, emptyBitrates) require.Equal(t, expectedResult, result) @@ -200,13 +209,14 @@ func TestForwarderAllocate(t *testing.T) { temporal: 1, } expectedResult = VideoAllocation{ - change: VideoStreamingChangeResuming, state: VideoAllocationStateOptimal, + change: VideoStreamingChangeResuming, bandwidthRequested: bitrates[2][1], bandwidthDelta: bitrates[2][1], availableLayers: []uint16{0}, bitrates: bitrates, targetLayers: expectedTargetLayers, + distanceToDesired: 0, } result = f.Allocate(ChannelCapacityInfinity-1, bitrates) require.Equal(t, expectedResult, result) @@ -220,13 +230,14 @@ func TestForwarderAllocate(t *testing.T) { temporal: 3, } expectedResult = VideoAllocation{ - change: VideoStreamingChangeNone, state: VideoAllocationStateDeficient, + change: VideoStreamingChangeNone, bandwidthRequested: bitrates[1][3], bandwidthDelta: bitrates[1][3] - bitrates[2][1], availableLayers: []uint16{0}, bitrates: bitrates, targetLayers: expectedTargetLayers, + distanceToDesired: 1, } result = f.Allocate(bitrates[2][1]-1, bitrates) require.Equal(t, expectedResult, result) @@ -236,13 +247,14 @@ func TestForwarderAllocate(t *testing.T) { // give it a bitrate that cannot fit any layer expectedResult = VideoAllocation{ - change: VideoStreamingChangePausing, state: VideoAllocationStateDeficient, + change: VideoStreamingChangePausing, bandwidthRequested: 0, bandwidthDelta: 0 - bitrates[1][3], availableLayers: []uint16{0}, bitrates: bitrates, targetLayers: InvalidLayers, + distanceToDesired: 5, } result = f.Allocate(bitrates[0][0]-1, bitrates) require.Equal(t, expectedResult, result) @@ -251,83 +263,87 @@ func TestForwarderAllocate(t *testing.T) { require.Equal(t, InvalidLayers, f.TargetLayers()) } -func TestForwarderTryAllocate(t *testing.T) { +func TestForwarderProvisionalAllocate(t *testing.T) { f := NewForwarder(testutils.TestVP8Codec, webrtc.RTPCodecTypeVideo) - // adjust target layers per given additional channel capacity (which can be negative), - bitrates := [DefaultMaxLayerSpatial + 1][DefaultMaxLayerTemporal + 1]int64{ - {2, 3, 0, 0}, - {4, 0, 0, 5}, - {0, 7, 0, 0}, + bitrates := Bitrates{ + {1, 2, 3, 4}, + {5, 6, 7, 8}, + {9, 10, 11, 12}, } - f.lastAllocation.state = VideoAllocationStateDeficient - f.lastAllocation.bandwidthRequested = bitrates[1][3] - f.targetLayers = VideoLayers{ - spatial: 1, - temporal: 3, - } + f.ProvisionalAllocatePrepare(bitrates) + usedBitrate := f.ProvisionalAllocate(bitrates[2][3], VideoLayers{spatial: 0, temporal: 0}) + require.Equal(t, bitrates[0][0], usedBitrate) + + usedBitrate = f.ProvisionalAllocate(bitrates[2][3], VideoLayers{spatial: 1, temporal: 2}) + require.Equal(t, bitrates[1][2], usedBitrate) + + // available not enough to reach (2, 2), allocating at (2, 2) should not succeed + usedBitrate = f.ProvisionalAllocate(bitrates[2][2]-bitrates[1][2]-1, VideoLayers{spatial: 2, temporal: 2}) + require.Equal(t, int64(0), usedBitrate) + + // committing should set target to (1, 2) expectedTargetLayers := VideoLayers{ spatial: 1, - temporal: 0, + temporal: 2, } expectedResult := VideoAllocation{ - change: VideoStreamingChangeNone, state: VideoAllocationStateDeficient, - bandwidthRequested: bitrates[1][0], - bandwidthDelta: bitrates[1][0] - bitrates[1][3], + change: VideoStreamingChangeResuming, + bandwidthRequested: bitrates[1][2], + bandwidthDelta: bitrates[1][2], availableLayers: nil, bitrates: bitrates, targetLayers: expectedTargetLayers, + distanceToDesired: 5, } - result := f.TryAllocate(-1, bitrates) + result := f.ProvisionalAllocateCommit() require.Equal(t, expectedResult, result) require.Equal(t, expectedResult, f.lastAllocation) - require.Equal(t, InvalidLayers, f.CurrentLayers()) require.Equal(t, expectedTargetLayers, f.TargetLayers()) +} - // but should not pause even if no layer fits, i. e. preserve current - expectedResult = VideoAllocation{ - change: VideoStreamingChangeNone, - state: VideoAllocationStateDeficient, - bandwidthRequested: bitrates[1][0], - bandwidthDelta: -1, - availableLayers: nil, - bitrates: bitrates, - targetLayers: expectedTargetLayers, - } - result = f.TryAllocate(-3, bitrates) - require.Equal(t, expectedResult, result) - require.Equal(t, expectedResult, f.lastAllocation) - require.Equal(t, InvalidLayers, f.CurrentLayers()) - require.Equal(t, expectedTargetLayers, f.TargetLayers()) +func TestForwarderProvisionalAllocateMute(t *testing.T) { + f := NewForwarder(testutils.TestVP8Codec, webrtc.RTPCodecTypeVideo) - // can catch up to optimal given enough additional channel capacity - expectedTargetLayers = VideoLayers{ - spatial: 2, - temporal: 1, + bitrates := Bitrates{ + {1, 2, 3, 4}, + {5, 6, 7, 8}, + {9, 10, 11, 12}, } - expectedResult = VideoAllocation{ + + f.Mute(true) + f.ProvisionalAllocatePrepare(bitrates) + + usedBitrate := f.ProvisionalAllocate(bitrates[2][3], VideoLayers{spatial: 0, temporal: 0}) + require.Equal(t, int64(0), usedBitrate) + + usedBitrate = f.ProvisionalAllocate(bitrates[2][3], VideoLayers{spatial: 1, temporal: 2}) + require.Equal(t, int64(0), usedBitrate) + + // committing should set target to InvalidLayers as track is muted + expectedResult := VideoAllocation{ + state: VideoAllocationStateMuted, change: VideoStreamingChangeNone, - state: VideoAllocationStateOptimal, - bandwidthRequested: bitrates[2][1], - bandwidthDelta: bitrates[2][1] - bitrates[1][0], + bandwidthRequested: 0, + bandwidthDelta: 0, availableLayers: nil, bitrates: bitrates, - targetLayers: expectedTargetLayers, + targetLayers: InvalidLayers, + distanceToDesired: 0, } - result = f.TryAllocate(10, bitrates) + result := f.ProvisionalAllocateCommit() require.Equal(t, expectedResult, result) require.Equal(t, expectedResult, f.lastAllocation) - require.Equal(t, InvalidLayers, f.CurrentLayers()) - require.Equal(t, expectedTargetLayers, f.TargetLayers()) + require.Equal(t, InvalidLayers, f.TargetLayers()) } func TestForwarderFinalizeAllocate(t *testing.T) { f := NewForwarder(testutils.TestVP8Codec, webrtc.RTPCodecTypeVideo) - bitrates := [DefaultMaxLayerSpatial + 1][DefaultMaxLayerTemporal + 1]int64{ + bitrates := Bitrates{ {1, 2, 3, 4}, {5, 6, 7, 8}, {9, 10, 11, 12}, @@ -338,34 +354,36 @@ func TestForwarderFinalizeAllocate(t *testing.T) { require.Equal(t, VideoAllocationDefault, f.lastAllocation) f.lastAllocation.state = VideoAllocationStateMuted - f.disable() + disable(f) expectedResult := VideoAllocation{ - change: VideoStreamingChangeNone, state: VideoAllocationStateMuted, + change: VideoStreamingChangeNone, bandwidthRequested: 0, bandwidthDelta: 0, availableLayers: nil, bitrates: [3][4]int64{}, targetLayers: InvalidLayers, + distanceToDesired: 0, } result = f.FinalizeAllocate(bitrates) require.Equal(t, expectedResult, result) require.Equal(t, expectedResult, f.lastAllocation) f.lastAllocation.state = VideoAllocationStateAwaitingMeasurement - f.disable() + disable(f) expectedTargetLayers := VideoLayers{ spatial: 2, temporal: 3, } expectedResult = VideoAllocation{ - change: VideoStreamingChangeResuming, state: VideoAllocationStateOptimal, + change: VideoStreamingChangeResuming, bandwidthRequested: bitrates[2][3], bandwidthDelta: bitrates[2][3], availableLayers: nil, bitrates: bitrates, targetLayers: expectedTargetLayers, + distanceToDesired: 0, } result = f.FinalizeAllocate(bitrates) require.Equal(t, expectedResult, result) @@ -375,56 +393,59 @@ func TestForwarderFinalizeAllocate(t *testing.T) { // no layers available => feed dry f.lastAllocation.state = VideoAllocationStateAwaitingMeasurement - f.disable() + disable(f) expectedResult = VideoAllocation{ - change: VideoStreamingChangeNone, state: VideoAllocationStateFeedDry, + change: VideoStreamingChangeNone, bandwidthRequested: 0, bandwidthDelta: 0 - bitrates[2][3], availableLayers: nil, - bitrates: [DefaultMaxLayerSpatial + 1][DefaultMaxLayerTemporal + 1]int64{}, + bitrates: Bitrates{}, targetLayers: InvalidLayers, + distanceToDesired: 0, } - result = f.FinalizeAllocate([DefaultMaxLayerSpatial + 1][DefaultMaxLayerTemporal + 1]int64{}) + result = f.FinalizeAllocate(Bitrates{}) require.Equal(t, expectedResult, result) require.Equal(t, expectedResult, f.lastAllocation) // layers available, but still awaiting measurement f.lastAllocation.state = VideoAllocationStateAwaitingMeasurement - f.disable() + disable(f) f.UptrackLayersChange([]uint16{0, 1}) expectedResult = VideoAllocation{ - change: VideoStreamingChangeNone, state: VideoAllocationStateAwaitingMeasurement, + change: VideoStreamingChangeNone, bandwidthRequested: 0, bandwidthDelta: -12, availableLayers: nil, - bitrates: [DefaultMaxLayerSpatial + 1][DefaultMaxLayerTemporal + 1]int64{}, + bitrates: Bitrates{}, targetLayers: InvalidLayers, + distanceToDesired: 0, } - result = f.FinalizeAllocate([DefaultMaxLayerSpatial + 1][DefaultMaxLayerTemporal + 1]int64{}) + result = f.FinalizeAllocate(Bitrates{}) require.Equal(t, expectedResult, result) require.Equal(t, expectedResult, f.lastAllocation) // sparse layers - bitrates = [DefaultMaxLayerSpatial + 1][DefaultMaxLayerTemporal + 1]int64{ + bitrates = Bitrates{ {1, 2, 0, 0}, {5, 0, 0, 6}, {0, 0, 0, 0}, } - f.disable() + disable(f) expectedTargetLayers = VideoLayers{ spatial: 1, temporal: 3, } expectedResult = VideoAllocation{ - change: VideoStreamingChangeResuming, state: VideoAllocationStateOptimal, + change: VideoStreamingChangeResuming, bandwidthRequested: bitrates[1][3], bandwidthDelta: bitrates[1][3], availableLayers: []uint16{0, 1}, bitrates: bitrates, targetLayers: expectedTargetLayers, + distanceToDesired: 0, } result = f.FinalizeAllocate(bitrates) require.Equal(t, expectedResult, result) @@ -436,8 +457,8 @@ func TestForwarderFinalizeAllocate(t *testing.T) { func TestForwarderAllocateNextHigher(t *testing.T) { f := NewForwarder(testutils.TestOpusCodec, webrtc.RTPCodecTypeAudio) - emptyBitrates := [DefaultMaxLayerSpatial + 1][DefaultMaxLayerTemporal + 1]int64{} - bitrates := [DefaultMaxLayerSpatial + 1][DefaultMaxLayerTemporal + 1]int64{ + emptyBitrates := Bitrates{} + bitrates := Bitrates{ {2, 3, 0, 0}, {4, 0, 0, 5}, {0, 7, 0, 0}, @@ -459,13 +480,14 @@ func TestForwarderAllocateNextHigher(t *testing.T) { f.lastAllocation.state = VideoAllocationStateDeficient f.targetLayers.spatial = 0 expectedResult := VideoAllocation{ - change: VideoStreamingChangeNone, state: VideoAllocationStateDeficient, + change: VideoStreamingChangeNone, bandwidthRequested: 0, bandwidthDelta: 0, availableLayers: nil, bitrates: emptyBitrates, targetLayers: InvalidLayers, + distanceToDesired: 0, } result, boosted = f.AllocateNextHigher(bitrates) require.Equal(t, expectedResult, result) @@ -484,13 +506,14 @@ func TestForwarderAllocateNextHigher(t *testing.T) { // empty bitrates cannot increase layer expectedResult = VideoAllocation{ - change: VideoStreamingChangeNone, state: VideoAllocationStateDeficient, + change: VideoStreamingChangeNone, bandwidthRequested: 2, bandwidthDelta: 0, availableLayers: nil, bitrates: emptyBitrates, targetLayers: InvalidLayers, + distanceToDesired: 0, } result, boosted = f.AllocateNextHigher(emptyBitrates) require.Equal(t, expectedResult, result) @@ -503,13 +526,14 @@ func TestForwarderAllocateNextHigher(t *testing.T) { temporal: 1, } expectedResult = VideoAllocation{ - change: VideoStreamingChangeNone, state: VideoAllocationStateDeficient, + change: VideoStreamingChangeNone, bandwidthRequested: 3, bandwidthDelta: 1, availableLayers: nil, bitrates: bitrates, targetLayers: expectedTargetLayers, + distanceToDesired: 3, } result, boosted = f.AllocateNextHigher(bitrates) require.Equal(t, expectedResult, result) @@ -524,13 +548,14 @@ func TestForwarderAllocateNextHigher(t *testing.T) { temporal: 0, } expectedResult = VideoAllocation{ - change: VideoStreamingChangeNone, state: VideoAllocationStateDeficient, + change: VideoStreamingChangeNone, bandwidthRequested: 4, bandwidthDelta: 1, availableLayers: nil, bitrates: bitrates, targetLayers: expectedTargetLayers, + distanceToDesired: 2, } result, boosted = f.AllocateNextHigher(bitrates) require.Equal(t, expectedResult, result) @@ -546,13 +571,14 @@ func TestForwarderAllocateNextHigher(t *testing.T) { temporal: 3, } expectedResult = VideoAllocation{ - change: VideoStreamingChangeNone, state: VideoAllocationStateDeficient, + change: VideoStreamingChangeNone, bandwidthRequested: 5, bandwidthDelta: 1, availableLayers: nil, bitrates: bitrates, targetLayers: expectedTargetLayers, + distanceToDesired: 1, } result, boosted = f.AllocateNextHigher(bitrates) require.Equal(t, expectedResult, result) @@ -567,13 +593,14 @@ func TestForwarderAllocateNextHigher(t *testing.T) { temporal: 1, } expectedResult = VideoAllocation{ - change: VideoStreamingChangeNone, state: VideoAllocationStateOptimal, + change: VideoStreamingChangeNone, bandwidthRequested: 7, bandwidthDelta: 2, availableLayers: nil, bitrates: bitrates, targetLayers: expectedTargetLayers, + distanceToDesired: 0, } result, boosted = f.AllocateNextHigher(bitrates) require.Equal(t, expectedResult, result) @@ -590,8 +617,8 @@ func TestForwarderAllocateNextHigher(t *testing.T) { require.Equal(t, expectedTargetLayers, f.TargetLayers()) require.False(t, boosted) - // turn off everything, allocating next layer should result - f.disable() + // turn off everything, allocating next layer should result in streaming lowest layers + disable(f) f.lastAllocation.state = VideoAllocationStateDeficient f.lastAllocation.bandwidthRequested = 0 @@ -600,13 +627,14 @@ func TestForwarderAllocateNextHigher(t *testing.T) { temporal: 0, } expectedResult = VideoAllocation{ - change: VideoStreamingChangeResuming, state: VideoAllocationStateDeficient, + change: VideoStreamingChangeResuming, bandwidthRequested: 2, bandwidthDelta: 2, availableLayers: nil, bitrates: bitrates, targetLayers: expectedTargetLayers, + distanceToDesired: 4, } result, boosted = f.AllocateNextHigher(bitrates) require.Equal(t, expectedResult, result) @@ -615,6 +643,67 @@ func TestForwarderAllocateNextHigher(t *testing.T) { require.True(t, boosted) } +func TestForwarderPause(t *testing.T) { + f := NewForwarder(testutils.TestVP8Codec, webrtc.RTPCodecTypeVideo) + + bitrates := Bitrates{ + {1, 2, 3, 4}, + {5, 6, 7, 8}, + {9, 10, 11, 12}, + } + + f.ProvisionalAllocatePrepare(bitrates) + f.ProvisionalAllocate(bitrates[2][3], VideoLayers{spatial: 0, temporal: 0}) + // should have set target at (0, 0) + f.ProvisionalAllocateCommit() + + expectedResult := VideoAllocation{ + state: VideoAllocationStateDeficient, + change: VideoStreamingChangePausing, + bandwidthRequested: 0, + bandwidthDelta: 0 - bitrates[0][0], + availableLayers: nil, + bitrates: bitrates, + targetLayers: InvalidLayers, + distanceToDesired: 12, + } + result := f.Pause(bitrates) + require.Equal(t, expectedResult, result) + require.Equal(t, expectedResult, f.lastAllocation) + require.Equal(t, InvalidLayers, f.TargetLayers()) +} + +func TestForwarderPauseMute(t *testing.T) { + f := NewForwarder(testutils.TestVP8Codec, webrtc.RTPCodecTypeVideo) + + bitrates := Bitrates{ + {1, 2, 3, 4}, + {5, 6, 7, 8}, + {9, 10, 11, 12}, + } + + f.ProvisionalAllocatePrepare(bitrates) + f.ProvisionalAllocate(bitrates[2][3], VideoLayers{spatial: 0, temporal: 0}) + // should have set target at (0, 0) + f.ProvisionalAllocateCommit() + + f.Mute(true) + expectedResult := VideoAllocation{ + state: VideoAllocationStateMuted, + change: VideoStreamingChangeNone, + bandwidthRequested: 0, + bandwidthDelta: 0 - bitrates[0][0], + availableLayers: nil, + bitrates: bitrates, + targetLayers: InvalidLayers, + distanceToDesired: 0, + } + result := f.Pause(bitrates) + require.Equal(t, expectedResult, result) + require.Equal(t, expectedResult, f.lastAllocation) + require.Equal(t, InvalidLayers, f.TargetLayers()) +} + func TestForwarderGetTranslationParamsMuted(t *testing.T) { f := NewForwarder(testutils.TestVP8Codec, webrtc.RTPCodecTypeVideo) f.Mute(true) @@ -1180,7 +1269,7 @@ func TestForwardGetSnTsForPadding(t *testing.T) { f.GetTranslationParams(extPkt, 0) // pause stream and get padding, it should still work - f.disable() + disable(f) // should get back frame end needed as the last packet did not have RTP marker set snts, err := f.GetSnTsForPadding(5) diff --git a/pkg/sfu/receiver.go b/pkg/sfu/receiver.go index 0a08412d8..9735bd923 100644 --- a/pkg/sfu/receiver.go +++ b/pkg/sfu/receiver.go @@ -16,17 +16,20 @@ import ( "github.com/livekit/livekit-server/pkg/sfu/buffer" ) +type Bitrates [DefaultMaxLayerSpatial + 1][DefaultMaxLayerTemporal + 1]int64 + // TrackReceiver defines a interface receive media from remote peer type TrackReceiver interface { TrackID() string StreamID() string - GetBitrateTemporalCumulative() [3][4]int64 + GetBitrateTemporalCumulative() Bitrates ReadRTP(buf []byte, layer uint8, sn uint16) (int, error) AddDownTrack(track TrackSender) DeleteDownTrack(peerID string) SendPLI(layer int32) GetSenderReportTime(layer int32) (rtpTS uint32, ntpTS uint64) Codec() webrtc.RTPCodecCapability + IsSimulcast() bool } // Receiver defines a interface for a track receivers @@ -34,11 +37,12 @@ type Receiver interface { TrackID() string StreamID() string Codec() webrtc.RTPCodecCapability + IsSimulcast() bool AddUpTrack(track *webrtc.TrackRemote, buffer *buffer.Buffer) AddDownTrack(track TrackSender) SetUpTrackPaused(paused bool) NumAvailableSpatialLayers() int - GetBitrateTemporalCumulative() [3][4]int64 + GetBitrateTemporalCumulative() Bitrates ReadRTP(buf []byte, layer uint8, sn uint16) (int, error) DeleteDownTrack(ID string) OnCloseHandler(fn func()) @@ -62,7 +66,7 @@ type WebRTCReceiver struct { onCloseHandler func() closeOnce sync.Once closed atomicBool - trackers [3]*StreamTracker + trackers [DefaultMaxLayerSpatial + 1]*StreamTracker useTrackers bool rtcpMu sync.Mutex @@ -71,10 +75,10 @@ type WebRTCReceiver struct { pliThrottle int64 bufferMu sync.RWMutex - buffers [3]*buffer.Buffer + buffers [DefaultMaxLayerSpatial + 1]*buffer.Buffer upTrackMu sync.RWMutex - upTracks [3]*webrtc.TrackRemote + upTracks [DefaultMaxLayerSpatial + 1]*webrtc.TrackRemote downTrackMu sync.RWMutex downTracks []TrackSender @@ -117,12 +121,13 @@ func WithLoadBalanceThreshold(downTracks int) ReceiverOpts { // NewWebRTCReceiver creates a new webrtc track receivers func NewWebRTCReceiver(receiver *webrtc.RTPReceiver, track *webrtc.TrackRemote, pid string, opts ...ReceiverOpts) Receiver { w := &WebRTCReceiver{ - peerID: pid, - receiver: receiver, - trackID: track.ID(), - streamID: track.StreamID(), - codec: track.Codec(), - kind: track.Kind(), + peerID: pid, + receiver: receiver, + trackID: track.ID(), + streamID: track.StreamID(), + codec: track.Codec(), + kind: track.Kind(), + // LK-TODO: this should be based on VideoLayers protocol message rather than RID based isSimulcast: len(track.RID()) > 0, pliThrottle: 500e6, downTracks: make([]TrackSender, 0), @@ -170,6 +175,10 @@ func (w *WebRTCReceiver) Kind() webrtc.RTPCodecType { return w.kind } +func (w *WebRTCReceiver) IsSimulcast() bool { + return w.isSimulcast +} + func (w *WebRTCReceiver) AddUpTrack(track *webrtc.TrackRemote, buff *buffer.Buffer) { if w.closed.get() { return @@ -240,7 +249,6 @@ func (w *WebRTCReceiver) AddDownTrack(track TrackSender) { return } - track.SetTrackType(w.isSimulcast) if w.Kind() == webrtc.RTPCodecTypeVideo { // notify added downtrack of available layers w.upTrackMu.RLock() @@ -331,14 +339,14 @@ func (w *WebRTCReceiver) removeAvailableLayer(layer uint16) { w.downtrackLayerChange(newLayers) } -func (w *WebRTCReceiver) GetBitrateTemporalCumulative() [3][4]int64 { +func (w *WebRTCReceiver) GetBitrateTemporalCumulative() Bitrates { // LK-TODO: For SVC tracks, need to accumulate across spatial layers also - var br [3][4]int64 + var br Bitrates w.bufferMu.RLock() defer w.bufferMu.RUnlock() for i, buff := range w.buffers { if buff != nil { - tls := make([]int64, 4) + tls := make([]int64, DefaultMaxLayerTemporal+1) if w.hasSpatialLayer(int32(i)) { tls = buff.BitrateTemporalCumulative() } diff --git a/pkg/sfu/streamallocator.go b/pkg/sfu/streamallocator.go index 22ff623c8..a11bad8b7 100644 --- a/pkg/sfu/streamallocator.go +++ b/pkg/sfu/streamallocator.go @@ -1,100 +1,3 @@ -// -// Design of StreamAllocator -// -// Each participant uses one peer connection for all downstream -// traffic. It is possible that the downstream peer connection -// gets congested. In such an event, the SFU (sender on that -// peer connection) should take measures to mitigate the -// media loss and latency that would result from such a congestion. -// -// This module is supposed to aggregate down stream tracks and -// drive bandwidth allocation with the goals of -// - Try and send highest quality media -// - React as quickly as possible to mitigate congestion -// -// Setup: -// ------ -// The following should be done to set up a stream allocator -// - There will be one of these per subscriber peer connection. -// Created in livekit-sever/transport.go for subscriber type -// peer connections. -// - In `AddSubscribedTrack` of livekit-server/participant.go, the created -// downTrack is added to the stream allocator. -// - In `RemoveSubscribedTrack` of livekit-server/participant.go, -// the downTrack is removed from the stream allocator. -// - Both video and audio tracks are added to this module. Although the -// stream allocator does not act on audio track forwarding, audio track -// information like loss rate may be used to adjust available bandwidth. -// -// Callbacks: -// ---------- -// StreamAllocator registers the following callbacks on all registered down tracks -// - OnREMB: called when down track receives RTCP REMB. Note that REMB is a -// peer connection level aggregate metric. But, it contains all the SSRCs -// used in the calculation of that REMB. So, there could be multiple -// callbacks per RTCP REMB received (one each from down track pertaining -// to the contained SSRCs) with the same estimated channel capacity. -// - AddReceiverReportListener: called when down track received RTCP RR (Receiver Report). -// - OnAvailableLayersChanged: called when the feeding track changes its layers. -// This could happen due to publisher throttling layers due to upstream congestion -// in its path. -// - OnSubscriptionChanged: called when a down track settings are changed resulting -// from client side requests (muting/unmuting) -// - OnSubscribedLayersChanged: called when a down track settings are changed resulting -// from client side requests (limiting maximum layer). -// - OnPacketSent: called when a media packet is forwarded by the down track. As -// this happens once per forwarded packet, processing in this callback should be -// kept to a minimum. -// -// The following may be needed depending on the StreamAllocator algorithm -// - OnBitrateUpdate: called periodically to update the bit rate at which a down track -// is forwarding. This can be used to measure any overshoot and adjust allocations -// accordingly. This may have granular information like primary bitrate, retransmitted -// bitrate and padding bitrate. -// -// State machine: -// -------------- -// The most critical component. It should monitor current state of channel and -// take actions to provide the best user experience by striving to achieve the -// goals outlined earlier -// -// States: -// ------ -// - StateStable: When all streams are forwarded at their optimal requested layers. -// -// Before the first estimate is committed, estimated channel capacity -// is initialized to some arbitrarily high value to start streaming -// immediately. Serves two purposes -// 1. Gives the bandwidth estimation algorithms data -// 2. Start streaming as soon as a user joins. Imagine -// a user joining a room with 10 participants already -// in it. That user should start receiving streams -// from everybody as soon as possible. -// -// In this state, it is also possible to probe for extra capacity -// to be prepared for cases like new participant joining and streaming OR -// an existing participant starting a new stream like enabling camera or -// screen share. -// - StateDeficient: When at least one stream is not able to forward optimal requested layers. -// -// Signals: -// ------- -// Each state should take action based on these signals and advance the state machine based -// on the result of the action. -// - SignalAddTrack: A new track has been added. -// - SignalRemoveTrack: An existing track has been removed. -// - SignalEstimate: A new channel capacity estimate has been received. -// Note that when channel gets congested, it is possible to -// get several of these in a very short time window. -// - SignalReceiverReport: An RTCP Receiver Report received from some down track. -// - SignalAvailableLayersChange: Available layers of publisher changed. -// - SignalSubscriptionChange: Subscription changed (mute/unmute) -// - SignalSubscribedLayersChange: Subscribed layers changed (requested layers changed). -// - SignalPeriodicPing: Periodic ping. -// - SignalSendProbe: Request from Prober to send padding probes. -// -// There are several interesting challenges which are documented in relevant code below. -// package sfu import ( @@ -211,9 +114,10 @@ type StreamAllocator struct { lastGratuitousProbeTime time.Time - audioTracks map[string]*Track - videoTracks map[string]*Track - videoTracksSorted TrackSorter + audioTracks map[string]*Track + videoTracks map[string]*Track + exemptVideoTracksSorted TrackSorter + managedVideoTracksSorted TrackSorter prober *Prober @@ -266,10 +170,11 @@ func (s *StreamAllocator) OnStreamStateChange(f func(update *StreamStateUpdate) s.onStreamStateChange = f } -func (s *StreamAllocator) AddTrack(downTrack *DownTrack) { +func (s *StreamAllocator) AddTrack(downTrack *DownTrack, isManaged bool) { s.postEvent(Event{ Signal: SignalAddTrack, DownTrack: downTrack, + Data: isManaged, }) if downTrack.Kind() == webrtc.RTPCodecTypeVideo { @@ -333,17 +238,11 @@ func (s *StreamAllocator) onSubscriptionChanged(downTrack *DownTrack) { } // called when subscribed layers changes (limiting max layers) -func (s *StreamAllocator) onSubscribedLayersChanged(downTrack *DownTrack, layers VideoLayers, layerPref LayerPreference) { +func (s *StreamAllocator) onSubscribedLayersChanged(downTrack *DownTrack, layers VideoLayers) { s.postEvent(Event{ Signal: SignalSubscribedLayersChange, DownTrack: downTrack, - Data: struct { - layers VideoLayers - layerPref LayerPreference - }{ - layers: layers, - layerPref: layerPref, - }, + Data: layers, }) } @@ -425,15 +324,22 @@ func (s *StreamAllocator) handleEvent(event *Event) { } func (s *StreamAllocator) handleSignalAddTrack(event *Event) { - track := newTrack(event.DownTrack) + isManaged, _ := event.Data.(bool) + track := newTrack(event.DownTrack, isManaged) + switch event.DownTrack.Kind() { case webrtc.RTPCodecTypeAudio: s.audioTracks[event.DownTrack.ID()] = track case webrtc.RTPCodecTypeVideo: s.videoTracks[event.DownTrack.ID()] = track - s.videoTracksSorted = append(s.videoTracksSorted, track) - sort.Sort(s.videoTracksSorted) + if isManaged { + s.managedVideoTracksSorted = append(s.managedVideoTracksSorted, track) + sort.Sort(s.managedVideoTracksSorted) + } else { + s.exemptVideoTracksSorted = append(s.exemptVideoTracksSorted, track) + sort.Sort(s.exemptVideoTracksSorted) + } s.allocateTrack(track) } @@ -455,23 +361,35 @@ func (s *StreamAllocator) handleSignalRemoveTrack(event *Event) { delete(s.videoTracks, event.DownTrack.ID()) - n := len(s.videoTracksSorted) - for idx, videoTrack := range s.videoTracksSorted { - if videoTrack.DownTrack() == event.DownTrack { - s.videoTracksSorted[idx] = s.videoTracksSorted[n-1] - s.videoTracksSorted = s.videoTracksSorted[:n-1] - break + if track.IsManaged() { + n := len(s.managedVideoTracksSorted) + for idx, videoTrack := range s.managedVideoTracksSorted { + if videoTrack.DownTrack() == event.DownTrack { + s.managedVideoTracksSorted[idx] = s.managedVideoTracksSorted[n-1] + s.managedVideoTracksSorted = s.managedVideoTracksSorted[:n-1] + break + } } + sort.Sort(s.managedVideoTracksSorted) + } else { + n := len(s.exemptVideoTracksSorted) + for idx, videoTrack := range s.exemptVideoTracksSorted { + if videoTrack.DownTrack() == event.DownTrack { + s.exemptVideoTracksSorted[idx] = s.exemptVideoTracksSorted[n-1] + s.exemptVideoTracksSorted = s.exemptVideoTracksSorted[:n-1] + break + } + } + sort.Sort(s.exemptVideoTracksSorted) } - sort.Sort(s.videoTracksSorted) - // re-initialize estimate if all tracks are removed, let it get a fresh start - if len(s.videoTracksSorted) == 0 { + // re-initialize estimate if all managed tracks are removed, let it get a fresh start + if len(s.managedVideoTracksSorted) == 0 { s.initializeEstimate() return } - s.unallocateTrack(track) + // LK-TODO: use any saved bandwidth to re-distribute } } @@ -498,7 +416,7 @@ func (s *StreamAllocator) handleSignalEstimate(event *Event) { // LK-TODO-END // if there are no video tracks, ignore any straggler REMB - if len(s.videoTracksSorted) == 0 { + if len(s.managedVideoTracksSorted) == 0 { return } @@ -600,12 +518,13 @@ func (s *StreamAllocator) handleSignalSubscribedLayersChange(event *Event) { return } - data := event.Data.(struct { - layers VideoLayers - layerPref LayerPreference - }) - track.UpdatePriority(data.layers, data.layerPref) - sort.Sort(s.videoTracksSorted) + layers := event.Data.(VideoLayers) + track.UpdateMaxLayers(layers) + if track.IsManaged() { + sort.Sort(s.managedVideoTracksSorted) + } else { + sort.Sort(s.exemptVideoTracksSorted) + } s.allocateTrack(track) } @@ -653,7 +572,7 @@ func (s *StreamAllocator) setState(state State) { } func (s *StreamAllocator) adjustState() { - for _, videoTrack := range s.videoTracksSorted { + for _, videoTrack := range s.managedVideoTracksSorted { if videoTrack.IsDeficient() { s.setState(StateDeficient) return @@ -705,7 +624,7 @@ func (s *StreamAllocator) maybeCommitEstimate() (isDecreasing bool) { func (s *StreamAllocator) allocateTrack(track *Track) { // if not deficient, free pass allocate track - if s.state == StateStable { + if s.state == StateStable || !track.IsManaged() { update := NewStreamStateUpdate() allocation := track.Allocate(ChannelCapacityInfinity) update.HandleStreamingChange(allocation.change, track) @@ -713,153 +632,78 @@ func (s *StreamAllocator) allocateTrack(track *Track) { return } - // slice into higher priority tracks and lower priority tracks - var hpTracks []*Track - var lpTracks []*Track - for idx, t := range s.videoTracksSorted { - if t == track { - hpTracks = s.videoTracksSorted[:idx] - lpTracks = s.videoTracksSorted[idx+1:] - break - } - } - - // check how much can be stolen from lower priority tracks - lpExpectedBps := int64(0) - for _, t := range lpTracks { - lpExpectedBps += t.BandwidthRequested() - } - - // - // Note that there might be no lower priority tracks and nothing to steal. - // But, a TryAllocate is done irrespective of any stolen bits as the - // track may be downgrading due to mute or reduction in subscribed layers - // and actually giving back some bits. - // - update := NewStreamStateUpdate() - - allocation := track.TryAllocate(lpExpectedBps) - - update.HandleStreamingChange(allocation.change, track) - - delta := lpExpectedBps - allocation.bandwidthDelta - if delta > 0 { - // gotten some bits back, check if any deficient higher priority track can make use of it - delta = s.tryAllocateTracks(hpTracks, delta, update) - } - - // allocate all lower priority tracks with left over capacity - if delta < 0 { - // stolen too much - delta = 0 - } - - for _, t := range lpTracks { - allocation := t.Allocate(delta) - update.HandleStreamingChange(allocation.change, t) - - delta -= allocation.bandwidthRequested - if delta < 0 { - delta = 0 - } - } - - s.maybeSendUpdate(update) - - s.adjustState() -} - -func (s *StreamAllocator) unallocateTrack(track *Track) { - if s.state == StateStable { - return - } - - update := NewStreamStateUpdate() - - unallocatedBps := track.BandwidthRequested() - if unallocatedBps > 0 { - s.tryAllocateTracks(s.videoTracksSorted, unallocatedBps, update) - } - - s.maybeSendUpdate(update) - - s.adjustState() -} - -func (s *StreamAllocator) tryAllocateTracks(tracks []*Track, additionalBps int64, update *StreamStateUpdate) int64 { - for _, t := range tracks { - if !t.IsDeficient() { - continue - } - - allocation := t.TryAllocate(additionalBps) - update.HandleStreamingChange(allocation.change, t) - - additionalBps -= allocation.bandwidthDelta - if additionalBps <= 0 { - // used up all the extra bits - break - } - } - - return additionalBps + // LK-TODO-START + // Two possible scenarios + // - Down allocate as necessary, take saved bits and re-distribute + // - If up shifting, try to steal from tracks which are closest to the desired + // LK-TODO-END } func (s *StreamAllocator) allocateAllTracks() { s.resetBoost() // - // LK-TODO-START - // Calculate the aggregate loss. This may or may not - // be necessary depending on the algorithm we choose. In this - // pass, we could also calculate audio & video track loss - // separately and use different rules. + // Goals: + // 1. Stream as many tracks as possible, i. e. no pauses. + // 2. Try to give fair allocation to all track. // - // The loss calculation should be for the window between last - // allocation and now. The `lastPackets*` field in - // `Track` structure is used to cache the packet stats - // at the last allocation. Potentially need to think about - // giving higher weight to recent losses. So, might have - // to update the `lastPackets*` periodically even when - // there is no allocation for a long time to ensure loss calculation - // remains fresh. - // LK-TODO-END + // Start with the lowest layers and give each track a chance at that layer and keep going up. + // As long as there is enough bandwidth for tracks to stream at the lowest layers, the first goal is achieved. // - + // Tracks that have higher subscribed layers can use any additional available bandwidth. This tried to achieve the second goal. // - // Ask down tracks adjust their forwarded layers. + // If there is not enough bandwidth even for the lowest layers, tracks at lower priorities will be paused. // update := NewStreamStateUpdate() availableChannelCapacity := s.committedChannelCapacity - for _, track := range s.videoTracksSorted { - // - // `video` tracks could do one of the following - // - no change, i. e. currently forwarding optimal available - // layer and there is enough bandwidth for that. - // - adjust layers up or down - // - pause if there is not enough capacity for any layer - // - allocation := track.Allocate(availableChannelCapacity) + // + // This pass is just to find out if there is any left over channel capacity. + // Infinite channel capacity is given so that exempt tracks do not stall + // + for _, track := range s.exemptVideoTracksSorted { + allocation := track.Allocate(ChannelCapacityInfinity) update.HandleStreamingChange(allocation.change, track) + // LK-TODO: optimistic allocation before bitrate is available will return 0. How to account for that? availableChannelCapacity -= allocation.bandwidthRequested - if availableChannelCapacity < 0 || allocation.state == VideoAllocationStateDeficient { - // - // This is walking down tracks in priortized order. - // Once one of those streams do not fit, set - // the availableChannelCapacity to 0 so that no - // other lower priority stream gets forwarded. - // Note that a lower priority stream may have - // a layer which might fit in the left over - // capacity. This is one type of policy - // implementation. There may be other policies - // which might allow lower priority to go through too. - // So, we need some sort of policy framework here - // to decide which streams get priority - // - availableChannelCapacity = 0 + } + + if availableChannelCapacity < 0 { + availableChannelCapacity = 0 + } + if availableChannelCapacity == 0 { + // nothing left for managed tracks, pause them all + for _, track := range s.managedVideoTracksSorted { + allocation := track.Pause() + update.HandleStreamingChange(allocation.change, track) + } + } else { + for _, track := range s.managedVideoTracksSorted { + track.ProvisionalAllocatePrepare() + } + + for spatial := int32(0); spatial <= DefaultMaxLayerSpatial; spatial++ { + for temporal := int32(0); temporal <= DefaultMaxLayerTemporal; temporal++ { + layers := VideoLayers{ + spatial: spatial, + temporal: temporal, + } + + for _, track := range s.managedVideoTracksSorted { + usedChannelCapacity := track.ProvisionalAllocate(availableChannelCapacity, layers) + availableChannelCapacity -= usedChannelCapacity + if availableChannelCapacity < 0 { + availableChannelCapacity = 0 + } + } + } + } + + for _, track := range s.managedVideoTracksSorted { + allocation := track.ProvisionalAllocateCommit() + update.HandleStreamingChange(allocation.change, track) } } @@ -883,7 +727,7 @@ func (s *StreamAllocator) maybeSendUpdate(update *StreamStateUpdate) { } func (s *StreamAllocator) finalizeTracks() { - for _, t := range s.videoTracksSorted { + for _, t := range s.managedVideoTracksSorted { t.FinalizeAllocate() } @@ -942,8 +786,14 @@ func (s *StreamAllocator) maybeProbe() { } func (s *StreamAllocator) maybeBoostLayer() { + var distanceSorted MaxDistanceSorter + for _, track := range s.managedVideoTracksSorted { + distanceSorted = append(distanceSorted, track) + } + sort.Sort(distanceSorted) + // boost first deficient track in priority order - for _, track := range s.videoTracksSorted { + for _, track := range distanceSorted { if !track.IsDeficient() { continue } @@ -978,7 +828,7 @@ func (s *StreamAllocator) resetBoost() { } func (s *StreamAllocator) maybeGratuitousProbe() bool { - if time.Since(s.lastEstimateDecreaseTime) < GratuitousProbeWaitMs || len(s.videoTracksSorted) == 0 { + if time.Since(s.lastEstimateDecreaseTime) < GratuitousProbeWaitMs || len(s.managedVideoTracksSorted) == 0 { return false } @@ -1065,46 +915,24 @@ func (s *StreamStateUpdate) Empty() bool { //------------------------------------------------ -type ForwardingState int - -const ( - ForwardingStateOptimistic ForwardingState = iota - ForwardingStateDryFeed - ForwardingStateDeficient - ForwardingStateOptimal -) - -func (f ForwardingState) String() string { - switch f { - case ForwardingStateOptimistic: - return "OPTIMISTIC" - case ForwardingStateDryFeed: - return "DRY_FEED" - case ForwardingStateDeficient: - return "DEFICIENT" - case ForwardingStateOptimal: - return "OPTIMAL" - default: - return fmt.Sprintf("%d", int(f)) - } -} - type Track struct { downTrack *DownTrack + isManaged bool highestSN uint32 packetsLost uint32 lastHighestSN uint32 lastPacketsLost uint32 - priority int32 + maxLayers VideoLayers } -func newTrack(downTrack *DownTrack) *Track { +func newTrack(downTrack *DownTrack, isManaged bool) *Track { t := &Track{ downTrack: downTrack, + isManaged: isManaged, } - t.UpdatePriority(downTrack.MaxLayers(), downTrack.GetLayerPreference()) + t.UpdateMaxLayers(downTrack.MaxLayers()) return t } @@ -1113,6 +941,10 @@ func (t *Track) DownTrack() *DownTrack { return t.downTrack } +func (t *Track) IsManaged() bool { + return t.isManaged +} + func (t *Track) ID() string { return t.downTrack.ID() } @@ -1136,17 +968,8 @@ func (t *Track) UpdatePacketStats(rr *rtcp.ReceiverReport) { } } -func (t *Track) UpdatePriority(layers VideoLayers, pref LayerPreference) { - switch pref { - case LayerPreferenceSpatial: - t.priority = layers.spatial*10 + layers.temporal - case LayerPreferenceTemporal: - t.priority = layers.temporal*10 + layers.spatial - } -} - -func (t *Track) Priority() int32 { - return t.priority +func (t *Track) UpdateMaxLayers(layers VideoLayers) { + t.maxLayers = layers } func (t *Track) GetPacketStats() (uint32, uint32) { @@ -1161,41 +984,44 @@ func (t *Track) Allocate(availableChannelCapacity int64) VideoAllocation { return t.downTrack.Allocate(availableChannelCapacity) } -func (t *Track) TryAllocate(additionalChannelCapacity int64) VideoAllocation { - return t.downTrack.TryAllocate(additionalChannelCapacity) +func (t *Track) ProvisionalAllocatePrepare() { + t.downTrack.ProvisionalAllocatePrepare() } -func (t *Track) FinalizeAllocate() { - t.downTrack.FinalizeAllocate() +func (t *Track) ProvisionalAllocate(availableChannelCapacity int64, layers VideoLayers) int64 { + return t.downTrack.ProvisionalAllocate(availableChannelCapacity, layers) +} + +func (t *Track) ProvisionalAllocateCommit() VideoAllocation { + return t.downTrack.ProvisionalAllocateCommit() } func (t *Track) AllocateNextHigher() (VideoAllocation, bool) { return t.downTrack.AllocateNextHigher() } +func (t *Track) FinalizeAllocate() { + t.downTrack.FinalizeAllocate() +} + +func (t *Track) Pause() VideoAllocation { + return t.downTrack.Pause() +} + func (t *Track) IsDeficient() bool { - return t.downTrack.LastAllocation().state == VideoAllocationStateDeficient + return t.downTrack.IsDeficient() } func (t *Track) BandwidthRequested() int64 { - return t.downTrack.LastAllocation().bandwidthRequested + return t.downTrack.BandwidthRequested() +} + +func (t *Track) DistanceToDesired() int32 { + return t.downTrack.DistanceToDesired() } //------------------------------------------------ -// LK-TODO-START -// Typically, in a system like this, there are track priorities. -// It is either implemented as policy -// Examples: -// 1. active speaker gets hi-res, all else lo-res -// 2. screen share streams get hi-res, all else lo-res -// OR -// It is left up to the clients to subscribe explicitly to the quality they want. -// -// This sorter is prioritizing tracks by max layer subscribed and layer preference. -// But, with simple tracks, there is only one layer. But, it is possible they should -// be higher priority, for e.g. screen share track. -// LK-TODO-END type TrackSorter []*Track func (t TrackSorter) Len() int { @@ -1207,7 +1033,27 @@ func (t TrackSorter) Swap(i, j int) { } func (t TrackSorter) Less(i, j int) bool { - return t[i].priority > t[j].priority + if t[i].maxLayers.spatial != t[j].maxLayers.spatial { + return t[i].maxLayers.spatial > t[j].maxLayers.spatial + } + + return t[i].maxLayers.temporal > t[j].maxLayers.temporal +} + +//------------------------------------------------ + +type MaxDistanceSorter []*Track + +func (m MaxDistanceSorter) Len() int { + return len(m) +} + +func (m MaxDistanceSorter) Swap(i, j int) { + m[i], m[j] = m[j], m[i] +} + +func (m MaxDistanceSorter) Less(i, j int) bool { + return m[i].DistanceToDesired() > m[j].DistanceToDesired() } //------------------------------------------------ diff --git a/pkg/telemetry/analytics.go b/pkg/telemetry/analytics.go index 373fffce9..f853a8296 100644 --- a/pkg/telemetry/analytics.go +++ b/pkg/telemetry/analytics.go @@ -3,8 +3,8 @@ package telemetry import ( "context" - "github.com/livekit/protocol/logger" livekit "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/logger" "github.com/livekit/livekit-server/pkg/config" "github.com/livekit/livekit-server/pkg/routing" diff --git a/test/integration_helpers.go b/test/integration_helpers.go index 032344193..c221f705d 100644 --- a/test/integration_helpers.go +++ b/test/integration_helpers.go @@ -10,8 +10,8 @@ import ( "github.com/go-redis/redis/v8" "github.com/livekit/protocol/auth" - "github.com/livekit/protocol/logger" livekit "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/logger" "github.com/livekit/protocol/utils" "github.com/twitchtv/twirp" diff --git a/test/scenarios.go b/test/scenarios.go index 5da1f8bdd..d36325ccd 100644 --- a/test/scenarios.go +++ b/test/scenarios.go @@ -4,8 +4,8 @@ import ( "testing" "time" - "github.com/livekit/protocol/logger" livekit "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/logger" "github.com/livekit/protocol/utils" "github.com/stretchr/testify/require" diff --git a/test/webhook_test.go b/test/webhook_test.go index a20500b6f..57b8fdca6 100644 --- a/test/webhook_test.go +++ b/test/webhook_test.go @@ -9,8 +9,8 @@ import ( "testing" "github.com/livekit/protocol/auth" - "github.com/livekit/protocol/logger" livekit "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/logger" "github.com/livekit/protocol/utils" "github.com/livekit/protocol/webhook" "github.com/stretchr/testify/require"