diff --git a/pkg/rtc/subscribedtrack.go b/pkg/rtc/subscribedtrack.go index 54fa405f1..ea6796c13 100644 --- a/pkg/rtc/subscribedtrack.go +++ b/pkg/rtc/subscribedtrack.go @@ -137,8 +137,6 @@ func NewSubscribedTrack(params SubscribedTrackParams) (*SubscribedTrack, error) // Strip packet trailer if track has packet trailer but subscriber does not have cap stripPacketTrailer := params.MediaTrack.HasPacketTrailer() && !subSupportsPacketTrailer downTrack, err := sfu.NewDownTrack(sfu.DownTrackParams{ - EnableFlexFEC: params.SubscriberConfig.FlexFEC.Enabled, - OnFECSent: prometheus.RecordFECDownstream, Codecs: codecs, IsEncrypted: isEncrypted, Source: params.MediaTrack.Source(), @@ -160,6 +158,8 @@ func NewSubscribedTrack(params SubscribedTrackParams) (*SubscribedTrack, error) DisableSenderReportPassThrough: params.Subscriber.GetDisableSenderReportPassThrough(), SupportsCodecChange: params.Subscriber.SupportsCodecChange(), EnableStartAtDesiredQuality: params.EnableStartAtDesiredQuality, + EnableFlexFEC: params.SubscriberConfig.FlexFEC.Enabled, + OnFECSent: prometheus.RecordFECDownstream, Listener: s, }) if err != nil { @@ -214,6 +214,7 @@ func (t *SubscribedTrack) Bound(err error) { if t.settings != nil { if t.params.AdaptiveStream { // remove `disabled` flag to force a visibility update + t.settings = utils.CloneProto(t.settings) t.settings.Disabled = false t.logger.Debugw("enabling subscriber track settings on bind", "settings", logger.Proto(t.settings)) } @@ -352,6 +353,7 @@ func (t *SubscribedTrack) applySettings() { t.settingsVersion = t.versionGenerator.Next() settingsVersion := t.settingsVersion + settings := t.settings t.settingsLock.Unlock() dt := t.DownTrack() @@ -359,30 +361,30 @@ func (t *SubscribedTrack) applySettings() { temporal := buffer.InvalidLayerTemporal if dt.Kind() == webrtc.RTPCodecTypeVideo { mt := t.MediaTrack() - quality := t.settings.Quality + quality := settings.Quality mimeType := dt.Mime() - if t.settings.Width > 0 { - quality = mt.GetQualityForDimension(mimeType, t.settings.Width, t.settings.Height) + if settings.Width > 0 { + quality = mt.GetQualityForDimension(mimeType, settings.Width, settings.Height) } spatial = buffer.GetSpatialLayerForVideoQuality(mimeType, quality, mt.ToProto()) - if t.settings.Fps > 0 { - temporal = mt.GetTemporalLayerForSpatialFps(mimeType, spatial, t.settings.Fps) + if settings.Fps > 0 { + temporal = mt.GetTemporalLayerForSpatialFps(mimeType, spatial, settings.Fps) } } t.settingsLock.Lock() - if settingsVersion != t.settingsVersion { + if settingsVersion != t.settingsVersion || settings != t.settings { // a newer settings has superseded this one t.settingsLock.Unlock() return } - t.logger.Debugw("applying subscriber track settings", "settings", logger.Proto(t.settings)) - if t.settings.Fec != nil { - dt.SetFECProtection(*t.settings.Fec) + t.logger.Debugw("applying subscriber track settings", "settings", logger.Proto(settings)) + if settings.Fec != nil { + dt.SetFECProtection(*settings.Fec) } - if t.settings.Disabled { + if settings.Disabled { dt.Mute(true) t.settingsLock.Unlock() return diff --git a/pkg/rtc/subscribedtrack_fec_test.go b/pkg/rtc/subscribedtrack_fec_test.go index 38b3c78cc..3ad390518 100644 --- a/pkg/rtc/subscribedtrack_fec_test.go +++ b/pkg/rtc/subscribedtrack_fec_test.go @@ -15,17 +15,21 @@ package rtc import ( + "sync" "testing" + "github.com/pion/webrtc/v4" + "github.com/stretchr/testify/require" + + "github.com/livekit/protocol/codecs/mime" + "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/logger" + "github.com/livekit/protocol/utils" + "github.com/livekit/livekit-server/pkg/rtc/types/typesfakes" "github.com/livekit/livekit-server/pkg/sfu" "github.com/livekit/livekit-server/pkg/sfu/buffer" "github.com/livekit/livekit-server/pkg/sfu/sfufakes" - "github.com/livekit/protocol/livekit" - "github.com/livekit/protocol/logger" - "github.com/livekit/protocol/utils" - "github.com/pion/webrtc/v4" - "github.com/stretchr/testify/require" ) func TestSubscriberFECSettings(t *testing.T) { @@ -80,6 +84,71 @@ func TestSubscriberFECPersistsAcrossResubscription(t *testing.T) { } func TestSubscriberFECAppliesToDownTrack(t *testing.T) { + sub, _ := newSubscriberFECTestTrack(t) + dt := sub.DownTrack() + for _, tc := range []struct { + level *livekit.FECProtection + want int64 + }{ + {nil, 100_000}, + {livekit.FECProtection_FEC_MEDIUM.Enum(), 125_000}, + {nil, 125_000}, + {livekit.FECProtection_FEC_NONE.Enum(), 100_000}, + {nil, 100_000}, + {livekit.FECProtection_FEC_LOW.Enum(), 115_000}, + {livekit.FECProtection_FEC_HIGH.Enum(), 135_000}, + {livekit.FECProtection(99).Enum(), 100_000}, + } { + sub.UpdateSubscriberSettings(&livekit.UpdateTrackSettings{Quality: livekit.VideoQuality_HIGH, Fec: tc.level}, true) + dt.AllocateOptimal(false, false) + require.Equal(t, tc.want, dt.BandwidthRequested(), "client settings must reach the negotiated encoder and allocator") + } +} + +func TestSubscriberFECSettingsSupersededDuringApply(t *testing.T) { + sub, mediaTrack := newSubscriberFECTestTrack(t) + sub.UpdateSubscriberSettings(&livekit.UpdateTrackSettings{Quality: livekit.VideoQuality_HIGH}, true) + var pending func() + sub.debouncer = func(f func()) { pending = f } + mediaTrack.GetQualityForDimensionCalls(func(mime.MimeType, uint32, uint32) livekit.VideoQuality { + // A settings update can arrive while layer selection runs without settingsLock. + sub.UpdateSubscriberSettings(&livekit.UpdateTrackSettings{ + Quality: livekit.VideoQuality_HIGH, + Fec: livekit.FECProtection_FEC_HIGH.Enum(), + }, false) + return livekit.VideoQuality_HIGH + }) + sub.UpdateSubscriberSettings(&livekit.UpdateTrackSettings{ + Width: 640, + Fec: livekit.FECProtection_FEC_MEDIUM.Enum(), + }, true) + dt := sub.DownTrack() + dt.AllocateOptimal(false, false) + require.EqualValues(t, 100_000, dt.BandwidthRequested(), "superseded settings must not apply part of a newer update") + require.NotNil(t, pending) + pending() + dt.AllocateOptimal(false, false) + require.EqualValues(t, 135_000, dt.BandwidthRequested()) +} + +func TestSubscriberFECConcurrentSettings(t *testing.T) { + sub, _ := newSubscriberFECTestTrack(t) + var wg sync.WaitGroup + for level := range 4 { + wg.Go(func() { + for range 100 { + sub.UpdateSubscriberSettings(&livekit.UpdateTrackSettings{ + Width: 640, + Fec: livekit.FECProtection(level).Enum(), + }, true) + } + }) + } + wg.Wait() +} + +func newSubscriberFECTestTrack(t *testing.T) (*SubscribedTrack, *typesfakes.FakeMediaTrack) { + t.Helper() codec := webrtc.RTPCodecParameters{ RTPCodecCapability: webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeVP8, ClockRate: 90000}, PayloadType: 96, @@ -115,23 +184,7 @@ func TestSubscriberFECAppliesToDownTrack(t *testing.T) { logger: logger.GetLogger(), versionGenerator: utils.NewDefaultTimedVersionGenerator(), } - for _, tc := range []struct { - level *livekit.FECProtection - want int64 - }{ - {nil, 100_000}, - {livekit.FECProtection_FEC_MEDIUM.Enum(), 125_000}, - {nil, 125_000}, - {livekit.FECProtection_FEC_NONE.Enum(), 100_000}, - {nil, 100_000}, - {livekit.FECProtection_FEC_LOW.Enum(), 115_000}, - {livekit.FECProtection_FEC_HIGH.Enum(), 135_000}, - {livekit.FECProtection(99).Enum(), 100_000}, - } { - sub.UpdateSubscriberSettings(&livekit.UpdateTrackSettings{Quality: livekit.VideoQuality_HIGH, Fec: tc.level}, true) - dt.AllocateOptimal(false, false) - require.Equal(t, tc.want, dt.BandwidthRequested(), "client settings must reach the negotiated encoder and allocator") - } + return sub, mediaTrack } type subscriberFECTrackContext struct { diff --git a/pkg/rtc/transport.go b/pkg/rtc/transport.go index 1659da944..60eac4c91 100644 --- a/pkg/rtc/transport.go +++ b/pkg/rtc/transport.go @@ -1022,6 +1022,12 @@ func (t *PCTransport) queueOrConfigureSender( enableAudioStereo bool, enableAudioNACK bool, ) { + keepFlexFEC := t.params.Config.Subscriber.FlexFEC.Enabled + if transceiver.Direction() == webrtc.RTPTransceiverDirectionSendrecv && t.params.DirectionConfig.FlexFEC.Enabled { + // A shared codec list must retain upstream repair support even when + // this subscription does not permit downstream FEC generation. + keepFlexFEC = true + } params := configureSenderParams{ transceiver: transceiver, enabledCodecs: enabledCodecs, @@ -1029,7 +1035,7 @@ func (t *PCTransport) queueOrConfigureSender( filterOutH264HighProfile: !t.params.IsOfferer, enableAudioStereo: enableAudioStereo, enableAudioNACK: enableAudioNACK, - keepFlexFEC: t.params.Config.Subscriber.FlexFEC.Enabled, + keepFlexFEC: keepFlexFEC, } if !t.params.IsOfferer { t.sendersPendingConfigMu.Lock() diff --git a/pkg/rtc/transport_fec_test.go b/pkg/rtc/transport_fec_test.go index a9f530657..c79ef515d 100644 --- a/pkg/rtc/transport_fec_test.go +++ b/pkg/rtc/transport_fec_test.go @@ -255,3 +255,24 @@ func TestSinglePCDownstreamFlexFEC(t *testing.T) { }) } } + +func TestSinglePCUpstreamFlexFECWithSubscription(t *testing.T) { + codecs := []*livekit.Codec{{Mime: "video/VP8"}, {Mime: "video/rtx"}} + conf := &WebRTCConfig{Publisher: DirectionConfig{FlexFEC: FlexFECDirectionConfig{Enabled: true}}} + server, err := NewPCTransport(TransportParams{ + Config: conf, DirectionConfig: conf.Publisher, IsSendSide: true, + Handler: &transportfakes.FakeHandler{}, + EnabledPublishCodecs: codecs, EnabledSubscribeCodecs: codecs, + }) + require.NoError(t, err) + defer server.Close() + tr, err := server.pc.AddTransceiverFromKind(webrtc.RTPCodecTypeVideo, webrtc.RTPTransceiverInit{Direction: webrtc.RTPTransceiverDirectionSendrecv}) + require.NoError(t, err) + server.queueOrConfigureSender(tr, codecs, RTCPFeedbackConfig{}, false, true) + require.Len(t, server.sendersPendingConfig, 1) + configureSender(server.sendersPendingConfig[0], nil) + server.restrictReceiverCodecsToPublishList() + offer, err := server.pc.CreateOffer(nil) + require.NoError(t, err) + require.Contains(t, offer.SDP, "flexfec-03/90000", "subscribing on a shared m-section must retain upstream FEC when downstream FEC is disabled") +} diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index a90fb2397..296d2b910 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -184,6 +184,7 @@ type DownTrackState struct { DeltaStatsRTXSenderSnapshotId uint32 ForwarderState *livekit.RTPForwarderState PlayoutDelayControllerState PlayoutDelayControllerState + FECState flexfec.EncoderState } func (d DownTrackState) MarshalLogObject(e zapcore.ObjectEncoder) error { @@ -193,6 +194,7 @@ func (d DownTrackState) MarshalLogObject(e zapcore.ObjectEncoder) error { e.AddUint32("DeltaStatsRTXSenderSnapshotId", d.DeltaStatsRTXSenderSnapshotId) e.AddObject("ForwarderState", logger.Proto(d.ForwarderState)) e.AddObject("PlayoutDelayControllerState", d.PlayoutDelayControllerState) + e.AddObject("FECState", d.FECState) return nil } @@ -227,8 +229,6 @@ var _ TrackSender = (*DownTrack)(nil) type ReceiverReportListener func(dt *DownTrack, report *rtcp.ReceiverReport) type DownTrackParams struct { - EnableFlexFEC bool - OnFECSent func(packets int, bytes int) Codecs []webrtc.RTPCodecParameters IsEncrypted bool Source livekit.TrackSource @@ -246,6 +246,8 @@ type DownTrackParams struct { SupportsCodecChange bool StripPacketTrailer bool EnableStartAtDesiredQuality bool + EnableFlexFEC bool + OnFECSent func(packets int, bytes int) Listener DownTrackListener } @@ -258,18 +260,20 @@ type DownTrackParams struct { // - closed // once closed, a DownTrack cannot be re-used. type DownTrack struct { + params DownTrackParams + id livekit.TrackID + kind webrtc.RTPCodecType + ssrc uint32 + ssrcRTX uint32 + payloadType atomic.Uint32 + payloadTypeRTX atomic.Uint32 + sequencer *sequencer + rtxSequenceNumber atomic.Uint64 + + fecLock sync.Mutex fecEncoder atomic.Pointer[flexfec.Encoder] fecProtectionPercent atomic.Uint32 - fecLock sync.Mutex - params DownTrackParams - id livekit.TrackID - kind webrtc.RTPCodecType - ssrc uint32 - ssrcRTX uint32 - payloadType atomic.Uint32 - payloadTypeRTX atomic.Uint32 - sequencer *sequencer - rtxSequenceNumber atomic.Uint64 + fecState flexfec.EncoderState receiverLock sync.RWMutex receiver TrackReceiver @@ -1459,6 +1463,7 @@ func (d *DownTrack) CloseWithFlush(flush bool, isEnding bool) { d.setBindStateLocked(bindStateUnbound) d.Receiver().DeleteDownTrack(d.SubscriberID()) + d.closeFEC() if d.rtcpReader != nil && isEnding { d.params.Logger.Debugw("downtrack close rtcp reader") @@ -1476,7 +1481,6 @@ func (d *DownTrack) CloseWithFlush(flush bool, isEnding bool) { d.rtpStats.Stop() d.rtpStatsRTX.Stop() - d.closeFEC() d.params.Logger.Debugw( "rtp stats", "direction", "downstream", @@ -1540,6 +1544,7 @@ func (d *DownTrack) GetState() DownTrackState { RTPStatsRTX: d.rtpStatsRTX, DeltaStatsRTXSenderSnapshotId: d.deltaStatsRTXSenderSnapshotId, ForwarderState: d.forwarder.GetState(), + FECState: d.getFECState(), } if d.playoutDelay != nil { @@ -1570,6 +1575,7 @@ func (d *DownTrack) SeedState(state DownTrackState) { d.rtxSequenceNumber.Store(d.rtpStatsRTX.ExtHighestSequenceNumber()) } d.forwarder.SeedState(state.ForwarderState) + d.seedFECState(state.FECState) } func (d *DownTrack) StopWriteAndGetState() DownTrackState { @@ -1577,6 +1583,7 @@ func (d *DownTrack) StopWriteAndGetState() DownTrackState { d.bindLock.Lock() d.writable.Store(false) d.writeStopped.Store(true) + d.closeFEC() d.bindLock.Unlock() return d.GetState() diff --git a/pkg/sfu/downtrack_fec.go b/pkg/sfu/downtrack_fec.go index 9d71ae65c..cb77aea67 100644 --- a/pkg/sfu/downtrack_fec.go +++ b/pkg/sfu/downtrack_fec.go @@ -17,24 +17,26 @@ package sfu import ( "strings" - "github.com/livekit/protocol/livekit" "github.com/pion/webrtc/v4" + "github.com/livekit/protocol/livekit" + "github.com/livekit/livekit-server/pkg/sfu/flexfec" ) // bindFEC runs under bindLock, using the complete negotiated codec list saved // before Pion narrows the TrackLocalContext to the selected media codec. func (d *DownTrack) bindFEC(t webrtc.TrackLocalContext) { - d.closeFEC() d.fecLock.Lock() defer d.fecLock.Unlock() + d.closeFECLocked() if !d.params.EnableFlexFEC || d.kind != webrtc.RTPCodecTypeVideo || t.SSRCForwardErrorCorrection() == 0 { return } 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.SeedState(d.fecState) encoder.SetProtectionPercent(d.fecProtectionPercent.Load()) d.fecEncoder.Store(encoder) return @@ -45,8 +47,35 @@ func (d *DownTrack) bindFEC(t webrtc.TrackLocalContext) { func (d *DownTrack) closeFEC() { d.fecLock.Lock() defer d.fecLock.Unlock() + d.closeFECLocked() +} + +func (d *DownTrack) closeFECLocked() { if encoder := d.fecEncoder.Swap(nil); encoder != nil { encoder.Close() + // Close first so queued media cannot advance the sequence after it is saved. + d.fecState = encoder.GetState() + } +} + +func (d *DownTrack) getFECState() flexfec.EncoderState { + d.fecLock.Lock() + defer d.fecLock.Unlock() + if encoder := d.fecEncoder.Load(); encoder != nil { + return encoder.GetState() + } + return d.fecState +} + +func (d *DownTrack) seedFECState(state flexfec.EncoderState) { + if state.SSRC == 0 { + return + } + d.fecLock.Lock() + defer d.fecLock.Unlock() + d.fecState = state + if encoder := d.fecEncoder.Load(); encoder != nil { + encoder.SeedState(state) } } diff --git a/pkg/sfu/downtrack_fec_integration_test.go b/pkg/sfu/downtrack_fec_integration_test.go index e8b550a6e..95abba6e1 100644 --- a/pkg/sfu/downtrack_fec_integration_test.go +++ b/pkg/sfu/downtrack_fec_integration_test.go @@ -247,6 +247,50 @@ func TestDownTrackFlexFECOnWire(t *testing.T) { } } } + + // Resume a subscription on its cached transceiver, including the + // repair sequence. The receiver's existing SRTP replay window must + // accept every packet without another negotiation. + dt.CloseWithFlush(false, false) + state := dt.GetState() + require.Equal(t, sequence+1, state.FECState.NextSequenceNumber) + next, err := sfu.NewDownTrack(sfu.DownTrackParams{ + Codecs: []webrtc.RTPCodecParameters{vp8CodecParams}, + Receiver: newFakeTrackReceiver(vp8CodecParams), + BufferFactory: factory, + Pacer: p, + Logger: logger.GetLogger(), + Listener: &sfufakes.FakeDownTrackListener{}, + StreamID: "fec-stream", + SubID: "fec-sub", + MaxTrack: 500, + EnableFlexFEC: true, + }) + require.NoError(t, err) + next.OnBinding(func(err error) { + if err == nil { + next.SeedState(state) + } + }) + next.SetTransceiver(tr) + next.SetFECProtection(livekit.FECProtection_FEC_MEDIUM) + require.NoError(t, tr.Sender().ReplaceTrack(next)) + dt = next + dt.SetConnected() + require.Eventually(t, dt.IsWritableForTest, 5*time.Second, 10*time.Millisecond) + 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) + repairs := 0 + for _, packet := range capture.all()[before:] { + if packet.SSRC == fecSSRC { + require.Equal(t, sequence+1, packet.SequenceNumber) + sequence = packet.SequenceNumber + repairs++ + } + } + require.Equal(t, 5, repairs) } }) } diff --git a/pkg/sfu/downtrack_fec_test.go b/pkg/sfu/downtrack_fec_test.go index 651c9fb01..1fe62ad0b 100644 --- a/pkg/sfu/downtrack_fec_test.go +++ b/pkg/sfu/downtrack_fec_test.go @@ -17,11 +17,13 @@ package sfu import ( "testing" - "github.com/livekit/livekit-server/pkg/sfu/flexfec" - "github.com/livekit/protocol/livekit" "github.com/pion/rtp" "github.com/pion/webrtc/v4" "github.com/stretchr/testify/require" + + "github.com/livekit/protocol/livekit" + + "github.com/livekit/livekit-server/pkg/sfu/flexfec" ) type fecTrackContext struct { @@ -54,6 +56,7 @@ func TestDownTrackFECNegotiationAndLifecycle(t *testing.T) { old := d.fecEncoder.Load() require.NotNil(t, old) p := rtp.Header{Version: 2, SSRC: 123, PayloadType: 96} + var lastSequenceNumber uint16 for i := range flexfec.MediaPacketsPerGroup { p.SequenceNumber++ repair := old.Encode(&p, []byte{1, 2, 3}) @@ -61,6 +64,7 @@ func TestDownTrackFECNegotiationAndLifecycle(t *testing.T) { require.Len(t, repair, 1) require.EqualValues(t, 456, repair[0].SSRC) require.EqualValues(t, 118, repair[0].PayloadType, "use negotiated PT") + lastSequenceNumber = repair[0].SequenceNumber } } d.bindFEC(c) @@ -68,6 +72,9 @@ func TestDownTrackFECNegotiationAndLifecycle(t *testing.T) { for range flexfec.MediaPacketsPerGroup { p.SequenceNumber++ 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") + } } d.closeFEC() require.Nil(t, d.fecEncoder.Load()) diff --git a/pkg/sfu/flexfec/encoder.go b/pkg/sfu/flexfec/encoder.go index 60de5c631..293dbd5a2 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/zap/zapcore" "github.com/livekit/protocol/utils/mono" ) @@ -37,6 +38,24 @@ const ( maxEncoderMediaPacketSize = maxMediaPacketSize - pionflexfec.BaseFec03HeaderSize - 16 ) +// EncoderState preserves the repair sequence when a sender SSRC is reused. +// Partial media groups are never carried over to a new binding. +type EncoderState struct { + SSRC uint32 + NextSequenceNumber uint16 +} + +func (s EncoderState) MarshalLogObject(e zapcore.ObjectEncoder) error { + e.AddUint32("SSRC", s.SSRC) + e.AddUint16("NextSequenceNumber", s.NextSequenceNumber) + return nil +} + +type encoderMediaGroup struct { + packets [MediaPacketsPerGroup]rtp.Packet + storage [MediaPacketsPerGroup][maxEncoderMediaPacketSize]byte +} + // Encoder batches final, outgoing video packets for Pion's FlexFEC-03 encoder. // It owns reusable wire storage: the caller may recycle headers, extensions and // payloads as soon as Encode returns. It does not delay media or start workers. @@ -44,11 +63,11 @@ const ( type Encoder struct { mu sync.Mutex encoder *pionflexfec.FlexEncoder03 - packets [MediaPacketsPerGroup]rtp.Packet - storage [MediaPacketsPerGroup][maxEncoderMediaPacketSize]byte + media *encoderMediaGroup count int startedAt int64 sequenceNumber uint16 + ssrc uint32 closed bool onSent func(packets int, bytes int) protectionPercent uint32 @@ -59,11 +78,31 @@ func NewEncoder(payloadType uint8, ssrc uint32, onSent func(packets int, bytes i return &Encoder{ encoder: pionflexfec.NewFlexEncoder03(payloadType, ssrc), sequenceNumber: uint16(rand.Uint32()), + ssrc: ssrc, onSent: onSent, protectionPercent: DefaultProtectionPercent, } } +func (e *Encoder) GetState() EncoderState { + e.mu.Lock() + defer e.mu.Unlock() + return EncoderState{SSRC: e.ssrc, NextSequenceNumber: e.sequenceNumber} +} + +// SeedState must be called before forwarding starts. Only reuse sequencing for +// the same repair SSRC; a new SSRC keeps its randomized initial sequence number. +func (e *Encoder) SeedState(state EncoderState) { + e.mu.Lock() + defer e.mu.Unlock() + if e.closed || state.SSRC == 0 || state.SSRC != e.ssrc { + return + } + 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. @@ -81,8 +120,13 @@ func (e *Encoder) Encode(header *rtp.Header, payload []byte) []rtp.Packet { e.count = 0 return nil } + if e.media == nil { + // Like the upstream recovery buffer, allocate packet storage only once + // it is needed. Negotiated tracks with the default "none" stay small. + e.media = &encoderMediaGroup{} + } if e.count != 0 { - previous := &e.packets[e.count-1] + previous := &e.media.packets[e.count-1] if header.SequenceNumber != previous.SequenceNumber+1 || header.SSRC != previous.SSRC || header.PayloadType != previous.PayloadType || now-e.startedAt > int64(maxEncoderGroupAge) { @@ -93,14 +137,14 @@ func (e *Encoder) Encode(header *rtp.Header, payload []byte) []rtp.Packet { e.startedAt = now } - raw := e.storage[e.count][:size] + raw := e.media.storage[e.count][:size] n, err := header.MarshalTo(raw) if err != nil { e.count = 0 return nil } copy(raw[n:], payload) - if err = e.packets[e.count].Unmarshal(raw); err != nil { + if err = e.media.packets[e.count].Unmarshal(raw); err != nil { e.count = 0 return nil } @@ -119,9 +163,9 @@ func (e *Encoder) Encode(header *rtp.Header, payload []byte) []rtp.Packet { if numRepair == 0 { return nil } - repair := e.encoder.EncodeFec(e.packets[:], numRepair) + repair := e.encoder.EncodeFec(e.media.packets[:], numRepair) for i := range repair { - // Pion's encoder uses a constant timestamp and initial sequence number. + // 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. repair[i].Timestamp = header.Timestamp repair[i].SequenceNumber = e.sequenceNumber @@ -157,5 +201,7 @@ func (e *Encoder) Close() { e.mu.Lock() e.closed = true e.count = 0 + e.media = nil + e.encoder = nil e.mu.Unlock() } diff --git a/pkg/sfu/flexfec/encoder_test.go b/pkg/sfu/flexfec/encoder_test.go index 20188cefb..f4e1d4688 100644 --- a/pkg/sfu/flexfec/encoder_test.go +++ b/pkg/sfu/flexfec/encoder_test.go @@ -19,9 +19,10 @@ import ( "sync" "testing" - "github.com/livekit/protocol/logger" "github.com/pion/rtp" "github.com/stretchr/testify/require" + + "github.com/livekit/protocol/logger" ) func TestEncoderDefaultsToNoProtection(t *testing.T) { @@ -32,6 +33,32 @@ func TestEncoderDefaultsToNoProtection(t *testing.T) { require.Empty(t, encoder.Encode(&p, []byte{1, 2, 3})) } 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) { + e := NewEncoder(testFECPT, testFECSSRC, nil) + 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) + } + } + e.Close() + state := e.GetState() + require.Equal(t, sn+1, state.NextSequenceNumber) + e = NewEncoder(testFECPT, testFECSSRC, nil) + e.SeedState(state) + e.SetProtectionPercent(20) + } + state := e.GetState() + e.SeedState(EncoderState{SSRC: testFECSSRC + 1, NextSequenceNumber: state.NextSequenceNumber + 100}) + require.Equal(t, state, e.GetState(), "do not seed a different repair SSRC") } func TestEncoderRecoveryWithReusedMemory(t *testing.T) { diff --git a/pkg/sfu/pacer/pacer.go b/pkg/sfu/pacer/pacer.go index bf02ce988..88a2edac5 100644 --- a/pkg/sfu/pacer/pacer.go +++ b/pkg/sfu/pacer/pacer.go @@ -43,8 +43,6 @@ const ( ) type Packet struct { - // FEC is set only on original video media, never padding, RTX or probes. - FEC *flexfec.Encoder Header *rtp.Header HeaderPool *sync.Pool HeaderSize int @@ -57,6 +55,8 @@ type Packet struct { WriteStream webrtc.TrackLocalWriter Pool *sync.Pool PoolEntity *[]byte + // FEC is set only on original video media, never padding, RTX or probes. + FEC *flexfec.Encoder } type Pacer interface { diff --git a/pkg/sfu/rtpstats/rtpstats_sender.go b/pkg/sfu/rtpstats/rtpstats_sender.go index eafa7bc7b..09f37c6e7 100644 --- a/pkg/sfu/rtpstats/rtpstats_sender.go +++ b/pkg/sfu/rtpstats/rtpstats_sender.go @@ -394,8 +394,15 @@ func NewRTPStatsSender(params RTPStatsParams, cacheSize int) *RTPStatsSender { } func (r *RTPStatsSender) Seed(from *RTPStatsSender) { + if from == nil || from == r { + return + } r.lock.Lock() defer r.lock.Unlock() + // The previous sender's final stats callback can still update snapshots + // after Stop. Serialize the state copy with those updates. + from.lock.RLock() + defer from.lock.RUnlock() if !r.seed(from.rtpStatsBase) { return