diff --git a/pkg/sfu/buffer/buffer.go b/pkg/sfu/buffer/buffer.go index 26fb22aca..361162b37 100644 --- a/pkg/sfu/buffer/buffer.go +++ b/pkg/sfu/buffer/buffer.go @@ -567,7 +567,7 @@ func (b *Buffer) getExtPacket(rtpPacket *rtp.Packet, arrivalTime int64) *ExtPack } else { // vp8 with DependencyDescriptor enabled, use the TID from the descriptor vp8Packet.TID = uint8(ep.Temporal) - ep.Spatial = InvalidLayerSpatial // vp8 don't have spatial scalability, reset to -1 + ep.Spatial = InvalidLayerSpatial // vp8 don't have spatial scalability, reset to invalid } ep.Payload = vp8Packet case "video/vp9": diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index ccd68e5ca..256f5defc 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -130,8 +130,11 @@ type DownTrackStreamAllocatorListener interface { // video layer bitrate availability changed OnBitrateAvailabilityChanged(dt *DownTrack) - // max published video layer changed - OnMaxPublishedLayerChanged(dt *DownTrack) + // max published spatial layer changed + OnMaxPublishedSpatialChanged(dt *DownTrack) + + // max published temporal layer changed + OnMaxPublishedTemporalChanged(dt *DownTrack) // subscription changed - mute/unmute OnSubscriptionChanged(dt *DownTrack) @@ -593,7 +596,7 @@ func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) error { d.streamAllocatorBytesCounter.Add(uint32(hdr.MarshalSize() + len(payload))) - if tp.isSwitchingToMaxLayer && d.onMaxSubscribedLayerChanged != nil && d.kind == webrtc.RTPCodecTypeVideo { + if tp.isSwitchingToMaxSpatial && d.onMaxSubscribedLayerChanged != nil && d.kind == webrtc.RTPCodecTypeVideo { d.onMaxSubscribedLayerChanged(d, layer) } @@ -601,8 +604,9 @@ func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) error { d.isNACKThrottled.Store(false) d.rtpStats.UpdateKeyFrame(1) d.logger.Debugw("forwarding key frame", "layer", layer) + } - // SVC-TODO - no need for key frame always when using SVC + if tp.isSwitchingToRequestSpatial { locked, _ := d.forwarder.CheckSync() if locked { d.stopKeyFrameRequester() @@ -921,15 +925,19 @@ func (d *DownTrack) UpTrackBitrateAvailabilityChange() { } func (d *DownTrack) UpTrackMaxPublishedLayerChange(maxPublishedLayer int32) { - d.forwarder.SetMaxPublishedLayer(maxPublishedLayer) - - if sal := d.getStreamAllocatorListener(); sal != nil { - sal.OnMaxPublishedLayerChanged(d) + if d.forwarder.SetMaxPublishedLayer(maxPublishedLayer) { + if sal := d.getStreamAllocatorListener(); sal != nil { + sal.OnMaxPublishedSpatialChanged(d) + } } } func (d *DownTrack) UpTrackMaxTemporalLayerSeenChange(maxTemporalLayerSeen int32) { - d.forwarder.SetMaxTemporalLayerSeen(maxTemporalLayerSeen) + if d.forwarder.SetMaxTemporalLayerSeen(maxTemporalLayerSeen) { + if sal := d.getStreamAllocatorListener(); sal != nil { + sal.OnMaxPublishedTemporalChanged(d) + } + } } func (d *DownTrack) maybeAddTransition(_bitrate int64, distance float64) { diff --git a/pkg/sfu/forwarder.go b/pkg/sfu/forwarder.go index 8d75db2f9..a0d419ecc 100644 --- a/pkg/sfu/forwarder.go +++ b/pkg/sfu/forwarder.go @@ -123,13 +123,14 @@ func (v VideoTransition) String() string { // ------------------------------------------------------------------- type TranslationParams struct { - shouldDrop bool - isResuming bool - isSwitchingToMaxLayer bool - rtp *TranslationParamsRTP - codecBytes []byte - ddBytes []byte - marker bool + shouldDrop bool + isResuming bool + isSwitchingToRequestSpatial bool + isSwitchingToMaxSpatial bool + rtp *TranslationParamsRTP + codecBytes []byte + ddBytes []byte + marker bool } // ------------------------------------------------------------------- @@ -202,30 +203,32 @@ func NewForwarder( return f } -func (f *Forwarder) SetMaxPublishedLayer(maxPublishedLayer int32) { +func (f *Forwarder) SetMaxPublishedLayer(maxPublishedLayer int32) bool { f.lock.Lock() defer f.lock.Unlock() existingMaxSeen := f.vls.GetMaxSeen() if maxPublishedLayer <= existingMaxSeen.Spatial { - return + return false } f.vls.SetMaxSeenSpatial(maxPublishedLayer) f.logger.Debugw("setting max published layer", "maxPublishedLayer", maxPublishedLayer) + return true } -func (f *Forwarder) SetMaxTemporalLayerSeen(maxTemporalLayerSeen int32) { +func (f *Forwarder) SetMaxTemporalLayerSeen(maxTemporalLayerSeen int32) bool { f.lock.Lock() defer f.lock.Unlock() existingMaxSeen := f.vls.GetMaxSeen() if maxTemporalLayerSeen <= existingMaxSeen.Temporal { - return + return false } f.vls.SetMaxSeenTemporal(maxTemporalLayerSeen) f.logger.Debugw("setting max temporal layer seen", "maxTemporalLayerSeen", maxTemporalLayerSeen) + return true } func (f *Forwarder) OnParkedLayerExpired(fn func()) { @@ -536,15 +539,24 @@ func (f *Forwarder) AllocateOptimal(availableLayers []int32, brs Bitrates, allow } alloc.BandwidthNeeded = optimalBandwidthNeeded + getMaxTemporal := func() int32 { + maxTemporal := maxLayer.Temporal + if maxSeenLayer.Temporal != buffer.InvalidLayerTemporal && maxSeenLayer.Temporal < maxTemporal { + maxTemporal = maxSeenLayer.Temporal + } + return maxTemporal + } + opportunisticAlloc := func() { // opportunistically latch on to anything maxSpatial := maxLayer.Spatial if allowOvershoot && f.vls.IsOvershootOkay() && maxSeenLayer.Spatial > maxSpatial { maxSpatial = maxSeenLayer.Spatial } + alloc.TargetLayer = buffer.VideoLayer{ Spatial: int32(math.Min(float64(maxSeenLayer.Spatial), float64(maxSpatial))), - Temporal: maxLayer.Temporal, + Temporal: getMaxTemporal(), } } @@ -599,7 +611,7 @@ func (f *Forwarder) AllocateOptimal(availableLayers []int32, brs Bitrates, allow alloc.TargetLayer.Spatial = l } } - alloc.TargetLayer.Temporal = maxLayer.Temporal + alloc.TargetLayer.Temporal = getMaxTemporal() alloc.RequestLayerSpatial = alloc.TargetLayer.Spatial } else { @@ -608,7 +620,7 @@ func (f *Forwarder) AllocateOptimal(availableLayers []int32, brs Bitrates, allow // current is locked to desired, stay there alloc.TargetLayer = buffer.VideoLayer{ Spatial: requestSpatial, - Temporal: maxLayer.Temporal, + Temporal: getMaxTemporal(), } alloc.RequestLayerSpatial = requestSpatial } else { @@ -1487,10 +1499,11 @@ func (f *Forwarder) getTranslationParamsVideo(extPkt *buffer.ExtPacket, layer in } return tp, nil } - tp.isSwitchingToMaxLayer = result.IsSwitchingToMaxSpatial tp.isResuming = result.IsResuming - tp.marker = result.RTPMarker + tp.isSwitchingToRequestSpatial = result.IsSwitchingToRequestSpatial + tp.isSwitchingToMaxSpatial = result.IsSwitchingToMaxSpatial tp.ddBytes = result.DependencyDescriptorExtension + tp.marker = result.RTPMarker if FlagPauseOnDowngrade && f.isDeficientLocked() && f.vls.GetTarget().Spatial < f.vls.GetCurrent().Spatial { // diff --git a/pkg/sfu/forwarder_test.go b/pkg/sfu/forwarder_test.go index e2f3dff44..46a86edf2 100644 --- a/pkg/sfu/forwarder_test.go +++ b/pkg/sfu/forwarder_test.go @@ -1409,8 +1409,8 @@ func TestForwarderGetTranslationParamsVideo(t *testing.T) { marshalledVP8, err := expectedVP8.Marshal() require.NoError(t, err) expectedTP = TranslationParams{ - isSwitchingToMaxLayer: true, - isResuming: true, + isSwitchingToMaxSpatial: true, + isResuming: true, rtp: &TranslationParamsRTP{ snOrdering: SequenceNumberOrderingContiguous, sequenceNumber: 23333, @@ -1721,7 +1721,7 @@ func TestForwarderGetTranslationParamsVideo(t *testing.T) { marshalledVP8, err = expectedVP8.Marshal() require.NoError(t, err) expectedTP = TranslationParams{ - isSwitchingToMaxLayer: true, + isSwitchingToMaxSpatial: true, rtp: &TranslationParamsRTP{ snOrdering: SequenceNumberOrderingContiguous, sequenceNumber: 23339, diff --git a/pkg/sfu/receiver.go b/pkg/sfu/receiver.go index 0868a574e..2e38540d3 100644 --- a/pkg/sfu/receiver.go +++ b/pkg/sfu/receiver.go @@ -626,10 +626,10 @@ func (w *WebRTCReceiver) forwardRTP(layer int32) { return } - // svc packet, dispatch to correct tracker spatialTracker := tracker spatialLayer := layer if pkt.Spatial >= 0 { + // svc packet, dispatch to correct tracker spatialLayer = pkt.Spatial spatialTracker = w.streamTrackerManager.GetTracker(pkt.Spatial) if spatialTracker == nil { diff --git a/pkg/sfu/streamallocator/streamallocator.go b/pkg/sfu/streamallocator/streamallocator.go index c1b64fe62..bc7b51276 100644 --- a/pkg/sfu/streamallocator/streamallocator.go +++ b/pkg/sfu/streamallocator/streamallocator.go @@ -386,8 +386,13 @@ func (s *StreamAllocator) OnBitrateAvailabilityChanged(downTrack *sfu.DownTrack) s.maybePostEventAllocateTrack(downTrack) } -// called when feeding track's max publisher layer changes -func (s *StreamAllocator) OnMaxPublishedLayerChanged(downTrack *sfu.DownTrack) { +// called when feeding track's max published spatial layer changes +func (s *StreamAllocator) OnMaxPublishedSpatialChanged(downTrack *sfu.DownTrack) { + s.maybePostEventAllocateTrack(downTrack) +} + +// called when feeding track's max published temporal layer changes +func (s *StreamAllocator) OnMaxPublishedTemporalChanged(downTrack *sfu.DownTrack) { s.maybePostEventAllocateTrack(downTrack) } diff --git a/pkg/sfu/videolayerselector/dependencydescriptor.go b/pkg/sfu/videolayerselector/dependencydescriptor.go index 8b39049ce..09d46515e 100644 --- a/pkg/sfu/videolayerselector/dependencydescriptor.go +++ b/pkg/sfu/videolayerselector/dependencydescriptor.go @@ -124,11 +124,13 @@ func (d *DependencyDescriptor) Select(extPkt *buffer.ExtPacket, _layer int32) (r if dti == dd.DecodeTargetSwitch { // dependency descriptor decode target switch is enabled at all potential switch points. // So, setting current layer on every switch point will change current layer a lot. - // currentLayer is not needed for layer selection in this selector. + // + // However `currentLayer` is not needed for layer selection in this selector. // But, it is needed to signal things in the selector checks outside of this selector. + // // The following cases are handled - // 1. To detect resumption - so by setting it to incoming layer only when current lqyer is invalid, that is taken care of - // 2. To detect target achieved - set current to target if it is not the same + // 1. To detect resumption + // 2. To detect target achieved so that key frame requests can be stopped // 3. To detect reaching max spatial layer - checked when current hits target if !d.currentLayer.IsValid() { result.IsResuming = true @@ -153,9 +155,13 @@ func (d *DependencyDescriptor) Select(extPkt *buffer.ExtPacket, _layer int32) (r if d.currentLayer != d.targetLayer { if d.currentLayer.Spatial != d.targetLayer.Spatial && int32(extPkt.DependencyDescriptor.FrameDependencies.SpatialId) == d.targetLayer.Spatial { d.currentLayer.Spatial = d.targetLayer.Spatial + + if d.currentLayer.Spatial == d.requestSpatial { + result.IsSwitchingToRequestSpatial = true + } + if d.currentLayer.Spatial == d.maxLayer.Spatial { result.IsSwitchingToMaxSpatial = true - d.logger.Infow( "reached max layer", "current", d.currentLayer, diff --git a/pkg/sfu/videolayerselector/simulcast.go b/pkg/sfu/videolayerselector/simulcast.go index ec875e11e..fc15c4db0 100644 --- a/pkg/sfu/videolayerselector/simulcast.go +++ b/pkg/sfu/videolayerselector/simulcast.go @@ -89,10 +89,15 @@ func (s *Simulcast) Select(extPkt *buffer.ExtPacket, layer int32) (result VideoL if !isActive { result.IsResuming = true } + s.SetParked(buffer.InvalidLayer) + + if s.currentLayer.Spatial == s.requestSpatial { + result.IsSwitchingToRequestSpatial = true + } + if s.currentLayer.Spatial >= s.maxLayer.Spatial { result.IsSwitchingToMaxSpatial = true - s.logger.Infow( "reached max layer", "current", s.currentLayer, @@ -112,27 +117,29 @@ func (s *Simulcast) Select(extPkt *buffer.ExtPacket, layer int32) (result VideoL } // if locked to higher than max layer due to overshoot, check if it can be dialed back - if s.currentLayer.Spatial > s.maxLayer.Spatial { - if layer <= s.maxLayer.Spatial && extPkt.KeyFrame { - s.logger.Infow( - "adjusting overshoot", - "current", s.currentLayer, - "target", s.targetLayer, - "max", s.maxLayer, - "layer", layer, - "req", s.requestSpatial, - "maxSeen", s.maxSeenLayer, - "feed", extPkt.Packet.SSRC, - ) - s.currentLayer.Spatial = layer + if s.currentLayer.Spatial > s.maxLayer.Spatial && layer <= s.maxLayer.Spatial && extPkt.KeyFrame { + s.logger.Infow( + "adjusting overshoot", + "current", s.currentLayer, + "target", s.targetLayer, + "max", s.maxLayer, + "layer", layer, + "req", s.requestSpatial, + "maxSeen", s.maxSeenLayer, + "feed", extPkt.Packet.SSRC, + ) + s.currentLayer.Spatial = layer - if s.currentLayer.Spatial >= s.maxLayer.Spatial { - result.IsSwitchingToMaxSpatial = true - } + if s.currentLayer.Spatial == s.requestSpatial { + result.IsSwitchingToRequestSpatial = true + } - if s.currentLayer.Spatial >= s.maxLayer.Spatial || s.currentLayer.Spatial == s.maxSeenLayer.Spatial { - s.targetLayer.Spatial = layer - } + if s.currentLayer.Spatial >= s.maxLayer.Spatial { + result.IsSwitchingToMaxSpatial = true + } + + if s.currentLayer.Spatial >= s.maxLayer.Spatial || s.currentLayer.Spatial == s.maxSeenLayer.Spatial { + s.targetLayer.Spatial = layer } } diff --git a/pkg/sfu/videolayerselector/temporallayerselector/vp8.go b/pkg/sfu/videolayerselector/temporallayerselector/vp8.go index fbf697f33..a83765526 100644 --- a/pkg/sfu/videolayerselector/temporallayerselector/vp8.go +++ b/pkg/sfu/videolayerselector/temporallayerselector/vp8.go @@ -34,8 +34,8 @@ func (v *VP8) Select(extPkt *buffer.ExtPacket, current int32, target int32) (thi next = tid } } else { - if tid < current && tid >= target && extPkt.Packet.Marker { - next = tid + if extPkt.Packet.Marker { + next = target } } return diff --git a/pkg/sfu/videolayerselector/videolayerselector.go b/pkg/sfu/videolayerselector/videolayerselector.go index 548f736dc..14229e182 100644 --- a/pkg/sfu/videolayerselector/videolayerselector.go +++ b/pkg/sfu/videolayerselector/videolayerselector.go @@ -9,6 +9,7 @@ type VideoLayerSelectorResult struct { IsSelected bool IsRelevant bool IsResuming bool + IsSwitchingToRequestSpatial bool IsSwitchingToMaxSpatial bool RTPMarker bool DependencyDescriptorExtension []byte diff --git a/pkg/sfu/videolayerselector/vp9.go b/pkg/sfu/videolayerselector/vp9.go index 0e7783d98..9f7533352 100644 --- a/pkg/sfu/videolayerselector/vp9.go +++ b/pkg/sfu/videolayerselector/vp9.go @@ -42,29 +42,34 @@ func (v *VP9) Select(extPkt *buffer.ExtPacket, _layer int32) (result VideoLayerS } updatedLayer = extPkt.VideoLayer - currentLayer = extPkt.VideoLayer } else { - // temporal scale up/down - if v.currentLayer.Temporal < v.targetLayer.Temporal { - if extPkt.VideoLayer.Temporal > v.currentLayer.Temporal && extPkt.VideoLayer.Temporal <= v.targetLayer.Temporal && vp9.U && vp9.B { - updatedLayer.Temporal = extPkt.VideoLayer.Temporal - currentLayer.Temporal = extPkt.VideoLayer.Temporal - } - } else { - if extPkt.VideoLayer.Temporal < v.currentLayer.Temporal && extPkt.VideoLayer.Temporal >= v.targetLayer.Temporal && vp9.E { - updatedLayer.Temporal = extPkt.VideoLayer.Temporal + if v.currentLayer.Temporal != v.targetLayer.Temporal { + if v.currentLayer.Temporal < v.targetLayer.Temporal { + // temporal scale up + if extPkt.VideoLayer.Temporal > v.currentLayer.Temporal && extPkt.VideoLayer.Temporal <= v.targetLayer.Temporal && vp9.U && vp9.B { + currentLayer.Temporal = extPkt.VideoLayer.Temporal + updatedLayer.Temporal = extPkt.VideoLayer.Temporal + } + } else { + // temporal scale down + if vp9.E { + updatedLayer.Temporal = v.targetLayer.Temporal + } } } - // spatial scale up/down - if v.currentLayer.Spatial < v.targetLayer.Spatial { - if extPkt.VideoLayer.Spatial > v.currentLayer.Spatial && extPkt.VideoLayer.Spatial <= v.targetLayer.Spatial && !vp9.P && vp9.B { - updatedLayer.Spatial = extPkt.VideoLayer.Spatial - currentLayer.Spatial = extPkt.VideoLayer.Spatial - } - } else { - if extPkt.VideoLayer.Spatial < v.currentLayer.Spatial && extPkt.VideoLayer.Spatial >= v.targetLayer.Spatial && vp9.E { - updatedLayer.Spatial = extPkt.VideoLayer.Spatial + if v.currentLayer.Spatial != v.targetLayer.Spatial { + if v.currentLayer.Spatial < v.targetLayer.Spatial { + // spatial scale up + if extPkt.VideoLayer.Spatial > v.currentLayer.Spatial && extPkt.VideoLayer.Spatial <= v.targetLayer.Spatial && !vp9.P && vp9.B { + currentLayer.Spatial = extPkt.VideoLayer.Spatial + updatedLayer.Spatial = extPkt.VideoLayer.Spatial + } + } else { + // spatial scale down + if vp9.E { + updatedLayer.Spatial = v.targetLayer.Spatial + } } } } @@ -74,6 +79,10 @@ 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 v.logger.Infow( @@ -93,7 +102,7 @@ func (v *VP9) Select(extPkt *buffer.ExtPacket, _layer int32) (result VideoLayerS } result.RTPMarker = extPkt.Packet.Marker - if extPkt.VideoLayer.Spatial == v.currentLayer.Spatial && vp9.E { + if vp9.E && extPkt.VideoLayer.Spatial == currentLayer.Spatial && (vp9.P || v.targetLayer.Spatial <= v.currentLayer.Spatial) { result.RTPMarker = true } result.IsSelected = !extPkt.VideoLayer.GreaterThan(currentLayer)