sfu/buffer: parse the dependency descriptor without allocating (#4953)

* sfu/buffer: parse the dependency descriptor without allocating Every VP9 and AV1 SVC packet allocated six objects in the dependency descriptor parse: the descriptor, a clone of the frame template with its three slices, a resolution copy, and a frame entity that took the address of a parameter. The pooled ExtDependencyDescriptor now owns the storage for the descriptor, the template and the resolution, with fixed arrays for the template's slices. The reader copies the matching template into that storage. Delta frame parse: 6 allocs, 176 B, 174 ns -> 0 allocs, 0 B, 122 ns. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * sfu/dd: never copy a parsed frame into a structure template A caller can hand Unmarshal a descriptor whose FrameDependencies points at one of the structure's templates. Copying the packet's custom fields into it would change the template for every later packet, so the reader allocates a fresh template in that case. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> --------- Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
2026-10-15 09:00:14 +00:00 · 2026-10-06 19:46:20 +05:30
parent 514e54b1e0
commit bc849eb47d
7 changed files with 201 additions and 39 deletions
@@ -107,6 +107,15 @@ type ExtDependencyDescriptor struct {
 	// increase when the stream restarts, clear and reinitialize all dd state includes
 	// attached structure, frame chain, decode target.
 	RestartGeneration int
+
+	// storage for Descriptor, zeroed with the rest of the struct on release,
+	// a frame with more than 8 frame diffs spills to the heap
+	descriptor        dd.DependencyDescriptor
+	frameDependencies dd.FrameDependencyTemplate
+	resolution        dd.RenderResolution
+	dtis              [dd.MaxDecodeTargets]dd.DecodeTargetIndication
+	chainDiffs        [dd.MaxDecodeTargets]int
+	frameDiffs        [8]int
 }

 func (r *DependencyDescriptorParser) Parse(pkt *rtp.Packet) (*ExtDependencyDescriptor, VideoLayer, error) {
@@ -120,9 +129,19 @@ func (r *DependencyDescriptorParser) Parse(pkt *rtp.Packet) (*ExtDependencyDescr
 		return nil, videoLayer, ErrDDExtentionNotFound
 	}

-	var ddVal dd.DependencyDescriptor
+	extDD := ExtDependencyDescriptorFactory.Get().(*ExtDependencyDescriptor)
+	extDD.frameDependencies = dd.FrameDependencyTemplate{
+		DecodeTargetIndications: extDD.dtis[:0],
+		FrameDiffs:              extDD.frameDiffs[:0],
+		ChainDiffs:              extDD.chainDiffs[:0],
+	}
+	ddVal := &extDD.descriptor
+	*ddVal = dd.DependencyDescriptor{
+		FrameDependencies: &extDD.frameDependencies,
+		Resolution:        &extDD.resolution,
+	}
 	ext := &dd.DependencyDescriptorExtension{
-		Descriptor: &ddVal,
+		Descriptor: ddVal,
 		Structure:  r.structure,
 	}
 	_, err := ext.Unmarshal(ddBuf)
@@ -130,6 +149,7 @@ func (r *DependencyDescriptorParser) Parse(pkt *rtp.Packet) (*ExtDependencyDescr
 		if err != dd.ErrDDReaderNoStructure && err != dd.ErrDDReaderInvalidTemplateIndex {
 			r.logger.Infow("failed to parse generic dependency descriptor", err, "payload", pkt.PayloadType, "ddbufLen", len(ddBuf))
 		}
+		ReleaseExtDependencyDescriptor(extDD)
 		return nil, videoLayer, err
 	}

@@ -168,18 +188,16 @@ func (r *DependencyDescriptorParser) Parse(pkt *rtp.Packet) (*ExtDependencyDescr
 			"unwrappedFN", unwrapped,
 			"frameWrapAround", r.frameWrapAround,
 		)
+		ReleaseExtDependencyDescriptor(extDD)
 		return nil, videoLayer, ErrFrameEarlierThanKeyFrame
 	}

-	r.frameChecker.AddPacket(extSeq, extFN, &ddVal)
+	r.frameChecker.AddPacket(extSeq, extFN, ddVal)

-	extDD := ExtDependencyDescriptorFactory.Get().(*ExtDependencyDescriptor)
-	*extDD = ExtDependencyDescriptor{
-		Descriptor:        &ddVal,
-		ExtFrameNum:       extFN,
-		Integrity:         r.frameChecker.FrameIntegrity(extFN),
-		RestartGeneration: r.restartGeneration,
-	}
+	extDD.Descriptor = ddVal
+	extDD.ExtFrameNum = extFN
+	extDD.Integrity = r.frameChecker.FrameIntegrity(extFN)
+	extDD.RestartGeneration = r.restartGeneration

 	if ddVal.AttachedStructure != nil {
 		if !ddVal.FirstPacketInFrame {
@@ -186,3 +186,51 @@ func TestDependencyDescriptorParserExtKeyFrameNum(t *testing.T) {
 	require.NoError(t, err)
 	require.EqualValues(t, 4, extDD.ExtKeyFrameNum)
 }
+
+// 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) {
+	f := newDDTestFeeder(t)
+	structure := newL1T1Structure(0)
+	structure.Resolutions = []dd.RenderResolution{{Width: 640, Height: 360}}
+	extDD, err := f.keyFrame(0, structure)
+	require.NoError(t, err)
+	require.Equal(t, 0, len(extDD.Descriptor.FrameDependencies.FrameDiffs))
+	require.Equal(t, dd.RenderResolution{Width: 640, Height: 360}, *extDD.Descriptor.Resolution)
+	ReleaseExtDependencyDescriptor(extDD)
+
+	extDD, err = f.deltaFrame(1)
+	require.NoError(t, err)
+	require.Equal(t, []int{1}, extDD.Descriptor.FrameDependencies.FrameDiffs)
+	require.Equal(t, []int{1}, extDD.Descriptor.FrameDependencies.ChainDiffs)
+	require.Nil(t, extDD.Descriptor.AttachedStructure)
+	ReleaseExtDependencyDescriptor(extDD)
+
+	// the parsed template is a copy, the structure's template is untouched
+	require.Equal(t, []int{1}, structure.Templates[1].FrameDiffs)
+
+	// marshal the packets first, only Parse runs under the allocation count
+	bufs := make([][]byte, 128)
+	for i := range bufs {
+		ddVal := &dd.DependencyDescriptor{
+			FirstPacketInFrame: true,
+			LastPacketInFrame:  true,
+			FrameNumber:        uint16(i + 2),
+			FrameDependencies:  structure.Templates[1],
+		}
+		bufs[i], err = (&dd.DependencyDescriptorExtension{Descriptor: ddVal, Structure: structure}).Marshal()
+		require.NoError(t, err)
+	}
+	pkt := &rtp.Packet{Header: rtp.Header{SequenceNumber: f.seq}}
+	require.NoError(t, pkt.SetExtension(ddTestExtID, bufs[0]))
+	i := 0
+	allocs := testing.AllocsPerRun(100, func() {
+		pkt.SequenceNumber++
+		require.NoError(t, pkt.SetExtension(ddTestExtID, bufs[i]))
+		i++
+		extDD, _, err := f.parser.Parse(pkt)
+		require.NoError(t, err)
+		ReleaseExtDependencyDescriptor(extDD)
+	})
+	require.Zero(t, allocs)
+}
@@ -26,13 +26,12 @@ import (
 var minFramesForCalculation = [...]int{8, 15, 40, 60}

 type frameInfo struct {
-	startSeq  uint16
-	endSeq    uint16
-	ts        uint32
-	fn        uint16
-	spatial   int32
-	temporal  int32
-	frameDiff []int
+	startSeq uint16
+	endSeq   uint16
+	ts       uint32
+	fn       uint16
+	spatial  int32
+	temporal int32
 }

 type FrameRateCalculator interface {
@@ -425,11 +424,10 @@ func (f *FrameRateCalculatorDD) RecvPacket(ep *ExtPacket) bool {
 	}

 	fi := &frameInfo{
-		ts:        ep.Packet.Timestamp,
-		fn:        fn,
-		temporal:  temporal,
-		spatial:   spatial,
-		frameDiff: ep.DependencyDescriptor.Descriptor.FrameDependencies.FrameDiffs,
+		ts:       ep.Packet.Timestamp,
+		fn:       fn,
+		temporal: temporal,
+		spatial:  spatial,
 	}
 	f.fnReceived[baseDiff] = fi

@@ -19,9 +19,9 @@ import (
 )

 type FrameEntity struct {
-	startSeq  *uint64
-	endSeq    *uint64
-	integrity bool
+	startSeq, endSeq       uint64
+	hasStartSeq, hasEndSeq bool
+	integrity              bool

 	pktHistory *PacketHistory
 }
@@ -32,15 +32,15 @@ func (fe *FrameEntity) AddPacket(extSeq uint64, ddVal *dd.DependencyDescriptor)
 		return
 	}

-	if fe.startSeq == nil && ddVal.FirstPacketInFrame {
-		fe.startSeq = &extSeq
+	if !fe.hasStartSeq && ddVal.FirstPacketInFrame {
+		fe.startSeq, fe.hasStartSeq = extSeq, true
 	}
-	if fe.endSeq == nil && ddVal.LastPacketInFrame {
-		fe.endSeq = &extSeq
+	if !fe.hasEndSeq && ddVal.LastPacketInFrame {
+		fe.endSeq, fe.hasEndSeq = extSeq, true
 	}

-	if fe.startSeq != nil && fe.endSeq != nil {
-		if fe.pktHistory.PacketsConsecutive(*fe.startSeq, *fe.endSeq) {
+	if fe.hasStartSeq && fe.hasEndSeq {
+		if fe.pktHistory.PacketsConsecutive(fe.startSeq, fe.endSeq) {
 			fe.integrity = true
 		}
 	}
@@ -48,7 +48,7 @@ func (fe *FrameEntity) AddPacket(extSeq uint64, ddVal *dd.DependencyDescriptor)

 func (fe *FrameEntity) Reset() {
 	fe.integrity = false
-	fe.startSeq, fe.endSeq = nil, nil
+	fe.hasStartSeq, fe.hasEndSeq = false, false
 }

 func (fe *FrameEntity) Integrity() bool {
@@ -113,9 +113,10 @@ const (
 // ------------------------------------------------------------------------------

 type DependencyDescriptor struct {
-	FirstPacketInFrame         bool
-	LastPacketInFrame          bool
-	FrameNumber                uint16
+	FirstPacketInFrame bool
+	LastPacketInFrame  bool
+	FrameNumber        uint16
+	// Unmarshal writes into FrameDependencies and Resolution when they are set, else it allocates them
 	FrameDependencies          *FrameDependencyTemplate
 	Resolution                 *RenderResolution
 	ActiveDecodeTargetsBitmask *uint32
@@ -201,6 +202,15 @@ func (t *FrameDependencyTemplate) Clone() *FrameDependencyTemplate {
 	return t2
 }

+// CopyInto copies t into dst and reuses the slices of dst
+func (t *FrameDependencyTemplate) CopyInto(dst *FrameDependencyTemplate) {
+	dst.SpatialId = t.SpatialId
+	dst.TemporalId = t.TemporalId
+	dst.DecodeTargetIndications = append(dst.DecodeTargetIndications[:0], t.DecodeTargetIndications...)
+	dst.FrameDiffs = append(dst.FrameDiffs[:0], t.FrameDiffs...)
+	dst.ChainDiffs = append(dst.ChainDiffs[:0], t.ChainDiffs...)
+}
+
 // ------------------------------------------------------------------------------

 type FrameDependencyStructure struct {
@@ -17,6 +17,8 @@ package dependencydescriptor
 import (
 	"encoding/hex"
 	"testing"
+
+	"github.com/stretchr/testify/require"
 )

 func TestDependencyDescriptorUnmarshal(t *testing.T) {
@@ -63,3 +65,80 @@ func TestDependencyDescriptorUnmarshal(t *testing.T) {
 		t.Log(ddVal.String())
 	}
 }
+
+// Unmarshal into a descriptor whose FrameDependencies and Resolution are set writes into that
+// storage, gives the same result as a fresh descriptor, and leaves the structure's templates intact
+func TestDependencyDescriptorUnmarshalIntoStorage(t *testing.T) {
+	hexes := []string{
+		"c1017280081485214eafffaaaa863cf0430c10c302afc0aaa0063c00430010c002a000a80006000040001d954926e082b04a0941b820ac1282503157f974000ca864330e222222eca8655304224230eca877530077004200ef008601df010d",
+		"86017340fc",
+		"46017340fc",
+		"c3017540fc",
+		"88017640fc",
+		"48017640fc",
+		"c2017840fc",
+	}
+
+	var structure *FrameDependencyStructure
+	var fd FrameDependencyTemplate
+	var res RenderResolution
+	for _, h := range hexes {
+		buf, err := hex.DecodeString(h)
+		require.NoError(t, err)
+
+		var fresh DependencyDescriptor
+		_, err = (&DependencyDescriptorExtension{Structure: structure, Descriptor: &fresh}).Unmarshal(buf)
+		require.NoError(t, err)
+
+		reused := DependencyDescriptor{FrameDependencies: &fd, Resolution: &res}
+		_, err = (&DependencyDescriptorExtension{Structure: structure, Descriptor: &reused}).Unmarshal(buf)
+		require.NoError(t, err)
+
+		require.Same(t, &fd, reused.FrameDependencies)
+		require.Equal(t, fresh.FrameDependencies, reused.FrameDependencies)
+		require.Equal(t, fresh.Resolution, reused.Resolution)
+		require.Equal(t, fresh.String(), reused.String())
+
+		if fresh.AttachedStructure != nil {
+			structure = fresh.AttachedStructure
+		}
+	}
+
+	// the templates are copied out, not shared
+	for _, template := range structure.Templates {
+		require.NotSame(t, template, &fd)
+	}
+}
+
+// a caller may hand Unmarshal a descriptor that still points at one of the structure's templates,
+// custom fields of the packet must not change that template
+func TestDependencyDescriptorUnmarshalDoesNotWriteIntoTemplate(t *testing.T) {
+	fixture := newDependencyDescriptorMarshalFixture(t)
+	structure := fixture.Structure
+
+	// a per-packet descriptor using a template with frame diffs, with one custom diff
+	var template *FrameDependencyTemplate
+	for _, tmpl := range structure.Templates {
+		if len(tmpl.FrameDiffs) > 0 {
+			template = tmpl
+			break
+		}
+	}
+	require.NotNil(t, template)
+	want := append([]int(nil), template.FrameDiffs...)
+
+	custom := template.Clone()
+	custom.FrameDiffs[0] += 2
+	buf, err := (&DependencyDescriptorExtension{
+		Descriptor: &DependencyDescriptor{FrameNumber: 7, FrameDependencies: custom},
+		Structure:  structure,
+	}).Marshal()
+	require.NoError(t, err)
+
+	parsed := DependencyDescriptor{FrameDependencies: template}
+	_, err = (&DependencyDescriptorExtension{Structure: structure, Descriptor: &parsed}).Unmarshal(buf)
+	require.NoError(t, err)
+	require.Equal(t, custom.FrameDiffs, parsed.FrameDependencies.FrameDiffs)
+	require.NotSame(t, template, parsed.FrameDependencies)
+	require.Equal(t, want, template.FrameDiffs)
+}
@@ -16,6 +16,7 @@ package dependencydescriptor

 import (
 	"errors"
+	"slices"
 )

 var (
@@ -356,8 +357,14 @@ func (r *DependencyDescriptorReader) readFrameDependencyDefinition() error {
 		return ErrDDReaderInvalidTemplateIndex
 	}

-	// Copy all the fields from the matching template
-	r.descriptor.FrameDependencies = r.structure.Templates[templateIndex].Clone()
+	// Copy all the fields from the matching template into caller storage,
+	// never into one of the structure's templates, custom fields would change the template
+	dst := r.descriptor.FrameDependencies
+	if dst == nil || slices.Contains(r.structure.Templates, dst) {
+		dst = &FrameDependencyTemplate{}
+		r.descriptor.FrameDependencies = dst
+	}
+	r.structure.Templates[templateIndex].CopyInto(dst)

 	if r.customDtisFlag {
 		err := r.readFrameDtis()
@@ -389,8 +396,10 @@ func (r *DependencyDescriptorReader) readFrameDependencyDefinition() error {
 			r.buffer.Invalidate()
 			return ErrDDReaderInvalidSpatialLayer
 		}
-		res := r.structure.Resolutions[r.descriptor.FrameDependencies.SpatialId]
-		r.descriptor.Resolution = &res
+		if r.descriptor.Resolution == nil {
+			r.descriptor.Resolution = &RenderResolution{}
+		}
+		*r.descriptor.Resolution = r.structure.Resolutions[r.descriptor.FrameDependencies.SpatialId]
 	}

 	return nil