From b30426000a3dfb6353789c8e63b325fabcf9fa43 Mon Sep 17 00:00:00 2001 From: David Chen Date: Wed, 21 Oct 2026 14:41:54 -0700 Subject: [PATCH] migrate fec into buffer_base --- pkg/rtc/mediatrack.go | 5 - pkg/sfu/buffer/buffer.go | 224 +++--------------------------- pkg/sfu/buffer/buffer_base.go | 203 +++++++++++++++++++++++++-- pkg/sfu/buffer/buffer_fec_test.go | 153 ++++++++++++++------ 4 files changed, 314 insertions(+), 271 deletions(-) diff --git a/pkg/rtc/mediatrack.go b/pkg/rtc/mediatrack.go index 895216f41..9793e2103 100644 --- a/pkg/rtc/mediatrack.go +++ b/pkg/rtc/mediatrack.go @@ -37,7 +37,6 @@ import ( "github.com/livekit/livekit-server/pkg/sfu/connectionquality" "github.com/livekit/livekit-server/pkg/sfu/interceptor" "github.com/livekit/livekit-server/pkg/telemetry" - "github.com/livekit/livekit-server/pkg/telemetry/prometheus" util "github.com/livekit/mediatransportutil" ) @@ -582,10 +581,6 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track sfu.TrackRe buff.OnNotifyRTX(t.MediaTrackReceiver.setLayerRtxInfo) - buff.OnFECRecovery(func(received int, recovered int, discarded int, bytesReceived int) { - prometheus.RecordFECUpstream(received, recovered, discarded, uint64(bytesReceived)) - }) - // if subscriber request fps before fps calculated, update them after fps updated. buff.OnFpsChanged(func() { t.MediaTrackSubscriptions.UpdateVideoLayers() diff --git a/pkg/sfu/buffer/buffer.go b/pkg/sfu/buffer/buffer.go index cca32e2e8..cecdf81d7 100644 --- a/pkg/sfu/buffer/buffer.go +++ b/pkg/sfu/buffer/buffer.go @@ -23,7 +23,6 @@ import ( "github.com/pion/rtp" "github.com/pion/webrtc/v4" - "github.com/livekit/livekit-server/pkg/sfu/flexfec" sutils "github.com/livekit/livekit-server/pkg/utils" "github.com/livekit/mediatransportutil/pkg/bucket" "github.com/livekit/mediatransportutil/pkg/twcc" @@ -42,10 +41,7 @@ const ( maxPendingFECRepairPackets = 100 ) -var ( - errInvalidCodec = errors.New("invalid codec") - errFECMediaPacketNotFound = errors.New("fec media packet not found") -) +var errInvalidCodec = errors.New("invalid codec") var _ BufferProvider = (*Buffer)(nil) @@ -54,21 +50,6 @@ type pendingPacket struct { packet []byte } -type fecRecoveryDelta struct { - received int - recovered int - discarded int - bytesReceived int -} - -func (d fecRecoveryDelta) invoke(callback func(received int, recovered int, discarded int, bytesReceived int)) { - if callback != nil && (d.recovered > 0 || d.received > 0 || d.discarded > 0) { - callback(d.received, d.recovered, d.discarded, d.bytesReceived) - } -} - -type cbFECRecovery func(received int, recovered int, discarded int, bytesReceived int) - // Buffer contains all packets type Buffer struct { *BufferBase @@ -96,10 +77,6 @@ type Buffer struct { primaryBufferForFEC *Buffer isFECRepair bool - fecSSRC uint32 - fecDecoder *flexfec.Decoder - fecPktBuf []byte - onFECRecovery cbFECRecovery streamInfoProbe *StreamInfoProbe warnedPendingOverflow bool @@ -123,13 +100,12 @@ type StreamInfoProbe struct { func NewBuffer(ssrc uint32, maxVideoPkts, maxAudioPkts int) *Buffer { b := &Buffer{} b.BufferBase = NewBufferBase(BufferBaseParams{ - SSRC: ssrc, - MaxVideoPkts: maxVideoPkts, - MaxAudioPkts: maxAudioPkts, - LoggerComponents: []string{sutils.ComponentPub, sutils.ComponentSFU}, - SendPLI: b.sendPLI, - OnStreamRestartLocked: b.recreateFECDecoderLocked, - IsReportingEnabled: true, + SSRC: ssrc, + MaxVideoPkts: maxVideoPkts, + MaxAudioPkts: maxAudioPkts, + LoggerComponents: []string{sutils.ComponentPub, sutils.ComponentSFU}, + SendPLI: b.sendPLI, + IsReportingEnabled: true, }) return b } @@ -175,10 +151,9 @@ func (b *Buffer) Bind(params webrtc.RTPParameters, codec webrtc.RTPCodecCapabili rtcpPackets := b.getRTCPPackets(now) b.isBound = true - b.maybeCreateFECDecoderLocked() b.Unlock() - b.packetPostProcess(rtcpPackets, fecRecoveryDelta{}, nil) + b.packetPostProcess(rtcpPackets) return nil } @@ -282,20 +257,10 @@ func (b *Buffer) Write(pkt []byte) (n int, err error) { b.ingestLocked(pkt, rtpPacket, now, false, false) - var fecDelta fecRecoveryDelta - var onFECRecovery cbFECRecovery - if b.fecDecoder != nil { - // feed media into the FEC decoder, a media arrival can complete a - // 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) - } - rtcpPackets := b.getRTCPPackets(now) b.Unlock() - b.packetPostProcess(rtcpPackets, fecDelta, onFECRecovery) + b.packetPostProcess(rtcpPackets) return } @@ -415,16 +380,10 @@ func (b *Buffer) writeRTX(rtxPkt *rtp.Packet, arrivalTime int64) { b.ingestLocked(b.rtxPktBuf[:n], &repairedPkt, arrivalTime, false, true) - var fecDelta fecRecoveryDelta - var onFECRecovery cbFECRecovery - if b.fecDecoder != nil { - fecDelta, onFECRecovery = b.feedFECLocked(&repairedPkt, arrivalTime) - } - rtcpPackets := b.getRTCPPackets(arrivalTime) b.Unlock() - b.packetPostProcess(rtcpPackets, fecDelta, onFECRecovery) + b.packetPostProcess(rtcpPackets) } func (b *Buffer) SetPrimaryBufferForFEC(primaryBuffer *Buffer) { @@ -437,7 +396,7 @@ func (b *Buffer) SetPrimaryBufferForFEC(primaryBuffer *Buffer) { // Let the primary know the repair stream SSRC so its decoder is ready // before the first FEC packet arrives. - primaryBuffer.setFECSSRC(ssrc) + primaryBuffer.SetFECSSRC(ssrc) for _, pp := range pkts { var rtpPacket rtp.Packet @@ -471,84 +430,8 @@ func (b *Buffer) setFECRepairBufferLimit() { b.pPackets = retained } -func (b *Buffer) setFECSSRC(ssrc uint32) { - b.Lock() - b.fecSSRC = ssrc - b.maybeCreateFECDecoderLocked() - b.Unlock() -} - -// maybeCreateFECDecoderLocked creates the FEC decoder as soon as the repair -// stream SSRC is known and the buffer is bound with a negotiated flexfec -// payload type. Protected media is read from the primary RTP packet bucket. -func (b *Buffer) maybeCreateFECDecoderLocked() { - if b.fecDecoder != nil || b.fecSSRC == 0 || !b.isBound || b.fecPayloadType == 0 { - return - } - - b.fecDecoder = flexfec.NewDecoder(b.fecSSRC, b.BufferBase.SSRC(), b.getFECMediaPacketLocked, b.logger) - b.logger.Debugw("flexfec decoder created", "fecSSRC", b.fecSSRC, "mediaSSRC", b.BufferBase.SSRC()) -} - -// recreateFECDecoderLocked drops FlexFEC state from the previous sequence -// space. Stream restart resyncs the packet bucket and sequence map, so repair -// packets retained by the old decoder would recover against the wrong media -// or be treated as duplicates of the new repair stream. -func (b *Buffer) recreateFECDecoderLocked() { - if b.fecDecoder == nil { - return - } - - b.fecDecoder = nil - b.maybeCreateFECDecoderLocked() -} - -func (b *Buffer) getFECMediaPacketLocked(sequenceNumber uint16, dst []byte) (int, error) { - if b.bucket == nil || b.rtpStats == nil { - return 0, errFECMediaPacketNotFound - } - - // FlexFEC masks use the publisher's sequence-number space. BufferBase - // removes padding-only packets from the downstream space, so resolve the - // original extended sequence number and apply the same adjustment used - // when the packet was inserted into the bucket. - highestSequenceNumber := b.rtpStats.ExtendedHighestSequenceNumber() - extendedSequenceNumber := int64(highestSequenceNumber) + int64(int16(sequenceNumber-uint16(highestSequenceNumber))) - if extendedSequenceNumber < 0 { - return 0, errFECMediaPacketNotFound - } - - extendedSN := uint64(extendedSequenceNumber) - sequenceNumberAdjustment, err := b.snRangeMap.GetValue(extendedSN) - if err != nil || sequenceNumberAdjustment > extendedSN { - return 0, errFECMediaPacketNotFound - } - - return b.bucket.GetPacket(dst, extendedSN-sequenceNumberAdjustment) -} - -// OnFECRecovery is called with counter deltas whenever FEC packets are -// processed since the previous callback: FEC packets received, media packets -// recovered, FEC packets discarded and FEC bytes received. -func (b *Buffer) OnFECRecovery(fn cbFECRecovery) { - b.Lock() - b.onFECRecovery = fn - b.Unlock() -} - -// FECDecoderStats returns cumulative FlexFEC decode counters of the buffer. -func (b *Buffer) FECDecoderStats() flexfec.DecoderStats { - b.RLock() - defer b.RUnlock() - - if b.fecDecoder == nil { - return flexfec.DecoderStats{} - } - return b.fecDecoder.Stats() -} - -// writeFEC handles a packet of the coupled FlexFEC repair stream, recovered -// media packets are injected into the regular packet pipeline. +// writeFEC accounts for transport feedback on the coupled repair stream. +// BufferBase handles decoding and injecting recovered media packets. func (b *Buffer) writeFEC(fecPkt *rtp.Packet, arrivalTime int64) { b.Lock() if !b.isBound { @@ -565,83 +448,14 @@ func (b *Buffer) writeFEC(fecPkt *rtp.Packet, arrivalTime int64) { } } - if b.fecPayloadType == 0 || fecPkt.PayloadType != b.fecPayloadType { - b.logger.Debugw("unexpected fec payload type", "expected", b.fecPayloadType, "actual", fecPkt.PayloadType) + if !b.BufferBase.HandleFECPacketLocked(fecPkt, arrivalTime) { b.Unlock() return } - - if b.fecDecoder == nil { - // normally created when the pair is declared, fall back to the - // observed repair stream SSRC - b.fecSSRC = fecPkt.SSRC - b.maybeCreateFECDecoderLocked() - if b.fecDecoder == nil { - b.logger.Infow("flexfec decoder unavailable after creation attempt", "fecSSRC", b.fecSSRC, "mediaSSRC", b.BufferBase.SSRC(), "payloadType", b.fecPayloadType) - b.Unlock() - return - } - } - - fecDelta, onFECRecovery := b.feedFECLocked(fecPkt, arrivalTime) rtcpPackets := b.getRTCPPackets(arrivalTime) b.Unlock() - b.packetPostProcess(rtcpPackets, fecDelta, onFECRecovery) -} - -// feedFECLocked runs a media or FEC packet through the FEC decoder and -// injects recovered packets into the packet pipeline. Must be called with the -// buffer lock held and a non-nil decoder. -func (b *Buffer) feedFECLocked( - pkt *rtp.Packet, - arrivalTime int64, -) (fecRecoveryDelta, cbFECRecovery) { - statsBefore := b.fecDecoder.Stats() - recovered := b.fecDecoder.DecodeFEC(pkt) - statsAfter := b.fecDecoder.Stats() - - fecPackets := statsAfter.FECPacketsReceived - statsBefore.FECPacketsReceived - fecBytes := statsAfter.FECBytesReceived - statsBefore.FECBytesReceived - fecPacketsDiscarded := statsAfter.FECPacketsDiscarded - statsBefore.FECPacketsDiscarded - fecPacketsRecovered := statsAfter.PacketsRecovered - statsBefore.PacketsRecovered - if b.rtpStats != nil { - b.rtpStats.UpdateFEC( - fecPackets, - fecBytes, - fecPacketsDiscarded, - fecPacketsRecovered, - ) - } - - if len(recovered) > 0 && b.fecPktBuf == nil { - b.fecPktBuf = make([]byte, bucket.RTPMaxPktSize) - } - for _, rp := range recovered { - n, err := rp.MarshalTo(b.fecPktBuf) - if err != nil { - b.logger.Warnw("could not marshal fec recovered packet", err, "ssrc", b.BufferBase.SSRC(), "sn", rp.SequenceNumber) - continue - } - - // 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. 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 { - return fecRecoveryDelta{ - received: int(fecPackets), - recovered: int(fecPacketsRecovered), - discarded: int(fecPacketsDiscarded), - bytesReceived: int(fecBytes), - }, cb - } - - return fecRecoveryDelta{}, nil + b.packetPostProcess(rtcpPackets) } func (b *Buffer) Read(buff []byte) (n int, err error) { @@ -738,13 +552,7 @@ func (b *Buffer) ingestLocked(rawPkt []byte, rtpPacket *rtp.Packet, arrivalTime // `rtcpPackets` should be RTCP feedback from the current NACK state, which // already reflects any FEC recovery performed during this write. // This is to ensure that packets recovered via FEC is not NACKed. -func (b *Buffer) packetPostProcess( - rtcpPackets []rtcp.Packet, - fecDelta fecRecoveryDelta, - onFECRecovery cbFECRecovery, -) { - fecDelta.invoke(onFECRecovery) - +func (b *Buffer) packetPostProcess(rtcpPackets []rtcp.Packet) { if len(rtcpPackets) != 0 { if cb := b.getOnRtcpFeedback(); cb != nil { cb(rtcpPackets) diff --git a/pkg/sfu/buffer/buffer_base.go b/pkg/sfu/buffer/buffer_base.go index ca831e212..568e2d87e 100644 --- a/pkg/sfu/buffer/buffer_base.go +++ b/pkg/sfu/buffer/buffer_base.go @@ -32,10 +32,12 @@ import ( "go.uber.org/atomic" "github.com/livekit/livekit-server/pkg/sfu/audio" + "github.com/livekit/livekit-server/pkg/sfu/flexfec" act "github.com/livekit/livekit-server/pkg/sfu/rtpextension/abscapturetime" dd "github.com/livekit/livekit-server/pkg/sfu/rtpextension/dependencydescriptor" "github.com/livekit/livekit-server/pkg/sfu/rtpstats" "github.com/livekit/livekit-server/pkg/sfu/utils" + "github.com/livekit/livekit-server/pkg/telemetry/prometheus" "github.com/livekit/mediatransportutil/pkg/bucket" "github.com/livekit/mediatransportutil/pkg/codec" "github.com/livekit/mediatransportutil/pkg/nack" @@ -46,6 +48,8 @@ import ( ) var ( + errFECMediaPacketNotFound = errors.New("fec media packet not found") + ExtPacketFactory = &sync.Pool{ New: func() any { return &ExtPacket{} @@ -152,15 +156,14 @@ const ( ) type BufferBaseParams struct { - SSRC uint32 - MaxVideoPkts int - MaxAudioPkts int - LoggerComponents []string - SendPLI func() - OnStreamRestartLocked func() - IsReportingEnabled bool - IsOOBSequenceNumber bool - IsDDRestartEnabled bool + SSRC uint32 + MaxVideoPkts int + MaxAudioPkts int + LoggerComponents []string + SendPLI func() + IsReportingEnabled bool + IsOOBSequenceNumber bool + IsDDRestartEnabled bool } type BufferBase struct { @@ -188,6 +191,9 @@ type BufferBase struct { payloadType uint8 rtxPayloadType uint8 fecPayloadType uint8 + fecSSRC uint32 + fecDecoder *flexfec.Decoder + fecPktBuf []byte snRangeMap *utils.RangeMap[uint64, uint64] @@ -427,6 +433,7 @@ func (b *BufferBase) BindLocked(rtpParameters webrtc.RTPParameters, codec webrtc } b.StartKeyFrameSeeder() + b.maybeCreateFECDecoderLocked() return nil } @@ -600,9 +607,7 @@ func (b *BufferBase) restartStreamLocked(reason string, isDetected bool) { b.StartKeyFrameSeeder() - if f := b.params.OnStreamRestartLocked; f != nil { - f() - } + b.recreateFECDecoderLocked() if isDetected { b.isRestartPending = true @@ -773,6 +778,32 @@ func (b *BufferBase) HandleIncomingPacketLocked( } } + sequenceNumber := rtpPacket.SequenceNumber + extSequenceNumber, err := b.handleIncomingPacketLocked( + rawPkt, rtpPacket, arrivalTime, isBuffered, isRTX, skippedSeqs, oobSequenceNumber, + ) + if b.fecDecoder != nil { + // Media can complete a retained FEC window. Recover before callers + // generate feedback, using the publisher's sequence number because + // ingestion adjusts it when padding has been removed. + mediaPacket := *rtpPacket + mediaPacket.SequenceNumber = sequenceNumber + b.feedFECLocked(&mediaPacket, arrivalTime) + } + return extSequenceNumber, err +} + +// handleIncomingPacketLocked ingests media without re-entering the decoder. +// Recovered packets use this path because DecodeFEC already chains recovery. +func (b *BufferBase) handleIncomingPacketLocked( + rawPkt []byte, + rtpPacket *rtp.Packet, + arrivalTime int64, + isBuffered bool, + isRTX bool, + skippedSeqs []uint16, + oobSequenceNumber uint16, +) (uint64, error) { b.processAudioSsrcLevelHeaderExtension(rtpPacket, arrivalTime) if len(skippedSeqs) > 0 { @@ -959,6 +990,154 @@ func (b *BufferBase) HandleIncomingPacketLocked( return ep.ExtSequenceNumber, nil } +// SetFECSSRC identifies the repair stream protecting this buffer. It can be +// called before or after Bind; decoding starts once FlexFEC is negotiated. +func (b *BufferBase) SetFECSSRC(ssrc uint32) { + b.Lock() + b.fecSSRC = ssrc + b.maybeCreateFECDecoderLocked() + b.Unlock() +} + +// maybeCreateFECDecoderLocked creates the FEC decoder as soon as the repair +// stream SSRC is known and the buffer is bound with a negotiated flexfec +// payload type. Protected media is read from the primary RTP packet bucket. +func (b *BufferBase) maybeCreateFECDecoderLocked() { + if b.fecDecoder != nil || b.fecSSRC == 0 || b.bucket == nil || b.fecPayloadType == 0 { + return + } + + b.fecDecoder = flexfec.NewDecoder(b.fecSSRC, b.SSRC(), b.getFECMediaPacketLocked, b.logger) + b.logger.Debugw("flexfec decoder created", "fecSSRC", b.fecSSRC, "mediaSSRC", b.SSRC()) +} + +// recreateFECDecoderLocked drops FlexFEC state from the previous sequence +// space. Stream restart resyncs the packet bucket and sequence map, so repair +// packets retained by the old decoder would recover against the wrong media +// or be treated as duplicates of the new repair stream. +func (b *BufferBase) recreateFECDecoderLocked() { + if b.fecDecoder == nil { + return + } + + b.fecDecoder = nil + b.maybeCreateFECDecoderLocked() +} + +func (b *BufferBase) getFECMediaPacketLocked(sequenceNumber uint16, dst []byte) (int, error) { + if b.bucket == nil || b.rtpStats == nil { + return 0, errFECMediaPacketNotFound + } + + // FlexFEC masks use the publisher's sequence-number space. BufferBase + // removes padding-only packets from the downstream space, so resolve the + // original extended sequence number and apply the same adjustment used + // when the packet was inserted into the bucket. + highestSequenceNumber := b.rtpStats.ExtendedHighestSequenceNumber() + extendedSequenceNumber := int64(highestSequenceNumber) + int64(int16(sequenceNumber-uint16(highestSequenceNumber))) + if extendedSequenceNumber < 0 { + return 0, errFECMediaPacketNotFound + } + + extendedSN := uint64(extendedSequenceNumber) + sequenceNumberAdjustment, err := b.snRangeMap.GetValue(extendedSN) + if err != nil || sequenceNumberAdjustment > extendedSN { + return 0, errFECMediaPacketNotFound + } + + return b.bucket.GetPacket(dst, extendedSN-sequenceNumberAdjustment) +} + +// FECDecoderStats returns cumulative FlexFEC decode counters of the buffer. +func (b *BufferBase) FECDecoderStats() flexfec.DecoderStats { + b.RLock() + defer b.RUnlock() + + if b.fecDecoder == nil { + return flexfec.DecoderStats{} + } + return b.fecDecoder.Stats() +} + +// HandleFECPacket decodes a packet from the coupled FlexFEC repair stream +// and injects recovered media into the regular packet pipeline. It reports +// whether the packet was passed to the decoder. +func (b *BufferBase) HandleFECPacket(fecPkt *rtp.Packet, arrivalTime int64) bool { + b.Lock() + defer b.Unlock() + + return b.HandleFECPacketLocked(fecPkt, arrivalTime) +} + +// HandleFECPacketLocked requires the buffer lock to be held. Recovery updates +// NACK state before callers generate feedback. +func (b *BufferBase) HandleFECPacketLocked(fecPkt *rtp.Packet, arrivalTime int64) bool { + if b.isClosed.Load() || b.bucket == nil { + return false + } + + if b.fecPayloadType == 0 || fecPkt.PayloadType != b.fecPayloadType { + b.logger.Debugw("unexpected fec payload type", "expected", b.fecPayloadType, "actual", fecPkt.PayloadType) + return false + } + + if b.fecDecoder == nil { + // normally created when the pair is declared, fall back to the + // observed repair stream SSRC + b.fecSSRC = fecPkt.SSRC + b.maybeCreateFECDecoderLocked() + if b.fecDecoder == nil { + b.logger.Infow("flexfec decoder unavailable after creation attempt", "fecSSRC", b.fecSSRC, "mediaSSRC", b.SSRC(), "payloadType", b.fecPayloadType) + return false + } + } + + b.feedFECLocked(fecPkt, arrivalTime) + return true +} + +// feedFECLocked runs a media or FEC packet through the FEC decoder and +// injects recovered packets into the packet pipeline. Must be called with the +// buffer lock held and a non-nil decoder. +func (b *BufferBase) feedFECLocked(pkt *rtp.Packet, arrivalTime int64) { + statsBefore := b.fecDecoder.Stats() + recovered := b.fecDecoder.DecodeFEC(pkt) + statsAfter := b.fecDecoder.Stats() + + fecPackets := statsAfter.FECPacketsReceived - statsBefore.FECPacketsReceived + fecBytes := statsAfter.FECBytesReceived - statsBefore.FECBytesReceived + fecPacketsDiscarded := statsAfter.FECPacketsDiscarded - statsBefore.FECPacketsDiscarded + fecPacketsRecovered := statsAfter.PacketsRecovered - statsBefore.PacketsRecovered + if b.rtpStats != nil { + b.rtpStats.UpdateFEC( + fecPackets, + fecBytes, + fecPacketsDiscarded, + fecPacketsRecovered, + ) + } + + prometheus.RecordFECUpstream(int(fecPackets), int(fecPacketsRecovered), int(fecPacketsDiscarded), fecBytes) + + if len(recovered) > 0 && b.fecPktBuf == nil { + b.fecPktBuf = make([]byte, bucket.RTPMaxPktSize) + } + for _, rp := range recovered { + n, err := rp.MarshalTo(b.fecPktBuf) + if err != nil { + b.logger.Warnw("could not marshal fec recovered packet", err, "ssrc", b.SSRC(), "sn", rp.SequenceNumber) + continue + } + + // 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. RTCP is generated once by the caller + // after all recovered packets have updated NACK state. + b.handleIncomingPacketLocked(b.fecPktBuf[:n], rp, arrivalTime, false, true, nil, 0) + } +} + func (b *BufferBase) updateNACKState(sequenceNumber uint16, flowState rtpstats.RTPFlowState) { if b.nacker == nil { return diff --git a/pkg/sfu/buffer/buffer_fec_test.go b/pkg/sfu/buffer/buffer_fec_test.go index 65f92adbd..786b40d28 100644 --- a/pkg/sfu/buffer/buffer_fec_test.go +++ b/pkg/sfu/buffer/buffer_fec_test.go @@ -27,11 +27,15 @@ import ( "github.com/pion/rtp" "github.com/pion/transport/v4/packetio" "github.com/pion/webrtc/v4" + promclient "github.com/prometheus/client_golang/prometheus" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "github.com/livekit/livekit-server/pkg/sfu/testutils/rtptest" + "github.com/livekit/livekit-server/pkg/telemetry/prometheus" "github.com/livekit/mediatransportutil/pkg/twcc" + "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/utils/mono" ) const ( @@ -49,14 +53,43 @@ var flexFECCodec = webrtc.RTPCodecParameters{ PayloadType: webrtc.PayloadType(fecTestFECPT), } -func bindFECTestBuffer(t *testing.T, buff *Buffer) { +func bindFECTestBuffer(t *testing.T, buff interface { + Bind(webrtc.RTPParameters, webrtc.RTPCodecCapability, int) error +}) { t.Helper() - buff.codecType = webrtc.RTPCodecTypeVideo + require.NoError(t, prometheus.Init("test", livekit.NodeType_SERVER)) require.NoError(t, buff.Bind(webrtc.RTPParameters{ Codecs: []webrtc.RTPCodecParameters{vp8Codec, flexFECCodec}, }, vp8Codec.RTPCodecCapability, 0)) } +func fecUpstreamCounters(t *testing.T) map[string]float64 { + t.Helper() + families, err := promclient.DefaultGatherer.Gather() + require.NoError(t, err) + counters := make(map[string]float64) + for _, family := range families { + if family.GetName() != "livekit_fec_total" && family.GetName() != "livekit_fec_bytes" { + continue + } + for _, metric := range family.Metric { + var direction, kind string + for _, label := range metric.Label { + switch label.GetName() { + case "direction": + direction = label.GetValue() + case "type": + kind = label.GetValue() + } + } + if direction == "incoming" { + counters[family.GetName()+"/"+kind] += metric.GetCounter().GetValue() + } + } + } + return counters +} + func writePacket(t *testing.T, buff *Buffer, pkt *rtp.Packet) { t.Helper() raw, err := pkt.Marshal() @@ -160,12 +193,7 @@ func TestBufferFECRecoversDroppedPacket(t *testing.T) { factory.SetFECPair(fecTestFECSSRC, fecTestMediaSSRC) bindFECTestBuffer(t, primary) - - var recoveredDelta, receivedDelta int - primary.OnFECRecovery(func(received int, recovered int, discarded int, bytesReceived int) { - recoveredDelta += recovered - receivedDelta += received - }) + metricsBefore := fecUpstreamCounters(t) media := rtptest.GenerateVP8Packets(100, 10, uint8(vp8Codec.PayloadType), fecTestMediaSSRC) encoder := pionflexfec.NewFlexEncoder03(fecTestFECPT, fecTestFECSSRC) @@ -187,8 +215,6 @@ func TestBufferFECRecoversDroppedPacket(t *testing.T) { assert.EqualValues(t, len(fecPackets), stats.FECPacketsReceived) assert.EqualValues(t, 1, stats.PacketsRecovered) assert.EqualValues(t, 0, stats.FECPacketsDiscarded) - assert.Equal(t, 1, recoveredDelta) - assert.Equal(t, len(fecPackets), receivedDelta) fecPayloadBytes := 0 for i := range fecPackets { @@ -200,6 +226,10 @@ func TestBufferFECRecoversDroppedPacket(t *testing.T) { assert.EqualValues(t, fecPayloadBytes, rtpStats.FecBytes) assert.EqualValues(t, 0, rtpStats.FecPacketsDiscarded) assert.EqualValues(t, 1, rtpStats.FecPacketsRecovered) + metricsAfter := fecUpstreamCounters(t) + assert.EqualValues(t, len(fecPackets), metricsAfter["livekit_fec_total/received"]-metricsBefore["livekit_fec_total/received"]) + assert.EqualValues(t, 1, metricsAfter["livekit_fec_total/recovered"]-metricsBefore["livekit_fec_total/recovered"]) + assert.EqualValues(t, fecPayloadBytes, metricsAfter["livekit_fec_bytes/received"]-metricsBefore["livekit_fec_bytes/received"]) // the 9 received packets flow through the ext packet pipeline, the // recovered one fills the bucket like an RTX repair @@ -262,46 +292,77 @@ func TestBufferFECRecoversEncryptedPayload(t *testing.T) { assert.Equal(t, plaintexts[droppedIdx], decrypted) } -func TestBufferFECRecoveryCallbackCanReenterBuffer(t *testing.T) { - factory := NewFactoryOfBufferFactory(500, 200).CreateBufferFactory() +func TestBufferBaseFECRecovery(t *testing.T) { + for _, test := range []struct { + name string + pairOnBind bool + lateMedia bool + lateRTX bool + parsePacket bool + }{ + {name: "paired before bind", pairOnBind: true}, + {name: "paired after bind"}, + {name: "late media", lateMedia: true, parsePacket: true}, + {name: "late RTX", lateMedia: true, lateRTX: true}, + } { + t.Run(test.name, func(t *testing.T) { + base := NewBufferBase(BufferBaseParams{ + SSRC: fecTestMediaSSRC, + MaxVideoPkts: 500, + MaxAudioPkts: 200, + }) + t.Cleanup(func() { base.CloseWithReason("test") }) + if test.pairOnBind { + base.SetFECSSRC(fecTestFECSSRC) + require.Nil(t, base.fecDecoder) + } + bindFECTestBuffer(t, base) + if !test.pairOnBind { + base.SetFECSSRC(fecTestFECSSRC) + } + require.NotNil(t, base.fecDecoder) - primary := factory.GetOrNew(packetio.RTPBufferPacket, fecTestMediaSSRC).(*Buffer) - fecBuff := factory.GetOrNew(packetio.RTPBufferPacket, fecTestFECSSRC).(*Buffer) - factory.SetFECPair(fecTestFECSSRC, fecTestMediaSSRC) - bindFECTestBuffer(t, primary) + media := rtptest.GenerateVP8Packets(150, 5, uint8(vp8Codec.PayloadType), fecTestMediaSSRC) + fecPackets := pionflexfec.NewFlexEncoder03(fecTestFECPT, fecTestFECSSRC).EncodeFec(media, 1) + require.Len(t, fecPackets, 1) - callbackDone := make(chan struct{}, 1) - primary.OnFECRecovery(func(received int, recovered int, discarded int, bytesReceived int) { - primary.FECDecoderStats() - callbackDone <- struct{}{} - }) + ingest := func(pkt *rtp.Packet, isRTX bool) { + raw, err := pkt.Marshal() + require.NoError(t, err) + parsed := pkt + if test.parsePacket { + parsed = nil + } + _, err = base.HandleIncomingPacket(raw, parsed, mono.UnixNano(), false, isRTX, nil, 0) + require.NoError(t, err) + } + for i := range media { + if i != 2 && (i != 3 || !test.lateMedia) { + ingest(&media[i], false) + } + } + base.HandleFECPacket(&fecPackets[0], mono.UnixNano()) + if test.lateMedia { + require.Zero(t, base.FECDecoderStats().PacketsRecovered) + ingest(&media[3], test.lateRTX) + } - media := rtptest.GenerateVP8Packets(150, 5, uint8(vp8Codec.PayloadType), fecTestMediaSSRC) - for i := range media { - writePacket(t, primary, &media[i]) - } - fecPackets := pionflexfec.NewFlexEncoder03(fecTestFECPT, fecTestFECSSRC).EncodeFec(media, 1) - require.NotEmpty(t, fecPackets) + stats := base.GetStats() + require.EqualValues(t, 1, stats.FecPackets) + require.EqualValues(t, len(fecPackets[0].Payload), stats.FecBytes) + require.EqualValues(t, 1, stats.FecPacketsRecovered) + require.EqualValues(t, 1, base.FECDecoderStats().PacketsRecovered) + require.Empty(t, base.GetNACKPairs()) - rawFEC, err := fecPackets[0].Marshal() - require.NoError(t, err) - writeDone := make(chan error, 1) - go func() { - _, err := fecBuff.Write(rawFEC) - writeDone <- err - }() - - select { - case <-callbackDone: - case <-time.After(time.Second): - t.Fatal("FEC recovery callback deadlocked while re-entering the buffer") - } - - select { - case err := <-writeDone: - require.NoError(t, err) - case <-time.After(time.Second): - t.Fatal("FEC packet write did not return after callback") + var raw [1500]byte + n, err := base.GetPacket(raw[:], uint64(media[2].SequenceNumber)) + require.NoError(t, err) + var recovered rtp.Packet + require.NoError(t, recovered.Unmarshal(raw[:n])) + assert.Equal(t, media[2].Header, recovered.Header) + assert.Equal(t, media[2].Payload, recovered.Payload) + assert.Equal(t, len(media), base.extPackets.Len()) + }) } }