fix(sfu): stop allocating per packet for abs-capture-time and retransmits (#5003)

* sfu: stop allocating per packet for abs-capture-time and retransmits

* sfu: drop unused abs-capture-time marshal errors

* sfu: size held extensions locally, guard nil holds, keep sequencer clones

pkg/sfu/buffer/buffer_base.go

@@ -93,9 +93,10 @@ type ExtPacket struct {
 	pkt        rtp.Packet
 	extensions [8]rtp.Extension
 
-	// backing storage for VP8 and VP9, so parsing does not allocate
-	vp8 codec.VP8
-	vp9 codecs.VP9Packet
+	// backing storage for VP8, VP9 and abs-capture-time, so parsing does not allocate
+	vp8            codec.VP8
+	vp9            codecs.VP9Packet
+	absCaptureTime act.AbsCaptureTime
 }
 
 type BufferProvider interface {
@@ -1072,9 +1073,8 @@ func (b *BufferBase) getExtPacket(
 
 	if b.absCaptureTimeExtID != 0 {
 		if extData := rtpPacket.GetExtension(b.absCaptureTimeExtID); len(extData) != 0 {
-			actExt := &act.AbsCaptureTime{}
-			if err := actExt.Unmarshal(extData); err == nil {
-				ep.AbsCaptureTimeExt = actExt
+			if err := ep.absCaptureTime.Unmarshal(extData); err == nil {
+				ep.AbsCaptureTimeExt = &ep.absCaptureTime
 			}
 		}
 	}

pkg/sfu/downtrack.go

@@ -1110,13 +1110,13 @@ func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) int32 {
 
 	// add extensions
 	//
-	// the header refers to the descriptor until the pacer marshals it, so hold it in the pacer packet, which one send owns
-	ddBytes := tp.ddBytesSpill
+	// the header refers to extensions until the pacer marshals them, so hold the ones on the stack in the pacer packet, which one send owns
+	ddBytes, ddExt := tp.ddBytesSpill, tp.ddBytesSpill
 	if inline := tp.ddInline(); ddBytes == nil && len(inline) != 0 {
-		ddBytes = pacerPacket.HoldExtension(inline)
+		ddBytes, ddExt = inline, pacerPacket.HoldExtension(inline)
 	}
-	if d.dependencyDescriptorExtID != 0 && len(ddBytes) != 0 {
-		hdr.SetExtension(uint8(d.dependencyDescriptorExtID), ddBytes)
+	if d.dependencyDescriptorExtID != 0 && len(ddExt) != 0 {
+		hdr.SetExtension(uint8(d.dependencyDescriptorExtID), ddExt)
 	}
 	if d.playoutDelayExtID != 0 && d.playoutDelay != nil {
 		if val := d.playoutDelay.GetDelayExtension(hdr.SequenceNumber); val != nil {
@@ -1133,13 +1133,12 @@ func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) int32 {
 	}
 	var actBytes []byte
 	if extPkt.AbsCaptureTimeExt != nil && d.absCaptureTimeExtID != 0 {
+		var actBuf [act.MaxSize]byte
+		n := 0
 		if !d.params.DisableSenderReportPassThrough {
 			// pass through the original publisher capture time verbatim, consistent
 			// with sender reports also being passed through unchanged.
-			actBytes, err = extPkt.AbsCaptureTimeExt.Marshal()
-			if err == nil {
-				hdr.SetExtension(uint8(d.absCaptureTimeExtID), actBytes)
-			}
+			n, _ = extPkt.AbsCaptureTimeExt.MarshalTo(actBuf[:])
 		} else {
 			// normalize capture time to SFU clock.
 			// NOTE: even if there is estimated offset populated, just re-map the
@@ -1153,26 +1152,29 @@ func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) int32 {
 				if err = actExtCopy.Rewrite(
 					rtpstats.RTCPSenderReportPropagationDelay(refSenderReport),
 				); err == nil {
-					actBytes, err = actExtCopy.Marshal()
-					if err == nil {
-						hdr.SetExtension(uint8(d.absCaptureTimeExtID), actBytes)
-					}
+					n, _ = actExtCopy.MarshalTo(actBuf[:])
 				}
 			}
 		}
+		if n != 0 {
+			actBytes = actBuf[:n]
+			if actExt := pacerPacket.HoldExtension(actBytes); actExt != nil {
+				hdr.SetExtension(uint8(d.absCaptureTimeExtID), actExt)
+			}
+		}
 	}
 	d.addDummyExtensions(hdr)
 
-	// the sequencer copies the descriptor, which has to happen before Enqueue frees the scratch
+	// the sequencer copies from the translation params and the incoming packet, not from the pacer packet that Enqueue frees
 	if d.sequencer != nil {
 		d.sequencer.push(
 			extPkt.Arrival,
 			extPkt.ExtSequenceNumber,
 			tp.rtp.extSequenceNumber,
 			tp.rtp.extTimestamp,
-			hdr.Marker,
+			tp.marker,
 			int8(layer),
-			payload[:len(codecBytes)],
+			codecBytes,
 			tp.incomingHeaderSize,
 			ddBytes,
 			actBytes,
@@ -2315,8 +2317,8 @@ func (d *DownTrack) retransmitPacket(epm *extPacketMeta, sourcePkt []byte, isPro
 			hdr.SetExtension(uint8(d.dependencyDescriptorExtID), ddBytes)
 		}
 	}
-	if d.absCaptureTimeExtID != 0 && len(epm.actBytes) != 0 {
-		hdr.SetExtension(uint8(d.absCaptureTimeExtID), epm.actBytes)
+	if d.absCaptureTimeExtID != 0 && epm.actBytesSize != 0 {
+		hdr.SetExtension(uint8(d.absCaptureTimeExtID), epm.actBytes[:epm.actBytesSize])
 	}
 	d.addDummyExtensions(hdr)
 
@@ -2469,7 +2471,9 @@ func (d *DownTrack) retransmitPackets(nacks []uint16) {
 	nackAcks := uint32(0)
 	nackMisses := uint32(0)
 	numRepeatedNACKs := uint32(0)
-	for _, epm := range d.sequencer.getExtPacketMetas(filtered) {
+	epms := d.sequencer.getExtPacketMetas(filtered)
+	for i := range epms {
+		epm := &epms[i]
 		if disallowedLayers[epm.layer] {
 			continue
 		}
@@ -2490,7 +2494,7 @@ func (d *DownTrack) retransmitPackets(nacks []uint16) {
 			numRepeatedNACKs++
 		}
 
-		d.retransmitPacket(&epm, pktBuff[:n], false)
+		d.retransmitPacket(epm, pktBuff[:n], false)
 	}
 
 	d.totalRepeatedNACKs.Add(numRepeatedNACKs)

pkg/sfu/downtrack_downstream_integration_test.go

@@ -69,9 +69,11 @@ import (
 	"github.com/livekit/livekit-server/pkg/sfu/ccutils"
 	"github.com/livekit/livekit-server/pkg/sfu/pacer"
 	"github.com/livekit/livekit-server/pkg/sfu/packettrailer"
+	act "github.com/livekit/livekit-server/pkg/sfu/rtpextension/abscapturetime"
 	"github.com/livekit/livekit-server/pkg/sfu/sfufakes"
 	"github.com/livekit/livekit-server/pkg/sfu/streamallocator"
 	"github.com/livekit/livekit-server/pkg/sfu/testutils"
+	"github.com/livekit/livekit-server/pkg/sfu/utils"
 	"github.com/livekit/livekit-server/pkg/testutils/vnettest"
 )
 
@@ -356,6 +358,54 @@ func TestDownTrackForwardsMedia(t *testing.T) {
 	}
 }
 
+type discardWriteStream struct{}
+
+func (discardWriteStream) WriteRTP(*rtp.Header, []byte) (int, error) { return 0, nil }
+func (discardWriteStream) Write([]byte) (int, error)                 { return 0, nil }
+
+type discardPacer struct {
+	*pacer.PassThrough
+}
+
+func (d discardPacer) Enqueue(p *pacer.Packet) {
+	p.WriteStream = discardWriteStream{}
+	d.PassThrough.Enqueue(p)
+}
+
+func TestDownTrackWriteRTPNoAlloc(t *testing.T) {
+	h := vnettest.NewHosts(t)
+	factory := buffer.NewFactoryOfBufferFactory(500, 500).CreateBufferFactory()
+	p := discardPacer{pacer.NewPassThrough(logger.GetLogger(), newNullBWE())}
+	t.Cleanup(p.Stop)
+
+	dh := newBoundDownTrack(t, h, factory, opusCodecParams, p, vnettest.MediaEngineConfig{})
+	dh.dt.SetAbsCaptureTimeExtIDForTest(5)
+
+	ep, err := testutils.GetTestExtPacket(&testutils.TestExtPacketParams{
+		SequenceNumber: 1000,
+		Timestamp:      5000,
+		SSRC:           0x12345678,
+		PayloadType:    111,
+		PayloadSize:    40,
+	})
+	require.NoError(t, err)
+	ep.AbsCaptureTimeExt = act.AbsCaptureTimeFromValue(0x0102030405060708, 9)
+
+	const runs = 1000
+	forwarded := int32(0)
+	allocs := testing.AllocsPerRun(runs, func() {
+		ep.Packet.SequenceNumber++
+		ep.ExtSequenceNumber++
+		ep.Packet.Timestamp += 960
+		ep.ExtTimestamp += 960
+		forwarded += dh.dt.WriteRTP(ep, 0)
+	})
+	require.EqualValues(t, runs+1, forwarded, "every write, including the AllocsPerRun warm up, should forward")
+	if !utils.RaceEnabled {
+		require.Zero(t, allocs, "allocations per WriteRTP")
+	}
+}
+
 // -----------------------------------------------------------------------------
 // retransmitPacket: RTX / retransmission
 // -----------------------------------------------------------------------------

pkg/sfu/export_test.go

@@ -50,6 +50,8 @@ func (d *DownTrack) HandleRTCPForTest(pkt []byte) { d.handleRTCP(pkt) }
 
 func (d *DownTrack) SetRTPHeaderExtensionsForTest() { d.setRTPHeaderExtensions() }
 
+func (d *DownTrack) SetAbsCaptureTimeExtIDForTest(id int) { d.absCaptureTimeExtID = id }
+
 // ForceForwardLayerForTest forces the forwarder onto a valid current/target layer so a
 // (non-simulcast) video packet will be forwarded without driving full allocation.
 func (d *DownTrack) ForceForwardLayerForTest(layer buffer.VideoLayer) {

pkg/sfu/pacer/base_test.go

@@ -25,7 +25,6 @@ import (
 	"github.com/livekit/protocol/logger"
 
 	"github.com/livekit/livekit-server/pkg/sfu/bwe"
-	dd "github.com/livekit/livekit-server/pkg/sfu/rtpextension/dependencydescriptor"
 	"github.com/livekit/livekit-server/pkg/sfu/utils"
 )
 
@@ -94,16 +93,16 @@ func TestHoldExtension(t *testing.T) {
 	ext := []byte{1, 2, 3}
 	held := p.HoldExtension(ext)
 	require.Equal(t, ext, held)
-	require.Same(t, &p.extBuf[0], &held[0], "the extension has to be held in the packet's scratch")
+	require.Same(t, &p.extBufs[0][0], &held[0], "the extension has to be held in the packet's scratch")
 
 	ext[0] = 9
 	require.EqualValues(t, 1, held[0], "the held copy has to be independent of the caller's slice")
 
-	// exactly the one byte extension profile payload cap
-	exact := bytes.Repeat([]byte{7}, dd.MaxInlineExtensionSize)
+	// exactly the largest extension the scratch holds
+	exact := bytes.Repeat([]byte{7}, maxHeldExtensionSize)
 	held = p.HoldExtension(exact)
 	require.Equal(t, exact, held)
 
 	// one byte more does not fit
-	require.Nil(t, p.HoldExtension(bytes.Repeat([]byte{7}, dd.MaxInlineExtensionSize+1)))
+	require.Nil(t, (&Packet{}).HoldExtension(bytes.Repeat([]byte{7}, maxHeldExtensionSize+1)))
 }

pkg/sfu/pacer/pacer.go

@@ -19,6 +19,7 @@ import (
 	"time"
 
 	"github.com/livekit/livekit-server/pkg/sfu/ccutils"
+	act "github.com/livekit/livekit-server/pkg/sfu/rtpextension/abscapturetime"
 	dd "github.com/livekit/livekit-server/pkg/sfu/rtpextension/dependencydescriptor"
 	"github.com/pion/rtp"
 	"github.com/pion/webrtc/v4"
@@ -42,6 +43,9 @@ const (
 	PacerBehaviorLeakybucket PacerBehavior = "leaky-bucket"
 )
 
+// largest extension HoldExtension copies, an inline dependency descriptor or an abs-capture-time
+const maxHeldExtensionSize = max(dd.MaxInlineExtensionSize, act.MaxSize)
+
 type Packet struct {
 	Header             *rtp.Header
 	HeaderPool         *sync.Pool
@@ -56,19 +60,23 @@ type Packet struct {
 	Pool               *sync.Pool
 	PoolEntity         *[]byte
 
-	// per-packet scratch for header extensions patched at send time,
+	// per-packet scratch for header extensions held before Enqueue or patched at send time,
 	// the Packet is owned by one send until SendPacket returns
 	absSendTimeBuf [3]byte
 	twccBuf        [2]byte
-	extBuf         [dd.MaxInlineExtensionSize]byte
+	extBufs        [2][maxHeldExtensionSize]byte
+	numExtHeld     int
 }
 
-// HoldExtension copies ext into scratch the header can point at until SendPacket returns, nil if it does not fit
+// HoldExtension copies ext into the next free scratch slot, which the header can point at until SendPacket returns.
+// It returns nil when ext does not fit or every slot is held.
 func (p *Packet) HoldExtension(ext []byte) []byte {
-	if len(ext) > len(p.extBuf) {
+	if p.numExtHeld == len(p.extBufs) || len(ext) > len(p.extBufs[p.numExtHeld]) {
 		return nil
 	}
-	return p.extBuf[:copy(p.extBuf[:], ext)]
+	buf := &p.extBufs[p.numExtHeld]
+	p.numExtHeld++
+	return buf[:copy(buf[:], ext)]
 }
 
 type Pacer interface {

pkg/sfu/rtpextension/abscapturetime/abscapturetime.go

@@ -24,6 +24,9 @@ import (
 
 const (
 	AbsCaptureTimeURI = "http://www.webrtc.org/experiments/rtp-hdrext/abs-capture-time"
+
+	// MaxSize is the marshaled size with the estimated capture clock offset
+	MaxSize = 16
 )
 
 var (
@@ -84,21 +87,23 @@ func (a *AbsCaptureTime) Rewrite(offset time.Duration) error {
 	return nil
 }
 
-func (a *AbsCaptureTime) Marshal() ([]byte, error) {
+func (a *AbsCaptureTime) MarshalTo(buf []byte) (int, error) {
 	if a.absoluteCaptureTimestamp == 0 {
-		return nil, errInvalidData
+		return 0, errInvalidData
 	}
 
 	size := 8
 	if a.estimatedCaptureClockOffset != 0 {
 		size += 8
 	}
-	marshalled := make([]byte, size)
-	binary.BigEndian.PutUint64(marshalled, uint64(a.absoluteCaptureTimestamp))
-	if a.estimatedCaptureClockOffset != 0 {
-		binary.BigEndian.PutUint64(marshalled[8:], uint64(a.estimatedCaptureClockOffset))
+	if len(buf) < size {
+		return 0, errTooSmall
 	}
-	return marshalled, nil
+	binary.BigEndian.PutUint64(buf, uint64(a.absoluteCaptureTimestamp))
+	if a.estimatedCaptureClockOffset != 0 {
+		binary.BigEndian.PutUint64(buf[8:], uint64(a.estimatedCaptureClockOffset))
+	}
+	return size, nil
 }
 
 func (a *AbsCaptureTime) Unmarshal(marshalled []byte) error {

pkg/sfu/rtpextension/dependencydescriptor/dependencydescriptorextension.go

@@ -102,7 +102,7 @@ const (
 	MaxDecodeTargets = 32
 	MaxTemplates     = 64
 
-	// MaxInlineExtensionSize is the inline storage the forwarding path and the pacer reserve for a marshaled extension, pion's one byte profile cap
+	// MaxInlineExtensionSize is the inline storage the forwarding path reserves for a marshaled extension, pion's one byte profile cap
 	MaxInlineExtensionSize = 16
 
 	AllChainsAreActive = uint32(0)

pkg/sfu/sequencer.go

@@ -19,6 +19,7 @@ import (
 	"sync"
 	"time"
 
+	act "github.com/livekit/livekit-server/pkg/sfu/rtpextension/abscapturetime"
 	"github.com/livekit/livekit-server/pkg/sfu/utils"
 	"github.com/livekit/protocol/logger"
 	"go.uber.org/zap/zapcore"
@@ -65,7 +66,8 @@ type packetMeta struct {
 	ddBytesSize  uint8
 	ddBytesSlice []byte
 	// abs-capture-time of packet
-	actBytes []byte
+	actBytes     [act.MaxSize]byte
+	actBytesSize uint8
 	// number of packet trailer bytes stripped when the packet was forwarded
 	trailerStripped uint8
 }
@@ -86,8 +88,8 @@ func (pm packetMeta) MarshalLogObject(e zapcore.ObjectEncoder) error {
 	} else {
 		e.AddUint8("ddBytesSize", pm.ddBytesSize)
 	}
-	if len(pm.actBytes) != 0 {
-		e.AddInt("actBytes", len(pm.actBytes))
+	if pm.actBytesSize != 0 {
+		e.AddUint8("actBytes", pm.actBytesSize)
 	}
 	if pm.trailerStripped != 0 {
 		e.AddUint8("trailerStripped", pm.trailerStripped)
@@ -271,7 +273,7 @@ func (s *sequencer) push(
 		copy(pm.ddBytes[:pm.ddBytesSize], ddBytes)
 	}
 
-	pm.actBytes = append([]byte{}, actBytes...)
+	pm.actBytesSize = uint8(copy(pm.actBytes[:], actBytes))
 
 	if extModifiedSN > s.extHighestSN {
 		s.extHighestSN = extModifiedSN
@@ -397,7 +399,6 @@ func (s *sequencer) getExtPacketMetas(seqNo []uint16) []extPacketMeta {
 			}
 			epm.codecBytesSlice = append([]byte{}, meta.codecBytesSlice...)
 			epm.ddBytesSlice = append([]byte{}, meta.ddBytesSlice...)
-			epm.actBytes = append([]byte{}, meta.actBytes...)
 			extPacketMetas = append(extPacketMetas, epm)
 		}
 	}
@@ -447,7 +448,6 @@ func (s *sequencer) lookupExtPacketMeta(extSN uint64) *extPacketMeta {
 	}
 	epm.codecBytesSlice = append([]byte{}, meta.codecBytesSlice...)
 	epm.ddBytesSlice = append([]byte{}, meta.ddBytesSlice...)
-	epm.actBytes = append([]byte{}, meta.actBytes...)
 	return &epm
 }
 

pkg/sfu/sequencer_test.go

@@ -247,7 +247,7 @@ func Test_sequencer_getNACKSeqNo_exclusion(t *testing.T) {
 					require.Equal(t, uint8(len(tt.fields.codecBytesOversized)), sn.numCodecBytesIn)
 					require.Equal(t, tt.fields.ddBytesOversized, sn.ddBytesSlice)
 					require.Equal(t, uint8(len(tt.fields.codecBytesOversized)), sn.ddBytesSize)
-					require.Equal(t, tt.fields.actBytesOdd, sn.actBytes)
+					require.Equal(t, tt.fields.actBytesOdd, sn.actBytes[:sn.actBytesSize])
 				} else {
 					if sn.sourceSeqNo%2 == 0 {
 						require.Equal(t, tt.fields.markerEven, sn.marker)
@@ -255,14 +255,14 @@ func Test_sequencer_getNACKSeqNo_exclusion(t *testing.T) {
 						require.Equal(t, uint8(tt.fields.numCodecBytesInEven), sn.numCodecBytesIn)
 						require.Equal(t, tt.fields.ddBytesEven, sn.ddBytes[:sn.ddBytesSize])
 						require.Equal(t, uint8(len(tt.fields.ddBytesEven)), sn.ddBytesSize)
-						require.Equal(t, tt.fields.actBytesEven, sn.actBytes)
+						require.Equal(t, tt.fields.actBytesEven, sn.actBytes[:sn.actBytesSize])
 					} else {
 						require.Equal(t, tt.fields.markerOdd, sn.marker)
 						require.Equal(t, tt.fields.codecBytesOdd, sn.codecBytes[:sn.numCodecBytesOut])
 						require.Equal(t, uint8(tt.fields.numCodecBytesInOdd), sn.numCodecBytesIn)
 						require.Equal(t, tt.fields.ddBytesOdd, sn.ddBytes[:sn.ddBytesSize])
 						require.Equal(t, uint8(len(tt.fields.ddBytesOdd)), sn.ddBytesSize)
-						require.Equal(t, tt.fields.actBytesOdd, sn.actBytes)
+						require.Equal(t, tt.fields.actBytesOdd, sn.actBytes[:sn.actBytesSize])
 					}
 				}
 			}
@@ -388,14 +388,14 @@ func Test_sequencer_getNACKSeqNo_no_exclusion(t *testing.T) {
 					require.Equal(t, uint8(tt.fields.numCodecBytesInEven), sn.numCodecBytesIn)
 					require.Equal(t, tt.fields.ddBytesEven, sn.ddBytes[:sn.ddBytesSize])
 					require.Equal(t, uint8(len(tt.fields.ddBytesEven)), sn.ddBytesSize)
-					require.Equal(t, tt.fields.actBytesEven, sn.actBytes)
+					require.Equal(t, tt.fields.actBytesEven, sn.actBytes[:sn.actBytesSize])
 				} else {
 					require.Equal(t, tt.fields.markerOdd, sn.marker)
 					require.Equal(t, tt.fields.codecBytesOdd, sn.codecBytes[:sn.numCodecBytesOut])
 					require.Equal(t, uint8(tt.fields.numCodecBytesInOdd), sn.numCodecBytesIn)
 					require.Equal(t, tt.fields.ddBytesOdd, sn.ddBytes[:sn.ddBytesSize])
 					require.Equal(t, uint8(len(tt.fields.ddBytesOdd)), sn.ddBytesSize)
-					require.Equal(t, tt.fields.actBytesOdd, sn.actBytes)
+					require.Equal(t, tt.fields.actBytesOdd, sn.actBytes[:sn.actBytesSize])
 				}
 			}
 			if !reflect.DeepEqual(got, tt.want) {