diff --git a/config-sample.yaml b/config-sample.yaml index 0559d7712..db567553f 100644 --- a/config-sample.yaml +++ b/config-sample.yaml @@ -115,6 +115,24 @@ rtc: # packet_buffer_size_video: 500 # # number of packets to buffer in the SFU for audio, defaults to 200 # packet_buffer_size_audio: 200 + # # FlexFEC-03 (RFC 8627) video forward error correction. Disabled by default. + # # FlexFEC is terminated at the SFU: it cannot be passed through unchanged because the + # # SFU rewrites SSRC/sequence numbers per subscriber. The two directions are independent. + # flexfec: + # # generate a fresh FlexFEC repair stream per subscriber (downstream loss recovery) + # subscriber: false + # # negotiate, receive and decode a publisher's FlexFEC repair stream (uplink loss recovery) + # publisher: false + # # dynamic payload type used for the flexfec-03 codec, defaults to 49 + # payload_type: 49 + # # subscriber-side generation is a fixed, block-based scheme: for every + # # num_media_packets media packets, num_fec_packets repair packets are emitted. + # # bandwidth overhead ~= num_fec_packets/num_media_packets, and a block can recover + # # at most num_fec_packets losses out of num_media_packets. smaller num_media_packets + # # recovers faster (lower added latency) but costs more overhead. defaults below give + # # ~20% overhead; the pion library default (no tuning) is a much heavier 5/2 = 40%. + # num_media_packets: 10 + # num_fec_packets: 2 # # minimum amount of time between pli/fir rtcp packets being sent to an individual # # producer. Increasing these times can lead to longer black screens when new participants join, # # while reducing them can lead to higher stream bitrate. diff --git a/pkg/config/config.go b/pkg/config/config.go index f155eb59d..439dea728 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -138,6 +138,67 @@ type RTCConfig struct { // enable rtp stream restart detection for published tracks EnableRTPStreamRestartDetection bool `yaml:"enable_rtp_stream_restart_detection,omitempty"` + + // FlexFEC (RFC 8627 / flexfec-03) support + FlexFEC FlexFECConfig `yaml:"flexfec,omitempty"` +} + +// FlexFECConfig controls FlexFEC-03 (video) forward error correction. +// +// FlexFEC cannot be passed through an SFU as-is because the SFU rewrites SSRC and +// sequence numbers per subscriber. Instead, FlexFEC is terminated at the SFU and +// the two directions are handled independently: +// - Subscriber: the SFU generates a fresh FlexFEC repair stream per downstream so a +// subscriber can recover packets lost on the SFU -> subscriber path. +// - Publisher: the SFU negotiates, receives and decodes a publisher's FlexFEC repair +// stream to recover packets lost on the publisher -> SFU path. +// +// Both are disabled by default. FlexFEC is video-only here; audio keeps RED / in-band +// Opus FEC. +type FlexFECConfig struct { + // generate FlexFEC for subscribers (downstream loss recovery) + Subscriber bool `yaml:"subscriber,omitempty"` + // negotiate, receive and decode FlexFEC from publishers (uplink loss recovery) + Publisher bool `yaml:"publisher,omitempty"` + // dynamic payload type used for the flexfec-03 codec + PayloadType int `yaml:"payload_type,omitempty"` + // NumMediaPackets is the number of media packets per FEC protection block for + // subscriber-side generation. The repair packets for a block are only emitted + // after this many media packets, so smaller blocks recover faster (lower added + // latency) but cost proportionally more overhead. Defaults to 10. + NumMediaPackets uint32 `yaml:"num_media_packets,omitempty"` + // NumFecPackets is the number of repair packets generated per block. Bandwidth + // overhead is approximately NumFecPackets/NumMediaPackets, and a block can recover + // at most NumFecPackets lost packets out of its NumMediaPackets. Note this is a + // fixed (non-adaptive), block-based scheme: it protects scattered loss well but a + // burst exceeding NumFecPackets within one block is unrecoverable by FEC. + // Defaults to 2 (i.e. ~20% overhead with the default NumMediaPackets). + NumFecPackets uint32 `yaml:"num_fec_packets,omitempty"` +} + +func (f FlexFECConfig) Enabled() bool { + return (f.Subscriber || f.Publisher) && f.PayloadType != 0 +} + +// EncoderParams returns the (numMediaPackets, numFecPackets) used to configure the +// subscriber-side FlexFEC encoder, applying sane defaults and clamping so the values +// are always usable (1 <= numFec < numMedia, numMedia >= 2). +func (f FlexFECConfig) EncoderParams() (numMediaPackets uint32, numFecPackets uint32) { + numMediaPackets = f.NumMediaPackets + numFecPackets = f.NumFecPackets + if numMediaPackets == 0 { + numMediaPackets = 10 + } + if numFecPackets == 0 { + numFecPackets = 2 + } + if numMediaPackets < 2 { + numMediaPackets = 2 + } + if numFecPackets >= numMediaPackets { + numFecPackets = numMediaPackets - 1 + } + return numMediaPackets, numFecPackets } type TURNServer struct { @@ -390,6 +451,13 @@ var DefaultConfig = Config{ SendSideBWEPacer: string(pacer.PacerBehaviorNoQueue), SendSideBWE: sendsidebwe.DefaultSendSideBWEConfig, }, + FlexFEC: FlexFECConfig{ + Subscriber: false, + Publisher: false, + PayloadType: 49, + NumMediaPackets: 10, + NumFecPackets: 2, + }, }, Audio: sfu.DefaultAudioConfig, Video: VideoConfig{ diff --git a/pkg/rtc/config.go b/pkg/rtc/config.go index 2f30060fd..db25634a8 100644 --- a/pkg/rtc/config.go +++ b/pkg/rtc/config.go @@ -37,6 +37,7 @@ type WebRTCConfig struct { Receiver ReceiverConfig Publisher DirectionConfig Subscriber DirectionConfig + FlexFEC config.FlexFECConfig } type ReceiverConfig struct { @@ -88,6 +89,7 @@ func NewWebRTCConfig(conf *config.Config) (*WebRTCConfig, error) { }, Publisher: getPublisherConfig(false), Subscriber: getSubscriberConfig(rtcConf.CongestionControl.UseSendSideBWEInterceptor || rtcConf.CongestionControl.UseSendSideBWE), + FlexFEC: rtcConf.FlexFEC, }, nil } diff --git a/pkg/rtc/mediaengine.go b/pkg/rtc/mediaengine.go index cfc53eaaf..824ed2e55 100644 --- a/pkg/rtc/mediaengine.go +++ b/pkg/rtc/mediaengine.go @@ -86,6 +86,28 @@ func registerCodecs(me *webrtc.MediaEngine, codecs []*livekit.Codec, rtcpFeedbac return nil } +// flexFEC03CodecParameters returns the flexfec-03 (RFC 8627 draft-03) video codec +// parameters for the given dynamic payload type. The repair-window fmtp matches the +// value pion's webrtc.ConfigureFlexFEC03 advertises so publisher and subscriber sides +// negotiate consistently. +func flexFEC03CodecParameters(payloadType webrtc.PayloadType) webrtc.RTPCodecParameters { + return webrtc.RTPCodecParameters{ + RTPCodecCapability: webrtc.RTPCodecCapability{ + MimeType: webrtc.MimeTypeFlexFEC03, + ClockRate: 90000, + SDPFmtpLine: "repair-window=10000000", + }, + PayloadType: payloadType, + } +} + +// registerFlexFECCodec registers the flexfec-03 codec on the media engine so that the +// remote peer can negotiate sending a FlexFEC repair stream to the SFU. This is the +// receive (publisher) side - it does NOT add the encoder interceptor. +func registerFlexFECCodec(me *webrtc.MediaEngine, payloadType webrtc.PayloadType) error { + return me.RegisterCodec(flexFEC03CodecParameters(payloadType), webrtc.RTPCodecTypeVideo) +} + func registerHeaderExtensions(me *webrtc.MediaEngine, rtpHeaderExtension RTPHeaderExtensionConfig) error { for _, extension := range rtpHeaderExtension.Video { if err := me.RegisterHeaderExtension(webrtc.RTPHeaderExtensionCapability{URI: extension}, webrtc.RTPCodecTypeVideo); err != nil { @@ -178,6 +200,14 @@ func filterCodecs( continue } + // FlexFEC is an infrastructure codec (like RTX): it only appears in the codec + // list when it was registered on the media engine for this peer connection, so + // retain it unconditionally. The room enabled-codec list only tracks media codecs. + if strings.EqualFold(c.RTPCodecCapability.MimeType, webrtc.MimeTypeFlexFEC03) { + filteredCodecs = append(filteredCodecs, c) + continue + } + for _, enabledCodec := range enabledCodecs { if mime.NormalizeMimeType(enabledCodec.Mime) == mime.NormalizeMimeType(c.RTPCodecCapability.MimeType) { if !mime.IsMimeTypeStringEqual(c.RTPCodecCapability.MimeType, mime.MimeTypeRTX.String()) { diff --git a/pkg/rtc/transport.go b/pkg/rtc/transport.go index 8f6a5e50a..884a81a2a 100644 --- a/pkg/rtc/transport.go +++ b/pkg/rtc/transport.go @@ -49,6 +49,7 @@ import ( "github.com/livekit/livekit-server/pkg/sfu/bwe/remotebwe" "github.com/livekit/livekit-server/pkg/sfu/bwe/sendsidebwe" "github.com/livekit/livekit-server/pkg/sfu/datachannel" + "github.com/livekit/livekit-server/pkg/sfu/flexfec" sfuinterceptor "github.com/livekit/livekit-server/pkg/sfu/interceptor" "github.com/livekit/livekit-server/pkg/sfu/pacer" pd "github.com/livekit/livekit-server/pkg/sfu/rtpextension/playoutdelay" @@ -443,6 +444,74 @@ func newPeerConnection( se.LoggerFactory = pionlogger.NewLoggerFactory(params.Logger) ir := &interceptor.Registry{} + + // FlexFEC (flexfec-03) setup. + // + // FlexFEC is terminated at the SFU; the two directions are independent: + // - Subscriber side (this PC sends media to the client): register the flexfec-03 + // codec AND the pion encoder interceptor so a fresh repair stream is generated + // per downstream. ConfigureFlexFEC03 must be added before any interceptor that + // mutates outgoing RTP (e.g. the TWCC header extension interceptor below). + // - Publisher side (this PC receives media from the client): register only the + // flexfec-03 codec so the client negotiates sending us a repair stream, which the + // buffer layer associates and decodes. + if flexFEC := params.Config.FlexFEC; flexFEC.Enabled() { + fecPayloadType := webrtc.PayloadType(flexFEC.PayloadType) + // in single-PC / one-shot mode a single PC both sends and receives, so Transport is + // PUBLISHER while IsSendSide is true - it then needs both behaviors. + isSubscriberRole := params.IsSendSide + isPublisherRole := params.Transport == livekit.SignalTarget_PUBLISHER + wantEncoder := isSubscriberRole && flexFEC.Subscriber + wantReceiveCodec := isPublisherRole && flexFEC.Publisher + switch { + case wantEncoder: + // Register the flexfec-03 codec so the sender allocates a repair SSRC and the + // stream info carries the FEC payload type / SSRC, then add our own encoder + // interceptor. We deliberately do NOT use pion's webrtc.ConfigureFlexFEC03: + // pion's encoder interceptor retains each media packet's payload slice by + // reference and only XORs the block once it is full, but LiveKit recycles RTP + // payload buffers back to a sync.Pool as soon as the packet is written (see + // pacer.Base.SendPacket). By the time pion computed the parity, earlier + // payloads in the block had been overwritten, producing FEC that protects + // garbage and causing receiver-side frame corruption / PLI storms. Our + // interceptor copies the payload when buffering to avoid this. + // + // It is added to the registry here, before the TWCC header extension + // interceptor below, so that when the interceptor-based send-side BWE is in + // use the generated FEC packets are assigned transport-wide sequence numbers + // and accounted for by congestion control. + numMediaPackets, numFecPackets := flexFEC.EncoderParams() + if err := registerFlexFECCodec(me, fecPayloadType); err != nil { + params.Logger.Warnw("failed to register flexfec-03 codec for generation", err) + } else { + ir.Add(flexfec.NewEncoderInterceptorFactory(numMediaPackets, numFecPackets)) + params.Logger.Infow("flexfec-03 subscriber generation enabled", + "payloadType", fecPayloadType, + "numMediaPackets", numMediaPackets, + "numFecPackets", numFecPackets, + ) + // FEC packets are generated inside our encoder interceptor, downstream of + // LiveKit's pacer. With the RemoteBWE path the added load is observed + // through receiver feedback (loss / REMB), and with the interceptor-based + // send-side BWE the packets are TWCC-tagged (see ordering above). With + // LiveKit's own pacer-based send-side BWE the FEC overhead is not fed back + // into the estimator, so warn to make the limitation explicit. + if params.CongestionControlConfig.UseSendSideBWE && !params.CongestionControlConfig.UseSendSideBWEInterceptor { + params.Logger.Warnw( + "flexfec-03 generation enabled with pacer-based send-side BWE; "+ + "FEC overhead is not accounted for by congestion control", nil, + ) + } + } + case wantReceiveCodec: + if err := registerFlexFECCodec(me, fecPayloadType); err != nil { + params.Logger.Warnw("failed to register flexfec-03 codec", err) + } else { + params.Logger.Debugw("flexfec-03 publisher reception enabled", "payloadType", fecPayloadType) + } + } + } + if params.IsSendSide { if params.CongestionControlConfig.UseSendSideBWEInterceptor && !params.CongestionControlConfig.UseSendSideBWE { params.Logger.Infow("using send side BWE - interceptor") @@ -1649,6 +1718,15 @@ func (t *PCTransport) HandleRemoteDescription(sd webrtc.SessionDescription, remo t.params.Config.BufferFactory.SetRTXPair(repair, base, "") } } + if t.params.Config.FlexFEC.Publisher { + fecRepairs := nonSimulcastFECRepairsFromSDP(parsed, t.params.Logger) + if len(fecRepairs) > 0 { + t.params.Logger.Debugw("flexfec pairs found from sdp", "ssrcs", fecRepairs) + for repair, base := range fecRepairs { + t.params.Config.BufferFactory.SetFECPair(repair, base) + } + } + } return nil } @@ -2874,6 +2952,16 @@ func (t *PCTransport) handleRemoteOfferReceived(sd *webrtc.SessionDescription, o } } + if t.params.Config.FlexFEC.Publisher { + fecRepairs := nonSimulcastFECRepairsFromSDP(parsed, t.params.Logger) + if len(fecRepairs) > 0 { + t.params.Logger.Debugw("flexfec pairs found from sdp", "ssrcs", fecRepairs) + for repair, base := range fecRepairs { + t.params.Config.BufferFactory.SetFECPair(repair, base) + } + } + } + if t.currentOfferIceCredential == "" || offerRestartICE { t.currentOfferIceCredential = iceCredential } @@ -3245,6 +3333,47 @@ func nonSimulcastRTXRepairsFromSDP(s *sdp.SessionDescription, logger logger.Logg return rtxRepairFlows } +// nonSimulcastFECRepairsFromSDP extracts FlexFEC repair flows from non-simulcast media +// sections, returning a map of FEC(repair) SSRC -> source(base) SSRC. It mirrors +// nonSimulcastRTXRepairsFromSDP but parses `a=ssrc-group:FEC-FR ` lines +// (RFC 5956), where the first SSRC is the protected source stream and the second is the +// FlexFEC repair stream. +func nonSimulcastFECRepairsFromSDP(s *sdp.SessionDescription, logger logger.Logger) map[uint32]uint32 { + fecRepairFlows := map[uint32]uint32{} + for _, media := range s.MediaDescriptions { + var ridFound bool + fecPairs := make(map[uint32]uint32) + findFEC: + for _, attr := range media.Attributes { + switch attr.Key { + case "rid": + ridFound = true + break findFEC + case sdp.AttrKeySSRCGroup: + split := strings.Split(attr.Value, " ") + if split[0] == sdp.SemanticTokenForwardErrorCorrectionFramework && len(split) == 3 { + baseSsrc, err := strconv.ParseUint(split[1], 10, 32) + if err != nil { + logger.Warnw("Failed to parse FEC source SSRC", err, "ssrc", split[1]) + continue + } + fecRepairFlow, err := strconv.ParseUint(split[2], 10, 32) + if err != nil { + logger.Warnw("Failed to parse FEC repair SSRC", err, "ssrc", split[2]) + continue + } + fecPairs[uint32(fecRepairFlow)] = uint32(baseSsrc) + } + } + } + if !ridFound { + maps.Copy(fecRepairFlows, fecPairs) + } + } + + return fecRepairFlows +} + // ---------------------- type iceCandidatePairStatsEncoder struct { diff --git a/pkg/sfu/buffer/buffer.go b/pkg/sfu/buffer/buffer.go index 5b376a490..c91b5dd7e 100644 --- a/pkg/sfu/buffer/buffer.go +++ b/pkg/sfu/buffer/buffer.go @@ -23,6 +23,7 @@ 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" @@ -33,6 +34,9 @@ import ( const ( rtcpReceiverReportDelta = 1e9 + // fecReportInterval bounds how often a buffer logs a FlexFEC recovery summary. + fecReportInterval = 5e9 + InitPacketBufferSizeVideo = 300 InitPacketBufferSizeAudio = 70 ) @@ -72,6 +76,28 @@ type Buffer struct { primaryBufferForRTX *Buffer rtxPktBuf []byte + + // FlexFEC (publisher -> SFU recovery). + // primaryBufferForFEC is set on a repair (FEC) buffer and points at the source + // buffer it protects. fecDecoder lives on the source buffer and reconstructs lost + // source packets from the repair stream. fecPktBuf is scratch for re-marshaling + // recovered packets before injecting them back into the source buffer. + primaryBufferForFEC *Buffer + fecDecoder *flexfec.Decoder + fecPktBuf []byte + + // FlexFEC observability. Counters are cumulative for the lifetime of the source + // buffer; the *Reported fields snapshot the last logged values so the periodic + // summary can report deltas. + fecPacketsReceived uint64 + fecPacketsRecovered uint64 + fecReportedAt int64 + fecReportedReceived uint64 + fecReportedRecovered uint64 + + // Debug-only simulated publisher ("robot") uplink impairment; nil unless the + // LK_UPLINK_* env vars are set. See buffer_impair.go. + impair *uplinkImpair } func NewBuffer(ssrc uint32, maxVideoPkts, maxAudioPkts int) *Buffer { @@ -84,6 +110,7 @@ func NewBuffer(ssrc uint32, maxVideoPkts, maxAudioPkts int) *Buffer { SendPLI: b.sendPLI, IsReportingEnabled: true, }) + b.initUplinkImpair() return b } @@ -138,7 +165,24 @@ func (b *Buffer) Bind(params webrtc.RTPParameters, codec webrtc.RTPCodecCapabili } // Write adds an RTP Packet, ordering is not guaranteed, newer packets may arrive later -func (b *Buffer) Write(pkt []byte) (n int, err error) { +// Write is the inbound entry point for the publisher's RTP (media, RTX, and FEC SSRCs). +// It applies the optional debug uplink impairment (loss + one-way delay) before handing +// the packet to writeNow; both are no-ops unless the LK_UPLINK_* env vars are set. +func (b *Buffer) Write(pkt []byte) (int, error) { + if b.impair != nil { + if b.impair.dropInbound() { + // Packet "lost" on the simulated 5G uplink; report success so pion's read + // loop is unaffected. + return len(pkt), nil + } + if b.impair.delayInbound(b, pkt) { + return len(pkt), nil + } + } + return b.writeNow(pkt) +} + +func (b *Buffer) writeNow(pkt []byte) (n int, err error) { var rtpPacket rtp.Packet err = rtpPacket.Unmarshal(pkt) if err != nil { @@ -179,6 +223,13 @@ func (b *Buffer) Write(pkt []byte) (n int, err error) { return } + // handle FlexFEC repair packet + if pb := b.primaryBufferForFEC; pb != nil { + b.Unlock() + pb.writeFEC(&rtpPacket, now) + return + } + if !b.isBound { packet := make([]byte, len(pkt)) copy(packet, pkt) @@ -203,16 +254,33 @@ func (b *Buffer) Write(pkt []byte) (n int, err error) { } rtcpPackets := b.calc(pkt, &rtpPacket, now, false, false) + + // feed received source packets to the FlexFEC decoder so it can reconstruct + // previously lost packets, and inject any recoveries back into this buffer. + if b.fecDecoder != nil { + b.injectRecoveredLocked(b.fecDecoder.Decode(rtpPacket), now) + } + b.Unlock() if len(rtcpPackets) != 0 { - if cb := b.getOnRtcpFeedback(); cb != nil { - cb(rtcpPackets) - } + b.emitRTCPFeedback(rtcpPackets) } return } +// emitRTCPFeedback delivers SFU->publisher feedback (NACK/RR/etc). When the debug uplink +// delay is set it defers delivery by the one-way delay so a NACK->RTX recovery costs a +// full round trip; otherwise it fires immediately. +func (b *Buffer) emitRTCPFeedback(rtcpPackets []rtcp.Packet) { + if b.impair != nil && b.impair.delayFeedback(b, rtcpPackets) { + return + } + if cb := b.getOnRtcpFeedback(); cb != nil { + cb(rtcpPackets) + } +} + func (b *Buffer) SetPrimaryBufferForRTX(primaryBuffer *Buffer) { b.Lock() b.primaryBufferForRTX = primaryBuffer @@ -233,6 +301,114 @@ func (b *Buffer) SetPrimaryBufferForRTX(primaryBuffer *Buffer) { } } +// SetPrimaryBufferForFEC associates this (repair/FEC) buffer with the source buffer it +// protects. It installs a FlexFEC decoder on the source buffer and replays any FEC +// packets that arrived before the association was known. +func (b *Buffer) SetPrimaryBufferForFEC(primaryBuffer *Buffer) { + primaryBuffer.enableFECDecoder(b.BufferBase.SSRC()) + + b.Lock() + b.primaryBufferForFEC = primaryBuffer + pkts := b.pPackets + b.pPackets = nil + b.Unlock() + + for _, pp := range pkts { + var rtpPacket rtp.Packet + if err := rtpPacket.Unmarshal(pp.packet); err != nil { + continue + } + primaryBuffer.writeFEC(&rtpPacket, pp.arrivalTime) + } +} + +// enableFECDecoder installs a FlexFEC decoder on this (source) buffer for the given +// repair SSRC. Safe to call before the buffer is bound. +func (b *Buffer) enableFECDecoder(fecSSRC uint32) { + b.Lock() + defer b.Unlock() + + if b.fecDecoder == nil { + b.fecDecoder = flexfec.NewDecoder(fecSSRC, b.BufferBase.SSRC()) + b.logger.Infow("flexfec decoder enabled", "fecSSRC", fecSSRC, "mediaSSRC", b.BufferBase.SSRC()) + } +} + +// writeFEC feeds a received FlexFEC repair packet to this (source) buffer's decoder and +// injects any recovered source packets back into the buffer. Called on the source buffer. +func (b *Buffer) writeFEC(fecPkt *rtp.Packet, arrivalTime int64) { + b.Lock() + defer b.Unlock() + + if b.fecDecoder == nil { + return + } + + b.fecPacketsReceived++ + b.injectRecoveredLocked(b.fecDecoder.Decode(*fecPkt), arrivalTime) + b.maybeLogFECStatsLocked(arrivalTime) +} + +// maybeLogFECStatsLocked emits a periodic FlexFEC recovery summary at most once per +// fecReportInterval, and only when there has been FEC activity since the last summary. +// b.Lock must be held. +func (b *Buffer) maybeLogFECStatsLocked(now int64) { + if b.fecReportedAt == 0 { + b.fecReportedAt = now + b.fecReportedReceived = b.fecPacketsReceived + b.fecReportedRecovered = b.fecPacketsRecovered + return + } + + if now-b.fecReportedAt < fecReportInterval { + return + } + + receivedDelta := b.fecPacketsReceived - b.fecReportedReceived + recoveredDelta := b.fecPacketsRecovered - b.fecReportedRecovered + + b.fecReportedAt = now + b.fecReportedReceived = b.fecPacketsReceived + b.fecReportedRecovered = b.fecPacketsRecovered + + if receivedDelta == 0 && recoveredDelta == 0 { + return + } + + b.logger.Infow( + "flexfec recovery stats", + "mediaSSRC", b.BufferBase.SSRC(), + "fecPacketsReceived", b.fecPacketsReceived, + "fecPacketsReceivedDelta", receivedDelta, + "packetsRecovered", b.fecPacketsRecovered, + "packetsRecoveredDelta", recoveredDelta, + ) +} + +// injectRecoveredLocked re-injects FlexFEC-recovered source packets into the buffer as +// repaired packets (treated like RTX repairs, so they become NACK/forward eligible). +// b.Lock must be held. +func (b *Buffer) injectRecoveredLocked(recovered []rtp.Packet, arrivalTime int64) { + if len(recovered) == 0 || !b.isBound { + return + } + + if b.fecPktBuf == nil { + b.fecPktBuf = make([]byte, bucket.RTPMaxPktSize) + } + + for i := range recovered { + pkt := recovered[i] + n, err := pkt.MarshalTo(b.fecPktBuf) + if err != nil { + b.logger.Errorw("could not marshal flexfec recovered packet", err, "sn", pkt.SequenceNumber) + continue + } + b.calc(b.fecPktBuf[:n], &pkt, arrivalTime, false, true) + b.fecPacketsRecovered++ + } +} + func (b *Buffer) NotifyRTX(ssrc uint32, repairSSRC uint32, rsid string) { if onNotifyRTX := b.getOnNotifyRTX(); onNotifyRTX != nil { onNotifyRTX(ssrc, repairSSRC, rsid) diff --git a/pkg/sfu/buffer/buffer_impair.go b/pkg/sfu/buffer/buffer_impair.go new file mode 100644 index 000000000..05abc483f --- /dev/null +++ b/pkg/sfu/buffer/buffer_impair.go @@ -0,0 +1,204 @@ +// Copyright 2024 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package buffer + +import ( + "math/rand" + "os" + "strconv" + "sync" + "time" + + "github.com/pion/rtcp" +) + +// Debug-only simulated publisher ("robot") uplink impairment for the local FlexFEC test +// harness (scripts/flexfec). It models the lossy, higher-RTT first hop a teleop robot sees +// on a wireless/5G link: random loss on inbound RTP (media + RTX + FEC, since real loss +// hits everything) and a one-way delay applied symmetrically to inbound media AND the +// SFU->publisher feedback (NACK), so a NACK->RTX recovery costs a full round trip while a +// normal packet pays one one-way delay. Publisher-side FlexFEC repairs that loss inline at +// the SFU without any round trip -- which is the latency it saves here. +// +// CRITICAL ORDERING NOTE: a real link is ONE pipe -- it delays/loses every packet of every +// SSRC (media, RTX, FEC) together while preserving their global arrival order. The SFU's +// FlexFEC decoder depends on that ordering (it XORs a repair packet against the surviving +// media packets of its block). An earlier design gave each Buffer (media/RTX/FEC SSRC) its +// own delay goroutine; the independent goroutines reordered the streams relative to each +// other and FEC recovery dropped to zero. So this impairment is a single process-global +// FIFO delay line with one drain goroutine: with a constant delay, FIFO dequeue preserves +// the exact arrival order (just time-shifted), and recovery behaves identically to the +// no-delay path. +// +// Controlled by env vars; absent/zero values make this a complete no-op: +// +// LK_PUB_LOSS fractional loss on the publisher->SFU path, e.g. 0.03 (3%) +// LK_PUB_DELAY_MS one-way publisher<->SFU delay in ms (inbound media + outbound NACK) +// +// (These are the robot-link knobs. The operator/subscriber-side knobs live in the sfu +// package: LK_DOWNLINK_LOSS, LK_DOWNLINK_DELAY_MS, LK_UPLINK_DELAY_MS.) +const ( + envPubLoss = "LK_PUB_LOSS" + envPubDelayMs = "LK_PUB_DELAY_MS" + + impairQueueDepth = 8192 +) + +type delayedInbound struct { + b *Buffer + pkt []byte + releaseAt int64 // unix nanos +} + +type delayedFeedback struct { + b *Buffer + pkts []rtcp.Packet + releaseAt int64 // unix nanos +} + +type uplinkImpair struct { + loss float64 + delay time.Duration + + rngMu sync.Mutex + rng *rand.Rand + + inQueue chan delayedInbound + fbQueue chan delayedFeedback +} + +var ( + globalImpairOnce sync.Once + globalImpair *uplinkImpair +) + +// getUplinkImpair lazily builds the process-global uplink impairment from the LK_PUB_* env +// vars, or returns nil when none are set. All inbound Buffers share the one instance so the +// delay line preserves global packet ordering across every SSRC (see the note above). +func getUplinkImpair() *uplinkImpair { + globalImpairOnce.Do(func() { + loss := parseEnvFloat(envPubLoss) + delay := time.Duration(parseEnvInt(envPubDelayMs)) * time.Millisecond + if loss <= 0 && delay <= 0 { + return + } + imp := &uplinkImpair{ + loss: loss, + delay: delay, + rng: rand.New(rand.NewSource(time.Now().UnixNano())), + } + if delay > 0 { + imp.inQueue = make(chan delayedInbound, impairQueueDepth) + imp.fbQueue = make(chan delayedFeedback, impairQueueDepth) + go imp.runInboundDelayLine() + go imp.runFeedbackDelayLine() + } + globalImpair = imp + }) + return globalImpair +} + +// initUplinkImpair attaches the shared uplink impairment to this Buffer (nil unless the +// LK_PUB_* env vars are set). Called once per Buffer from NewBuffer. +func (b *Buffer) initUplinkImpair() { + b.impair = getUplinkImpair() + if b.impair != nil && b.logger != nil { + b.logger.Infow("uplink impairment enabled (debug)", "lossFraction", b.impair.loss, "delay", b.impair.delay) + } +} + +func (u *uplinkImpair) dropInbound() bool { + if u.loss <= 0 { + return false + } + u.rngMu.Lock() + r := u.rng.Float64() + u.rngMu.Unlock() + return r < u.loss +} + +// delayInbound returns true when the packet was deferred onto the shared delay line (caller +// should return immediately); false when no delay is configured and the caller should +// proceed inline. +func (u *uplinkImpair) delayInbound(b *Buffer, pkt []byte) bool { + if u.delay <= 0 { + return false + } + // pion recycles pkt after Write returns, so copy before deferring. + cp := make([]byte, len(pkt)) + copy(cp, pkt) + item := delayedInbound{b: b, pkt: cp, releaseAt: time.Now().Add(u.delay).UnixNano()} + select { + case u.inQueue <- item: + default: + // Queue full: drop rather than block pion's read loop. + } + return true +} + +// delayFeedback returns true when feedback was deferred onto the shared delay line; false +// when no delay is configured and the caller should send immediately. +func (u *uplinkImpair) delayFeedback(b *Buffer, pkts []rtcp.Packet) bool { + if u.delay <= 0 { + return false + } + item := delayedFeedback{b: b, pkts: pkts, releaseAt: time.Now().Add(u.delay).UnixNano()} + select { + case u.fbQueue <- item: + default: + } + return true +} + +// runInboundDelayLine releases delayed inbound packets, in global FIFO order, into the +// owning buffer's writeNow. A constant delay keeps releaseAt monotonic, so a single +// goroutine preserves the exact arrival order across all SSRCs. +func (u *uplinkImpair) runInboundDelayLine() { + for item := range u.inQueue { + if w := item.releaseAt - time.Now().UnixNano(); w > 0 { + time.Sleep(time.Duration(w)) + } + _, _ = item.b.writeNow(item.pkt) + } +} + +// runFeedbackDelayLine releases delayed SFU->publisher feedback (NACK/RR) after the one-way +// uplink delay. +func (u *uplinkImpair) runFeedbackDelayLine() { + for item := range u.fbQueue { + if w := item.releaseAt - time.Now().UnixNano(); w > 0 { + time.Sleep(time.Duration(w)) + } + if cb := item.b.getOnRtcpFeedback(); cb != nil { + cb(item.pkts) + } + } +} + +func parseEnvFloat(key string) float64 { + v, err := strconv.ParseFloat(os.Getenv(key), 64) + if err != nil { + return 0 + } + return v +} + +func parseEnvInt(key string) int { + v, err := strconv.Atoi(os.Getenv(key)) + if err != nil { + return 0 + } + return v +} diff --git a/pkg/sfu/buffer/factory.go b/pkg/sfu/buffer/factory.go index 4a73ab057..1e973745f 100644 --- a/pkg/sfu/buffer/factory.go +++ b/pkg/sfu/buffer/factory.go @@ -40,6 +40,7 @@ func (f *FactoryOfBufferFactory) CreateBufferFactory() *Factory { rtpBuffers: make(map[uint32]*Buffer), rtcpReaders: make(map[uint32]*RTCPReader), rtxPair: make(map[uint32]uint32), + fecPair: make(map[uint32]uint32), } } @@ -50,6 +51,7 @@ type Factory struct { rtpBuffers map[uint32]*Buffer rtcpReaders map[uint32]*RTCPReader rtxPair map[uint32]uint32 // repair -> base + fecPair map[uint32]uint32 // flexfec repair -> base } func (f *Factory) GetOrNew(packetType packetio.BufferPacketType, ssrc uint32) io.ReadWriteCloser { @@ -89,10 +91,24 @@ func (f *Factory) GetOrNew(packetType packetio.BufferPacketType, ssrc uint32) io break } } + for repair, base := range f.fecPair { + if repair == ssrc { + if baseBuffer, ok := f.rtpBuffers[base]; ok { + buffer.SetPrimaryBufferForFEC(baseBuffer) + } + break + } else if base == ssrc { + if repairBuffer, ok := f.rtpBuffers[repair]; ok { + repairBuffer.SetPrimaryBufferForFEC(buffer) + } + break + } + } buffer.OnClose(func() { f.Lock() delete(f.rtpBuffers, ssrc) delete(f.rtxPair, ssrc) + delete(f.fecPair, ssrc) f.Unlock() }) return buffer @@ -132,3 +148,19 @@ func (f *Factory) SetRTXPair(repair, base uint32, rsid string) { } } } + +// SetFECPair associates a FlexFEC repair stream SSRC with the source (base) SSRC it +// protects, as signalled by an a=ssrc-group:FEC-FR line. If both buffers already exist +// the association is wired immediately, otherwise it is remembered and applied when the +// missing buffer is created. +func (f *Factory) SetFECPair(repair, base uint32) { + f.Lock() + repairBuffer, baseBuffer := f.rtpBuffers[repair], f.rtpBuffers[base] + if repairBuffer == nil || baseBuffer == nil { + f.fecPair[repair] = base + } + f.Unlock() + if repairBuffer != nil && baseBuffer != nil { + repairBuffer.SetPrimaryBufferForFEC(baseBuffer) + } +} diff --git a/pkg/sfu/buffer/flexfec_test.go b/pkg/sfu/buffer/flexfec_test.go new file mode 100644 index 000000000..4f4e3f5d3 --- /dev/null +++ b/pkg/sfu/buffer/flexfec_test.go @@ -0,0 +1,148 @@ +// Copyright 2024 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package buffer + +import ( + "testing" + "time" + + "github.com/pion/interceptor/pkg/flexfec" + "github.com/pion/rtp" + "github.com/pion/transport/v4/packetio" + "github.com/pion/webrtc/v4" + "github.com/stretchr/testify/require" +) + +// TestFlexFECRecoveryThroughFactory exercises the full SFU receive path: the source and +// repair buffers are created through the factory (as the SRTP layer would), associated +// via SetFECPair, and a lost source packet is recovered from the FlexFEC repair stream +// and forwarded out of the source buffer. +func TestFlexFECRecoveryThroughFactory(t *testing.T) { + const ( + mediaSSRC = uint32(0xAABBCCDD) + fecSSRC = uint32(0x11223344) + baseSeq = uint16(5000) + numMedia = 10 + dropIdx = 6 + ) + + factory := NewFactoryOfBufferFactory(InitPacketBufferSizeVideo, InitPacketBufferSizeAudio).CreateBufferFactory() + + mediaBuf, ok := factory.GetOrNew(packetio.RTPBufferPacket, mediaSSRC).(*Buffer) + require.True(t, ok) + mediaBuf.codecType = webrtc.RTPCodecTypeAudio + require.NoError(t, mediaBuf.Bind( + webrtc.RTPParameters{Codecs: []webrtc.RTPCodecParameters{opusCodec}}, + opusCodec.RTPCodecCapability, + 0, + )) + + fecBuf, ok := factory.GetOrNew(packetio.RTPBufferPacket, fecSSRC).(*Buffer) + require.True(t, ok) + + // associate the repair stream with the source stream (as a=ssrc-group:FEC-FR would) + factory.SetFECPair(fecSSRC, mediaSSRC) + require.NotNil(t, mediaBuf.fecDecoder) + require.Equal(t, mediaBuf, fecBuf.primaryBufferForFEC) + + // build media packets and the protecting FlexFEC packet + media := make([]rtp.Packet, numMedia) + for i := 0; i < numMedia; i++ { + payload := make([]byte, 16) + for j := range payload { + payload[j] = byte((i*11 + j*5 + 1) & 0xff) + } + media[i] = rtp.Packet{ + Header: rtp.Header{ + Version: 2, + PayloadType: uint8(opusCodec.PayloadType), + SequenceNumber: baseSeq + uint16(i), + Timestamp: uint32(8000 + i*960), + SSRC: mediaSSRC, + }, + Payload: payload, + } + } + + encoder := flexfec.NewFlexEncoder03(uint8(49), fecSSRC) + mediaForFec := make([]rtp.Packet, numMedia) + copy(mediaForFec, media) + fecPackets := encoder.EncodeFec(mediaForFec, 1) + require.NotEmpty(t, fecPackets) + + // collect forwarded packets out of the source buffer + got := make(chan *ExtPacket, numMedia*2) + go func() { + var buf [1500]byte + for { + ep, err := mediaBuf.ReadExtended(buf[:]) + if err != nil { + return + } + if ep != nil { + clone := *ep + got <- &clone + } + } + }() + + // deliver all source packets except the dropped one + for i := 0; i < numMedia; i++ { + if i == dropIdx { + continue + } + raw, err := media[i].Marshal() + require.NoError(t, err) + _, err = mediaBuf.Write(raw) + require.NoError(t, err) + } + + // deliver the repair packet to the FEC buffer -> triggers recovery + injection + for _, fp := range fecPackets { + raw, err := fp.Marshal() + require.NoError(t, err) + _, err = fecBuf.Write(raw) + require.NoError(t, err) + } + + // the dropped source packet must be recovered and forwarded + recoveredSeq := baseSeq + uint16(dropIdx) + deadline := time.After(2 * time.Second) + seen := map[uint16]*ExtPacket{} + for { + select { + case ep := <-got: + seen[ep.Packet.SequenceNumber] = ep + if rec, found := seen[recoveredSeq]; found { + require.Equal(t, mediaSSRC, rec.Packet.SSRC) + require.Equal(t, media[dropIdx].Payload, rec.Packet.Payload) + require.Equal(t, media[dropIdx].Timestamp, rec.Packet.Timestamp) + _ = mediaBuf.Close() + return + } + case <-deadline: + _ = mediaBuf.Close() + t.Fatalf("recovered packet seq %d not forwarded; saw %v", recoveredSeq, seqKeys(seen)) + } + } +} + +func seqKeys(m map[uint16]*ExtPacket) []uint16 { + keys := make([]uint16, 0, len(m)) + for k := range m { + keys = append(keys, k) + } + return keys +} diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index 38cf31569..eeea37eb7 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -630,18 +630,22 @@ func (d *DownTrack) Bind(t webrtc.TrackLocalContext) (webrtc.RTPCodecParameters, ) d.params.Logger.Debugw("DownTrack.Bind", logFields...) - d.writeStream = t.WriteStream() + // maybeWrapDownlinkImpairment is a no-op unless the LK_DOWNLINK_* debug env vars + // are set (local FlexFEC test harness only); see downtrack_impair.go. + d.writeStream = maybeWrapDownlinkImpairment(t.WriteStream(), d.params.Logger) if rr := d.params.BufferFactory.GetOrNew(packetio.RTCPBufferPacket, d.ssrc).(*buffer.RTCPReader); rr != nil { - rr.OnPacket(func(pkt []byte) { + // maybeDelayInboundRTCP is a no-op unless the LK_UPLINK_DELAY_MS debug env var + // is set (local FlexFEC test harness only); see downtrack_impair.go. + rr.OnPacket(maybeDelayInboundRTCP(func(pkt []byte) { d.handleRTCP(pkt) - }) + }, d.params.Logger)) d.rtcpReader = rr } if d.ssrcRTX != 0 { if rr := d.params.BufferFactory.GetOrNew(packetio.RTCPBufferPacket, d.ssrcRTX).(*buffer.RTCPReader); rr != nil { - rr.OnPacket(func(pkt []byte) { + rr.OnPacket(maybeDelayInboundRTCP(func(pkt []byte) { d.handleRTCPRTX(pkt) - }) + }, d.params.Logger)) d.rtcpReaderRTX = rr } } diff --git a/pkg/sfu/downtrack_impair.go b/pkg/sfu/downtrack_impair.go new file mode 100644 index 000000000..87a4590a4 --- /dev/null +++ b/pkg/sfu/downtrack_impair.go @@ -0,0 +1,231 @@ +// Copyright 2024 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package sfu + +import ( + "math/rand" + "os" + "strconv" + "sync" + "time" + + "github.com/pion/rtp" + "github.com/pion/webrtc/v4" + + "github.com/livekit/protocol/logger" +) + +// Debug-only downlink network impairment for the local FlexFEC test harness +// (scripts/flexfec). When the SFU, publisher and subscriber all run on one macOS box, +// media never leaves the host, and pf/dummynet (netem.sh) cannot shape same-host UDP +// because the kernel short-circuits local delivery before the pf OUTPUT hook. To run a +// meaningful FEC-vs-RTX comparison we instead inject loss + one-way delay in software, +// right at the subscriber DownTrack egress. +// +// Controlled entirely by env vars; absent/zero values make wrapping a no-op so this has +// ZERO effect on normal operation: +// +// LK_DOWNLINK_LOSS fractional packet loss applied to SFU->subscriber, e.g. 0.01 (1%) +// LK_DOWNLINK_DELAY_MS one-way delay added to SFU->subscriber in milliseconds, e.g. 40 +// LK_UPLINK_DELAY_MS one-way delay added to the subscriber->SFU feedback path (inbound +// RTCP, i.e. NACK/RR/REMB/PLI) in milliseconds, e.g. 40 +// +// Set DOWNLINK and UPLINK to the same value to model a symmetric link: a normal frame pays +// one one-way downlink delay, whereas a NACK->RTX recovery pays uplink (delayed NACK) + +// downlink (delayed RTX) = a full RTT. That extra round trip is precisely the latency +// FlexFEC avoids by repairing inline, so a symmetric delay is the fair test of FEC's value. +const ( + envDownlinkLoss = "LK_DOWNLINK_LOSS" + envDownlinkDelayMs = "LK_DOWNLINK_DELAY_MS" + envUplinkDelayMs = "LK_UPLINK_DELAY_MS" + + // Bounded queue so a stalled subscriber can't grow memory without limit; at ~300 + // pkt/s and tens of ms of delay only a handful are ever in flight. + impairedQueueDepth = 2048 +) + +// maybeWrapDownlinkImpairment returns w wrapped with loss/delay impairment when the +// LK_DOWNLINK_* env vars request it, otherwise returns w unchanged. +func maybeWrapDownlinkImpairment(w webrtc.TrackLocalWriter, log logger.Logger) webrtc.TrackLocalWriter { + loss := parseEnvFloat(envDownlinkLoss) + delay := time.Duration(parseEnvInt(envDownlinkDelayMs)) * time.Millisecond + if loss <= 0 && delay <= 0 { + return w + } + if log != nil { + log.Infow("downlink impairment enabled (debug)", "lossFraction", loss, "delay", delay) + } + iw := &impairedWriter{ + inner: w, + loss: loss, + delay: delay, + rng: rand.New(rand.NewSource(time.Now().UnixNano())), + } + if delay > 0 { + iw.queue = make(chan delayedWrite, impairedQueueDepth) + go iw.runDelayLine() + } + return iw +} + +func parseEnvFloat(key string) float64 { + v, err := strconv.ParseFloat(os.Getenv(key), 64) + if err != nil { + return 0 + } + return v +} + +func parseEnvInt(key string) int { + v, err := strconv.Atoi(os.Getenv(key)) + if err != nil { + return 0 + } + return v +} + +type delayedWrite struct { + header *rtp.Header + payload []byte + raw []byte + releaseAt time.Time +} + +type impairedWriter struct { + inner webrtc.TrackLocalWriter + loss float64 + delay time.Duration + + mu sync.Mutex + rng *rand.Rand + + queue chan delayedWrite +} + +func (iw *impairedWriter) drop() bool { + if iw.loss <= 0 { + return false + } + iw.mu.Lock() + r := iw.rng.Float64() + iw.mu.Unlock() + return r < iw.loss +} + +func (iw *impairedWriter) WriteRTP(header *rtp.Header, payload []byte) (int, error) { + if iw.drop() { + // Report success so DownTrack stats/sequencing are unaffected; the packet is + // simply "lost" in the simulated network. + return header.MarshalSize() + len(payload), nil + } + if iw.delay <= 0 { + return iw.inner.WriteRTP(header, payload) + } + + // The caller's header/payload come from pooled buffers that are recycled the moment + // this returns, so everything retained past this call must be copied. + dw := delayedWrite{ + header: cloneRTPHeader(header), + payload: append([]byte(nil), payload...), + releaseAt: time.Now().Add(iw.delay), + } + n := header.MarshalSize() + len(payload) + select { + case iw.queue <- dw: + default: + // Queue full (subscriber stalled): drop rather than block the write path. + } + return n, nil +} + +func (iw *impairedWriter) Write(b []byte) (int, error) { + if iw.drop() { + return len(b), nil + } + if iw.delay <= 0 { + return iw.inner.Write(b) + } + dw := delayedWrite{ + raw: append([]byte(nil), b...), + releaseAt: time.Now().Add(iw.delay), + } + select { + case iw.queue <- dw: + default: + } + return len(b), nil +} + +// runDelayLine releases queued writes in order once their release time arrives. A constant +// delay keeps releaseAt monotonic, so a single goroutine preserves packet order. +func (iw *impairedWriter) runDelayLine() { + for dw := range iw.queue { + if d := time.Until(dw.releaseAt); d > 0 { + time.Sleep(d) + } + if dw.raw != nil { + _, _ = iw.inner.Write(dw.raw) + continue + } + _, _ = iw.inner.WriteRTP(dw.header, dw.payload) + } +} + +func cloneRTPHeader(h *rtp.Header) *rtp.Header { + // rtp.Header.Clone deep-copies CSRC and extension payloads, so the result is safe to + // retain after the caller recycles its pooled buffers. + c := h.Clone() + return &c +} + +// maybeDelayInboundRTCP wraps an RTCP packet handler so that inbound RTCP (the +// subscriber->SFU feedback path: NACK/RR/REMB/PLI) is processed LK_UPLINK_DELAY_MS later, +// modelling the uplink leg of the round trip. With this and a matching downlink delay, a +// NACK->RTX recovery pays a full RTT while normal media pays only one one-way delay. +// Returns handler unchanged when LK_UPLINK_DELAY_MS is unset/zero. +func maybeDelayInboundRTCP(handler func([]byte), log logger.Logger) func([]byte) { + delay := time.Duration(parseEnvInt(envUplinkDelayMs)) * time.Millisecond + if delay <= 0 { + return handler + } + if log != nil { + log.Infow("uplink RTCP delay enabled (debug)", "delay", delay) + } + queue := make(chan delayedRTCP, impairedQueueDepth) + go func() { + for d := range queue { + if w := time.Until(d.releaseAt); w > 0 { + time.Sleep(w) + } + handler(d.pkt) + } + }() + return func(pkt []byte) { + // The reader recycles pkt after this returns, so copy before deferring. + dr := delayedRTCP{ + pkt: append([]byte(nil), pkt...), + releaseAt: time.Now().Add(delay), + } + select { + case queue <- dr: + default: + } + } +} + +type delayedRTCP struct { + pkt []byte + releaseAt time.Time +} diff --git a/pkg/sfu/flexfec/decoder.go b/pkg/sfu/flexfec/decoder.go new file mode 100644 index 000000000..952b0e72c --- /dev/null +++ b/pkg/sfu/flexfec/decoder.go @@ -0,0 +1,482 @@ +// Copyright 2024 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Package flexfec implements a FlexFEC-03 decoder used by the SFU to recover +// RTP packets lost on the publisher -> SFU path. +// +// FlexFEC (RFC 8627 / draft-ietf-payload-flexible-fec-scheme-03) sends repair +// packets on a dedicated SSRC. Each repair packet references the protected +// source packets via a base sequence number plus a bitmask and carries the XOR +// of the protected packets' headers and payloads. When a single protected +// packet is missing, it can be reconstructed by XOR-ing the repair packet with +// the other protected source packets. +// +// This is a port of pion/interceptor's (unexported) FlexFEC-03 decoder, adapted +// to be reusable from the SFU buffer layer. The decoder is NOT safe for +// concurrent use; the caller must serialize access (the SFU feeds it from a +// single SRTP read goroutine). +package flexfec + +import ( + "encoding/binary" + "errors" + "fmt" + "sort" + + "github.com/pion/rtp" +) + +var ( + errPacketTruncated = errors.New("packet truncated") + errRetransmissionBitSet = errors.New("packet with retransmission bit set not supported") + errInflexibleGeneratorMatrix = errors.New("packet with inflexible generator matrix not supported") + errMultipleSSRCProtection = errors.New("multiple ssrc protection not supported") + errLastOptionalMaskKBitSetToFalse = errors.New("k-bit of last optional mask is set to false") +) + +const ( + defaultMaxMediaPackets = 100 + defaultMaxFECPackets = 100 + recoveredPacketLimit = 192 +) + +// Decoder reconstructs lost source RTP packets from a FlexFEC-03 repair stream. +// +// Feed it both the received source packets (SSRC == ProtectedSSRC) and the +// received repair packets (SSRC == SSRC) via Decode. Each call returns any +// source packets recovered as a result of the newly inserted packet. +type Decoder struct { + ssrc uint32 + protectedStreamSSRC uint32 + maxMediaPackets int + maxFECPackets int + recoveredPackets []rtp.Packet + receivedFECPackets []fecPacketState +} + +// NewDecoder creates a decoder for a single (repair SSRC, protected SSRC) pair. +func NewDecoder(fecSSRC uint32, protectedStreamSSRC uint32) *Decoder { + return &Decoder{ + ssrc: fecSSRC, + protectedStreamSSRC: protectedStreamSSRC, + maxMediaPackets: defaultMaxMediaPackets, + maxFECPackets: defaultMaxFECPackets, + recoveredPackets: make([]rtp.Packet, 0), + receivedFECPackets: make([]fecPacketState, 0), + } +} + +// SSRC returns the FlexFEC repair stream SSRC this decoder handles. +func (d *Decoder) SSRC() uint32 { + return d.ssrc +} + +// ProtectedSSRC returns the source stream SSRC this decoder protects. +func (d *Decoder) ProtectedSSRC() uint32 { + return d.protectedStreamSSRC +} + +// Decode inserts a received packet (either a source packet on ProtectedSSRC or +// a repair packet on SSRC) and returns any source packets recovered as a +// result. The supplied packet is cloned, so the caller may reuse the backing +// buffer after Decode returns. +func (d *Decoder) Decode(receivedPacket rtp.Packet) []rtp.Packet { + if receivedPacket.SSRC != d.ssrc && receivedPacket.SSRC != d.protectedStreamSSRC { + return nil + } + + pkt := clonePacket(receivedPacket) + + if len(d.recoveredPackets) == d.maxMediaPackets { + backRecoveredPacket := d.recoveredPackets[len(d.recoveredPackets)-1] + if backRecoveredPacket.SSRC == pkt.SSRC { + if seqDiff(pkt.SequenceNumber, backRecoveredPacket.SequenceNumber) > uint16(d.maxMediaPackets) { + d.recoveredPackets = nil + d.receivedFECPackets = nil + } + } + } + + d.insertPacket(pkt) + + return d.attemptRecovery() +} + +func (d *Decoder) insertPacket(receivedPkt rtp.Packet) { + // Discard old FEC packets such that the sequence numbers in receivedFECPackets + // span at most 1/2 of the sequence number space. This keeps the slice sorted + // and reduces incorrect decoding due to sequence number wrap-around. + if len(d.receivedFECPackets) > 0 && receivedPkt.SSRC == d.ssrc { + toRemove := 0 + for _, fecPkt := range d.receivedFECPackets { + if abs(int(receivedPkt.SequenceNumber)-int(fecPkt.packet.SequenceNumber)) > 0x3fff { + toRemove++ + } else { + break + } + } + if toRemove > 0 { + d.receivedFECPackets = d.receivedFECPackets[toRemove:] + } + } + + switch receivedPkt.SSRC { + case d.ssrc: + d.insertFECPacket(receivedPkt) + case d.protectedStreamSSRC: + d.insertMediaPacket(receivedPkt) + } + + d.discardOldRecoveredPackets() +} + +func (d *Decoder) insertMediaPacket(receivedPkt rtp.Packet) { + for _, recoveredPacket := range d.recoveredPackets { + if recoveredPacket.SequenceNumber == receivedPkt.SequenceNumber { + return + } + } + + d.recoveredPackets = append(d.recoveredPackets, receivedPkt) + sort.Slice(d.recoveredPackets, func(i, j int) bool { + return isNewerSeq(d.recoveredPackets[i].SequenceNumber, d.recoveredPackets[j].SequenceNumber) + }) + d.updateCoveringFecPackets(receivedPkt) +} + +func (d *Decoder) updateCoveringFecPackets(receivedPkt rtp.Packet) { + pkt := receivedPkt + for _, fecPkt := range d.receivedFECPackets { + for _, protectedPacket := range fecPkt.protectedPackets { + if protectedPacket.seq == pkt.SequenceNumber { + protectedPacket.packet = &pkt + } + } + } +} + +func (d *Decoder) insertFECPacket(fecPkt rtp.Packet) { + for _, existingFECPacket := range d.receivedFECPackets { + if existingFECPacket.packet.SequenceNumber == fecPkt.SequenceNumber { + return + } + } + + fec, err := parseFlexFEC03Header(fecPkt.Payload) + if err != nil { + return + } + + if fec.protectedSSRC != d.protectedStreamSSRC { + return + } + + protectedSeqs := decodeMask(uint64(fec.mask0), 15, fec.seqNumBase) + if fec.mask1 != 0 { + protectedSeqs = append(protectedSeqs, decodeMask(uint64(fec.mask1), 31, fec.seqNumBase+15)...) + } + if fec.mask2 != 0 { + protectedSeqs = append(protectedSeqs, decodeMask(fec.mask2, 63, fec.seqNumBase+46)...) + } + + if len(protectedSeqs) == 0 { + return + } + + protectedPackets := make([]*protectedPacket, 0, len(protectedSeqs)) + protectedSeqIt := 0 + recoveredPacketIt := 0 + + for protectedSeqIt < len(protectedSeqs) && recoveredPacketIt < len(d.recoveredPackets) { + switch { + case isNewerSeq(protectedSeqs[protectedSeqIt], d.recoveredPackets[recoveredPacketIt].SequenceNumber): + protectedPackets = append(protectedPackets, &protectedPacket{ + seq: protectedSeqs[protectedSeqIt], + packet: nil, + }) + protectedSeqIt++ + case isNewerSeq(d.recoveredPackets[recoveredPacketIt].SequenceNumber, protectedSeqs[protectedSeqIt]): + recoveredPacketIt++ + default: + protectedPackets = append(protectedPackets, &protectedPacket{ + seq: protectedSeqs[protectedSeqIt], + packet: &d.recoveredPackets[recoveredPacketIt], + }) + protectedSeqIt++ + recoveredPacketIt++ + } + } + + for protectedSeqIt < len(protectedSeqs) { + protectedPackets = append(protectedPackets, &protectedPacket{ + seq: protectedSeqs[protectedSeqIt], + packet: nil, + }) + protectedSeqIt++ + } + + d.receivedFECPackets = append(d.receivedFECPackets, fecPacketState{ + packet: fecPkt, + flexFec: fec, + protectedPackets: protectedPackets, + }) + + sort.Slice(d.receivedFECPackets, func(i, j int) bool { + return isNewerSeq(d.receivedFECPackets[i].packet.SequenceNumber, d.receivedFECPackets[j].packet.SequenceNumber) + }) + + if len(d.receivedFECPackets) > d.maxFECPackets { + d.receivedFECPackets = d.receivedFECPackets[1:] + } +} + +func (d *Decoder) attemptRecovery() []rtp.Packet { + recoveredPackets := make([]rtp.Packet, 0) + for { + packetsRecovered := 0 + for i := range d.receivedFECPackets { + fecPkt := d.receivedFECPackets[i] + packetsMissing := 0 + for _, pkt := range fecPkt.protectedPackets { + if pkt.packet == nil { + packetsMissing++ + if packetsMissing > 1 { + break + } + } + } + + if packetsMissing != 1 { + continue + } + + recovered, err := d.recoverPacket(&fecPkt) + if err != nil { + continue + } + + recoveredPackets = append(recoveredPackets, recovered) + d.recoveredPackets = append(d.recoveredPackets, recovered) + sort.Slice(d.recoveredPackets, func(i, j int) bool { + return isNewerSeq(d.recoveredPackets[i].SequenceNumber, d.recoveredPackets[j].SequenceNumber) + }) + + d.updateCoveringFecPackets(recovered) + d.discardOldRecoveredPackets() + packetsRecovered++ + } + + if packetsRecovered == 0 { + break + } + } + + return recoveredPackets +} + +func (d *Decoder) recoverPacket(fec *fecPacketState) (rtp.Packet, error) { + // https://datatracker.ietf.org/doc/html/draft-ietf-payload-flexible-fec-scheme-03#section-6.3.2 + + // extract the FEC bit string as the first 80 bits of the FEC header. + headerRecovery := make([]byte, 12) + if len(fec.packet.Payload) < 10 { + return rtp.Packet{}, errPacketTruncated + } + copy(headerRecovery, fec.packet.Payload[:10]) + + var seqnum uint16 + for _, protectedPacket := range fec.protectedPackets { + if protectedPacket.packet != nil { + // for each received source packet, compute the 80-bit string by + // concatenating the first 64 bits of its RTP header and the 16-bit + // network-ordered representation of its length in bytes minus 12. + receivedHeader, err := protectedPacket.packet.Header.Marshal() + if err != nil { + return rtp.Packet{}, fmt.Errorf("marshal received header: %w", err) + } + binary.BigEndian.PutUint16(receivedHeader[2:4], uint16(protectedPacket.packet.MarshalSize()-12)) + for i := range 8 { + headerRecovery[i] ^= receivedHeader[i] + } + } else { + seqnum = protectedPacket.seq + } + } + + // set version to 2 + headerRecovery[0] |= 0x80 + headerRecovery[0] &= 0xbf + payloadLength := binary.BigEndian.Uint16(headerRecovery[2:4]) + binary.BigEndian.PutUint16(headerRecovery[2:4], seqnum) + binary.BigEndian.PutUint32(headerRecovery[8:12], d.protectedStreamSSRC) + + payloadRecovery := make([]byte, payloadLength) + copy(payloadRecovery, fec.flexFec.payload) + for _, protectedPacket := range fec.protectedPackets { + if protectedPacket.packet != nil { + packet, err := protectedPacket.packet.Marshal() + if err != nil { + return rtp.Packet{}, fmt.Errorf("marshal protected packet: %w", err) + } + for i := 0; i < min(int(payloadLength), len(packet)-12); i++ { + payloadRecovery[i] ^= packet[12+i] + } + } + } + + headerRecovery = append(headerRecovery, payloadRecovery...) + + var packet rtp.Packet + if err := packet.Unmarshal(headerRecovery); err != nil { + return rtp.Packet{}, fmt.Errorf("unmarshal recovered: %w", err) + } + + return packet, nil +} + +func (d *Decoder) discardOldRecoveredPackets() { + if len(d.recoveredPackets) > recoveredPacketLimit { + d.recoveredPackets = d.recoveredPackets[len(d.recoveredPackets)-recoveredPacketLimit:] + } +} + +func decodeMask(mask uint64, bitCount uint16, seqNumBase uint16) []uint16 { + res := make([]uint16, 0) + for i := uint16(0); i < bitCount; i++ { + if (mask>>(bitCount-1-i))&1 == 1 { + res = append(res, seqNumBase+i) + } + } + + return res +} + +type fecPacketState struct { + packet rtp.Packet + flexFec flexFec + protectedPackets []*protectedPacket +} + +type flexFec struct { + protectedSSRC uint32 + seqNumBase uint16 + mask0 uint16 + mask1 uint32 + mask2 uint64 + payload []byte +} + +type protectedPacket struct { + seq uint16 + packet *rtp.Packet +} + +func parseFlexFEC03Header(data []byte) (flexFec, error) { + if len(data) < 20 { + return flexFec{}, fmt.Errorf("%w: length %d", errPacketTruncated, len(data)) + } + + rBit := (data[0] & 0x80) != 0 + if rBit { + return flexFec{}, errRetransmissionBitSet + } + + fBit := (data[0] & 0x40) != 0 + if fBit { + return flexFec{}, errInflexibleGeneratorMatrix + } + + ssrcCount := data[8] + if ssrcCount != 1 { + return flexFec{}, fmt.Errorf("%w: count %d", errMultipleSSRCProtection, ssrcCount) + } + + protectedSSRC := binary.BigEndian.Uint32(data[12:]) + seqNumBase := binary.BigEndian.Uint16(data[16:]) + rawPacketMask := data[18:] + var payload []byte + + kBit0 := (rawPacketMask[0] & 0x80) != 0 + maskPart0 := binary.BigEndian.Uint16(rawPacketMask[0:2]) & 0x7FFF + var maskPart1 uint32 + var maskPart2 uint64 + + if kBit0 { + payload = rawPacketMask[2:] + } else { + if len(data) < 24 { + return flexFec{}, fmt.Errorf("%w: length %d", errPacketTruncated, len(data)) + } + + kBit1 := (rawPacketMask[2] & 0x80) != 0 + maskPart1 = binary.BigEndian.Uint32(rawPacketMask[2:]) & 0x7FFFFFFF + + if kBit1 { + payload = rawPacketMask[6:] + } else { + if len(data) < 32 { + return flexFec{}, fmt.Errorf("%w: length %d", errPacketTruncated, len(data)) + } + + kBit2 := (rawPacketMask[6] & 0x80) != 0 + maskPart2 = binary.BigEndian.Uint64(rawPacketMask[6:]) & 0x7FFFFFFFFFFFFFFF + + if kBit2 { + payload = rawPacketMask[14:] + } else { + return flexFec{}, errLastOptionalMaskKBitSetToFalse + } + } + } + + return flexFec{ + protectedSSRC: protectedSSRC, + seqNumBase: seqNumBase, + mask0: maskPart0, + mask1: maskPart1, + mask2: maskPart2, + payload: payload, + }, nil +} + +func clonePacket(pkt rtp.Packet) rtp.Packet { + cloned := pkt + cloned.Header = pkt.Header.Clone() + if pkt.Payload != nil { + cloned.Payload = make([]byte, len(pkt.Payload)) + copy(cloned.Payload, pkt.Payload) + } + return cloned +} + +func seqDiff(a, b uint16) uint16 { + return min(a-b, b-a) +} + +func abs(x int) int { + if x >= 0 { + return x + } + + return -x +} + +func isNewerSeq(prevValue, value uint16) bool { + breakpoint := uint16(0x8000) + if value-prevValue == breakpoint { + return value > prevValue + } + + return value != prevValue && (value-prevValue) < breakpoint +} diff --git a/pkg/sfu/flexfec/decoder_test.go b/pkg/sfu/flexfec/decoder_test.go new file mode 100644 index 000000000..26e83f6f0 --- /dev/null +++ b/pkg/sfu/flexfec/decoder_test.go @@ -0,0 +1,173 @@ +// Copyright 2024 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package flexfec + +import ( + "testing" + + "github.com/pion/interceptor/pkg/flexfec" + "github.com/pion/rtp" + "github.com/stretchr/testify/require" +) + +const ( + testMediaSSRC = uint32(0x11223344) + testFECSSRC = uint32(0x55667788) + testFECPayloadT = uint8(49) + testMediaPT = uint8(96) +) + +func makeMediaPackets(baseSeq uint16, count int) []rtp.Packet { + pkts := make([]rtp.Packet, 0, count) + for i := 0; i < count; i++ { + // vary the payload length to exercise length recovery + payloadLen := 20 + (i % 5) + payload := make([]byte, payloadLen) + for j := range payload { + payload[j] = byte((i*7 + j*3) & 0xff) + } + pkts = append(pkts, rtp.Packet{ + Header: rtp.Header{ + Version: 2, + PayloadType: testMediaPT, + SequenceNumber: baseSeq + uint16(i), + Timestamp: uint32(1000 + i*960), + SSRC: testMediaSSRC, + }, + Payload: payload, + }) + } + return pkts +} + +func requirePacketsEqual(t *testing.T, want, got rtp.Packet) { + t.Helper() + require.Equal(t, want.SequenceNumber, got.SequenceNumber, "sequence number") + require.Equal(t, want.SSRC, got.SSRC, "ssrc") + require.Equal(t, want.PayloadType, got.PayloadType, "payload type") + require.Equal(t, want.Timestamp, got.Timestamp, "timestamp") + require.Equal(t, want.Payload, got.Payload, "payload") +} + +func TestDecoderRecoversSingleLossMediaThenFEC(t *testing.T) { + media := makeMediaPackets(1000, 10) + + encoder := flexfec.NewFlexEncoder03(testFECPayloadT, testFECSSRC) + fecPackets := encoder.EncodeFec(cloneAll(media), 1) + require.NotEmpty(t, fecPackets, "encoder should produce a FEC packet") + + dropped := 4 + dec := NewDecoder(testFECSSRC, testMediaSSRC) + + // deliver all media packets except the dropped one + for i, p := range media { + if i == dropped { + continue + } + require.Empty(t, dec.Decode(p), "no recovery expected while feeding media") + } + + // deliver the FEC packet - now exactly one protected packet is missing + var recovered []rtp.Packet + for _, fp := range fecPackets { + recovered = append(recovered, dec.Decode(fp)...) + } + + require.Len(t, recovered, 1) + requirePacketsEqual(t, media[dropped], recovered[0]) +} + +func TestDecoderRecoversWhenFECArrivesFirst(t *testing.T) { + media := makeMediaPackets(2000, 8) + + encoder := flexfec.NewFlexEncoder03(testFECPayloadT, testFECSSRC) + fecPackets := encoder.EncodeFec(cloneAll(media), 1) + require.NotEmpty(t, fecPackets) + + dropped := 2 + dec := NewDecoder(testFECSSRC, testMediaSSRC) + + // FEC arrives before any media: nothing can be recovered yet + for _, fp := range fecPackets { + require.Empty(t, dec.Decode(fp)) + } + + var recovered []rtp.Packet + for i, p := range media { + if i == dropped { + continue + } + recovered = append(recovered, dec.Decode(p)...) + } + + require.Len(t, recovered, 1) + requirePacketsEqual(t, media[dropped], recovered[0]) +} + +func TestDecoderNoLossNoRecovery(t *testing.T) { + media := makeMediaPackets(3000, 6) + + encoder := flexfec.NewFlexEncoder03(testFECPayloadT, testFECSSRC) + fecPackets := encoder.EncodeFec(cloneAll(media), 1) + require.NotEmpty(t, fecPackets) + + dec := NewDecoder(testFECSSRC, testMediaSSRC) + for _, p := range media { + require.Empty(t, dec.Decode(p)) + } + for _, fp := range fecPackets { + require.Empty(t, dec.Decode(fp), "no recovery expected when nothing is lost") + } +} + +func TestDecoderCannotRecoverDoubleLoss(t *testing.T) { + media := makeMediaPackets(4000, 10) + + encoder := flexfec.NewFlexEncoder03(testFECPayloadT, testFECSSRC) + fecPackets := encoder.EncodeFec(cloneAll(media), 1) + require.NotEmpty(t, fecPackets) + + dec := NewDecoder(testFECSSRC, testMediaSSRC) + // drop two packets covered by a single FEC packet -> unrecoverable + for i, p := range media { + if i == 3 || i == 6 { + continue + } + dec.Decode(p) + } + + var recovered []rtp.Packet + for _, fp := range fecPackets { + recovered = append(recovered, dec.Decode(fp)...) + } + require.Empty(t, recovered, "single FEC packet cannot recover two losses") +} + +func TestDecoderIgnoresUnrelatedSSRC(t *testing.T) { + dec := NewDecoder(testFECSSRC, testMediaSSRC) + pkt := rtp.Packet{ + Header: rtp.Header{Version: 2, SSRC: 0xdeadbeef, SequenceNumber: 1}, + Payload: []byte{1, 2, 3}, + } + require.Empty(t, dec.Decode(pkt)) +} + +func cloneAll(pkts []rtp.Packet) []rtp.Packet { + out := make([]rtp.Packet, len(pkts)) + for i, p := range pkts { + out[i] = clonePacket(p) + } + return out +} diff --git a/pkg/sfu/flexfec/encoder.go b/pkg/sfu/flexfec/encoder.go new file mode 100644 index 000000000..1c8c45ba4 --- /dev/null +++ b/pkg/sfu/flexfec/encoder.go @@ -0,0 +1,142 @@ +// Copyright 2024 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package flexfec + +import ( + "sync" + + "github.com/pion/interceptor" + pionflexfec "github.com/pion/interceptor/pkg/flexfec" + "github.com/pion/rtp" +) + +// EncoderInterceptorFactory builds EncoderInterceptors with the configured block size. +type EncoderInterceptorFactory struct { + numMediaPackets uint32 + numFecPackets uint32 +} + +// NewEncoderInterceptorFactory returns a factory for a FlexFEC-03 encoder interceptor +// that generates numFecPackets repair packets for every numMediaPackets media packets. +// +// It exists because pion's own flexfec.FecInterceptor is unsafe to use with LiveKit's +// pooled RTP payloads. That interceptor retains each media packet's payload slice *by +// reference* and only XORs the bytes once a full block has accumulated. LiveKit, however, +// returns the underlying payload buffer to a sync.Pool as soon as the packet has been +// written downstream (see pacer.Base.SendPacket -> Pool.Put). By the time pion computes +// the parity over a block, the earlier packets' buffers have already been recycled and +// overwritten, so the generated FEC protects garbage. Receivers that lose a protected +// packet then "recover" corrupt bytes and inject them into the media stream, causing +// frame corruption and PLI storms even at very low loss rates. +// +// This interceptor is behaviourally identical to pion's, except that it copies the +// payload when buffering so the parity is computed over the bytes that were actually +// sent. The copy is bounded by numMediaPackets per stream. +func NewEncoderInterceptorFactory(numMediaPackets, numFecPackets uint32) *EncoderInterceptorFactory { + return &EncoderInterceptorFactory{ + numMediaPackets: numMediaPackets, + numFecPackets: numFecPackets, + } +} + +// NewInterceptor constructs a new EncoderInterceptor. +func (f *EncoderInterceptorFactory) NewInterceptor(_ string) (interceptor.Interceptor, error) { + return &EncoderInterceptor{ + streams: make(map[uint32]*encoderStream), + numMediaPackets: f.numMediaPackets, + numFecPackets: f.numFecPackets, + }, nil +} + +type encoderStream struct { + mu sync.Mutex + encoder pionflexfec.FlexEncoder + packetBuffer []rtp.Packet +} + +// EncoderInterceptor generates a FlexFEC-03 repair stream for outgoing media. +type EncoderInterceptor struct { + interceptor.NoOp + mu sync.Mutex + streams map[uint32]*encoderStream + numMediaPackets uint32 + numFecPackets uint32 +} + +// UnbindLocalStream removes per-stream encoder state. +func (e *EncoderInterceptor) UnbindLocalStream(info *interceptor.StreamInfo) { + e.mu.Lock() + delete(e.streams, info.SSRC) + e.mu.Unlock() +} + +// BindLocalStream wraps the writer so FEC repair packets are emitted alongside media. +func (e *EncoderInterceptor) BindLocalStream( + info *interceptor.StreamInfo, writer interceptor.RTPWriter, +) interceptor.RTPWriter { + // No FEC negotiated for this stream; pass through untouched. + if info.PayloadTypeForwardErrorCorrection == 0 || info.SSRCForwardErrorCorrection == 0 { + return writer + } + + mediaSSRC := info.SSRC + + stream := &encoderStream{ + encoder: pionflexfec.FlexEncoder03Factory{}.NewEncoder( + info.PayloadTypeForwardErrorCorrection, + info.SSRCForwardErrorCorrection, + ), + } + + e.mu.Lock() + e.streams[mediaSSRC] = stream + e.mu.Unlock() + + return interceptor.RTPWriterFunc( + func(header *rtp.Header, payload []byte, attributes interceptor.Attributes) (int, error) { + // Only the protected media stream feeds the encoder. + if header.SSRC != mediaSSRC { + return writer.Write(header, payload, attributes) + } + + var fecPackets []rtp.Packet + stream.mu.Lock() + // Copy the payload: LiveKit hands us a pooled buffer that is recycled the + // moment this write returns, so it cannot be retained by reference. + payloadCopy := make([]byte, len(payload)) + copy(payloadCopy, payload) + stream.packetBuffer = append(stream.packetBuffer, rtp.Packet{ + Header: *header, + Payload: payloadCopy, + }) + if len(stream.packetBuffer) == int(e.numMediaPackets) { + fecPackets = stream.encoder.EncodeFec(stream.packetBuffer, e.numFecPackets) + stream.packetBuffer = nil + } + stream.mu.Unlock() + + result, err := writer.Write(header, payload, attributes) + + for i := range fecPackets { + fecHeader := fecPackets[i].Header + if _, fecErr := writer.Write(&fecHeader, fecPackets[i].Payload, attributes); fecErr != nil && err == nil { + err = fecErr + } + } + + return result, err + }, + ) +} diff --git a/pkg/sfu/flexfec/encoder_test.go b/pkg/sfu/flexfec/encoder_test.go new file mode 100644 index 000000000..06f31d394 --- /dev/null +++ b/pkg/sfu/flexfec/encoder_test.go @@ -0,0 +1,160 @@ +// Copyright 2024 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package flexfec + +import ( + "testing" + + "github.com/pion/interceptor" + pionflexfec "github.com/pion/interceptor/pkg/flexfec" + "github.com/pion/rtp" + "github.com/stretchr/testify/require" +) + +// recyclingWriter mimics LiveKit's pooled-payload lifecycle: it captures a deep copy of +// every packet it is asked to write (the bytes that are actually sent on the wire), then +// overwrites the caller-supplied payload buffer to emulate pacer.Base.SendPacket +// returning that buffer to the sync.Pool. Any encoder that retained the payload slice by +// reference will subsequently XOR garbage. +type recyclingWriter struct { + fecSSRC uint32 + media []rtp.Packet + fec []rtp.Packet +} + +func (w *recyclingWriter) Write(header *rtp.Header, payload []byte, _ interceptor.Attributes) (int, error) { + captured := make([]byte, len(payload)) + copy(captured, payload) + pkt := rtp.Packet{Header: *header, Payload: captured} + if header.SSRC == w.fecSSRC { + w.fec = append(w.fec, pkt) + } else { + w.media = append(w.media, pkt) + } + + // Recycle the caller's buffer, as the pacer does once the write returns. + for i := range payload { + payload[i] = 0xff + } + return len(payload), nil +} + +func fecStreamInfo() *interceptor.StreamInfo { + return &interceptor.StreamInfo{ + SSRC: testMediaSSRC, + SSRCForwardErrorCorrection: testFECSSRC, + PayloadTypeForwardErrorCorrection: testFECPayloadT, + } +} + +// drivePooled writes count media packets through wr. Each packet is backed by its own +// freshly allocated buffer (standing in for a pooled buffer) so that recyclingWriter can +// safely scribble over it after each write. +func drivePooled(wr interceptor.RTPWriter, baseSeq uint16, count int) { + for i := 0; i < count; i++ { + payloadLen := 20 + (i % 5) + buf := make([]byte, payloadLen) + for j := range buf { + buf[j] = byte((i*7 + j*3) & 0xff) + } + hdr := &rtp.Header{ + Version: 2, + PayloadType: testMediaPT, + SequenceNumber: baseSeq + uint16(i), + Timestamp: uint32(1000 + i*960), + SSRC: testMediaSSRC, + } + _, _ = wr.Write(hdr, buf, nil) + } +} + +// TestEncoderInterceptorRecoversWithPooledPayloads is the regression test for the FEC +// generation bug: our interceptor must copy the payload before retaining it, so that the +// parity it computes matches the bytes actually sent even though the caller recycles the +// payload buffer immediately after each write. +func TestEncoderInterceptorRecoversWithPooledPayloads(t *testing.T) { + const numMedia, numFec = 5, uint32(1) + + wr := &recyclingWriter{fecSSRC: testFECSSRC} + factory := NewEncoderInterceptorFactory(numMedia, numFec) + itc, err := factory.NewInterceptor("") + require.NoError(t, err) + writer := itc.BindLocalStream(fecStreamInfo(), interceptor.RTPWriterFunc(wr.Write)) + + drivePooled(writer, 1000, numMedia) + + require.Len(t, wr.media, numMedia, "all media packets forwarded") + require.Len(t, wr.fec, int(numFec), "one FEC packet generated for the block") + + // Drop one media packet, deliver the rest plus FEC, and confirm the recovered + // packet exactly matches the bytes that were sent. + const dropped = 2 + dec := NewDecoder(testFECSSRC, testMediaSSRC) + for i, p := range wr.media { + if i == dropped { + continue + } + require.Empty(t, dec.Decode(p)) + } + + var recovered []rtp.Packet + for _, fp := range wr.fec { + recovered = append(recovered, dec.Decode(fp)...) + } + + require.Len(t, recovered, 1, "FEC must recover the single dropped packet") + requirePacketsEqual(t, wr.media[dropped], recovered[0]) +} + +// TestPionEncoderCorruptsWithPooledPayloads documents the upstream behaviour that makes +// pion's own encoder interceptor unsafe with LiveKit's pooled payloads: because it +// retains the payload by reference, recycling the buffer corrupts the parity and the +// "recovered" packet does not match what was sent. This is the bug our interceptor fixes. +func TestPionEncoderCorruptsWithPooledPayloads(t *testing.T) { + const numMedia, numFec = 5, uint32(1) + + wr := &recyclingWriter{fecSSRC: testFECSSRC} + factory, err := pionflexfec.NewFecInterceptor( + pionflexfec.NumMediaPackets(numMedia), + pionflexfec.NumFECPackets(numFec), + ) + require.NoError(t, err) + itc, err := factory.NewInterceptor("") + require.NoError(t, err) + writer := itc.BindLocalStream(fecStreamInfo(), interceptor.RTPWriterFunc(wr.Write)) + + drivePooled(writer, 1000, numMedia) + require.Len(t, wr.media, numMedia) + require.Len(t, wr.fec, int(numFec)) + + const dropped = 2 + dec := NewDecoder(testFECSSRC, testMediaSSRC) + for i, p := range wr.media { + if i == dropped { + continue + } + dec.Decode(p) + } + var recovered []rtp.Packet + for _, fp := range wr.fec { + recovered = append(recovered, dec.Decode(fp)...) + } + + // pion still "recovers" a packet, but its payload is garbage because the parity was + // computed over recycled buffers. + require.Len(t, recovered, 1) + require.NotEqual(t, wr.media[dropped].Payload, recovered[0].Payload, + "pion encoder is expected to produce corrupt recovery with pooled payloads") +} diff --git a/scripts/flexfec/README.md b/scripts/flexfec/README.md new file mode 100644 index 000000000..63c7a36d5 --- /dev/null +++ b/scripts/flexfec/README.md @@ -0,0 +1,145 @@ +# Local FlexFEC-03 test harness + +Scripts to validate end-to-end FlexFEC-03 between the rust-sdks `local_video` +examples and a locally-built SFU, under simulated packet loss. + +FlexFEC is terminated at the SFU (it can't be passed through because the SFU +rewrites SSRC/sequence numbers per subscriber), so the two directions are +independent and tested separately: + +| Direction | What's exercised | Loss to inject | +| --------- | -------------------------------------------------- | -------------- | +| Downlink | SFU **generates** a repair stream per subscriber | `down` | +| Uplink | SFU **receives + decodes** a publisher repair flow | `up` | + +## Prerequisites + +- Go toolchain (to build this SFU). +- A `rust-sdks5` checkout with the FlexFEC changes, next to this repo + (`../rust-sdks5`) or pointed at via `RUST_SDKS_DIR`. Its `webrtc-sys` must be + built (the C++ field trials that enable FlexFEC-03 live there). +- A camera (the publisher captures real video). +- `sudo` for traffic shaping. macOS uses `dnctl`/`pfctl` (dummynet); Linux uses + `tc netem`. + +All processes run on one machine, so media flows over loopback. The SFU is pinned +to UDP `7882` (see `flexfec-local.yaml`) so shaping can target it by port. + +## Files + +| File | Purpose | +| -------------------- | ---------------------------------------------------- | +| `flexfec-local.yaml` | SFU config: flexfec on (pub+sub), fixed loopback port | +| `run-sfu.sh` | Build + run the SFU (`--dev`, devkey/secret) | +| `netem.sh` | Start/stop/inspect packet-loss shaping on udp/7882 | +| `run-publisher.sh` | Run the publisher with `--flex-fec` | +| `run-subscriber.sh` | Run the subscriber, surface `Video FEC ...` logs | + +## Workflow + +Use four terminals. + +**1 — SFU** + +```bash +scripts/flexfec/run-sfu.sh +``` + +Connects on `ws://127.0.0.1:7880` (API key `devkey` / secret `secret`). + +**2 — Subscriber** + +```bash +scripts/flexfec/run-subscriber.sh +# or only show FEC lines on the console: +FEC_ONLY=1 scripts/flexfec/run-subscriber.sh +``` + +**3 — Publisher** (grant camera permission on first run) + +```bash +scripts/flexfec/run-publisher.sh +``` + +**4 — Packet loss** (needs sudo) + +```bash +# Uplink test: drop 8% of publisher -> SFU packets +sudo scripts/flexfec/netem.sh start 8 up + +# Downlink test: drop 8% of SFU -> subscriber packets (+ optional delay_ms, see A/B below) +sudo scripts/flexfec/netem.sh start 8 down +sudo scripts/flexfec/netem.sh start 8 down 150 # 8% loss + 150ms one-way delay + +sudo scripts/flexfec/netem.sh status +sudo scripts/flexfec/netem.sh stop # always clean up when done +``` + +## What success looks like + +**Downlink (`down`)** — the subscriber recovers SFU-generated repair packets. +In the subscriber log (`logs/subscriber.log`): + +``` +Video FEC active: fec_ssrc=..., received_total=..., discarded_total=..., accepted_recovery_payload_total=... +Video FEC recovery payload accepted: +N packets (accepted_total=...) +``` + +`accepted_recovery_payload_total` should climb while loss is applied. + +**Uplink (`up`)** — the SFU receives + decodes the publisher's repair stream. +In the SFU log: + +``` +flexfec decoder enabled {"fecSSRC": ..., "mediaSSRC": ...} +flexfec recovery stats {"fecPacketsReceived": ..., "packetsRecovered": ..., "packetsRecoveredDelta": ...} +``` + +`packetsRecovered` increasing confirms the SFU rebuilt lost uplink packets. + +To also see FlexFEC SDP negotiation (`flexfec ... from sdp`, codec registration), +set `logging.level: debug` in `flexfec-local.yaml`. + +### Subscriber-side A/B (isolating FEC from RTX) + +On loopback RTT~=0, so NACK/RTX recovers nearly all loss before FlexFEC matters and +its benefit is invisible. To make FEC the primary recovery path, add one-way delay so +retransmissions arrive too late, and compare FEC on vs off under the same conditions: + +```bash +# A) FEC ON (flexfec.subscriber: true). With SFU + pub + sub running: +sudo scripts/flexfec/netem.sh start 8 down 150 # 8% loss + 150ms delay, downlink +# ...observe subscriber for ~20s, then... +sudo scripts/flexfec/netem.sh stop + +# B) FEC OFF: set flexfec.subscriber: false in flexfec-local.yaml, restart the SFU, +# reconnect the subscriber, then apply the SAME shaping and observe. +``` + +The subscriber logs a `Video quality:` line every ~3s: + +``` +Video quality: frames_decoded=..., frames_dropped=..., freezes=N (Xs), nack=..., pli=..., + packets_lost=..., rtx_recovered=..., fec_recv=..., fec_discarded=... +``` + +With FEC ON you expect fewer `freezes`/`frames_dropped`, lower net `packets_lost`, and a +meaningful `fec_recv`; with FEC OFF the same loss leans entirely on `rtx_recovered` and +(because of the delay) typically shows more freezes and dropped frames. + +## Notes & caveats + +- **Loopback shaping (macOS):** `netem.sh` loads its dummynet rules into a dedicated + `flexfec-loss` pf anchor and re-applies `/etc/pf.conf` non-destructively. `stop` + flushes the pipes and restores pf to its prior enabled/disabled state. If pf is set + to `skip on lo0` the script warns and shaping won't apply. +- **Direction isolation is approximate.** Both clients share the SFU UDP port, so + `up`/`down` are distinguished by source vs. destination port. The publisher is the + dominant sender to the SFU and the subscriber the dominant receiver, so this is good + enough to isolate each path in practice (some RTCP is also affected). +- **`dnctl -q flush` clears all dummynet pipes**, not just ours — fine on a dev box. +- **Send-side BWE:** if the SFU uses pacer-based send-side BWE, subscriber FEC overhead + isn't accounted for by congestion control (logged as a warning by the SFU). The + interceptor-based send-side BWE path TWCC-tags FEC packets correctly. +- Increase loss gradually (e.g. 3% → 10%). Beyond what a single FlexFEC repair packet + can cover, bursts become unrecoverable (expected). diff --git a/scripts/flexfec/flexfec-linux.yaml b/scripts/flexfec/flexfec-linux.yaml new file mode 100644 index 000000000..e6b58b86b --- /dev/null +++ b/scripts/flexfec/flexfec-linux.yaml @@ -0,0 +1,45 @@ +# Linux FlexFEC-03 test config for the netns + netem harness (scripts/flexfec/netns.sh). +# +# Unlike flexfec-local.yaml (macOS, loopback + in-SFU software impairment), this runs the +# SFU in the root netns and advertises the veth host IP 10.200.0.1 so that: +# - the publisher in the `robot` netns reaches the SFU over the veth pair (shaped by netem) +# - the subscriber in the root netns reaches the SFU at 10.200.0.1 locally (clean link) +# +# Run with --dev (injects devkey/secret): +# ~/go1.26/bin/... # build first, then: +# sudo scripts/flexfec/netns.sh up +# /tmp/flexfec-livekit-server --dev --config scripts/flexfec/flexfec-linux.yaml + +port: 7880 + +# Bind signalling (HTTP/WS + TCP) to the veth host IP so the publisher in the `robot` netns +# can reach it, plus loopback for the in-root-ns subscriber. Setting a non-loopback bind +# address also stops --dev from pinning everything to 127.0.0.1. +bind_addresses: + - 10.200.0.1 + - 127.0.0.1 + +rtc: + udp_port: 7882 + tcp_port: 7881 + # Advertise the veth host IP, not a public IP. + use_external_ip: false + enable_loopback_candidate: true + ips: + includes: + # veth subnet: 10.200.0.1 (SFU/root) <-> 10.200.0.2 (robot netns) + - 10.200.0.0/24 + - 127.0.0.0/8 + + # Asymmetric teleop topology: robot publishes with FlexFEC (uplink recovery at the SFU); + # the operator/subscriber is on a clean wired link so downlink FEC is off. + flexfec: + subscriber: false + publisher: true + payload_type: 49 + num_media_packets: 5 + num_fec_packets: 1 + +logging: + level: debug + json: false diff --git a/scripts/flexfec/flexfec-local.yaml b/scripts/flexfec/flexfec-local.yaml new file mode 100644 index 000000000..919c3c38e --- /dev/null +++ b/scripts/flexfec/flexfec-local.yaml @@ -0,0 +1,48 @@ +# Local FlexFEC-03 test config for the LiveKit SFU. +# +# Intended to be run together with `--dev` (which injects the devkey/secret API key +# pair and binds to loopback): +# +# go run ./cmd/server --dev --config scripts/flexfec/flexfec-local.yaml +# +# Media is pinned to a single UDP port on loopback so traffic shaping (see netem.sh) +# can target it deterministically. + +port: 7880 + +rtc: + # Single, fixed media port (no port_range_* so this takes effect). Shape this port + # with scripts/flexfec/netem.sh to simulate packet loss. + udp_port: 7882 + tcp_port: 7881 + # Keep everything on the local machine; do not try to discover a public IP. + use_external_ip: false + # Gather loopback (127.0.0.1) candidates and pin ICE to IPv4 loopback so all media flows + # over a single, stable path. NOTE: on an all-on-one-box macOS setup, pf/dummynet + # (netem.sh) cannot shape this traffic -- the kernel short-circuits same-host UDP delivery + # before the pf hook -- so loss/delay is injected in-SFU instead via the LK_DOWNLINK_* + # env vars (see pkg/sfu/downtrack_impair.go and run-sfu.sh). The media path is therefore + # irrelevant to shaping, so loopback (no dependency on a volatile LAN IP) is preferred. + enable_loopback_candidate: true + ips: + includes: + - 127.0.0.0/8 + + # FlexFEC-03 for both directions: + # publisher: receive + decode a publisher's repair stream (uplink recovery) + # subscriber: generate a repair stream per subscriber (downlink recovery) + flexfec: + subscriber: false + publisher: true + payload_type: 49 + # Low-latency teleop tuning: 5-packet block with 1 repair = 20% overhead, but the + # repair lands ~16ms after the block (vs ~32ms for a 10-block), well under an 80ms + # RTX round trip. Recovers 1 scattered loss per 5-block (ample for ~1% random loss). + num_media_packets: 5 + num_fec_packets: 1 + +logging: + # info surfaces the "flexfec decoder enabled" and periodic "flexfec recovery stats" + # logs. Bump to debug to also see SDP-level FEC negotiation ("flexfec ... from sdp"). + level: debug + json: false diff --git a/scripts/flexfec/netem.sh b/scripts/flexfec/netem.sh new file mode 100755 index 000000000..1b679cb61 --- /dev/null +++ b/scripts/flexfec/netem.sh @@ -0,0 +1,289 @@ +#!/usr/bin/env bash +# +# Simulate packet loss on the SFU's media UDP port so FlexFEC recovery can be +# observed end-to-end. Works on macOS (dnctl/dummynet + pfctl) and Linux (tc netem). +# +# Because publisher, SFU and subscriber all run on one machine, media flows over the +# loopback interface. Loss is matched by the SFU UDP port (default 7882): +# - "down" drops packets coming FROM the SFU port -> SFU -> subscriber path +# (exercises subscriber-side / downlink FEC generation+recovery) +# - "up" drops packets going TO the SFU port -> publisher -> SFU path +# (exercises publisher-side / uplink FEC reception+recovery) +# - "both" drops in both directions +# +# Usage: +# sudo scripts/flexfec/netem.sh start [loss_percent] [up|down|both] [delay_ms] # uniform loss; default: 5 down 0 +# sudo scripts/flexfec/netem.sh burst [loss_percent] [burst_ms] [gap_ms] [up|down|both] [delay_ms] # bursty loss +# sudo scripts/flexfec/netem.sh stop +# sudo scripts/flexfec/netem.sh status +# +# delay_ms adds one-way latency (so RTT ~= 2*delay_ms on loopback). Models the +# robot's access-network round trip; RTX recovery costs ~1 RTT of buffering. +# +# "burst" emulates handoff/fade-style correlated loss: it toggles the loss rate +# between loss_percent (for burst_ms) and 0 (for gap_ms) on a duty cycle, holding +# the delay constant. Average loss ~= loss_percent * burst_ms/(burst_ms+gap_ms). +# Run it in its own terminal; Ctrl-C stops and restores the network. +# This is where RTX (retransmit storms, each costing an RTT) and FlexFEC (fixed +# inline overhead) diverge most. +# +# Env overrides: +# SFU_UDP_PORT media port to shape (default 7882) +# NETEM_IFACE (Linux only) interface to shape (default lo) +set -euo pipefail + +PORT="${SFU_UDP_PORT:-7882}" +ANCHOR="flexfec-loss" +STATE_DIR="${TMPDIR:-/tmp}/flexfec-netem" +PF_STATE="$STATE_DIR/pf.enabled" +OS="$(uname -s)" + +require_root() { + if [[ "$(id -u)" -ne 0 ]]; then + echo "error: this command needs root; re-run with sudo" >&2 + exit 1 + fi +} + +validate_dir() { + case "$1" in + up | down | both) ;; + *) + echo "error: direction must be up|down|both (got '$1')" >&2 + exit 1 + ;; + esac +} + +# ----------------------------------------------------------------------------- macOS + +# Install the pf anchor + dummynet rules for the requested direction. Pipes are +# (re)configured separately so loss can be toggled live for burst mode. +macos_setup() { + local dir="$1" + + mkdir -p "$STATE_DIR" + + # Remember whether pf was already enabled so `stop` can restore it. + if [[ ! -f "$PF_STATE" ]]; then + if pfctl -s info 2>/dev/null | grep -q 'Status: Enabled'; then + echo enabled >"$PF_STATE" + else + echo disabled >"$PF_STATE" + fi + fi + + local rules="" + case "$dir" in + down | both) rules+="dummynet out proto udp from any port $PORT to any pipe 1"$'\n' ;; + esac + case "$dir" in + up | both) rules+="dummynet out proto udp from any to any port $PORT pipe 2"$'\n' ;; + esac + + # Re-load the system ruleset plus references to our named anchor (non-destructive: + # existing /etc/pf.conf rules are preserved), then install our rules in the anchor. + { + cat /etc/pf.conf 2>/dev/null || true + echo "dummynet-anchor \"$ANCHOR\"" + echo "anchor \"$ANCHOR\"" + } | pfctl -f - 2>/dev/null + printf '%s' "$rules" | pfctl -a "$ANCHOR" -f - 2>/dev/null + pfctl -e 2>/dev/null || true + + if pfctl -sa 2>/dev/null | grep -qi 'skip on lo'; then + echo "warning: pf is configured to 'skip on lo'; loopback shaping may not apply" >&2 + fi +} + +# macos_set_loss +macos_set_loss() { + local plr + plr="$(awk "BEGIN { printf \"%.4f\", $1 / 100 }")" + dnctl pipe 1 config plr "$plr" delay "$2" >/dev/null + dnctl pipe 2 config plr "$plr" delay "$2" >/dev/null +} + +macos_start() { + local loss="$1" dir="$2" delay="$3" + macos_setup "$dir" + macos_set_loss "$loss" "$delay" + echo "==> macOS dummynet loss enabled: ${loss}% loss, ${delay}ms delay (${dir}) on udp/${PORT}" +} + +macos_burst() { + local loss="$1" burst_ms="$2" gap_ms="$3" dir="$4" delay="$5" + local burst_s gap_s avg + burst_s="$(awk "BEGIN { printf \"%.3f\", $burst_ms / 1000 }")" + gap_s="$(awk "BEGIN { printf \"%.3f\", $gap_ms / 1000 }")" + avg="$(awk "BEGIN { printf \"%.1f\", $loss * $burst_ms / ($burst_ms + $gap_ms) }")" + + macos_setup "$dir" + macos_set_loss 0 "$delay" + + trap 'echo; echo "==> stopping burst"; macos_stop; exit 0' INT TERM + echo "==> macOS bursty loss: ${loss}% for ${burst_ms}ms every $((burst_ms + gap_ms))ms (~${avg}% avg), ${delay}ms delay (${dir}) on udp/${PORT}" + echo " Ctrl-C to stop and restore the network." + while true; do + macos_set_loss "$loss" "$delay" + printf '\r [burst ON %s%% loss] ' "$loss" + sleep "$burst_s" + macos_set_loss 0 "$delay" + printf '\r [burst off 0%% loss] ' "$loss" + sleep "$gap_s" + done +} + +macos_stop() { + dnctl -q flush 2>/dev/null || true + printf '' | pfctl -a "$ANCHOR" -f - 2>/dev/null || true + pfctl -f /etc/pf.conf 2>/dev/null || true + if [[ -f "$PF_STATE" && "$(cat "$PF_STATE")" == "disabled" ]]; then + pfctl -d 2>/dev/null || true + fi + rm -f "$PF_STATE" + echo "==> macOS dummynet loss disabled" +} + +macos_status() { + echo "== dummynet pipes ==" + dnctl list 2>/dev/null || true + echo "== pf anchor '$ANCHOR' ==" + pfctl -a "$ANCHOR" -s rules 2>/dev/null || true + echo "== pf status ==" + pfctl -s info 2>/dev/null | head -1 || true +} + +# ----------------------------------------------------------------------------- Linux + +# linux_setup : build prio qdisc + netem band (loss starts at 0) +# and funnel matching UDP packets into it. +linux_setup() { + local dir="$1" delay="$2" + local dev="${NETEM_IFACE:-lo}" + + local netem_args=(loss 0%) + if [[ "$delay" -gt 0 ]]; then + netem_args+=(delay "${delay}ms") + fi + + tc qdisc del dev "$dev" root 2>/dev/null || true + tc qdisc add dev "$dev" root handle 1: prio + tc qdisc add dev "$dev" parent 1:3 handle 30: netem "${netem_args[@]}" + + if [[ "$dir" == "down" || "$dir" == "both" ]]; then + tc filter add dev "$dev" parent 1:0 protocol ip prio 3 u32 \ + match ip protocol 17 0xff match ip sport "$PORT" 0xffff flowid 1:3 + fi + if [[ "$dir" == "up" || "$dir" == "both" ]]; then + tc filter add dev "$dev" parent 1:0 protocol ip prio 3 u32 \ + match ip protocol 17 0xff match ip dport "$PORT" 0xffff flowid 1:3 + fi +} + +# linux_set_loss +linux_set_loss() { + local dev="${NETEM_IFACE:-lo}" + if [[ "$2" -gt 0 ]]; then + tc qdisc change dev "$dev" parent 1:3 handle 30: netem loss "${1}%" delay "${2}ms" + else + tc qdisc change dev "$dev" parent 1:3 handle 30: netem loss "${1}%" + fi +} + +linux_start() { + local loss="$1" dir="$2" delay="$3" + local dev="${NETEM_IFACE:-lo}" + linux_setup "$dir" "$delay" + linux_set_loss "$loss" "$delay" + echo "==> Linux netem enabled: ${loss}% loss, ${delay}ms delay (${dir}) on udp/${PORT} dev ${dev}" +} + +linux_burst() { + local loss="$1" burst_ms="$2" gap_ms="$3" dir="$4" delay="$5" + local dev="${NETEM_IFACE:-lo}" + local burst_s gap_s avg + burst_s="$(awk "BEGIN { printf \"%.3f\", $burst_ms / 1000 }")" + gap_s="$(awk "BEGIN { printf \"%.3f\", $gap_ms / 1000 }")" + avg="$(awk "BEGIN { printf \"%.1f\", $loss * $burst_ms / ($burst_ms + $gap_ms) }")" + + linux_setup "$dir" "$delay" + + trap 'echo; echo "==> stopping burst"; linux_stop; exit 0' INT TERM + echo "==> Linux bursty loss: ${loss}% for ${burst_ms}ms every $((burst_ms + gap_ms))ms (~${avg}% avg), ${delay}ms delay (${dir}) on udp/${PORT} dev ${dev}" + echo " Ctrl-C to stop and restore the network." + while true; do + linux_set_loss "$loss" "$delay" + printf '\r [burst ON %s%% loss] ' "$loss" + sleep "$burst_s" + linux_set_loss 0 "$delay" + printf '\r [burst off 0%% loss] ' "$loss" + sleep "$gap_s" + done +} + +linux_stop() { + local dev="${NETEM_IFACE:-lo}" + tc qdisc del dev "$dev" root 2>/dev/null || true + echo "==> Linux netem loss disabled (dev ${dev})" +} + +linux_status() { + local dev="${NETEM_IFACE:-lo}" + echo "== qdisc (dev ${dev}) ==" + tc qdisc show dev "$dev" || true + echo "== filters (dev ${dev}) ==" + tc filter show dev "$dev" || true +} + +# ----------------------------------------------------------------------------- dispatch + +cmd="${1:-}" + +case "$cmd" in + start) + require_root + loss="${2:-5}" + dir="${3:-down}" + delay="${4:-0}" + validate_dir "$dir" + case "$OS" in + Darwin) macos_start "$loss" "$dir" "$delay" ;; + Linux) linux_start "$loss" "$dir" "$delay" ;; + *) echo "error: unsupported OS '$OS'" >&2; exit 1 ;; + esac + ;; + burst) + require_root + loss="${2:-40}" + burst_ms="${3:-200}" + gap_ms="${4:-2000}" + dir="${5:-down}" + delay="${6:-20}" + validate_dir "$dir" + case "$OS" in + Darwin) macos_burst "$loss" "$burst_ms" "$gap_ms" "$dir" "$delay" ;; + Linux) linux_burst "$loss" "$burst_ms" "$gap_ms" "$dir" "$delay" ;; + *) echo "error: unsupported OS '$OS'" >&2; exit 1 ;; + esac + ;; + stop) + require_root + case "$OS" in + Darwin) macos_stop ;; + Linux) linux_stop ;; + esac + ;; + status) + case "$OS" in + Darwin) macos_status ;; + Linux) linux_status ;; + esac + ;; + *) + echo "usage: sudo $0 {start [loss%] [up|down|both] [delay_ms]" >&2 + echo " | burst [loss%] [burst_ms] [gap_ms] [up|down|both] [delay_ms]" >&2 + echo " | stop | status}" >&2 + exit 1 + ;; +esac diff --git a/scripts/flexfec/netns.sh b/scripts/flexfec/netns.sh new file mode 100644 index 000000000..4828abb72 --- /dev/null +++ b/scripts/flexfec/netns.sh @@ -0,0 +1,131 @@ +#!/usr/bin/env bash +# +# Linux network-namespace + netem harness for the asymmetric teleop FlexFEC test. +# +# Topology (models a robot on a lossy 5G uplink talking to a wired operator): +# +# robot netns (publisher) root netns (SFU + subscriber/operator) +# 10.200.0.2 veth-robot <=========> veth-host 10.200.0.1 [SFU :7880/:7882] +# netem here [subscriber -> 10.200.0.1] +# +# Only the robot<->SFU leg traverses the veth pair, so netem on the veth shapes the +# robot uplink/downlink ONLY. The subscriber (operator) sits in the root netns and reaches +# the SFU at 10.200.0.1 locally, so its link stays clean -- exactly the teleop case where +# the robot is on 5G and the operator is on a stable hardline. +# +# Loss is applied to the robot->SFU (uplink) direction; delay is applied symmetrically so a +# NACK->RTX recovery costs a full RTT (the latency publisher-side FlexFEC avoids by letting +# the SFU repair the stream inline). +# +# Usage: +# sudo scripts/flexfec/netns.sh up # create ns + veth (no impairment) +# sudo scripts/flexfec/netns.sh shape # apply/replace netem +# sudo scripts/flexfec/netns.sh clear # remove netem (clean link) +# sudo scripts/flexfec/netns.sh status # show qdiscs + addrs +# sudo scripts/flexfec/netns.sh down # tear everything down +# scripts/flexfec/netns.sh exec # run cmd inside robot netns (as $SUDO_USER) +# +# Example: +# sudo scripts/flexfec/netns.sh up +# sudo scripts/flexfec/netns.sh shape 3 20 # 3% uplink loss, 20ms each way (40ms RTT) +set -euo pipefail + +NS=robot +VETH_HOST=veth-host +VETH_ROBOT=veth-robot +HOST_IP=10.200.0.1 +ROBOT_IP=10.200.0.2 +PREFIX=24 + +# Privileged commands are run via passwordless sudo for ip/tc only (see the sudoers drop-in +# in the harness docs). The script itself does NOT need to run as root. +if [[ "$(id -u)" == "0" ]]; then + SUDO="" +else + SUDO="sudo -n" +fi + +cmd_up() { + # Idempotent: tear down any prior instance first. + $SUDO ip netns del "$NS" 2>/dev/null || true + $SUDO ip link del "$VETH_HOST" 2>/dev/null || true + + $SUDO ip netns add "$NS" + $SUDO ip link add "$VETH_HOST" type veth peer name "$VETH_ROBOT" + $SUDO ip link set "$VETH_ROBOT" netns "$NS" + + $SUDO ip addr add "$HOST_IP/$PREFIX" dev "$VETH_HOST" + $SUDO ip link set "$VETH_HOST" up + + $SUDO ip netns exec "$NS" ip addr add "$ROBOT_IP/$PREFIX" dev "$VETH_ROBOT" + $SUDO ip netns exec "$NS" ip link set "$VETH_ROBOT" up + $SUDO ip netns exec "$NS" ip link set lo up + + echo "up: $NS netns ready" + echo " root: $VETH_HOST $HOST_IP/$PREFIX robot: $VETH_ROBOT $ROBOT_IP/$PREFIX" + echo " SFU should advertise $HOST_IP (use flexfec-linux.yaml); publisher connects to ws://$HOST_IP:7880" +} + +# shape [jitter_ms] [corr_pct] +# When jitter_ms/corr_pct are given they feed netem's loss correlation + delay jitter to +# approximate bursty (e.g. 5G handoff) conditions rather than uniform random loss. +cmd_shape() { + local loss="${1:?usage: shape [jitter_ms] [loss_corr%]}" + local owd="${2:?usage: shape [jitter_ms] [loss_corr%]}" + local jitter="${3:-0}" + local corr="${4:-0}" + + local delayspec="delay ${owd}ms" + [[ "$jitter" != "0" ]] && delayspec="delay ${owd}ms ${jitter}ms distribution normal" + local lossspec="loss ${loss}%" + [[ "$corr" != "0" ]] && lossspec="loss ${loss}% ${corr}%" + + # Uplink (robot -> SFU): loss + delay. Applied on robot-side egress. + $SUDO ip netns exec "$NS" tc qdisc replace dev "$VETH_ROBOT" root netem $delayspec $lossspec + # Downlink (SFU -> robot, carries NACK/RTCP/PLI): delay only, no loss (operator link is + # clean, but the return path still costs propagation delay so RTX pays a full RTT). + $SUDO tc qdisc replace dev "$VETH_HOST" root netem delay "${owd}ms" + + echo "shape: uplink ${lossspec} ${delayspec} ; downlink delay=${owd}ms (RTT≈$((owd*2))ms)" +} + +cmd_clear() { + $SUDO ip netns exec "$NS" tc qdisc del dev "$VETH_ROBOT" root 2>/dev/null || true + $SUDO tc qdisc del dev "$VETH_HOST" root 2>/dev/null || true + echo "clear: netem removed (clean link)" +} + +cmd_status() { + echo "=== root ns: $VETH_HOST ===" + $SUDO ip addr show "$VETH_HOST" 2>/dev/null | sed -n '2,4p' || echo " (absent)" + $SUDO tc qdisc show dev "$VETH_HOST" 2>/dev/null || true + echo "=== robot ns: $VETH_ROBOT ===" + $SUDO ip netns exec "$NS" ip addr show "$VETH_ROBOT" 2>/dev/null | sed -n '2,4p' || echo " (absent)" + $SUDO ip netns exec "$NS" tc qdisc show dev "$VETH_ROBOT" 2>/dev/null || true +} + +cmd_down() { + $SUDO ip netns del "$NS" 2>/dev/null || true + $SUDO ip link del "$VETH_HOST" 2>/dev/null || true + echo "down: torn down" +} + +# exec : run inside robot netns, dropping back to the invoking user. The outer +# `sudo ip netns exec` runs as root; the inner `sudo -u $user` drops privileges (root -> user +# needs no password), so the publisher runs as the normal user with its env intact. +cmd_exec() { + local user="${SUDO_USER:-$USER}" + exec $SUDO ip netns exec "$NS" sudo -u "$user" --preserve-env=LIVEKIT_URL,LIVEKIT_API_KEY,LIVEKIT_API_SECRET,RUST_LOG,ROOM,PUB_IDENTITY,PATH,HOME,CARGO_HOME,RUSTUP_HOME "$@" +} + +action="${1:-}" +shift || true +case "$action" in + up) cmd_up "$@" ;; + shape) cmd_shape "$@" ;; + clear) cmd_clear "$@" ;; + status) cmd_status "$@" ;; + down) cmd_down "$@" ;; + exec) cmd_exec "$@" ;; + *) echo "usage: $0 {up|shape |clear|status|down|exec }" >&2; exit 2 ;; +esac diff --git a/scripts/flexfec/run-leg-linux.sh b/scripts/flexfec/run-leg-linux.sh new file mode 100644 index 000000000..168560105 --- /dev/null +++ b/scripts/flexfec/run-leg-linux.sh @@ -0,0 +1,133 @@ +#!/usr/bin/env bash +# +# Run one FlexFEC test "leg" on the Linux box: start the SFU, a publisher (robot), and a +# subscriber (operator), let them run, then print a summary of SFU FEC recovery + the +# subscriber's frame-latency / video-quality stats. +# +# Two modes: +# - loopback (default): everything on 127.0.0.1, no impairment. Smoke test. +# - netns (USE_NETNS=1): publisher runs inside the `robot` netns (set up + shaped via +# netns.sh), SFU + subscriber run in the root netns reaching the SFU at 10.200.0.1. +# This models a lossy robot uplink with a clean operator downlink. +# +# Env: +# LABEL leg label for the banner +# DURATION seconds to run (default 35) +# USE_NETNS 1 to run publisher in the robot netns (default 0 = loopback) +# SFU_BIN path to SFU binary (default /tmp/flexfec-livekit-server) +# RUST_DIR rust-sdks5 dir (default ~/workspace/rust-sdks5) +# LK_DIR livekit-flexfec dir (default ~/workspace/livekit-flexfec) +# PUB_EXTRA extra publisher args (e.g. "--max-playout-delay 30") +set -uo pipefail + +LABEL="${LABEL:-leg}" +DURATION="${DURATION:-35}" +USE_NETNS="${USE_NETNS:-0}" +SFU_BIN="${SFU_BIN:-/tmp/flexfec-livekit-server}" +RUST_DIR="${RUST_DIR:-$HOME/workspace/rust-sdks5}" +LK_DIR="${LK_DIR:-$HOME/workspace/livekit-flexfec}" +PUB_EXTRA="${PUB_EXTRA:-}" + +if [[ "$USE_NETNS" == "1" ]]; then + SFU_HOST="10.200.0.1" +else + SFU_HOST="127.0.0.1" +fi + +# FEC=on (default) uses flexfec-linux.yaml (publisher: true). FEC=off writes a sibling +# config with publisher: false so the SFU does not negotiate uplink FlexFEC and the robot +# falls back to RTX-only -- the A/B baseline. +BASE_CONFIG="$LK_DIR/scripts/flexfec/flexfec-linux.yaml" +if [[ "${FEC:-on}" == "off" ]]; then + CONFIG="/tmp/flexfec-linux-fecoff.yaml" + sed -E 's/^( *publisher:) *true/\1 false/' "$BASE_CONFIG" > "$CONFIG" +else + CONFIG="$BASE_CONFIG" +fi + +LOGDIR=/tmp/flexfec-logs +mkdir -p "$LOGDIR" +SFU_LOG="$LOGDIR/sfu.log" +PUB_LOG="$LOGDIR/pub.log" +SUB_LOG="$LOGDIR/sub.log" + +cleanup() { + pkill -f flexfec-livekit-server 2>/dev/null + pkill -f "target/debug/publisher" 2>/dev/null + pkill -f "target/debug/subscriber" 2>/dev/null +} +cleanup +sleep 1 + +echo "==================== LEG: $LABEL (netns=$USE_NETNS, ${DURATION}s) ====================" + +# In netns mode, apply netem shaping to the robot uplink. NETEM_LOSS/% NETEM_OWD ms, plus +# optional NETEM_JITTER ms and NETEM_CORR % for bursty (5G-like) loss/jitter. +if [[ "$USE_NETNS" == "1" ]]; then + NETEM_LOSS="${NETEM_LOSS:-0}" + NETEM_OWD="${NETEM_OWD:-0}" + if [[ "$NETEM_LOSS" != "0" || "$NETEM_OWD" != "0" ]]; then + "$LK_DIR/scripts/flexfec/netns.sh" shape "$NETEM_LOSS" "$NETEM_OWD" "${NETEM_JITTER:-0}" "${NETEM_CORR:-0}" + else + "$LK_DIR/scripts/flexfec/netns.sh" clear + fi +fi + +# --- SFU (root netns) --- +# In-SFU software impairment (used when netem/root is unavailable). All no-ops unless set. +# PUB_LOSS / PUB_DELAY_MS robot uplink (publisher->SFU) loss + one-way delay +# DOWNLINK_LOSS / DOWNLINK_DELAY_MS / UPLINK_DELAY_MS operator downlink + NACK delay +( cd "$LK_DIR" && \ + LK_PUB_LOSS="${PUB_LOSS:-0}" LK_PUB_DELAY_MS="${PUB_DELAY_MS:-0}" \ + LK_DOWNLINK_LOSS="${DOWNLINK_LOSS:-0}" LK_DOWNLINK_DELAY_MS="${DOWNLINK_DELAY_MS:-0}" \ + LK_UPLINK_DELAY_MS="${UPLINK_DELAY_MS:-0}" \ + "$SFU_BIN" --dev --config "$CONFIG" > "$SFU_LOG" 2>&1 ) & +sleep 3 + +export LIVEKIT_URL="ws://$SFU_HOST:7880" +export LIVEKIT_API_KEY=devkey +export LIVEKIT_API_SECRET=secret +export RUST_LOG="${RUST_LOG:-info}" +ROOM=flexfec-test + +PUB="$RUST_DIR/target/debug/publisher" +SUB="$RUST_DIR/target/debug/subscriber" + +# --- Publisher (robot) --- +PUB_ARGS=(--flex-fec --room-name "$ROOM" --identity flexfec-pub --test-pattern --animate-test-pattern --attach-timestamp --attach-frame-id) +# shellcheck disable=SC2206 +[[ -n "$PUB_EXTRA" ]] && PUB_ARGS+=($PUB_EXTRA) + +if [[ "$USE_NETNS" == "1" ]]; then + # netns.sh sudoes ip/tc internally; do NOT wrap it in sudo (the script isn't allowlisted). + "$LK_DIR/scripts/flexfec/netns.sh" exec env \ + LIVEKIT_URL="$LIVEKIT_URL" LIVEKIT_API_KEY=devkey LIVEKIT_API_SECRET=secret RUST_LOG="$RUST_LOG" \ + "$PUB" "${PUB_ARGS[@]}" > "$PUB_LOG" 2>&1 & +else + "$PUB" "${PUB_ARGS[@]}" > "$PUB_LOG" 2>&1 & +fi +sleep 4 + +# --- Subscriber (operator, root netns) --- +# The subscriber opens an eframe/wgpu window (frame-latency stats are produced by its render +# loop), so it needs a display. The box has a real X server on :1; use it unless DISPLAY is +# already set. xvfb-run is used as a fallback if available. +SUB_DISPLAY="${DISPLAY:-:1}" +SUB_XAUTH="${XAUTHORITY:-/run/user/$(id -u)/gdm/Xauthority}" +DISPLAY="$SUB_DISPLAY" XAUTHORITY="$SUB_XAUTH" "$SUB" --room-name "$ROOM" --identity flexfec-sub > "$SUB_LOG" 2>&1 & + +sleep "$DURATION" +cleanup +sleep 1 + +echo "-- SFU FlexFEC recovery (publisher uplink), last 3 --" +grep -iE 'flexfec recovery stats' "$SFU_LOG" 2>/dev/null | tail -3 | sed -E 's/.*"fecPacketsReceived"/fecRecv/; s/, "fecReportedReceived.*//' +echo "-- SFU flexfec decoder enabled? --" +grep -iE 'flexfec decoder enabled' "$SFU_LOG" 2>/dev/null | tail -1 +echo "-- frame latency, last 6 windows --" +grep -iE 'frame latency' "$SUB_LOG" 2>/dev/null | tail -6 +echo "-- video quality (last) --" +grep -iE 'Video quality' "$SUB_LOG" 2>/dev/null | tail -1 +echo "-- jitter buffer (last) --" +grep -iE 'jitter buffer' "$SUB_LOG" 2>/dev/null | tail -1 +echo "============================================================================" diff --git a/scripts/flexfec/run-publisher.sh b/scripts/flexfec/run-publisher.sh new file mode 100755 index 000000000..4ce98ce3f --- /dev/null +++ b/scripts/flexfec/run-publisher.sh @@ -0,0 +1,66 @@ +#!/usr/bin/env bash +# +# Run the rust-sdks local_video publisher with FlexFEC publishing enabled. +# +# Connects to the local dev SFU (run-sfu.sh) and publishes camera video with a +# FlexFEC-03 repair stream (--flex-fec). Output is teed to a log file. +# +# Usage: +# scripts/flexfec/run-publisher.sh [extra publisher args...] +# +# Env overrides: +# RUST_SDKS_DIR path to the rust-sdks5 checkout (default: ../rust-sdks5) +# LIVEKIT_URL default ws://127.0.0.1:7880 +# LIVEKIT_API_KEY / LIVEKIT_API_SECRET default devkey / secret +# ROOM room name (default flexfec-test) +# PUB_IDENTITY participant identity (default flexfec-pub) +# RUST_LOG log filter (default info) +# LOG_DIR where to write logs (default scripts/flexfec/logs) +# TEST_PATTERN 1 = SMPTE test pattern (no camera); 0 = use camera (default 1) +# ATTACH_META 1 = attach per-frame timestamp + frame id so the subscriber can log +# end-to-end frame latency; 0 = off (default 1) +# MIN_PLAYOUT_DELAY / MAX_PLAYOUT_DELAY ms; when set, recreates the room with a +# subscriber playout-delay cap (passed through to the publisher) +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +REPO_ROOT="$(cd "$SCRIPT_DIR/../.." && pwd)" +RUST_SDKS_DIR="${RUST_SDKS_DIR:-$REPO_ROOT/../rust-sdks5}" +LOG_DIR="${LOG_DIR:-$SCRIPT_DIR/logs}" + +if [[ ! -d "$RUST_SDKS_DIR" ]]; then + echo "error: rust-sdks checkout not found at '$RUST_SDKS_DIR' (set RUST_SDKS_DIR)" >&2 + exit 1 +fi + +export LIVEKIT_URL="${LIVEKIT_URL:-ws://127.0.0.1:7880}" +export LIVEKIT_API_KEY="${LIVEKIT_API_KEY:-devkey}" +export LIVEKIT_API_SECRET="${LIVEKIT_API_SECRET:-secret}" +export RUST_LOG="${RUST_LOG:-info}" +ROOM="${ROOM:-flexfec-test}" +PUB_IDENTITY="${PUB_IDENTITY:-flexfec-pub}" + +PUB_ARGS=(--flex-fec --room-name "$ROOM" --identity "$PUB_IDENTITY") +if [[ "${TEST_PATTERN:-1}" == "1" ]]; then + PUB_ARGS+=(--test-pattern) +fi +if [[ "${ATTACH_META:-1}" == "1" ]]; then + PUB_ARGS+=(--attach-timestamp --attach-frame-id) +fi +if [[ -n "${MIN_PLAYOUT_DELAY:-}" ]]; then + PUB_ARGS+=(--min-playout-delay "$MIN_PLAYOUT_DELAY") +fi +if [[ -n "${MAX_PLAYOUT_DELAY:-}" ]]; then + PUB_ARGS+=(--max-playout-delay "$MAX_PLAYOUT_DELAY") +fi + +mkdir -p "$LOG_DIR" +echo "==> Publisher -> $LIVEKIT_URL room=$ROOM identity=$PUB_IDENTITY (FlexFEC on)" +echo " args: ${PUB_ARGS[*]}" +echo " log: $LOG_DIR/publisher.log" + +cd "$RUST_SDKS_DIR" +set -o pipefail +cargo run -p local_video --features desktop --bin publisher -- \ + "${PUB_ARGS[@]}" \ + "$@" 2>&1 | tee "$LOG_DIR/publisher.log" diff --git a/scripts/flexfec/run-sfu.sh b/scripts/flexfec/run-sfu.sh new file mode 100755 index 000000000..0fd855778 --- /dev/null +++ b/scripts/flexfec/run-sfu.sh @@ -0,0 +1,62 @@ +#!/usr/bin/env bash +# +# Build and run the LiveKit SFU locally with FlexFEC-03 enabled. +# +# Uses `--dev` (devkey/secret API keys, loopback bind) together with the local +# FlexFEC config. Media is pinned to UDP 7882 so netem.sh can shape it. +# +# Usage: +# scripts/flexfec/run-sfu.sh [extra go-run/server args...] +# +# Env overrides: +# CONFIG path to the SFU config (default: scripts/flexfec/flexfec-local.yaml) +# Operator/subscriber side (SFU<->operator): +# DOWNLINK_LOSS fractional SFU->subscriber loss, e.g. 0.01 for 1% (unset = none) +# DOWNLINK_DELAY_MS one-way SFU->subscriber delay in ms (unset = none) +# UPLINK_DELAY_MS one-way subscriber->SFU feedback (NACK/RTCP) delay in ms +# Robot/publisher side (robot<->SFU), e.g. a lossy 5G first hop: +# PUB_LOSS fractional publisher->SFU loss, e.g. 0.03 for 3% (unset = none) +# PUB_DELAY_MS one-way robot<->SFU delay in ms (inbound media + outbound NACK) +# +# These inject impairment in-SFU (pkg/sfu/downtrack_impair.go for the subscriber egress, +# pkg/sfu/buffer/buffer_impair.go for the publisher ingress), replacing netem.sh for the +# all-on-one-box setup where macOS pf/dummynet can't shape same-host UDP. Delays apply to +# retransmissions too; set the two delays equal for a symmetric link so a NACK->RTX +# recovery costs a full RTT -- the latency FlexFEC avoids by repairing inline. +# +# Teleop topology example (lossy 5G robot uplink, clean wired operator): +# PUB_LOSS=0.03 PUB_DELAY_MS=40 scripts/flexfec/run-sfu.sh # + flexfec.publisher: true +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +REPO_ROOT="$(cd "$SCRIPT_DIR/../.." && pwd)" +CONFIG="${CONFIG:-$SCRIPT_DIR/flexfec-local.yaml}" + +cd "$REPO_ROOT" + +if [[ -n "${DOWNLINK_LOSS:-}" ]]; then + export LK_DOWNLINK_LOSS="$DOWNLINK_LOSS" +fi +if [[ -n "${DOWNLINK_DELAY_MS:-}" ]]; then + export LK_DOWNLINK_DELAY_MS="$DOWNLINK_DELAY_MS" +fi +if [[ -n "${UPLINK_DELAY_MS:-}" ]]; then + export LK_UPLINK_DELAY_MS="$UPLINK_DELAY_MS" +fi +if [[ -n "${PUB_LOSS:-}" ]]; then + export LK_PUB_LOSS="$PUB_LOSS" +fi +if [[ -n "${PUB_DELAY_MS:-}" ]]; then + export LK_PUB_DELAY_MS="$PUB_DELAY_MS" +fi + +echo "==> Building & running LiveKit SFU with FlexFEC" +echo " repo: $REPO_ROOT" +echo " config: $CONFIG" +echo " url: ws://127.0.0.1:7880 (devkey / secret)" +echo " media: udp/7882 on loopback" +echo " operator downlink: loss=${LK_DOWNLINK_LOSS:-none} delay_ms=${LK_DOWNLINK_DELAY_MS:-none} nack_delay_ms=${LK_UPLINK_DELAY_MS:-none}" +echo " robot uplink: loss=${LK_PUB_LOSS:-none} delay_ms=${LK_PUB_DELAY_MS:-none}" +echo + +exec go run ./cmd/server --dev --config "$CONFIG" "$@" diff --git a/scripts/flexfec/run-subscriber.sh b/scripts/flexfec/run-subscriber.sh new file mode 100755 index 000000000..072541780 --- /dev/null +++ b/scripts/flexfec/run-subscriber.sh @@ -0,0 +1,59 @@ +#!/usr/bin/env bash +# +# Run the rust-sdks local_video subscriber against the local dev SFU and surface +# FlexFEC stats. The subscriber logs "Video FEC ..." lines as repair packets arrive +# and recovery payload is accepted. +# +# Usage: +# scripts/flexfec/run-subscriber.sh [extra subscriber args...] +# +# Env overrides: +# RUST_SDKS_DIR path to the rust-sdks5 checkout (default: ../rust-sdks5) +# LIVEKIT_URL default ws://127.0.0.1:7880 +# LIVEKIT_API_KEY / LIVEKIT_API_SECRET default devkey / secret +# ROOM room name (default flexfec-test) +# SUB_IDENTITY participant identity (default flexfec-sub) +# RUST_LOG log filter (default info) +# LOG_DIR where to write logs (default scripts/flexfec/logs) +# FEC_ONLY if set to 1, only print FEC + frame-latency lines to the console +# +# Frame latency: when the publisher runs with --attach-timestamp/--attach-frame-id, the +# subscriber emits "Subscriber frame latency: ..." lines (capture->receive->decode). The +# receive_to_decode value is the jitter-buffer wait, which is where FEC-ON and RTX-only +# diverge; capture_to_decode is the end-to-end glass-to-decode latency. +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +REPO_ROOT="$(cd "$SCRIPT_DIR/../.." && pwd)" +RUST_SDKS_DIR="${RUST_SDKS_DIR:-$REPO_ROOT/../rust-sdks5}" +LOG_DIR="${LOG_DIR:-$SCRIPT_DIR/logs}" + +if [[ ! -d "$RUST_SDKS_DIR" ]]; then + echo "error: rust-sdks checkout not found at '$RUST_SDKS_DIR' (set RUST_SDKS_DIR)" >&2 + exit 1 +fi + +export LIVEKIT_URL="${LIVEKIT_URL:-ws://127.0.0.1:7880}" +export LIVEKIT_API_KEY="${LIVEKIT_API_KEY:-devkey}" +export LIVEKIT_API_SECRET="${LIVEKIT_API_SECRET:-secret}" +export RUST_LOG="${RUST_LOG:-info}" +ROOM="${ROOM:-flexfec-test}" +SUB_IDENTITY="${SUB_IDENTITY:-flexfec-sub}" + +mkdir -p "$LOG_DIR" +echo "==> Subscriber -> $LIVEKIT_URL room=$ROOM identity=$SUB_IDENTITY" +echo " log: $LOG_DIR/subscriber.log (grep 'Video FEC' for FEC, 'frame latency' for latency)" + +cd "$RUST_SDKS_DIR" +set -o pipefail +if [[ "${FEC_ONLY:-0}" == "1" ]]; then + cargo run -p local_video --features desktop --bin subscriber -- \ + --room-name "$ROOM" \ + --identity "$SUB_IDENTITY" \ + "$@" 2>&1 | tee "$LOG_DIR/subscriber.log" | grep --line-buffered -iE 'fec|recover|frame latency' || true +else + cargo run -p local_video --features desktop --bin subscriber -- \ + --room-name "$ROOM" \ + --identity "$SUB_IDENTITY" \ + "$@" 2>&1 | tee "$LOG_DIR/subscriber.log" +fi