diff --git a/pkg/sfu/codecmunger/codecmunger.go b/pkg/sfu/codecmunger/codecmunger.go index efab8d6b5..add55f333 100644 --- a/pkg/sfu/codecmunger/codecmunger.go +++ b/pkg/sfu/codecmunger/codecmunger.go @@ -23,8 +23,8 @@ import ( var ( ErrNotVP8 = errors.New("not VP8") ErrOutOfOrderVP8PictureIdCacheMiss = errors.New("out-of-order VP8 picture id not found in cache") - ErrFilteredVP8TemporalLayer = errors.New("filtered VP8 temporal layer") - ErrFilteredVP8OutOfOrder = errors.New("filtered out-of-order VP8 packet") + ErrFilteredTemporalLayer = errors.New("filtered temporal layer") + ErrFilteredOutOfOrder = errors.New("filtered out-of-order packet") ) // MaxHeaderSize is the largest codec header the mungers produce (VP8: 6 bytes). diff --git a/pkg/sfu/codecmunger/vp8dd.go b/pkg/sfu/codecmunger/ddtemporal.go similarity index 57% rename from pkg/sfu/codecmunger/vp8dd.go rename to pkg/sfu/codecmunger/ddtemporal.go index 97bea6df6..7fcbd116e 100644 --- a/pkg/sfu/codecmunger/vp8dd.go +++ b/pkg/sfu/codecmunger/ddtemporal.go @@ -17,20 +17,22 @@ package codecmunger import ( "github.com/livekit/livekit-server/pkg/sfu/buffer" "github.com/livekit/mediatransportutil/pkg/codec" + "github.com/livekit/protocol/codecs/mime" "github.com/livekit/protocol/logger" ) // frames with a remembered decision, behind the highest frame number -const vp8DDFrameWindow = 64 +const ddTemporalFrameWindow = 64 -// VP8DD filters VP8 temporal layers with the dependency descriptor. libwebrtc sends a -// one byte VP8 payload descriptor with the dependency descriptor, so there is no picture -// ID, TL0PICIDX or key index to munge, and the header goes out unchanged. Every packet -// names its frame and temporal layer, so the munger decides once per frame and applies -// that decision to all packets of the frame, a late one included. It needs the one byte -// descriptor: the forwarder uses the VP8 munger for a stream that keeps picture IDs. -type VP8DD struct { +// DDTemporal filters the temporal layers of a simulcast stream with the dependency +// descriptor, for any codec: it leaves the payload unchanged. Every packet names its frame +// and temporal layer, so the munger decides once per frame and applies that decision to all +// packets of the frame, a late one included. For VP8, libwebrtc sends a one byte payload +// descriptor with the dependency descriptor, so there is no picture ID to munge; the +// forwarder uses the VP8 munger for a stream that keeps picture IDs. +type DDTemporal struct { logger logger.Logger + isVP8 bool // padding needs a VP8 payload descriptor initialized bool highestFrame uint64 @@ -38,29 +40,30 @@ type VP8DD struct { forwardedMask uint64 // bit i: frame highestFrame - i is forwarded } -func NewVP8DD(logger logger.Logger) *VP8DD { - return &VP8DD{ +func NewDDTemporal(mimeType mime.MimeType, logger logger.Logger) *DDTemporal { + return &DDTemporal{ logger: logger, + isVP8: mimeType == mime.MimeTypeVP8, } } -func (v *VP8DD) GetState() any { +func (v *DDTemporal) GetState() any { return nil } -func (v *VP8DD) SeedState(_state any) { +func (v *DDTemporal) SeedState(_state any) { } -func (v *VP8DD) SetLast(_extPkt *buffer.ExtPacket) { +func (v *DDTemporal) SetLast(_extPkt *buffer.ExtPacket) { v.initialized = false } -func (v *VP8DD) UpdateOffsets(_extPkt *buffer.ExtPacket) { +func (v *DDTemporal) UpdateOffsets(_extPkt *buffer.ExtPacket) { // a layer switch starts a stream with its own frame numbers v.initialized = false } -func (v *VP8DD) UpdateAndGet(extPkt *buffer.ExtPacket, snOutOfOrder bool, _snHasGap bool, maxTemporalLayer int32) (int, [MaxHeaderSize]byte, int, error) { +func (v *DDTemporal) UpdateAndGet(extPkt *buffer.ExtPacket, snOutOfOrder bool, _snHasGap bool, maxTemporalLayer int32) (int, [MaxHeaderSize]byte, int, error) { var hdr [MaxHeaderSize]byte ddwdt := extPkt.DependencyDescriptor if ddwdt == nil { @@ -81,14 +84,18 @@ func (v *VP8DD) UpdateAndGet(extPkt *buffer.ExtPacket, snOutOfOrder bool, _snHas case isForwarded: return 0, hdr, 0, nil case snOutOfOrder: - return 0, hdr, 0, ErrFilteredVP8OutOfOrder + return 0, hdr, 0, ErrFilteredOutOfOrder default: - return 0, hdr, 0, ErrFilteredVP8TemporalLayer + return 0, hdr, 0, ErrFilteredTemporalLayer } } -func (v *VP8DD) UpdateAndGetPadding(_newPicture bool) ([]byte, error) { - // the one byte payload descriptor of a key frame that starts a partition +func (v *DDTemporal) UpdateAndGetPadding(_newPicture bool) ([]byte, error) { + if !v.isVP8 { + return nil, nil + } + + // the one byte VP8 payload descriptor of a key frame that starts a partition vp8Packet := &codec.VP8{ FirstByte: 0x10, IsKeyFrame: true, @@ -97,17 +104,17 @@ func (v *VP8DD) UpdateAndGetPadding(_newPicture bool) ([]byte, error) { return vp8Packet.Marshal() } -func (v *VP8DD) decision(frame uint64) (isForwarded bool, ok bool) { - if !v.initialized || frame > v.highestFrame || v.highestFrame-frame >= vp8DDFrameWindow { +func (v *DDTemporal) decision(frame uint64) (isForwarded bool, ok bool) { + if !v.initialized || frame > v.highestFrame || v.highestFrame-frame >= ddTemporalFrameWindow { return false, false } bit := uint64(1) << (v.highestFrame - frame) return v.forwardedMask&bit != 0, v.decidedMask&bit != 0 } -func (v *VP8DD) record(frame uint64, isForwarded bool) bool { +func (v *DDTemporal) record(frame uint64, isForwarded bool) bool { switch { - case !v.initialized || frame >= v.highestFrame+vp8DDFrameWindow: + case !v.initialized || frame >= v.highestFrame+ddTemporalFrameWindow: v.initialized = true v.highestFrame = frame v.decidedMask = 0 @@ -118,7 +125,7 @@ func (v *VP8DD) record(frame uint64, isForwarded bool) bool { v.forwardedMask <<= frame - v.highestFrame v.highestFrame = frame - case v.highestFrame-frame >= vp8DDFrameWindow: + case v.highestFrame-frame >= ddTemporalFrameWindow: return false } diff --git a/pkg/sfu/codecmunger/vp8dd_test.go b/pkg/sfu/codecmunger/ddtemporal_test.go similarity index 72% rename from pkg/sfu/codecmunger/vp8dd_test.go rename to pkg/sfu/codecmunger/ddtemporal_test.go index c8453dbf9..46973017f 100644 --- a/pkg/sfu/codecmunger/vp8dd_test.go +++ b/pkg/sfu/codecmunger/ddtemporal_test.go @@ -21,6 +21,7 @@ import ( "github.com/livekit/livekit-server/pkg/sfu/buffer" "github.com/livekit/livekit-server/pkg/sfu/testutils" + "github.com/livekit/protocol/codecs/mime" "github.com/livekit/protocol/logger" ) @@ -38,63 +39,70 @@ func vp8DDPacket(t *testing.T, frameNum uint64, tid int) *buffer.ExtPacket { return extPkt } -func TestVP8DDUpdateAndGet(t *testing.T) { +func TestDDTemporalUpdateAndGet(t *testing.T) { t.Run("filters by temporal layer, header unchanged", func(t *testing.T) { - v := NewVP8DD(logger.GetLogger()) + v := NewDDTemporal(mime.MimeTypeVP8, logger.GetLogger()) inputSize, _, n, err := v.UpdateAndGet(vp8DDPacket(t, 0, 0), false, false, 0) require.NoError(t, err) require.Zero(t, inputSize) require.Zero(t, n) _, _, _, err = v.UpdateAndGet(vp8DDPacket(t, 1, 2), false, false, 0) - require.ErrorIs(t, err, ErrFilteredVP8TemporalLayer) + require.ErrorIs(t, err, ErrFilteredTemporalLayer) }) t.Run("a gap does not exempt a filtered layer", func(t *testing.T) { - v := NewVP8DD(logger.GetLogger()) + v := NewDDTemporal(mime.MimeTypeVP8, logger.GetLogger()) _, _, _, err := v.UpdateAndGet(vp8DDPacket(t, 0, 0), false, false, 0) require.NoError(t, err) for frameNum := uint64(2); frameNum < 6; frameNum++ { _, _, _, err = v.UpdateAndGet(vp8DDPacket(t, frameNum, 1), false, true, 0) - require.ErrorIs(t, err, ErrFilteredVP8TemporalLayer) + require.ErrorIs(t, err, ErrFilteredTemporalLayer) } }) t.Run("every packet of a frame gets the decision of its first packet", func(t *testing.T) { - v := NewVP8DD(logger.GetLogger()) + v := NewDDTemporal(mime.MimeTypeVP8, logger.GetLogger()) _, _, _, err := v.UpdateAndGet(vp8DDPacket(t, 1, 2), false, false, 0) - require.ErrorIs(t, err, ErrFilteredVP8TemporalLayer) + require.ErrorIs(t, err, ErrFilteredTemporalLayer) _, _, _, err = v.UpdateAndGet(vp8DDPacket(t, 2, 1), false, false, 1) require.NoError(t, err) // the layer changed in between, late packets keep the decision of their frame _, _, _, err = v.UpdateAndGet(vp8DDPacket(t, 1, 2), true, false, 2) - require.ErrorIs(t, err, ErrFilteredVP8OutOfOrder) + require.ErrorIs(t, err, ErrFilteredOutOfOrder) _, _, _, err = v.UpdateAndGet(vp8DDPacket(t, 2, 1), true, false, 0) require.NoError(t, err) }) t.Run("late packet of a frame older than the window", func(t *testing.T) { - v := NewVP8DD(logger.GetLogger()) + v := NewDDTemporal(mime.MimeTypeVP8, logger.GetLogger()) _, _, _, err := v.UpdateAndGet(vp8DDPacket(t, 100, 0), false, false, 2) require.NoError(t, err) - _, _, _, err = v.UpdateAndGet(vp8DDPacket(t, 100-vp8DDFrameWindow, 0), true, false, 2) - require.ErrorIs(t, err, ErrFilteredVP8OutOfOrder) + _, _, _, err = v.UpdateAndGet(vp8DDPacket(t, 100-ddTemporalFrameWindow, 0), true, false, 2) + require.ErrorIs(t, err, ErrFilteredOutOfOrder) }) t.Run("layer switch forgets decisions", func(t *testing.T) { - v := NewVP8DD(logger.GetLogger()) + v := NewDDTemporal(mime.MimeTypeVP8, logger.GetLogger()) _, _, _, err := v.UpdateAndGet(vp8DDPacket(t, 1, 2), false, false, 0) - require.ErrorIs(t, err, ErrFilteredVP8TemporalLayer) + require.ErrorIs(t, err, ErrFilteredTemporalLayer) v.UpdateOffsets(vp8DDPacket(t, 1, 0)) _, _, _, err = v.UpdateAndGet(vp8DDPacket(t, 1, 2), false, false, 2) require.NoError(t, err) }) t.Run("padding", func(t *testing.T) { - v := NewVP8DD(logger.GetLogger()) + v := NewDDTemporal(mime.MimeTypeVP8, logger.GetLogger()) hdr, err := v.UpdateAndGetPadding(true) require.NoError(t, err) require.Equal(t, []byte{0x10}, hdr) + + // other codecs have no payload descriptor to prepend + for _, mimeType := range []mime.MimeType{mime.MimeTypeVP9, mime.MimeTypeAV1} { + hdr, err = NewDDTemporal(mimeType, logger.GetLogger()).UpdateAndGetPadding(true) + require.NoError(t, err) + require.Nil(t, hdr) + } }) } diff --git a/pkg/sfu/codecmunger/vp8.go b/pkg/sfu/codecmunger/vp8.go index 748acfcef..dc1d78b00 100644 --- a/pkg/sfu/codecmunger/vp8.go +++ b/pkg/sfu/codecmunger/vp8.go @@ -263,7 +263,7 @@ func (v *VP8) UpdateAndGet(extPkt *buffer.ExtPacket, snOutOfOrder bool, snHasGap v.pictureIdOffset += 1 } - return 0, hdr, 0, ErrFilteredVP8TemporalLayer + return 0, hdr, 0, ErrFilteredTemporalLayer } } } diff --git a/pkg/sfu/codecmunger/vp8_test.go b/pkg/sfu/codecmunger/vp8_test.go index 7c6d1a36f..79bb6ef8e 100644 --- a/pkg/sfu/codecmunger/vp8_test.go +++ b/pkg/sfu/codecmunger/vp8_test.go @@ -296,7 +296,7 @@ func TestTemporalLayerFiltering(t *testing.T) { // translate nIn, _, nOut, err := v.UpdateAndGet(extPkt, false, false, 0) require.Error(t, err) - require.ErrorIs(t, err, ErrFilteredVP8TemporalLayer) + require.ErrorIs(t, err, ErrFilteredTemporalLayer) require.Equal(t, 0, nIn) require.Equal(t, 0, nOut) dropped, _ := v.droppedPictureIds.Get(13467) @@ -310,7 +310,7 @@ func TestTemporalLayerFiltering(t *testing.T) { nIn, _, nOut, err = v.UpdateAndGet(extPkt, false, false, 0) require.Error(t, err) - require.ErrorIs(t, err, ErrFilteredVP8TemporalLayer) + require.ErrorIs(t, err, ErrFilteredTemporalLayer) require.Equal(t, 0, nIn) require.Equal(t, 0, nOut) dropped, _ = v.droppedPictureIds.Get(13467) @@ -324,7 +324,7 @@ func TestTemporalLayerFiltering(t *testing.T) { nIn, _, nOut, err = v.UpdateAndGet(extPkt, false, false, 0) require.Error(t, err) - require.ErrorIs(t, err, ErrFilteredVP8TemporalLayer) + require.ErrorIs(t, err, ErrFilteredTemporalLayer) require.Equal(t, 0, nIn) require.Equal(t, 0, nOut) dropped, _ = v.droppedPictureIds.Get(13467) diff --git a/pkg/sfu/forwarder.go b/pkg/sfu/forwarder.go index 99e229d9e..7e378ad20 100644 --- a/pkg/sfu/forwarder.go +++ b/pkg/sfu/forwarder.go @@ -424,95 +424,110 @@ func (f *Forwarder) DetermineCodec(codec webrtc.RTPCodecCapability, extensions [ f.ddForwarder = nil switch f.mime { case mime.MimeTypeVP8: - f.isDDAvailable = ddAvailable(extensions) - if f.isDDAvailable { - f.codecMunger = codecmunger.NewVP8DD(f.logger) - // the downtrack keeps its SSRC, so the frame numbers continue - f.ddForwarder = videolayerselector.NewDDForwarderFromOther(prevDDForwarder, f.logger) - f.ddForwarder.SeedState(f.ddForwarderSeed) - f.ddForwarderSeed = nil + f.setSimulcastSelector() + if ddAvailable(extensions) { + f.setDDTemporalFilter(prevDDForwarder) } else { + f.isDDAvailable = false f.codecMunger = codecmunger.NewVP8FromOther(f.codecMunger, f.logger) - } - if f.vls != nil { - if vls := videolayerselector.NewSimulcastFromOther(f.vls); vls != nil { - f.vls = vls - } else { - f.logger.Errorw("failed to create simulcast on codec change", nil) - } - } else { - f.vls = videolayerselector.NewSimulcast(f.logger) - } - if f.isDDAvailable { - f.vls.SetTemporalLayerSelector(temporallayerselector.NewDependencyDescriptor(f.logger)) - } else { f.vls.SetTemporalLayerSelector(temporallayerselector.NewVP8(f.logger)) } case mime.MimeTypeH264, mime.MimeTypeH265: - f.codecMunger = codecmunger.NewNull(f.logger) - if f.vls != nil { - if vls := videolayerselector.NewSimulcastFromOther(f.vls); vls != nil { - f.vls = vls - } else { - f.logger.Errorw("failed to create simulcast on codec change", nil) - } - } else { - f.vls = videolayerselector.NewSimulcast(f.logger) - } + f.setSimulcastSelector() + f.setUnfilteredTemporal() case mime.MimeTypeVP9: - f.codecMunger = codecmunger.NewNull(f.logger) if sfuutils.IsSimulcastMode(videoLayerMode) { - if f.vls != nil { - f.vls = videolayerselector.NewSimulcastFromOther(f.vls) + f.setSimulcastSelector() + if ddAvailable(extensions) { + f.setDDTemporalFilter(prevDDForwarder) } else { - f.vls = videolayerselector.NewSimulcast(f.logger) - } - } else { - f.isDDAvailable = ddAvailable(extensions) - if f.isDDAvailable { - if f.vls != nil { - f.vls = videolayerselector.NewDependencyDescriptorFromOther(f.vls) - } else { - f.vls = videolayerselector.NewDependencyDescriptor(f.logger) - } - } else { - if f.vls != nil { - f.vls = videolayerselector.NewVP9FromOther(f.vls) - } else { - f.vls = videolayerselector.NewVP9(f.logger) - } + f.setUnfilteredTemporal() } + break } - case mime.MimeTypeAV1: f.codecMunger = codecmunger.NewNull(f.logger) - if sfuutils.IsSimulcastMode(videoLayerMode) { + f.isDDAvailable = ddAvailable(extensions) + if f.isDDAvailable { if f.vls != nil { - f.vls = videolayerselector.NewSimulcastFromOther(f.vls) + f.vls = videolayerselector.NewDependencyDescriptorFromOther(f.vls) } else { - f.vls = videolayerselector.NewSimulcast(f.logger) + f.vls = videolayerselector.NewDependencyDescriptor(f.logger) } } else { - f.isDDAvailable = ddAvailable(extensions) - if f.isDDAvailable { - if f.vls != nil { - f.vls = videolayerselector.NewDependencyDescriptorFromOther(f.vls) - } else { - f.vls = videolayerselector.NewDependencyDescriptor(f.logger) - } + if f.vls != nil { + f.vls = videolayerselector.NewVP9FromOther(f.vls) } else { - if f.vls != nil { - f.vls = videolayerselector.NewSimulcastFromOther(f.vls) - } else { - f.vls = videolayerselector.NewSimulcast(f.logger) - } + f.vls = videolayerselector.NewVP9(f.logger) } } + // the svc selectors pick temporal layers themselves + f.vls.SetTemporalLayerSelector(nil) + + case mime.MimeTypeAV1: + if sfuutils.IsSimulcastMode(videoLayerMode) { + f.setSimulcastSelector() + if ddAvailable(extensions) { + f.setDDTemporalFilter(prevDDForwarder) + } else { + f.setUnfilteredTemporal() + } + break + } + + f.codecMunger = codecmunger.NewNull(f.logger) + f.isDDAvailable = ddAvailable(extensions) + if f.isDDAvailable { + if f.vls != nil { + f.vls = videolayerselector.NewDependencyDescriptorFromOther(f.vls) + } else { + f.vls = videolayerselector.NewDependencyDescriptor(f.logger) + } + f.vls.SetTemporalLayerSelector(nil) + } else { + // without the dependency descriptor nothing tells the temporal layers apart + f.setSimulcastSelector() + f.setUnfilteredTemporal() + } } } +// setSimulcastSelector switches to the simulcast selector, keeping the selector state +// across a codec change +func (f *Forwarder) setSimulcastSelector() { + if f.vls == nil { + f.vls = videolayerselector.NewSimulcast(f.logger) + return + } + if vls := videolayerselector.NewSimulcastFromOther(f.vls); vls != nil { + f.vls = vls + } else { + f.logger.Errorw("failed to create simulcast on codec change", nil) + } +} + +// setDDTemporalFilter filters the temporal layers of a simulcast stream by frame with the +// dependency descriptor, and forwards the descriptor to the subscriber +func (f *Forwarder) setDDTemporalFilter(prevDDForwarder *videolayerselector.DDForwarder) { + f.isDDAvailable = true + f.codecMunger = codecmunger.NewDDTemporal(f.mime, f.logger) + f.vls.SetTemporalLayerSelector(temporallayerselector.NewDependencyDescriptor(f.logger)) + // the downtrack keeps its SSRC, so the frame numbers continue + f.ddForwarder = videolayerselector.NewDDForwarderFromOther(prevDDForwarder, f.logger) + f.ddForwarder.SeedState(f.ddForwarderSeed) + f.ddForwarderSeed = nil +} + +// setUnfilteredTemporal forwards every temporal layer of a simulcast stream that has +// nothing to filter them with +func (f *Forwarder) setUnfilteredTemporal() { + f.isDDAvailable = false + f.codecMunger = codecmunger.NewNull(f.logger) + f.vls.SetTemporalLayerSelector(temporallayerselector.NewNull()) +} + func (f *Forwarder) GetState() *livekit.RTPForwarderState { f.lock.RLock() defer f.lock.RUnlock() @@ -2211,21 +2226,26 @@ func (f *Forwarder) isEndOfLayerFrame(extPkt *buffer.ExtPacket) bool { // should be called with lock held func (f *Forwarder) getTranslationParamsVideo(extPkt *buffer.ExtPacket, layer int32) (TranslationParams, error) { tp := TranslationParams{} - if f.isDDAvailable && f.mime == mime.MimeTypeVP8 && extPkt.DependencyDescriptor == nil && len(extPkt.Packet.Payload) != 0 { + if f.ddForwarder != nil && extPkt.DependencyDescriptor == nil && len(extPkt.Packet.Payload) != 0 { // The buffer stopped parsing the dependency descriptor because the publisher does - // not send it, so switch temporal layers with the VP8 payload descriptor. Padding - // never carries the descriptor, so it does not count. - f.logger.Infow("turning off dependency descriptor for VP8 temporal layers", "layer", layer) - f.isDDAvailable = false + // not send it. VP8 switches temporal layers with its payload descriptor, other + // codecs forward every temporal layer. Padding never carries the descriptor, so it + // does not count. + f.logger.Infow("turning off dependency descriptor for simulcast temporal layers", "layer", layer) f.ddForwarder = nil - f.codecMunger = codecmunger.NewVP8FromOther(f.codecMunger, f.logger) - if f.started { - // the munger takes its picture ID and TL0PICIDX from the first packet it sees - f.codecMunger.SetLast(extPkt) + if f.mime == mime.MimeTypeVP8 { + f.isDDAvailable = false + f.codecMunger = codecmunger.NewVP8FromOther(f.codecMunger, f.logger) + if f.started { + // the munger takes its picture ID and TL0PICIDX from the first packet it sees + f.codecMunger.SetLast(extPkt) + } + f.vls.SetTemporalLayerSelector(temporallayerselector.NewVP8(f.logger)) + } else { + f.setUnfilteredTemporal() } - f.vls.SetTemporalLayerSelector(temporallayerselector.NewVP8(f.logger)) } - if _, ok := f.codecMunger.(*codecmunger.VP8DD); ok && extPkt.DependencyDescriptor != nil { + if _, ok := f.codecMunger.(*codecmunger.DDTemporal); ok && f.mime == mime.MimeTypeVP8 && extPkt.DependencyDescriptor != nil { if extPkt.VP8 != nil && extPkt.VP8.I { // The publisher sends picture IDs with the dependency descriptor, so they need // rewriting. The VP8 munger does that, the temporal layers still switch with the @@ -2246,7 +2266,8 @@ func (f *Forwarder) getTranslationParamsVideo(extPkt *buffer.ExtPacket, layer in result := f.vls.Select(extPkt, layer) if !result.IsSelected { - if f.isDDAvailable && f.mime != mime.MimeTypeVP8 && extPkt.DependencyDescriptor == nil { + // the svc selector, simulcast with the dependency descriptor turns it off above + if f.isDDAvailable && f.ddForwarder == nil && extPkt.DependencyDescriptor == nil { f.logger.Infow( "turning off dependency descriptor", "layer", layer, @@ -2261,7 +2282,8 @@ func (f *Forwarder) getTranslationParamsVideo(extPkt *buffer.ExtPacket, layer in case mime.MimeTypeVP9: f.vls = videolayerselector.NewVP9FromOther(f.vls) case mime.MimeTypeAV1: - f.vls = videolayerselector.NewSimulcastFromOther(f.vls) + f.setSimulcastSelector() + f.setUnfilteredTemporal() } } tp.shouldDrop = true @@ -2340,10 +2362,10 @@ func (f *Forwarder) translateCodecHeader(extPkt *buffer.ExtPacket, tp *Translati ) if err != nil { tp.shouldDrop = true - if err == codecmunger.ErrFilteredVP8TemporalLayer || err == codecmunger.ErrOutOfOrderVP8PictureIdCacheMiss || err == codecmunger.ErrFilteredVP8OutOfOrder { + if err == codecmunger.ErrFilteredTemporalLayer || err == codecmunger.ErrOutOfOrderVP8PictureIdCacheMiss || err == codecmunger.ErrFilteredOutOfOrder { switch tp.rtp.snOrdering { case SequenceNumberOrderingContiguous: - if err == codecmunger.ErrFilteredVP8TemporalLayer { + if err == codecmunger.ErrFilteredTemporalLayer { // filtered temporal layer, update sequence number offset to prevent holes f.rtpMunger.PacketDropped(extPkt) } diff --git a/pkg/sfu/forwarder_test.go b/pkg/sfu/forwarder_test.go index 398a3ee94..567a79096 100644 --- a/pkg/sfu/forwarder_test.go +++ b/pkg/sfu/forwarder_test.go @@ -1980,7 +1980,7 @@ func TestForwarderVP8TemporalUpSwitchAtLayerSync(t *testing.T) { func TestForwarderVP8TemporalUpSwitchWithDependencyDescriptor(t *testing.T) { f := NewForwarder(webrtc.RTPCodecTypeVideo, logger.GetLogger(), true, true, true, nil) f.DetermineCodec(testutils.TestVP8Codec, []webrtc.RTPHeaderExtensionParameter{{URI: dd.ExtensionURI}}, livekit.VideoLayer_MODE_UNUSED) - require.IsType(t, &codecmunger.VP8DD{}, f.codecMunger) + require.IsType(t, &codecmunger.DDTemporal{}, f.codecMunger) f.vls.SetCurrent(buffer.VideoLayer{Spatial: 0, Temporal: 0}) f.vls.SetTarget(buffer.VideoLayer{Spatial: 0, Temporal: 2}) @@ -2090,7 +2090,7 @@ func TestForwarderVP8DDWithPictureID(t *testing.T) { f := NewForwarder(webrtc.RTPCodecTypeVideo, logger.GetLogger(), true, true, true, nil) f.DetermineCodec(testutils.TestVP8Codec, []webrtc.RTPHeaderExtensionParameter{{URI: dd.ExtensionURI}}, livekit.VideoLayer_MODE_UNUSED) f.vls.SetTarget(buffer.VideoLayer{Spatial: 0, Temporal: 2}) - require.IsType(t, &codecmunger.VP8DD{}, f.codecMunger) + require.IsType(t, &codecmunger.DDTemporal{}, f.codecMunger) // the one byte descriptor keeps the dependency descriptor munger extPkt, err := testutils.GetTestExtPacketVP8DD( @@ -2099,7 +2099,7 @@ func TestForwarderVP8DDWithPictureID(t *testing.T) { ) require.NoError(t, err) _, _ = f.GetTranslationParams(extPkt, 0) - require.IsType(t, &codecmunger.VP8DD{}, f.codecMunger) + require.IsType(t, &codecmunger.DDTemporal{}, f.codecMunger) require.True(t, f.started) // a picture ID next to the dependency descriptor needs the VP8 munger @@ -2115,30 +2115,40 @@ func TestForwarderVP8DDWithPictureID(t *testing.T) { require.True(t, f.isDDAvailable, "temporal layers still switch with the dependency descriptor") } -func TestForwarderVP8ForwardsDependencyDescriptor(t *testing.T) { - // an L3T3 key frame descriptor with its dependency structure, the capture the - // dependency descriptor package tests with +// testKeyFrameDescriptor is an L3T3 key frame descriptor with its dependency structure, the +// capture the dependency descriptor package tests with +func testKeyFrameDescriptor(t *testing.T) dd.DependencyDescriptor { raw, err := hex.DecodeString("c1017280081485214eafffaaaa863cf0430c10c302afc0aaa0063c00430010c002a000a80006000040001d954926e082b04a0941b820ac1282503157f974000ca864330e222222eca8655304224230eca877530077004200ef008601df010d") require.NoError(t, err) keyFrameDescriptor := dd.DependencyDescriptor{} _, err = (&dd.DependencyDescriptorExtension{Descriptor: &keyFrameDescriptor}).Unmarshal(raw) require.NoError(t, err) require.NotNil(t, keyFrameDescriptor.AttachedStructure) + return keyFrameDescriptor +} +// testDDPacket is a packet of frame frameNum on temporalID, with keyFrameDescriptor as its +// dependency descriptor, its structure attached only on a key frame +func testDDPacket(t *testing.T, keyFrameDescriptor dd.DependencyDescriptor, sn uint16, frameNum uint64, temporalID int, isKeyFrame bool) *buffer.ExtPacket { + extPkt, err := testutils.GetTestExtPacketVP8DD( + &testutils.TestExtPacketParams{SequenceNumber: sn, Timestamp: 0xabcdef + uint32(frameNum)*3000, SSRC: 0x12345678, PayloadSize: 20, Marker: true, IsKeyFrame: isKeyFrame}, + &testutils.TestDDParams{FirstPacketInFrame: true, ExtFrameNum: frameNum, TemporalID: temporalID, DTIs: "SSS", DecodeTargets: testutils.TestVP8DecodeTargets}, + ) + require.NoError(t, err) + descriptor := keyFrameDescriptor + descriptor.FrameNumber = uint16(frameNum) + if !isKeyFrame { + descriptor.AttachedStructure = nil + descriptor.ActiveDecodeTargetsBitmask = nil + } + extPkt.DependencyDescriptor.Descriptor = &descriptor + return extPkt +} + +func TestForwarderVP8ForwardsDependencyDescriptor(t *testing.T) { + keyFrameDescriptor := testKeyFrameDescriptor(t) packet := func(sn uint16, frameNum uint64, isKeyFrame bool) *buffer.ExtPacket { - extPkt, err := testutils.GetTestExtPacketVP8DD( - &testutils.TestExtPacketParams{SequenceNumber: sn, Timestamp: 0xabcdef + uint32(frameNum)*3000, SSRC: 0x12345678, PayloadSize: 20, Marker: true, IsKeyFrame: isKeyFrame}, - &testutils.TestDDParams{FirstPacketInFrame: true, ExtFrameNum: frameNum, DTIs: "SSS", DecodeTargets: testutils.TestVP8DecodeTargets}, - ) - require.NoError(t, err) - descriptor := keyFrameDescriptor - descriptor.FrameNumber = uint16(frameNum) - if !isKeyFrame { - descriptor.AttachedStructure = nil - descriptor.ActiveDecodeTargetsBitmask = nil - } - extPkt.DependencyDescriptor.Descriptor = &descriptor - return extPkt + return testDDPacket(t, keyFrameDescriptor, sn, frameNum, 0, isKeyFrame) } f := NewForwarder(webrtc.RTPCodecTypeVideo, logger.GetLogger(), true, true, true, nil) @@ -2692,3 +2702,192 @@ func TestForwarderNeedsPeriodicSyncCheck(t *testing.T) { f.DetermineCodec(av1, []webrtc.RTPHeaderExtensionParameter{{URI: dd.ExtensionURI}}, livekit.VideoLayer_MODE_UNUSED) require.True(t, f.NeedsPeriodicSyncCheck()) } + +// TestForwarderSimulcastTemporalLayerAfterSwitch checks that, after a spatial layer switch on a +// key frame (which is always on temporal layer 0), the forwarder reaches its target layer so that +// GetNextHigherTransition (and hence stream allocator probing) is not blocked forever. +// +// VP9, AV1 and H.264 simulcast without the dependency descriptor forward every temporal layer of +// the selected spatial layer, so the current temporal layer has to follow the target. VP8 steps +// up through its temporal layer selector. +func TestForwarderSimulcastTemporalLayerAfterSwitch(t *testing.T) { + bitrates := Bitrates{ + {100, 150, 200, 0}, + {300, 450, 600, 0}, + {900, 1350, 1800, 0}, + } + + testCases := []struct { + name string + previousCodec *webrtc.RTPCodecCapability + codec webrtc.RTPCodecCapability + target buffer.VideoLayer + expectedCurrent buffer.VideoLayer + }{ + { + name: "VP9 simulcast, three temporal layers", + codec: webrtc.RTPCodecCapability{MimeType: mime.MimeTypeVP9.String(), ClockRate: 90000}, + target: buffer.VideoLayer{Spatial: 1, Temporal: 2}, + expectedCurrent: buffer.VideoLayer{Spatial: 1, Temporal: 2}, + }, + { + name: "AV1 simulcast, three temporal layers", + codec: webrtc.RTPCodecCapability{MimeType: mime.MimeTypeAV1.String(), ClockRate: 90000}, + target: buffer.VideoLayer{Spatial: 1, Temporal: 2}, + expectedCurrent: buffer.VideoLayer{Spatial: 1, Temporal: 2}, + }, + { + name: "VP9 simulcast, one temporal layer", + codec: webrtc.RTPCodecCapability{MimeType: mime.MimeTypeVP9.String(), ClockRate: 90000}, + target: buffer.VideoLayer{Spatial: 1, Temporal: 0}, + expectedCurrent: buffer.VideoLayer{Spatial: 1, Temporal: 0}, + }, + { + name: "H.264 simulcast", + codec: webrtc.RTPCodecCapability{MimeType: mime.MimeTypeH264.String(), ClockRate: 90000}, + target: buffer.VideoLayer{Spatial: 1, Temporal: 0}, + expectedCurrent: buffer.VideoLayer{Spatial: 1, Temporal: 0}, + }, + { + // the VP8 temporal layer selector moves to the target on a key frame + name: "VP8 simulcast, three temporal layers", + codec: testutils.TestVP8Codec, + target: buffer.VideoLayer{Spatial: 1, Temporal: 2}, + expectedCurrent: buffer.VideoLayer{Spatial: 1, Temporal: 2}, + }, + { + // the VP8 temporal layer selector must not be carried over on codec change + name: "VP8 to VP9 simulcast, three temporal layers", + previousCodec: &testutils.TestVP8Codec, + codec: webrtc.RTPCodecCapability{MimeType: mime.MimeTypeVP9.String(), ClockRate: 90000}, + target: buffer.VideoLayer{Spatial: 1, Temporal: 2}, + expectedCurrent: buffer.VideoLayer{Spatial: 1, Temporal: 2}, + }, + { + name: "VP8 to AV1 simulcast, three temporal layers", + previousCodec: &testutils.TestVP8Codec, + codec: webrtc.RTPCodecCapability{MimeType: mime.MimeTypeAV1.String(), ClockRate: 90000}, + target: buffer.VideoLayer{Spatial: 1, Temporal: 2}, + expectedCurrent: buffer.VideoLayer{Spatial: 1, Temporal: 2}, + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + f := NewForwarder( + webrtc.RTPCodecTypeVideo, + logger.GetLogger(), + true, // skipReferenceTS + true, // disableOpportunisticAllocation + false, // enableStartAtDesiredQuality + nil, + ) + if tc.previousCodec != nil { + f.DetermineCodec(*tc.previousCodec, nil, livekit.VideoLayer_ONE_SPATIAL_LAYER_PER_STREAM) + } + f.DetermineCodec(tc.codec, nil, livekit.VideoLayer_ONE_SPATIAL_LAYER_PER_STREAM) + f.SetMaxSpatialLayer(buffer.DefaultMaxLayerSpatial) + f.SetMaxTemporalLayer(buffer.DefaultMaxLayerTemporal) + f.SetMaxPublishedLayer(buffer.DefaultMaxLayerSpatial) + f.SetMaxTemporalLayerSeen(buffer.DefaultMaxLayerTemporal) + + // down-switched for bandwidth: was on the top layer, target is now lower + f.lastAllocation.IsDeficient = true + f.vls.SetCurrent(buffer.VideoLayer{Spatial: 2, Temporal: tc.target.Temporal}) + f.vls.SetTarget(tc.target) + + // target pending (spatial switch not done yet): no higher transition + _, available := f.GetNextHigherTransition(bitrates, false) + require.False(t, available) + + // key frame of the target spatial layer, temporal layer 0 + params := &testutils.TestExtPacketParams{ + SequenceNumber: 23333, + Timestamp: 0xabcdef, + SSRC: 0x12345678, + PayloadSize: 20, + IsKeyFrame: true, + VideoLayer: buffer.VideoLayer{Spatial: tc.target.Spatial, Temporal: 0}, + } + var extPkt *buffer.ExtPacket + var err error + if tc.codec.MimeType == testutils.TestVP8Codec.MimeType { + extPkt, err = testutils.GetTestExtPacketVP8(params, &codec.VP8{ + FirstByte: 25, + I: true, + M: true, + PictureID: 13467, + L: true, + TL0PICIDX: 233, + T: true, + TID: 0, + Y: true, + K: true, + KEYIDX: 23, + HeaderSize: 6, + IsKeyFrame: true, + }) + } else { + extPkt, err = testutils.GetTestExtPacket(params) + } + require.NoError(t, err) + + tp, err := f.GetTranslationParams(extPkt, tc.target.Spatial) + require.NoError(t, err) + require.False(t, tp.shouldDrop) + require.True(t, tp.isSwitching) + + require.Equal(t, tc.expectedCurrent, f.vls.GetCurrent()) + // target reached: the next higher transition (and hence probing) is available + _, available = f.GetNextHigherTransition(bitrates, false) + require.True(t, available) + }) + } +} + +// VP9 and AV1 simulcast with the dependency descriptor filter temporal layers by frame and +// forward the descriptor, as VP8 does. A publisher that stops sending the descriptor gets +// every temporal layer forwarded. +func TestForwarderSimulcastDependencyDescriptor(t *testing.T) { + for _, mimeType := range []mime.MimeType{mime.MimeTypeVP9, mime.MimeTypeAV1} { + t.Run(mimeType.String(), func(t *testing.T) { + keyFrameDescriptor := testKeyFrameDescriptor(t) + f := NewForwarder(webrtc.RTPCodecTypeVideo, logger.GetLogger(), true, true, true, nil) + f.DetermineCodec( + webrtc.RTPCodecCapability{MimeType: mimeType.String(), ClockRate: 90000}, + []webrtc.RTPHeaderExtensionParameter{{URI: dd.ExtensionURI}}, + livekit.VideoLayer_ONE_SPATIAL_LAYER_PER_STREAM, + ) + require.IsType(t, &codecmunger.DDTemporal{}, f.codecMunger) + require.NotNil(t, f.ddForwarder) + f.vls.SetCurrent(buffer.VideoLayer{Spatial: 0, Temporal: 0}) + f.vls.SetTarget(buffer.VideoLayer{Spatial: 0, Temporal: 0}) + + tp, err := f.GetTranslationParams(testDDPacket(t, keyFrameDescriptor, 10, 100, 0, true), 0) + require.NoError(t, err) + require.False(t, tp.shouldDrop) + require.NotEmpty(t, tp.ddBytesSpill) + + // TL2 is above the target and is filtered, its slot is rolled back + tp, err = f.GetTranslationParams(testDDPacket(t, keyFrameDescriptor, 11, 101, 2, false), 0) + require.NoError(t, err) + require.True(t, tp.shouldDrop) + + tp, err = f.GetTranslationParams(testDDPacket(t, keyFrameDescriptor, 12, 102, 0, false), 0) + require.NoError(t, err) + require.False(t, tp.shouldDrop) + require.Equal(t, uint64(11), tp.rtp.extSequenceNumber) + require.NotEmpty(t, tp.ddInline()) + + // the publisher stops sending the descriptor + noDD, err := testutils.GetTestExtPacket(&testutils.TestExtPacketParams{SequenceNumber: 13, Timestamp: 0xabcdef + 103*3000, SSRC: 0x12345678, PayloadSize: 20, Marker: true}) + require.NoError(t, err) + tp, err = f.GetTranslationParams(noDD, 0) + require.NoError(t, err) + require.False(t, tp.shouldDrop) + require.Nil(t, f.ddForwarder) + require.IsType(t, &codecmunger.Null{}, f.codecMunger) + require.Empty(t, tp.ddInline()) + }) + } +} diff --git a/pkg/sfu/videolayerselector/ddforwarder.go b/pkg/sfu/videolayerselector/ddforwarder.go index f5e525579..2080f34d7 100644 --- a/pkg/sfu/videolayerselector/ddforwarder.go +++ b/pkg/sfu/videolayerselector/ddforwarder.go @@ -101,7 +101,8 @@ func (d *DDForwarder) Forward(extPkt *buffer.ExtPacket, temporal int32, result * descriptor := ddwdt.Descriptor extFrameNum := ddwdt.ExtFrameNum - hasStructure := extPkt.IsKeyFrame && descriptor.AttachedStructure != nil + // the parser accepts a structure only on the first packet of a frame + hasStructure := descriptor.AttachedStructure != nil switch { case !d.initialized || extPkt.Packet.SSRC != d.ssrc: // a stream starts on a key frame, a late packet of the old layer is not described diff --git a/pkg/sfu/videolayerselector/temporallayerselector/null.go b/pkg/sfu/videolayerselector/temporallayerselector/null.go index 51f2fb591..52358b5f1 100644 --- a/pkg/sfu/videolayerselector/temporallayerselector/null.go +++ b/pkg/sfu/videolayerselector/temporallayerselector/null.go @@ -18,14 +18,17 @@ import ( "github.com/livekit/livekit-server/pkg/sfu/buffer" ) +// Null is for a stream whose temporal layers cannot be filtered, simulcast without the +// dependency descriptor for example: every temporal layer is forwarded. It reports the +// target as the current layer, so a target below the highest layer does not stay pending +// and block further allocation. The forwarder pairs it with the null codec munger, which +// ignores the temporal layer it returns for the packet. type Null struct{} func NewNull() *Null { return &Null{} } -func Select(_extPkt *buffer.ExtPacket, current int32, _target int32) (this int32, next int32) { - this = current - next = current - return +func (n *Null) Select(_extPkt *buffer.ExtPacket, _current int32, target int32) (this int32, next int32) { + return target, target } diff --git a/pkg/sfu/videolayerselector/temporallayerselector/null_test.go b/pkg/sfu/videolayerselector/temporallayerselector/null_test.go new file mode 100644 index 000000000..a86ec71ca --- /dev/null +++ b/pkg/sfu/videolayerselector/temporallayerselector/null_test.go @@ -0,0 +1,35 @@ +// Copyright 2026 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 temporallayerselector + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +func TestNullFollowsTarget(t *testing.T) { + n := NewNull() + + // a switch key frame on temporal layer 0 reaches a higher target right away + this, next := n.Select(nil, 0, 2) + require.Equal(t, int32(2), this) + require.Equal(t, int32(2), next) + + // and a lower one + this, next = n.Select(nil, 2, 1) + require.Equal(t, int32(1), this) + require.Equal(t, int32(1), next) +}