From bc849eb47d4422e6b8387c21dd97cff1e9f2ad1e Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Tue, 6 Oct 2026 19:46:20 +0530 Subject: [PATCH] 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 * 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 --------- Co-authored-by: Claude Fable 5.1 --- pkg/sfu/buffer/dependencydescriptorparser.go | 38 ++++++--- .../buffer/dependencydescriptorparser_test.go | 48 +++++++++++ pkg/sfu/buffer/fps.go | 22 +++--- pkg/sfu/buffer/frameintegrity.go | 20 ++--- .../dependencydescriptorextension.go | 16 +++- .../dependencydescriptorextension_test.go | 79 +++++++++++++++++++ .../dependencydescriptorreader.go | 17 +++- 7 files changed, 201 insertions(+), 39 deletions(-) diff --git a/pkg/sfu/buffer/dependencydescriptorparser.go b/pkg/sfu/buffer/dependencydescriptorparser.go index 09a0d3b52..343eafc24 100644 --- a/pkg/sfu/buffer/dependencydescriptorparser.go +++ b/pkg/sfu/buffer/dependencydescriptorparser.go @@ -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 { diff --git a/pkg/sfu/buffer/dependencydescriptorparser_test.go b/pkg/sfu/buffer/dependencydescriptorparser_test.go index 652b0b637..7a09b013c 100644 --- a/pkg/sfu/buffer/dependencydescriptorparser_test.go +++ b/pkg/sfu/buffer/dependencydescriptorparser_test.go @@ -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) +} diff --git a/pkg/sfu/buffer/fps.go b/pkg/sfu/buffer/fps.go index 9ac69cd8f..d41dfe375 100644 --- a/pkg/sfu/buffer/fps.go +++ b/pkg/sfu/buffer/fps.go @@ -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 diff --git a/pkg/sfu/buffer/frameintegrity.go b/pkg/sfu/buffer/frameintegrity.go index e2cbe8a43..9e7814a8b 100644 --- a/pkg/sfu/buffer/frameintegrity.go +++ b/pkg/sfu/buffer/frameintegrity.go @@ -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 { diff --git a/pkg/sfu/rtpextension/dependencydescriptor/dependencydescriptorextension.go b/pkg/sfu/rtpextension/dependencydescriptor/dependencydescriptorextension.go index 28812d09a..3a1bb4760 100644 --- a/pkg/sfu/rtpextension/dependencydescriptor/dependencydescriptorextension.go +++ b/pkg/sfu/rtpextension/dependencydescriptor/dependencydescriptorextension.go @@ -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 { diff --git a/pkg/sfu/rtpextension/dependencydescriptor/dependencydescriptorextension_test.go b/pkg/sfu/rtpextension/dependencydescriptor/dependencydescriptorextension_test.go index 95c3aba35..58efec953 100644 --- a/pkg/sfu/rtpextension/dependencydescriptor/dependencydescriptorextension_test.go +++ b/pkg/sfu/rtpextension/dependencydescriptor/dependencydescriptorextension_test.go @@ -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) +} diff --git a/pkg/sfu/rtpextension/dependencydescriptor/dependencydescriptorreader.go b/pkg/sfu/rtpextension/dependencydescriptor/dependencydescriptorreader.go index b3baf6f36..0b7d8ac0a 100644 --- a/pkg/sfu/rtpextension/dependencydescriptor/dependencydescriptorreader.go +++ b/pkg/sfu/rtpextension/dependencydescriptor/dependencydescriptorreader.go @@ -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