diff --git a/pkg/sfu/buffer/dependencydescriptorparser.go b/pkg/sfu/buffer/dependencydescriptorparser.go index 343eafc24..332c38a28 100644 --- a/pkg/sfu/buffer/dependencydescriptorparser.go +++ b/pkg/sfu/buffer/dependencydescriptorparser.go @@ -69,6 +69,7 @@ type DependencyDescriptorParser struct { // repeating the same structure (e. g. screen content), valid late/retransmitted // frames would be dropped as "earlier than current structure". structureChangeExtFrameNum uint64 + hasActiveDecodeTargets bool // the first mask is taken whatever its sequence number activeDecodeTargetsExtSeq uint64 activeDecodeTargetsMask uint32 frameChecker *FrameIntegrityChecker @@ -99,8 +100,11 @@ type ExtDependencyDescriptor struct { DecodeTargets []DependencyDescriptorDecodeTarget StructureUpdated bool ActiveDecodeTargetsUpdated bool - Integrity bool - ExtFrameNum uint64 + // the stream's current active decode targets, on every packet, also when the mask + // changed on a packet the subscriber does not get + ActiveDecodeTargetsMask uint32 + Integrity bool + ExtFrameNum uint64 // the frame number of the keyframe which the current frame depends on ExtKeyFrameNum uint64 @@ -262,7 +266,8 @@ func (r *DependencyDescriptorParser) Parse(pkt *rtp.Packet) (*ExtDependencyDescr // so don't need to notify max layer change here. } - if mask := ddVal.ActiveDecodeTargetsBitmask; mask != nil && extSeq > r.activeDecodeTargetsExtSeq { + if mask := ddVal.ActiveDecodeTargetsBitmask; mask != nil && (!r.hasActiveDecodeTargets || extSeq > r.activeDecodeTargetsExtSeq) { + r.hasActiveDecodeTargets = true r.activeDecodeTargetsExtSeq = extSeq if *mask != r.activeDecodeTargetsMask { r.activeDecodeTargetsMask = *mask @@ -285,6 +290,7 @@ func (r *DependencyDescriptorParser) Parse(pkt *rtp.Packet) (*ExtDependencyDescr extDD.DecodeTargets = r.decodeTargets extDD.ExtKeyFrameNum = r.structureExtFrameNum + extDD.ActiveDecodeTargetsMask = r.activeDecodeTargetsMask return extDD, videoLayer, nil } @@ -294,6 +300,7 @@ func (r *DependencyDescriptorParser) restart() { r.structure = nil r.structureExtFrameNum = 0 r.structureChangeExtFrameNum = 0 + r.hasActiveDecodeTargets = false r.activeDecodeTargetsExtSeq = 0 r.activeDecodeTargetsMask = 0 r.decodeTargets = r.decodeTargets[:0] diff --git a/pkg/sfu/buffer/dependencydescriptorparser_test.go b/pkg/sfu/buffer/dependencydescriptorparser_test.go index 7a09b013c..370c53023 100644 --- a/pkg/sfu/buffer/dependencydescriptorparser_test.go +++ b/pkg/sfu/buffer/dependencydescriptorparser_test.go @@ -75,19 +75,34 @@ func (f *ddTestFeeder) deltaFrame(frameNumber uint16) (*ExtDependencyDescriptor, return f.feed(frameNumber, f.structure.Templates[1], nil) } +// deltaFrameWithMask sends a delta frame that changes the active decode targets +func (f *ddTestFeeder) deltaFrameWithMask(frameNumber uint16, activeDecodeTargets uint32) (*ExtDependencyDescriptor, error) { + return f.feedWithMask(frameNumber, f.structure.Templates[1], nil, &activeDecodeTargets) +} + func (f *ddTestFeeder) feed( frameNumber uint16, template *dd.FrameDependencyTemplate, attachedStructure *dd.FrameDependencyStructure, +) (*ExtDependencyDescriptor, error) { + return f.feedWithMask(frameNumber, template, attachedStructure, nil) +} + +func (f *ddTestFeeder) feedWithMask( + frameNumber uint16, + template *dd.FrameDependencyTemplate, + attachedStructure *dd.FrameDependencyStructure, + activeDecodeTargets *uint32, ) (*ExtDependencyDescriptor, error) { f.t.Helper() ddVal := &dd.DependencyDescriptor{ - FirstPacketInFrame: true, - LastPacketInFrame: true, - FrameNumber: frameNumber, - FrameDependencies: template, - AttachedStructure: attachedStructure, + FirstPacketInFrame: true, + LastPacketInFrame: true, + FrameNumber: frameNumber, + FrameDependencies: template, + AttachedStructure: attachedStructure, + ActiveDecodeTargetsBitmask: activeDecodeTargets, } buf, err := (&dd.DependencyDescriptorExtension{Descriptor: ddVal, Structure: f.structure}).Marshal() require.NoError(f.t, err) @@ -187,6 +202,45 @@ func TestDependencyDescriptorParserExtKeyFrameNum(t *testing.T) { require.EqualValues(t, 4, extDD.ExtKeyFrameNum) } +// every packet carries the stream's current active decode targets, also after the mask +// changed on a packet the forwarder dropped +func TestDependencyDescriptorParserActiveDecodeTargetsMask(t *testing.T) { + f := newDDTestFeeder(t) + structure := newL1T1Structure(0) + + extDD, err := f.keyFrame(0, structure) + require.NoError(t, err) + require.EqualValues(t, 1, extDD.ActiveDecodeTargetsMask) + + extDD, err = f.deltaFrameWithMask(1, 0) + require.NoError(t, err) + require.EqualValues(t, 0, extDD.ActiveDecodeTargetsMask) + + extDD, err = f.deltaFrame(2) + require.NoError(t, err) + require.EqualValues(t, 0, extDD.ActiveDecodeTargetsMask) + + extDD, err = f.keyFrame(3, structure) + require.NoError(t, err) + require.EqualValues(t, 1, extDD.ActiveDecodeTargetsMask) +} + +// a first key frame at RTP sequence number 0 still makes its structure's decode targets +// active +func TestDependencyDescriptorParserActiveDecodeTargetsMaskAtSequenceZero(t *testing.T) { + f := newDDTestFeeder(t) + f.seq = 0xffff // the first packet goes out with sequence number 0 + structure := newL1T1Structure(0) + + extDD, err := f.keyFrame(0, structure) + require.NoError(t, err) + require.EqualValues(t, 1, extDD.ActiveDecodeTargetsMask) + + extDD, err = f.deltaFrame(1) + require.NoError(t, err) + require.EqualValues(t, 1, extDD.ActiveDecodeTargetsMask) +} + // a pooled ExtDependencyDescriptor carries the storage for its descriptor, so parsing a delta // frame must not allocate and a reused descriptor must not show the previous packet's values func TestDependencyDescriptorParserNoAlloc(t *testing.T) { diff --git a/pkg/sfu/forwarder.go b/pkg/sfu/forwarder.go index f0eed8824..ef5b0548a 100644 --- a/pkg/sfu/forwarder.go +++ b/pkg/sfu/forwarder.go @@ -281,6 +281,11 @@ type Forwarder struct { rtpMunger *RTPMunger vls videolayerselector.VideoLayerSelector + // set for VP8 with the dependency descriptor, forwards the descriptor to the subscriber + ddForwarder *videolayerselector.DDForwarder + // temporal layer of the packet being translated, the selector's current layer is + // already the next one on the last packet before a down switch + packetTemporal int32 codecMunger codecmunger.CodecMunger } @@ -413,11 +418,15 @@ func (f *Forwarder) DetermineCodec(codec webrtc.RTPCodecCapability, extensions [ return false } + prevDDForwarder := f.ddForwarder + 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) } else { f.codecMunger = codecmunger.NewVP8FromOther(f.codecMunger, f.logger) } @@ -2193,6 +2202,7 @@ func (f *Forwarder) getTranslationParamsVideo(extPkt *buffer.ExtPacket, layer in // 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.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 @@ -2261,6 +2271,7 @@ func (f *Forwarder) getTranslationParamsVideo(extPkt *buffer.ExtPacket, layer in tp.ddBytes, tp.ddBytesLen, tp.ddBytesSpill = result.DDBytes, result.DDBytesLen, result.DDBytesSpill tp.marker = result.RTPMarker tp.isEndOfLayerFrame = f.isEndOfLayerFrame(extPkt) + f.packetTemporal = f.vls.GetCurrent().Temporal starting, err := f.getTranslationParamsCommon(extPkt, layer, &tp) tp.isStarting = starting @@ -2291,12 +2302,21 @@ func (f *Forwarder) getTranslationParamsVideo(extPkt *buffer.ExtPacket, layer in return tp, nil } + if f.ddForwarder != nil { + // after translateCodecHeader, which picks the packet's temporal layer + var ddResult videolayerselector.VideoLayerSelectorResult + if f.ddForwarder.Forward(extPkt, f.packetTemporal, &ddResult) { + tp.ddBytes, tp.ddBytesLen, tp.ddBytesSpill = ddResult.DDBytes, ddResult.DDBytesLen, ddResult.DDBytesSpill + } + } + return tp, nil } func (f *Forwarder) translateCodecHeader(extPkt *buffer.ExtPacket, tp *TranslationParams) error { // codec specific forwarding check and any needed packet munging tl := f.vls.SelectTemporal(extPkt) + f.packetTemporal = tl inputSize, codecBytes, codecBytesLen, err := f.codecMunger.UpdateAndGet( extPkt, tp.rtp.snOrdering == SequenceNumberOrderingOutOfOrder, diff --git a/pkg/sfu/forwarder_test.go b/pkg/sfu/forwarder_test.go index 6d82db34e..36f2e92b7 100644 --- a/pkg/sfu/forwarder_test.go +++ b/pkg/sfu/forwarder_test.go @@ -15,6 +15,7 @@ package sfu import ( + "encoding/hex" "testing" "github.com/pion/rtp/codecs" @@ -2114,6 +2115,80 @@ 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 + 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) + + 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 + } + + 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: 0}) + + // the key frame carries the structure, too big to hold inline + tp, err := f.GetTranslationParams(packet(10, 100, true), 0) + require.NoError(t, err) + require.False(t, tp.shouldDrop) + require.NotEmpty(t, tp.ddBytesSpill) + + tp, err = f.GetTranslationParams(packet(11, 101, false), 0) + require.NoError(t, err) + require.False(t, tp.shouldDrop) + require.NotEmpty(t, tp.ddInline()) + + // a restarted upstream receiver sets the codec again, the frame numbers continue on + // the same downtrack + f.DetermineCodec(testutils.TestVP8Codec, []webrtc.RTPHeaderExtensionParameter{{URI: dd.ExtensionURI}}, livekit.VideoLayer_MODE_UNUSED) + tp, err = f.GetTranslationParams(packet(12, 50, true), 0) + require.NoError(t, err) + require.False(t, tp.shouldDrop) + out := dd.DependencyDescriptor{} + _, err = (&dd.DependencyDescriptorExtension{Descriptor: &out, Structure: keyFrameDescriptor.AttachedStructure}).Unmarshal(tp.ddBytesSpill) + require.NoError(t, err) + require.Equal(t, uint16(102), out.FrameNumber) + + // the last packet of a TL2 frame that starts a down switch to TL0 is forwarded at TL2, + // its descriptor names the TL2 decode target + f.vls.SetCurrent(buffer.VideoLayer{Spatial: 0, Temporal: 2}) + f.vls.SetTarget(buffer.VideoLayer{Spatial: 0, Temporal: 0}) + tl2 := packet(13, 51, false) + tl2.Temporal = 2 + tp, err = f.GetTranslationParams(tl2, 0) + require.NoError(t, err) + require.False(t, tp.shouldDrop) + require.Equal(t, int32(0), f.vls.GetCurrent().Temporal, "the switch takes effect after this packet") + out = dd.DependencyDescriptor{} + _, err = (&dd.DependencyDescriptorExtension{Descriptor: &out, Structure: keyFrameDescriptor.AttachedStructure}).Unmarshal(tp.ddInline()) + require.NoError(t, err) + require.Equal(t, uint32(1<<0|1<<1|1<<2), *out.ActiveDecodeTargetsBitmask) + + // without the extension the forwarder writes no descriptor + f = NewForwarder(webrtc.RTPCodecTypeVideo, logger.GetLogger(), true, true, true, nil) + f.DetermineCodec(testutils.TestVP8Codec, nil, livekit.VideoLayer_MODE_UNUSED) + require.Nil(t, f.ddForwarder) +} + func TestForwarderPadDroppedSlot(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) diff --git a/pkg/sfu/testutils/data.go b/pkg/sfu/testutils/data.go index 1056e4119..6ac6a9f2b 100644 --- a/pkg/sfu/testutils/data.go +++ b/pkg/sfu/testutils/data.go @@ -143,6 +143,8 @@ func GetTestExtPacketVP8DD(params *TestExtPacketParams, ddParams *TestDDParams) }, DecodeTargets: ddParams.DecodeTargets, ExtFrameNum: ddParams.ExtFrameNum, + // all decode targets active, as the parser reports after a structure + ActiveDecodeTargetsMask: 1< d.extLastFrameNum { + d.hasLast = true + d.extLastFrameNum = extOutFrameNum + } + + // tell the subscriber which decode targets to expect: the publisher's active ones up to + // the forwarded temporal layer. On a key frame too, where the structure alone would make + // all of them active. + d.activeDecodeTargetsBitmask = 0 + for _, dt := range ddwdt.DecodeTargets { + if dt.Layer.Temporal <= temporal { + d.activeDecodeTargetsBitmask |= 1 << dt.Target + } + } + d.activeDecodeTargetsBitmask &= ddwdt.ActiveDecodeTargetsMask + d.descriptor = *descriptor + d.descriptor.FrameNumber = uint16(extOutFrameNum) + d.descriptor.ActiveDecodeTargetsBitmask = &d.activeDecodeTargetsBitmask + + if err := result.marshalDependencyDescriptorExtension(&dd.DependencyDescriptorExtension{ + Descriptor: &d.descriptor, + Structure: d.structure, + }); err != nil { + d.logger.Warnw("could not marshal dependency descriptor", err, "extFrameNum", extFrameNum) + return false + } + return true +} + +// rebase moves the frame numbers up when a new stream would go back past the last one sent +func (d *DDForwarder) rebase(extFrameNum uint64) { + if d.hasLast && extFrameNum+d.frameNumOffset <= d.extLastFrameNum { + d.frameNumOffset = d.extLastFrameNum + 1 - extFrameNum + } +} diff --git a/pkg/sfu/videolayerselector/ddforwarder_test.go b/pkg/sfu/videolayerselector/ddforwarder_test.go new file mode 100644 index 000000000..cc4a47a4a --- /dev/null +++ b/pkg/sfu/videolayerselector/ddforwarder_test.go @@ -0,0 +1,234 @@ +// 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 videolayerselector + +import ( + "encoding/hex" + "testing" + + "github.com/pion/rtp" + "github.com/stretchr/testify/require" + + "github.com/livekit/livekit-server/pkg/sfu/buffer" + dd "github.com/livekit/livekit-server/pkg/sfu/rtpextension/dependencydescriptor" + "github.com/livekit/protocol/logger" +) + +// ddForwarderTest builds packets with the L3T3 fixture descriptor and decodes what a +// DDForwarder writes +type ddForwarderTest struct { + t *testing.T + keyFrameDescriptor dd.DependencyDescriptor + structure *dd.FrameDependencyStructure + decodeTargets []buffer.DependencyDescriptorDecodeTarget +} + +func newDDForwarderTest(t *testing.T) *ddForwarderTest { + raw, err := hex.DecodeString(dependencyDescriptorFixture) + require.NoError(t, err) + keyFrameDescriptor := dd.DependencyDescriptor{} + _, err = (&dd.DependencyDescriptorExtension{Descriptor: &keyFrameDescriptor}).Unmarshal(raw) + require.NoError(t, err) + require.NotNil(t, keyFrameDescriptor.AttachedStructure) + + return &ddForwarderTest{ + t: t, + keyFrameDescriptor: keyFrameDescriptor, + structure: keyFrameDescriptor.AttachedStructure, + decodeTargets: []buffer.DependencyDescriptorDecodeTarget{ + {Target: 2, Layer: buffer.VideoLayer{Temporal: 2}}, + {Target: 1, Layer: buffer.VideoLayer{Temporal: 1}}, + {Target: 0, Layer: buffer.VideoLayer{Temporal: 0}}, + }, + } +} + +func (d *ddForwarderTest) packet(ssrc uint32, extFrameNum uint64, isKeyFrame bool, restartGeneration int) *buffer.ExtPacket { + descriptor := d.keyFrameDescriptor + descriptor.FrameNumber = uint16(extFrameNum) + if !isKeyFrame { + descriptor.AttachedStructure = nil + descriptor.ActiveDecodeTargetsBitmask = nil + } + return &buffer.ExtPacket{ + Packet: &rtp.Packet{Header: rtp.Header{SSRC: ssrc}}, + IsKeyFrame: isKeyFrame, + DependencyDescriptor: &buffer.ExtDependencyDescriptor{ + Descriptor: &descriptor, + DecodeTargets: d.decodeTargets, + ExtFrameNum: extFrameNum, + RestartGeneration: restartGeneration, + // all decode targets active, as the parser reports after a structure + ActiveDecodeTargetsMask: 1<