sfu: forward the dependency descriptor for VP8 simulcast (#4987)

* sfu: forward the dependency descriptor for VP8 simulcast

libwebrtc sends VP8 with a one byte payload descriptor when it sends the
dependency descriptor, so a VP8 subscriber gets neither picture IDs nor the
descriptor and finds frame references by sequence number alone. Under loss
with temporal filtering it stalls until a key frame.

DDForwarder rewrites the publisher's descriptor for each forwarded packet:
frame numbers continue across simulcast layer switches and stream restarts,
the structure goes out with the current layer's key frame, and the active
decode targets follow the forwarded temporal layer.

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

* sfu: keep VP8 descriptor frame numbers from going back

Rebase frame numbers only when a new stream would go back past the last
number sent: on a layer switch, a stream restart (also when it starts with a
delta frame) and a new forwarder on the same downtrack, which now carries
over the last number sent.

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

* sfu: set VP8 descriptor decode targets from the packet's layer

The active decode targets use the temporal layer the packet is forwarded at,
not the selector's next layer on the last packet before a down switch. They
also go out on key frames, combined with the publisher's own mask.

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

* sfu: carry the current active decode targets on every DD packet

The parser puts the stream's current mask on every packet, so a mask change
on a dropped packet still reaches the VP8 descriptor forwarder, which no
longer tracks the publisher's mask itself.

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

* sfu: take the first DD active decode targets mask at any sequence number

A publisher whose first packet has RTP sequence number 0 never had its key
frame's mask applied, so every packet reported no active decode targets.

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 12:08:49 +05:30
committed by GitHub
co-authored by Claude Opus 5.5
parent b1c801b979
commit 099e03fdf8
7 changed files with 541 additions and 8 deletions
+10 -3
View File
@@ -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]
@@ -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) {
+20
View File
@@ -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,
+75
View File
@@ -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)
+2
View File
@@ -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<<len(ddParams.DecodeTargets) - 1,
}
return ep, nil
}
+141
View File
@@ -0,0 +1,141 @@
// 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 (
"github.com/livekit/livekit-server/pkg/sfu/buffer"
dd "github.com/livekit/livekit-server/pkg/sfu/rtpextension/dependencydescriptor"
"github.com/livekit/protocol/logger"
)
// DDForwarder forwards the dependency descriptor of a simulcast stream. Without it, a
// subscriber of a stream without picture IDs, VP8 from libwebrtc for example, finds frame
// references by sequence number alone and stalls on every sequence number it cannot account
// for.
//
// libwebrtc numbers frames with one counter for all simulcast layers of a sender, so the
// frame numbers of a new layer are close to those of the old one. The numbers sent to the
// subscriber must still never go back: a key frame older than the last structure the
// subscriber received is dropped. So whenever the stream changes, a layer switch, a stream
// restart or a new forwarder, the numbers are moved up past the last one sent if needed.
// Otherwise they pass through with the current offset, so frame diffs and chain diffs stay
// valid. A migrated subscriber starts with the publisher's numbers, which continue its old
// ones in most cases but not all, as the last number sent is not part of the migrated state.
type DDForwarder struct {
logger logger.Logger
initialized bool // a key frame with a structure set the stream
ssrc uint32
restartGeneration int
structure *dd.FrameDependencyStructure
frameNumOffset uint64 // added to the incoming extended frame number
hasLast bool
extLastFrameNum uint64 // highest outgoing extended frame number
descriptor dd.DependencyDescriptor
activeDecodeTargetsBitmask uint32
}
func NewDDForwarder(logger logger.Logger) *DDForwarder {
return &DDForwarder{
logger: logger,
}
}
// NewDDForwarderFromOther continues the frame numbers of a forwarder it replaces on the
// same downtrack, a restarted upstream receiver for example.
func NewDDForwarderFromOther(other *DDForwarder, logger logger.Logger) *DDForwarder {
d := NewDDForwarder(logger)
if other != nil && other.hasLast {
d.hasLast = true
d.extLastFrameNum = other.extLastFrameNum
}
return d
}
// Forward marshals the descriptor of a forwarded packet into result. temporal is the
// temporal layer the packet is forwarded at. It returns false when there is nothing to
// send, before the first key frame of the stream for example.
func (d *DDForwarder) Forward(extPkt *buffer.ExtPacket, temporal int32, result *VideoLayerSelectorResult) bool {
ddwdt := extPkt.DependencyDescriptor
if ddwdt == nil || ddwdt.Descriptor == nil || ddwdt.Descriptor.FrameDependencies == nil {
return false
}
descriptor := ddwdt.Descriptor
extFrameNum := ddwdt.ExtFrameNum
hasStructure := extPkt.IsKeyFrame && 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
if !hasStructure {
return false
}
d.rebase(extFrameNum)
d.initialized = true
d.ssrc = extPkt.Packet.SSRC
d.restartGeneration = ddwdt.RestartGeneration
case ddwdt.RestartGeneration != d.restartGeneration:
// the restarted stream numbers its frames from a new start and has its own structure,
// the first packet can be a delta frame
d.rebase(extFrameNum)
d.restartGeneration = ddwdt.RestartGeneration
d.structure = nil
}
if hasStructure {
d.structure = descriptor.AttachedStructure
}
if d.structure == nil {
return false
}
extOutFrameNum := extFrameNum + d.frameNumOffset
if !d.hasLast || extOutFrameNum > 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
}
}
@@ -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<<len(d.decodeTargets) - 1,
},
}
}
// forward returns the decoded descriptor, nil when f writes none
func (d *ddForwarderTest) forward(f *DDForwarder, extPkt *buffer.ExtPacket, temporal int32) *dd.DependencyDescriptor {
result := VideoLayerSelectorResult{}
if !f.Forward(extPkt, temporal, &result) {
return nil
}
ddBytes := result.DDBytesSpill
if ddBytes == nil {
ddBytes = result.DDBytes[:result.DDBytesLen]
}
out := &dd.DependencyDescriptor{}
_, err := (&dd.DependencyDescriptorExtension{Descriptor: out, Structure: d.structure}).Unmarshal(ddBytes)
require.NoError(d.t, err)
return out
}
func TestDDForwarder(t *testing.T) {
d := newDDForwarderTest(t)
f := NewDDForwarder(logger.GetLogger())
// nothing before the first key frame, there is no structure to write it with
require.Nil(t, d.forward(f, d.packet(1, 999, false, 0), 0))
// the first key frame keeps the publisher's frame number and carries the structure,
// with the decode targets the subscriber gets, not all of them
out := d.forward(f, d.packet(1, 1000, true, 0), 0)
require.Equal(t, uint16(1000), out.FrameNumber)
require.NotNil(t, out.AttachedStructure)
require.Equal(t, uint32(1<<0), *out.ActiveDecodeTargetsBitmask)
// delta frames name the active decode targets, frames dropped in between leave gaps
out = d.forward(f, d.packet(1, 1001, false, 0), 0)
require.Equal(t, uint16(1001), out.FrameNumber)
require.Equal(t, uint32(1<<0), *out.ActiveDecodeTargetsBitmask)
out = d.forward(f, d.packet(1, 1003, false, 0), 1)
require.Equal(t, uint16(1003), out.FrameNumber)
require.Equal(t, uint32(1<<0|1<<1), *out.ActiveDecodeTargetsBitmask)
// a layer switch: the new layer's lower frame numbers continue after the last one
out = d.forward(f, d.packet(2, 40, true, 0), 0)
require.Equal(t, uint16(1004), out.FrameNumber)
require.NotNil(t, out.AttachedStructure)
out = d.forward(f, d.packet(2, 41, false, 0), 0)
require.Equal(t, uint16(1005), out.FrameNumber)
// a retransmitted key frame packet keeps its number
out = d.forward(f, d.packet(2, 40, true, 0), 0)
require.Equal(t, uint16(1004), out.FrameNumber)
// a late packet of the previous layer is not described
require.Nil(t, d.forward(f, d.packet(1, 1004, false, 0), 0))
// a restarted stream starts its frame numbers over, they still continue
out = d.forward(f, d.packet(2, 5, true, 1), 0)
require.Equal(t, uint16(1006), out.FrameNumber)
out = d.forward(f, d.packet(2, 6, false, 1), 0)
require.Equal(t, uint16(1007), out.FrameNumber)
}
func TestDDForwarderSharedFrameCounter(t *testing.T) {
// libwebrtc numbers the frames of all simulcast layers with one counter
d := newDDForwarderTest(t)
f := NewDDForwarder(logger.GetLogger())
require.Equal(t, uint16(100), d.forward(f, d.packet(1, 100, true, 0), 0).FrameNumber)
require.Equal(t, uint16(102), d.forward(f, d.packet(1, 102, false, 0), 0).FrameNumber)
// the new layer's key frame was encoded before the last frame sent, it moves up
require.Equal(t, uint16(103), d.forward(f, d.packet(2, 101, true, 0), 0).FrameNumber)
require.Equal(t, uint16(106), d.forward(f, d.packet(2, 104, false, 0), 0).FrameNumber)
// a key frame ahead of the last frame sent keeps the current offset
require.Equal(t, uint16(112), d.forward(f, d.packet(1, 110, true, 0), 0).FrameNumber)
}
func TestDDForwarderFromOther(t *testing.T) {
d := newDDForwarderTest(t)
f := NewDDForwarder(logger.GetLogger())
require.Equal(t, uint16(100), d.forward(f, d.packet(1, 100, true, 0), 0).FrameNumber)
require.Equal(t, uint16(103), d.forward(f, d.packet(1, 103, false, 0), 0).FrameNumber)
// a replacement on the same downtrack, a restarted upstream receiver for example, gets
// a key frame behind the last frame sent: it continues after the last frame sent
f = NewDDForwarderFromOther(f, logger.GetLogger())
require.Nil(t, d.forward(f, d.packet(1, 101, false, 0), 0))
require.Equal(t, uint16(104), d.forward(f, d.packet(1, 101, true, 0), 0).FrameNumber)
require.Equal(t, uint16(105), d.forward(f, d.packet(1, 102, false, 0), 0).FrameNumber)
// a replacement of nothing keeps the publisher's frame numbers
f = NewDDForwarderFromOther(nil, logger.GetLogger())
require.Equal(t, uint16(50), d.forward(f, d.packet(2, 50, true, 0), 0).FrameNumber)
}
func TestDDForwarderRestartOnDeltaFrame(t *testing.T) {
d := newDDForwarderTest(t)
f := NewDDForwarder(logger.GetLogger())
require.Equal(t, uint16(1000), d.forward(f, d.packet(1, 1000, true, 0), 0).FrameNumber)
require.Equal(t, uint16(1001), d.forward(f, d.packet(1, 1001, false, 0), 0).FrameNumber)
// the restarted stream's first packet is a delta frame: it has no structure to be
// written with, but its numbers already continue
require.Nil(t, d.forward(f, d.packet(1, 3, false, 1), 0))
require.Equal(t, uint16(1003), d.forward(f, d.packet(1, 4, true, 1), 0).FrameNumber)
require.Equal(t, uint16(1004), d.forward(f, d.packet(1, 5, false, 1), 0).FrameNumber)
}
func TestDDForwarderPublisherActiveDecodeTargets(t *testing.T) {
d := newDDForwarderTest(t)
f := NewDDForwarder(logger.GetLogger())
require.Equal(t, uint32(1<<0|1<<1|1<<2), *d.forward(f, d.packet(1, 100, true, 0), 2).ActiveDecodeTargetsBitmask)
// the publisher turned off its upper decode targets on a packet that was not forwarded,
// the next forwarded packet still reports it
delta := d.packet(1, 102, false, 0)
delta.DependencyDescriptor.ActiveDecodeTargetsMask = 1 << 0
require.Equal(t, uint32(1<<0), *d.forward(f, delta, 2).ActiveDecodeTargetsBitmask)
// and turned them on again
require.Equal(t, uint32(1<<0|1<<1|1<<2), *d.forward(f, d.packet(1, 103, true, 0), 2).ActiveDecodeTargetsBitmask)
}
func TestDDForwarderAllocations(t *testing.T) {
raw, err := hex.DecodeString(dependencyDescriptorFixture)
require.NoError(t, err)
keyFrameDescriptor := dd.DependencyDescriptor{}
_, err = (&dd.DependencyDescriptorExtension{Descriptor: &keyFrameDescriptor}).Unmarshal(raw)
require.NoError(t, err)
decodeTargets := []buffer.DependencyDescriptorDecodeTarget{{Target: 0, Layer: buffer.VideoLayer{Temporal: 0}}}
keyFrame := &buffer.ExtPacket{
Packet: &rtp.Packet{Header: rtp.Header{SSRC: 1}},
IsKeyFrame: true,
DependencyDescriptor: &buffer.ExtDependencyDescriptor{Descriptor: &keyFrameDescriptor, DecodeTargets: decodeTargets, ExtFrameNum: 100, ActiveDecodeTargetsMask: 1},
}
deltaDescriptor := keyFrameDescriptor
deltaDescriptor.AttachedStructure = nil
deltaDescriptor.ActiveDecodeTargetsBitmask = nil
delta := &buffer.ExtPacket{
Packet: &rtp.Packet{Header: rtp.Header{SSRC: 1}},
DependencyDescriptor: &buffer.ExtDependencyDescriptor{Descriptor: &deltaDescriptor, DecodeTargets: decodeTargets, ExtFrameNum: 101, ActiveDecodeTargetsMask: 1},
}
f := NewDDForwarder(logger.GetLogger())
result := VideoLayerSelectorResult{}
require.True(t, f.Forward(keyFrame, 0, &result))
// a delta frame's descriptor fits inline
allocs := testing.AllocsPerRun(100, func() {
delta.DependencyDescriptor.ExtFrameNum++
result = VideoLayerSelectorResult{}
if !f.Forward(delta, 0, &result) {
t.Fatal("not forwarded")
}
})
require.Zero(t, allocs)
}