Internal bytes pool

Signed-off-by: Valery Piashchynski <[email protected]>

.golangci.yml

@@ -54,7 +54,6 @@ linters: # All available linters list: <https://golangci-lint.run/usage/linters/
     - errorlint # find code that will cause problems with the error wrapping scheme introduced in Go 1.13
     - exhaustive # check exhaustiveness of enum switch statements
     - exportloopref # checks for pointers to enclosing loop variables
-    - gochecknoglobals # Checks that no globals are present in Go code
     - gochecknoinits # Checks that no init functions are present in Go code
     - goconst # Finds repeated strings that could be replaced by a constant
     - gocritic # The most opinionated Go source code linter

internal/bpool.go

@@ -0,0 +1,77 @@
+package internal
+
+import (
+	"sync"
+)
+
+var frameChunkedPool = &sync.Map{}
+var preallocate = &sync.Once{}
+
+const (
+	OneMB  uint32 = 1024 * 1024 * 1
+	FiveMB uint32 = 1024 * 1024 * 5
+	TenMB  uint32 = 1024 * 1024 * 10
+)
+
+func Preallocate() {
+	preallocate.Do(internalAllocate)
+}
+
+func internalAllocate() {
+	pool1 := &sync.Pool{
+		New: func() interface{} {
+			data := make([]byte, OneMB)
+			return &data
+		},
+	}
+	pool5 := &sync.Pool{
+		New: func() interface{} {
+			data := make([]byte, FiveMB)
+			return &data
+		},
+	}
+	pool10 := &sync.Pool{
+		New: func() interface{} {
+			data := make([]byte, TenMB)
+			return &data
+		},
+	}
+
+	frameChunkedPool.Store(OneMB, pool1)
+	frameChunkedPool.Store(FiveMB, pool5)
+	frameChunkedPool.Store(TenMB, pool10)
+}
+
+func get(size uint32) *[]byte {
+	switch {
+	case size <= OneMB:
+		val, _ := frameChunkedPool.Load(OneMB)
+		return val.(*sync.Pool).Get().(*[]byte)
+	case size <= FiveMB:
+		val, _ := frameChunkedPool.Load(FiveMB)
+		return val.(*sync.Pool).Get().(*[]byte)
+	case size <= TenMB:
+		val, _ := frameChunkedPool.Load(TenMB)
+		return val.(*sync.Pool).Get().(*[]byte)
+	default:
+		data := make([]byte, size)
+		return &data
+	}
+}
+
+func put(size uint32, data *[]byte) {
+	switch {
+	case size <= OneMB:
+		pool, _ := frameChunkedPool.Load(OneMB)
+		pool.(*sync.Pool).Put(data)
+		return
+	case size <= FiveMB:
+		pool, _ := frameChunkedPool.Load(FiveMB)
+		pool.(*sync.Pool).Put(data)
+		return
+	default:
+		pool, _ := frameChunkedPool.Load(TenMB)
+		pool.(*sync.Pool).Put(data)
+		return
+	}
+}

internal/receive.go

@@ -56,12 +56,14 @@ func ReceiveFrame(relay io.Reader, fr *frame.Frame) error {
 		return nil
 	}
 
-	pb := make([]byte, pl)
-	_, err = io.ReadFull(relay, pb)
+	pb := get(pl)
+	_, err = io.ReadFull(relay, (*pb)[:pl])
 	if err != nil {
+		put(pl, pb)
 		return errors.E(op, err)
 	}
 
-	fr.WritePayload(pb)
+	fr.WritePayload((*pb)[:pl])
+	put(pl, pb)
 	return nil
 }

pkg/frame/frame.go

@@ -428,16 +428,8 @@ func (f *Frame) Payload() []byte {
 	return f.payload
 }
 
-func (f *Frame) AllocatePayload(l int) {
-	f.payload = make([]byte, l)
-}
-
 // WritePayload writes payload
 func (f *Frame) WritePayload(data []byte) {
-	if len(f.payload) == len(data) {
-		copy(f.payload, data)
-		return
-	}
 	f.payload = make([]byte, len(data))
 	copy(f.payload, data)
 }

pkg/pipe/pipe.go

@@ -17,6 +17,7 @@ type Relay struct {
 
 // NewPipeRelay creates new pipe based data relay.
 func NewPipeRelay(in io.ReadCloser, out io.WriteCloser) *Relay {
+	internal.Preallocate()
 	return &Relay{in: in, out: out}
 }
 

pkg/socket/socket.go

@@ -15,6 +15,7 @@ type Relay struct {
 
 // NewSocketRelay creates new socket based data relay.
 func NewSocketRelay(rwc io.ReadWriteCloser) *Relay {
+	internal.Preallocate()
 	return &Relay{rwc: rwc}
 }