dandri/livekit
1
0
Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-10-06 05:37:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

do not NACK on fec recovered packet

Browse Source
This commit is contained in:
David Chen
2026-09-22 10:41:13 -07:00
parent fd47ab4319
commit b08f67f61a
2 changed files with 116 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/sfu/buffer/buffer.go
+42 -18
View File
@@ -279,22 +279,17 @@ func (b *Buffer) Write(pkt []byte) (n int, err error) {
return
}
rtcpPackets := b.calc(pkt, &rtpPacket, now, false, false)
b.ingestLocked(pkt, &rtpPacket, now, false, false)
var fecDelta fecRecoveryDelta
var onFECRecovery func(received int, recovered int, discarded int, bytesReceived int)
if b.fecDecoder != nil {
// feed media into the FEC decoder, a media arrival can complete a
// previously unrecoverable FEC window
// previously unrecoverable FEC window. Recovered packets update NACK
// state before feedback is generated below, so a packet repaired by
// FEC is not needlessly NACKed.
fecDelta, onFECRecovery = b.feedFECLocked(&rtpPacket, now)
}
b.Unlock()
fecDelta.invoke(onFECRecovery)
if len(rtcpPackets) != 0 {
if cb := b.getOnRtcpFeedback(); cb != nil {
cb(rtcpPackets)
}
}
b.finishWriteLocked(now, fecDelta, onFECRecovery)
return
}
@@ -412,14 +407,13 @@ func (b *Buffer) writeRTX(rtxPkt *rtp.Packet, arrivalTime int64) {
return
}
b.calc(b.rtxPktBuf[:n], &repairedPkt, arrivalTime, false, true)
b.ingestLocked(b.rtxPktBuf[:n], &repairedPkt, arrivalTime, false, true)
var fecDelta fecRecoveryDelta
var onFECRecovery func(received int, recovered int, discarded int, bytesReceived int)
if b.fecDecoder != nil {
fecDelta, onFECRecovery = b.feedFECLocked(&repairedPkt, arrivalTime)
}
b.Unlock()
fecDelta.invoke(onFECRecovery)
b.finishWriteLocked(arrivalTime, fecDelta, onFECRecovery)
}
func (b *Buffer) SetPrimaryBufferForFEC(primaryBuffer *Buffer) {
@@ -565,8 +559,7 @@ func (b *Buffer) writeFEC(fecPkt *rtp.Packet, arrivalTime int64) {
}
fecDelta, onFECRecovery := b.feedFECLocked(fecPkt, arrivalTime)
b.Unlock()
fecDelta.invoke(onFECRecovery)
b.finishWriteLocked(arrivalTime, fecDelta, onFECRecovery)
}
// feedFECLocked runs a media or FEC packet through the FEC decoder and
@@ -592,8 +585,9 @@ func (b *Buffer) feedFECLocked(
// recovered packets flow through the regular pipeline: they are
// forwarded downstream and stop NACKs for the lost sequence numbers.
// They do not re-enter the decoder because chained recovery already
// completed within DecodeFEC.
b.calc(b.fecPktBuf[:n], rp, arrivalTime, false, true)
// completed within DecodeFEC. RTCP is generated once by the caller
// after all recovered packets have updated NACK state.
b.ingestLocked(b.fecPktBuf[:n], rp, arrivalTime, false, true)
}
if cb := b.onFECRecovery; cb != nil {
@@ -683,7 +677,11 @@ func (b *Buffer) sendPLI() {
}
}
func (b *Buffer) calc(rawPkt []byte, rtpPacket *rtp.Packet, arrivalTime int64, isBuffered bool, isRTX bool) []rtcp.Packet {
// ingestLocked runs a packet through the incoming packet pipeline, updating RTP
// stats, NACK state and the packet bucket. It intentionally does not generate
// RTCP so that FEC recovery can update NACK state before feedback is generated
// once. Must be called with the buffer lock held.
func (b *Buffer) ingestLocked(rawPkt []byte, rtpPacket *rtp.Packet, arrivalTime int64, isBuffered bool, isRTX bool) {
b.BufferBase.HandleIncomingPacketLocked(
rawPkt,
rtpPacket,
@@ -693,10 +691,36 @@ func (b *Buffer) calc(rawPkt []byte, rtpPacket *rtp.Packet, arrivalTime int64, i
nil,
0,
)
}
func (b *Buffer) calc(rawPkt []byte, rtpPacket *rtp.Packet, arrivalTime int64, isBuffered bool, isRTX bool) []rtcp.Packet {
b.ingestLocked(rawPkt, rtpPacket, arrivalTime, isBuffered, isRTX)
return b.getRTCPPackets(arrivalTime)
}
// finishWriteLocked generates RTCP feedback from the current NACK state, which
// already reflects any FEC recovery performed during this write, releases the
// buffer lock, then invokes the FEC recovery callback and delivers RTCP
// feedback. Generating RTCP after recovery ensures sequence numbers repaired by
// FEC are not needlessly NACKed. Must be called with the buffer lock held; it
// releases the lock before returning.
func (b *Buffer) finishWriteLocked(
arrivalTime int64,
fecDelta fecRecoveryDelta,
onFECRecovery func(received int, recovered int, discarded int, bytesReceived int),
) {
rtcpPackets := b.getRTCPPackets(arrivalTime)
b.Unlock()
fecDelta.invoke(onFECRecovery)
if len(rtcpPackets) != 0 {
if cb := b.getOnRtcpFeedback(); cb != nil {
cb(rtcpPackets)
}
}
}
func (b *Buffer) getRTCPPackets(arrivalTime int64) []rtcp.Packet {
var pkts []rtcp.Packet
if nackPkt := b.getNACKPacket(); nackPkt != nil {
pkg/sfu/buffer/buffer_fec_test.go
+74
View File
@@ -19,10 +19,12 @@ import (
"crypto/cipher"
"encoding/binary"
"math/rand"
"sync"
"testing"
"time"
pionflexfec "github.com/pion/interceptor/pkg/flexfec"
"github.com/pion/rtcp"
"github.com/pion/rtp"
"github.com/pion/transport/v4/packetio"
"github.com/pion/webrtc/v4"
@@ -598,3 +600,75 @@ func TestBufferFECNACKSuppression(t *testing.T) {
require.Empty(t, primary.nacker.Nacks(), "NACK for recovered packet not suppressed")
}
func TestBufferFECNACKNotSentForRecoveredPacket(t *testing.T) {
// A late media arrival can complete a retained FEC window. RTCP feedback
// must be generated after FEC recovery so the sequence number repaired by
// FEC is not NACKed, while an unrelated unrecoverable gap is still NACKed.
factory := NewFactoryOfBufferFactory(500, 200).CreateBufferFactory()
primary := factory.GetOrNew(packetio.RTPBufferPacket, fecTestMediaSSRC).(*Buffer)
fecBuff := factory.GetOrNew(packetio.RTPBufferPacket, fecTestFECSSRC).(*Buffer)
factory.SetFECPair(fecTestFECSSRC, fecTestMediaSSRC)
bindFECTestBuffer(t, primary)
var mu sync.Mutex
nackedSNs := make(map[uint16]bool)
primary.OnRtcpFeedback(func(fb []rtcp.Packet) {
mu.Lock()
defer mu.Unlock()
for _, pkt := range fb {
if nack, ok := pkt.(*rtcp.TransportLayerNack); ok {
for _, pair := range nack.Nacks {
pair.Range(func(sn uint16) bool {
nackedSNs[sn] = true
return true
})
}
}
}
})
// single FEC packet protecting the whole window; dropping two of the
// protected packets keeps the window unrecoverable until one arrives
media := fecTestMediaPackets(t, 100, 5) // 100..104
fecPackets := pionflexfec.NewFlexEncoder03(fecTestFECPT, fecTestFECSSRC).EncodeFec(media, 1)
require.Len(t, fecPackets, 1)
writePacket(t, primary, &media[0]) // 100
writePacket(t, primary, &media[1]) // 101
writePacket(t, primary, &media[4]) // 104 -> gap 102, 103
// an unrelated gap outside the FEC-protected window that FEC cannot recover
extra := fecTestMediaPackets(t, 105, 3) // 105..107
writePacket(t, primary, &extra[0]) // 105
writePacket(t, primary, &extra[2]) // 107 -> gap 106
const unrecoverableSN = uint16(106)
// FEC packet retained: two protected packets (102, 103) still missing
writePacket(t, fecBuff, &fecPackets[0])
require.EqualValues(t, 0, primary.FECDecoderStats().PacketsRecovered)
require.Len(t, primary.nacker.Nacks(), 3, "expected queued NACKs for 102, 103, 106")
// let the queued NACKs age past the minimum NACK interval so they would be
// sent on the next RTCP generation, and ignore any feedback emitted during
// setup
time.Sleep(30 * time.Millisecond)
mu.Lock()
nackedSNs = make(map[uint16]bool)
mu.Unlock()
// late arrival of 102 leaves only 103 missing, recovered in the same write
writePacket(t, primary, &media[2]) // 102
require.EqualValues(t, 1, primary.FECDecoderStats().PacketsRecovered)
// only the unrecoverable gap should remain queued; 102 arrived and 103 was
// recovered, both cleared from the NACK queue
require.Len(t, primary.nacker.Nacks(), 1)
mu.Lock()
defer mu.Unlock()
assert.True(t, nackedSNs[unrecoverableSN], "expected NACK for unrecoverable gap %d", unrecoverableSN)
assert.False(t, nackedSNs[media[2].SequenceNumber], "arrived packet %d must not be NACKed", media[2].SequenceNumber)
assert.False(t, nackedSNs[media[3].SequenceNumber], "FEC-recovered packet %d must not be NACKed", media[3].SequenceNumber)
}