From b48dfc9ac449ba3e06eaab2b02df3ba690b211a3 Mon Sep 17 00:00:00 2001 From: David Chen Date: Sat, 29 Aug 2026 12:20:05 -0700 Subject: [PATCH] Avoid allocations for complete FlexFEC packets --- pkg/sfu/flexfec/decoder.go | 66 +++++++++++++---- pkg/sfu/flexfec/decoder_benchmark_test.go | 90 +++++++++++++++++++++++ pkg/sfu/flexfec/decoder_test.go | 28 +++++++ 3 files changed, 170 insertions(+), 14 deletions(-) create mode 100644 pkg/sfu/flexfec/decoder_benchmark_test.go diff --git a/pkg/sfu/flexfec/decoder.go b/pkg/sfu/flexfec/decoder.go index 98ad33e17..80bca306c 100644 --- a/pkg/sfu/flexfec/decoder.go +++ b/pkg/sfu/flexfec/decoder.go @@ -56,6 +56,8 @@ const ( maxFECPackets = 100 // seen/recovered media packets retained for XOR recovery recoveredPacketsLimit = 192 + // maximum number of sequence numbers represented by the three packet masks + maxProtectedPackets = fecMask0Bits + fecMask1Bits + fecMask2Bits ) // FlexFEC-03 header bit fields. @@ -123,8 +125,13 @@ func (d *Decoder) DecodeFec(receivedPacket *rtp.Packet) []*rtp.Packet { d.stats.FECBytesReceived += uint64(len(receivedPacket.Payload)) } - // the caller reuses packet memory, keep an owned copy - pkt := receivedPacket.Clone() + // Media packets remain in the recovery window and need an owned copy. FEC + // packets are cloned only if insertFECPacket determines that their state + // must outlive this call. + pkt := receivedPacket + if receivedPacket.SSRC == d.protectedSSRC { + pkt = receivedPacket.Clone() + } if len(d.recoveredPackets) >= maxMediaPackets { backRecoveredPacket := d.recoveredPackets[len(d.recoveredPackets)-1] @@ -227,12 +234,13 @@ func (d *Decoder) insertFECPacket(fecPkt *rtp.Packet) { return } - protectedSeqs := decodeMask(uint64(fec.mask0), fecMask0Bits, fec.seqNumBase) + var protectedSeqBuf [maxProtectedPackets]uint16 + protectedSeqs := appendMaskSequences(protectedSeqBuf[:0], uint64(fec.mask0), fecMask0Bits, fec.seqNumBase) if fec.mask1 != 0 { - protectedSeqs = append(protectedSeqs, decodeMask(uint64(fec.mask1), fecMask1Bits, fec.seqNumBase+fecMask0Bits)...) + protectedSeqs = appendMaskSequences(protectedSeqs, uint64(fec.mask1), fecMask1Bits, fec.seqNumBase+fecMask0Bits) } if fec.mask2 != 0 { - protectedSeqs = append(protectedSeqs, decodeMask(fec.mask2, fecMask2Bits, fec.seqNumBase+fecMask0Bits+fecMask1Bits)...) + protectedSeqs = appendMaskSequences(protectedSeqs, fec.mask2, fecMask2Bits, fec.seqNumBase+fecMask0Bits+fecMask1Bits) } if len(protectedSeqs) == 0 { @@ -241,6 +249,10 @@ func (d *Decoder) insertFECPacket(fecPkt *rtp.Packet) { return } + if countMissingSequences(protectedSeqs, d.recoveredPackets) == 0 { + return + } + protectedPackets := make([]protectedPacket, 0, len(protectedSeqs)) protectedSeqIt := 0 recoveredPacketIt := 0 @@ -273,15 +285,20 @@ func (d *Decoder) insertFECPacket(fecPkt *rtp.Packet) { protectedSeqIt++ } - // No recovery is needed when all protected media packets are already - // available. Do not retain the FEC packet and its cloned payload. - if countMissingPackets(protectedPackets) == 0 { + // The caller may reuse packet memory after DecodeFec returns. Take + // ownership only now that this FEC state needs to be retained. + ownedFECPkt := fecPkt.Clone() + ownedFEC, err := parseFlexFEC03Header(ownedFECPkt.Payload) + if err != nil { + // Parsing the same bytes succeeded above, so this should be unreachable. + d.stats.FECPacketsDiscarded++ + d.logger.Debugw("flexfec: failed to parse cloned header", "error", err) return } d.receivedFECPackets = append(d.receivedFECPackets, fecPacketState{ - packet: fecPkt, - flexFec: fec, + packet: ownedFECPkt, + flexFec: ownedFEC, protectedPackets: protectedPackets, }) @@ -352,6 +369,28 @@ func countMissingPackets(protectedPackets []protectedPacket) int { return missing } +func countMissingSequences(protectedSeqs []uint16, recoveredPackets []*rtp.Packet) int { + missing := 0 + protectedSeqIt := 0 + recoveredPacketIt := 0 + + for protectedSeqIt < len(protectedSeqs) && recoveredPacketIt < len(recoveredPackets) { + switch { + case isNewerSeq(protectedSeqs[protectedSeqIt], recoveredPackets[recoveredPacketIt].SequenceNumber): + missing++ + protectedSeqIt++ + case isNewerSeq(recoveredPackets[recoveredPacketIt].SequenceNumber, protectedSeqs[protectedSeqIt]): + recoveredPacketIt++ + default: + protectedSeqIt++ + recoveredPacketIt++ + } + } + + missing += len(protectedSeqs) - protectedSeqIt + return missing +} + func (d *Decoder) removeFECPacketAt(index int) { last := len(d.receivedFECPackets) - 1 copy(d.receivedFECPackets[index:], d.receivedFECPackets[index+1:]) @@ -426,15 +465,14 @@ func (d *Decoder) discardOldRecoveredPackets() { } } -func decodeMask(mask uint64, bitCount uint16, seqNumBase uint16) []uint16 { - res := make([]uint16, 0) +func appendMaskSequences(dst []uint16, mask uint64, bitCount uint16, seqNumBase uint16) []uint16 { for i := uint16(0); i < bitCount; i++ { if (mask>>(bitCount-1-i))&1 == 1 { - res = append(res, seqNumBase+i) + dst = append(dst, seqNumBase+i) } } - return res + return dst } type fecPacketState struct { diff --git a/pkg/sfu/flexfec/decoder_benchmark_test.go b/pkg/sfu/flexfec/decoder_benchmark_test.go new file mode 100644 index 000000000..6ebf179c8 --- /dev/null +++ b/pkg/sfu/flexfec/decoder_benchmark_test.go @@ -0,0 +1,90 @@ +// 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 flexfec + +import ( + "testing" + + pionflexfec "github.com/pion/interceptor/pkg/flexfec" + "github.com/pion/rtp" + + "github.com/livekit/protocol/logger" +) + +func benchmarkMediaPackets(count, payloadSize int) []rtp.Packet { + packets := make([]rtp.Packet, count) + for i := range packets { + packets[i] = rtp.Packet{ + Header: rtp.Header{ + Version: 2, + PayloadType: testMediaPT, + SequenceNumber: 100 + uint16(i), + Timestamp: 3000 * uint32(i), + SSRC: testMediaSSRC, + }, + Payload: make([]byte, payloadSize), + } + } + return packets +} + +func BenchmarkDecoderMediaSteadyState1200(b *testing.B) { + decoder := NewDecoder(testFECSSRC, testMediaSSRC, logger.GetLogger()) + packet := benchmarkMediaPackets(1, 1200)[0] + + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + packet.SequenceNumber = uint16(i) + packet.Timestamp = uint32(i) * 3000 + decoder.DecodeFec(&packet) + } +} + +func BenchmarkDecoderCompleteWindow10x1200(b *testing.B) { + media := benchmarkMediaPackets(10, 1200) + fecPackets := pionflexfec.NewFlexEncoder03(testFECPT, testFECSSRC).EncodeFec(media, 1) + + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + decoder := NewDecoder(testFECSSRC, testMediaSSRC, logger.GetLogger()) + for j := range media { + decoder.DecodeFec(&media[j]) + } + if recovered := decoder.DecodeFec(&fecPackets[0]); len(recovered) != 0 { + b.Fatalf("expected no recovered packets, got %d", len(recovered)) + } + } +} + +func BenchmarkDecoderRecoveryWindow10x1200(b *testing.B) { + media := benchmarkMediaPackets(10, 1200) + fecPackets := pionflexfec.NewFlexEncoder03(testFECPT, testFECSSRC).EncodeFec(media, 1) + + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + decoder := NewDecoder(testFECSSRC, testMediaSSRC, logger.GetLogger()) + for j := range media { + if j != 4 { + decoder.DecodeFec(&media[j]) + } + } + if recovered := decoder.DecodeFec(&fecPackets[0]); len(recovered) != 1 { + b.Fatalf("expected one recovered packet, got %d", len(recovered)) + } + } +} diff --git a/pkg/sfu/flexfec/decoder_test.go b/pkg/sfu/flexfec/decoder_test.go index 967fa4a04..622b5870d 100644 --- a/pkg/sfu/flexfec/decoder_test.go +++ b/pkg/sfu/flexfec/decoder_test.go @@ -310,6 +310,34 @@ func TestDecoderInputMemoryReuse(t *testing.T) { requirePacketEqual(t, &media[2], recovered[0]) } +func TestDecoderRetainedFECMemoryReuse(t *testing.T) { + media := makeMediaPackets(t, 550, 5) + fec := encodeFEC(t, media, 1) + + decoder := NewDecoder(testFECSSRC, testMediaSSRC, logger.GetLogger()) + scratch := &rtp.Packet{} + feed := func(src *rtp.Packet) []*rtp.Packet { + buf, err := src.Marshal() + require.NoError(t, err) + require.NoError(t, scratch.Unmarshal(buf)) + out := decoder.DecodeFec(scratch) + for i := range scratch.Payload { + scratch.Payload[i] = 0xde + } + return out + } + + for _, i := range []int{0, 3, 4} { + require.Empty(t, feed(&media[i])) + } + require.Empty(t, feed(&fec[0])) + require.Len(t, decoder.receivedFECPackets, 1) + + recovered := feed(&media[1]) + require.Len(t, recovered, 1) + requirePacketEqual(t, &media[2], recovered[0]) +} + func TestDecoderTwoFECPacketsTwoLosses(t *testing.T) { // with 2 FEC packets over 10 media packets, the coverage interleaves, so // two losses landing in different coverage groups are both recoverable