From e22f47b41649c508afff751045cac0f9ab4c769c Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Tue, 1 Feb 2022 13:24:42 +0530 Subject: [PATCH] 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 | 4 +- pkg/sfu/forwarder.go | 6 +- pkg/sfu/forwarder_test.go | 30 ++--- pkg/sfu/receiver.go | 200 ++++++------------------------- pkg/sfu/streamtracker.go | 68 +++++------ pkg/sfu/streamtrackermanager.go | 201 ++++++++++++++++++++++++++++++++ 6 files changed, 280 insertions(+), 229 deletions(-) create mode 100644 pkg/sfu/streamtrackermanager.go diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index 8b823751b..f5e25651e 100644 --- a/pkg/sfu/downtrack.go +++ b/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 { diff --git a/pkg/sfu/forwarder.go b/pkg/sfu/forwarder.go index b107813c6..544e0d98e 100644 --- a/pkg/sfu/forwarder.go +++ b/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() diff --git a/pkg/sfu/forwarder_test.go b/pkg/sfu/forwarder_test.go index 11b4bae18..71b015fc4 100644 --- a/pkg/sfu/forwarder_test.go +++ b/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, diff --git a/pkg/sfu/receiver.go b/pkg/sfu/receiver.go index 09a6a6f28..7832beae0 100644 --- a/pkg/sfu/receiver.go +++ b/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 { diff --git a/pkg/sfu/streamtracker.go b/pkg/sfu/streamtracker.go index 10ec76b94..29e7471fd 100644 --- a/pkg/sfu/streamtracker.go +++ b/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 } diff --git a/pkg/sfu/streamtrackermanager.go b/pkg/sfu/streamtrackermanager.go new file mode 100644 index 000000000..2900d4c60 --- /dev/null +++ b/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) + } +}