diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index c6dbb34fd..e3dd65a04 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -42,6 +42,8 @@ type TrackSender interface { HandleRTCPSenderReportData(payloadType webrtc.PayloadType, layer int32, srData *buffer.RTCPSenderReportData) error } +// ------------------------------------------------------------------- + const ( RTPPaddingMaxPayloadSize = 255 RTPPaddingEstimatedHeaderSize = 20 @@ -60,6 +62,8 @@ const ( maxPaddingOnMuteDuration = 5 * time.Second ) +// ------------------------------------------------------------------- + var ( ErrUnknownKind = errors.New("unknown kind of codec") ErrOutOfOrderSequenceNumberCacheMiss = errors.New("out-of-order sequence number not found in cache") @@ -197,14 +201,13 @@ type DownTrack struct { transceiver *webrtc.RTPTransceiver writeStream webrtc.TrackLocalWriter rtcpReader *buffer.RTCPReader - onCloseHandler func(willBeResumed bool) - onBinding func(error) listenerLock sync.RWMutex receiverReportListeners []ReceiverReportListener - bindLock sync.Mutex - bound atomic.Bool + bindLock sync.Mutex + bound atomic.Bool + onBinding func(error) isClosed atomic.Bool connected atomic.Bool @@ -235,14 +238,13 @@ type DownTrack struct { pacer pacer.Pacer - // update stats - onStatsUpdate func(dt *DownTrack, stat *livekit.AnalyticsStat) + maxLayerNotifierCh chan struct{} - // when max subscribed layer changes + cbMu sync.RWMutex + onStatsUpdate func(dt *DownTrack, stat *livekit.AnalyticsStat) onMaxSubscribedLayerChanged func(dt *DownTrack, layer int32) - - // update rtt - onRttUpdate func(dt *DownTrack, rtt uint32) + onRttUpdate func(dt *DownTrack, rtt uint32) + onCloseHandler func(willBeResumed bool) } // NewDownTrack returns a DownTrack. @@ -266,17 +268,18 @@ func NewDownTrack( } d := &DownTrack{ - logger: logger, - id: r.TrackID(), - subscriberID: subID, - maxTrack: mt, - streamID: r.StreamID(), - bufferFactory: bf, - receiver: r, - upstreamCodecs: codecs, - kind: kind, - codec: codecs[0].RTPCodecCapability, - pacer: pacer, + logger: logger, + id: r.TrackID(), + subscriberID: subID, + maxTrack: mt, + streamID: r.StreamID(), + bufferFactory: bf, + receiver: r, + upstreamCodecs: codecs, + kind: kind, + codec: codecs[0].RTPCodecCapability, + pacer: pacer, + maxLayerNotifierCh: make(chan struct{}, 20), } d.forwarder = NewForwarder( d.kind, @@ -307,11 +310,15 @@ func NewDownTrack( Logger: d.logger.WithValues("direction", "down"), }) d.connectionStats.OnStatsUpdate(func(_cs *connectionquality.ConnectionStats, stat *livekit.AnalyticsStat) { - if d.onStatsUpdate != nil { - d.onStatsUpdate(d, stat) + if onStatsUpdate := d.getOnStatsUpdate(); onStatsUpdate != nil { + onStatsUpdate(d, stat) } }) + if d.kind == webrtc.RTPCodecTypeVideo { + go d.maxLayerNotifierWorker() + } + return d, nil } @@ -541,6 +548,7 @@ func (d *DownTrack) keyFrameRequester(generation uint32, layer int32) { if d.IsClosed() || layer == buffer.InvalidLayerSpatial { return } + interval := 2 * d.rtpStats.GetRtt() if interval < keyFrameIntervalMin { interval = keyFrameIntervalMin @@ -550,7 +558,13 @@ func (d *DownTrack) keyFrameRequester(generation uint32, layer int32) { } ticker := time.NewTicker(time.Duration(interval) * time.Millisecond) defer ticker.Stop() + for { + locked, _ := d.forwarder.CheckSync() + if locked { + return + } + if d.connected.Load() { d.logger.Debugw("sending PLI for layer lock", "generation", generation, "layer", layer) d.receiver.SendPLI(layer, false) @@ -565,6 +579,34 @@ func (d *DownTrack) keyFrameRequester(generation uint32, layer int32) { } } +func (d *DownTrack) postMaxLayerNotifierEvent() { + if d.IsClosed() { + return + } + + select { + case d.maxLayerNotifierCh <- struct{}{}: + default: + d.logger.Warnw("max layer notifier event queue full", nil) + } +} + +func (d *DownTrack) maxLayerNotifierWorker() { + more := true + for more { + _, more = <-d.maxLayerNotifierCh + + maxLayerSpatial := buffer.InvalidLayerSpatial + if more { + maxLayerSpatial = d.forwarder.GetMaxSubscribedSpatial() + } + if onMaxSubscribedLayerChanged := d.getOnMaxLayerChanged(); onMaxSubscribedLayerChanged != nil { + d.logger.Infow("max subscribed layer changed", maxLayerSpatial) + onMaxSubscribedLayerChanged(d, maxLayerSpatial) + } + } +} + // WriteRTP writes an RTP Packet to the DownTrack func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) error { if !d.bound.Load() || !d.connected.Load() { @@ -725,17 +767,17 @@ func (d *DownTrack) WritePaddingRTP(bytesToSend int, paddingOnMute bool, forceMa // Mute enables or disables media forwarding - subscriber triggered func (d *DownTrack) Mute(muted bool) { - changed, maxLayer := d.forwarder.Mute(muted) - d.handleMute(muted, false, changed, maxLayer) + changed := d.forwarder.Mute(muted) + d.handleMute(muted, changed) } // PubMute enables or disables media forwarding - publisher side func (d *DownTrack) PubMute(pubMuted bool) { - changed, maxLayer := d.forwarder.PubMute(pubMuted) - d.handleMute(pubMuted, true, changed, maxLayer) + changed := d.forwarder.PubMute(pubMuted) + d.handleMute(pubMuted, changed) } -func (d *DownTrack) handleMute(muted bool, isPub bool, changed bool, maxLayer buffer.VideoLayer) { +func (d *DownTrack) handleMute(muted bool, changed bool) { if !changed { return } @@ -762,18 +804,7 @@ func (d *DownTrack) handleMute(muted bool, isPub bool, changed bool, maxLayer bu // Note that while publisher mute is active, subscriber changes can also happen // and that could turn on/off layers on publisher side. // - if !isPub && d.onMaxSubscribedLayerChanged != nil && d.kind == webrtc.RTPCodecTypeVideo { - notifyLayer := buffer.InvalidLayerSpatial - if !muted { - // - // When unmuting, don't wait for layer lock as - // client might need to be notified to start layers - // before locking can happen in the forwarder. - // - notifyLayer = maxLayer.Spatial - } - d.onMaxSubscribedLayerChanged(d, notifyLayer) - } + d.postMaxLayerNotifierEvent() if sal := d.getStreamAllocatorListener(); sal != nil { sal.OnSubscriptionChanged(d) @@ -856,12 +887,10 @@ func (d *DownTrack) CloseWithFlush(flush bool) { d.rtpStats.Stop() d.logger.Infow("rtp stats", "direction", "downstream", "mime", d.mime, "ssrc", d.ssrc, "stats", d.rtpStats.ToString()) - if d.onMaxSubscribedLayerChanged != nil && d.kind == webrtc.RTPCodecTypeVideo { - d.onMaxSubscribedLayerChanged(d, buffer.InvalidLayerSpatial) - } + close(d.maxLayerNotifierCh) - if d.onCloseHandler != nil { - d.onCloseHandler(!flush) + if onCloseHandler := d.getOnCloseHandler(); onCloseHandler != nil { + onCloseHandler(!flush) } d.stopKeyFrameRequester() @@ -869,21 +898,12 @@ func (d *DownTrack) CloseWithFlush(flush bool) { } func (d *DownTrack) SetMaxSpatialLayer(spatialLayer int32) { - changed, maxLayer, currentLayer := d.forwarder.SetMaxSpatialLayer(spatialLayer) + changed, maxLayer := d.forwarder.SetMaxSpatialLayer(spatialLayer) if !changed { return } - if d.onMaxSubscribedLayerChanged != nil && d.kind == webrtc.RTPCodecTypeVideo && maxLayer.SpatialGreaterThanOrEqual(currentLayer) { - // - // Notify when new max is - // 1. Equal to current -> already locked to the new max - // 2. Greater than current -> two scenarios - // a. is higher than previous max -> client may need to start higher layer before forwarder can lock - // b. is lower than previous max -> client can stop higher layer(s) - // - d.onMaxSubscribedLayerChanged(d, maxLayer.Spatial) - } + d.postMaxLayerNotifierEvent() if sal := d.getStreamAllocatorListener(); sal != nil { sal.OnSubscribedLayerChanged(d, maxLayer) @@ -891,7 +911,7 @@ func (d *DownTrack) SetMaxSpatialLayer(spatialLayer int32) { } func (d *DownTrack) SetMaxTemporalLayer(temporalLayer int32) { - changed, maxLayer, _ := d.forwarder.SetMaxTemporalLayer(temporalLayer) + changed, maxLayer := d.forwarder.SetMaxTemporalLayer(temporalLayer) if !changed { return } @@ -973,10 +993,23 @@ func (d *DownTrack) UpTrackBitrateReport(availableLayers []int32, bitrates Bitra // OnCloseHandler method to be called on remote tracked removed func (d *DownTrack) OnCloseHandler(fn func(willBeResumed bool)) { + d.cbMu.Lock() + defer d.cbMu.Unlock() + d.onCloseHandler = fn } +func (d *DownTrack) getOnCloseHandler() func(willBeResumed bool) { + d.cbMu.RLock() + defer d.cbMu.RUnlock() + + return d.onCloseHandler +} + func (d *DownTrack) OnBinding(fn func(error)) { + d.bindLock.Lock() + defer d.bindLock.Unlock() + d.onBinding = fn } @@ -988,17 +1021,47 @@ func (d *DownTrack) AddReceiverReportListener(listener ReceiverReportListener) { } func (d *DownTrack) OnStatsUpdate(fn func(dt *DownTrack, stat *livekit.AnalyticsStat)) { + d.cbMu.Lock() + defer d.cbMu.Unlock() + d.onStatsUpdate = fn } +func (d *DownTrack) getOnStatsUpdate() func(dt *DownTrack, stat *livekit.AnalyticsStat) { + d.cbMu.RLock() + defer d.cbMu.RUnlock() + + return d.onStatsUpdate +} + func (d *DownTrack) OnRttUpdate(fn func(dt *DownTrack, rtt uint32)) { + d.cbMu.Lock() + defer d.cbMu.Unlock() + d.onRttUpdate = fn } +func (d *DownTrack) getOnRttUpdate() func(dt *DownTrack, rtt uint32) { + d.cbMu.RLock() + defer d.cbMu.RUnlock() + + return d.onRttUpdate +} + func (d *DownTrack) OnMaxLayerChanged(fn func(dt *DownTrack, layer int32)) { + d.cbMu.Lock() + defer d.cbMu.Unlock() + d.onMaxSubscribedLayerChanged = fn } +func (d *DownTrack) getOnMaxLayerChanged() func(dt *DownTrack, layer int32) { + d.cbMu.RLock() + defer d.cbMu.RUnlock() + + return d.onMaxSubscribedLayerChanged +} + func (d *DownTrack) IsDeficient() bool { return d.forwarder.IsDeficient() } @@ -1355,8 +1418,8 @@ func (d *DownTrack) handleRTCP(bytes []byte) { d.sequencer.setRTT(rttToReport) } - if d.onRttUpdate != nil { - d.onRttUpdate(d, rttToReport) + if onRttUpdate := d.getOnRttUpdate(); onRttUpdate != nil { + onRttUpdate(d, rttToReport) } } } @@ -1744,15 +1807,8 @@ func (d *DownTrack) packetSent(md interface{}, hdr *rtp.Header, payloadSize int, } if spmd.tp != nil { - if spmd.tp.isSwitchingToMaxSpatial && d.onMaxSubscribedLayerChanged != nil && d.kind == webrtc.RTPCodecTypeVideo { - d.onMaxSubscribedLayerChanged(d, spmd.tp.maxSpatialLayer) - } - - if spmd.tp.isSwitchingToRequestSpatial { - locked, _ := d.forwarder.CheckSync() - if locked { - d.stopKeyFrameRequester() - } + if spmd.tp.isSwitching { + d.postMaxLayerNotifierEvent() } if spmd.tp.isResuming { diff --git a/pkg/sfu/forwarder.go b/pkg/sfu/forwarder.go index 1da413ee8..23a8d694d 100644 --- a/pkg/sfu/forwarder.go +++ b/pkg/sfu/forwarder.go @@ -129,15 +129,13 @@ func (v VideoTransition) String() string { // ------------------------------------------------------------------- type TranslationParams struct { - shouldDrop bool - isResuming bool - isSwitchingToRequestSpatial bool - isSwitchingToMaxSpatial bool - maxSpatialLayer int32 - rtp *TranslationParamsRTP - codecBytes []byte - ddBytes []byte - marker bool + shouldDrop bool + isResuming bool + isSwitching bool + rtp *TranslationParamsRTP + codecBytes []byte + ddBytes []byte + marker bool } // ------------------------------------------------------------------- @@ -360,12 +358,12 @@ func (f *Forwarder) SeedState(state ForwarderState) { f.refTSOffset = state.RefTSOffset } -func (f *Forwarder) Mute(muted bool) (bool, buffer.VideoLayer) { +func (f *Forwarder) Mute(muted bool) bool { f.lock.Lock() defer f.lock.Unlock() if f.muted == muted { - return false, f.vls.GetMax() + return false } // Do not mute when paused due to bandwidth limitation. @@ -384,7 +382,7 @@ func (f *Forwarder) Mute(muted bool) (bool, buffer.VideoLayer) { // the case of intentional mute. if muted && f.isDeficientLocked() && f.lastAllocation.PauseReason == VideoPauseReasonBandwidth { f.logger.Infow("ignoring forwarder mute, paused due to congestion") - return false, f.vls.GetMax() + return false } f.logger.Debugw("setting forwarder mute", "muted", muted) @@ -395,7 +393,7 @@ func (f *Forwarder) Mute(muted bool) (bool, buffer.VideoLayer) { f.resyncLocked() } - return true, f.vls.GetMax() + return true } func (f *Forwarder) IsMuted() bool { @@ -405,12 +403,12 @@ func (f *Forwarder) IsMuted() bool { return f.muted } -func (f *Forwarder) PubMute(pubMuted bool) (bool, buffer.VideoLayer) { +func (f *Forwarder) PubMute(pubMuted bool) bool { f.lock.Lock() defer f.lock.Unlock() if f.pubMuted == pubMuted { - return false, f.vls.GetMax() + return false } f.logger.Debugw("setting forwarder pub mute", "pubMuted", pubMuted) @@ -432,7 +430,7 @@ func (f *Forwarder) PubMute(pubMuted bool) (bool, buffer.VideoLayer) { } } - return true, f.vls.GetMax() + return true } func (f *Forwarder) IsPubMuted() bool { @@ -449,17 +447,17 @@ func (f *Forwarder) IsAnyMuted() bool { return f.muted || f.pubMuted } -func (f *Forwarder) SetMaxSpatialLayer(spatialLayer int32) (bool, buffer.VideoLayer, buffer.VideoLayer) { +func (f *Forwarder) SetMaxSpatialLayer(spatialLayer int32) (bool, buffer.VideoLayer) { f.lock.Lock() defer f.lock.Unlock() if f.kind == webrtc.RTPCodecTypeAudio { - return false, buffer.InvalidLayer, buffer.InvalidLayer + return false, buffer.InvalidLayer } existingMax := f.vls.GetMax() if spatialLayer == existingMax.Spatial { - return false, existingMax, f.vls.GetCurrent() + return false, existingMax } f.logger.Debugw("setting max spatial layer", "layer", spatialLayer) @@ -467,20 +465,20 @@ func (f *Forwarder) SetMaxSpatialLayer(spatialLayer int32) (bool, buffer.VideoLa f.clearParkedLayer() - return true, f.vls.GetMax(), f.vls.GetCurrent() + return true, f.vls.GetMax() } -func (f *Forwarder) SetMaxTemporalLayer(temporalLayer int32) (bool, buffer.VideoLayer, buffer.VideoLayer) { +func (f *Forwarder) SetMaxTemporalLayer(temporalLayer int32) (bool, buffer.VideoLayer) { f.lock.Lock() defer f.lock.Unlock() if f.kind == webrtc.RTPCodecTypeAudio { - return false, buffer.InvalidLayer, buffer.InvalidLayer + return false, buffer.InvalidLayer } existingMax := f.vls.GetMax() if temporalLayer == existingMax.Temporal { - return false, existingMax, f.vls.GetCurrent() + return false, existingMax } f.logger.Debugw("setting max temporal layer", "layer", temporalLayer) @@ -488,7 +486,7 @@ func (f *Forwarder) SetMaxTemporalLayer(temporalLayer int32) (bool, buffer.Video f.clearParkedLayer() - return true, f.vls.GetMax(), f.vls.GetCurrent() + return true, f.vls.GetMax() } func (f *Forwarder) MaxLayer() buffer.VideoLayer { @@ -512,6 +510,25 @@ func (f *Forwarder) TargetLayer() buffer.VideoLayer { return f.vls.GetTarget() } +func (f *Forwarder) GetMaxSubscribedSpatial() int32 { + f.lock.RLock() + defer f.lock.RUnlock() + + layer := buffer.InvalidLayerSpatial // covers muted case + if !f.muted { + layer = f.vls.GetMax().Spatial + + // If current is higher, mark the current layer as max subscribed layer + // to prevent the current layer from stopping before forwarder switches + // to the new and lower max layer, + if layer < f.vls.GetCurrent().Spatial { + layer = f.vls.GetCurrent().Spatial + } + } + + return layer +} + func (f *Forwarder) isDeficientLocked() bool { return f.lastAllocation.IsDeficient } @@ -1690,9 +1707,7 @@ func (f *Forwarder) getTranslationParamsVideo(extPkt *buffer.ExtPacket, layer in return tp, nil } tp.isResuming = result.IsResuming - tp.isSwitchingToRequestSpatial = result.IsSwitchingToRequestSpatial - tp.isSwitchingToMaxSpatial = result.IsSwitchingToMaxSpatial - tp.maxSpatialLayer = result.MaxSpatialLayer + tp.isSwitching = result.IsSwitching tp.ddBytes = result.DependencyDescriptorExtension tp.marker = result.RTPMarker diff --git a/pkg/sfu/forwarder_test.go b/pkg/sfu/forwarder_test.go index a0e340800..2da1c524b 100644 --- a/pkg/sfu/forwarder_test.go +++ b/pkg/sfu/forwarder_test.go @@ -26,13 +26,13 @@ func newForwarder(codec webrtc.RTPCodecCapability, kind webrtc.RTPCodecType) *Fo func TestForwarderMute(t *testing.T) { f := newForwarder(testutils.TestOpusCodec, webrtc.RTPCodecTypeAudio) require.False(t, f.IsMuted()) - muted, _ := f.Mute(false) + muted := f.Mute(false) require.False(t, muted) // no change in mute state require.False(t, f.IsMuted()) - muted, _ = f.Mute(true) + muted = f.Mute(true) require.True(t, muted) require.True(t, f.IsMuted()) - muted, _ = f.Mute(false) + muted = f.Mute(false) require.True(t, muted) require.False(t, f.IsMuted()) } @@ -45,15 +45,13 @@ func TestForwarderLayersAudio(t *testing.T) { require.Equal(t, buffer.InvalidLayer, f.CurrentLayer()) require.Equal(t, buffer.InvalidLayer, f.TargetLayer()) - changed, maxLayer, currentLayer := f.SetMaxSpatialLayer(1) + changed, maxLayer := f.SetMaxSpatialLayer(1) require.False(t, changed) require.Equal(t, buffer.InvalidLayer, maxLayer) - require.Equal(t, buffer.InvalidLayer, currentLayer) - changed, maxLayer, currentLayer = f.SetMaxTemporalLayer(1) + changed, maxLayer = f.SetMaxTemporalLayer(1) require.False(t, changed) require.Equal(t, buffer.InvalidLayer, maxLayer) - require.Equal(t, buffer.InvalidLayer, currentLayer) require.Equal(t, buffer.InvalidLayer, f.MaxLayer()) } @@ -72,12 +70,11 @@ func TestForwarderLayersVideo(t *testing.T) { Spatial: buffer.DefaultMaxLayerSpatial, Temporal: buffer.DefaultMaxLayerTemporal, } - changed, maxLayer, currentLayer := f.SetMaxSpatialLayer(buffer.DefaultMaxLayerSpatial) + changed, maxLayer := f.SetMaxSpatialLayer(buffer.DefaultMaxLayerSpatial) require.True(t, changed) require.Equal(t, expectedLayers, maxLayer) - require.Equal(t, buffer.InvalidLayer, currentLayer) - changed, maxLayer, currentLayer = f.SetMaxSpatialLayer(buffer.DefaultMaxLayerSpatial - 1) + changed, maxLayer = f.SetMaxSpatialLayer(buffer.DefaultMaxLayerSpatial - 1) require.True(t, changed) expectedLayers = buffer.VideoLayer{ Spatial: buffer.DefaultMaxLayerSpatial - 1, @@ -85,21 +82,18 @@ func TestForwarderLayersVideo(t *testing.T) { } require.Equal(t, expectedLayers, maxLayer) require.Equal(t, expectedLayers, f.MaxLayer()) - require.Equal(t, buffer.InvalidLayer, currentLayer) f.vls.SetCurrent(buffer.VideoLayer{Spatial: 0, Temporal: 1}) - changed, maxLayer, currentLayer = f.SetMaxSpatialLayer(buffer.DefaultMaxLayerSpatial - 1) + changed, maxLayer = f.SetMaxSpatialLayer(buffer.DefaultMaxLayerSpatial - 1) require.False(t, changed) require.Equal(t, expectedLayers, maxLayer) require.Equal(t, expectedLayers, f.MaxLayer()) - require.Equal(t, buffer.VideoLayer{Spatial: 0, Temporal: 1}, currentLayer) - changed, maxLayer, currentLayer = f.SetMaxTemporalLayer(buffer.DefaultMaxLayerTemporal) + changed, maxLayer = f.SetMaxTemporalLayer(buffer.DefaultMaxLayerTemporal) require.False(t, changed) require.Equal(t, expectedLayers, maxLayer) - require.Equal(t, buffer.VideoLayer{Spatial: 0, Temporal: 1}, currentLayer) - changed, maxLayer, currentLayer = f.SetMaxTemporalLayer(buffer.DefaultMaxLayerTemporal - 1) + changed, maxLayer = f.SetMaxTemporalLayer(buffer.DefaultMaxLayerTemporal - 1) require.True(t, changed) expectedLayers = buffer.VideoLayer{ Spatial: buffer.DefaultMaxLayerSpatial - 1, @@ -107,7 +101,6 @@ func TestForwarderLayersVideo(t *testing.T) { } require.Equal(t, expectedLayers, maxLayer) require.Equal(t, expectedLayers, f.MaxLayer()) - require.Equal(t, buffer.VideoLayer{Spatial: 0, Temporal: 1}, currentLayer) } func TestForwarderAllocateOptimal(t *testing.T) { @@ -1404,8 +1397,8 @@ func TestForwarderGetTranslationParamsVideo(t *testing.T) { marshalledVP8, err := expectedVP8.Marshal() require.NoError(t, err) expectedTP = TranslationParams{ - isSwitchingToMaxSpatial: true, - isResuming: true, + isSwitching: true, + isResuming: true, rtp: &TranslationParamsRTP{ snOrdering: SequenceNumberOrderingContiguous, sequenceNumber: 23333, @@ -1716,8 +1709,7 @@ func TestForwarderGetTranslationParamsVideo(t *testing.T) { marshalledVP8, err = expectedVP8.Marshal() require.NoError(t, err) expectedTP = TranslationParams{ - isSwitchingToMaxSpatial: true, - maxSpatialLayer: 1, + isSwitching: true, rtp: &TranslationParamsRTP{ snOrdering: SequenceNumberOrderingContiguous, sequenceNumber: 23339, diff --git a/pkg/sfu/streamallocator/streamallocator.go b/pkg/sfu/streamallocator/streamallocator.go index 475b38e94..f0c17db9e 100644 --- a/pkg/sfu/streamallocator/streamallocator.go +++ b/pkg/sfu/streamallocator/streamallocator.go @@ -548,6 +548,10 @@ func (s *StreamAllocator) postEvent(event Event) { func (s *StreamAllocator) processEvents() { for event := range s.eventCh { + if s.isStopped.Load() { + break + } + s.handleEvent(&event) } diff --git a/pkg/sfu/videolayerselector/dependencydescriptor.go b/pkg/sfu/videolayerselector/dependencydescriptor.go index 7b8c95310..34f8b0b55 100644 --- a/pkg/sfu/videolayerselector/dependencydescriptor.go +++ b/pkg/sfu/videolayerselector/dependencydescriptor.go @@ -204,26 +204,6 @@ func (d *DependencyDescriptor) Select(extPkt *buffer.ExtPacket, _layer int32) (r d.previousActiveDecodeTargetsBitmask = d.activeDecodeTargetsBitmask d.activeDecodeTargetsBitmask = buffer.GetActiveDecodeTargetBitmask(d.currentLayer, ddwdt.DecodeTargets) - - if d.currentLayer.Spatial == d.requestSpatial { - result.IsSwitchingToRequestSpatial = true - } - if d.currentLayer.Spatial == d.maxLayer.Spatial { - result.IsSwitchingToMaxSpatial = true - result.MaxSpatialLayer = d.currentLayer.Spatial - d.logger.Infow( - "reached max layer", - "previous", d.previousLayer, - "current", d.currentLayer, - "previousTarget", d.previousTargetLayer, - "target", d.targetLayer, - "max", d.maxLayer, - "layer", fd.SpatialId, - "req", d.requestSpatial, - "maxSeen", d.maxSeenLayer, - "feed", extPkt.Packet.SSRC, - ) - } } ddExtension := &dede.DependencyDescriptorExtension{ diff --git a/pkg/sfu/videolayerselector/simulcast.go b/pkg/sfu/videolayerselector/simulcast.go index 4d0c9294d..06e0bad72 100644 --- a/pkg/sfu/videolayerselector/simulcast.go +++ b/pkg/sfu/videolayerselector/simulcast.go @@ -35,19 +35,6 @@ func (s *Simulcast) Select(extPkt *buffer.ExtPacket, layer int32) (result VideoL result.IsResuming = true } - if s.currentLayer.Spatial == s.requestSpatial { - result.IsSwitchingToRequestSpatial = true - } - - if s.currentLayer.Spatial >= s.maxLayer.Spatial { - result.IsSwitchingToMaxSpatial = true - result.MaxSpatialLayer = s.currentLayer.Spatial - if reason != "" { - reason += ", " - } - reason += "reached max layer" - } - if reason != "" { s.logger.Infow( reason, diff --git a/pkg/sfu/videolayerselector/videolayerselector.go b/pkg/sfu/videolayerselector/videolayerselector.go index ffbb9f42c..f17d745d1 100644 --- a/pkg/sfu/videolayerselector/videolayerselector.go +++ b/pkg/sfu/videolayerselector/videolayerselector.go @@ -10,9 +10,6 @@ type VideoLayerSelectorResult struct { IsRelevant bool IsSwitching bool IsResuming bool - IsSwitchingToRequestSpatial bool - IsSwitchingToMaxSpatial bool - MaxSpatialLayer int32 RTPMarker bool DependencyDescriptorExtension []byte } diff --git a/pkg/sfu/videolayerselector/vp9.go b/pkg/sfu/videolayerselector/vp9.go index ed1a165b6..508cdf289 100644 --- a/pkg/sfu/videolayerselector/vp9.go +++ b/pkg/sfu/videolayerselector/vp9.go @@ -80,26 +80,6 @@ func (v *VP9) Select(extPkt *buffer.ExtPacket, _layer int32) (result VideoLayerS result.IsResuming = true } - if v.currentLayer.Spatial != v.requestSpatial && updatedLayer.Spatial == v.requestSpatial { - result.IsSwitchingToRequestSpatial = true - } - - if v.currentLayer.Spatial != v.maxLayer.Spatial && updatedLayer.Spatial == v.maxLayer.Spatial { - result.IsSwitchingToMaxSpatial = true - result.MaxSpatialLayer = updatedLayer.Spatial - v.logger.Infow( - "reached max layer", - "current", v.currentLayer, - "updated", updatedLayer, - "target", v.targetLayer, - "max", v.maxLayer, - "layer", extPkt.VideoLayer.Spatial, - "req", v.requestSpatial, - "maxSeen", v.maxSeenLayer, - "feed", extPkt.Packet.SSRC, - ) - } - v.previousLayer = v.currentLayer v.currentLayer = updatedLayer }