sfu: filter VP9 and AV1 simulcast temporal layers with the dependency descriptor (#4991)

* sfu: filter VP9 and AV1 simulcast temporal layers with the dependency descriptor

Every simulcast codec gets an explicit temporal layer selector. With the
dependency descriptor, VP8, VP9 and AV1 filter temporal layers by frame and
forward the descriptor. Without it, a pass-through selector forwards every
temporal layer and reports the target as reached, so a pending target cannot
block allocation. VP8DD becomes the codec-neutral DDTemporal.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* sfu: VP8-only padding header in DDTemporal, structure gate on presence

DDTemporal returns the one byte VP8 payload descriptor for padding only for
VP8. The descriptor forwarder takes a structure from any packet that carries
one, as the parser accepts it only on the first packet of a frame.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Raja Subramanian
2026-10-12 16:53:09 +05:30
committed by GitHub
co-authored by Claude Opus 5.5
parent d621849e3a
commit 93296ad4f6
10 changed files with 423 additions and 148 deletions
+2 -2
View File
@@ -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).
@@ -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
}
@@ -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)
}
})
}
+1 -1
View File
@@ -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
}
}
}
+3 -3
View File
@@ -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)
+102 -80
View File
@@ -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)
}
+218 -19
View File
@@ -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())
})
}
}
+2 -1
View File
@@ -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
@@ -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
}
@@ -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)
}