dandri/livekit
1
0
Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-09-12 02:46:09 +00:00
Code Issues Packages Projects Releases Wiki Activity

Avoid allocations for complete FlexFEC packets

Browse Source
This commit is contained in:
David Chen
2026-08-29 12:20:05 -07:00
parent 861b034d16
commit b48dfc9ac4
3 changed files with 170 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/sfu/flexfec/decoder.go
+52 -14
View File
@@ -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 {
pkg/sfu/flexfec/decoder_benchmark_test.go
+90
View File
@@ -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))
}
}
}
pkg/sfu/flexfec/decoder_test.go
+28
View File
@@ -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