Buffer optimizations

mqtt/mempool/bufpool.go

@@ -0,0 +1,81 @@
+package mempool
+
+import (
+	"bytes"
+	"sync"
+)
+
+var bufPool = NewBuffer(0)
+
+// GetBuffer takes a Buffer from the default buffer pool
+func GetBuffer() *bytes.Buffer { return bufPool.Get() }
+
+// PutBuffer returns Buffer to the default buffer pool
+func PutBuffer(x *bytes.Buffer) { bufPool.Put(x) }
+
+type BufferPool interface {
+	Get() *bytes.Buffer
+	Put(x *bytes.Buffer)
+}
+
+// NewBuffer returns a buffer pool. The max specify the max capacity of the Buffer the pool will
+// return. If the Buffer becoomes large than max, it will no longer be returned to the pool. If
+// max <= 0, no limit will be enforced.
+func NewBuffer(max int) BufferPool {
+	if max > 0 {
+		return newBufferWithCap(max)
+	}
+
+	return newBuffer()
+}
+
+// Buffer is a Buffer pool.
+type Buffer struct {
+	pool *sync.Pool
+}
+
+func newBuffer() *Buffer {
+	return &Buffer{
+		pool: &sync.Pool{
+			New: func() any { return new(bytes.Buffer) },
+		},
+	}
+}
+
+// Get a Buffer from the pool.
+func (b *Buffer) Get() *bytes.Buffer {
+	return b.pool.Get().(*bytes.Buffer)
+}
+
+// Put the Buffer back into pool. It resets the Buffer for reuse.
+func (b *Buffer) Put(x *bytes.Buffer) {
+	x.Reset()
+	b.pool.Put(x)
+}
+
+// BufferWithCap is a Buffer pool that
+type BufferWithCap struct {
+	bp  *Buffer
+	max int
+}
+
+func newBufferWithCap(max int) *BufferWithCap {
+	return &BufferWithCap{
+		bp:  newBuffer(),
+		max: max,
+	}
+}
+
+// Get a Buffer from the pool.
+func (b *BufferWithCap) Get() *bytes.Buffer {
+	return b.bp.Get()
+}
+
+// Put the Buffer back into the pool if the capacity doesn't exceed the limit. It resets the Buffer
+// for reuse.
+func (b *BufferWithCap) Put(x *bytes.Buffer) {
+	if x.Cap() > b.max {
+		return
+	}
+	b.bp.Put(x)
+}

mqtt/mempool/bufpool_test.go

@@ -0,0 +1,96 @@
+package mempool
+
+import (
+	"bytes"
+	"reflect"
+	"runtime/debug"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestNewBuffer(t *testing.T) {
+	defer debug.SetGCPercent(debug.SetGCPercent(-1))
+	bp := NewBuffer(1000)
+	require.Equal(t, "*mempool.BufferWithCap", reflect.TypeOf(bp).String())
+
+	bp = NewBuffer(0)
+	require.Equal(t, "*mempool.Buffer", reflect.TypeOf(bp).String())
+
+	bp = NewBuffer(-1)
+	require.Equal(t, "*mempool.Buffer", reflect.TypeOf(bp).String())
+}
+
+func TestBuffer(t *testing.T) {
+	defer debug.SetGCPercent(debug.SetGCPercent(-1))
+	Size := 101
+
+	bp := NewBuffer(0)
+	buf := bp.Get()
+
+	for i := 0; i < Size; i++ {
+		buf.WriteByte('a')
+	}
+
+	bp.Put(buf)
+	buf = bp.Get()
+	require.Equal(t, 0, buf.Len())
+}
+
+func TestBufferWithCap(t *testing.T) {
+	defer debug.SetGCPercent(debug.SetGCPercent(-1))
+	Size := 101
+	bp := NewBuffer(100)
+	buf := bp.Get()
+
+	for i := 0; i < Size; i++ {
+		buf.WriteByte('a')
+	}
+
+	bp.Put(buf)
+	buf = bp.Get()
+	require.Equal(t, 0, buf.Len())
+	require.Equal(t, 0, buf.Cap())
+}
+
+func BenchmarkBufferPool(b *testing.B) {
+	bp := NewBuffer(0)
+
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		b := bp.Get()
+		b.WriteString("this is a test")
+		bp.Put(b)
+	}
+}
+
+func BenchmarkBufferPoolWithCapLarger(b *testing.B) {
+	bp := NewBuffer(64 * 1024)
+
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		b := bp.Get()
+		b.WriteString("this is a test")
+		bp.Put(b)
+	}
+}
+
+func BenchmarkBufferPoolWithCapLesser(b *testing.B) {
+	bp := NewBuffer(10)
+
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		b := bp.Get()
+		b.WriteString("this is a test")
+		bp.Put(b)
+	}
+}
+
+func BenchmarkBufferWithoutPool(b *testing.B) {
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		b := new(bytes.Buffer)
+		b.WriteString("this is a test")
+		_ = b
+	}
+}