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