diff --git a/pkg/sfu/flexfec/decoder.go b/pkg/sfu/flexfec/decoder.go index fdac3f11d..0c34911fb 100644 --- a/pkg/sfu/flexfec/decoder.go +++ b/pkg/sfu/flexfec/decoder.go @@ -161,6 +161,7 @@ func (d *Decoder) insertPacket(receivedPkt *rtp.Packet) { } } if toRemove > 0 { + clear(d.receivedFECPackets[:toRemove]) d.receivedFECPackets = d.receivedFECPackets[toRemove:] } } @@ -270,6 +271,13 @@ 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 { + return + } + d.receivedFECPackets = append(d.receivedFECPackets, fecPacketState{ packet: fecPkt, flexFec: fec, @@ -281,7 +289,7 @@ func (d *Decoder) insertFECPacket(fecPkt *rtp.Packet) { }) if len(d.receivedFECPackets) > maxFECPackets { - d.receivedFECPackets = d.receivedFECPackets[1:] + d.removeFECPacketAt(0) } } @@ -289,28 +297,27 @@ func (d *Decoder) attemptRecovery() []*rtp.Packet { var recoveredPackets []*rtp.Packet for { packetsRecovered := 0 - for i := range d.receivedFECPackets { + for i := 0; i < len(d.receivedFECPackets); { fecPkt := &d.receivedFECPackets[i] - packetsMissing := 0 - for _, pkt := range fecPkt.protectedPackets { - if pkt.packet == nil { - packetsMissing++ - if packetsMissing > 1 { - break - } - } + packetsMissing := countMissingPackets(fecPkt.protectedPackets) + if packetsMissing == 0 { + d.removeFECPacketAt(i) + continue } if packetsMissing != 1 { + i++ continue } recovered, err := d.recoverPacket(fecPkt) if err != nil { d.logger.Debugw("flexfec: failed to recover packet", "error", err) + i++ continue } + d.removeFECPacketAt(i) recoveredPackets = append(recoveredPackets, recovered) d.recoveredPackets = append(d.recoveredPackets, recovered) sort.Slice(d.recoveredPackets, func(i, j int) bool { @@ -330,6 +337,27 @@ func (d *Decoder) attemptRecovery() []*rtp.Packet { return recoveredPackets } +func countMissingPackets(protectedPackets []*protectedPacket) int { + missing := 0 + for _, pkt := range protectedPackets { + if pkt.packet == nil { + missing++ + if missing > 1 { + break + } + } + } + + return missing +} + +func (d *Decoder) removeFECPacketAt(index int) { + last := len(d.receivedFECPackets) - 1 + copy(d.receivedFECPackets[index:], d.receivedFECPackets[index+1:]) + d.receivedFECPackets[last] = fecPacketState{} + d.receivedFECPackets = d.receivedFECPackets[:last] +} + func (d *Decoder) recoverPacket(fec *fecPacketState) (*rtp.Packet, error) { // https://datatracker.ietf.org/doc/html/draft-ietf-payload-flexible-fec-scheme-03#section-6.3.2 @@ -391,7 +419,9 @@ func (d *Decoder) recoverPacket(fec *fecPacketState) (*rtp.Packet, error) { func (d *Decoder) discardOldRecoveredPackets() { if len(d.recoveredPackets) > recoveredPacketsLimit { - d.recoveredPackets = d.recoveredPackets[len(d.recoveredPackets)-recoveredPacketsLimit:] + toRemove := len(d.recoveredPackets) - recoveredPacketsLimit + clear(d.recoveredPackets[:toRemove]) + d.recoveredPackets = d.recoveredPackets[toRemove:] } } diff --git a/pkg/sfu/flexfec/decoder_test.go b/pkg/sfu/flexfec/decoder_test.go index 3431c099e..967fa4a04 100644 --- a/pkg/sfu/flexfec/decoder_test.go +++ b/pkg/sfu/flexfec/decoder_test.go @@ -96,6 +96,7 @@ func TestDecoderRecoversSingleLoss(t *testing.T) { require.Len(t, recovered, 1) requirePacketEqual(t, &media[2], recovered[0]) + assert.Empty(t, decoder.receivedFECPackets) stats := decoder.Stats() assert.Equal(t, uint64(len(fec)), stats.FECPacketsReceived) @@ -121,11 +122,13 @@ func TestDecoderRecoversWithLateMedia(t *testing.T) { } // two packets missing from the protected window, nothing recoverable yet require.Empty(t, recovered) + require.Len(t, decoder.receivedFECPackets, 1) // late arrival of media[1] leaves only media[2] missing recovered = decoder.DecodeFec(&media[1]) require.Len(t, recovered, 1) requirePacketEqual(t, &media[2], recovered[0]) + assert.Empty(t, decoder.receivedFECPackets) } func TestDecoderRecoversMultipleWindows(t *testing.T) { @@ -248,7 +251,7 @@ func TestDecoderDiscardsDuplicateFEC(t *testing.T) { fec := encodeFEC(t, media, 1) decoder := NewDecoder(testFECSSRC, testMediaSSRC, logger.GetLogger()) - for i := range media { + for _, i := range []int{0, 3, 4} { decoder.DecodeFec(&media[i]) } require.Empty(t, decoder.DecodeFec(&fec[0])) @@ -259,6 +262,19 @@ func TestDecoderDiscardsDuplicateFEC(t *testing.T) { assert.Equal(t, uint64(1), stats.FECPacketsDiscarded) } +func TestDecoderDoesNotRetainCompleteFECState(t *testing.T) { + media := makeMediaPackets(t, 450, 5) + fec := encodeFEC(t, media, 1) + + decoder := NewDecoder(testFECSSRC, testMediaSSRC, logger.GetLogger()) + for i := range media { + require.Empty(t, decoder.DecodeFec(&media[i])) + } + + require.Empty(t, decoder.DecodeFec(&fec[0])) + assert.Empty(t, decoder.receivedFECPackets) +} + func TestDecoderInputMemoryReuse(t *testing.T) { // the decoder must not retain references to caller-owned packet memory media := makeMediaPackets(t, 500, 5)