diff --git a/pkg/sfu/buffer/buffer_base.go b/pkg/sfu/buffer/buffer_base.go index ec7ff6ac6..e124b1311 100644 --- a/pkg/sfu/buffer/buffer_base.go +++ b/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 } } } diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index 29e35747b..efb10a3d0 100644 --- a/pkg/sfu/downtrack.go +++ b/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) diff --git a/pkg/sfu/downtrack_downstream_integration_test.go b/pkg/sfu/downtrack_downstream_integration_test.go index 07fd81880..99f1c3d0f 100644 --- a/pkg/sfu/downtrack_downstream_integration_test.go +++ b/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 // ----------------------------------------------------------------------------- diff --git a/pkg/sfu/export_test.go b/pkg/sfu/export_test.go index 9812b2c33..8baa42ade 100644 --- a/pkg/sfu/export_test.go +++ b/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) { diff --git a/pkg/sfu/pacer/base_test.go b/pkg/sfu/pacer/base_test.go index 82599f049..7fdc53645 100644 --- a/pkg/sfu/pacer/base_test.go +++ b/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))) } diff --git a/pkg/sfu/pacer/pacer.go b/pkg/sfu/pacer/pacer.go index 6b89b6d1f..d5460b166 100644 --- a/pkg/sfu/pacer/pacer.go +++ b/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 { diff --git a/pkg/sfu/rtpextension/abscapturetime/abscapturetime.go b/pkg/sfu/rtpextension/abscapturetime/abscapturetime.go index 1eb3f8d2a..503daa0be 100644 --- a/pkg/sfu/rtpextension/abscapturetime/abscapturetime.go +++ b/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 { diff --git a/pkg/sfu/rtpextension/dependencydescriptor/dependencydescriptorextension.go b/pkg/sfu/rtpextension/dependencydescriptor/dependencydescriptorextension.go index 3a1bb4760..fe2fbbfcf 100644 --- a/pkg/sfu/rtpextension/dependencydescriptor/dependencydescriptorextension.go +++ b/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) diff --git a/pkg/sfu/sequencer.go b/pkg/sfu/sequencer.go index f737ccbb5..724f04e26 100644 --- a/pkg/sfu/sequencer.go +++ b/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 } diff --git a/pkg/sfu/sequencer_test.go b/pkg/sfu/sequencer_test.go index 4d769e096..dd314a870 100644 --- a/pkg/sfu/sequencer_test.go +++ b/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) {