From 8a7478fd0e39e3b86cf3836275c4d94a3c79e14e Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Wed, 8 Feb 2023 20:03:32 +0530 Subject: [PATCH] Send stream state paused only when it is paused due to bandwidth limitation. (#1391) * WIP commit * Send stream state paused only when it is paused due to bandwidth limitation. When stream is resumed after a stream is paused, an active update is sent. Note that this means if there are intervening events like mute/unmute between pause and resume, resume will be sent. --- pkg/sfu/downtrack.go | 15 ++- pkg/sfu/forwarder.go | 191 ++++++++++++++-------------------- pkg/sfu/forwarder_test.go | 110 ++++++-------------- pkg/sfu/streamallocator.go | 82 ++++++++++++--- pkg/sfu/videolayerselector.go | 2 +- 5 files changed, 194 insertions(+), 206 deletions(-) diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index 11c68acea..e8ad4c7c8 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -201,6 +201,9 @@ type DownTrack struct { // max layer change callback onSubscribedLayersChanged func(dt *DownTrack, layers VideoLayers) + // target layer found callback + onTargetLayerFound func(dt *DownTrack) + // packet sent callback onPacketSent []func(dt *DownTrack, size int) @@ -543,7 +546,7 @@ func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) error { d.onMaxLayerChanged(d, layer) } - if extPkt.KeyFrame || tp.switchingToTargetLayer { + if extPkt.KeyFrame || tp.isSwitchingToTargetLayer { d.isNACKThrottled.Store(false) d.rtpStats.UpdateKeyFrame(1) @@ -552,8 +555,12 @@ func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) error { d.stopKeyFrameRequester() } - if !tp.switchingToTargetLayer { + if !tp.isSwitchingToTargetLayer { d.logger.Debugw("forwarding key frame", "layer", layer) + } else { + if d.onTargetLayerFound != nil { + d.onTargetLayerFound(d) + } } } @@ -910,6 +917,10 @@ func (d *DownTrack) OnSubscribedLayersChanged(fn func(dt *DownTrack, layers Vide d.onSubscribedLayersChanged = fn } +func (d *DownTrack) OnTargetLayerFound(fn func(dt *DownTrack)) { + d.onTargetLayerFound = fn +} + func (d *DownTrack) OnPacketSent(fn func(dt *DownTrack, size int)) { d.onPacketSent = append(d.onPacketSent, fn) } diff --git a/pkg/sfu/forwarder.go b/pkg/sfu/forwarder.go index 2ce1fcdbe..62d19b772 100644 --- a/pkg/sfu/forwarder.go +++ b/pkg/sfu/forwarder.go @@ -35,22 +35,28 @@ const ( // ------------------------------------------------------------------- -type VideoStreamingChange int +type VideoPauseReason int const ( - VideoStreamingChangeNone VideoStreamingChange = iota - VideoStreamingChangePausing - VideoStreamingChangeResuming + VideoPauseReasonNone VideoPauseReason = iota + VideoPauseReasonMuted + VideoPauseReasonPubMuted + VideoPauseReasonFeedDry + VideoPauseReasonBandwidth ) -func (v VideoStreamingChange) String() string { +func (v VideoPauseReason) String() string { switch v { - case VideoStreamingChangeNone: + case VideoPauseReasonNone: return "NONE" - case VideoStreamingChangePausing: - return "PAUSING" - case VideoStreamingChangeResuming: - return "RESUMING" + case VideoPauseReasonMuted: + return "MUTED" + case VideoPauseReasonPubMuted: + return "PUB_MUTED" + case VideoPauseReasonFeedDry: + return "FEED_DRY" + case VideoPauseReasonBandwidth: + return "BANDWIDTH" default: return fmt.Sprintf("%d", int(v)) } @@ -58,42 +64,9 @@ func (v VideoStreamingChange) String() string { // ------------------------------------------------------------------- -type VideoAllocationState int - -const ( - VideoAllocationStateNone VideoAllocationState = iota - VideoAllocationStateMuted - VideoAllocationStatePubMuted - VideoAllocationStateFeedDry - VideoAllocationStateAwaitingMeasurement - VideoAllocationStateOptimal - VideoAllocationStateDeficient -) - -func (v VideoAllocationState) String() string { - switch v { - case VideoAllocationStateNone: - return "NONE" - case VideoAllocationStateMuted: - return "MUTED" - case VideoAllocationStatePubMuted: - return "PUB_MUTED" - case VideoAllocationStateFeedDry: - return "FEED_DRY" - case VideoAllocationStateAwaitingMeasurement: - return "AWAITING_MEASUREMENT" - case VideoAllocationStateOptimal: - return "OPTIMAL" - case VideoAllocationStateDeficient: - return "DEFICIENT" - default: - return fmt.Sprintf("%d", int(v)) - } -} - type VideoAllocation struct { - state VideoAllocationState - change VideoStreamingChange + pauseReason VideoPauseReason + isDeficient bool bandwidthRequested int64 bandwidthDelta int64 availableLayers []int32 @@ -104,8 +77,8 @@ type VideoAllocation struct { } func (v VideoAllocation) String() string { - return fmt.Sprintf("VideoAllocation{state: %s, change: %s, bw: %d, del: %d, avail: %+v, exempt: %+v, rates: %+v, target: %s, dist: %d}", - v.state, v.change, v.bandwidthRequested, v.bandwidthDelta, v.availableLayers, v.exemptedLayers, v.bitrates, v.targetLayers, v.distanceToDesired) + return fmt.Sprintf("VideoAllocation{pause: %s, def: %+v, bw: %d, del: %d, avail: %+v, exempt: %+v, rates: %+v, target: %s, dist: %d}", + v.pauseReason, v.isDeficient, v.bandwidthRequested, v.bandwidthDelta, v.availableLayers, v.exemptedLayers, v.bitrates, v.targetLayers, v.distanceToDesired) } var ( @@ -125,7 +98,6 @@ type VideoAllocationProvisional struct { maxLayers VideoLayers currentLayers VideoLayers parkedLayers VideoLayers - allocatedLayers VideoLayers } @@ -155,7 +127,7 @@ type TranslationParams struct { // indicates this frame has 'Switch' decode indication for target layer // TODO : in theory, we need check frame chain is not broken for the target // but we don't have frame queue now, so just use decode target indication - switchingToTargetLayer bool + isSwitchingToTargetLayer bool } // ------------------------------------------------------------------- @@ -526,6 +498,11 @@ func (f *Forwarder) getOptimalBandwidthNeeded(brs Bitrates, maxLayers VideoLayer } } + // could be 0 due to either + // 1. publisher has stopped all layers ==> feed dry. + // 2. stream tracker has declared all layers stopped, functionally same as above. + // But, listed differently as this could be a mis-detection. + // 3. Bitrate measurement is pending. return 0 } @@ -609,7 +586,7 @@ func (f *Forwarder) getDistanceToDesired(brs Bitrates, targetLayers VideoLayers, } func (f *Forwarder) isDeficientLocked() bool { - return f.lastAllocation.state == VideoAllocationStateDeficient + return f.lastAllocation.isDeficient } func (f *Forwarder) IsDeficient() bool { @@ -646,6 +623,7 @@ func (f *Forwarder) AllocateOptimal(brs Bitrates, allowOvershoot bool) VideoAllo } alloc := VideoAllocation{ + pauseReason: VideoPauseReasonNone, availableLayers: f.availableLayers, exemptedLayers: f.exemptedLayers, bitrates: brs, @@ -654,28 +632,24 @@ func (f *Forwarder) AllocateOptimal(brs Bitrates, allowOvershoot bool) VideoAllo switch { case f.muted: - alloc.state = VideoAllocationStateMuted + alloc.pauseReason = VideoPauseReasonMuted case f.pubMuted: - alloc.state = VideoAllocationStatePubMuted + alloc.pauseReason = VideoPauseReasonPubMuted // leave it at current layers for opportunistic resume alloc.targetLayers = f.currentLayers case f.parkedLayers.IsValid(): // if parked on a layer, let it continue - alloc.state = VideoAllocationStateOptimal alloc.targetLayers = f.parkedLayers case !f.bitrateAvailable(brs): - // feed bitrate not yet calculated for all available layers - alloc.state = VideoAllocationStateAwaitingMeasurement - + // Feed bitrate not yet calculated for all available layers // // Target the highest layer available <= max subscribed layer. // // If target already set, move allocation to the highest available layer <= max subscribed layer only if that is higher than existing target. // It is possible that in situations like coming out of publisher mute, target is already set higher due to parked layers. - // if f.availableLayers[len(f.availableLayers)-1] > f.targetLayers.Spatial { alloc.targetLayers = VideoLayers{ Spatial: int32(math.Min(float64(f.maxLayers.Spatial), float64(f.availableLayers[len(f.availableLayers)-1]))), @@ -699,7 +673,6 @@ func (f *Forwarder) AllocateOptimal(brs Bitrates, allowOvershoot bool) VideoAllo } alloc.bandwidthRequested = brs[s][t] - alloc.state = VideoAllocationStateOptimal break } @@ -727,7 +700,6 @@ func (f *Forwarder) AllocateOptimal(brs Bitrates, allowOvershoot bool) VideoAllo } alloc.bandwidthRequested = brs[s][t] - alloc.state = VideoAllocationStateOptimal f.logger.Infow("allowing overshoot", "maxLayer", f.maxLayers, "targetLayers", alloc.targetLayers) break } @@ -740,7 +712,7 @@ func (f *Forwarder) AllocateOptimal(brs Bitrates, allowOvershoot bool) VideoAllo // feed may be dry, leave target at current if already started for opportunistic resume if alloc.bandwidthRequested == 0 && f.maxLayers.IsValid() { - alloc.state = VideoAllocationStateFeedDry + alloc.pauseReason = VideoPauseReasonFeedDry if f.started { alloc.targetLayers = f.currentLayers } else { @@ -1080,44 +1052,42 @@ func (f *Forwarder) ProvisionalAllocateCommit() VideoAllocation { switch { case f.provisional.muted: - alloc.state = VideoAllocationStateMuted + alloc.pauseReason = VideoPauseReasonMuted case f.provisional.pubMuted: - alloc.state = VideoAllocationStatePubMuted + alloc.pauseReason = VideoPauseReasonPubMuted - case len(f.provisional.availableLayers) == 0: - // feed is dry - alloc.state = VideoAllocationStateFeedDry + case f.getOptimalBandwidthNeeded(f.provisional.bitrates, f.provisional.maxLayers) == 0: + if f.provisional.allocatedLayers.IsValid() { + // overshoot + alloc.bandwidthRequested = f.provisional.bitrates[f.provisional.allocatedLayers.Spatial][f.provisional.allocatedLayers.Temporal] + alloc.bandwidthDelta = alloc.bandwidthRequested - f.lastAllocation.bandwidthRequested + } else { + // some exempt layer may be available, but technically the feed is dry + alloc.pauseReason = VideoPauseReasonFeedDry - case !f.provisional.allocatedLayers.IsValid(): - alloc.state = VideoAllocationStateDeficient - - // leave target at current if current layer is exempted for opportunistic forwarding - if f.provisional.currentLayers.IsValid() { - for _, s := range f.provisional.exemptedLayers { - if s <= f.provisional.maxLayers.Spatial && f.provisional.currentLayers.Spatial == s { - f.provisional.allocatedLayers = f.provisional.currentLayers - alloc.targetLayers = f.provisional.allocatedLayers - break - } + // leave target at current for opportunistic forwarding + if f.provisional.currentLayers.IsValid() && f.provisional.currentLayers.Spatial <= f.provisional.maxLayers.Spatial { + f.provisional.allocatedLayers = f.provisional.currentLayers + alloc.targetLayers = f.provisional.allocatedLayers } } + default: - optimalBandwidthNeeded := f.getOptimalBandwidthNeeded(f.provisional.bitrates, f.provisional.maxLayers) - bandwidthRequested := f.provisional.bitrates[f.provisional.allocatedLayers.Spatial][f.provisional.allocatedLayers.Temporal] + if f.provisional.allocatedLayers.IsValid() { + alloc.bandwidthRequested = f.provisional.bitrates[f.provisional.allocatedLayers.Spatial][f.provisional.allocatedLayers.Temporal] + } + alloc.bandwidthDelta = alloc.bandwidthRequested - f.lastAllocation.bandwidthRequested - alloc.bandwidthRequested = bandwidthRequested - alloc.bandwidthDelta = bandwidthRequested - f.lastAllocation.bandwidthRequested - - if f.provisional.allocatedLayers.GreaterThan(f.provisional.maxLayers) || (optimalBandwidthNeeded > 0 && bandwidthRequested >= optimalBandwidthNeeded) { + if f.provisional.allocatedLayers.GreaterThan(f.provisional.maxLayers) || + alloc.bandwidthRequested >= f.getOptimalBandwidthNeeded(f.provisional.bitrates, f.provisional.maxLayers) { // could be greater than optimal if overshooting - alloc.state = VideoAllocationStateOptimal + alloc.isDeficient = false } else { - // - // Optimal bandwidth could be 0 if using exempted layer. - // Exempted layer is still treated as deficient. - // - alloc.state = VideoAllocationStateDeficient + alloc.isDeficient = true + if !f.provisional.allocatedLayers.IsValid() { + alloc.pauseReason = VideoPauseReasonBandwidth + } } } @@ -1130,19 +1100,16 @@ func (f *Forwarder) AllocateNextHigher(availableChannelCapacity int64, brs Bitra defer f.lock.Unlock() if f.kind == webrtc.RTPCodecTypeAudio { - f.lastAllocation.change = VideoStreamingChangeNone return f.lastAllocation, false } // if not deficient, nothing to do - if f.lastAllocation.state != VideoAllocationStateDeficient { - f.lastAllocation.change = VideoStreamingChangeNone + if !f.isDeficientLocked() { return f.lastAllocation, false } // if targets are still pending, don't increase if f.targetLayers.IsValid() && f.targetLayers != f.currentLayers { - f.lastAllocation.change = VideoStreamingChangeNone return f.lastAllocation, false } @@ -1166,13 +1133,12 @@ func (f *Forwarder) AllocateNextHigher(availableChannelCapacity int64, brs Bitra if !allowOvershoot && bandwidthRequested-alreadyAllocated > availableChannelCapacity { // next higher available layer does not fit, return - f.lastAllocation.change = VideoStreamingChangeNone return true, f.lastAllocation, false } targetLayers := VideoLayers{Spatial: s, Temporal: t} alloc := VideoAllocation{ - state: VideoAllocationStateDeficient, + isDeficient: true, bandwidthRequested: bandwidthRequested, bandwidthDelta: bandwidthRequested - alreadyAllocated, availableLayers: f.availableLayers, @@ -1181,8 +1147,8 @@ func (f *Forwarder) AllocateNextHigher(availableChannelCapacity int64, brs Bitra targetLayers: targetLayers, distanceToDesired: f.getDistanceToDesired(brs, targetLayers, f.maxLayers), } - if targetLayers.GreaterThan(f.maxLayers) || (optimalBandwidthNeeded > 0 && bandwidthRequested >= optimalBandwidthNeeded) { - alloc.state = VideoAllocationStateOptimal + if targetLayers.GreaterThan(f.maxLayers) || bandwidthRequested >= optimalBandwidthNeeded { + alloc.isDeficient = false } return true, f.updateAllocation(alloc, "next-higher"), true @@ -1226,7 +1192,6 @@ func (f *Forwarder) AllocateNextHigher(availableChannelCapacity int64, brs Bitra } } - f.lastAllocation.change = VideoStreamingChangeNone return f.lastAllocation, false } @@ -1239,7 +1204,7 @@ func (f *Forwarder) GetNextHigherTransition(brs Bitrates, allowOvershoot bool) ( } // if not deficient, nothing to do - if f.lastAllocation.state != VideoAllocationStateDeficient { + if !f.isDeficientLocked() { return VideoTransition{}, false } @@ -1330,18 +1295,19 @@ func (f *Forwarder) Pause(brs Bitrates) VideoAllocation { switch { case f.muted: - alloc.state = VideoAllocationStateMuted + alloc.pauseReason = VideoPauseReasonMuted case f.pubMuted: - alloc.state = VideoAllocationStatePubMuted + alloc.pauseReason = VideoPauseReasonPubMuted - case len(f.availableLayers) == 0: - // feed is dry - alloc.state = VideoAllocationStateFeedDry + case f.getOptimalBandwidthNeeded(brs, f.maxLayers) == 0: + // some exempt layer may be available, but technically the feed is dry + alloc.pauseReason = VideoPauseReasonFeedDry default: - // feed bitrate is not yet calculated or pausing due to lack of bandwidth - alloc.state = VideoAllocationStateDeficient + // pausing due to lack of bandwidth + alloc.isDeficient = true + alloc.pauseReason = VideoPauseReasonBandwidth } f.clearParkedLayers() @@ -1349,14 +1315,10 @@ func (f *Forwarder) Pause(brs Bitrates) VideoAllocation { } func (f *Forwarder) updateAllocation(alloc VideoAllocation, reason string) VideoAllocation { - if !f.targetLayers.IsValid() && alloc.targetLayers.IsValid() { - alloc.change = VideoStreamingChangeResuming - } else if f.targetLayers.IsValid() && !alloc.targetLayers.IsValid() { - alloc.change = VideoStreamingChangePausing - } - - if alloc.state != f.lastAllocation.state || alloc.targetLayers != f.lastAllocation.targetLayers { - f.logger.Debugw(fmt.Sprintf("stream allocation: %s", reason), "allocation", alloc) + if alloc.isDeficient != f.lastAllocation.isDeficient || + alloc.pauseReason != f.lastAllocation.pauseReason || + alloc.targetLayers != f.lastAllocation.targetLayers { + f.logger.Infow(fmt.Sprintf("stream allocation: %s", reason), "allocation", alloc) } f.lastAllocation = alloc @@ -1568,7 +1530,7 @@ func (f *Forwarder) getTranslationParamsVideo(extPkt *buffer.ExtPacket, layer in tp.shouldDrop = true f.rtpMunger.PacketDropped(extPkt) return tp, nil - } else if tp.switchingToTargetLayer { + } else if tp.isSwitchingToTargetLayer { // lock to target layer f.logger.Infow("locking to target layer", "current", f.currentLayers, "target", f.targetLayers) f.currentLayers.Spatial = f.targetLayers.Spatial @@ -1619,6 +1581,7 @@ func (f *Forwarder) getTranslationParamsVideo(extPkt *buffer.ExtPacket, layer in } if found { + tp.isSwitchingToTargetLayer = true f.clearParkedLayers() if f.currentLayers.Spatial >= f.maxLayers.Spatial || f.currentLayers.Spatial == (f.numAdvertisedLayers-1) { tp.isSwitchingToMaxLayer = true diff --git a/pkg/sfu/forwarder_test.go b/pkg/sfu/forwarder_test.go index d86371f51..40bc7097e 100644 --- a/pkg/sfu/forwarder_test.go +++ b/pkg/sfu/forwarder_test.go @@ -181,8 +181,7 @@ func TestForwarderAllocateOptimal(t *testing.T) { f.Mute(true) disable(f) expectedResult := VideoAllocation{ - state: VideoAllocationStateMuted, - change: VideoStreamingChangeNone, + pauseReason: VideoPauseReasonMuted, bandwidthRequested: 0, bandwidthDelta: 0, availableLayers: nil, @@ -196,17 +195,15 @@ func TestForwarderAllocateOptimal(t *testing.T) { f.Mute(false) - // feed dry state, target set to maximum for opportunistic start + // feed dry, target set to maximum for opportunistic start f.SetNumAdvertisedLayers(3) - f.lastAllocation.state = VideoAllocationStateNone disable(f) expectedTargetLayers := VideoLayers{ Spatial: 2, Temporal: DefaultMaxLayerTemporal, } expectedResult = VideoAllocation{ - state: VideoAllocationStateFeedDry, - change: VideoStreamingChangeResuming, + pauseReason: VideoPauseReasonFeedDry, bandwidthRequested: 0, bandwidthDelta: 0, availableLayers: nil, @@ -224,8 +221,6 @@ func TestForwarderAllocateOptimal(t *testing.T) { Temporal: 1, } expectedResult = VideoAllocation{ - state: VideoAllocationStateOptimal, - change: VideoStreamingChangeNone, bandwidthRequested: 0, bandwidthDelta: 0, availableLayers: nil, @@ -240,7 +235,6 @@ func TestForwarderAllocateOptimal(t *testing.T) { f.parkedLayers = InvalidLayers // awaiting measurement, i.e. bitrates are not available, but layers available - f.lastAllocation.state = VideoAllocationStateNone disable(f) f.UpTrackLayersChange([]int32{0}, []int32{}) expectedTargetLayers = VideoLayers{ @@ -248,8 +242,6 @@ func TestForwarderAllocateOptimal(t *testing.T) { Temporal: DefaultMaxLayerTemporal, } expectedResult = VideoAllocation{ - state: VideoAllocationStateAwaitingMeasurement, - change: VideoStreamingChangeResuming, bandwidthRequested: 0, bandwidthDelta: 0, availableLayers: []int32{0}, @@ -265,6 +257,8 @@ func TestForwarderAllocateOptimal(t *testing.T) { // layers are available, but all layers <= max are exempted. // forwarder not started yet, so should set target to max. + // pauseReason should be populated to indicate feed is dry + // because exempt layers are actually detected as stopped. f.SetMaxSpatialLayer(0) f.currentLayers = VideoLayers{Spatial: 0, Temporal: 0} f.UpTrackLayersChange([]int32{0, 1, 2}, []int32{0}) @@ -273,8 +267,7 @@ func TestForwarderAllocateOptimal(t *testing.T) { Temporal: DefaultMaxLayerTemporal, } expectedResult = VideoAllocation{ - state: VideoAllocationStateFeedDry, - change: VideoStreamingChangeNone, + pauseReason: VideoPauseReasonFeedDry, bandwidthRequested: 0, bandwidthDelta: 0, availableLayers: []int32{0, 1, 2}, @@ -295,8 +288,7 @@ func TestForwarderAllocateOptimal(t *testing.T) { Temporal: 0, } expectedResult = VideoAllocation{ - state: VideoAllocationStateFeedDry, - change: VideoStreamingChangeNone, + pauseReason: VideoPauseReasonFeedDry, bandwidthRequested: 0, bandwidthDelta: 0, availableLayers: []int32{0, 1, 2}, @@ -317,8 +309,7 @@ func TestForwarderAllocateOptimal(t *testing.T) { Temporal: 0, } expectedResult = VideoAllocation{ - state: VideoAllocationStateFeedDry, - change: VideoStreamingChangeNone, + pauseReason: VideoPauseReasonFeedDry, bandwidthRequested: 0, bandwidthDelta: 0, availableLayers: []int32{0, 1, 2}, @@ -334,7 +325,6 @@ func TestForwarderAllocateOptimal(t *testing.T) { // allocate using bitrates, allocation should choose optimal f.currentLayers = InvalidLayers - f.SetMaxSpatialLayer(DefaultMaxLayerSpatial) f.UpTrackLayersChange([]int32{0, 1, 2}, []int32{}) f.maxLayers = VideoLayers{Spatial: 1, Temporal: 3} expectedTargetLayers = VideoLayers{ @@ -342,8 +332,6 @@ func TestForwarderAllocateOptimal(t *testing.T) { Temporal: 3, } expectedResult = VideoAllocation{ - state: VideoAllocationStateOptimal, - change: VideoStreamingChangeNone, bandwidthRequested: bitrates[1][3], bandwidthDelta: bitrates[1][3], availableLayers: []int32{0, 1, 2}, @@ -369,8 +357,6 @@ func TestForwarderAllocateOptimal(t *testing.T) { Temporal: 1, } expectedResult = VideoAllocation{ - state: VideoAllocationStateOptimal, - change: VideoStreamingChangeNone, bandwidthRequested: sparseBitrates[2][1], bandwidthDelta: sparseBitrates[2][1] - bitrates[1][3], availableLayers: []int32{2}, @@ -384,10 +370,9 @@ func TestForwarderAllocateOptimal(t *testing.T) { require.Equal(t, InvalidLayers, f.CurrentLayers()) require.Equal(t, expectedTargetLayers, f.TargetLayers()) - // when not allowing overshoot, should not be able to find a layer + // when not allowing overshoot, should leave it at current (current is InvalidLayers in this test) expectedResult = VideoAllocation{ - state: VideoAllocationStateFeedDry, - change: VideoStreamingChangePausing, + pauseReason: VideoPauseReasonFeedDry, bandwidthRequested: 0, bandwidthDelta: -sparseBitrates[2][1], availableLayers: []int32{2}, @@ -439,8 +424,7 @@ func TestForwarderProvisionalAllocate(t *testing.T) { Temporal: 2, } expectedResult := VideoAllocation{ - state: VideoAllocationStateDeficient, - change: VideoStreamingChangeResuming, + isDeficient: true, bandwidthRequested: bitrates[1][2], bandwidthDelta: bitrates[1][2], availableLayers: availableLayers, @@ -465,8 +449,7 @@ func TestForwarderProvisionalAllocate(t *testing.T) { Temporal: 0, } expectedResult = VideoAllocation{ - state: VideoAllocationStateDeficient, - change: VideoStreamingChangeResuming, + isDeficient: true, bandwidthRequested: bitrates[0][0], bandwidthDelta: bitrates[0][0] - bitrates[1][2], availableLayers: availableLayers, @@ -514,8 +497,6 @@ func TestForwarderProvisionalAllocate(t *testing.T) { Temporal: 3, } expectedResult = VideoAllocation{ - state: VideoAllocationStateOptimal, - change: VideoStreamingChangeNone, bandwidthRequested: bitrates[1][3], bandwidthDelta: bitrates[1][3] - 1, // 1 is the last allocation bandwith requested availableLayers: availableLayers, @@ -566,8 +547,7 @@ func TestForwarderProvisionalAllocate(t *testing.T) { Temporal: 2, } expectedResult = VideoAllocation{ - state: VideoAllocationStateDeficient, - change: VideoStreamingChangeNone, + pauseReason: VideoPauseReasonFeedDry, bandwidthRequested: bitrates[0][2], bandwidthDelta: bitrates[0][2] - 8, // 8 is the last allocation bandwith requested availableLayers: availableLayers, @@ -598,7 +578,7 @@ func TestForwarderProvisionalAllocate(t *testing.T) { f.currentLayers = VideoLayers{Spatial: 1, Temporal: 2} f.ProvisionalAllocatePrepare(bitrates) - // all the provisional allocations below should fall back to exempted layer + // all the provisional allocations below should not succeed usedBitrate = f.ProvisionalAllocate(bitrates[2][3], VideoLayers{Spatial: 0, Temporal: 0}, false, true) require.Equal(t, int64(0), usedBitrate) @@ -611,8 +591,7 @@ func TestForwarderProvisionalAllocate(t *testing.T) { require.Equal(t, int64(0), usedBitrate) expectedResult = VideoAllocation{ - state: VideoAllocationStateDeficient, - change: VideoStreamingChangePausing, + pauseReason: VideoPauseReasonFeedDry, bandwidthRequested: 0, bandwidthDelta: 0, availableLayers: availableLayers, @@ -647,8 +626,7 @@ func TestForwarderProvisionalAllocateMute(t *testing.T) { // committing should set target to InvalidLayers as track is muted expectedResult := VideoAllocation{ - state: VideoAllocationStateMuted, - change: VideoStreamingChangeNone, + pauseReason: VideoPauseReasonMuted, bandwidthRequested: 0, bandwidthDelta: 0, availableLayers: nil, @@ -689,8 +667,7 @@ func TestForwarderProvisionalAllocateGetCooperativeTransition(t *testing.T) { // committing should set target to (0, 0) expectedLayers := VideoLayers{Spatial: 0, Temporal: 0} expectedResult := VideoAllocation{ - state: VideoAllocationStateDeficient, - change: VideoStreamingChangeResuming, + isDeficient: true, bandwidthRequested: 1, bandwidthDelta: 1, availableLayers: availableLayers, @@ -718,8 +695,6 @@ func TestForwarderProvisionalAllocateGetCooperativeTransition(t *testing.T) { // committing should set target to (2, 1) expectedLayers = VideoLayers{Spatial: 2, Temporal: 1} expectedResult = VideoAllocation{ - state: VideoAllocationStateOptimal, - change: VideoStreamingChangeNone, bandwidthRequested: 10, bandwidthDelta: 0, availableLayers: availableLayers, @@ -791,8 +766,6 @@ func TestForwarderProvisionalAllocateGetCooperativeTransition(t *testing.T) { // committing should set target to (1, 0) expectedLayers = VideoLayers{Spatial: 1, Temporal: 0} expectedResult = VideoAllocation{ - state: VideoAllocationStateOptimal, - change: VideoStreamingChangeResuming, bandwidthRequested: 5, bandwidthDelta: 5, availableLayers: availableLayers, @@ -836,8 +809,6 @@ func TestForwarderProvisionalAllocateGetCooperativeTransition(t *testing.T) { // committing should set target to (0, 2) expectedLayers = VideoLayers{Spatial: 0, Temporal: 2} expectedResult = VideoAllocation{ - state: VideoAllocationStateDeficient, - change: VideoStreamingChangeResuming, bandwidthRequested: 0, bandwidthDelta: -5, availableLayers: availableLayers, @@ -870,8 +841,6 @@ func TestForwarderProvisionalAllocateGetCooperativeTransition(t *testing.T) { // committing should set target to current layers to enable opportunistic forwarding expectedResult = VideoAllocation{ - state: VideoAllocationStateOptimal, - change: VideoStreamingChangeNone, bandwidthRequested: 0, bandwidthDelta: 0, availableLayers: availableLayers, @@ -931,20 +900,18 @@ func TestForwarderAllocateNextHigher(t *testing.T) { f.SetMaxTemporalLayer(DefaultMaxLayerTemporal) // when not in deficient state, does not boost - f.lastAllocation.state = VideoAllocationStateNone result, boosted = f.AllocateNextHigher(ChannelCapacityInfinity, bitrates, false) require.Equal(t, VideoAllocationDefault, result) require.False(t, boosted) - // if layers have not caught up, should not allocate next layer - f.lastAllocation.state = VideoAllocationStateDeficient + // if layers have not caught up, should not allocate next layer even if deficient + f.lastAllocation.isDeficient = true f.targetLayers = VideoLayers{ Spatial: 0, Temporal: 0, } expectedResult := VideoAllocation{ - state: VideoAllocationStateDeficient, - change: VideoStreamingChangeNone, + isDeficient: true, bandwidthRequested: 0, bandwidthDelta: 0, availableLayers: nil, @@ -969,8 +936,7 @@ func TestForwarderAllocateNextHigher(t *testing.T) { // empty bitrates cannot increase layer expectedResult = VideoAllocation{ - state: VideoAllocationStateDeficient, - change: VideoStreamingChangeNone, + isDeficient: true, bandwidthRequested: 2, bandwidthDelta: 0, availableLayers: nil, @@ -989,8 +955,7 @@ func TestForwarderAllocateNextHigher(t *testing.T) { Temporal: 1, } expectedResult = VideoAllocation{ - state: VideoAllocationStateDeficient, - change: VideoStreamingChangeNone, + isDeficient: true, bandwidthRequested: 3, bandwidthDelta: 1, availableLayers: nil, @@ -1011,8 +976,7 @@ func TestForwarderAllocateNextHigher(t *testing.T) { Temporal: 0, } expectedResult = VideoAllocation{ - state: VideoAllocationStateDeficient, - change: VideoStreamingChangeNone, + isDeficient: true, bandwidthRequested: 4, bandwidthDelta: 1, availableLayers: nil, @@ -1034,8 +998,7 @@ func TestForwarderAllocateNextHigher(t *testing.T) { Temporal: 3, } expectedResult = VideoAllocation{ - state: VideoAllocationStateDeficient, - change: VideoStreamingChangeNone, + isDeficient: true, bandwidthRequested: 5, bandwidthDelta: 1, availableLayers: nil, @@ -1056,8 +1019,6 @@ func TestForwarderAllocateNextHigher(t *testing.T) { Temporal: 1, } expectedResult = VideoAllocation{ - state: VideoAllocationStateOptimal, - change: VideoStreamingChangeNone, bandwidthRequested: 7, bandwidthDelta: 2, availableLayers: nil, @@ -1082,7 +1043,7 @@ func TestForwarderAllocateNextHigher(t *testing.T) { // turn off everything, allocating next layer should result in streaming lowest layers disable(f) - f.lastAllocation.state = VideoAllocationStateDeficient + f.lastAllocation.isDeficient = true f.lastAllocation.bandwidthRequested = 0 expectedTargetLayers = VideoLayers{ @@ -1090,8 +1051,7 @@ func TestForwarderAllocateNextHigher(t *testing.T) { Temporal: 0, } expectedResult = VideoAllocation{ - state: VideoAllocationStateDeficient, - change: VideoStreamingChangeResuming, + isDeficient: true, bandwidthRequested: 2, bandwidthDelta: 2, availableLayers: nil, @@ -1107,8 +1067,7 @@ func TestForwarderAllocateNextHigher(t *testing.T) { // no new available capacity cannot bump up layer expectedResult = VideoAllocation{ - state: VideoAllocationStateDeficient, - change: VideoStreamingChangeNone, + isDeficient: true, bandwidthRequested: 2, bandwidthDelta: 2, availableLayers: nil, @@ -1138,8 +1097,6 @@ func TestForwarderAllocateNextHigher(t *testing.T) { Temporal: 0, } expectedResult = VideoAllocation{ - state: VideoAllocationStateOptimal, - change: VideoStreamingChangeNone, bandwidthRequested: bitrates[1][0], bandwidthDelta: bitrates[1][0], availableLayers: nil, @@ -1175,8 +1132,8 @@ func TestForwarderPause(t *testing.T) { f.ProvisionalAllocateCommit() expectedResult := VideoAllocation{ - state: VideoAllocationStateDeficient, - change: VideoStreamingChangePausing, + pauseReason: VideoPauseReasonBandwidth, + isDeficient: true, bandwidthRequested: 0, bandwidthDelta: 0 - bitrates[0][0], availableLayers: availableLayers, @@ -1210,8 +1167,7 @@ func TestForwarderPauseMute(t *testing.T) { f.Mute(true) expectedResult := VideoAllocation{ - state: VideoAllocationStateMuted, - change: VideoStreamingChangePausing, + pauseReason: VideoPauseReasonMuted, bandwidthRequested: 0, bandwidthDelta: 0 - bitrates[0][0], availableLayers: availableLayers, @@ -1456,7 +1412,8 @@ func TestForwarderGetTranslationParamsVideo(t *testing.T) { } extPkt, _ = testutils.GetTestExtPacketVP8(params, vp8) expectedTP = TranslationParams{ - isSwitchingToMaxLayer: true, + isSwitchingToMaxLayer: true, + isSwitchingToTargetLayer: true, rtp: &TranslationParamsRTP{ snOrdering: SequenceNumberOrderingContiguous, sequenceNumber: 23333, @@ -1710,7 +1667,8 @@ func TestForwarderGetTranslationParamsVideo(t *testing.T) { extPkt, _ = testutils.GetTestExtPacketVP8(params, vp8) expectedTP = TranslationParams{ - isSwitchingToMaxLayer: true, + isSwitchingToMaxLayer: true, + isSwitchingToTargetLayer: true, rtp: &TranslationParamsRTP{ snOrdering: SequenceNumberOrderingContiguous, sequenceNumber: 23338, diff --git a/pkg/sfu/streamallocator.go b/pkg/sfu/streamallocator.go index f4a995fab..83a1c2440 100644 --- a/pkg/sfu/streamallocator.go +++ b/pkg/sfu/streamallocator.go @@ -96,6 +96,7 @@ const ( streamAllocatorSignalPeriodicPing streamAllocatorSignalSendProbe streamAllocatorSignalProbeClusterDone + streamAllocatorSignalTargetLayerFound ) func (s streamAllocatorSignal) String() string { @@ -114,6 +115,8 @@ func (s streamAllocatorSignal) String() string { return "SEND_PROBE" case streamAllocatorSignalProbeClusterDone: return "PROBE_CLUSTER_DONE" + case streamAllocatorSignalTargetLayerFound: + return "TARGET_LAYER_FOUND" default: return fmt.Sprintf("%d", int(s)) } @@ -240,6 +243,7 @@ func (s *StreamAllocator) AddTrack(downTrack *DownTrack, params AddTrackParams) downTrack.OnSubscriptionChanged(s.onSubscriptionChanged) downTrack.OnSubscribedLayersChanged(s.onSubscribedLayersChanged) downTrack.OnPacketSent(s.onPacketSent) + downTrack.OnTargetLayerFound(s.onTargetLayerFound) s.maybePostEventAllocateTrack(downTrack) } @@ -429,6 +433,14 @@ func (s *StreamAllocator) onProbeClusterDone(info ProbeClusterInfo) { }) } +// called when forwarder finds a target layer +func (s *StreamAllocator) onTargetLayerFound(downTrack *DownTrack) { + s.postEvent(Event{ + Signal: streamAllocatorSignalTargetLayerFound, + TrackID: livekit.TrackID(downTrack.ID()), + }) +} + func (s *StreamAllocator) maybePostEventAllocateTrack(downTrack *DownTrack) { shouldPost := false s.videoTracksMu.Lock() @@ -500,6 +512,8 @@ func (s *StreamAllocator) handleEvent(event *Event) { s.handleSignalSendProbe(event) case streamAllocatorSignalProbeClusterDone: s.handleSignalProbeClusterDone(event) + case streamAllocatorSignalTargetLayerFound: + s.handleSignalTargetLayerFound(event) } } @@ -599,6 +613,20 @@ func (s *StreamAllocator) handleSignalProbeClusterDone(event *Event) { s.probeEndTime = s.lastProbeStartTime.Add(queueWait) } +func (s *StreamAllocator) handleSignalTargetLayerFound(event *Event) { + s.videoTracksMu.Lock() + track := s.videoTracks[event.TrackID] + s.videoTracksMu.Unlock() + + if track != nil { + update := NewStreamStateUpdate() + if track.SetPaused(false) { + update.HandleStreamingChange(false, track) + } + s.maybeSendUpdate(update) + } +} + func (s *StreamAllocator) setState(state streamAllocatorState) { if s.state == state { return @@ -719,7 +747,9 @@ func (s *StreamAllocator) allocateTrack(track *Track) { if !s.params.Config.Enabled || s.state == streamAllocatorStateStable || !track.IsManaged() { update := NewStreamStateUpdate() allocation := track.AllocateOptimal(FlagAllowOvershootWhileOptimal) - update.HandleStreamingChange(allocation.change, track) + if allocation.pauseReason == VideoPauseReasonBandwidth && track.SetPaused(true) { + update.HandleStreamingChange(true, track) + } s.maybeSendUpdate(update) return } @@ -744,7 +774,9 @@ func (s *StreamAllocator) allocateTrack(track *Track) { allocation := track.ProvisionalAllocateCommit() update := NewStreamStateUpdate() - update.HandleStreamingChange(allocation.change, track) + if allocation.pauseReason == VideoPauseReasonBandwidth && track.SetPaused(true) { + update.HandleStreamingChange(true, track) + } s.maybeSendUpdate(update) s.adjustState() @@ -785,7 +817,9 @@ func (s *StreamAllocator) allocateTrack(track *Track) { // commit the tracks that contributed for _, t := range contributingTracks { allocation := t.ProvisionalAllocateCommit() - update.HandleStreamingChange(allocation.change, t) + if allocation.pauseReason == VideoPauseReasonBandwidth && track.SetPaused(true) { + update.HandleStreamingChange(true, t) + } } // LK-TODO if got too much extra, can potentially give it to some deficient track @@ -794,7 +828,9 @@ func (s *StreamAllocator) allocateTrack(track *Track) { // commit the track that needs change if enough could be acquired or pause not allowed if !s.params.Config.AllowPause || bandwidthAcquired >= transition.bandwidthDelta { allocation := track.ProvisionalAllocateCommit() - update.HandleStreamingChange(allocation.change, track) + if allocation.pauseReason == VideoPauseReasonBandwidth && track.SetPaused(true) { + update.HandleStreamingChange(true, track) + } } s.maybeSendUpdate(update) @@ -864,7 +900,9 @@ func (s *StreamAllocator) maybeBoostDeficientTracks() { continue } - update.HandleStreamingChange(allocation.change, track) + if allocation.pauseReason == VideoPauseReasonBandwidth && track.SetPaused(true) { + update.HandleStreamingChange(true, track) + } availableChannelCapacity -= allocation.bandwidthDelta if availableChannelCapacity <= 0 { @@ -918,7 +956,9 @@ func (s *StreamAllocator) allocateAllTracks() { } allocation := track.AllocateOptimal(FlagAllowOvershootExemptTrackWhileDeficient) - update.HandleStreamingChange(allocation.change, track) + if allocation.pauseReason == VideoPauseReasonBandwidth && track.SetPaused(true) { + update.HandleStreamingChange(true, track) + } // LK-TODO: optimistic allocation before bitrate is available will return 0. How to account for that? availableChannelCapacity -= allocation.bandwidthRequested @@ -935,7 +975,9 @@ func (s *StreamAllocator) allocateAllTracks() { } allocation := track.Pause() - update.HandleStreamingChange(allocation.change, track) + if allocation.pauseReason == VideoPauseReasonBandwidth && track.SetPaused(true) { + update.HandleStreamingChange(true, track) + } } } else { sorted := s.getSorted() @@ -962,7 +1004,9 @@ func (s *StreamAllocator) allocateAllTracks() { for _, track := range sorted { allocation := track.ProvisionalAllocateCommit() - update.HandleStreamingChange(allocation.change, track) + if allocation.pauseReason == VideoPauseReasonBandwidth && track.SetPaused(true) { + update.HandleStreamingChange(true, track) + } } } @@ -1113,7 +1157,9 @@ func (s *StreamAllocator) maybeProbeWithMedia() { } update := NewStreamStateUpdate() - update.HandleStreamingChange(allocation.change, track) + if allocation.pauseReason == VideoPauseReasonBandwidth && track.SetPaused(true) { + update.HandleStreamingChange(true, track) + } s.maybeSendUpdate(update) s.lastProbeStartTime = time.Now() @@ -1235,15 +1281,14 @@ func NewStreamStateUpdate() *StreamStateUpdate { return &StreamStateUpdate{} } -func (s *StreamStateUpdate) HandleStreamingChange(change VideoStreamingChange, track *Track) { - switch change { - case VideoStreamingChangePausing: +func (s *StreamStateUpdate) HandleStreamingChange(isPaused bool, track *Track) { + if isPaused { s.StreamStates = append(s.StreamStates, &StreamStateInfo{ ParticipantID: track.PublisherID(), TrackID: track.ID(), State: StreamStatePaused, }) - case VideoStreamingChangeResuming: + } else { s.StreamStates = append(s.StreamStates, &StreamStateInfo{ ParticipantID: track.PublisherID(), TrackID: track.ID(), @@ -1272,6 +1317,8 @@ type Track struct { totalRepeatedNacks uint32 isDirty bool + + isPaused bool } func newTrack( @@ -1303,6 +1350,15 @@ func (t *Track) SetDirty(isDirty bool) bool { return true } +func (t *Track) SetPaused(isPaused bool) bool { + if t.isPaused == isPaused { + return false + } + + t.isPaused = isPaused + return true +} + func (t *Track) SetPriority(priority uint8) bool { if priority == 0 { switch t.source { diff --git a/pkg/sfu/videolayerselector.go b/pkg/sfu/videolayerselector.go index baad634ed..507bfe49f 100644 --- a/pkg/sfu/videolayerselector.go +++ b/pkg/sfu/videolayerselector.go @@ -98,7 +98,7 @@ func (s *DDVideoLayerSelector) Select(expPkt *buffer.ExtPacket, tp *TranslationP // expPkt.DependencyDescriptor.FrameDependencies.SpatialId, expPkt.DependencyDescriptor.FrameDependencies.TemporalId)) return false } else if dti == dd.DecodeTargetSwitch { - tp.switchingToTargetLayer = true + tp.isSwitchingToTargetLayer = true } // DD-TODO : add frame to forwarded queue if entire frame is forwarded