From 7b02897ce235d29be53bb4eedd08ae0d4c8a971e Mon Sep 17 00:00:00 2001 From: David Chen Date: Sat, 12 Sep 2026 14:24:34 -0700 Subject: [PATCH] Align downstream FEC grouping with frame boundaries Generate repairs at each frame marker, splitting large frames into blocks of at most 48 packets. Match upstream Q8 repair-count rounding and minimum-one protection, and lazily reuse packet buffers. Reserve measured RTP repair overhead during layer allocation so small frames have enough headroom. Cover sparse frames, large-frame tails, buffer reuse, real Pion wire recovery, and live preset changes. --- config-sample.yaml | 5 +- pkg/sfu/downtrack.go | 1 + pkg/sfu/downtrack_fec.go | 27 +- pkg/sfu/downtrack_fec_integration_test.go | 57 +-- pkg/sfu/downtrack_fec_test.go | 50 ++- pkg/sfu/flexfec/encoder.go | 103 ++++-- pkg/sfu/flexfec/encoder_test.go | 402 +++++++++++++--------- pkg/sfu/pacer/base.go | 5 +- pkg/sfu/pacer/base_fec_test.go | 6 +- 9 files changed, 416 insertions(+), 240 deletions(-) diff --git a/config-sample.yaml b/config-sample.yaml index 68ea80813..e8a360d91 100644 --- a/config-sample.yaml +++ b/config-sample.yaml @@ -125,7 +125,10 @@ rtc: # # Allow repairs for subscribers that negotiate FlexFEC. Clients select fec # # per track in UpdateTrackSettings: FEC_NONE (default), FEC_LOW (15%), # # FEC_MEDIUM (25%), or FEC_HIGH (35%) repair packet overhead, matching publish - # # track presets. Media is sent immediately; incomplete groups + # # track presets. These are approximate: each protected frame/block gets at + # # least one repair, so small frames can exceed the nominal percentage. + # # Media is sent immediately; repairs follow the end of each frame, with + # # large frames split into blocks of at most 48 packets. Stale partial frames # # spanning more than 200ms and packets without room for repair headers are skipped. # downstream_enabled: false # # allows automatic connection fallback to TCP and TURN/TLS (if configured) when UDP has been unstable, default true diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index 296d2b910..084048707 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -273,6 +273,7 @@ type DownTrack struct { fecLock sync.Mutex fecEncoder atomic.Pointer[flexfec.Encoder] fecProtectionPercent atomic.Uint32 + fecNotifiedOverhead atomic.Uint32 fecState flexfec.EncoderState receiverLock sync.RWMutex diff --git a/pkg/sfu/downtrack_fec.go b/pkg/sfu/downtrack_fec.go index cb77aea67..b545df9e1 100644 --- a/pkg/sfu/downtrack_fec.go +++ b/pkg/sfu/downtrack_fec.go @@ -35,9 +35,10 @@ func (d *DownTrack) bindFEC(t webrtc.TrackLocalContext) { } for _, codec := range d.negotiatedCodecParameters { if strings.EqualFold(codec.MimeType, webrtc.MimeTypeFlexFEC03) { - encoder := flexfec.NewEncoder(uint8(codec.PayloadType), uint32(t.SSRCForwardErrorCorrection()), d.params.OnFECSent) + encoder := flexfec.NewEncoder(uint8(codec.PayloadType), uint32(t.SSRCForwardErrorCorrection()), d.onFECSent) encoder.SeedState(d.fecState) encoder.SetProtectionPercent(d.fecProtectionPercent.Load()) + d.fecNotifiedOverhead.Store(encoder.OverheadPercent()) d.fecEncoder.Store(encoder) return } @@ -79,12 +80,27 @@ func (d *DownTrack) seedFECState(state flexfec.EncoderState) { } } -// Include the nominal repair overhead in every allocator input, so layer -// selection leaves room for repair traffic. The pacer/BWE account actual bytes. +func (d *DownTrack) onFECSent(packets int, bytes int) { + if d.params.OnFECSent != nil { + d.params.OnFECSent(packets, bytes) + } + if encoder := d.fecEncoder.Load(); encoder != nil { + percent := encoder.OverheadPercent() + if d.fecNotifiedOverhead.Swap(percent) != percent { + if listener := d.getStreamAllocatorListener(); listener != nil { + listener.OnSubscriptionChanged(d) + } + } + } +} + +// Reserve the greater of configured and measured repair overhead in every +// allocator input. Small frames can require much more than the nominal preset. +// The pacer/BWE account actual bytes, including all repair headers. func (d *DownTrack) getLayeredBitrateWithFEC() ([]int32, Bitrates) { layers, bitrates := d.Receiver().GetLayeredBitrate() - if d.fecEncoder.Load() != nil { - percent := int64(d.fecProtectionPercent.Load()) + if encoder := d.fecEncoder.Load(); encoder != nil { + percent := int64(encoder.OverheadPercent()) if percent == 0 { return layers, bitrates } @@ -120,6 +136,7 @@ func (d *DownTrack) SetFECProtection(level livekit.FECProtection) { encoder := d.fecEncoder.Load() if encoder != nil { encoder.SetProtectionPercent(percent) + d.fecNotifiedOverhead.Store(percent) } d.fecLock.Unlock() if encoder != nil { diff --git a/pkg/sfu/downtrack_fec_integration_test.go b/pkg/sfu/downtrack_fec_integration_test.go index 95abba6e1..71bb5812d 100644 --- a/pkg/sfu/downtrack_fec_integration_test.go +++ b/pkg/sfu/downtrack_fec_integration_test.go @@ -62,16 +62,20 @@ func TestDownTrackFlexFECOnWire(t *testing.T) { pacing string level *livekit.FECProtection protectionPercent uint32 + framePackets, frameRepairs int }{ - {"pass through", true, true, false, "pass", livekit.FECProtection_FEC_MEDIUM.Enum(), 25}, - {"queued", true, true, false, "queue", livekit.FECProtection_FEC_MEDIUM.Enum(), 25}, - {"low", true, true, false, "queue", livekit.FECProtection_FEC_LOW.Enum(), 15}, - {"high", true, true, false, "leaky", livekit.FECProtection_FEC_HIGH.Enum(), 35}, - {"none", true, true, false, "pass", livekit.FECProtection_FEC_NONE.Enum(), 0}, - {"default none", true, true, false, "pass", nil, 0}, - {"paced encrypted", true, true, true, "leaky", livekit.FECProtection_FEC_MEDIUM.Enum(), 25}, - {"subscriber declines", true, false, false, "pass", livekit.FECProtection_FEC_HIGH.Enum(), 35}, - {"disabled", false, true, false, "pass", livekit.FECProtection_FEC_HIGH.Enum(), 35}, + {"pass through", true, true, false, "pass", livekit.FECProtection_FEC_MEDIUM.Enum(), 25, 20, 5}, + {"queued", true, true, false, "queue", livekit.FECProtection_FEC_MEDIUM.Enum(), 25, 20, 5}, + {"low", true, true, false, "queue", livekit.FECProtection_FEC_LOW.Enum(), 15, 20, 3}, + {"high", true, true, false, "leaky", livekit.FECProtection_FEC_HIGH.Enum(), 35, 20, 7}, + {"none", true, true, false, "pass", livekit.FECProtection_FEC_NONE.Enum(), 0, 20, 0}, + {"default none", true, true, false, "pass", nil, 0, 20, 0}, + {"paced encrypted", true, true, true, "leaky", livekit.FECProtection_FEC_MEDIUM.Enum(), 25, 20, 5}, + {"subscriber declines", true, false, false, "pass", livekit.FECProtection_FEC_HIGH.Enum(), 35, 20, 0}, + {"disabled", false, true, false, "pass", livekit.FECProtection_FEC_HIGH.Enum(), 35, 20, 0}, + {"single packet frames", true, true, false, "pass", livekit.FECProtection_FEC_LOW.Enum(), 15, 1, 1}, + {"small frames", true, true, false, "queue", livekit.FECProtection_FEC_LOW.Enum(), 15, 2, 1}, + {"large frames", true, true, false, "leaky", livekit.FECProtection_FEC_HIGH.Enum(), 35, 99, 35}, } { t.Run(tc.name, func(t *testing.T) { h := vnettest.NewHosts(t) @@ -150,13 +154,15 @@ func TestDownTrackFlexFECOnWire(t *testing.T) { require.Zero(t, fecSSRC) } - const mediaCount = 20 + const frameCount = 2 + mediaCount := frameCount * tc.framePackets sendMedia := func(start, count int) { for i := start; i < start+count; i++ { + frameIndex := i / tc.framePackets ep, err := testutils.GetTestExtPacketVP8(&testutils.TestExtPacketParams{ - SequenceNumber: uint16(6000 + i), Timestamp: uint32(270000 + i*3000), SSRC: 0x44444444, - PayloadType: 96, PayloadSize: 100 + i, IsKeyFrame: true, Marker: true, - }, &codec.VP8{FirstByte: 0x10, S: true, PictureID: uint16(i + 1), IsKeyFrame: true}) + SequenceNumber: uint16(6000 + i), Timestamp: uint32(270000 + frameIndex*3000), SSRC: 0x44444444, + PayloadType: 96, PayloadSize: 100 + i, IsKeyFrame: true, Marker: i%tc.framePackets == tc.framePackets-1, + }, &codec.VP8{FirstByte: 0x10, S: i%tc.framePackets == 0, PictureID: uint16(frameIndex + 1), IsKeyFrame: true}) require.NoError(t, err) ep.Packet.Payload = distinctivePayload(byte(i), 100+i) require.EqualValues(t, 1, dt.WriteRTP(ep, 0)) @@ -166,7 +172,7 @@ func TestDownTrackFlexFECOnWire(t *testing.T) { sendMedia(0, mediaCount) expected := mediaCount if negotiated { - expected += mediaCount * int(tc.protectionPercent) / 100 + expected += frameCount * tc.frameRepairs } require.Eventually(t, func() bool { return capture.count() >= expected }, 10*time.Second, 10*time.Millisecond) require.Equal(t, expected, capture.count()) @@ -187,16 +193,15 @@ func TestDownTrackFlexFECOnWire(t *testing.T) { require.Empty(t, repair) return } - require.Len(t, repair, mediaCount*int(tc.protectionPercent)/100) + require.Len(t, repair, frameCount*tc.frameRepairs) for i := 1; i < len(repair); i++ { require.Equal(t, repair[i-1].SequenceNumber+1, repair[i].SequenceNumber) } - // Drop media in the second and third groups (protected by every enabled - // preset) and recover from the actual wire + // Drop media from the first and last frames and recover from the actual wire // representation, including translated SSRC/PT/sequence and payload. stored := map[uint16][]byte{} for i, packet := range media { - if i != 7 && i != 12 { + if i != 0 && i != mediaCount-1 { stored[packet.SequenceNumber], _ = packet.Marshal() } } @@ -217,7 +222,7 @@ func TestDownTrackFlexFECOnWire(t *testing.T) { } } require.Len(t, recoveredPackets, 2) - for _, lost := range []*rtp.Packet{media[7], media[12]} { + for _, lost := range []*rtp.Packet{media[0], media[mediaCount-1]} { want, _ := lost.Marshal() require.Equal(t, want, recoveredPackets[lost.SequenceNumber]) } @@ -226,18 +231,18 @@ func TestDownTrackFlexFECOnWire(t *testing.T) { sequence := repair[len(repair)-1].SequenceNumber start := mediaCount for _, update := range []struct { - level livekit.FECProtection - percent int + level livekit.FECProtection + repairsPerFrame int }{ - {livekit.FECProtection_FEC_HIGH, 35}, + {livekit.FECProtection_FEC_HIGH, 7}, {livekit.FECProtection_FEC_NONE, 0}, - {livekit.FECProtection_FEC_LOW, 15}, + {livekit.FECProtection_FEC_LOW, 3}, } { before := capture.count() dt.SetFECProtection(update.level) sendMedia(start, mediaCount) start += mediaCount - numRepair := mediaCount * update.percent / 100 + numRepair := frameCount * update.repairsPerFrame require.Eventually(t, func() bool { return capture.count() >= before+mediaCount+numRepair }, 5*time.Second, 10*time.Millisecond) require.Equal(t, before+mediaCount+numRepair, capture.count()) for _, packet := range capture.all()[before:] { @@ -281,7 +286,7 @@ func TestDownTrackFlexFECOnWire(t *testing.T) { dt.ForceForwardLayerForTest(buffer.VideoLayer{Spatial: 0, Temporal: 0}) before := capture.count() sendMedia(start, mediaCount) - require.Eventually(t, func() bool { return capture.count() >= before+25 }, 5*time.Second, 10*time.Millisecond) + require.Eventually(t, func() bool { return capture.count() >= before+mediaCount+frameCount*5 }, 5*time.Second, 10*time.Millisecond) repairs := 0 for _, packet := range capture.all()[before:] { if packet.SSRC == fecSSRC { @@ -290,7 +295,7 @@ func TestDownTrackFlexFECOnWire(t *testing.T) { repairs++ } } - require.Equal(t, 5, repairs) + require.Equal(t, frameCount*5, repairs) } }) } diff --git a/pkg/sfu/downtrack_fec_test.go b/pkg/sfu/downtrack_fec_test.go index 1fe62ad0b..378f25af1 100644 --- a/pkg/sfu/downtrack_fec_test.go +++ b/pkg/sfu/downtrack_fec_test.go @@ -57,10 +57,11 @@ func TestDownTrackFECNegotiationAndLifecycle(t *testing.T) { require.NotNil(t, old) p := rtp.Header{Version: 2, SSRC: 123, PayloadType: 96} var lastSequenceNumber uint16 - for i := range flexfec.MediaPacketsPerGroup { + for i := range 5 { p.SequenceNumber++ + p.Marker = i == 4 repair := old.Encode(&p, []byte{1, 2, 3}) - if i == flexfec.MediaPacketsPerGroup-1 { + if i == 4 { require.Len(t, repair, 1) require.EqualValues(t, 456, repair[0].SSRC) require.EqualValues(t, 118, repair[0].PayloadType, "use negotiated PT") @@ -69,8 +70,9 @@ func TestDownTrackFECNegotiationAndLifecycle(t *testing.T) { } d.bindFEC(c) require.NotSame(t, old, d.fecEncoder.Load()) - for range flexfec.MediaPacketsPerGroup { + for i := range 5 { p.SequenceNumber++ + p.Marker = i == 4 require.Empty(t, old.Encode(&p, []byte{1, 2, 3}), "old queued writes cannot generate repair") for _, repair := range d.fecEncoder.Load().Encode(&p, []byte{1, 2, 3}) { require.Equal(t, lastSequenceNumber+1, repair.SequenceNumber, "reusing an SSRC must preserve repair sequencing") @@ -89,7 +91,9 @@ func TestDownTrackFECBandwidthReservation(t *testing.T) { _, got := d.getLayeredBitrateWithFEC() require.Equal(t, rates, got) d.SetFECProtection(livekit.FECProtection_FEC_MEDIUM) - d.fecEncoder.Store(flexfec.NewEncoder(115, 456, nil)) + encoder := flexfec.NewEncoder(115, 456, nil) + encoder.SetProtectionPercent(25) + d.fecEncoder.Store(encoder) layers, got := d.getLayeredBitrateWithFEC() require.Equal(t, []int32{0, 1}, layers) require.EqualValues(t, 125_000, got[0][0]) @@ -105,6 +109,41 @@ type fecTrackReceiver struct { func (r *fecTrackReceiver) GetLayeredBitrate() ([]int32, Bitrates) { return []int32{0, 1}, r.rates } +type fecAllocatorListener struct { + DownTrackStreamAllocatorListener + changes int +} + +func (l *fecAllocatorListener) OnSubscriptionChanged(*DownTrack) { l.changes++ } + +func TestDownTrackFECMeasuredBandwidthReservation(t *testing.T) { + listener := &fecAllocatorListener{} + d := &DownTrack{ + params: DownTrackParams{EnableFlexFEC: true}, kind: webrtc.RTPCodecTypeVideo, + receiver: &fecTrackReceiver{rates: Bitrates{{100_000}}}, + negotiatedCodecParameters: []webrtc.RTPCodecParameters{{RTPCodecCapability: webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeFlexFEC03}, PayloadType: 115}}, + streamAllocatorListener: listener, + } + d.SetFECProtection(livekit.FECProtection_FEC_LOW) + d.bindFEC(fecTrackContext{ssrcFEC: 456}) + e := d.fecEncoder.Load() + p := rtp.Header{Version: 2, SSRC: 123, PayloadType: 96, Marker: true} + for range 2 { + p.SequenceNumber++ + repairs := e.Encode(&p, make([]byte, 100)) + require.Len(t, repairs, 1) + e.RecordSent(1, len(repairs[0].Payload), repairs[0].MarshalSize()) + } + _, rates := d.getLayeredBitrateWithFEC() + require.EqualValues(t, 218_000, rates[0][0], "reserve actual small-frame overhead, including RTP/FEC headers") + require.Equal(t, 1, listener.changes, "reallocate on an overhead change, not on every repair") + d.SetFECProtection(livekit.FECProtection_FEC_NONE) + _, rates = d.getLayeredBitrateWithFEC() + require.EqualValues(t, 100_000, rates[0][0]) + require.Equal(t, 2, listener.changes) + d.closeFEC() +} + func TestDownTrackFECProtectionChanges(t *testing.T) { d := &DownTrack{ params: DownTrackParams{EnableFlexFEC: true}, kind: webrtc.RTPCodecTypeVideo, @@ -121,8 +160,9 @@ func TestDownTrackFECProtectionChanges(t *testing.T) { p := rtp.Header{Version: 2, SSRC: 123, PayloadType: 96} send := func() int { count := 0 - for range 20 { + for i := range 20 { p.SequenceNumber++ + p.Marker = i == 19 count += len(d.fecEncoder.Load().Encode(&p, []byte{1, 2, 3})) } return count diff --git a/pkg/sfu/flexfec/encoder.go b/pkg/sfu/flexfec/encoder.go index 293dbd5a2..e0e863c28 100644 --- a/pkg/sfu/flexfec/encoder.go +++ b/pkg/sfu/flexfec/encoder.go @@ -21,6 +21,7 @@ import ( pionflexfec "github.com/pion/interceptor/pkg/flexfec" "github.com/pion/rtp" + "go.uber.org/atomic" "go.uber.org/zap/zapcore" "github.com/livekit/protocol/utils/mono" @@ -30,12 +31,14 @@ const ( // DefaultProtectionPercent is the repair-to-media packet ratio when unspecified. DefaultProtectionPercent uint32 = 0 MaxProtectionPercent uint32 = 100 - // MediaPacketsPerGroup bounds both retained packet memory and encoding work. - MediaPacketsPerGroup = 5 - maxEncoderGroupAge = 200 * time.Millisecond - // Reserve the FEC header and up to 16 bytes of outer RTP extensions (AST/TWCC). - // Groups of at most 15 packets use only the first FlexFEC packet mask. - maxEncoderMediaPacketSize = maxMediaPacketSize - pionflexfec.BaseFec03HeaderSize - 16 + // MaxMediaPacketsPerGroup bounds memory and work for large frames. Like + // libwebrtc, use at most 48 media packets in a protection block. + MaxMediaPacketsPerGroup = 48 + maxEncoderGroupAge = 200 * time.Millisecond + overheadUpdateInterval = time.Second + // Reserve all three FEC masks and up to 16 bytes of outer RTP extensions + // (AST/TWCC). A 48-packet group can require the third mask. + maxEncoderMediaPacketSize = maxMediaPacketSize - pionflexfec.BaseFec03HeaderSize - 12 - 16 ) // EncoderState preserves the repair sequence when a sender SSRC is reused. @@ -52,8 +55,8 @@ func (s EncoderState) MarshalLogObject(e zapcore.ObjectEncoder) error { } type encoderMediaGroup struct { - packets [MediaPacketsPerGroup]rtp.Packet - storage [MediaPacketsPerGroup][maxEncoderMediaPacketSize]byte + packets [MaxMediaPacketsPerGroup]rtp.Packet + storage [MaxMediaPacketsPerGroup]*[maxEncoderMediaPacketSize]byte } // Encoder batches final, outgoing video packets for Pion's FlexFEC-03 encoder. @@ -71,7 +74,12 @@ type Encoder struct { closed bool onSent func(packets int, bytes int) protectionPercent uint32 - repairCredit uint32 + protectionFactor uint32 + + overheadPercent atomic.Uint32 + overheadUpdatedAt int64 + sentMediaBytes uint64 + sentRepairBytes uint64 } func NewEncoder(payloadType uint8, ssrc uint32, onSent func(packets int, bytes int)) *Encoder { @@ -100,12 +108,13 @@ func (e *Encoder) SeedState(state EncoderState) { } e.sequenceNumber = state.NextSequenceNumber e.count = 0 - e.repairCredit = 0 } -// Encode returns independently owned repair packets for complete groups only. -// Gaps (including skipped padding), codec switches and stale groups start a new -// group, since Pion requires consecutive sequence numbers. RTP wrap is valid. +// Encode returns independently owned repairs at the end of a video frame (the +// RTP marker), splitting large frames at MaxMediaPacketsPerGroup. Timestamp +// changes discard an incomplete previous frame. Gaps, codec switches and stale +// groups also reset the block; Pion requires consecutive sequence numbers. +// Media is never delayed. No timer or worker is needed for sparse frames. // Call after the media write, including any interceptor header modifications. func (e *Encoder) Encode(header *rtp.Header, payload []byte) []rtp.Packet { e.mu.Lock() @@ -116,6 +125,7 @@ func (e *Encoder) Encode(header *rtp.Header, payload []byte) []rtp.Packet { now := mono.UnixNano() size := header.MarshalSize() + len(payload) + e.sentMediaBytes += uint64(size) if header.Padding || len(payload) == 0 || size > maxEncoderMediaPacketSize { e.count = 0 return nil @@ -129,6 +139,7 @@ func (e *Encoder) Encode(header *rtp.Header, payload []byte) []rtp.Packet { previous := &e.media.packets[e.count-1] if header.SequenceNumber != previous.SequenceNumber+1 || header.SSRC != previous.SSRC || header.PayloadType != previous.PayloadType || + header.Timestamp != previous.Timestamp || now-e.startedAt > int64(maxEncoderGroupAge) { e.count = 0 } @@ -137,6 +148,10 @@ func (e *Encoder) Encode(header *rtp.Header, payload []byte) []rtp.Packet { e.startedAt = now } + if e.media.storage[e.count] == nil { + // Retain only the packet buffers this stream has actually needed. + e.media.storage[e.count] = new([maxEncoderMediaPacketSize]byte) + } raw := e.media.storage[e.count][:size] n, err := header.MarshalTo(raw) if err != nil { @@ -149,21 +164,17 @@ func (e *Encoder) Encode(header *rtp.Header, payload []byte) []rtp.Packet { return nil } e.count++ - if e.count < MediaPacketsPerGroup { + if !header.Marker && e.count < MaxMediaPacketsPerGroup { return nil } + count := e.count e.count = 0 - // Carry fractional packets between groups rather than rounding each group - // up (which would turn e.g. 30% into 40%). Storage and latency stay bounded - // at five media packets, independent of the configured percentage. - e.repairCredit += e.protectionPercent * MediaPacketsPerGroup - numRepair := e.repairCredit / 100 - e.repairCredit %= 100 - if numRepair == 0 { - return nil - } - repair := e.encoder.EncodeFec(e.media.packets[:], numRepair) + // Match libwebrtc's ForwardErrorCorrection::NumFecPackets: round the Q8 + // protection factor, with at least one repair for every protected block. + // In particular, a one-packet frame is protected immediately at every preset. + numRepair := max(uint32(1), (uint32(count)*e.protectionFactor+128)>>8) + repair := e.encoder.EncodeFec(e.media.packets[:count], numRepair) for i := range repair { // Pion's encoder uses a constant timestamp and a fixed initial sequence number. // Use this stream's media clock and a randomized, continuous repair sequence. @@ -175,24 +186,54 @@ func (e *Encoder) Encode(header *rtp.Header, payload []byte) []rtp.Packet { } // SetProtectionPercent updates the packet ratio without changing the repair -// SSRC or sequence number. A new value discards partial groups and old credit. +// SSRC or sequence number. A new value discards partial groups and measurements. // Values above 100 are capped; zero disables encoding before any packet copies. func (e *Encoder) SetProtectionPercent(percent uint32) { percent = min(percent, MaxProtectionPercent) e.mu.Lock() if e.protectionPercent != percent { e.protectionPercent = percent + // Publish-track presets use the same conversion to libwebrtc's Q8 rate. + e.protectionFactor = percent * 255 / 100 e.count = 0 - e.repairCredit = 0 + e.overheadPercent.Store(percent) + e.overheadUpdatedAt = 0 + e.sentMediaBytes = 0 + e.sentRepairBytes = 0 } e.mu.Unlock() } -// RecordSent reports successful repair writes, with RTP payload bytes like the -// upstream counters. Callbacks run outside the encoder lock. -func (e *Encoder) RecordSent(packets int, bytes int) { - if e.onSent != nil && packets != 0 { - e.onSent(packets, bytes) +// OverheadPercent reserves at least the configured rate, using measured RTP +// bytes when frame rounding or repair headers increase the actual overhead. +// Reads by the allocator do not contend with encoding. +func (e *Encoder) OverheadPercent() uint32 { + return e.overheadPercent.Load() +} + +// RecordSent accounts successful repairs, including headers for the allocator +// and payload bytes for the upstream-compatible telemetry callback. Refresh the +// estimate on the first repair and at most once per second thereafter. +// Callbacks run outside the encoder lock. +func (e *Encoder) RecordSent(packets int, payloadBytes int, rtpBytes int) { + if packets == 0 { + return + } + e.mu.Lock() + if !e.closed && e.protectionPercent != 0 { + e.sentRepairBytes += uint64(rtpBytes) + now := mono.UnixNano() + if e.sentMediaBytes != 0 && (e.overheadUpdatedAt == 0 || now-e.overheadUpdatedAt >= int64(overheadUpdateInterval)) { + percent := uint32((e.sentRepairBytes*100 + e.sentMediaBytes - 1) / e.sentMediaBytes) + e.overheadPercent.Store(max(e.protectionPercent, percent)) + e.overheadUpdatedAt = now + e.sentMediaBytes = 0 + e.sentRepairBytes = 0 + } + } + e.mu.Unlock() + if e.onSent != nil { + e.onSent(packets, payloadBytes) } } diff --git a/pkg/sfu/flexfec/encoder_test.go b/pkg/sfu/flexfec/encoder_test.go index f4e1d4688..e4394e5ff 100644 --- a/pkg/sfu/flexfec/encoder_test.go +++ b/pkg/sfu/flexfec/encoder_test.go @@ -25,15 +25,74 @@ import ( "github.com/livekit/protocol/logger" ) +func makeEncoderFrame(t *testing.T, base uint16, count int) []rtp.Packet { + t.Helper() + packets := makeMediaPackets(t, base, count) + for i := range packets { + packets[i].Timestamp = uint32(base) * 3000 + } + return packets +} + func TestEncoderDefaultsToNoProtection(t *testing.T) { - encoder := NewEncoder(testFECPT, testFECSSRC, nil) - p := rtp.Header{Version: 2, SSRC: testMediaSSRC, PayloadType: testMediaPT} + e := NewEncoder(testFECPT, testFECSSRC, nil) + p := rtp.Header{Version: 2, SSRC: testMediaSSRC, PayloadType: testMediaPT, Marker: true} for range 100 { p.SequenceNumber++ - require.Empty(t, encoder.Encode(&p, []byte{1, 2, 3})) + require.Empty(t, e.Encode(&p, []byte{1, 2, 3})) + } + require.Zero(t, e.count) + require.Nil(t, e.media, "negotiation alone must not allocate packet storage") +} + +func TestEncoderFrameProtectionCounts(t *testing.T) { + // Expected counts from upstream's 38/63/89 Q8 presets, including minimum-one + // protection and rounding cases where a nominal percentage alone differs. + for _, tc := range []struct{ packets, low, medium, high int }{ + {1, 1, 1, 1}, {2, 1, 1, 1}, {3, 1, 1, 1}, {5, 1, 1, 2}, + {6, 1, 1, 2}, {8, 1, 2, 3}, {20, 3, 5, 7}, {48, 7, 12, 17}, + } { + for _, level := range []struct { + percent uint32 + want int + }{ + {0, 0}, {15, tc.low}, {25, tc.medium}, {35, tc.high}, {100, tc.packets}, {101, tc.packets}, + } { + t.Run(fmt.Sprintf("packets_%d/percent_%d", tc.packets, level.percent), func(t *testing.T) { + e := NewEncoder(testFECPT, testFECSSRC, nil) + e.SetProtectionPercent(level.percent) + for i, p := range makeEncoderFrame(t, 100, tc.packets) { + repairs := e.Encode(&p.Header, p.Payload) + if i == tc.packets-1 { + require.Len(t, repairs, level.want) + for _, repair := range repairs { + require.Equal(t, p.Timestamp, repair.Timestamp) + } + } else { + require.Empty(t, repairs, "do not close the group before the frame ends") + } + } + require.Zero(t, e.count) + }) + } + } +} + +func TestEncoderSparseFrames(t *testing.T) { + e := NewEncoder(testFECPT, testFECSSRC, nil) + e.SetProtectionPercent(15) + // No wall-clock wait is necessary: complete frames must not depend on the + // previous frame's group age or on the arrival of a later frame. + for i := range 3 { + e.startedAt -= int64(10 * maxEncoderGroupAge) + p := makeEncoderFrame(t, uint16(100+i), 1)[0] + repairs := e.Encode(&p.Header, p.Payload) + require.Len(t, repairs, 1, "even the first isolated frame needs immediate protection") + decoder := newTestDecoder(testFECSSRC, testMediaSSRC, logger.GetLogger()) + recovered := decoder.DecodeFEC(&repairs[0]) + require.Len(t, recovered, 1) + requirePacketEqual(t, &p, recovered[0]) } - require.Zero(t, encoder.count, "disabled protection must not retain media") - require.Nil(t, encoder.media, "negotiation alone must not allocate packet storage") } func TestEncoderState(t *testing.T) { @@ -41,14 +100,10 @@ func TestEncoderState(t *testing.T) { e.SetProtectionPercent(20) e.SeedState(EncoderState{SSRC: testFECSSRC, NextSequenceNumber: 65535}) for _, sn := range []uint16{65535, 0} { - media := makeMediaPackets(t, 100, MediaPacketsPerGroup) - for i := range media { - repair := e.Encode(&media[i].Header, media[i].Payload) - if i == MediaPacketsPerGroup-1 { - require.Len(t, repair, 1) - require.Equal(t, sn, repair[0].SequenceNumber) - } - } + p := makeEncoderFrame(t, 100, 1)[0] + repair := e.Encode(&p.Header, p.Payload) + require.Len(t, repair, 1) + require.Equal(t, sn, repair[0].SequenceNumber) e.Close() state := e.GetState() require.Equal(t, sn+1, state.NextSequenceNumber) @@ -63,10 +118,10 @@ func TestEncoderState(t *testing.T) { func TestEncoderRecoveryWithReusedMemory(t *testing.T) { for _, base := range []uint16{100, 65533} { - media := makeMediaPackets(t, base, 2*MediaPacketsPerGroup) - encoder := NewEncoder(testFECPT, testFECSSRC, nil) - encoder.SetProtectionPercent(20) - encoder.sequenceNumber = 65535 + media := append(makeEncoderFrame(t, base, 5), makeEncoderFrame(t, base+5, 5)...) + e := NewEncoder(testFECPT, testFECSSRC, nil) + e.SetProtectionPercent(20) + e.sequenceNumber = 65535 decoder := newTestDecoder(testFECSSRC, testMediaSSRC, logger.GetLogger()) var repair []rtp.Packet for i := range media { @@ -74,8 +129,7 @@ func TestEncoderRecoveryWithReusedMemory(t *testing.T) { require.NoError(t, media[i].SetExtension(3, []byte{byte(i), 2, 3})) require.NoError(t, media[i].SetExtension(5, []byte{7, byte(i)})) packet := media[i].Clone() - repair = append(repair, encoder.Encode(&packet.Header, packet.Payload)...) - // The caller owns and immediately recycles every part of the input. + repair = append(repair, e.Encode(&packet.Header, packet.Payload)...) clear(packet.Payload) clear(packet.CSRC) clear(packet.GetExtension(3)) @@ -88,28 +142,23 @@ func TestEncoderRecoveryWithReusedMemory(t *testing.T) { require.Equal(t, uint16(65535), repair[0].SequenceNumber) require.Zero(t, repair[1].SequenceNumber) for i := range repair { - require.Equal(t, media[(i+1)*MediaPacketsPerGroup-1].Timestamp, repair[i].Timestamp) recovered := decoder.DecodeFEC(&repair[i]) require.Len(t, recovered, 1) - expected, err := media[i*MediaPacketsPerGroup+2].Marshal() - require.NoError(t, err) - actual, err := recovered[0].Marshal() - require.NoError(t, err) - require.Equal(t, expected, actual) + requirePacketEqual(t, &media[i*5+2], recovered[0]) } } } func TestEncoderGroupBoundaries(t *testing.T) { - for _, name := range []string{"gap", "duplicate", "out of order", "codec", "ssrc", "stale", "oversized", "padding", "empty"} { + for _, name := range []string{"gap", "duplicate", "out of order", "codec", "ssrc", "timestamp", "stale", "oversized", "padding", "empty"} { t.Run(name, func(t *testing.T) { e := NewEncoder(testFECPT, testFECSSRC, nil) e.SetProtectionPercent(20) - media := makeMediaPackets(t, 100, 10) - for i := 0; i < 4; i++ { + media := makeEncoderFrame(t, 100, 5) + for i := 0; i < 3; i++ { require.Empty(t, e.Encode(&media[i].Header, media[i].Payload)) } - p := &media[4] + p := &media[3] switch name { case "gap": p.SequenceNumber++ @@ -121,6 +170,8 @@ func TestEncoderGroupBoundaries(t *testing.T) { p.PayloadType++ case "ssrc": p.SSRC++ + case "timestamp": + p.Timestamp++ case "stale": e.startedAt -= int64(maxEncoderGroupAge) case "oversized": @@ -130,56 +181,75 @@ func TestEncoderGroupBoundaries(t *testing.T) { case "empty": p.Payload = nil } - require.Empty(t, e.Encode(&p.Header, p.Payload), "must not protect a discontinuous group") - fresh := makeMediaPackets(t, 200, MediaPacketsPerGroup) - for i := range fresh { - out := e.Encode(&fresh[i].Header, fresh[i].Payload) - if i == MediaPacketsPerGroup-1 { - require.Len(t, out, 1, "resume protection after the discontinuity") - } else { - require.Empty(t, out) - } - } + require.Empty(t, e.Encode(&p.Header, p.Payload)) + require.LessOrEqual(t, e.count, 1, "discard the previous partial block") + fresh := makeEncoderFrame(t, 200, 1)[0] + repairs := e.Encode(&fresh.Header, fresh.Payload) + require.Len(t, repairs, 1) + decoder := newTestDecoder(testFECSSRC, testMediaSSRC, logger.GetLogger()) + recovered := decoder.DecodeFEC(&repairs[0]) + require.Len(t, recovered, 1, "a missing marker must not mix consecutive frames") + requirePacketEqual(t, &fresh, recovered[0]) }) } } +func TestEncoderLargeFrame(t *testing.T) { + e := NewEncoder(testFECPT, testFECSSRC, nil) + e.SetProtectionPercent(35) + media := makeEncoderFrame(t, 65510, 2*MaxMediaPacketsPerGroup+3) + decoder := newTestDecoder(testFECSSRC, testMediaSSRC, logger.GetLogger()) + var repair []rtp.Packet + for i := range media { + out := e.Encode(&media[i].Header, media[i].Payload) + switch i { + case 47, 95: + require.Len(t, out, 17) + case 98: + require.Len(t, out, 1, "protect the tail of the large frame") + default: + require.Empty(t, out) + } + require.Less(t, e.count, MaxMediaPacketsPerGroup) + repair = append(repair, out...) + if i != 47 && i != 95 && i != 98 { + decoder.DecodeFEC(&media[i]) + } + } + var recovered []*rtp.Packet + for i := range repair { + recovered = append(recovered, decoder.DecodeFEC(&repair[i])...) + } + require.Len(t, recovered, 3) + for i, lost := range []int{47, 95, 98} { + requirePacketEqual(t, &media[lost], recovered[i]) + } +} + func TestEncoderSizeLimitAndClose(t *testing.T) { e := NewEncoder(testFECPT, testFECSSRC, nil) - e.SetProtectionPercent(20) - p := &rtp.Packet{Header: rtp.Header{Version: 2, SSRC: testMediaSSRC, PayloadType: testMediaPT}, Payload: make([]byte, maxEncoderMediaPacketSize-12)} - for i := range MediaPacketsPerGroup { + e.SetProtectionPercent(1) // one repair covering all three packet masks + p := rtp.Header{Version: 2, SSRC: testMediaSSRC, PayloadType: testMediaPT} + payload := make([]byte, maxEncoderMediaPacketSize-12) + for i := range MaxMediaPacketsPerGroup { p.SequenceNumber++ - out := e.Encode(&p.Header, p.Payload) - if i == MediaPacketsPerGroup-1 { + out := e.Encode(&p, payload) + if i == MaxMediaPacketsPerGroup-1 { require.Len(t, out, 1) - require.NoError(t, out[0].SetExtension(3, []byte{1, 2, 3})) - require.NoError(t, out[0].SetExtension(5, []byte{1, 2})) + out[0].Extension = true + out[0].ExtensionProfile = 0x1000 // RFC 8285 two-byte extensions + require.NoError(t, out[0].SetExtension(20, []byte{1, 2, 3})) + require.NoError(t, out[0].SetExtension(22, []byte{1, 2})) require.LessOrEqual(t, out[0].MarshalSize(), maxMediaPacketSize) } } e.Close() - for range 2 * MediaPacketsPerGroup { + p.Marker = true + for range 10 { p.SequenceNumber++ - require.Empty(t, e.Encode(&p.Header, p.Payload)) + require.Empty(t, e.Encode(&p, payload)) } -} - -func TestEncoderConcurrentClose(t *testing.T) { - e := NewEncoder(testFECPT, testFECSSRC, nil) - e.SetProtectionPercent(20) - var wg sync.WaitGroup - for n := range 4 { - wg.Go(func() { - p := rtp.Header{Version: 2, SSRC: testMediaSSRC, SequenceNumber: uint16(n * 1000)} - for range 100 { - e.Encode(&p, []byte{1, 2, 3}) - p.SequenceNumber++ - } - }) - } - e.Close() - wg.Wait() + require.Nil(t, e.media) } func TestEncoderAccumulationReusesStorage(t *testing.T) { @@ -188,9 +258,12 @@ func TestEncoderAccumulationReusesStorage(t *testing.T) { p := rtp.Header{Version: 2, SSRC: testMediaSSRC} require.NoError(t, p.SetExtension(3, []byte{1, 2, 3})) payload := make([]byte, 1200) + require.Empty(t, e.Encode(&p, payload)) + require.NotNil(t, e.media.storage[0]) + require.Nil(t, e.media.storage[1], "small frames do not allocate large-frame payload storage") allocs := testing.AllocsPerRun(100, func() { e.count = 0 - for range MediaPacketsPerGroup - 1 { + for range MaxMediaPacketsPerGroup - 1 { p.SequenceNumber++ e.Encode(&p, payload) } @@ -198,124 +271,119 @@ func TestEncoderAccumulationReusesStorage(t *testing.T) { require.Zero(t, allocs) } -func BenchmarkEncoder(b *testing.B) { - for _, percent := range []uint32{0, 15, 25, 35} { - b.Run(fmt.Sprintf("percent_%d", percent), func(b *testing.B) { - e := NewEncoder(testFECPT, testFECSSRC, nil) - e.SetProtectionPercent(percent) - p := rtp.Header{Version: 2, SSRC: testMediaSSRC, PayloadType: testMediaPT} - _ = p.SetExtension(3, []byte{1, 2, 3}) - payload := make([]byte, 1200) - // Warm the reusable packet slots and Pion's coverage cache. - for range 2 * MediaPacketsPerGroup { - p.SequenceNumber++ - e.Encode(&p, payload) - } - b.ReportAllocs() - b.SetBytes(int64(MediaPacketsPerGroup * len(payload))) - b.ResetTimer() - for b.Loop() { - for range MediaPacketsPerGroup { - p.SequenceNumber++ - e.Encode(&p, payload) - } - } - }) - } -} - -func TestEncoderProtectionPercent(t *testing.T) { - for percent := uint32(0); percent <= 101; percent++ { - t.Run(fmt.Sprint(percent), func(t *testing.T) { - encoder := NewEncoder(testFECPT, testFECSSRC, nil) - encoder.SetProtectionPercent(percent) - p := rtp.Header{Version: 2, SSRC: testMediaSSRC, PayloadType: testMediaPT} - var count int - var previous uint16 - for range 100 { - p.SequenceNumber++ - repairs := encoder.Encode(&p, []byte{1, 2, 3}) - require.LessOrEqual(t, len(repairs), MediaPacketsPerGroup) - for _, repair := range repairs { - if count > 0 { - require.Equal(t, previous+1, repair.SequenceNumber) - } - count++ - previous = repair.SequenceNumber - } - } - require.EqualValues(t, min(percent, MaxProtectionPercent), count) - }) - } -} - func TestEncoderProtectionPercentChanges(t *testing.T) { - encoder := NewEncoder(testFECPT, testFECSSRC, nil) - encoder.SetProtectionPercent(20) - p := rtp.Header{Version: 2, SSRC: testMediaSSRC, PayloadType: testMediaPT} - send := func(count int) []rtp.Packet { - var result []rtp.Packet - for range count { - p.SequenceNumber++ - result = append(result, encoder.Encode(&p, []byte{1, 2, 3})...) - } - return result - } - first := send(MediaPacketsPerGroup) + e := NewEncoder(testFECPT, testFECSSRC, nil) + e.SetProtectionPercent(20) + p := makeEncoderFrame(t, 100, 1)[0] + first := e.Encode(&p.Header, p.Payload) require.Len(t, first, 1) - require.Empty(t, send(4)) - encoder.SetProtectionPercent(40) - require.Empty(t, send(4), "discard the old partial group") - repairs := send(1) - require.Len(t, repairs, 2) + p.Marker = false + for range 3 { + p.SequenceNumber++ + require.Empty(t, e.Encode(&p.Header, p.Payload)) + } + e.SetProtectionPercent(40) + p.SequenceNumber++ + p.Marker = true + repairs := e.Encode(&p.Header, p.Payload) + require.Len(t, repairs, 1, "discard the old partial frame on a setting change") require.Equal(t, first[0].SequenceNumber+1, repairs[0].SequenceNumber) - encoder.SetProtectionPercent(0) - require.Empty(t, send(100)) - require.Zero(t, encoder.count) - encoder.SetProtectionPercent(10) - require.Empty(t, send(5)) - encoder.SetProtectionPercent(10) // unchanged settings must retain fractional credit - last := send(5) - require.Len(t, last, 1) - require.Equal(t, repairs[1].SequenceNumber+1, last[0].SequenceNumber) + e.SetProtectionPercent(0) + require.Empty(t, e.Encode(&p.Header, p.Payload)) + require.Zero(t, e.OverheadPercent()) + e.SetProtectionPercent(40) + p.Marker = false + for range 4 { + p.SequenceNumber++ + require.Empty(t, e.Encode(&p.Header, p.Payload)) + } + e.SetProtectionPercent(40) + p.SequenceNumber++ + p.Marker = true + last := e.Encode(&p.Header, p.Payload) + require.Len(t, last, 2, "unchanged settings retain the partial frame") + require.Equal(t, repairs[0].SequenceNumber+1, last[0].SequenceNumber) } -func TestEncoderMultipleRepairRecovery(t *testing.T) { - encoder := NewEncoder(testFECPT, testFECSSRC, nil) - encoder.SetProtectionPercent(60) - media := makeMediaPackets(t, 100, MediaPacketsPerGroup) - decoder := newTestDecoder(testFECSSRC, testMediaSSRC, logger.GetLogger()) - var repair []rtp.Packet - for i := range media { - repair = append(repair, encoder.Encode(&media[i].Header, media[i].Payload)...) - if i >= 3 { - decoder.DecodeFEC(&media[i]) - } - } - require.Len(t, repair, 3) - for i := range repair { - recovered := decoder.DecodeFEC(&repair[i]) - require.Len(t, recovered, 1) - requirePacketEqual(t, &media[i], recovered[0]) +func TestEncoderMeasuredOverhead(t *testing.T) { + var e *Encoder + callbacks := 0 + e = NewEncoder(testFECPT, testFECSSRC, func(int, int) { + e.GetState() // callbacks must be outside the encoder lock + callbacks++ + }) + e.SetProtectionPercent(15) + require.EqualValues(t, 15, e.OverheadPercent()) + p := rtp.Header{Version: 2, SSRC: testMediaSSRC, Marker: true} + payload := make([]byte, 100) + repairs := e.Encode(&p, payload) + require.Len(t, repairs, 1) + e.RecordSent(1, len(repairs[0].Payload), repairs[0].MarshalSize()) + require.EqualValues(t, 118, e.OverheadPercent(), "132 repair bytes / 112 media bytes, rounded up") + p.Marker = false + for i := range 20 { + p.SequenceNumber++ + p.Marker = i == 19 + repairs = e.Encode(&p, payload) } + require.Len(t, repairs, 3) + e.overheadUpdatedAt -= int64(overheadUpdateInterval) + e.RecordSent(3, 3*len(repairs[0].Payload), 3*repairs[0].MarshalSize()) + require.EqualValues(t, 19, e.OverheadPercent(), "adapt when frames become larger, including the longer FEC masks") + require.Equal(t, 2, callbacks) + e.SetProtectionPercent(25) + require.EqualValues(t, 25, e.OverheadPercent(), "clear stale measurements on a preset change") } -func TestEncoderConcurrentProtectionChanges(t *testing.T) { - encoder := NewEncoder(testFECPT, testFECSSRC, nil) - encoder.SetProtectionPercent(20) +func TestEncoderConcurrentProtectionChangesAndClose(t *testing.T) { + e := NewEncoder(testFECPT, testFECSSRC, nil) + e.SetProtectionPercent(20) var wg sync.WaitGroup wg.Go(func() { for i := range 1000 { - encoder.SetProtectionPercent(uint32(i % 101)) + e.SetProtectionPercent(uint32(i % 101)) + e.OverheadPercent() } }) wg.Go(func() { - p := rtp.Header{Version: 2, SSRC: testMediaSSRC, PayloadType: testMediaPT} + p := rtp.Header{Version: 2, SSRC: testMediaSSRC, Marker: true} for range 1000 { p.SequenceNumber++ - encoder.Encode(&p, []byte{1, 2, 3}) + repair := e.Encode(&p, []byte{1, 2, 3}) + for _, packet := range repair { + e.RecordSent(1, len(packet.Payload), packet.MarshalSize()) + } } }) + wg.Go(e.Close) wg.Wait() - encoder.Close() +} + +func BenchmarkEncoder(b *testing.B) { + for _, packets := range []int{1, 5, 20, MaxMediaPacketsPerGroup} { + for _, percent := range []uint32{0, 15, 25, 35} { + b.Run(fmt.Sprintf("packets_%d/percent_%d", packets, percent), func(b *testing.B) { + e := NewEncoder(testFECPT, testFECSSRC, nil) + e.SetProtectionPercent(percent) + p := rtp.Header{Version: 2, SSRC: testMediaSSRC, PayloadType: testMediaPT} + _ = p.SetExtension(3, []byte{1, 2, 3}) + payload := make([]byte, 1200) + send := func() { + p.Timestamp += 3000 + for i := range packets { + p.SequenceNumber++ + p.Marker = i == packets-1 + e.Encode(&p, payload) + } + } + send() + b.ReportAllocs() + b.SetBytes(int64(packets * len(payload))) + b.ResetTimer() + for b.Loop() { + send() + } + }) + } + } } diff --git a/pkg/sfu/pacer/base.go b/pkg/sfu/pacer/base.go index f1ea1aae2..13643f56f 100644 --- a/pkg/sfu/pacer/base.go +++ b/pkg/sfu/pacer/base.go @@ -87,7 +87,7 @@ func (b *Base) SendPacket(p *Packet) (int, error) { if p.FEC != nil && written > 0 && !p.IsRTX && !p.IsProbe { repair := p.FEC.Encode(p.Header, p.Payload) - sent, payloadBytes := 0, 0 + sent, payloadBytes, rtpBytes := 0, 0, 0 for i := range repair { packet := &repair[i] fec := PacketFactory.Get().(*Packet) @@ -114,9 +114,10 @@ func (b *Base) SendPacket(p *Packet) (int, error) { if fecErr == nil && n > 0 { sent++ payloadBytes += len(packet.Payload) + rtpBytes += packet.MarshalSize() } } - p.FEC.RecordSent(sent, payloadBytes) + p.FEC.RecordSent(sent, payloadBytes, rtpBytes) } return written, nil } diff --git a/pkg/sfu/pacer/base_fec_test.go b/pkg/sfu/pacer/base_fec_test.go index 17252ec7f..524ec791b 100644 --- a/pkg/sfu/pacer/base_fec_test.go +++ b/pkg/sfu/pacer/base_fec_test.go @@ -72,9 +72,9 @@ func TestPacerFECFinalHeadersAndAccounting(t *testing.T) { encoder := flexfec.NewEncoder(115, 456, func(n int, bytes int) { sent += n; payloadBytes += bytes }) encoder.SetProtectionPercent(20) total := 0 - for i := range flexfec.MediaPacketsPerGroup { + for i := range 5 { p := PacketFactory.Get().(*Packet) - h := &rtp.Header{Version: 2, SSRC: 123, PayloadType: 96, SequenceNumber: uint16(i), Timestamp: uint32(i * 3000)} + h := &rtp.Header{Version: 2, SSRC: 123, PayloadType: 96, SequenceNumber: uint16(i), Timestamp: 3000, Marker: i == 4} _ = h.SetExtension(3, []byte{0, 0, 0}) _ = h.SetExtension(5, []byte{0, 0}) _ = h.SetExtension(7, []byte{0, 0}) @@ -122,7 +122,7 @@ func TestPacerFECSkipsRTXProbesAndFailedWrites(t *testing.T) { e.SetProtectionPercent(20) for i := range 10 { p := PacketFactory.Get().(*Packet) - *p = Packet{Header: &rtp.Header{Version: 2, SSRC: 123, SequenceNumber: uint16(i)}, HeaderSize: 12, Payload: []byte{1}, FEC: e, WriteStream: w, IsRTX: mode == "rtx", IsProbe: mode == "probe"} + *p = Packet{Header: &rtp.Header{Version: 2, SSRC: 123, SequenceNumber: uint16(i), Marker: true}, HeaderSize: 12, Payload: []byte{1}, FEC: e, WriteStream: w, IsRTX: mode == "rtx", IsProbe: mode == "probe"} n, err := b.SendPacket(p) if w.failMedia { require.ErrorIs(t, err, io.ErrClosedPipe)