diff --git a/pkg/config/config.go b/pkg/config/config.go index 68d3e8a2c..4c613917f 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -207,6 +207,11 @@ type RoomConfig struct { EnableRemoteUnmute bool `yaml:"enable_remote_unmute,omitempty"` PlayoutDelay PlayoutDelayConfig `yaml:"playout_delay,omitempty"` SyncStreams bool `yaml:"sync_streams,omitempty"` + // when enabled, a subscriber acquires its requested (highest) video layer directly instead + // of opportunistically ramping up from a lower layer (avoids a visible low->high quality + // ramp). intended for live streaming where the initial quality matters more than the + // time-to-first-frame. when disabled (default), the original opportunistic behavior is used. + LiveStreamingMode bool `yaml:"live_streaming_mode,omitempty"` CreateRoomEnabled bool `yaml:"create_room_enabled,omitempty"` CreateRoomTimeout time.Duration `yaml:"create_room_timeout,omitempty"` CreateRoomAttempts int `yaml:"create_room_attempts,omitempty"` diff --git a/pkg/rtc/mediatracksubscriptions.go b/pkg/rtc/mediatracksubscriptions.go index 69575fe6b..d93edf469 100644 --- a/pkg/rtc/mediatracksubscriptions.go +++ b/pkg/rtc/mediatracksubscriptions.go @@ -109,6 +109,7 @@ func (t *MediaTrackSubscriptions) AddSubscriber(sub types.LocalParticipant, wr * Subscriber: sub, MediaTrack: t.params.MediaTrack, AdaptiveStream: sub.GetAdaptiveStream(), + LiveStreamingMode: sub.GetLiveStreamingMode(), TelemetryListener: sub.GetTelemetryListener(), WrappedReceiver: wr, IsRelayed: t.params.IsRelayed, diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 2b71174fb..8959a9832 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -189,6 +189,7 @@ type ParticipantParams struct { Migration bool Reconnect bool AdaptiveStream bool + LiveStreamingMode bool AllowTCPFallback bool TCPFallbackRTTThreshold int AllowUDPUnstableFallback bool @@ -500,6 +501,10 @@ func (p *ParticipantImpl) GetAdaptiveStream() bool { return p.params.AdaptiveStream } +func (p *ParticipantImpl) GetLiveStreamingMode() bool { + return p.params.LiveStreamingMode +} + func (p *ParticipantImpl) GetPacer() pacer.Pacer { return p.TransportManager.GetSubscriberPacer() } diff --git a/pkg/rtc/subscribedtrack.go b/pkg/rtc/subscribedtrack.go index 3a8a06148..a6dc38ad8 100644 --- a/pkg/rtc/subscribedtrack.go +++ b/pkg/rtc/subscribedtrack.go @@ -48,6 +48,7 @@ type SubscribedTrackParams struct { Subscriber types.LocalParticipant MediaTrack types.MediaTrack AdaptiveStream bool + LiveStreamingMode bool TelemetryListener types.ParticipantTelemetryListener WrappedReceiver *WrappedReceiver IsRelayed bool @@ -154,6 +155,7 @@ func NewSubscribedTrack(params SubscribedTrackParams) (*SubscribedTrack, error) RTCPWriter: params.Subscriber.WriteSubscriberRTCP, DisableSenderReportPassThrough: params.Subscriber.GetDisableSenderReportPassThrough(), SupportsCodecChange: params.Subscriber.SupportsCodecChange(), + LiveStreamingMode: params.LiveStreamingMode, Listener: s, }) if err != nil { @@ -212,7 +214,11 @@ func (t *SubscribedTrack) Bound(err error) { t.logger.Debugw("enabling subscriber track settings on bind", "settings", logger.Proto(t.settings)) } } else { - if t.params.AdaptiveStream { + if t.params.LiveStreamingMode { + // live streaming favors initial quality over time-to-first-frame: request the + // highest layer up front so the layer selector acquires it directly + t.settings = &livekit.UpdateTrackSettings{Quality: livekit.VideoQuality_HIGH} + } else if t.params.AdaptiveStream { t.settings = &livekit.UpdateTrackSettings{Quality: livekit.VideoQuality_LOW} } else { t.settings = &livekit.UpdateTrackSettings{Quality: livekit.VideoQuality_HIGH} diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index e77393e80..2570af891 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -403,6 +403,7 @@ type LocalParticipant interface { GetReporter() roomobs.ParticipantSessionReporter GetReporterResolver() roomobs.ParticipantReporterResolver GetAdaptiveStream() bool + GetLiveStreamingMode() bool ProtocolVersion() ProtocolVersion SupportsSyncStreamID() bool SupportsTransceiverReuse(mt MediaTrack) bool diff --git a/pkg/rtc/types/typesfakes/fake_local_participant.go b/pkg/rtc/types/typesfakes/fake_local_participant.go index 96686a0d1..689ee6d2a 100644 --- a/pkg/rtc/types/typesfakes/fake_local_participant.go +++ b/pkg/rtc/types/typesfakes/fake_local_participant.go @@ -366,6 +366,16 @@ type FakeLocalParticipant struct { getLastReliableSequenceReturnsOnCall map[int]struct { result1 uint32 } + GetLiveStreamingModeStub func() bool + getLiveStreamingModeMutex sync.RWMutex + getLiveStreamingModeArgsForCall []struct { + } + getLiveStreamingModeReturns struct { + result1 bool + } + getLiveStreamingModeReturnsOnCall map[int]struct { + result1 bool + } GetLoggerStub func() logger.Logger getLoggerMutex sync.RWMutex getLoggerArgsForCall []struct { @@ -3289,6 +3299,59 @@ func (fake *FakeLocalParticipant) GetLastReliableSequenceReturnsOnCall(i int, re }{result1} } +func (fake *FakeLocalParticipant) GetLiveStreamingMode() bool { + fake.getLiveStreamingModeMutex.Lock() + ret, specificReturn := fake.getLiveStreamingModeReturnsOnCall[len(fake.getLiveStreamingModeArgsForCall)] + fake.getLiveStreamingModeArgsForCall = append(fake.getLiveStreamingModeArgsForCall, struct { + }{}) + stub := fake.GetLiveStreamingModeStub + fakeReturns := fake.getLiveStreamingModeReturns + fake.recordInvocation("GetLiveStreamingMode", []interface{}{}) + fake.getLiveStreamingModeMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeLocalParticipant) GetLiveStreamingModeCallCount() int { + fake.getLiveStreamingModeMutex.RLock() + defer fake.getLiveStreamingModeMutex.RUnlock() + return len(fake.getLiveStreamingModeArgsForCall) +} + +func (fake *FakeLocalParticipant) GetLiveStreamingModeCalls(stub func() bool) { + fake.getLiveStreamingModeMutex.Lock() + defer fake.getLiveStreamingModeMutex.Unlock() + fake.GetLiveStreamingModeStub = stub +} + +func (fake *FakeLocalParticipant) GetLiveStreamingModeReturns(result1 bool) { + fake.getLiveStreamingModeMutex.Lock() + defer fake.getLiveStreamingModeMutex.Unlock() + fake.GetLiveStreamingModeStub = nil + fake.getLiveStreamingModeReturns = struct { + result1 bool + }{result1} +} + +func (fake *FakeLocalParticipant) GetLiveStreamingModeReturnsOnCall(i int, result1 bool) { + fake.getLiveStreamingModeMutex.Lock() + defer fake.getLiveStreamingModeMutex.Unlock() + fake.GetLiveStreamingModeStub = nil + if fake.getLiveStreamingModeReturnsOnCall == nil { + fake.getLiveStreamingModeReturnsOnCall = make(map[int]struct { + result1 bool + }) + } + fake.getLiveStreamingModeReturnsOnCall[i] = struct { + result1 bool + }{result1} +} + func (fake *FakeLocalParticipant) GetLogger() logger.Logger { fake.getLoggerMutex.Lock() ret, specificReturn := fake.getLoggerReturnsOnCall[len(fake.getLoggerArgsForCall)] diff --git a/pkg/service/roommanager.go b/pkg/service/roommanager.go index 9419e282e..627c8f25f 100644 --- a/pkg/service/roommanager.go +++ b/pkg/service/roommanager.go @@ -514,6 +514,7 @@ func (r *RoomManager) StartSession( SubscriptionLimitVideo: r.config.Limit.SubscriptionLimitVideo, PlayoutDelay: roomInternal.GetPlayoutDelay(), SyncStreams: roomInternal.GetSyncStreams(), + LiveStreamingMode: r.config.Room.LiveStreamingMode, ForwardStats: r.forwardStats, MetricConfig: r.config.Metric, UseOneShotSignallingMode: useOneShotSignallingMode, diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index 38cf31569..0dc7a3961 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -312,6 +312,7 @@ type DownTrackParams struct { DisableSenderReportPassThrough bool SupportsCodecChange bool StripPacketTrailer bool + LiveStreamingMode bool Listener DownTrackListener } @@ -463,6 +464,7 @@ func NewDownTrack(params DownTrackParams) (*DownTrack, error) { d.params.Logger, false, // skipReferenceTS false, // disableOpportunisticAllocation + d.params.LiveStreamingMode, d.rtpStats, ) @@ -1026,6 +1028,15 @@ func (d *DownTrack) keyFrameRequester() { d.Receiver().SendPLI(layer, false) d.rtpStats.UpdateLayerLockPliAndTime(1) } + + // if the initial-acquisition grace expired without latching the requested layer, force a + // re-allocation so the target falls back to the highest layer actually seen (rather than + // stalling while waiting for a requested layer that never showed up) + if d.forwarder.MaybeExpireAcquireGrace() { + if sal := d.getStreamAllocatorListener(); sal != nil { + sal.OnAvailableLayersChanged(d) + } + } } } diff --git a/pkg/sfu/forwarder.go b/pkg/sfu/forwarder.go index 4fac35962..3d6a7a818 100644 --- a/pkg/sfu/forwarder.go +++ b/pkg/sfu/forwarder.go @@ -52,6 +52,15 @@ const ( ResumeBehindHighThresholdSeconds = float64(2.0) // 2 seconds LayerSwitchBehindThresholdSeconds = float64(0.05) // 50ms SwitchAheadThresholdSeconds = float64(0.025) // 25ms + + // While a subscriber is acquiring its first layer and the requested (max) layer has not been + // seen on the wire yet, aim straight for the requested layer for this long instead of latching + // onto a lower layer that is detected first. Avoids a visible low -> high quality ramp, + // notably when the subscriber joined before the publisher started (so layers are detected, and + // `maxSeen` climbs, gradually). If the requested layer does not show up within this window the + // grace expires and forwarding falls back to the highest layer actually seen. + // See Forwarder.opportunisticAlloc / withinAcquireGraceLocked / MaybeExpireAcquireGrace. + initialLayerAcquisitionGrace = time.Second ) var ( @@ -222,6 +231,7 @@ type Forwarder struct { logger logger.Logger skipReferenceTS bool disableOpportunisticAllocation bool + liveStreamingMode bool rtpStats *rtpstats.RTPStatsSender muted bool @@ -230,6 +240,7 @@ type Forwarder struct { started bool preStartTime time.Time + acquireDeadline int64 // mono nanos; initial-acquisition grace deadline, 0 = inactive extFirstTS uint64 lastSSRC uint32 lastReferencePayloadType int8 @@ -256,6 +267,7 @@ func NewForwarder( logger logger.Logger, skipReferenceTS bool, disableOpportunisticAllocation bool, + liveStreamingMode bool, rtpStats *rtpstats.RTPStatsSender, ) *Forwarder { f := &Forwarder{ @@ -264,6 +276,7 @@ func NewForwarder( logger: logger, skipReferenceTS: skipReferenceTS, disableOpportunisticAllocation: disableOpportunisticAllocation, + liveStreamingMode: liveStreamingMode, rtpStats: rtpStats, referenceLayerSpatial: buffer.InvalidLayerSpatial, lastAllocation: VideoAllocationDefault, @@ -272,6 +285,7 @@ func NewForwarder( vls: videolayerselector.NewNull(logger), codecMunger: codecmunger.NewNull(logger), } + f.vls.SetLiveStreamingMode(liveStreamingMode) if f.kind == webrtc.RTPCodecTypeVideo { f.vls.SetMaxTemporal(buffer.DefaultMaxLayerTemporal) @@ -289,10 +303,36 @@ func (f *Forwarder) SetMaxPublishedLayer(maxPublishedLayer int32) bool { } f.vls.SetMaxSeenSpatial(maxPublishedLayer) + if f.liveStreamingMode && !f.vls.GetCurrent().IsValid() { + // A (higher) layer just became available while nothing is being forwarded yet. + // (Re)start the initial-acquisition grace so the target aims for the requested layer + // instead of ramping up gradually as more layers are detected. See opportunisticAlloc. + f.acquireDeadline = mono.UnixNano() + initialLayerAcquisitionGrace.Nanoseconds() + } f.logger.Debugw("setting max published layer", "layer", maxPublishedLayer) return true } +// withinAcquireGraceLocked reports whether the initial-acquisition grace window is still open. +func (f *Forwarder) withinAcquireGraceLocked() bool { + return f.acquireDeadline != 0 && mono.UnixNano() < f.acquireDeadline +} + +// MaybeExpireAcquireGrace returns true once when the initial-acquisition grace has expired while +// the forwarder is still not streaming any layer. The caller should trigger a re-allocation so the +// target falls back from the requested layer to the highest layer actually seen, avoiding a stall +// if the requested layer never shows up. The deadline is cleared so it fires at most once. +func (f *Forwarder) MaybeExpireAcquireGrace() bool { + f.lock.Lock() + defer f.lock.Unlock() + + if f.acquireDeadline == 0 || mono.UnixNano() < f.acquireDeadline { + return false + } + f.acquireDeadline = 0 + return !f.vls.GetCurrent().IsValid() +} + func (f *Forwarder) SetMaxTemporalLayerSeen(maxTemporalLayerSeen int32) bool { f.lock.Lock() defer f.lock.Unlock() @@ -409,6 +449,9 @@ func (f *Forwarder) DetermineCodec(codec webrtc.RTPCodecCapability, extensions [ } } } + + // the selector may have been (re)created above; keep its live-streaming mode in sync + f.vls.SetLiveStreamingMode(f.liveStreamingMode) } func (f *Forwarder) GetState() *livekit.RTPForwarderState { @@ -832,6 +875,17 @@ func (f *Forwarder) AllocateOptimal(availableLayers []int32, brs Bitrates, allow return maxTemporal } + // inAcquireGrace reports that we are acquiring the first layer, the requested layer has not + // been seen on the wire yet, and the grace window is still open. While true, aim straight for + // the requested layer instead of the highest seen so far, so acquisition does not ramp up + // gradually as layers are detected (`maxSeen` climbs). See initialLayerAcquisitionGrace. + inAcquireGrace := func(maxSpatial int32) bool { + return !currentLayer.IsValid() && maxSeenLayer.Spatial < maxSpatial && f.withinAcquireGraceLocked() + } + + // set when opportunisticAlloc aimed the target at the requested layer due to the acquisition + // grace (as opposed to overshoot), so the key frame request can be pointed at it too + acquireGraceApplied := false opportunisticAlloc := func() { // opportunistically latch on to anything maxSpatial := maxLayer.Spatial @@ -839,8 +893,14 @@ func (f *Forwarder) AllocateOptimal(availableLayers []int32, brs Bitrates, allow maxSpatial = maxSeenLayer.Spatial } + targetSpatial := min(maxSeenLayer.Spatial, maxSpatial) + if inAcquireGrace(maxSpatial) { + targetSpatial = maxSpatial + acquireGraceApplied = true + } + alloc.TargetLayer = buffer.VideoLayer{ - Spatial: min(maxSeenLayer.Spatial, maxSpatial), + Spatial: targetSpatial, Temporal: getMaxTemporal(), } } @@ -935,7 +995,11 @@ func (f *Forwarder) AllocateOptimal(availableLayers []int32, brs Bitrates, allow } else { // opportunistically latch on to anything opportunisticAlloc() - if requestLayerSpatial == buffer.InvalidLayerSpatial { + if acquireGraceApplied { + // in the acquisition grace, request a key frame for the requested layer we + // are waiting for (above what has been seen so far) + alloc.RequestLayerSpatial = alloc.TargetLayer.Spatial + } else if requestLayerSpatial == buffer.InvalidLayerSpatial { alloc.RequestLayerSpatial = maxLayerSpatialLimit } else { alloc.RequestLayerSpatial = requestLayerSpatial diff --git a/pkg/sfu/forwarder_test.go b/pkg/sfu/forwarder_test.go index cca1e69be..e45db66b5 100644 --- a/pkg/sfu/forwarder_test.go +++ b/pkg/sfu/forwarder_test.go @@ -37,8 +37,9 @@ func newForwarder(codec webrtc.RTPCodecCapability, kind webrtc.RTPCodecType) *Fo f := NewForwarder( kind, logger.GetLogger(), - true, // skipReferenceTS - true, // disableOpportunisticAllocation + true, // skipReferenceTS + true, // disableOpportunisticAllocation + false, // liveStreamingMode nil, ) f.DetermineCodec(codec, nil, livekit.VideoLayer_MODE_UNUSED) @@ -2145,3 +2146,43 @@ func TestForwarderGetPaddingVP8(t *testing.T) { require.NoError(t, err) require.Equal(t, marshalledVP8, buf) } + +func TestForwarderInitialAcquisitionGrace(t *testing.T) { + f := newForwarder(testutils.TestVP8Codec, webrtc.RTPCodecTypeVideo) + f.liveStreamingMode = true // the acquisition grace only applies in live streaming mode + + // subscriber requested the top spatial layer + f.SetMaxSpatialLayer(buffer.DefaultMaxLayerSpatial) + f.SetMaxTemporalLayer(buffer.DefaultMaxLayerTemporal) + f.SetMaxTemporalLayerSeen(buffer.DefaultMaxLayerTemporal) + + bitrates := Bitrates{ + {2, 3, 0, 0}, + {4, 0, 0, 5}, + {0, 7, 0, 0}, + } + + // subscriber-first: only layer 1 has been seen so far and nothing is being forwarded yet + // (current invalid). this arms the initial-acquisition grace. + require.True(t, f.SetMaxPublishedLayer(1)) + require.True(t, f.withinAcquireGraceLocked()) + + // during the grace, the target aims straight at the requested layer (2) and requests a key + // frame for it, even though only layer 1 has been seen - so acquisition does not ramp up + // gradually as higher layers are detected + alloc := f.AllocateOptimal([]int32{0, 1}, bitrates, false, false) + require.Equal(t, int32(2), alloc.TargetLayer.Spatial) + require.Equal(t, int32(2), alloc.RequestLayerSpatial) + + // force the grace to expire while still not streaming: must signal that a re-allocation is + // needed so the target can fall back + f.acquireDeadline = 1 // a deadline far in the past + require.True(t, f.MaybeExpireAcquireGrace()) + require.False(t, f.MaybeExpireAcquireGrace()) // only fires once + + // after the grace, the target falls back to the highest layer actually seen (1) instead of + // stalling while waiting for a requested layer that never showed up + alloc = f.AllocateOptimal([]int32{0, 1}, bitrates, false, false) + require.Equal(t, int32(1), alloc.TargetLayer.Spatial) + require.Equal(t, int32(1), alloc.RequestLayerSpatial) +} diff --git a/pkg/sfu/videolayerselector/base.go b/pkg/sfu/videolayerselector/base.go index 181e5d3fb..c89cc51b4 100644 --- a/pkg/sfu/videolayerselector/base.go +++ b/pkg/sfu/videolayerselector/base.go @@ -35,6 +35,8 @@ type Base struct { currentLayer buffer.VideoLayer previousLayer buffer.VideoLayer + + liveStreamingMode bool } func NewBase(logger logger.Logger) *Base { @@ -74,6 +76,10 @@ func (b *Base) SetMaxSpatial(layer int32) { b.maxLayer.Spatial = layer } +func (b *Base) SetLiveStreamingMode(enabled bool) { + b.liveStreamingMode = enabled +} + func (b *Base) SetMaxTemporal(layer int32) { b.maxLayer.Temporal = layer } diff --git a/pkg/sfu/videolayerselector/simulcast.go b/pkg/sfu/videolayerselector/simulcast.go index 664b2a706..5c95790ed 100644 --- a/pkg/sfu/videolayerselector/simulcast.go +++ b/pkg/sfu/videolayerselector/simulcast.go @@ -94,14 +94,32 @@ func (s *Simulcast) Select(extPkt *buffer.ExtPacket, layer int32) (result VideoL found := false reason := "" if extPkt.IsKeyFrame { - if layer > s.currentLayer.Spatial && layer <= s.targetLayer.Spatial { - reason = "upgrading layer" - found = true - } + if s.liveStreamingMode && !isActive { + // Live streaming mode, initial acquisition: latch directly onto the target layer + // instead of opportunistically latching onto the first key frame of any lower + // layer that happens to arrive first. This avoids a visible low-quality -> + // high-quality ramp (e.g. briefly decoding layer 0 before settling on a requested + // layer 2) for a subscriber that requested the higher layer. + // + // The target is chosen by the allocator: during the initial-acquisition grace it + // points at the requested layer (so we wait for it); if that layer never shows + // up the grace expires and the allocator drops the target to the highest layer + // actually seen, so we always end up latching onto a layer that is flowing. + if layer == s.targetLayer.Spatial { + reason = "acquiring target layer" + found = true + } + } else { + // default: opportunistically latch on to / step towards the target layer + if layer > s.currentLayer.Spatial && layer <= s.targetLayer.Spatial { + reason = "upgrading layer" + found = true + } - if layer < s.currentLayer.Spatial && layer >= s.targetLayer.Spatial { - reason = "downgrading layer" - found = true + if layer < s.currentLayer.Spatial && layer >= s.targetLayer.Spatial { + reason = "downgrading layer" + found = true + } } if found { diff --git a/pkg/sfu/videolayerselector/simulcast_test.go b/pkg/sfu/videolayerselector/simulcast_test.go new file mode 100644 index 000000000..f96ebad05 --- /dev/null +++ b/pkg/sfu/videolayerselector/simulcast_test.go @@ -0,0 +1,96 @@ +// Copyright 2023 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package videolayerselector + +import ( + "testing" + + "github.com/pion/rtp" + "github.com/stretchr/testify/require" + + "github.com/livekit/livekit-server/pkg/sfu/buffer" + "github.com/livekit/protocol/logger" +) + +func keyFrameOnLayer(spatial, temporal int32) *buffer.ExtPacket { + return &buffer.ExtPacket{ + VideoLayer: buffer.VideoLayer{Spatial: spatial, Temporal: temporal}, + Packet: &rtp.Packet{}, + IsKeyFrame: true, + } +} + +// On initial acquisition the selector must latch directly onto the target layer and ignore +// lower-layer key frames that arrive first, so a subscriber requesting the top layer does not +// briefly decode a lower layer (a visible quality ramp). +func TestSimulcastSelectAcquiresTargetLayerDirectly(t *testing.T) { + s := NewSimulcast(logger.GetLogger()) + s.SetLiveStreamingMode(true) + s.SetMax(buffer.VideoLayer{Spatial: 2, Temporal: 2}) + s.SetMaxSeen(buffer.VideoLayer{Spatial: 2, Temporal: 2}) + s.SetTarget(buffer.VideoLayer{Spatial: 2, Temporal: 2}) + s.SetRequestSpatial(2) + s.SetCurrent(buffer.InvalidLayer) + + // lower-layer key frames arriving first must NOT be latched + require.False(t, s.Select(keyFrameOnLayer(0, 2), 0).IsSelected) + require.False(t, s.GetCurrent().IsValid()) + require.False(t, s.Select(keyFrameOnLayer(1, 2), 1).IsSelected) + require.False(t, s.GetCurrent().IsValid()) + + // the target layer key frame latches directly + require.True(t, s.Select(keyFrameOnLayer(2, 2), 2).IsSelected) + require.Equal(t, int32(2), s.GetCurrent().Spatial) +} + +// Acquisition follows whatever target the allocator set: when the target is lowered (e.g. the +// acquisition grace expired and the allocator fell back to the highest layer seen), the selector +// latches that lower layer directly. This is the fallback path that prevents a stall when the +// originally requested layer never shows up. +func TestSimulcastSelectAcquiresLoweredTarget(t *testing.T) { + s := NewSimulcast(logger.GetLogger()) + s.SetLiveStreamingMode(true) + s.SetMax(buffer.VideoLayer{Spatial: 2, Temporal: 2}) + s.SetMaxSeen(buffer.VideoLayer{Spatial: 1, Temporal: 2}) + // allocator dropped the target to the highest layer actually seen + s.SetTarget(buffer.VideoLayer{Spatial: 1, Temporal: 2}) + s.SetRequestSpatial(1) + s.SetCurrent(buffer.InvalidLayer) + + require.False(t, s.Select(keyFrameOnLayer(0, 2), 0).IsSelected) + require.False(t, s.GetCurrent().IsValid()) + + require.True(t, s.Select(keyFrameOnLayer(1, 2), 1).IsSelected) + require.Equal(t, int32(1), s.GetCurrent().Spatial) +} + +// With live streaming mode disabled (the default), acquisition keeps the original opportunistic +// behavior: it latches onto the first lower-layer key frame that arrives (<= target) rather than +// waiting for the target layer. +func TestSimulcastSelectOpportunisticWhenLiveStreamingModeOff(t *testing.T) { + s := NewSimulcast(logger.GetLogger()) + // liveStreamingMode left at its default (false) + s.SetMax(buffer.VideoLayer{Spatial: 2, Temporal: 2}) + s.SetMaxSeen(buffer.VideoLayer{Spatial: 2, Temporal: 2}) + s.SetTarget(buffer.VideoLayer{Spatial: 2, Temporal: 2}) + s.SetRequestSpatial(2) + s.SetCurrent(buffer.InvalidLayer) + + // a lower-layer key frame is latched immediately (opportunistic), then it ramps up + require.True(t, s.Select(keyFrameOnLayer(0, 2), 0).IsSelected) + require.Equal(t, int32(0), s.GetCurrent().Spatial) + require.True(t, s.Select(keyFrameOnLayer(2, 2), 2).IsSelected) + require.Equal(t, int32(2), s.GetCurrent().Spatial) +} diff --git a/pkg/sfu/videolayerselector/videolayerselector.go b/pkg/sfu/videolayerselector/videolayerselector.go index 883e46aea..ddd450329 100644 --- a/pkg/sfu/videolayerselector/videolayerselector.go +++ b/pkg/sfu/videolayerselector/videolayerselector.go @@ -43,6 +43,8 @@ type VideoLayerSelector interface { SetMaxTemporal(layer int32) GetMax() buffer.VideoLayer + SetLiveStreamingMode(enabled bool) + SetTarget(targetLayer buffer.VideoLayer) GetTarget() buffer.VideoLayer