From 68aebb0106311d3ce71e0731b7c638224dd79364 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Wed, 13 Sep 2023 19:59:24 +0530 Subject: [PATCH] Do not mute when stream is paused. (#2069) --- pkg/sfu/downtrack.go | 46 ++++++++++--- pkg/sfu/forwarder.go | 78 +++++++++++----------- pkg/sfu/forwarder_test.go | 65 ++++++++++++------ pkg/sfu/streamallocator/streamallocator.go | 12 ++++ pkg/sfu/streamallocator/track.go | 4 ++ 5 files changed, 135 insertions(+), 70 deletions(-) diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index fb445f222..0ddb62c19 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -181,6 +181,9 @@ type DownTrackStreamAllocatorListener interface { // check if track should participate in BWE IsBWEEnabled(dt *DownTrack) bool + + // check if subscription mute can be applied + IsSubscribeMutable(dt *DownTrack) bool } type ReceiverReportListener func(dt *DownTrack, report *rtcp.ReceiverReport) @@ -830,7 +833,15 @@ func (d *DownTrack) WritePaddingRTP(bytesToSend int, paddingOnMute bool, forceMa // Mute enables or disables media forwarding - subscriber triggered func (d *DownTrack) Mute(muted bool) { - changed := d.forwarder.Mute(muted) + d.streamAllocatorLock.RLock() + listener := d.streamAllocatorListener + d.streamAllocatorLock.RUnlock() + + isSubscribeMutable := true + if listener != nil { + isSubscribeMutable = listener.IsSubscribeMutable(d) + } + changed := d.forwarder.Mute(muted, isSubscribeMutable) d.handleMute(muted, changed) } @@ -913,9 +924,8 @@ func (d *DownTrack) CloseWithFlush(flush bool) { d.bindLock.Lock() d.params.Logger.Debugw("close down track", "flushBlankFrame", flush) if d.bound.Load() { - if d.forwarder != nil { - d.forwarder.Mute(true) - } + d.forwarder.Mute(true, true) + // write blank frames after disabling so that other frames do not interfere. // Idea here is to send blank key frames to flush the decoder buffer at the remote end. // Otherwise, with transceiver re-use last frame from previous stream is held in the @@ -1159,14 +1169,24 @@ func (d *DownTrack) ProvisionalAllocate(availableChannelCapacity int64, layers b } func (d *DownTrack) ProvisionalAllocateGetCooperativeTransition(allowOvershoot bool) VideoTransition { - transition := d.forwarder.ProvisionalAllocateGetCooperativeTransition(allowOvershoot) - d.params.Logger.Debugw("stream: cooperative transition", "transition", transition) + transition, availableLayers, brs := d.forwarder.ProvisionalAllocateGetCooperativeTransition(allowOvershoot) + d.params.Logger.Debugw( + "stream: cooperative transition", + "transition", transition, + "availableLayers", availableLayers, + "bitrates", brs, + ) return transition } func (d *DownTrack) ProvisionalAllocateGetBestWeightedTransition() VideoTransition { - transition := d.forwarder.ProvisionalAllocateGetBestWeightedTransition() - d.params.Logger.Debugw("stream: best weighted transition", "transition", transition) + transition, availableLayers, brs := d.forwarder.ProvisionalAllocateGetBestWeightedTransition() + d.params.Logger.Debugw( + "stream: best weighted transition", + "transition", transition, + "availableLayers", availableLayers, + "bitrates", brs, + ) return transition } @@ -1186,9 +1206,15 @@ func (d *DownTrack) AllocateNextHigher(availableChannelCapacity int64, allowOver } func (d *DownTrack) GetNextHigherTransition(allowOvershoot bool) (VideoTransition, bool) { - _, brs := d.params.Receiver.GetLayeredBitrate() + availableLayers, brs := d.params.Receiver.GetLayeredBitrate() transition, available := d.forwarder.GetNextHigherTransition(brs, allowOvershoot) - d.params.Logger.Debugw("stream: get next higher layer", "transition", transition, "available", available, "bitrates", brs) + d.params.Logger.Debugw( + "stream: get next higher layer", + "transition", transition, + "available", available, + "availableLayers", availableLayers, + "bitrates", brs, + ) return transition, available } diff --git a/pkg/sfu/forwarder.go b/pkg/sfu/forwarder.go index 6e2805d09..f0c37a3c2 100644 --- a/pkg/sfu/forwarder.go +++ b/pkg/sfu/forwarder.go @@ -121,7 +121,7 @@ type VideoAllocationProvisional struct { pubMuted bool maxSeenLayer buffer.VideoLayer availableLayers []int32 - Bitrates Bitrates + bitrates Bitrates maxLayer buffer.VideoLayer currentLayer buffer.VideoLayer allocatedLayer buffer.VideoLayer @@ -372,7 +372,7 @@ func (f *Forwarder) SeedState(state ForwarderState) { f.refTSOffset = state.RefTSOffset } -func (f *Forwarder) Mute(muted bool) bool { +func (f *Forwarder) Mute(muted bool, isSubscribeMutable bool) bool { f.lock.Lock() defer f.lock.Unlock() @@ -394,7 +394,7 @@ func (f *Forwarder) Mute(muted bool) bool { // The work around here to ignore mute does ignore an intentional mute. // It could result in some bandwidth consumed for stream without visibility in // the case of intentional mute. - if muted && f.isDeficientLocked() && f.lastAllocation.PauseReason == VideoPauseReasonBandwidth { + if muted && !isSubscribeMutable { f.logger.Debugw("ignoring forwarder mute, paused due to congestion") return false } @@ -722,7 +722,7 @@ func (f *Forwarder) AllocateOptimal(availableLayers []int32, brs Bitrates, allow return f.updateAllocation(alloc, "optimal") } -func (f *Forwarder) ProvisionalAllocatePrepare(availableLayers []int32, Bitrates Bitrates) { +func (f *Forwarder) ProvisionalAllocatePrepare(availableLayers []int32, bitrates Bitrates) { f.lock.Lock() defer f.lock.Unlock() @@ -731,7 +731,7 @@ func (f *Forwarder) ProvisionalAllocatePrepare(availableLayers []int32, Bitrates muted: f.muted, pubMuted: f.pubMuted, maxSeenLayer: f.vls.GetMaxSeen(), - Bitrates: Bitrates, + bitrates: bitrates, maxLayer: f.vls.GetMax(), currentLayer: f.vls.GetCurrent(), } @@ -759,14 +759,14 @@ func (f *Forwarder) ProvisionalAllocate(availableChannelCapacity int64, layer bu return false, 0 } - requiredBitrate := f.provisional.Bitrates[layer.Spatial][layer.Temporal] + requiredBitrate := f.provisional.bitrates[layer.Spatial][layer.Temporal] if requiredBitrate == 0 { return false, 0 } alreadyAllocatedBitrate := int64(0) if f.provisional.allocatedLayer.IsValid() { - alreadyAllocatedBitrate = f.provisional.Bitrates[f.provisional.allocatedLayer.Spatial][f.provisional.allocatedLayer.Temporal] + alreadyAllocatedBitrate = f.provisional.bitrates[f.provisional.allocatedLayer.Spatial][f.provisional.allocatedLayer.Temporal] } // a layer under maximum fits, take it @@ -791,7 +791,7 @@ func (f *Forwarder) ProvisionalAllocate(availableChannelCapacity int64, layer bu return false, 0 } -func (f *Forwarder) ProvisionalAllocateGetCooperativeTransition(allowOvershoot bool) VideoTransition { +func (f *Forwarder) ProvisionalAllocateGetCooperativeTransition(allowOvershoot bool) (VideoTransition, []int32, Bitrates) { // // This is called when a track needs a change (could be mute/unmute, subscribed layers changed, published layers changed) // when channel is congested. @@ -820,8 +820,8 @@ func (f *Forwarder) ProvisionalAllocateGetCooperativeTransition(allowOvershoot b return VideoTransition{ From: existingTargetLayer, To: f.provisional.allocatedLayer, - BandwidthDelta: -getBandwidthNeeded(f.provisional.Bitrates, existingTargetLayer, f.lastAllocation.BandwidthRequested), - } + BandwidthDelta: -getBandwidthNeeded(f.provisional.bitrates, existingTargetLayer, f.lastAllocation.BandwidthRequested), + }, f.provisional.availableLayers, f.provisional.bitrates } // check if we should preserve current target @@ -831,9 +831,9 @@ func (f *Forwarder) ProvisionalAllocateGetCooperativeTransition(allowOvershoot b maximalBandwidthRequired := int64(0) for s := f.provisional.maxLayer.Spatial; s >= 0; s-- { for t := f.provisional.maxLayer.Temporal; t >= 0; t-- { - if f.provisional.Bitrates[s][t] != 0 { + if f.provisional.bitrates[s][t] != 0 { maximalLayer = buffer.VideoLayer{Spatial: s, Temporal: t} - maximalBandwidthRequired = f.provisional.Bitrates[s][t] + maximalBandwidthRequired = f.provisional.bitrates[s][t] break } } @@ -844,7 +844,7 @@ func (f *Forwarder) ProvisionalAllocateGetCooperativeTransition(allowOvershoot b } if maximalLayer.IsValid() { - if !existingTargetLayer.GreaterThan(maximalLayer) && f.provisional.Bitrates[existingTargetLayer.Spatial][existingTargetLayer.Temporal] != 0 { + if !existingTargetLayer.GreaterThan(maximalLayer) && f.provisional.bitrates[existingTargetLayer.Spatial][existingTargetLayer.Temporal] != 0 { // currently streaming and maybe wanting an upgrade (existingTargetLayer <= maximalLayer), // just preserve current target in the cooperative scheme of things f.provisional.allocatedLayer = existingTargetLayer @@ -852,7 +852,7 @@ func (f *Forwarder) ProvisionalAllocateGetCooperativeTransition(allowOvershoot b From: existingTargetLayer, To: existingTargetLayer, BandwidthDelta: 0, - } + }, f.provisional.availableLayers, f.provisional.bitrates } if existingTargetLayer.GreaterThan(maximalLayer) { @@ -861,8 +861,8 @@ func (f *Forwarder) ProvisionalAllocateGetCooperativeTransition(allowOvershoot b return VideoTransition{ From: existingTargetLayer, To: maximalLayer, - BandwidthDelta: maximalBandwidthRequired - getBandwidthNeeded(f.provisional.Bitrates, existingTargetLayer, f.lastAllocation.BandwidthRequested), - } + BandwidthDelta: maximalBandwidthRequired - getBandwidthNeeded(f.provisional.bitrates, existingTargetLayer, f.lastAllocation.BandwidthRequested), + }, f.provisional.availableLayers, f.provisional.bitrates } } } @@ -875,9 +875,9 @@ func (f *Forwarder) ProvisionalAllocateGetCooperativeTransition(allowOvershoot b bw := int64(0) for s := minSpatial; s <= maxSpatial; s++ { for t := minTemporal; t <= maxTemporal; t++ { - if f.provisional.Bitrates[s][t] != 0 { + if f.provisional.bitrates[s][t] != 0 { layers = buffer.VideoLayer{Spatial: s, Temporal: t} - bw = f.provisional.Bitrates[s][t] + bw = f.provisional.bitrates[s][t] break } } @@ -914,7 +914,7 @@ func (f *Forwarder) ProvisionalAllocateGetCooperativeTransition(allowOvershoot b if !targetLayer.IsValid() { targetLayer = f.provisional.currentLayer if targetLayer.IsValid() { - bandwidthRequired = f.provisional.Bitrates[targetLayer.Spatial][targetLayer.Temporal] + bandwidthRequired = f.provisional.bitrates[targetLayer.Spatial][targetLayer.Temporal] } } @@ -922,11 +922,11 @@ func (f *Forwarder) ProvisionalAllocateGetCooperativeTransition(allowOvershoot b return VideoTransition{ From: f.vls.GetTarget(), To: targetLayer, - BandwidthDelta: bandwidthRequired - getBandwidthNeeded(f.provisional.Bitrates, existingTargetLayer, f.lastAllocation.BandwidthRequested), - } + BandwidthDelta: bandwidthRequired - getBandwidthNeeded(f.provisional.bitrates, existingTargetLayer, f.lastAllocation.BandwidthRequested), + }, f.provisional.availableLayers, f.provisional.bitrates } -func (f *Forwarder) ProvisionalAllocateGetBestWeightedTransition() VideoTransition { +func (f *Forwarder) ProvisionalAllocateGetBestWeightedTransition() (VideoTransition, []int32, Bitrates) { // // This is called when a track needs a change (could be mute/unmute, subscribed layers changed, published layers changed) // when channel is congested. This is called on tracks other than the one needing the change. When the track @@ -951,14 +951,14 @@ func (f *Forwarder) ProvisionalAllocateGetBestWeightedTransition() VideoTransiti return VideoTransition{ From: targetLayer, To: f.provisional.allocatedLayer, - BandwidthDelta: 0 - getBandwidthNeeded(f.provisional.Bitrates, targetLayer, f.lastAllocation.BandwidthRequested), - } + BandwidthDelta: 0 - getBandwidthNeeded(f.provisional.bitrates, targetLayer, f.lastAllocation.BandwidthRequested), + }, f.provisional.availableLayers, f.provisional.bitrates } maxReachableLayerTemporal := buffer.InvalidLayerTemporal for t := f.provisional.maxLayer.Temporal; t >= 0; t-- { for s := f.provisional.maxLayer.Spatial; s >= 0; s-- { - if f.provisional.Bitrates[s][t] != 0 { + if f.provisional.bitrates[s][t] != 0 { maxReachableLayerTemporal = t break } @@ -976,13 +976,13 @@ func (f *Forwarder) ProvisionalAllocateGetBestWeightedTransition() VideoTransiti return VideoTransition{ From: targetLayer, To: f.provisional.allocatedLayer, - BandwidthDelta: 0 - getBandwidthNeeded(f.provisional.Bitrates, targetLayer, f.lastAllocation.BandwidthRequested), - } + BandwidthDelta: 0 - getBandwidthNeeded(f.provisional.bitrates, targetLayer, f.lastAllocation.BandwidthRequested), + }, f.provisional.availableLayers, f.provisional.bitrates } // starting from minimum to target, find transition which gives the best // transition taking into account bits saved vs cost of such a transition - existingBandwidthNeeded := getBandwidthNeeded(f.provisional.Bitrates, targetLayer, f.lastAllocation.BandwidthRequested) + existingBandwidthNeeded := getBandwidthNeeded(f.provisional.bitrates, targetLayer, f.lastAllocation.BandwidthRequested) bestLayer := buffer.InvalidLayer bestBandwidthDelta := int64(0) bestValue := float32(0) @@ -992,7 +992,7 @@ func (f *Forwarder) ProvisionalAllocateGetBestWeightedTransition() VideoTransiti break } - bandwidthDelta := int64(math.Max(float64(0), float64(existingBandwidthNeeded-f.provisional.Bitrates[s][t]))) + bandwidthDelta := int64(math.Max(float64(0), float64(existingBandwidthNeeded-f.provisional.bitrates[s][t]))) transitionCost := int32(0) // SVC-TODO: SVC will need a different cost transition @@ -1019,7 +1019,7 @@ func (f *Forwarder) ProvisionalAllocateGetBestWeightedTransition() VideoTransiti From: targetLayer, To: bestLayer, BandwidthDelta: -bestBandwidthDelta, - } + }, f.provisional.availableLayers, f.provisional.bitrates } func (f *Forwarder) ProvisionalAllocateCommit() VideoAllocation { @@ -1030,13 +1030,13 @@ func (f *Forwarder) ProvisionalAllocateCommit() VideoAllocation { f.provisional.muted, f.provisional.pubMuted, f.provisional.maxSeenLayer.Spatial, - f.provisional.Bitrates, + f.provisional.bitrates, f.provisional.maxLayer, ) alloc := VideoAllocation{ BandwidthRequested: 0, - BandwidthDelta: 0 - getBandwidthNeeded(f.provisional.Bitrates, f.vls.GetTarget(), f.lastAllocation.BandwidthRequested), - Bitrates: f.provisional.Bitrates, + BandwidthDelta: 0 - getBandwidthNeeded(f.provisional.bitrates, f.vls.GetTarget(), f.lastAllocation.BandwidthRequested), + Bitrates: f.provisional.bitrates, BandwidthNeeded: optimalBandwidthNeeded, TargetLayer: f.provisional.allocatedLayer, RequestLayerSpatial: f.provisional.allocatedLayer.Spatial, @@ -1046,7 +1046,7 @@ func (f *Forwarder) ProvisionalAllocateCommit() VideoAllocation { f.provisional.pubMuted, f.provisional.maxSeenLayer, f.provisional.availableLayers, - f.provisional.Bitrates, + f.provisional.bitrates, f.provisional.allocatedLayer, f.provisional.maxLayer, ), @@ -1062,8 +1062,8 @@ func (f *Forwarder) ProvisionalAllocateCommit() VideoAllocation { case optimalBandwidthNeeded == 0: if f.provisional.allocatedLayer.IsValid() { // overshoot - alloc.BandwidthRequested = f.provisional.Bitrates[f.provisional.allocatedLayer.Spatial][f.provisional.allocatedLayer.Temporal] - alloc.BandwidthDelta = alloc.BandwidthRequested - getBandwidthNeeded(f.provisional.Bitrates, f.vls.GetTarget(), f.lastAllocation.BandwidthRequested) + alloc.BandwidthRequested = f.provisional.bitrates[f.provisional.allocatedLayer.Spatial][f.provisional.allocatedLayer.Temporal] + alloc.BandwidthDelta = alloc.BandwidthRequested - getBandwidthNeeded(f.provisional.bitrates, f.vls.GetTarget(), f.lastAllocation.BandwidthRequested) } else { alloc.PauseReason = VideoPauseReasonFeedDry @@ -1077,16 +1077,16 @@ func (f *Forwarder) ProvisionalAllocateCommit() VideoAllocation { default: if f.provisional.allocatedLayer.IsValid() { - alloc.BandwidthRequested = f.provisional.Bitrates[f.provisional.allocatedLayer.Spatial][f.provisional.allocatedLayer.Temporal] + alloc.BandwidthRequested = f.provisional.bitrates[f.provisional.allocatedLayer.Spatial][f.provisional.allocatedLayer.Temporal] } - alloc.BandwidthDelta = alloc.BandwidthRequested - getBandwidthNeeded(f.provisional.Bitrates, f.vls.GetTarget(), f.lastAllocation.BandwidthRequested) + alloc.BandwidthDelta = alloc.BandwidthRequested - getBandwidthNeeded(f.provisional.bitrates, f.vls.GetTarget(), f.lastAllocation.BandwidthRequested) if f.provisional.allocatedLayer.GreaterThan(f.provisional.maxLayer) || alloc.BandwidthRequested >= getOptimalBandwidthNeeded( f.provisional.muted, f.provisional.pubMuted, f.provisional.maxSeenLayer.Spatial, - f.provisional.Bitrates, + f.provisional.bitrates, f.provisional.maxLayer, ) { // could be greater than optimal if overshooting diff --git a/pkg/sfu/forwarder_test.go b/pkg/sfu/forwarder_test.go index abd58144a..f621fd587 100644 --- a/pkg/sfu/forwarder_test.go +++ b/pkg/sfu/forwarder_test.go @@ -40,13 +40,19 @@ func newForwarder(codec webrtc.RTPCodecCapability, kind webrtc.RTPCodecType) *Fo func TestForwarderMute(t *testing.T) { f := newForwarder(testutils.TestOpusCodec, webrtc.RTPCodecTypeAudio) require.False(t, f.IsMuted()) - muted := f.Mute(false) + muted := f.Mute(false, true) require.False(t, muted) // no change in mute state require.False(t, f.IsMuted()) - muted = f.Mute(true) + + muted = f.Mute(true, false) + require.False(t, muted) + require.False(t, f.IsMuted()) + + muted = f.Mute(true, true) require.True(t, muted) require.True(t, f.IsMuted()) - muted = f.Mute(false) + + muted = f.Mute(false, true) require.True(t, muted) require.False(t, f.IsMuted()) } @@ -164,7 +170,7 @@ func TestForwarderAllocateOptimal(t *testing.T) { f.SetMaxPublishedLayer(buffer.DefaultMaxLayerSpatial) // muted should not consume any bandwidth - f.Mute(true) + f.Mute(true, true) disable(f) expectedResult = VideoAllocation{ PauseReason: VideoPauseReasonMuted, @@ -180,7 +186,7 @@ func TestForwarderAllocateOptimal(t *testing.T) { require.Equal(t, expectedResult, result) require.Equal(t, expectedResult, f.lastAllocation) - f.Mute(false) + f.Mute(false, true) // pub muted should not consume any bandwidth f.PubMute(true) @@ -616,7 +622,7 @@ func TestForwarderProvisionalAllocateMute(t *testing.T) { {9, 10, 11, 12}, } - f.Mute(true) + f.Mute(true, true) f.ProvisionalAllocatePrepare(nil, bitrates) isCandidate, usedBitrate := f.ProvisionalAllocate(bitrates[2][3], buffer.VideoLayer{Spatial: 0, Temporal: 0}, true, false) @@ -651,13 +657,14 @@ func TestForwarderProvisionalAllocateGetCooperativeTransition(t *testing.T) { f.SetMaxPublishedLayer(buffer.DefaultMaxLayerSpatial) f.SetMaxTemporalLayerSeen(buffer.DefaultMaxLayerTemporal) + availableLayers := []int32{0, 1, 2} bitrates := Bitrates{ {1, 2, 3, 4}, {5, 6, 7, 8}, {9, 10, 0, 0}, } - f.ProvisionalAllocatePrepare(nil, bitrates) + f.ProvisionalAllocatePrepare(availableLayers, bitrates) // from scratch (buffer.InvalidLayer) should give back layer (0, 0) expectedTransition := VideoTransition{ @@ -665,8 +672,10 @@ func TestForwarderProvisionalAllocateGetCooperativeTransition(t *testing.T) { To: buffer.VideoLayer{Spatial: 0, Temporal: 0}, BandwidthDelta: 1, } - transition := f.ProvisionalAllocateGetCooperativeTransition(false) + transition, al, brs := f.ProvisionalAllocateGetCooperativeTransition(false) require.Equal(t, expectedTransition, transition) + require.Equal(t, availableLayers, al) + require.Equal(t, bitrates, brs) // committing should set target to (0, 0) expectedLayers := buffer.VideoLayer{Spatial: 0, Temporal: 0} @@ -695,8 +704,10 @@ func TestForwarderProvisionalAllocateGetCooperativeTransition(t *testing.T) { To: targetLayer, BandwidthDelta: 0, } - transition = f.ProvisionalAllocateGetCooperativeTransition(false) + transition, al, brs = f.ProvisionalAllocateGetCooperativeTransition(false) require.Equal(t, expectedTransition, transition) + require.Equal(t, availableLayers, al) + require.Equal(t, bitrates, brs) // committing should set target to (2, 1) expectedLayers = buffer.VideoLayer{Spatial: 2, Temporal: 1} @@ -723,14 +734,16 @@ func TestForwarderProvisionalAllocateGetCooperativeTransition(t *testing.T) { To: buffer.VideoLayer{Spatial: 2, Temporal: 1}, BandwidthDelta: 0, } - transition = f.ProvisionalAllocateGetCooperativeTransition(false) + transition, al, brs = f.ProvisionalAllocateGetCooperativeTransition(false) require.Equal(t, expectedTransition, transition) + require.Equal(t, availableLayers, al) + require.Equal(t, bitrates, brs) f.ProvisionalAllocateCommit() // mute - f.Mute(true) - f.ProvisionalAllocatePrepare(nil, bitrates) + f.Mute(true, true) + f.ProvisionalAllocatePrepare(availableLayers, bitrates) // mute should send target to buffer.InvalidLayer expectedTransition = VideoTransition{ @@ -738,17 +751,20 @@ func TestForwarderProvisionalAllocateGetCooperativeTransition(t *testing.T) { To: buffer.InvalidLayer, BandwidthDelta: -10, } - transition = f.ProvisionalAllocateGetCooperativeTransition(false) + transition, al, brs = f.ProvisionalAllocateGetCooperativeTransition(false) require.Equal(t, expectedTransition, transition) + require.Equal(t, availableLayers, al) + require.Equal(t, bitrates, brs) f.ProvisionalAllocateCommit() // // Test allowOvershoot // - f.Mute(false) + f.Mute(false, true) f.SetMaxSpatialLayer(0) + availableLayers = []int32{1, 2} bitrates = Bitrates{ {0, 0, 0, 0}, {5, 6, 7, 8}, @@ -756,7 +772,7 @@ func TestForwarderProvisionalAllocateGetCooperativeTransition(t *testing.T) { } f.vls.SetTarget(buffer.InvalidLayer) - f.ProvisionalAllocatePrepare(nil, bitrates) + f.ProvisionalAllocatePrepare(availableLayers, bitrates) // from scratch (buffer.InvalidLayer) should go to a layer past maximum as overshoot is allowed expectedTransition = VideoTransition{ @@ -764,8 +780,10 @@ func TestForwarderProvisionalAllocateGetCooperativeTransition(t *testing.T) { To: buffer.VideoLayer{Spatial: 1, Temporal: 0}, BandwidthDelta: 5, } - transition = f.ProvisionalAllocateGetCooperativeTransition(true) + transition, al, brs = f.ProvisionalAllocateGetCooperativeTransition(true) require.Equal(t, expectedTransition, transition) + require.Equal(t, availableLayers, al) + require.Equal(t, bitrates, brs) // committing should set target to (1, 0) expectedLayers = buffer.VideoLayer{Spatial: 1, Temporal: 0} @@ -804,8 +822,10 @@ func TestForwarderProvisionalAllocateGetCooperativeTransition(t *testing.T) { To: buffer.VideoLayer{Spatial: 0, Temporal: 2}, BandwidthDelta: -5, // 5 was the bandwidth needed for the last allocation } - transition = f.ProvisionalAllocateGetCooperativeTransition(true) + transition, al, brs = f.ProvisionalAllocateGetCooperativeTransition(true) require.Equal(t, expectedTransition, transition) + require.Equal(t, []int32{}, al) + require.Equal(t, bitrates, brs) // committing should set target to (0, 2) expectedLayers = buffer.VideoLayer{Spatial: 0, Temporal: 2} @@ -844,13 +864,14 @@ func TestForwarderProvisionalAllocateGetBestWeightedTransition(t *testing.T) { f.SetMaxSpatialLayer(buffer.DefaultMaxLayerSpatial) f.SetMaxTemporalLayer(buffer.DefaultMaxLayerTemporal) + availableLayers := []int32{0, 1, 2} bitrates := Bitrates{ {1, 2, 3, 4}, {5, 6, 7, 8}, {9, 10, 11, 12}, } - f.ProvisionalAllocatePrepare(nil, bitrates) + f.ProvisionalAllocatePrepare(availableLayers, bitrates) f.vls.SetTarget(buffer.VideoLayer{Spatial: 2, Temporal: 2}) f.lastAllocation.BandwidthRequested = bitrates[2][2] @@ -859,8 +880,10 @@ func TestForwarderProvisionalAllocateGetBestWeightedTransition(t *testing.T) { To: buffer.VideoLayer{Spatial: 2, Temporal: 0}, BandwidthDelta: -2, } - transition := f.ProvisionalAllocateGetBestWeightedTransition() + transition, al, brs := f.ProvisionalAllocateGetBestWeightedTransition() require.Equal(t, expectedTransition, transition) + require.Equal(t, availableLayers, al) + require.Equal(t, bitrates, brs) } func TestForwarderAllocateNextHigher(t *testing.T) { @@ -1142,7 +1165,7 @@ func TestForwarderPauseMute(t *testing.T) { // should have set target at (0, 0) f.ProvisionalAllocateCommit() - f.Mute(true) + f.Mute(true, true) expectedResult := VideoAllocation{ PauseReason: VideoPauseReasonMuted, BandwidthRequested: 0, @@ -1161,7 +1184,7 @@ func TestForwarderPauseMute(t *testing.T) { func TestForwarderGetTranslationParamsMuted(t *testing.T) { f := newForwarder(testutils.TestVP8Codec, webrtc.RTPCodecTypeVideo) - f.Mute(true) + f.Mute(true, true) params := &testutils.TestExtPacketParams{ SequenceNumber: 23333, diff --git a/pkg/sfu/streamallocator/streamallocator.go b/pkg/sfu/streamallocator/streamallocator.go index e188d54c3..0c2d7d52b 100644 --- a/pkg/sfu/streamallocator/streamallocator.go +++ b/pkg/sfu/streamallocator/streamallocator.go @@ -515,6 +515,18 @@ func (s *StreamAllocator) IsBWEEnabled(downTrack *sfu.DownTrack) bool { return true } +// called to check if track subscription mute can be applied +func (s *StreamAllocator) IsSubscribeMutable(downTrack *sfu.DownTrack) bool { + s.videoTracksMu.Lock() + defer s.videoTracksMu.Unlock() + + if track := s.videoTracks[livekit.TrackID(downTrack.ID())]; track != nil { + return track.IsSubscribeMutable() + } + + return true +} + func (s *StreamAllocator) maybePostEventAllocateTrack(downTrack *sfu.DownTrack) { shouldPost := false s.videoTracksMu.Lock() diff --git a/pkg/sfu/streamallocator/track.go b/pkg/sfu/streamallocator/track.go index 528792928..be93b9e4b 100644 --- a/pkg/sfu/streamallocator/track.go +++ b/pkg/sfu/streamallocator/track.go @@ -101,6 +101,10 @@ func (t *Track) SetStreamState(streamState StreamState) bool { return true } +func (t *Track) IsSubscribeMutable() bool { + return t.streamState != StreamStatePaused +} + func (t *Track) SetPriority(priority uint8) bool { if priority == 0 { switch t.source {