api: added future/response releasing methods

Added method Release to Future and Response's interface,
that allows to free used data directly by calling.
Used to reduce allocations in `reader` function.

Closes #493

Author: ermyar

connection.go

@@ -39,6 +39,13 @@ var (
 		"to the current connection or connection pool")
 )
 
+var slicePool = &sync.Pool{
+	New: func() interface{} {
+		buf := make([]byte, 1024)
+		return &buf
+	},
+}
+
 const (
 	// Connected signals that connection is established or reestablished.
 	Connected ConnEventKind = iota + 1
@@ -843,12 +850,36 @@ func readWatchEvent(reader io.Reader) (connWatchEvent, error) {
 	return event, nil
 }
 
+func getSlice(length int) *[]byte {
+	bs, ok := slicePool.Get().(*[]byte)
+
+	if ok && cap(*bs) >= length {
+		clear((*bs)[:cap(*bs)])
+		*bs = (*bs)[:length]
+		return bs
+	}
+
+	if ok {
+		slicePool.Put(bs)
+	}
+
+	b := make([]byte, length)
+
+	return &b
+}
+
+func putSlice(buf *[]byte) {
+	slicePool.Put(buf)
+}
+
 func (conn *Connection) reader(r io.Reader, c Conn) {
 	events := make(chan connWatchEvent, 1024)
 	defer close(events)
 
 	go conn.eventer(events)
 
+	buf := &smallBuf{}
+
 	for atomic.LoadUint32(&conn.state) != connClosed {
 		respBytes, err := read(r, conn.lenbuf[:])
 		if err != nil {
@@ -860,8 +891,11 @@ func (conn *Connection) reader(r io.Reader, c Conn) {
 			return
 		}
 
-		buf := smallBuf{b: respBytes}
-		header, code, err := decodeHeader(conn.dec, &buf)
+		buf.b = *respBytes
+		buf.p = 0
+		buf.ptr = respBytes
+
+		header, code, err := decodeHeader(conn.dec, buf)
 
 		if err != nil {
 			err = ClientError{
@@ -874,7 +908,7 @@ func (conn *Connection) reader(r io.Reader, c Conn) {
 
 		var fut *future = nil
 		if code == iproto.IPROTO_EVENT {
-			if event, err := readWatchEvent(&buf); err == nil {
+			if event, err := readWatchEvent(buf); err == nil {
 				events <- event
 			} else {
 				err = ClientError{
@@ -888,7 +922,7 @@ func (conn *Connection) reader(r io.Reader, c Conn) {
 			conn.opts.Logger.Report(LogBoxSessionPushUnsupported, conn, header)
 		} else {
 			if fut = conn.fetchFuture(header.RequestId); fut != nil {
-				if err := fut.setResponse(header, &buf); err != nil {
+				if err := fut.setResponse(header, buf); err != nil {
 					fut.setError(fmt.Errorf("failed to set response: %w", err))
 				}
 				conn.markDone(fut)
@@ -1191,7 +1225,9 @@ func (conn *Connection) timeouts() {
 	}
 }
 
-func read(r io.Reader, lenbuf []byte) (response []byte, err error) {
+// read uses args to allocate slices for responses using sync.Pool.
+// data must be released later using Release.
+func read(r io.Reader, lenbuf []byte) (response *[]byte, err error) {
 	var length uint64
 
 	if _, err = io.ReadFull(r, lenbuf); err != nil {
@@ -1215,8 +1251,8 @@ func read(r io.Reader, lenbuf []byte) (response []byte, err error) {
 		return
 	}
 
-	response = make([]byte, length)
-	_, err = io.ReadFull(r, response)
+	response = getSlice(int(length))
+	_, err = io.ReadFull(r, *response)
 
 	return
 }

dial.go

@@ -623,17 +623,17 @@ func readResponse(ctx context.Context, conn Conn, req Request) (Response, error)
 		return nil, fmt.Errorf("read error: %w", err)
 	}
 
-	buf := smallBuf{b: respBytes, p: 0}
+	buf := &smallBuf{b: *respBytes, p: 0}
 
-	d := getDecoder(&buf)
+	d := getDecoder(buf)
 	defer putDecoder(d)
 
-	header, _, err := decodeHeader(d, &buf)
+	header, _, err := decodeHeader(d, buf)
 	if err != nil {
 		return nil, fmt.Errorf("decode response header error: %w", err)
 	}
 
-	resp, err := req.Response(header, &buf)
+	resp, err := req.Response(header, buf)
 	if err != nil {
 		return nil, fmt.Errorf("creating response error: %w", err)
 	}

request.go

@@ -620,11 +620,12 @@ func (req *SelectRequest) Context(ctx context.Context) *SelectRequest {
 
 // Response creates a response for the SelectRequest.
 func (req *SelectRequest) Response(header Header, body io.Reader) (Response, error) {
-	baseResp, err := createBaseResponse(header, body)
+	SelectResp, err := createSelectResponse(header, body)
 	if err != nil {
 		return nil, err
 	}
-	return &SelectResponse{baseResponse: *baseResp}, nil
+
+	return SelectResp, nil
 }
 
 // InsertRequest helps you to create an insert request object for execution

response.go

@@ -34,20 +34,31 @@ type baseResponse struct {
 }
 
 func createBaseResponse(header Header, body io.Reader) (*baseResponse, error) {
+	resp := &baseResponse{}
 	if body == nil {
-		return &baseResponse{header: header}, nil
+		resp.header = header
+		return resp, nil
 	}
 	if buf, ok := body.(*smallBuf); ok {
-		return &baseResponse{header: header, buf: *buf}, nil
+		resp.header = header
+		resp.buf.b = buf.b
+		resp.buf.p = buf.p
+		resp.buf.ptr = buf.ptr
+		return resp, nil
 	}
 	data, err := io.ReadAll(body)
 	if err != nil {
-		return nil, err
+		return resp, err
 	}
-	return &baseResponse{header: header, buf: smallBuf{b: data}}, nil
+	resp.header = header
+	resp.buf.b = data
+	return resp, nil
 }
 
 func (resp *baseResponse) Release() {
+	if resp.buf.ptr != nil {
+		putSlice(resp.buf.ptr)
+	}
 	*resp = baseResponse{}
 }
 
@@ -676,6 +687,28 @@ func (resp *baseResponse) Header() Header {
 	return resp.header
 }
 
+func createSelectResponse(header Header, body io.Reader) (*SelectResponse, error) {
+	resp := &SelectResponse{}
+	if body == nil {
+		resp.header = header
+		return resp, nil
+	}
+	if buf, ok := body.(*smallBuf); ok {
+		resp.header = header
+		resp.buf.b = buf.b
+		resp.buf.p = buf.p
+		resp.buf.ptr = buf.ptr
+		return resp, nil
+	}
+	data, err := io.ReadAll(body)
+	if err != nil {
+		return resp, err
+	}
+	resp.header = header
+	resp.buf.b = data
+	return resp, nil
+}
+
 func (resp *SelectResponse) Release() {
 	resp.baseResponse.Release()
 	resp.pos = nil

smallbuf.go

@@ -6,8 +6,9 @@ import (
 )
 
 type smallBuf struct {
-	b []byte
-	p int
+	b   []byte
+	p   int
+	ptr *[]byte
 }
 
 func (s *smallBuf) Read(d []byte) (l int, err error) {