Merge remote-tracking branch 'origin/master' into raja_min_packets

pkg/sfu/streamallocator/channelobserver.go

@@ -1,9 +1,23 @@
+// Copyright 2023 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 package streamallocator
 
 import (
 	"fmt"
-	"time"
 
+	"github.com/livekit/livekit-server/pkg/config"
 	"github.com/livekit/protocol/logger"
 )
 
@@ -56,14 +70,8 @@ func (c ChannelCongestionReason) String() string {
 // ------------------------------------------------
 
 type ChannelObserverParams struct {
-	Name                           string
-	EstimateRequiredSamples        int
-	EstimateDownwardTrendThreshold float64
-	EstimateValidityWindow         time.Duration
-	NackMinPPS                     uint32
-	NackWindowMinDuration          time.Duration
-	NackWindowMaxDuration          time.Duration
-	NackRatioThreshold             float64
+	Name   string
+	Config config.CongestionControlChannelObserverConfig
 }
 
 type ChannelObserver struct {
@@ -81,17 +89,17 @@ func NewChannelObserver(params ChannelObserverParams, logger logger.Logger) *Cha
 		estimateTrend: NewTrendDetector(TrendDetectorParams{
 			Name:                   params.Name + "-estimate",
 			Logger:                 logger,
-			RequiredSamples:        params.EstimateRequiredSamples,
-			DownwardTrendThreshold: params.EstimateDownwardTrendThreshold,
-			ValidityWindow:         params.EstimateValidityWindow,
+			RequiredSamples:        params.Config.EstimateRequiredSamples,
+			DownwardTrendThreshold: params.Config.EstimateDownwardTrendThreshold,
+			ValidityWindow:         params.Config.EstimateValidityWindow,
 		}),
 		nackTracker: NewNackTracker(NackTrackerParams{
 			Name:              params.Name + "-nack",
 			Logger:            logger,
-			MinPPS:            params.NackMinPPS,
-			WindowMinDuration: params.NackWindowMinDuration,
-			WindowMaxDuration: params.NackWindowMaxDuration,
-			RatioThreshold:    params.NackRatioThreshold,
+			MinPPS:            params.Config.NackMinPPS,
+			WindowMinDuration: params.Config.NackWindowMinDuration,
+			WindowMaxDuration: params.Config.NackWindowMaxDuration,
+			RatioThreshold:    params.Config.NackRatioThreshold,
 		}),
 	}
 }

pkg/sfu/streamallocator/nacktracker.go

@@ -1,3 +1,17 @@
+// Copyright 2023 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 package streamallocator
 
 import (

pkg/sfu/streamallocator/probe_controller.go

@@ -1,3 +1,17 @@
+// Copyright 2023 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 package streamallocator
 
 import (

pkg/sfu/streamallocator/prober.go

@@ -1,3 +1,17 @@
+// Copyright 2023 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 // Design of Prober
 //
 // Probing is used to check for existence of excess channel capacity.

pkg/sfu/streamallocator/ratemonitor.go

@@ -1,3 +1,17 @@
+// Copyright 2023 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 package streamallocator
 
 import (

pkg/sfu/streamallocator/streamallocator.go

@@ -1,3 +1,17 @@
+// Copyright 2023 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 package streamallocator
 
 import (
@@ -39,32 +53,6 @@ const (
 
 // ---------------------------------------------------------------------------
 
-var (
-	ChannelObserverParamsProbe = ChannelObserverParams{
-		Name:                           "probe",
-		EstimateRequiredSamples:        3,
-		EstimateDownwardTrendThreshold: 0.0,
-		EstimateValidityWindow:         10 * time.Second,
-		NackMinPPS:                     0,
-		NackWindowMinDuration:          500 * time.Millisecond,
-		NackWindowMaxDuration:          1 * time.Second,
-		NackRatioThreshold:             0.04,
-	}
-
-	ChannelObserverParamsNonProbe = ChannelObserverParams{
-		Name:                           "non-probe",
-		EstimateRequiredSamples:        8,
-		EstimateDownwardTrendThreshold: -0.5,
-		EstimateValidityWindow:         10 * time.Second,
-		NackMinPPS:                     50,
-		NackWindowMinDuration:          1 * time.Second,
-		NackWindowMaxDuration:          2 * time.Second,
-		NackRatioThreshold:             0.08,
-	}
-)
-
-// ---------------------------------------------------------------------------
-
 type streamAllocatorState int
 
 const (
@@ -79,7 +67,7 @@ func (s streamAllocatorState) String() string {
 	case streamAllocatorStateDeficient:
 		return "DEFICIENT"
 	default:
-		return fmt.Sprintf("%d", int(s))
+		return fmt.Sprintf("UNKNOWN: %d", int(s))
 	}
 }
 
@@ -696,8 +684,8 @@ func (s *StreamAllocator) handleSignalResume(event *Event) {
 
 	if track != nil {
 		update := NewStreamStateUpdate()
-		if track.SetPaused(false) {
-			update.HandleStreamingChange(false, track)
+		if track.SetStreamState(StreamStateActive) {
+			update.HandleStreamingChange(track, StreamStateActive)
 		}
 		s.maybeSendUpdate(update)
 	}
@@ -840,19 +828,29 @@ func (s *StreamAllocator) allocateTrack(track *Track) {
 	if !s.params.Config.Enabled || s.state == streamAllocatorStateStable || !track.IsManaged() {
 		update := NewStreamStateUpdate()
 		allocation := track.AllocateOptimal(FlagAllowOvershootWhileOptimal)
-		if allocation.PauseReason == sfu.VideoPauseReasonBandwidth && track.SetPaused(true) {
-			update.HandleStreamingChange(true, track)
-		}
+		updateStreamStateChange(track, allocation, update)
 		s.maybeSendUpdate(update)
 		return
 	}
 
 	//
 	// In DEFICIENT state,
-	//   1. Find cooperative transition from track that needs allocation.
-	//   2. If track is currently streaming at minimum, do not do anything.
-	//   3. If that track is giving back bits, apply the transition.
-	//   4. If this track needs more, ask for best offer from others and try to use it.
+	//   Two possibilities
+	//   1. Available headroom is enough to accommodate track that needs change.
+	//      Note that the track could be muted, hence stopping.
+	//   2. Have to steal bits from other tracks currently streaming.
+	//
+	//   For both cases, do
+	//     a. Find cooperative transition from track that needs allocation.
+	//     b. If track is currently streaming at minimum, do not do anything.
+	//     c. If track is giving back bits, apply the transition and use bits given
+	//        back to boost any deficient track(s).
+	//
+	//   If track needs more bits, i.e. upward transition (may need resume or higher layer subscription),
+	//     a. Try to allocate using existing headroom. This can be tried to get the best
+	//        possible fit for the available headroom.
+	//     b. If there is not enough headroom to allocate anything, ask for best offer from
+	//        other tracks that are currently streaming and try to use it.
 	//
 	track.ProvisionalAllocatePrepare()
 	transition := track.ProvisionalAllocateGetCooperativeTransition(FlagAllowOvershootWhileDeficient)
@@ -867,24 +865,60 @@ func (s *StreamAllocator) allocateTrack(track *Track) {
 		allocation := track.ProvisionalAllocateCommit()
 
 		update := NewStreamStateUpdate()
-		if allocation.PauseReason == sfu.VideoPauseReasonBandwidth && track.SetPaused(true) {
-			update.HandleStreamingChange(true, track)
-		}
+		updateStreamStateChange(track, allocation, update)
 		s.maybeSendUpdate(update)
 
 		s.adjustState()
-		return
-		// STREAM-ALLOCATOR-TODO-START
-		// Should use the bits given back to start any paused track.
+
+		// Use the bits given back to boost deficient track(s).
 		// Note layer downgrade may actually have positive delta (i.e. consume more bits)
-		// because of when the measurement is done. Watch for that.
-		// STREAM_ALLOCATOR-TODO-END
+		// because of when the measurement is done. But, only available headroom after
+		// applying the transition will be used to boost deficient track(s).
+		s.maybeBoostDeficientTracks()
+		return
 	}
 
-	//
-	// This track is currently not streaming and needs bits to start.
-	// Try to redistribute starting with tracks that are closest to their desired.
-	//
+	// this track is currently not streaming and needs bits to start.
+	// first try an allocation using available headroom
+	availableChannelCapacity := s.getAvailableHeadroom(false)
+	if availableChannelCapacity > 0 {
+		track.ProvisionalAllocateReset() // to reset allocation from co-operative transition above and try fresh
+
+		bestLayer := buffer.InvalidLayer
+
+	alloc_loop:
+		for spatial := int32(0); spatial <= buffer.DefaultMaxLayerSpatial; spatial++ {
+			for temporal := int32(0); temporal <= buffer.DefaultMaxLayerTemporal; temporal++ {
+				layer := buffer.VideoLayer{
+					Spatial:  spatial,
+					Temporal: temporal,
+				}
+
+				usedChannelCapacity := track.ProvisionalAllocate(availableChannelCapacity, layer, s.allowPause, FlagAllowOvershootWhileDeficient)
+				if availableChannelCapacity < usedChannelCapacity {
+					break alloc_loop
+				}
+
+				bestLayer = layer
+			}
+		}
+
+		if bestLayer.IsValid() {
+			// found layer that can fit in available headroom
+			update := NewStreamStateUpdate()
+			allocation := track.ProvisionalAllocateCommit()
+			updateStreamStateChange(track, allocation, update)
+			s.maybeSendUpdate(update)
+
+			s.adjustState()
+			return
+		}
+
+		track.ProvisionalAllocateReset()
+		transition = track.ProvisionalAllocateGetCooperativeTransition(FlagAllowOvershootWhileDeficient) // get transition again to reset above allocation attempt using available headroom
+	}
+
+	// if there is not enough headroom, try to redistribute starting with tracks that are closest to their desired.
 	bandwidthAcquired := int64(0)
 	var contributingTracks []*Track
 
@@ -910,9 +944,7 @@ func (s *StreamAllocator) allocateTrack(track *Track) {
 		// commit the tracks that contributed
 		for _, t := range contributingTracks {
 			allocation := t.ProvisionalAllocateCommit()
-			if allocation.PauseReason == sfu.VideoPauseReasonBandwidth && track.SetPaused(true) {
-				update.HandleStreamingChange(true, t)
-			}
+			updateStreamStateChange(t, allocation, update)
 		}
 
 		// STREAM-ALLOCATOR-TODO if got too much extra, can potentially give it to some deficient track
@@ -921,9 +953,11 @@ func (s *StreamAllocator) allocateTrack(track *Track) {
 	// commit the track that needs change if enough could be acquired or pause not allowed
 	if !s.allowPause || bandwidthAcquired >= transition.BandwidthDelta {
 		allocation := track.ProvisionalAllocateCommit()
-		if allocation.PauseReason == sfu.VideoPauseReasonBandwidth && track.SetPaused(true) {
-			update.HandleStreamingChange(true, track)
-		}
+		updateStreamStateChange(track, allocation, update)
+	} else {
+		// explicitly pause to ensure stream state update happens if a track coming out of mute cannot be allocated
+		allocation := track.Pause()
+		updateStreamStateChange(track, allocation, update)
 	}
 
 	s.maybeSendUpdate(update)
@@ -967,16 +1001,7 @@ func (s *StreamAllocator) onProbeDone(isNotFailing bool, isGoalReached bool) {
 }
 
 func (s *StreamAllocator) maybeBoostDeficientTracks() {
-	committedChannelCapacity := s.committedChannelCapacity
-	if s.params.Config.MinChannelCapacity > committedChannelCapacity {
-		committedChannelCapacity = s.params.Config.MinChannelCapacity
-		s.params.Logger.Debugw(
-			"stream allocator: overriding channel capacity",
-			"actual", s.committedChannelCapacity,
-			"override", committedChannelCapacity,
-		)
-	}
-	availableChannelCapacity := committedChannelCapacity - s.getExpectedBandwidthUsage()
+	availableChannelCapacity := s.getAvailableHeadroom(false)
 	if availableChannelCapacity <= 0 {
 		return
 	}
@@ -989,9 +1014,7 @@ func (s *StreamAllocator) maybeBoostDeficientTracks() {
 			continue
 		}
 
-		if allocation.PauseReason == sfu.VideoPauseReasonBandwidth && track.SetPaused(true) {
-			update.HandleStreamingChange(true, track)
-		}
+		updateStreamStateChange(track, allocation, update)
 
 		availableChannelCapacity -= allocation.BandwidthDelta
 		if availableChannelCapacity <= 0 {
@@ -1024,23 +1047,7 @@ func (s *StreamAllocator) allocateAllTracks() {
 	//
 	update := NewStreamStateUpdate()
 
-	availableChannelCapacity := s.committedChannelCapacity
-	if s.params.Config.MinChannelCapacity > availableChannelCapacity {
-		availableChannelCapacity = s.params.Config.MinChannelCapacity
-		s.params.Logger.Debugw(
-			"stream allocator: overriding channel capacity with min channel capacity",
-			"actual", s.committedChannelCapacity,
-			"override", availableChannelCapacity,
-		)
-	}
-	if s.overriddenChannelCapacity > 0 {
-		availableChannelCapacity = s.overriddenChannelCapacity
-		s.params.Logger.Debugw(
-			"stream allocator: overriding channel capacity",
-			"actual", s.committedChannelCapacity,
-			"override", availableChannelCapacity,
-		)
-	}
+	availableChannelCapacity := s.getAvailableChannelCapacity(true)
 
 	//
 	// This pass is to find out if there is any leftover channel capacity after allocating exempt tracks.
@@ -1053,9 +1060,7 @@ func (s *StreamAllocator) allocateAllTracks() {
 		}
 
 		allocation := track.AllocateOptimal(FlagAllowOvershootExemptTrackWhileDeficient)
-		if allocation.PauseReason == sfu.VideoPauseReasonBandwidth && track.SetPaused(true) {
-			update.HandleStreamingChange(true, track)
-		}
+		updateStreamStateChange(track, allocation, update)
 
 		// STREAM-ALLOCATOR-TODO: optimistic allocation before bitrate is available will return 0. How to account for that?
 		availableChannelCapacity -= allocation.BandwidthRequested
@@ -1072,9 +1077,7 @@ func (s *StreamAllocator) allocateAllTracks() {
 			}
 
 			allocation := track.Pause()
-			if allocation.PauseReason == sfu.VideoPauseReasonBandwidth && track.SetPaused(true) {
-				update.HandleStreamingChange(true, track)
-			}
+			updateStreamStateChange(track, allocation, update)
 		}
 	} else {
 		sorted := s.getSorted()
@@ -1101,9 +1104,7 @@ func (s *StreamAllocator) allocateAllTracks() {
 
 		for _, track := range sorted {
 			allocation := track.ProvisionalAllocateCommit()
-			if allocation.PauseReason == sfu.VideoPauseReasonBandwidth && track.SetPaused(true) {
-				update.HandleStreamingChange(true, track)
-			}
+			updateStreamStateChange(track, allocation, update)
 		}
 	}
 
@@ -1132,6 +1133,28 @@ func (s *StreamAllocator) maybeSendUpdate(update *StreamStateUpdate) {
 	}
 }
 
+func (s *StreamAllocator) getAvailableChannelCapacity(allowOverride bool) int64 {
+	availableChannelCapacity := s.committedChannelCapacity
+	if s.params.Config.MinChannelCapacity > availableChannelCapacity {
+		availableChannelCapacity = s.params.Config.MinChannelCapacity
+		s.params.Logger.Debugw(
+			"stream allocator: overriding channel capacity with min channel capacity",
+			"actual", s.committedChannelCapacity,
+			"override", availableChannelCapacity,
+		)
+	}
+	if allowOverride && s.overriddenChannelCapacity > 0 {
+		availableChannelCapacity = s.overriddenChannelCapacity
+		s.params.Logger.Debugw(
+			"stream allocator: overriding channel capacity",
+			"actual", s.committedChannelCapacity,
+			"override", availableChannelCapacity,
+		)
+	}
+
+	return availableChannelCapacity
+}
+
 func (s *StreamAllocator) getExpectedBandwidthUsage() int64 {
 	expected := int64(0)
 	for _, track := range s.getTracks() {
@@ -1141,6 +1164,10 @@ func (s *StreamAllocator) getExpectedBandwidthUsage() int64 {
 	return expected
 }
 
+func (s *StreamAllocator) getAvailableHeadroom(allowOverride bool) int64 {
+	return s.getAvailableChannelCapacity(allowOverride) - s.getExpectedBandwidthUsage()
+}
+
 func (s *StreamAllocator) getNackDelta() (uint32, uint32) {
 	aggPacketDelta := uint32(0)
 	aggRepeatedNackDelta := uint32(0)
@@ -1154,11 +1181,23 @@ func (s *StreamAllocator) getNackDelta() (uint32, uint32) {
 }
 
 func (s *StreamAllocator) newChannelObserverProbe() *ChannelObserver {
-	return NewChannelObserver(ChannelObserverParamsProbe, s.params.Logger)
+	return NewChannelObserver(
+		ChannelObserverParams{
+			Name:   "probe",
+			Config: s.params.Config.ChannelObserverProbeConfig,
+		},
+		s.params.Logger,
+	)
 }
 
 func (s *StreamAllocator) newChannelObserverNonProbe() *ChannelObserver {
-	return NewChannelObserver(ChannelObserverParamsNonProbe, s.params.Logger)
+	return NewChannelObserver(
+		ChannelObserverParams{
+			Name:   "non-probe",
+			Config: s.params.Config.ChannelObserverNonProbeConfig,
+		},
+		s.params.Logger,
+	)
 }
 
 func (s *StreamAllocator) initProbe(probeGoalDeltaBps int64) {
@@ -1225,9 +1264,7 @@ func (s *StreamAllocator) maybeProbeWithMedia() {
 		}
 
 		update := NewStreamStateUpdate()
-		if allocation.PauseReason == sfu.VideoPauseReasonBandwidth && track.SetPaused(true) {
-			update.HandleStreamingChange(true, track)
-		}
+		updateStreamStateChange(track, allocation, update)
 		s.maybeSendUpdate(update)
 
 		s.probeController.Reset()
@@ -1354,3 +1391,26 @@ func (s *StreamAllocator) getTracksHistory() map[livekit.TrackID]string {
 }
 
 // ------------------------------------------------
+
+func updateStreamStateChange(track *Track, allocation sfu.VideoAllocation, update *StreamStateUpdate) {
+	updated := false
+	streamState := StreamStateInactive
+	switch allocation.PauseReason {
+	case sfu.VideoPauseReasonMuted:
+		fallthrough
+
+	case sfu.VideoPauseReasonPubMuted:
+		streamState = StreamStateInactive
+		updated = track.SetStreamState(streamState)
+
+	case sfu.VideoPauseReasonBandwidth:
+		streamState = StreamStatePaused
+		updated = track.SetStreamState(streamState)
+	}
+
+	if updated {
+		update.HandleStreamingChange(track, streamState)
+	}
+}
+
+// ------------------------------------------------

pkg/sfu/streamallocator/streamstateupdate.go

@@ -1,6 +1,22 @@
+// Copyright 2023 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 package streamallocator
 
 import (
+	"fmt"
+
 	"github.com/livekit/protocol/livekit"
 )
 
@@ -9,18 +25,21 @@ import (
 type StreamState int
 
 const (
-	StreamStateActive StreamState = iota
+	StreamStateInactive StreamState = iota
+	StreamStateActive
 	StreamStatePaused
 )
 
 func (s StreamState) String() string {
 	switch s {
+	case StreamStateInactive:
+		return "INACTIVE"
 	case StreamStateActive:
-		return "active"
+		return "ACTIVE"
 	case StreamStatePaused:
-		return "paused"
+		return "PAUSED"
 	default:
-		return "unknown"
+		return fmt.Sprintf("UNKNOWN: %d", int(s))
 	}
 }
 
@@ -40,19 +59,22 @@ func NewStreamStateUpdate() *StreamStateUpdate {
 	return &StreamStateUpdate{}
 }
 
-func (s *StreamStateUpdate) HandleStreamingChange(isPaused bool, track *Track) {
-	if isPaused {
-		s.StreamStates = append(s.StreamStates, &StreamStateInfo{
-			ParticipantID: track.PublisherID(),
-			TrackID:       track.ID(),
-			State:         StreamStatePaused,
-		})
-	} else {
+func (s *StreamStateUpdate) HandleStreamingChange(track *Track, streamState StreamState) {
+	switch streamState {
+	case StreamStateInactive:
+		// inactive is not a notification, could get into this state because of mute
+	case StreamStateActive:
 		s.StreamStates = append(s.StreamStates, &StreamStateInfo{
 			ParticipantID: track.PublisherID(),
 			TrackID:       track.ID(),
 			State:         StreamStateActive,
 		})
+	case StreamStatePaused:
+		s.StreamStates = append(s.StreamStates, &StreamStateInfo{
+			ParticipantID: track.PublisherID(),
+			TrackID:       track.ID(),
+			State:         StreamStatePaused,
+		})
 	}
 }
 

pkg/sfu/streamallocator/track.go

@@ -1,3 +1,17 @@
+// Copyright 2023 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 package streamallocator
 
 import (
@@ -42,7 +56,7 @@ type Track struct {
 
 	isDirty bool
 
-	isPaused bool
+	streamState StreamState
 }
 
 func NewTrack(
@@ -61,7 +75,7 @@ func NewTrack(
 		nackInfos:             make(map[uint16]sfu.NackInfo),
 		nackHistory:           make([]string, 0, 10),
 		receiverReportHistory: make([]string, 0, 10),
-		isPaused:              true,
+		streamState:           StreamStateInactive,
 	}
 	t.SetPriority(0)
 	t.SetMaxLayer(downTrack.MaxLayer())
@@ -78,12 +92,12 @@ func (t *Track) SetDirty(isDirty bool) bool {
 	return true
 }
 
-func (t *Track) SetPaused(isPaused bool) bool {
-	if t.isPaused == isPaused {
+func (t *Track) SetStreamState(streamState StreamState) bool {
+	if t.streamState == streamState {
 		return false
 	}
 
-	t.isPaused = isPaused
+	t.streamState = streamState
 	return true
 }
 
@@ -146,6 +160,10 @@ func (t *Track) ProvisionalAllocatePrepare() {
 	t.downTrack.ProvisionalAllocatePrepare()
 }
 
+func (t *Track) ProvisionalAllocateReset() {
+	t.downTrack.ProvisionalAllocateReset()
+}
+
 func (t *Track) ProvisionalAllocate(availableChannelCapacity int64, layer buffer.VideoLayer, allowPause bool, allowOvershoot bool) int64 {
 	return t.downTrack.ProvisionalAllocate(availableChannelCapacity, layer, allowPause, allowOvershoot)
 }

pkg/sfu/streamallocator/trenddetector.go

@@ -1,3 +1,17 @@
+// Copyright 2023 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 package streamallocator
 
 import (