From b90b0ca1dfefb94c4746900e3ec55829780e064c Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Sun, 11 Oct 2026 00:47:42 +0530 Subject: [PATCH] sfu: switch VP8 temporal layers up with the dependency descriptor (#4974) libwebrtc strips the VP8 payload descriptor when it sends the dependency descriptor, so the layer sync bit that #4947 waits for is never set and a subscriber that lowers its frame rate stays at the base layer until a key frame. A new temporal layer selector reads only the dependency descriptor: it switches up at a switch point whose referenced frames were all forwarded. The forwarder uses it for VP8 when the extension is negotiated and falls back to the Y-bit selector when the descriptor is not sent. Fixes #4973. Co-authored-by: Claude Opus 5.5 --- pkg/sfu/forwarder.go | 18 +- pkg/sfu/forwarder_test.go | 58 ++++++ pkg/sfu/testutils/data.go | 60 ++++++ .../dependencydescriptor.go | 125 ++++++++++++ .../dependencydescriptor_test.go | 184 ++++++++++++++++++ 5 files changed, 443 insertions(+), 2 deletions(-) create mode 100644 pkg/sfu/videolayerselector/temporallayerselector/dependencydescriptor.go create mode 100644 pkg/sfu/videolayerselector/temporallayerselector/dependencydescriptor_test.go diff --git a/pkg/sfu/forwarder.go b/pkg/sfu/forwarder.go index fbdc28092..eaaf367cb 100644 --- a/pkg/sfu/forwarder.go +++ b/pkg/sfu/forwarder.go @@ -406,7 +406,12 @@ func (f *Forwarder) DetermineCodec(codec webrtc.RTPCodecCapability, extensions [ } else { f.vls = videolayerselector.NewSimulcast(f.logger) } - f.vls.SetTemporalLayerSelector(temporallayerselector.NewVP8(f.logger)) + f.isDDAvailable = ddAvailable(extensions) + 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) @@ -2160,6 +2165,15 @@ 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 { + // 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 + f.vls.SetTemporalLayerSelector(temporallayerselector.NewVP8(f.logger)) + } + if !f.vls.GetTarget().IsValid() { // stream is paused by streamallocator tp.shouldDrop = true @@ -2168,7 +2182,7 @@ func (f *Forwarder) getTranslationParamsVideo(extPkt *buffer.ExtPacket, layer in result := f.vls.Select(extPkt, layer) if !result.IsSelected { - if f.isDDAvailable && extPkt.DependencyDescriptor == nil { + if f.isDDAvailable && f.mime != mime.MimeTypeVP8 && extPkt.DependencyDescriptor == nil { f.logger.Infow( "turning off dependency descriptor", "layer", layer, diff --git a/pkg/sfu/forwarder_test.go b/pkg/sfu/forwarder_test.go index 146662802..7d4cf4d83 100644 --- a/pkg/sfu/forwarder_test.go +++ b/pkg/sfu/forwarder_test.go @@ -1975,6 +1975,64 @@ func TestForwarderVP8TemporalUpSwitchAtLayerSync(t *testing.T) { require.Equal(t, int32(2), f.vls.GetCurrent().Temporal) } +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) + f.vls.SetCurrent(buffer.VideoLayer{Spatial: 0, Temporal: 0}) + f.vls.SetTarget(buffer.VideoLayer{Spatial: 0, Temporal: 2}) + + // libwebrtc strips the VP8 payload descriptor when it sends the dependency + // descriptor: no TID and no layer sync bit + packet := func(frameNum uint64, tid int, dtis string, frameDiffs []int, isKeyFrame bool) *buffer.ExtPacket { + extPkt, err := testutils.GetTestExtPacketVP8DD( + &testutils.TestExtPacketParams{SequenceNumber: uint16(frameNum), PayloadSize: 20, Marker: true, IsKeyFrame: isKeyFrame}, + &testutils.TestDDParams{ + FirstPacketInFrame: true, + ExtFrameNum: frameNum, + TemporalID: tid, + DTIs: dtis, + FrameDiffs: frameDiffs, + DecodeTargets: testutils.TestVP8DecodeTargets, + }, + ) + require.NoError(t, err) + return extPkt + } + + f.vls.SetTarget(buffer.VideoLayer{Spatial: 0, Temporal: 0}) + require.Equal(t, int32(0), f.vls.SelectTemporal(packet(0, 0, "SSS", nil, true))) + f.vls.SetTarget(buffer.VideoLayer{Spatial: 0, Temporal: 2}) + require.Equal(t, int32(0), f.vls.SelectTemporal(packet(1, 2, "--D", []int{1}, false))) + require.Equal(t, int32(0), f.vls.GetCurrent().Temporal) + require.Equal(t, int32(2), f.vls.SelectTemporal(packet(2, 1, "-SS", []int{2}, false))) + require.Equal(t, int32(2), f.vls.GetCurrent().Temporal) + + // padding never carries the descriptor and leaves it on + f.vls.SetCurrent(buffer.VideoLayer{Spatial: 0, Temporal: 0}) + padding, err := testutils.GetTestExtPacket(&testutils.TestExtPacketParams{SequenceNumber: 3, PaddingSize: 20}) + require.NoError(t, err) + _, _ = f.getTranslationParamsVideo(padding, 0) + require.True(t, f.isDDAvailable) + + // a packet with a payload and without the descriptor means the buffer stopped + // parsing it, the Y bit then raises the layer + extPkt, err := testutils.GetTestExtPacketVP8( + &testutils.TestExtPacketParams{SequenceNumber: 4, PayloadSize: 20}, + &codec.VP8{S: true, I: true, PictureID: 4, T: true, TID: 0}, + ) + require.NoError(t, err) + _, _ = f.getTranslationParamsVideo(extPkt, 0) + require.False(t, f.isDDAvailable) + + extPkt, err = testutils.GetTestExtPacketVP8( + &testutils.TestExtPacketParams{SequenceNumber: 5, PayloadSize: 20}, + &codec.VP8{S: true, I: true, PictureID: 5, T: true, TID: 1, Y: true}, + ) + require.NoError(t, err) + require.Equal(t, int32(1), f.vls.SelectTemporal(extPkt)) + require.Equal(t, int32(1), f.vls.GetCurrent().Temporal) +} + func TestForwarderGetSnTsForPadding(t *testing.T) { f := newForwarder(testutils.TestVP8Codec, webrtc.RTPCodecTypeVideo) diff --git a/pkg/sfu/testutils/data.go b/pkg/sfu/testutils/data.go index 9b8a1386e..f4a89e49d 100644 --- a/pkg/sfu/testutils/data.go +++ b/pkg/sfu/testutils/data.go @@ -21,6 +21,7 @@ import ( "github.com/pion/webrtc/v4" "github.com/livekit/livekit-server/pkg/sfu/buffer" + dd "github.com/livekit/livekit-server/pkg/sfu/rtpextension/dependencydescriptor" "github.com/livekit/mediatransportutil/pkg/codec" ) @@ -96,6 +97,65 @@ func GetTestExtPacketVP8(params *TestExtPacketParams, vp8 *codec.VP8) (*buffer.E // -------------------------------------- +type TestDDParams struct { + FirstPacketInFrame bool + ExtFrameNum uint64 + TemporalID int + // decode target indications as one character per decode target: S, R, D or - + DTIs string + FrameDiffs []int + DecodeTargets []buffer.DependencyDescriptorDecodeTarget +} + +// GetTestExtPacketVP8DD returns a VP8 packet with a dependency descriptor and the +// minimized VP8 payload descriptor that libwebrtc sends with it: no TID, no Y bit and +// no picture ID. +func GetTestExtPacketVP8DD(params *TestExtPacketParams, ddParams *TestDDParams) (*buffer.ExtPacket, error) { + ep, err := GetTestExtPacketVP8(params, &codec.VP8{S: ddParams.FirstPacketInFrame, IsKeyFrame: params.IsKeyFrame}) + if err != nil { + return nil, err + } + + dtis := make([]dd.DecodeTargetIndication, 0, len(ddParams.DTIs)) + for _, c := range ddParams.DTIs { + switch c { + case 'S': + dtis = append(dtis, dd.DecodeTargetSwitch) + case 'R': + dtis = append(dtis, dd.DecodeTargetRequired) + case 'D': + dtis = append(dtis, dd.DecodeTargetDiscardable) + default: + dtis = append(dtis, dd.DecodeTargetNotPresent) + } + } + ep.Temporal = int32(ddParams.TemporalID) + ep.DependencyDescriptor = &buffer.ExtDependencyDescriptor{ + Descriptor: &dd.DependencyDescriptor{ + FirstPacketInFrame: ddParams.FirstPacketInFrame, + LastPacketInFrame: params.Marker, + FrameDependencies: &dd.FrameDependencyTemplate{ + TemporalId: ddParams.TemporalID, + DecodeTargetIndications: dtis, + FrameDiffs: ddParams.FrameDiffs, + }, + }, + DecodeTargets: ddParams.DecodeTargets, + ExtFrameNum: ddParams.ExtFrameNum, + } + return ep, nil +} + +// TestVP8DecodeTargets are the decode targets of libwebrtc's three layer VP8 structure, +// sorted from high to low. +var TestVP8DecodeTargets = []buffer.DependencyDescriptorDecodeTarget{ + {Target: 2, Layer: buffer.VideoLayer{Temporal: 2}}, + {Target: 1, Layer: buffer.VideoLayer{Temporal: 1}}, + {Target: 0, Layer: buffer.VideoLayer{Temporal: 0}}, +} + +// -------------------------------------- + var TestVP8Codec = webrtc.RTPCodecCapability{ MimeType: "video/vp8", ClockRate: 90000, diff --git a/pkg/sfu/videolayerselector/temporallayerselector/dependencydescriptor.go b/pkg/sfu/videolayerselector/temporallayerselector/dependencydescriptor.go new file mode 100644 index 000000000..1c0641669 --- /dev/null +++ b/pkg/sfu/videolayerselector/temporallayerselector/dependencydescriptor.go @@ -0,0 +1,125 @@ +// 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 ( + "github.com/livekit/livekit-server/pkg/sfu/buffer" + dd "github.com/livekit/livekit-server/pkg/sfu/rtpextension/dependencydescriptor" + "github.com/livekit/protocol/logger" +) + +// frames tracked behind the highest frame number, more than temporal references reach +const ddFrameWindow = 64 + +// DependencyDescriptor switches temporal layers with the dependency descriptor alone, for +// a stream with one spatial layer. It only picks the layer, the codec munger drops the +// packets above it. libwebrtc strips the VP8 payload descriptor when it sends the +// dependency descriptor, so for VP8 the temporal ID and the layer sync bit are only here. +type DependencyDescriptor struct { + logger logger.Logger + + initialized bool + highestFrame uint64 + forwardedMask uint64 // bit i: frame highestFrame - i was forwarded +} + +func NewDependencyDescriptor(logger logger.Logger) *DependencyDescriptor { + return &DependencyDescriptor{ + logger: logger, + } +} + +func (d *DependencyDescriptor) Select(extPkt *buffer.ExtPacket, current int32, target int32) (this int32, next int32) { + this = current + next = current + + ddwdt := extPkt.DependencyDescriptor + if ddwdt == nil || ddwdt.Descriptor == nil || ddwdt.Descriptor.FrameDependencies == nil { + return + } + + // A late first packet, a retransmission for example, starts a frame whose other + // packets were already decided. A key frame is the exception: it references nothing, + // and it restarts the frame numbers on a stream restart. + isFrameStart := ddwdt.Descriptor.FirstPacketInFrame && !extPkt.IsOutOfOrder + switch { + case current < target && extPkt.IsKeyFrame: + this = target + next = target + + case current < target && isFrameStart: + // A switch point of a decode target makes all later frames of that target + // decodable if this frame is decodable, so also every frame this one references + // has to have been forwarded. + if !d.areForwarded(ddwdt.ExtFrameNum, ddwdt.Descriptor.FrameDependencies.FrameDiffs) { + break + } + dtis := ddwdt.Descriptor.FrameDependencies.DecodeTargetIndications + for _, dt := range ddwdt.DecodeTargets { // sorted from high to low + if dt.Layer.Temporal <= current || dt.Layer.Temporal > target || dt.Target >= len(dtis) { + continue + } + if dtis[dt.Target] == dd.DecodeTargetSwitch { + this = dt.Layer.Temporal + next = dt.Layer.Temporal + break + } + } + + case current > target && extPkt.Packet.Marker: + next = target + } + + if extPkt.IsKeyFrame || isFrameStart { + d.onFrame(ddwdt.ExtFrameNum, extPkt.IsKeyFrame, extPkt.Temporal <= this) + } + return +} + +func (d *DependencyDescriptor) onFrame(frame uint64, isKeyFrame bool, isForwarded bool) { + switch { + case isKeyFrame || !d.initialized || frame >= d.highestFrame+ddFrameWindow: + d.initialized = true + d.highestFrame = frame + d.forwardedMask = 0 + + case frame > d.highestFrame: + d.forwardedMask <<= frame - d.highestFrame + d.highestFrame = frame + + case d.highestFrame-frame >= ddFrameWindow: + return + } + + if isForwarded { + d.forwardedMask |= 1 << (d.highestFrame - frame) + } +} + +func (d *DependencyDescriptor) areForwarded(frame uint64, frameDiffs []int) bool { + if !d.initialized { + return false + } + for _, diff := range frameDiffs { + if diff <= 0 || uint64(diff) > frame { + return false + } + ref := frame - uint64(diff) + if ref > d.highestFrame || d.highestFrame-ref >= ddFrameWindow || d.forwardedMask&(1<<(d.highestFrame-ref)) == 0 { + return false + } + } + return true +} diff --git a/pkg/sfu/videolayerselector/temporallayerselector/dependencydescriptor_test.go b/pkg/sfu/videolayerselector/temporallayerselector/dependencydescriptor_test.go new file mode 100644 index 000000000..319ecc9e6 --- /dev/null +++ b/pkg/sfu/videolayerselector/temporallayerselector/dependencydescriptor_test.go @@ -0,0 +1,184 @@ +// 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" + + "github.com/livekit/livekit-server/pkg/sfu/buffer" + "github.com/livekit/livekit-server/pkg/sfu/testutils" + "github.com/livekit/protocol/logger" +) + +type vp8DDFrame struct { + tid int + dtis string + frameDiffs []int + keyFrame bool +} + +// libwebrtc's three layer VP8 pattern: T0 refreshes "last", T1 refreshes "golden", and +// frame diffs point at the frames in those buffers +var vp8DDPattern = []vp8DDFrame{ + {tid: 0, dtis: "SSS", frameDiffs: []int{4}}, + {tid: 2, dtis: "--D", frameDiffs: []int{1}}, + {tid: 1, dtis: "-SS", frameDiffs: []int{2}}, + {tid: 2, dtis: "--D", frameDiffs: []int{1, 3}}, + {tid: 0, dtis: "SRR", frameDiffs: []int{4}}, + {tid: 2, dtis: "--D", frameDiffs: []int{1, 3}}, + {tid: 1, dtis: "-DS", frameDiffs: []int{2, 4}}, + {tid: 2, dtis: "--D", frameDiffs: []int{1, 3}}, +} + +func vp8DDPacket(t *testing.T, frameNum uint64, f vp8DDFrame, first bool, marker bool) *buffer.ExtPacket { + extPkt, err := testutils.GetTestExtPacketVP8DD( + &testutils.TestExtPacketParams{SequenceNumber: uint16(frameNum), PayloadSize: 20, Marker: marker, IsKeyFrame: f.keyFrame}, + &testutils.TestDDParams{ + FirstPacketInFrame: first, + ExtFrameNum: frameNum, + TemporalID: f.tid, + DTIs: f.dtis, + FrameDiffs: f.frameDiffs, + DecodeTargets: testutils.TestVP8DecodeTargets, + }, + ) + require.NoError(t, err) + return extPkt +} + +// runDependencyDescriptor sends one packet per frame: a key frame, the pattern at the start target until +// frame switchAt, then at target 2. It returns the frame number of the up-switch. +func runDependencyDescriptor(t *testing.T, startTarget int32, switchAt uint64) (uint64, bool) { + v := NewDependencyDescriptor(logger.GetLogger()) + current := int32(0) + for frameNum := uint64(0); frameNum < 40; frameNum++ { + f := vp8DDPattern[frameNum%uint64(len(vp8DDPattern))] + if frameNum == 0 { + f = vp8DDFrame{tid: 0, dtis: "SSS", keyFrame: true} + } + target := startTarget + if frameNum >= switchAt { + target = 2 + } + this, next := v.Select(vp8DDPacket(t, frameNum, f, true, true), current, target) + if frameNum >= switchAt && next == 2 { + require.Equal(t, int32(2), this) + return frameNum, true + } + current = next + } + return 0, false +} + +func TestDependencyDescriptorSelect(t *testing.T) { + t.Run("from TL0, up-switch at the next T0 switch point", func(t *testing.T) { + // frame 14 is the "-DS" T1 frame, it references T1 frame 10 that was dropped + frameNum, ok := runDependencyDescriptor(t, 0, 12) + require.True(t, ok) + require.Equal(t, uint64(16), frameNum) + + // frame 10 is the "-SS" T1 frame, it references only T0 + frameNum, ok = runDependencyDescriptor(t, 0, 10) + require.True(t, ok) + require.Equal(t, uint64(10), frameNum) + }) + + t.Run("from TL1, the -DS frame is a switch point", func(t *testing.T) { + frameNum, ok := runDependencyDescriptor(t, 1, 14) + require.True(t, ok) + require.Equal(t, uint64(14), frameNum) + }) + + t.Run("switch point that references a dropped frame", func(t *testing.T) { + // libwebrtc L1T4: a T2 frame that is a switch point for TL2 but references the + // previous T2 frame + v := NewDependencyDescriptor(logger.GetLogger()) + _, current := v.Select(vp8DDPacket(t, 0, vp8DDFrame{tid: 0, dtis: "SSS", keyFrame: true}, true, true), 1, 1) + _, current = v.Select(vp8DDPacket(t, 1, vp8DDFrame{tid: 2, dtis: "--D", frameDiffs: []int{1}}, true, true), current, 1) + this, next := v.Select(vp8DDPacket(t, 2, vp8DDFrame{tid: 2, dtis: "--S", frameDiffs: []int{1}}, true, true), current, 2) + require.Equal(t, int32(1), this) + require.Equal(t, int32(1), next) + }) + + t.Run("reference lost upstream", func(t *testing.T) { + v := NewDependencyDescriptor(logger.GetLogger()) + _, current := v.Select(vp8DDPacket(t, 0, vp8DDFrame{tid: 0, dtis: "SSS", keyFrame: true}, true, true), 0, 0) + // frame 1 never arrives + this, next := v.Select(vp8DDPacket(t, 2, vp8DDFrame{tid: 1, dtis: "-SS", frameDiffs: []int{1}}, true, true), current, 2) + require.Equal(t, int32(0), this) + require.Equal(t, int32(0), next) + }) + + t.Run("up-switch only on the first packet of a frame, in order", func(t *testing.T) { + v := NewDependencyDescriptor(logger.GetLogger()) + _, current := v.Select(vp8DDPacket(t, 0, vp8DDFrame{tid: 0, dtis: "SSS", keyFrame: true}, true, true), 0, 0) + f := vp8DDFrame{tid: 1, dtis: "-SS", frameDiffs: []int{1}} + + _, next := v.Select(vp8DDPacket(t, 1, f, false, true), current, 2) + require.Equal(t, int32(0), next, "second packet") + + late := vp8DDPacket(t, 1, f, true, true) + late.IsOutOfOrder = true + _, next = v.Select(late, current, 2) + require.Equal(t, int32(0), next, "late first packet") + + this, next := v.Select(vp8DDPacket(t, 1, f, true, true), current, 2) + require.Equal(t, int32(2), this) + require.Equal(t, int32(2), next) + }) + + t.Run("late key frame packet of a stream restart", func(t *testing.T) { + v := NewDependencyDescriptor(logger.GetLogger()) + _, current := v.Select(vp8DDPacket(t, 100, vp8DDFrame{tid: 0, dtis: "SSS", keyFrame: true}, true, true), 0, 0) + + // the restart moves the frame numbers back, and its key frame arrives out of order + keyFrame := vp8DDPacket(t, 5, vp8DDFrame{tid: 0, dtis: "SSS", keyFrame: true}, true, true) + keyFrame.IsOutOfOrder = true + _, current = v.Select(keyFrame, current, 0) + + this, next := v.Select(vp8DDPacket(t, 6, vp8DDFrame{tid: 1, dtis: "-SS", frameDiffs: []int{1}}, true, true), current, 2) + require.Equal(t, int32(2), this) + require.Equal(t, int32(2), next) + }) + + t.Run("key frame", func(t *testing.T) { + v := NewDependencyDescriptor(logger.GetLogger()) + this, next := v.Select(vp8DDPacket(t, 0, vp8DDFrame{tid: 0, dtis: "SSS", keyFrame: true}, true, false), 0, 2) + require.Equal(t, int32(2), this) + require.Equal(t, int32(2), next) + }) + + t.Run("no dependency descriptor", func(t *testing.T) { + v := NewDependencyDescriptor(logger.GetLogger()) + extPkt := vp8DDPacket(t, 0, vp8DDFrame{tid: 0, dtis: "SSS", keyFrame: true}, true, true) + extPkt.DependencyDescriptor = nil + this, next := v.Select(extPkt, 0, 2) + require.Equal(t, int32(0), this) + require.Equal(t, int32(0), next) + }) + + t.Run("down-switch at the end of a frame", func(t *testing.T) { + v := NewDependencyDescriptor(logger.GetLogger()) + f := vp8DDFrame{tid: 2, dtis: "--D", frameDiffs: []int{1}} + this, next := v.Select(vp8DDPacket(t, 1, f, true, false), 2, 0) + require.Equal(t, int32(2), this) + require.Equal(t, int32(2), next) + this, next = v.Select(vp8DDPacket(t, 1, f, false, true), 2, 0) + require.Equal(t, int32(2), this) + require.Equal(t, int32(0), next) + }) +}