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

@@ -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)
 }

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

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,

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 {

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