From e0d98eebb6f503bb151110110496f7898998c012 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Mon, 18 Sep 2023 18:42:39 +0530 Subject: [PATCH] Do stats update in line. (#2085) --- pkg/sfu/downtrack.go | 151 ++++++++++++++++++++--------------------- pkg/sfu/pacer/base.go | 8 +-- pkg/sfu/pacer/pacer.go | 5 +- 3 files changed, 80 insertions(+), 84 deletions(-) diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index 0ddb62c19..86931681f 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -672,23 +672,23 @@ func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) error { } var payload []byte - pool := PacketFactory.Get().(*[]byte) + poolEntity := PacketFactory.Get().(*[]byte) if len(tp.codecBytes) != 0 { incomingVP8, ok := extPkt.Payload.(buffer.VP8) if ok { - payload = d.translateVP8PacketTo(extPkt.Packet, &incomingVP8, tp.codecBytes, pool) + payload = d.translateVP8PacketTo(extPkt.Packet, &incomingVP8, tp.codecBytes, poolEntity) } } if payload == nil { - payload = (*pool)[:len(extPkt.Packet.Payload)] + payload = (*poolEntity)[:len(extPkt.Packet.Payload)] copy(payload, extPkt.Packet.Payload) } hdr, err := d.getTranslatedRTPHeader(extPkt, tp) if err != nil { d.params.Logger.Errorw("write rtp packet failed", err) - if pool != nil { - PacketFactory.Put(pool) + if poolEntity != nil { + PacketFactory.Put(poolEntity) } return err } @@ -715,6 +715,18 @@ func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) error { ) } + d.sendingPacket( + hdr, + len(payload), + &sendPacketMetadata{ + layer: layer, + packetTime: extPkt.Arrival, + extSequenceNumber: tp.rtp.extSequenceNumber, + extTimestamp: tp.rtp.extTimestamp, + isKeyFrame: extPkt.KeyFrame, + tp: tp, + }, + ) d.pacer.Enqueue(pacer.Packet{ Header: hdr, Extensions: extensions, @@ -722,16 +734,8 @@ func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) error { AbsSendTimeExtID: uint8(d.absSendTimeExtID), TransportWideExtID: uint8(d.transportWideExtID), WriteStream: d.writeStream, - Metadata: sendPacketMetadata{ - layer: layer, - arrival: extPkt.Arrival, - extSequenceNumber: tp.rtp.extSequenceNumber, - extTimestamp: tp.rtp.extTimestamp, - isKeyFrame: extPkt.KeyFrame, - tp: tp, - pool: pool, - }, - OnSent: d.packetSent, + Pool: PacketFactory, + PoolEntity: poolEntity, }) return nil } @@ -809,19 +813,22 @@ func (d *DownTrack) WritePaddingRTP(bytesToSend int, paddingOnMute bool, forceMa // last byte of padding has padding size including that byte payload[RTPPaddingMaxPayloadSize-1] = byte(RTPPaddingMaxPayloadSize) + d.sendingPacket( + &hdr, + len(payload), + &sendPacketMetadata{ + extSequenceNumber: snts[i].extSequenceNumber, + extTimestamp: snts[i].extTimestamp, + isPadding: true, + shouldDisableCounter: true, + }, + ) d.pacer.Enqueue(pacer.Packet{ Header: &hdr, Payload: payload, AbsSendTimeExtID: uint8(d.absSendTimeExtID), TransportWideExtID: uint8(d.transportWideExtID), WriteStream: d.writeStream, - Metadata: sendPacketMetadata{ - extSequenceNumber: snts[i].extSequenceNumber, - extTimestamp: snts[i].extTimestamp, - isPadding: true, - disableCounter: true, - }, - OnSent: d.packetSent, }) bytesSent += hdr.MarshalSize() + len(payload) @@ -1333,17 +1340,16 @@ func (d *DownTrack) writeBlankFrameRTP(duration float32, generation uint32) chan return } + d.sendingPacket(&hdr, len(payload), &sendPacketMetadata{ + extSequenceNumber: snts[i].extSequenceNumber, + extTimestamp: snts[i].extTimestamp, + }) d.pacer.Enqueue(pacer.Packet{ Header: &hdr, Payload: payload, AbsSendTimeExtID: uint8(d.absSendTimeExtID), TransportWideExtID: uint8(d.transportWideExtID), WriteStream: d.writeStream, - Metadata: sendPacketMetadata{ - extSequenceNumber: snts[i].extSequenceNumber, - extTimestamp: snts[i].extTimestamp, - }, - OnSent: d.packetSent, }) // only the first frame will need frameEndNeeded to close out the @@ -1607,22 +1613,33 @@ func (d *DownTrack) retransmitPackets(nacks []uint16) { pkt.Header.PayloadType = d.payloadType var payload []byte - pool := PacketFactory.Get().(*[]byte) + poolEntity := PacketFactory.Get().(*[]byte) if d.mime == "video/vp8" && len(pkt.Payload) > 0 && len(epm.codecBytes) != 0 { var incomingVP8 buffer.VP8 if err = incomingVP8.Unmarshal(pkt.Payload); err != nil { d.params.Logger.Errorw("unmarshalling VP8 packet err", err) - PacketFactory.Put(pool) + PacketFactory.Put(poolEntity) continue } - payload = d.translateVP8PacketTo(&pkt, &incomingVP8, epm.codecBytes, pool) + payload = d.translateVP8PacketTo(&pkt, &incomingVP8, epm.codecBytes, poolEntity) } if payload == nil { - payload = (*pool)[:len(pkt.Payload)] + payload = (*poolEntity)[:len(pkt.Payload)] copy(payload, pkt.Payload) } + d.sendingPacket( + &pkt.Header, + len(payload), + &sendPacketMetadata{ + layer: int32(epm.layer), + packetTime: time.Now(), + extSequenceNumber: epm.extSequenceNumber, + extTimestamp: epm.extTimestamp, + isRTX: true, + }, + ) d.pacer.Enqueue(pacer.Packet{ Header: &pkt.Header, Extensions: []pacer.ExtensionData{{ID: uint8(d.dependencyDescriptorExtID), Payload: epm.ddBytes}}, @@ -1630,13 +1647,8 @@ func (d *DownTrack) retransmitPackets(nacks []uint16) { AbsSendTimeExtID: uint8(d.absSendTimeExtID), TransportWideExtID: uint8(d.transportWideExtID), WriteStream: d.writeStream, - Metadata: sendPacketMetadata{ - extSequenceNumber: epm.extSequenceNumber, - extTimestamp: epm.extTimestamp, - isRTX: true, - pool: pool, - }, - OnSent: d.packetSent, + Pool: PacketFactory, + PoolEntity: poolEntity, }) } @@ -1839,19 +1851,22 @@ func (d *DownTrack) sendSilentFrameOnMuteForOpus() { return } + d.sendingPacket( + &hdr, + len(payload), + &sendPacketMetadata{ + extSequenceNumber: snts[i].extSequenceNumber, + extTimestamp: snts[i].extTimestamp, + // although this is using empty frames, mark as padding as these are used to trigger Pion OnTrack only + isPadding: true, + }, + ) d.pacer.Enqueue(pacer.Packet{ Header: &hdr, Payload: payload, AbsSendTimeExtID: uint8(d.absSendTimeExtID), TransportWideExtID: uint8(d.transportWideExtID), WriteStream: d.writeStream, - Metadata: sendPacketMetadata{ - extSequenceNumber: snts[i].extSequenceNumber, - extTimestamp: snts[i].extTimestamp, - // although this is using empty frames, mark as padding as these are used to trigger Pion OnTrack only - isPadding: true, - }, - OnSent: d.packetSent, }) } @@ -1868,34 +1883,20 @@ func (d *DownTrack) HandleRTCPSenderReportData(_payloadType webrtc.PayloadType, } type sendPacketMetadata struct { - layer int32 - arrival time.Time - extSequenceNumber uint64 - extTimestamp uint64 - isKeyFrame bool - isRTX bool - isPadding bool - disableCounter bool - tp *TranslationParams - pool *[]byte + layer int32 + packetTime time.Time + extSequenceNumber uint64 + extTimestamp uint64 + isKeyFrame bool + isRTX bool + isPadding bool + shouldDisableCounter bool + tp *TranslationParams } -func (d *DownTrack) packetSent(md interface{}, marker bool, hdrSize int, payloadSize int, sendTime time.Time, sendError error) { - spmd, ok := md.(sendPacketMetadata) - if !ok { - d.params.Logger.Errorw("invalid send packet metadata", nil) - return - } - - if spmd.pool != nil { - PacketFactory.Put(spmd.pool) - } - - if sendError != nil { - return - } - - if !spmd.disableCounter { +func (d *DownTrack) sendingPacket(hdr *rtp.Header, payloadSize int, spmd *sendPacketMetadata) { + hdrSize := hdr.MarshalSize() + if !spmd.shouldDisableCounter { // STREAM-ALLOCATOR-TODO: remove this stream allocator bytes counter once stream allocator changes fully to pull bytes counter size := uint32(hdrSize + payloadSize) d.streamAllocatorBytesCounter.Add(size) @@ -1907,14 +1908,10 @@ func (d *DownTrack) packetSent(md interface{}, marker bool, hdrSize int, payload } // update RTPStats - packetTime := spmd.arrival - if packetTime.IsZero() { - packetTime = sendTime - } if spmd.isPadding { - d.rtpStats.Update(packetTime, spmd.extSequenceNumber, spmd.extTimestamp, marker, hdrSize, 0, payloadSize) + d.rtpStats.Update(spmd.packetTime, spmd.extSequenceNumber, spmd.extTimestamp, hdr.Marker, hdrSize, 0, payloadSize) } else { - d.rtpStats.Update(packetTime, spmd.extSequenceNumber, spmd.extTimestamp, marker, hdrSize, payloadSize, 0) + d.rtpStats.Update(spmd.packetTime, spmd.extSequenceNumber, spmd.extTimestamp, hdr.Marker, hdrSize, payloadSize, 0) } if spmd.isKeyFrame { diff --git a/pkg/sfu/pacer/base.go b/pkg/sfu/pacer/base.go index a43ea6cd7..632a7e200 100644 --- a/pkg/sfu/pacer/base.go +++ b/pkg/sfu/pacer/base.go @@ -43,15 +43,13 @@ func (b *Base) SetBitrate(_bitrate int) { } func (b *Base) SendPacket(p *Packet) (int, error) { - var sendingAt time.Time - var err error defer func() { - if p.OnSent != nil { - p.OnSent(p.Metadata, p.Header.Marker, p.Header.MarshalSize(), len(p.Payload), sendingAt, err) + if p.Pool != nil && p.PoolEntity != nil { + p.Pool.Put(p.PoolEntity) } }() - sendingAt, err = b.writeRTPHeaderExtensions(p) + _, err := b.writeRTPHeaderExtensions(p) if err != nil { b.logger.Errorw("writing rtp header extensions err", err) return 0, err diff --git a/pkg/sfu/pacer/pacer.go b/pkg/sfu/pacer/pacer.go index 1464ea496..83e79a376 100644 --- a/pkg/sfu/pacer/pacer.go +++ b/pkg/sfu/pacer/pacer.go @@ -15,6 +15,7 @@ package pacer import ( + "sync" "time" "github.com/pion/rtp" @@ -33,8 +34,8 @@ type Packet struct { AbsSendTimeExtID uint8 TransportWideExtID uint8 WriteStream webrtc.TrackLocalWriter - Metadata interface{} - OnSent func(md interface{}, marker bool, hdrSize int, payloadSize int, sentTime time.Time, sendError error) + Pool *sync.Pool + PoolEntity *[]byte } type Pacer interface {