packet post process (#5020)

* packet post process

* remove redundant writeRTX

* send out RTCP (NACK and RR) from packetPostProcess only

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

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)
 }

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 {