Split out stream tracker manager (#394)

* Random clean up

* Split out StreamTrackerManager for re-use

* Reset the correct tracker

* use generation counter to exit coroutine

* start only for video and when enabled

* Add RemoveAllTrackers method

pkg/sfu/downtrack.go

@@ -28,7 +28,7 @@ const (
 
 // TrackSender defines an interface send media to remote peer
 type TrackSender interface {
-	UpTrackLayersChange(availableLayers []uint16)
+	UpTrackLayersChange(availableLayers []int32)
 	WriteRTP(p *buffer.ExtPacket, layer int32) error
 	Close()
 	// ID is the globally unique identifier for this Track.
@@ -549,7 +549,7 @@ func (d *DownTrack) GetForwardingStatus() ForwardingStatus {
 	return d.forwarder.GetForwardingStatus()
 }
 
-func (d *DownTrack) UpTrackLayersChange(availableLayers []uint16) {
+func (d *DownTrack) UpTrackLayersChange(availableLayers []int32) {
 	d.forwarder.UpTrackLayersChange(availableLayers)
 
 	if d.onAvailableLayersChanged != nil {

pkg/sfu/forwarder.go

@@ -83,7 +83,7 @@ type VideoAllocation struct {
 	change             VideoStreamingChange
 	bandwidthRequested int64
 	bandwidthDelta     int64
-	availableLayers    []uint16
+	availableLayers    []int32
 	bitrates           Bitrates
 	targetLayers       VideoLayers
 	distanceToDesired  int32
@@ -182,7 +182,7 @@ type Forwarder struct {
 
 	lastAllocation VideoAllocation
 
-	availableLayers []uint16
+	availableLayers []int32
 
 	rtpMunger *RTPMunger
 	vp8Munger *VP8Munger
@@ -301,7 +301,7 @@ func (f *Forwarder) GetForwardingStatus() ForwardingStatus {
 	return ForwardingStatusOptimal
 }
 
-func (f *Forwarder) UpTrackLayersChange(availableLayers []uint16) {
+func (f *Forwarder) UpTrackLayersChange(availableLayers []int32) {
 	f.lock.Lock()
 	defer f.lock.Unlock()
 

pkg/sfu/forwarder_test.go

@@ -119,7 +119,7 @@ func TestForwarderGetForwardingStatus(t *testing.T) {
 	require.Equal(t, ForwardingStatusOptimal, f.GetForwardingStatus())
 
 	// with available layers, should be off
-	availableLayers := []uint16{0, 1, 2}
+	availableLayers := []int32{0, 1, 2}
 	f.UpTrackLayersChange(availableLayers)
 	require.Equal(t, ForwardingStatusOff, f.GetForwardingStatus())
 
@@ -137,7 +137,7 @@ func TestForwarderGetForwardingStatus(t *testing.T) {
 	require.Equal(t, ForwardingStatusPartial, f.GetForwardingStatus())
 
 	// when available layers are lower than max subscribed, optimal as long as target is at max available
-	availableLayers = []uint16{0, 1}
+	availableLayers = []int32{0, 1}
 	f.UpTrackLayersChange(availableLayers)
 	require.Equal(t, ForwardingStatusOptimal, f.GetForwardingStatus())
 }
@@ -147,15 +147,15 @@ func TestForwarderUpTrackLayersChange(t *testing.T) {
 
 	require.Nil(t, f.availableLayers)
 
-	availableLayers := []uint16{0, 1, 2}
+	availableLayers := []int32{0, 1, 2}
 	f.UpTrackLayersChange(availableLayers)
 	require.Equal(t, availableLayers, f.availableLayers)
 
-	availableLayers = []uint16{0, 2}
+	availableLayers = []int32{0, 2}
 	f.UpTrackLayersChange(availableLayers)
 	require.Equal(t, availableLayers, f.availableLayers)
 
-	availableLayers = []uint16{}
+	availableLayers = []int32{}
 	f.UpTrackLayersChange(availableLayers)
 	require.Equal(t, availableLayers, f.availableLayers)
 }
@@ -208,7 +208,7 @@ func TestForwarderAllocate(t *testing.T) {
 	// awaiting measurement, i.e. bitrates are not available, but layers available
 	f.lastAllocation.state = VideoAllocationStateNone
 	disable(f)
-	f.UpTrackLayersChange([]uint16{0})
+	f.UpTrackLayersChange([]int32{0})
 	expectedTargetLayers := VideoLayers{
 		spatial:  0,
 		temporal: DefaultMaxLayerTemporal,
@@ -218,7 +218,7 @@ func TestForwarderAllocate(t *testing.T) {
 		change:             VideoStreamingChangeResuming,
 		bandwidthRequested: 0,
 		bandwidthDelta:     0,
-		availableLayers:    []uint16{0},
+		availableLayers:    []int32{0},
 		bitrates:           emptyBitrates,
 		targetLayers:       expectedTargetLayers,
 		distanceToDesired:  0,
@@ -235,7 +235,7 @@ func TestForwarderAllocate(t *testing.T) {
 		change:             VideoStreamingChangePausing,
 		bandwidthRequested: 0,
 		bandwidthDelta:     0,
-		availableLayers:    []uint16{0},
+		availableLayers:    []int32{0},
 		bitrates:           emptyBitrates,
 		targetLayers:       InvalidLayers,
 		distanceToDesired:  0,
@@ -256,7 +256,7 @@ func TestForwarderAllocate(t *testing.T) {
 		change:             VideoStreamingChangeResuming,
 		bandwidthRequested: 0,
 		bandwidthDelta:     0,
-		availableLayers:    []uint16{0},
+		availableLayers:    []int32{0},
 		bitrates:           emptyBitrates,
 		targetLayers:       expectedTargetLayers,
 		distanceToDesired:  0,
@@ -277,7 +277,7 @@ func TestForwarderAllocate(t *testing.T) {
 		change:             VideoStreamingChangeNone,
 		bandwidthRequested: bitrates[2][1],
 		bandwidthDelta:     bitrates[2][1],
-		availableLayers:    []uint16{0},
+		availableLayers:    []int32{0},
 		bitrates:           bitrates,
 		targetLayers:       expectedTargetLayers,
 		distanceToDesired:  0,
@@ -298,7 +298,7 @@ func TestForwarderAllocate(t *testing.T) {
 		change:             VideoStreamingChangeNone,
 		bandwidthRequested: bitrates[1][3],
 		bandwidthDelta:     bitrates[1][3] - bitrates[2][1],
-		availableLayers:    []uint16{0},
+		availableLayers:    []int32{0},
 		bitrates:           bitrates,
 		targetLayers:       expectedTargetLayers,
 		distanceToDesired:  1,
@@ -315,7 +315,7 @@ func TestForwarderAllocate(t *testing.T) {
 		change:             VideoStreamingChangePausing,
 		bandwidthRequested: 0,
 		bandwidthDelta:     0 - bitrates[1][3],
-		availableLayers:    []uint16{0},
+		availableLayers:    []int32{0},
 		bitrates:           bitrates,
 		targetLayers:       InvalidLayers,
 		distanceToDesired:  5,
@@ -336,7 +336,7 @@ func TestForwarderAllocate(t *testing.T) {
 		change:             VideoStreamingChangeResuming,
 		bandwidthRequested: bitrates[0][0],
 		bandwidthDelta:     bitrates[0][0],
-		availableLayers:    []uint16{0},
+		availableLayers:    []int32{0},
 		bitrates:           bitrates,
 		targetLayers:       expectedTargetLayers,
 		distanceToDesired:  4,
@@ -637,7 +637,7 @@ func TestForwarderFinalizeAllocate(t *testing.T) {
 	// layers available, but still awaiting measurement
 	f.lastAllocation.state = VideoAllocationStateAwaitingMeasurement
 	disable(f)
-	f.UpTrackLayersChange([]uint16{0, 1})
+	f.UpTrackLayersChange([]int32{0, 1})
 	expectedResult = VideoAllocation{
 		state:              VideoAllocationStateAwaitingMeasurement,
 		change:             VideoStreamingChangeNone,
@@ -668,7 +668,7 @@ func TestForwarderFinalizeAllocate(t *testing.T) {
 		change:             VideoStreamingChangeResuming,
 		bandwidthRequested: bitrates[1][3],
 		bandwidthDelta:     bitrates[1][3],
-		availableLayers:    []uint16{0, 1},
+		availableLayers:    []int32{0, 1},
 		bitrates:           bitrates,
 		targetLayers:       expectedTargetLayers,
 		distanceToDesired:  0,

pkg/sfu/receiver.go

@@ -4,7 +4,6 @@ import (
 	"io"
 	"math/rand"
 	"runtime"
-	"sort"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -46,20 +45,17 @@ type TrackReceiver interface {
 type WebRTCReceiver struct {
 	logger logger.Logger
 
-	peerID           livekit.ParticipantID
-	trackID          livekit.TrackID
-	streamID         string
-	kind             webrtc.RTPCodecType
-	receiver         *webrtc.RTPReceiver
-	codec            webrtc.RTPCodecParameters
-	isSimulcast      bool
-	availableLayers  atomic.Value
-	maxExpectedLayer int32
-	onCloseHandler   func()
-	closeOnce        sync.Once
-	closed           atomicBool
-	trackers         [DefaultMaxLayerSpatial + 1]*StreamTracker
-	useTrackers      bool
+	peerID         livekit.ParticipantID
+	trackID        livekit.TrackID
+	streamID       string
+	kind           webrtc.RTPCodecType
+	receiver       *webrtc.RTPReceiver
+	codec          webrtc.RTPCodecParameters
+	isSimulcast    bool
+	onCloseHandler func()
+	closeOnce      sync.Once
+	closed         atomicBool
+	useTrackers    bool
 
 	rtcpMu      sync.Mutex
 	rtcpCh      chan []rtcp.Packet
@@ -78,6 +74,8 @@ type WebRTCReceiver struct {
 	free        map[int]struct{}
 	numProcs    int
 	lbThreshold int
+
+	streamTrackerManager *StreamTrackerManager
 }
 
 func RidToLayer(rid string) int32 {
@@ -138,14 +136,15 @@ func NewWebRTCReceiver(
 		codec:    track.Codec(),
 		kind:     track.Kind(),
 		// LK-TODO: this should be based on VideoLayers protocol message rather than RID based
-		isSimulcast:      len(track.RID()) > 0,
-		maxExpectedLayer: DefaultMaxLayerSpatial,
-		pliThrottle:      500e6,
-		downTracks:       make([]TrackSender, 0),
-		index:            make(map[livekit.ParticipantID]int),
-		free:             make(map[int]struct{}),
-		numProcs:         runtime.NumCPU(),
+		isSimulcast:          len(track.RID()) > 0,
+		pliThrottle:          500e6,
+		downTracks:           make([]TrackSender, 0),
+		index:                make(map[livekit.ParticipantID]int),
+		free:                 make(map[int]struct{}),
+		numProcs:             runtime.NumCPU(),
+		streamTrackerManager: NewStreamTrackerManager(),
 	}
+	w.streamTrackerManager.OnAvailableLayersChanged(w.downTrackLayerChange)
 	if runtime.GOMAXPROCS(0) < w.numProcs {
 		w.numProcs = runtime.GOMAXPROCS(0)
 	}
@@ -203,7 +202,9 @@ func (w *WebRTCReceiver) AddUpTrack(track *webrtc.TrackRemote, buff *buffer.Buff
 	w.buffers[layer] = buff
 	w.bufferMu.Unlock()
 
-	w.setupTracker(layer)
+	if w.Kind() == webrtc.RTPCodecTypeVideo && w.useTrackers {
+		w.streamTrackerManager.AddTracker(layer)
+	}
 	go w.forwardRTP(layer)
 }
 
@@ -211,13 +212,7 @@ func (w *WebRTCReceiver) AddUpTrack(track *webrtc.TrackRemote, buff *buffer.Buff
 // this will reflect the "muted" status and will pause streamtracker to ensure we don't turn off
 // the layer
 func (w *WebRTCReceiver) SetUpTrackPaused(paused bool) {
-	w.upTrackMu.Lock()
-	defer w.upTrackMu.Unlock()
-	for _, tracker := range w.trackers {
-		if tracker != nil {
-			tracker.SetPaused(paused)
-		}
-	}
+	w.streamTrackerManager.SetPaused(paused)
 }
 
 func (w *WebRTCReceiver) AddDownTrack(track TrackSender) {
@@ -234,10 +229,8 @@ func (w *WebRTCReceiver) AddDownTrack(track TrackSender) {
 
 	if w.Kind() == webrtc.RTPCodecTypeVideo {
 		// notify added down track of available layers
-		w.upTrackMu.RLock()
-		layers, ok := w.availableLayers.Load().([]uint16)
-		w.upTrackMu.RUnlock()
-		if ok && len(layers) != 0 {
+		layers := w.streamTrackerManager.GetAvailableLayers()
+		if len(layers) != 0 {
 			track.UpTrackLayersChange(layers)
 		}
 	}
@@ -245,91 +238,15 @@ func (w *WebRTCReceiver) AddDownTrack(track TrackSender) {
 	w.storeDownTrack(track)
 }
 
-func (w *WebRTCReceiver) setupTracker(layer int32) {
-	w.upTrackMu.Lock()
-	defer w.upTrackMu.Unlock()
-
-	if w.Kind() != webrtc.RTPCodecTypeVideo || !w.useTrackers {
-		return
-	}
-
-	samplesRequired := uint32(5)
-	cyclesRequired := uint64(60) // 30s of continuous stream
-	if layer == 0 {
-		// be very forgiving for base layer
-		samplesRequired = 1
-		cyclesRequired = 4 // 2s of continuous stream
-	}
-	tracker := NewStreamTracker(samplesRequired, cyclesRequired, 500*time.Millisecond)
-	w.trackers[layer] = tracker
-	tracker.OnStatusChanged(func(status StreamStatus) {
-		if status == StreamStatusStopped {
-			w.removeAvailableLayer(uint16(layer))
-		} else {
-			w.addAvailableLayer(uint16(layer))
-		}
-	})
-	tracker.Start()
-}
-
-func (w *WebRTCReceiver) hasSpatialLayer(layer int32) bool {
-	layers, ok := w.availableLayers.Load().([]uint16)
-	if !ok {
-		return false
-	}
-	desired := uint16(layer)
-	for _, l := range layers {
-		if l == desired {
-			return true
-		}
-	}
-	return false
-}
-
 func (w *WebRTCReceiver) SetMaxExpectedSpatialLayer(layer int32) {
-	w.upTrackMu.Lock()
-	defer w.upTrackMu.Unlock()
-
-	if layer <= w.maxExpectedLayer {
-		// some higher layer(s) expected to stop, nothing else to do
-		w.maxExpectedLayer = layer
-		return
-	}
-
-	//
-	// Some higher layer is expected to start.
-	// If the layer was not stopped (i.e. it will still be in available layers),
-	// don't need to do anything. If not, reset the stream tracker so that
-	// the layer is declared available on the first packet
-	//
-	// NOTE: There may be a race between checking if a layer is available and
-	// resetting the tracker, i.e. the track may stop just after checking.
-	// But, those conditions should be rare. In those cases, the restart will
-	// take longer.
-	//
-	for l := w.maxExpectedLayer + 1; l <= layer; l++ {
-		if w.hasSpatialLayer(l) {
-			continue
-		}
-
-		tracker := w.trackers[l]
-		if tracker != nil {
-			tracker.Reset()
-		}
-	}
-	w.maxExpectedLayer = layer
+	w.streamTrackerManager.SetMaxExpectedSpatialLayer(layer)
 }
 
 func (w *WebRTCReceiver) NumAvailableSpatialLayers() int {
-	layers, ok := w.availableLayers.Load().([]uint16)
-	if !ok {
-		return 0
-	}
-
-	return len(layers)
+	return len(w.streamTrackerManager.GetAvailableLayers())
 }
 
-func (w *WebRTCReceiver) downTrackLayerChange(layers []uint16) {
+func (w *WebRTCReceiver) downTrackLayerChange(layers []int32) {
 	w.downTrackMu.RLock()
 	downTracks := w.downTracks
 	w.downTrackMu.RUnlock()
@@ -341,50 +258,6 @@ func (w *WebRTCReceiver) downTrackLayerChange(layers []uint16) {
 	}
 }
 
-func (w *WebRTCReceiver) addAvailableLayer(layer uint16) {
-	w.upTrackMu.Lock()
-	layers, ok := w.availableLayers.Load().([]uint16)
-	if !ok {
-		layers = []uint16{}
-	}
-	hasLayer := false
-	for _, l := range layers {
-		if l == layer {
-			hasLayer = true
-			break
-		}
-	}
-	if !hasLayer {
-		layers = append(layers, layer)
-	}
-	sort.Slice(layers, func(i, j int) bool { return layers[i] < layers[j] })
-	w.availableLayers.Store(layers)
-	w.upTrackMu.Unlock()
-
-	w.downTrackLayerChange(layers)
-}
-
-func (w *WebRTCReceiver) removeAvailableLayer(layer uint16) {
-	w.upTrackMu.Lock()
-	layers, ok := w.availableLayers.Load().([]uint16)
-	if !ok {
-		w.upTrackMu.Unlock()
-		return
-	}
-	newLayers := make([]uint16, 0, DefaultMaxLayerSpatial+1)
-	for _, l := range layers {
-		if l != layer {
-			newLayers = append(newLayers, l)
-		}
-	}
-	sort.Slice(newLayers, func(i, j int) bool { return newLayers[i] < newLayers[j] })
-	w.availableLayers.Store(newLayers)
-	w.upTrackMu.Unlock()
-
-	// need to immediately switch off unavailable layers
-	w.downTrackLayerChange(newLayers)
-}
-
 func (w *WebRTCReceiver) GetBitrateTemporalCumulative() Bitrates {
 	// LK-TODO: For SVC tracks, need to accumulate across spatial layers also
 	var br Bitrates
@@ -393,7 +266,7 @@ func (w *WebRTCReceiver) GetBitrateTemporalCumulative() Bitrates {
 	for i, buff := range w.buffers {
 		if buff != nil {
 			tls := make([]int64, DefaultMaxLayerTemporal+1)
-			if w.hasSpatialLayer(int32(i)) {
+			if w.streamTrackerManager.HasSpatialLayer(int32(i)) {
 				tls = buff.BitrateTemporalCumulative()
 			}
 
@@ -486,9 +359,7 @@ func (w *WebRTCReceiver) GetTotalBytes() uint64 {
 }
 
 func (w *WebRTCReceiver) forwardRTP(layer int32) {
-	w.upTrackMu.RLock()
-	tracker := w.trackers[layer]
-	w.upTrackMu.RUnlock()
+	tracker := w.streamTrackerManager.GetTracker(layer)
 
 	defer func() {
 		w.closeOnce.Do(func() {
@@ -496,12 +367,7 @@ func (w *WebRTCReceiver) forwardRTP(layer int32) {
 			w.closeTracks()
 		})
 
-		w.upTrackMu.Lock()
-		if tracker != nil {
-			tracker.Stop()
-			w.trackers[layer] = nil
-		}
-		w.upTrackMu.Unlock()
+		w.streamTrackerManager.RemoveTracker(layer)
 	}()
 
 	for {

pkg/sfu/streamtracker.go

@@ -4,6 +4,8 @@ import (
 	"sync"
 	"sync/atomic"
 	"time"
+
+	"github.com/livekit/protocol/utils"
 )
 
 type StreamStatus int32
@@ -37,7 +39,6 @@ type StreamTracker struct {
 
 	paused         atomicBool
 	countSinceLast uint32 // number of packets received since last check
-	running        chan struct{}
 	generation     atomicUint32
 
 	initMu      sync.Mutex
@@ -51,6 +52,8 @@ type StreamTracker struct {
 
 	// only access by the same goroutine as Observe
 	lastSN uint16
+
+	isStopped utils.AtomicFlag
 }
 
 func NewStreamTracker(samplesRequired uint32, cyclesRequired uint64, cycleDuration time.Duration) *StreamTracker {
@@ -74,41 +77,31 @@ func (s *StreamTracker) Status() StreamStatus {
 	return s.status
 }
 
-func (s *StreamTracker) maybeSetActive() {
+func (s *StreamTracker) maybeSetStatus(status StreamStatus) {
 	changed := false
 	s.statusMu.Lock()
-	if s.status != StreamStatusActive {
-		s.status = StreamStatusActive
+	if s.status != status {
+		s.status = status
 		changed = true
 	}
 	s.statusMu.Unlock()
 
 	if changed && s.onStatusChanged != nil {
-		s.onStatusChanged(StreamStatusActive)
+		s.onStatusChanged(status)
 	}
 }
 
-func (s *StreamTracker) maybeSetStopped() {
-	changed := false
-	s.statusMu.Lock()
-	if s.status != StreamStatusStopped {
-		s.status = StreamStatusStopped
-		changed = true
-	}
-	s.statusMu.Unlock()
+func (s *StreamTracker) maybeSetActive() {
+	s.maybeSetStatus(StreamStatusActive)
+}
 
-	if changed && s.onStatusChanged != nil {
-		s.onStatusChanged(StreamStatusStopped)
-	}
+func (s *StreamTracker) maybeSetStopped() {
+	s.maybeSetStatus(StreamStatusStopped)
 }
 
 func (s *StreamTracker) init() {
 	s.maybeSetActive()
 
-	if s.isRunning() {
-		return
-	}
-	s.running = make(chan struct{})
 	go s.detectWorker(s.generation.get())
 }
 
@@ -116,17 +109,23 @@ func (s *StreamTracker) Start() {
 }
 
 func (s *StreamTracker) Stop() {
-	if s.running != nil {
-		close(s.running)
-		s.running = nil
+	if !s.isStopped.TrySet(true) {
+		return
 	}
+
+	// bump generation to trigger exit of worker
+	s.generation.add(1)
 }
 
 func (s *StreamTracker) Reset() {
-	s.generation.add(1)
-	s.Stop()
+	if s.isStopped.Get() {
+		return
+	}
 
-	s.countSinceLast = 0
+	// bump generation to trigger exit of current worker
+	s.generation.add(1)
+
+	atomic.StoreUint32(&s.countSinceLast, 0)
 	s.cycleCount = 0
 
 	s.initMu.Lock()
@@ -142,18 +141,6 @@ func (s *StreamTracker) SetPaused(paused bool) {
 	s.paused.set(paused)
 }
 
-func (s *StreamTracker) isRunning() bool {
-	if s.running == nil {
-		return false
-	}
-	select {
-	case <-s.running:
-		return false
-	default:
-		return true
-	}
-}
-
 // Observe a packet that's received
 func (s *StreamTracker) Observe(sn uint16) {
 	if s.paused.get() {
@@ -187,11 +174,8 @@ func (s *StreamTracker) Observe(sn uint16) {
 func (s *StreamTracker) detectWorker(generation uint32) {
 	ticker := time.NewTicker(s.cycleDuration)
 
-	for s.isRunning() {
+	for {
 		<-ticker.C
-		if !s.isRunning() {
-			return
-		}
 		if generation != s.generation.get() {
 			return
 		}

pkg/sfu/streamtrackermanager.go

@@ -0,0 +1,201 @@
+package sfu
+
+import (
+	"sort"
+	"sync"
+	"time"
+)
+
+type StreamTrackerManager struct {
+	lock sync.RWMutex
+
+	trackers [DefaultMaxLayerSpatial + 1]*StreamTracker
+
+	availableLayers  []int32
+	maxExpectedLayer int32
+
+	onAvailableLayersChanged func(availableLayers []int32)
+}
+
+func NewStreamTrackerManager() *StreamTrackerManager {
+	return &StreamTrackerManager{
+		maxExpectedLayer: DefaultMaxLayerSpatial,
+	}
+}
+
+func (s *StreamTrackerManager) OnAvailableLayersChanged(f func(availableLayers []int32)) {
+	s.onAvailableLayersChanged = f
+}
+
+func (s *StreamTrackerManager) AddTracker(layer int32) {
+	samplesRequired := uint32(5)
+	cyclesRequired := uint64(60) // 30s of continuous stream
+	if layer == 0 {
+		// be very forgiving for base layer
+		samplesRequired = 1
+		cyclesRequired = 4 // 2s of continuous stream
+	}
+	tracker := NewStreamTracker(samplesRequired, cyclesRequired, 500*time.Millisecond)
+	tracker.OnStatusChanged(func(status StreamStatus) {
+		if status == StreamStatusStopped {
+			s.removeAvailableLayer(layer)
+		} else {
+			s.addAvailableLayer(layer)
+		}
+	})
+
+	s.lock.Lock()
+	s.trackers[layer] = tracker
+	s.lock.Unlock()
+
+	tracker.Start()
+}
+
+func (s *StreamTrackerManager) RemoveTracker(layer int32) {
+	s.lock.Lock()
+	tracker := s.trackers[layer]
+	s.trackers[layer] = nil
+	s.lock.Unlock()
+
+	if tracker != nil {
+		tracker.Stop()
+	}
+}
+
+func (s *StreamTrackerManager) RemoveAllTrackers() {
+	s.lock.Lock()
+	trackers := s.trackers
+	for layer := range s.trackers {
+		s.trackers[layer] = nil
+	}
+	s.lock.Unlock()
+
+	for _, tracker := range trackers {
+		if tracker != nil {
+			tracker.Stop()
+		}
+	}
+}
+
+func (s *StreamTrackerManager) GetTracker(layer int32) *StreamTracker {
+	s.lock.RLock()
+	defer s.lock.RUnlock()
+
+	return s.trackers[layer]
+}
+
+func (s *StreamTrackerManager) SetPaused(paused bool) {
+	s.lock.Lock()
+	trackers := s.trackers
+	s.lock.Unlock()
+
+	for _, tracker := range trackers {
+		if tracker != nil {
+			tracker.SetPaused(paused)
+		}
+	}
+}
+
+func (s *StreamTrackerManager) SetMaxExpectedSpatialLayer(layer int32) {
+	s.lock.Lock()
+	if layer <= s.maxExpectedLayer {
+		// some higher layer(s) expected to stop, nothing else to do
+		s.maxExpectedLayer = layer
+		s.lock.Unlock()
+		return
+	}
+
+	//
+	// Some higher layer is expected to start.
+	// If the layer was not stopped (i.e. it will still be in available layers),
+	// don't need to do anything. If not, reset the stream tracker so that
+	// the layer is declared available on the first packet
+	//
+	// NOTE: There may be a race between checking if a layer is available and
+	// resetting the tracker, i.e. the track may stop just after checking.
+	// But, those conditions should be rare. In those cases, the restart will
+	// take longer.
+	//
+	var trackersToReset []*StreamTracker
+	for l := s.maxExpectedLayer + 1; l <= layer; l++ {
+		if s.hasSpatialLayerLocked(l) {
+			continue
+		}
+
+		if s.trackers[l] != nil {
+			trackersToReset = append(trackersToReset, s.trackers[l])
+		}
+	}
+	s.maxExpectedLayer = layer
+	s.lock.Unlock()
+
+	for _, tracker := range trackersToReset {
+		tracker.Reset()
+	}
+}
+
+func (s *StreamTrackerManager) GetAvailableLayers() []int32 {
+	s.lock.RLock()
+	defer s.lock.RUnlock()
+
+	return s.availableLayers
+}
+
+func (s *StreamTrackerManager) HasSpatialLayer(layer int32) bool {
+	s.lock.RLock()
+	defer s.lock.RUnlock()
+
+	return s.hasSpatialLayerLocked(layer)
+}
+
+func (s *StreamTrackerManager) hasSpatialLayerLocked(layer int32) bool {
+	for _, l := range s.availableLayers {
+		if l == layer {
+			return true
+		}
+	}
+
+	return false
+}
+
+func (s *StreamTrackerManager) addAvailableLayer(layer int32) {
+	s.lock.Lock()
+	hasLayer := false
+	for _, l := range s.availableLayers {
+		if l == layer {
+			hasLayer = true
+			break
+		}
+	}
+	if hasLayer {
+		s.lock.Unlock()
+		return
+	}
+
+	s.availableLayers = append(s.availableLayers, layer)
+	sort.Slice(s.availableLayers, func(i, j int) bool { return s.availableLayers[i] < s.availableLayers[j] })
+	layers := s.availableLayers
+	s.lock.Unlock()
+
+	if s.onAvailableLayersChanged != nil {
+		s.onAvailableLayersChanged(layers)
+	}
+}
+
+func (s *StreamTrackerManager) removeAvailableLayer(layer int32) {
+	s.lock.Lock()
+	newLayers := make([]int32, 0, DefaultMaxLayerSpatial+1)
+	for _, l := range s.availableLayers {
+		if l != layer {
+			newLayers = append(newLayers, l)
+		}
+	}
+	sort.Slice(newLayers, func(i, j int) bool { return newLayers[i] < newLayers[j] })
+	s.availableLayers = newLayers
+	s.lock.Unlock()
+
+	// need to immediately switch off unavailable layers
+	if s.onAvailableLayersChanged != nil {
+		s.onAvailableLayersChanged(newLayers)
+	}
+}