api: updated response/future methods

Added method Release to Future and Response's interface,
that allows to free used data directly by calling.

Fixes #493

Author: ermyar

CHANGELOG.md

@@ -12,6 +12,8 @@ Versioning](http://semver.org/spec/v2.0.0.html) except to the first release.
 
 * New types for MessagePack extensions compatible with go-option (#459).
 * Added `box.MustNew` wrapper for `box.New` without an error (#448).
+* Method `Release` for `Future` and `Response` interface that allows
+  to free used data directly by calling (#493).
 
 ### Changed
 

connection.go

@@ -39,6 +39,18 @@ var (
 		"to the current connection or connection pool")
 )
 
+var smallBufPool = &sync.Pool{
+	New: func() interface{} {
+		return &smallBuf{}
+	},
+}
+
+var slicePool = &sync.Pool{
+	New: func() interface{} {
+		return make([]byte, 1024)
+	},
+}
+
 const (
 	// Connected signals that connection is established or reestablished.
 	Connected ConnEventKind = iota + 1
@@ -373,7 +385,6 @@ func Connect(ctx context.Context, dialer Dialer, opts Opts) (conn *Connection, e
 	}
 
 	conn.cond = sync.NewCond(&conn.mutex)
-
 	if conn.opts.Reconnect > 0 {
 		// We don't need these mutex.Lock()/mutex.Unlock() here, but
 		// runReconnects() expects mutex.Lock() to be set, so it's
@@ -849,7 +860,7 @@ func (conn *Connection) reader(r io.Reader, c Conn) {
 	go conn.eventer(events)
 
 	for atomic.LoadUint32(&conn.state) != connClosed {
-		respBytes, err := read(r, conn.lenbuf[:])
+		respBytes, err := read(r, conn.lenbuf[:], conn)
 		if err != nil {
 			err = ClientError{
 				ErrIoError,
@@ -858,8 +869,9 @@ func (conn *Connection) reader(r io.Reader, c Conn) {
 			conn.reconnect(err, c)
 			return
 		}
-		buf := smallBuf{b: respBytes}
-		header, code, err := decodeHeader(conn.dec, &buf)
+		buf := smallBufPool.Get().(*smallBuf)
+		buf.b = respBytes
+		header, code, err := decodeHeader(conn.dec, buf)
 		if err != nil {
 			err = ClientError{
 				ErrProtocolError,
@@ -871,7 +883,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{
@@ -885,7 +897,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)
@@ -925,7 +937,7 @@ func (conn *Connection) eventer(events <-chan connWatchEvent) {
 
 func (conn *Connection) newFuture(req Request) (fut *Future) {
 	ctx := req.Ctx()
-	fut = NewFuture(req)
+	fut = NewFuture(req, conn)
 	if conn.rlimit != nil && conn.opts.RLimitAction == RLimitDrop {
 		select {
 		case conn.rlimit <- struct{}{}:
@@ -1187,7 +1199,7 @@ func (conn *Connection) timeouts() {
 	}
 }
 
-func read(r io.Reader, lenbuf []byte) (response []byte, err error) {
+func read(r io.Reader, lenbuf []byte, conn ...*Connection) (response []byte, err error) {
 	var length uint64
 
 	if _, err = io.ReadFull(r, lenbuf); err != nil {
@@ -1211,7 +1223,18 @@ func read(r io.Reader, lenbuf []byte) (response []byte, err error) {
 		return
 	}
 
-	response = make([]byte, length)
+	if len(conn) == 0 {
+		response = make([]byte, length)
+	} else {
+		ptr := slicePool.Get().([]byte)
+		if cap(ptr) < int(length) {
+			response = make([]byte, length)
+			slicePool.Put(ptr) // nolint
+		} else {
+			response = ptr
+			response = response[:length]
+		}
+	}
 	_, err = io.ReadFull(r, response)
 
 	return
@@ -1232,7 +1255,7 @@ func (conn *Connection) nextRequestId(context bool) (requestId uint32) {
 func (conn *Connection) Do(req Request) *Future {
 	if connectedReq, ok := req.(ConnectedRequest); ok {
 		if connectedReq.Conn() != conn {
-			fut := NewFuture(req)
+			fut := NewFuture(req, conn)
 			fut.SetError(errUnknownRequest)
 			return fut
 		}

example_test.go

@@ -1291,6 +1291,7 @@ func ExampleConnection_Do_failure() {
 
 	// We got a future, the request actually not performed yet.
 	future := conn.Do(req)
+	defer future.Release()
 
 	// When the future receives the response, the result of the Future is set
 	// and becomes available. We could wait for that moment with Future.Get(),

future.go

@@ -38,7 +38,7 @@ func (fut *Future) isDone() bool {
 }
 
 // NewFuture creates a new empty Future for a given Request.
-func NewFuture(req Request) (fut *Future) {
+func NewFuture(req Request, conn ...*Connection) (fut *Future) {
 	fut = &Future{}
 	fut.done = make(chan struct{})
 	fut.req = req
@@ -89,6 +89,7 @@ func (fut *Future) GetResponse() (Response, error) {
 }
 
 // Get waits for Future to be filled and returns the data of the Response and error.
+// Also Release Future's data. After this, Future becomes invalid.
 //
 // The data will be []interface{}, so if you want more performance, use GetTyped method.
 //
@@ -105,6 +106,8 @@ func (fut *Future) Get() ([]interface{}, error) {
 // GetTyped waits for Future and calls msgpack.Decoder.Decode(result) if no error happens.
 // It is could be much faster than Get() function.
 //
+// Also Release Future's data. After this, Future becomes invalid.
+//
 // Note: Tarantool usually returns array of tuples (except for Eval and Call17 actions).
 func (fut *Future) GetTyped(result interface{}) error {
 	fut.wait()
@@ -127,3 +130,11 @@ func (fut *Future) WaitChan() <-chan struct{} {
 	}
 	return fut.done
 }
+
+// Release is freeing the Future resources.
+// After this, using this Future becomes invalid.
+func (fut *Future) Release() {
+	if fut.resp != nil {
+		fut.resp.Release()
+	}
+}

future_test.go

@@ -53,6 +53,10 @@ func (resp *futureMockResponse) Header() Header {
 	return resp.header
 }
 
+func (resp *futureMockResponse) Release() {
+	// Releasing futureMockResponse data.
+}
+
 func (resp *futureMockResponse) Decode() ([]interface{}, error) {
 	resp.decodeCnt++
 

prepared.go

@@ -90,7 +90,7 @@ func (req *PrepareRequest) Response(header Header, body io.Reader) (Response, er
 	if err != nil {
 		return nil, err
 	}
-	return &PrepareResponse{baseResponse: baseResp}, nil
+	return &PrepareResponse{baseResponse: *baseResp}, nil
 }
 
 // UnprepareRequest helps you to create an unprepare request object for
@@ -204,5 +204,5 @@ func (req *ExecutePreparedRequest) Response(header Header, body io.Reader) (Resp
 	if err != nil {
 		return nil, err
 	}
-	return &ExecuteResponse{baseResponse: baseResp}, nil
+	return &ExecuteResponse{baseResponse: *baseResp}, nil
 }

request.go

@@ -620,11 +620,11 @@ 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
@@ -1154,7 +1154,7 @@ func (req *ExecuteRequest) Response(header Header, body io.Reader) (Response, er
 	if err != nil {
 		return nil, err
 	}
-	return &ExecuteResponse{baseResponse: baseResp}, nil
+	return &ExecuteResponse{baseResponse: *baseResp}, nil
 }
 
 // WatchOnceRequest synchronously fetches the value currently associated with a

response.go

@@ -3,6 +3,7 @@ package tarantool
 import (
 	"fmt"
 	"io"
+	"sync"
 
 	"github.com/tarantool/go-iproto"
 	"github.com/vmihailenco/msgpack/v5"
@@ -12,6 +13,8 @@ import (
 type Response interface {
 	// Header returns a response header.
 	Header() Header
+	// Release free responses data and returns buffer's data.
+	Release()
 	// Decode decodes a response.
 	Decode() ([]interface{}, error)
 	// DecodeTyped decodes a response into a given container res.
@@ -31,24 +34,42 @@ type baseResponse struct {
 	err          error
 }
 
-func createBaseResponse(header Header, body io.Reader) (baseResponse, error) {
+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
+		return resp, nil
 	}
 	data, err := io.ReadAll(body)
 	if err != nil {
-		return baseResponse{}, 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) clear() {
+	*resp = baseResponse{}
+}
+
+func (resp *baseResponse) Release() {
+	slicePool.Put(resp.buf.b) // nolint
+	resp.buf.b = nil
+	resp.buf.p = 0
+	smallBufPool.Put(&resp.buf)
 }
 
 // DecodeBaseResponse parse response header and body.
 func DecodeBaseResponse(header Header, body io.Reader) (Response, error) {
 	resp, err := createBaseResponse(header, body)
-	return &resp, err
+	return resp, err
 }
 
 // SelectResponse is used for the select requests.
@@ -657,6 +678,41 @@ func (resp *baseResponse) Header() Header {
 	return resp.header
 }
 
+var selectResponsePool *sync.Pool = &sync.Pool{
+	New: func() interface{} {
+		return &SelectResponse{}
+	},
+}
+
+func createSelectResponse(header Header, body io.Reader) (*SelectResponse, error) {
+	resp := selectResponsePool.Get().(*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
+		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.baseResponse.clear()
+	resp.pos = nil
+
+	selectResponsePool.Put(resp)
+}
+
 // Pos returns a position descriptor of the last selected tuple for the SelectResponse.
 // If the response was not decoded, this method will call Decode().
 func (resp *SelectResponse) Pos() ([]byte, error) {

tarantool_test.go

@@ -220,14 +220,15 @@ func BenchmarkSync_naive_with_custom_type(b *testing.B) {
 	b.ResetTimer()
 
 	for b.Loop() {
-		err := conn.Do(req).GetTyped(&tuple)
-		if err != nil {
+		fut := conn.Do(req)
+		if err := fut.GetTyped(&tuple); err != nil {
 			b.Errorf("request error: %s", err)
 		}
 
 		if tuple.id != 1111 {
 			b.Errorf("invalid result")
 		}
+		fut.Release()
 	}
 }
 
@@ -529,6 +530,7 @@ func TestFutureMultipleGetGetTyped(t *testing.T) {
 	defer conn.Close()
 
 	fut := conn.Do(NewCall17Request("simple_concat").Args([]interface{}{"1"}))
+	defer fut.Release()
 
 	for i := 0; i < 30; i++ {
 		// [0, 10) fut.Get()
@@ -567,6 +569,7 @@ func TestFutureMultipleGetWithError(t *testing.T) {
 	defer conn.Close()
 
 	fut := conn.Do(NewCall17Request("non_exist").Args([]interface{}{"1"}))
+	defer fut.Release()
 
 	for i := 0; i < 2; i++ {
 		if _, err := fut.Get(); err == nil {
@@ -580,6 +583,7 @@ func TestFutureMultipleGetTypedWithError(t *testing.T) {
 	defer conn.Close()
 
 	fut := conn.Do(NewCall17Request("simple_concat").Args([]interface{}{"1"}))
+	defer fut.Release()
 
 	wrongTpl := struct {
 		Val int

test_helpers/response.go

@@ -54,6 +54,10 @@ func (resp *MockResponse) Header() tarantool.Header {
 	return resp.header
 }
 
+func (resp *MockResponse) Release() {
+	// Releasing MockResponse data
+}
+
 // Decode returns the result of decoding the response data as slice.
 func (resp *MockResponse) Decode() ([]interface{}, error) {
 	if resp.data == nil {