diff --git a/pkg/rtc/mediaengine.go b/pkg/rtc/mediaengine.go index 45df171bd..5d0828d8e 100644 --- a/pkg/rtc/mediaengine.go +++ b/pkg/rtc/mediaengine.go @@ -217,7 +217,7 @@ func filterCodecs( enabledCodecs []*livekit.Codec, rtcpFeedbackConfig RTCPFeedbackConfig, filterOutH264HighProfile bool, - keepFlexFEC bool, + enableFlexFEC bool, ) []webrtc.RTPCodecParameters { filteredCodecs := make([]webrtc.RTPCodecParameters, 0, len(codecs)) for _, c := range codecs { @@ -228,7 +228,7 @@ func filterCodecs( // flexfec-03 is not part of the enabled codec lists, retain it when // the transport direction has FlexFEC enabled if mime.IsMimeTypeStringFlexFEC03(c.RTPCodecCapability.MimeType) { - if keepFlexFEC { + if enableFlexFEC { filteredCodecs = append(filteredCodecs, c) } continue diff --git a/pkg/rtc/transport.go b/pkg/rtc/transport.go index 9b46c6885..0c73cf762 100644 --- a/pkg/rtc/transport.go +++ b/pkg/rtc/transport.go @@ -1042,7 +1042,7 @@ func (t *PCTransport) queueOrConfigureSender( filterOutH264HighProfile: !t.params.IsOfferer, enableAudioStereo: enableAudioStereo, enableAudioNACK: enableAudioNACK, - keepFlexFEC: t.params.DirectionConfig.FlexFEC.Enabled, + enableFlexFEC: t.params.DirectionConfig.FlexFEC.Enabled, } if !t.params.IsOfferer { t.sendersPendingConfigMu.Lock() @@ -3191,7 +3191,7 @@ type configureSenderParams struct { filterOutH264HighProfile bool enableAudioStereo bool enableAudioNACK bool - keepFlexFEC bool + enableFlexFEC bool } func configureSender(params configureSenderParams, offerAudioPT map[mime.MimeType]webrtc.PayloadType) { @@ -3200,7 +3200,7 @@ func configureSender(params configureSenderParams, offerAudioPT map[mime.MimeTyp params.enabledCodecs, params.rtcpFeedbackConfig, params.filterOutH264HighProfile, - params.keepFlexFEC, + params.enableFlexFEC, ) if params.transceiver.Kind() == webrtc.RTPCodecTypeAudio { @@ -3306,7 +3306,7 @@ func configureSenderCodecs( enabledCodecs []*livekit.Codec, rtcpFeedbackConfig RTCPFeedbackConfig, filterOutH264HighProfile bool, - keepFlexFEC bool, + enableFlexFEC bool, ) { if len(enabledCodecs) == 0 { return @@ -3322,7 +3322,7 @@ func configureSenderCodecs( enabledCodecs, rtcpFeedbackConfig, filterOutH264HighProfile, - keepFlexFEC, + enableFlexFEC, ) tr.SetCodecPreferences(filteredCodecs) } diff --git a/pkg/sfu/buffer/buffer.go b/pkg/sfu/buffer/buffer.go index 9ff362f51..49eea11d4 100644 --- a/pkg/sfu/buffer/buffer.go +++ b/pkg/sfu/buffer/buffer.go @@ -67,6 +67,8 @@ func (d fecRecoveryDelta) invoke(callback func(received int, recovered int, disc } } +type cbFECRecovery func(received int, recovered int, discarded int, bytesReceived int) + // Buffer contains all packets type Buffer struct { *BufferBase @@ -97,7 +99,7 @@ type Buffer struct { fecSSRC uint32 fecDecoder *flexfec.Decoder fecPktBuf []byte - onFECRecovery func(received int, recovered int, discarded int, bytesReceived int) + onFECRecovery cbFECRecovery streamInfoProbe *StreamInfoProbe warnedPendingOverflow bool @@ -158,27 +160,24 @@ func (b *Buffer) Bind(params webrtc.RTPParameters, codec webrtc.RTPCodecCapabili return err } - b.lastReportAt = mono.UnixNano() + now := mono.UnixNano() + b.lastReportAt = now if len(b.pPackets) != 0 { b.logger.Debugw("releasing queued packets on bind", "count", len(b.pPackets)) } - var rtcpPackets []rtcp.Packet for _, pp := range b.pPackets { - rtcpPackets = append(rtcpPackets, b.calc(pp.packet, nil, pp.arrivalTime, true, false)...) + b.ingestLocked(pp.packet, nil, pp.arrivalTime, true, false) } b.pPackets = nil + rtcpPackets := b.getRTCPPackets(now) + b.isBound = true b.maybeCreateFECDecoderLocked() b.Unlock() - if len(rtcpPackets) != 0 { - if cb := b.getOnRtcpFeedback(); cb != nil { - cb(rtcpPackets) - } - } - + b.packetPostProcess(rtcpPackets, fecRecoveryDelta{}, nil) return nil } @@ -281,8 +280,9 @@ func (b *Buffer) Write(pkt []byte) (n int, err error) { } b.ingestLocked(pkt, rtpPacket, now, false, false) + var fecDelta fecRecoveryDelta - var onFECRecovery func(received int, recovered int, discarded int, bytesReceived int) + var onFECRecovery cbFECRecovery if b.fecDecoder != nil { // feed media into the FEC decoder, a media arrival can complete a // previously unrecoverable FEC window. Recovered packets update NACK @@ -290,7 +290,11 @@ func (b *Buffer) Write(pkt []byte) (n int, err error) { // FEC is not needlessly NACKed. fecDelta, onFECRecovery = b.feedFECLocked(rtpPacket, now) } - b.finishWriteLocked(now, fecDelta, onFECRecovery) + + rtcpPackets := b.getRTCPPackets(now) + b.Unlock() + + b.packetPostProcess(rtcpPackets, fecDelta, onFECRecovery) return } @@ -374,25 +378,16 @@ func (b *Buffer) NotifyRTX(ssrc uint32, repairSSRC uint32, rsid string) { } func (b *Buffer) writeRTX(rtxPkt *rtp.Packet, arrivalTime int64) { - rtcpPackets := b.writeRTXLocked(rtxPkt, arrivalTime) - if len(rtcpPackets) != 0 { - if cb := b.getOnRtcpFeedback(); cb != nil { - cb(rtcpPackets) - } - } -} - -func (b *Buffer) writeRTXLocked(rtxPkt *rtp.Packet, arrivalTime int64) []rtcp.Packet { b.Lock() if !b.isBound { b.Unlock() - return nil + return } if rtxPkt.PayloadType != b.rtxPayloadType { b.logger.Debugw("unexpected rtx payload type", "expected", b.rtxPayloadType, "actual", rtxPkt.PayloadType) b.Unlock() - return nil + return } if b.rtxPktBuf == nil { @@ -402,7 +397,7 @@ func (b *Buffer) writeRTXLocked(rtxPkt *rtp.Packet, arrivalTime int64) []rtcp.Pa if len(rtxPkt.Payload) < 2 { b.logger.Warnw("rtx payload too short", nil, "size", len(rtxPkt.Payload)) b.Unlock() - return nil + return } repairedPkt := *rtxPkt @@ -414,17 +409,21 @@ func (b *Buffer) writeRTXLocked(rtxPkt *rtp.Packet, arrivalTime int64) []rtcp.Pa if err != nil { b.logger.Errorw("could not marshal repaired packet", err, "ssrc", b.BufferBase.SSRC(), "sn", repairedPkt.SequenceNumber) b.Unlock() - return nil + return } b.ingestLocked(b.rtxPktBuf[:n], &repairedPkt, arrivalTime, false, true) + var fecDelta fecRecoveryDelta - var onFECRecovery func(received int, recovered int, discarded int, bytesReceived int) + var onFECRecovery cbFECRecovery if b.fecDecoder != nil { fecDelta, onFECRecovery = b.feedFECLocked(&repairedPkt, arrivalTime) } - b.finishWriteLocked(arrivalTime, fecDelta, onFECRecovery) - return nil + + rtcpPackets := b.getRTCPPackets(arrivalTime) + b.Unlock() + + b.packetPostProcess(rtcpPackets, fecDelta, onFECRecovery) } func (b *Buffer) SetPrimaryBufferForFEC(primaryBuffer *Buffer) { @@ -517,7 +516,7 @@ func (b *Buffer) getFECMediaPacketLocked(sequenceNumber uint16, dst []byte) (int // OnFECRecovery is called with counter deltas whenever FEC packets are // processed since the previous callback: FEC packets received, media packets // recovered, FEC packets discarded and FEC bytes received. -func (b *Buffer) OnFECRecovery(fn func(received int, recovered int, discarded int, bytesReceived int)) { +func (b *Buffer) OnFECRecovery(fn cbFECRecovery) { b.Lock() b.onFECRecovery = fn b.Unlock() @@ -571,7 +570,10 @@ func (b *Buffer) writeFEC(fecPkt *rtp.Packet, arrivalTime int64) { } fecDelta, onFECRecovery := b.feedFECLocked(fecPkt, arrivalTime) - b.finishWriteLocked(arrivalTime, fecDelta, onFECRecovery) + rtcpPackets := b.getRTCPPackets(arrivalTime) + b.Unlock() + + b.packetPostProcess(rtcpPackets, fecDelta, onFECRecovery) } // feedFECLocked runs a media or FEC packet through the FEC decoder and @@ -580,7 +582,7 @@ func (b *Buffer) writeFEC(fecPkt *rtp.Packet, arrivalTime int64) { func (b *Buffer) feedFECLocked( pkt *rtp.Packet, arrivalTime int64, -) (fecRecoveryDelta, func(received int, recovered int, discarded int, bytesReceived int)) { +) (fecRecoveryDelta, cbFECRecovery) { statsBefore := b.fecDecoder.Stats() recovered := b.fecDecoder.DecodeFEC(pkt) statsAfter := b.fecDecoder.Stats() @@ -718,25 +720,15 @@ func (b *Buffer) ingestLocked(rawPkt []byte, rtpPacket *rtp.Packet, arrivalTime ) } -func (b *Buffer) calc(rawPkt []byte, rtpPacket *rtp.Packet, arrivalTime int64, isBuffered bool, isRTX bool) []rtcp.Packet { - b.ingestLocked(rawPkt, rtpPacket, arrivalTime, isBuffered, isRTX) - return b.getRTCPPackets(arrivalTime) -} - -// finishWriteLocked generates RTCP feedback from the current NACK state, which -// already reflects any FEC recovery performed during this write, releases the -// buffer lock, then invokes the FEC recovery callback and delivers RTCP -// feedback. Generating RTCP after recovery ensures sequence numbers repaired by -// FEC are not needlessly NACKed. Must be called with the buffer lock held; it -// releases the lock before returning. -func (b *Buffer) finishWriteLocked( - arrivalTime int64, +// packetPostProcess runs processing after a packet is handled. +// `rtcpPackets` should be RTCP feedback from the current NACK state, which +// already reflects any FEC recovery performed during this write. +// This is to ensure that packets recovered via FEC is not NACKed. +func (b *Buffer) packetPostProcess( + rtcpPackets []rtcp.Packet, fecDelta fecRecoveryDelta, - onFECRecovery func(received int, recovered int, discarded int, bytesReceived int), + onFECRecovery cbFECRecovery, ) { - rtcpPackets := b.getRTCPPackets(arrivalTime) - b.Unlock() - fecDelta.invoke(onFECRecovery) if len(rtcpPackets) != 0 {