Reducing memory allocations in read RTCP methods

application_defined.go

@@ -120,3 +120,9 @@ func (a *ApplicationDefined) MarshalSize() int {
 	}
 	return 12 + dataLength + paddingSize
 }
+
+// Release returns the packet to its pool and resets it
+func (p *ApplicationDefined) Release() {
+	*p = ApplicationDefined{} // Reset the packet
+	applicationDefinedPool.Put(p)
+}

compound_packet.go

@@ -159,3 +159,11 @@ func (c CompoundPacket) String() string {
 	out = strings.TrimSuffix(strings.ReplaceAll(out, "\n", "\n\t"), "\t")
 	return out
 }
+
+// Release returns the packet to its pool and resets it
+func (p *CompoundPacket) Release() {
+	// CompoundPacket is a slice of pointers, so we need to release each one
+	for _, packet := range *p {
+		packet.Release()
+	}
+}

extended_report.go

@@ -657,3 +657,9 @@ func (x *ExtendedReport) DestinationSSRC() []uint32 {
 func (x *ExtendedReport) String() string {
 	return stringify(x)
 }
+
+// Release returns the packet to its pool and resets it
+func (p *ExtendedReport) Release() {
+	*p = ExtendedReport{} // Reset the packet
+	extendedReportPool.Put(p)
+}

full_intra_request.go

@@ -114,3 +114,9 @@ func (p *FullIntraRequest) DestinationSSRC() []uint32 {
 	}
 	return ssrcs
 }
+
+// Release returns the packet to its pool and resets it
+func (p *FullIntraRequest) Release() {
+	*p = FullIntraRequest{} // Reset the packet
+	fullIntraRequestPool.Put(p)
+}

goodbye.go

@@ -161,3 +161,9 @@ func (g Goodbye) String() string {
 
 	return out
 }
+
+// Release returns the packet to its pool and resets it
+func (p *Goodbye) Release() {
+	*p = Goodbye{} // Reset the packet
+	goodbyePool.Put(p)
+}

packet.go

@@ -3,6 +3,11 @@
 
 package rtcp
 
+import (
+	"bytes"
+	"sync"
+)
+
 // Packet represents an RTCP packet, a protocol used for out-of-band statistics and control information for an RTP session
 type Packet interface {
 	// DestinationSSRC returns an array of SSRC values that this packet refers to.
@@ -11,19 +16,48 @@ type Packet interface {
 	Marshal() ([]byte, error)
 	Unmarshal(rawPacket []byte) error
 	MarshalSize() int
+
+	// Release returns the packet to its pool
+	Release()
 }
 
+//nolint:gochecknoglobals
+var (
+	senderReportPool                    = sync.Pool{New: func() interface{} { return new(SenderReport) }}
+	receiverReportPool                  = sync.Pool{New: func() interface{} { return new(ReceiverReport) }}
+	sourceDescriptionPool               = sync.Pool{New: func() interface{} { return new(SourceDescription) }}
+	goodbyePool                         = sync.Pool{New: func() interface{} { return new(Goodbye) }}
+	transportLayerNackPool              = sync.Pool{New: func() interface{} { return new(TransportLayerNack) }}
+	rapidResynchronizationRequestPool   = sync.Pool{New: func() interface{} { return new(RapidResynchronizationRequest) }}
+	transportLayerCCPool                = sync.Pool{New: func() interface{} { return new(TransportLayerCC) }}
+	ccFeedbackReportPool                = sync.Pool{New: func() interface{} { return new(CCFeedbackReport) }}
+	pictureLossIndicationPool           = sync.Pool{New: func() interface{} { return new(PictureLossIndication) }}
+	sliceLossIndicationPool             = sync.Pool{New: func() interface{} { return new(SliceLossIndication) }}
+	receiverEstimatedMaximumBitratePool = sync.Pool{New: func() interface{} { return new(ReceiverEstimatedMaximumBitrate) }}
+	fullIntraRequestPool                = sync.Pool{New: func() interface{} { return new(FullIntraRequest) }}
+	extendedReportPool                  = sync.Pool{New: func() interface{} { return new(ExtendedReport) }}
+	applicationDefinedPool              = sync.Pool{New: func() interface{} { return new(ApplicationDefined) }}
+	rawPacketPool                       = sync.Pool{New: func() interface{} { return new(RawPacket) }}
+)
+
 // Unmarshal takes an entire udp datagram (which may consist of multiple RTCP packets) and
 // returns the unmarshaled packets it contains.
 //
 // If this is a reduced-size RTCP packet a feedback packet (Goodbye, SliceLossIndication, etc)
 // will be returned. Otherwise, the underlying type of the returned packet will be
 // CompoundPacket.
 func Unmarshal(rawData []byte) ([]Packet, error) {
-	var packets []Packet
+	// Preallocate a slice with a reasonable initial capacity
+	estimatedPackets := len(rawData) / 100 // Estimate based on average packet size
+	packets := make([]Packet, 0, estimatedPackets)
+
 	for len(rawData) != 0 {
 		p, processed, err := unmarshal(rawData)
 		if err != nil {
+			// Release already allocated packets in case of error
+			for _, packet := range packets {
+				packet.Release()
+			}
 			return nil, err
 		}
 
@@ -43,15 +77,16 @@ func Unmarshal(rawData []byte) ([]Packet, error) {
 
 // Marshal takes an array of Packets and serializes them to a single buffer
 func Marshal(packets []Packet) ([]byte, error) {
-	out := make([]byte, 0)
+	var buf bytes.Buffer
 	for _, p := range packets {
 		data, err := p.Marshal()
 		if err != nil {
 			return nil, err
 		}
-		out = append(out, data...)
+		buf.Write(data)
+		p.Release()
 	}
-	return out, nil
+	return buf.Bytes(), nil
 }
 
 // unmarshal is a factory which pulls the first RTCP packet from a bytestream,
@@ -72,53 +107,53 @@ func unmarshal(rawData []byte) (packet Packet, bytesprocessed int, err error) {
 
 	switch h.Type {
 	case TypeSenderReport:
-		packet = new(SenderReport)
+		packet = senderReportPool.Get().(*SenderReport)
 
 	case TypeReceiverReport:
-		packet = new(ReceiverReport)
+		packet = receiverReportPool.Get().(*ReceiverReport)
 
 	case TypeSourceDescription:
-		packet = new(SourceDescription)
+		packet = sourceDescriptionPool.Get().(*SourceDescription)
 
 	case TypeGoodbye:
-		packet = new(Goodbye)
+		packet = goodbyePool.Get().(*Goodbye)
 
 	case TypeTransportSpecificFeedback:
 		switch h.Count {
 		case FormatTLN:
-			packet = new(TransportLayerNack)
+			packet = transportLayerNackPool.Get().(*TransportLayerNack)
 		case FormatRRR:
-			packet = new(RapidResynchronizationRequest)
+			packet = rapidResynchronizationRequestPool.Get().(*RapidResynchronizationRequest)
 		case FormatTCC:
-			packet = new(TransportLayerCC)
+			packet = transportLayerCCPool.Get().(*TransportLayerCC)
 		case FormatCCFB:
-			packet = new(CCFeedbackReport)
+			packet = ccFeedbackReportPool.Get().(*CCFeedbackReport)
 		default:
-			packet = new(RawPacket)
+			packet = rawPacketPool.Get().(*RawPacket)
 		}
 
 	case TypePayloadSpecificFeedback:
 		switch h.Count {
 		case FormatPLI:
-			packet = new(PictureLossIndication)
+			packet = pictureLossIndicationPool.Get().(*PictureLossIndication)
 		case FormatSLI:
-			packet = new(SliceLossIndication)
+			packet = sliceLossIndicationPool.Get().(*SliceLossIndication)
 		case FormatREMB:
-			packet = new(ReceiverEstimatedMaximumBitrate)
+			packet = receiverEstimatedMaximumBitratePool.Get().(*ReceiverEstimatedMaximumBitrate)
 		case FormatFIR:
-			packet = new(FullIntraRequest)
+			packet = fullIntraRequestPool.Get().(*FullIntraRequest)
 		default:
-			packet = new(RawPacket)
+			packet = rawPacketPool.Get().(*RawPacket)
 		}
 
 	case TypeExtendedReport:
-		packet = new(ExtendedReport)
+		packet = extendedReportPool.Get().(*ExtendedReport)
 
 	case TypeApplicationDefined:
-		packet = new(ApplicationDefined)
+		packet = applicationDefinedPool.Get().(*ApplicationDefined)
 
 	default:
-		packet = new(RawPacket)
+		packet = rawPacketPool.Get().(*RawPacket)
 	}
 
 	err = packet.Unmarshal(inPacket)

packet_test.go

@@ -83,6 +83,21 @@ func realPacket() []byte {
 	}
 }
 
+func BenchmarkUnmarshal(b *testing.B) {
+	packetData := realPacket()
+	for i := 0; i < b.N; i++ {
+		pkts, err := Unmarshal(packetData)
+		if err != nil {
+			b.Fatalf("Error unmarshalling packets: %s", err)
+		}
+
+		for _, pkt := range pkts {
+			pkt.Release()
+		}
+
+	}
+}
+
 func TestUnmarshal(t *testing.T) {
 	packet, err := Unmarshal(realPacket())
 	if err != nil {
@@ -144,3 +159,111 @@ func TestInvalidHeaderLength(t *testing.T) {
 		t.Fatalf("Unmarshal(nil) err = %v, want %v", got, want)
 	}
 }
+
+func TestPacketPool(t *testing.T) {
+	t.Run("SenderReport", func(t *testing.T) {
+		sr := senderReportPool.Get()
+		p, ok := sr.(*SenderReport)
+		assert.True(t, ok)
+
+		p.Release()
+	})
+
+	t.Run("ReceiverReport", func(t *testing.T) {
+		rr := receiverReportPool.Get()
+		p, ok := rr.(*ReceiverReport)
+		assert.True(t, ok)
+		p.Release()
+	})
+
+	t.Run("SourceDescription", func(t *testing.T) {
+		sd := sourceDescriptionPool.Get()
+		p, ok := sd.(*SourceDescription)
+		assert.True(t, ok)
+		p.Release()
+	})
+
+	t.Run("Goodbye", func(t *testing.T) {
+		gb := goodbyePool.Get()
+		p, ok := gb.(*Goodbye)
+		assert.True(t, ok)
+		p.Release()
+	})
+
+	t.Run("TransportLayerNack", func(t *testing.T) {
+		tln := transportLayerNackPool.Get()
+		p, ok := tln.(*TransportLayerNack)
+		assert.True(t, ok)
+		p.Release()
+	})
+
+	t.Run("RapidResynchronizationRequest", func(t *testing.T) {
+		rrr := rapidResynchronizationRequestPool.Get()
+		p, ok := rrr.(*RapidResynchronizationRequest)
+		assert.True(t, ok)
+		p.Release()
+	})
+
+	t.Run("TransportLayerCC", func(t *testing.T) {
+		tcc := transportLayerCCPool.Get()
+		p, ok := tcc.(*TransportLayerCC)
+		assert.True(t, ok)
+		p.Release()
+	})
+
+	t.Run("CCFeedbackReport", func(t *testing.T) {
+		ccfb := ccFeedbackReportPool.Get()
+		p, ok := ccfb.(*CCFeedbackReport)
+		assert.True(t, ok)
+		p.Release()
+	})
+
+	t.Run("PictureLossIndication", func(t *testing.T) {
+		pli := pictureLossIndicationPool.Get()
+		p, ok := pli.(*PictureLossIndication)
+		assert.True(t, ok)
+		p.Release()
+	})
+
+	t.Run("SliceLossIndication", func(t *testing.T) {
+		sli := sliceLossIndicationPool.Get()
+		p, ok := sli.(*SliceLossIndication)
+		assert.True(t, ok)
+		p.Release()
+	})
+
+	t.Run("ReceiverEstimatedMaximumBitrate", func(t *testing.T) {
+		remb := receiverEstimatedMaximumBitratePool.Get()
+		p, ok := remb.(*ReceiverEstimatedMaximumBitrate)
+		assert.True(t, ok)
+		p.Release()
+	})
+
+	t.Run("FullIntraRequest", func(t *testing.T) {
+		fir := fullIntraRequestPool.Get()
+		p, ok := fir.(*FullIntraRequest)
+		assert.True(t, ok)
+		p.Release()
+	})
+
+	t.Run("ExtendedReport", func(t *testing.T) {
+		er := extendedReportPool.Get()
+		p, ok := er.(*ExtendedReport)
+		assert.True(t, ok)
+		p.Release()
+	})
+
+	t.Run("ApplicationDefined", func(t *testing.T) {
+		ad := applicationDefinedPool.Get()
+		p, ok := ad.(*ApplicationDefined)
+		assert.True(t, ok)
+		p.Release()
+	})
+
+	t.Run("RawPacket", func(t *testing.T) {
+		rp := rawPacketPool.Get()
+		p, ok := rp.(*RawPacket)
+		assert.True(t, ok)
+		p.Release()
+	})
+}

picture_loss_indication.go

@@ -91,3 +91,9 @@ func (p *PictureLossIndication) String() string {
 func (p *PictureLossIndication) DestinationSSRC() []uint32 {
 	return []uint32{p.MediaSSRC}
 }
+
+// Release returns the packet to its pool and resets it
+func (p *PictureLossIndication) Release() {
+	*p = PictureLossIndication{} // Reset the packet
+	pictureLossIndicationPool.Put(p)
+}

rapid_resynchronization_request.go

@@ -92,3 +92,9 @@ func (p *RapidResynchronizationRequest) DestinationSSRC() []uint32 {
 func (p *RapidResynchronizationRequest) String() string {
 	return fmt.Sprintf("RapidResynchronizationRequest %x %x", p.SenderSSRC, p.MediaSSRC)
 }
+
+// Release returns the packet to its pool and resets it
+func (p *RapidResynchronizationRequest) Release() {
+	*p = RapidResynchronizationRequest{} // Reset the packet
+	rapidResynchronizationRequestPool.Put(p)
+}

raw_packet.go

@@ -48,3 +48,9 @@ func (r RawPacket) String() string {
 func (r RawPacket) MarshalSize() int {
 	return len(r)
 }
+
+// Release returns the packet to its pool and resets it
+func (p *RawPacket) Release() {
+	*p = RawPacket{} // Reset the packet
+	rawPacketPool.Put(p)
+}

receiver_estimated_maximum_bitrate.go

@@ -283,3 +283,9 @@ func (p *ReceiverEstimatedMaximumBitrate) String() string {
 func (p *ReceiverEstimatedMaximumBitrate) DestinationSSRC() []uint32 {
 	return p.SSRCs
 }
+
+// Release returns the packet to its pool and resets it
+func (p *ReceiverEstimatedMaximumBitrate) Release() {
+	*p = ReceiverEstimatedMaximumBitrate{} // Reset the packet
+	receiverEstimatedMaximumBitratePool.Put(p)
+}

receiver_report.go

@@ -193,3 +193,9 @@ func (r ReceiverReport) String() string {
 	out += fmt.Sprintf("\tProfile Extension Data: %v\n", r.ProfileExtensions)
 	return out
 }
+
+// Release returns the packet to its pool and resets it
+func (p *ReceiverReport) Release() {
+	*p = ReceiverReport{} // Reset the packet
+	receiverReportPool.Put(p)
+}