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

CHANGELOG.md

@@ -17,6 +17,8 @@ Versioning](http://semver.org/spec/v2.0.0.html) except to the first release.
 * Added Future.cond (sync.Cond) and Future.finished bool. Added Future.finish() marks Future as done (#496).
 * Added function String() for type datetime (#322).
 * New `Future` interface (#470).
+* Method `Release` for `Future` and `Response` interface that allows
+  to free used data directly by calling (#493).
 
 ### Changed
 

connection.go

@@ -373,7 +373,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
@@ -860,8 +859,10 @@ 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)
+
 		if err != nil {
 			err = ClientError{
 				ErrProtocolError,

dial.go

@@ -481,6 +481,7 @@ func identify(ctx context.Context, conn Conn) (ProtocolInfo, error) {
 		}
 		return info, err
 	}
+	defer resp.Release()
 	data, err := resp.Decode()
 	if err != nil {
 		return info, err
@@ -536,6 +537,7 @@ func checkProtocolInfo(required ProtocolInfo, actual ProtocolInfo) error {
 func authenticate(ctx context.Context, c Conn, auth Auth, user, pass, salt string) error {
 	var req Request
 	var err error
+	var resp Response
 
 	switch auth {
 	case ChapSha1Auth:
@@ -552,9 +554,10 @@ func authenticate(ctx context.Context, c Conn, auth Auth, user, pass, salt strin
 	if err = writeRequest(ctx, c, req); err != nil {
 		return err
 	}
-	if _, err = readResponse(ctx, c, req); err != nil {
+	if resp, err = readResponse(ctx, c, req); err != nil {
 		return err
 	}
+	resp.Release()
 	return nil
 }
 
@@ -620,7 +623,7 @@ func readResponse(ctx context.Context, conn Conn, req Request) (Response, error)
 		return nil, fmt.Errorf("read error: %w", err)
 	}
 
-	buf := smallBuf{b: respBytes}
+	buf := smallBuf{b: respBytes, p: 0}
 
 	d := getDecoder(&buf)
 	defer putDecoder(d)

example_test.go

@@ -1313,6 +1313,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

@@ -11,6 +11,7 @@ type Future interface {
 	Get() ([]interface{}, error)
 	GetTyped(result interface{}) error
 	GetResponse() (Response, error)
+	Release()
 	WaitChan() <-chan struct{}
 }
 
@@ -188,3 +189,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

@@ -49,12 +49,17 @@ type futureMockResponse struct {
 
 	decodeCnt      int
 	decodeTypedCnt int
+	released       bool
 }
 
 func (resp *futureMockResponse) Header() Header {
 	return resp.header
 }
 
+func (resp *futureMockResponse) Release() {
+	resp.released = true
+}
+
 func (resp *futureMockResponse) Decode() ([]interface{}, error) {
 	resp.decodeCnt++
 
@@ -133,6 +138,23 @@ func TestFuture_GetResponse(t *testing.T) {
 	assert.Equal(t, []interface{}{uint8('v'), uint8('2')}, data)
 }
 
+func TestFuture_Release(t *testing.T) {
+	fut, err := NewFutureWithResponse(&futureMockRequest{},
+		Header{}, bytes.NewReader([]byte{'v', '3'}))
+	assert.NoError(t, err)
+
+	resp, err := fut.GetResponse()
+	assert.NoError(t, err)
+	mockResp, ok := resp.(*futureMockResponse)
+	assert.True(t, ok)
+	assert.False(t, mockResp.released)
+
+	// Doing efficient work.
+
+	fut.Release()
+	assert.True(t, mockResp.released)
+}
+
 func BenchmarkFuture_Get(b *testing.B) {
 	fut, err := NewFutureWithResponse(&futureMockRequest{},
 		Header{}, bytes.NewReader([]byte{'v', '3'}))
@@ -219,6 +241,10 @@ func (*futureMock) WaitChan() <-chan struct{} {
 	return nil
 }
 
+func (*futureMock) Release() {
+	// Nothing to do.
+}
+
 func TestFuture(t *testing.T) {
 	fut := &futureMock{value: 5}
 

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

@@ -624,7 +624,7 @@ func (req *SelectRequest) Response(header Header, body io.Reader) (Response, err
 	if err != nil {
 		return nil, err
 	}
-	return &SelectResponse{baseResponse: baseResp}, nil
+	return &SelectResponse{baseResponse: *baseResp}, 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

@@ -12,6 +12,8 @@ import (
 type Response interface {
 	// Header returns a response header.
 	Header() Header
+	// Release free responses data.
+	Release()
 	// Decode decodes a response.
 	Decode() ([]interface{}, error)
 	// DecodeTyped decodes a response into a given container res.
@@ -31,24 +33,28 @@ type baseResponse struct {
 	err          error
 }
 
-func createBaseResponse(header Header, body io.Reader) (baseResponse, error) {
+func createBaseResponse(header Header, body io.Reader) (*baseResponse, error) {
 	if body == nil {
-		return baseResponse{header: header}, nil
+		return &baseResponse{header: header}, nil
 	}
 	if buf, ok := body.(*smallBuf); ok {
-		return baseResponse{header: header, buf: *buf}, nil
+		return &baseResponse{header: header, buf: *buf}, nil
 	}
 	data, err := io.ReadAll(body)
 	if err != nil {
-		return baseResponse{}, err
+		return nil, err
 	}
-	return baseResponse{header: header, buf: smallBuf{b: data}}, nil
+	return &baseResponse{header: header, buf: smallBuf{b: data}}, nil
+}
+
+func (resp *baseResponse) Release() {
+	*resp = baseResponse{}
 }
 
 // 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.
@@ -670,6 +676,11 @@ func (resp *baseResponse) Header() Header {
 	return resp.header
 }
 
+func (resp *SelectResponse) Release() {
+	resp.baseResponse.Release()
+	resp.pos = nil
+}
+
 // 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) {

response_test.go

@@ -54,3 +54,42 @@ func TestDecodeBaseResponse(t *testing.T) {
 		})
 	}
 }
+
+func TestBaseResponseRelease(t *testing.T) {
+	header := tarantool.Header{RequestId: 123}
+	buf := []byte{'v', '3'}
+
+	resp, err := tarantool.DecodeBaseResponse(header, encodeResponseData(t, buf))
+	require.NoError(t, err)
+
+	data, err := resp.Decode()
+	require.NoError(t, err)
+	require.Equal(t, resp.Header(), header)
+	require.Equal(t, []interface{}{buf}, data)
+
+	resp.Release()
+
+	data, err = resp.Decode()
+	require.NoError(t, err)
+	require.Equal(t, []interface{}(nil), data)
+}
+
+func TestSelectResponseRelease(t *testing.T) {
+	header := tarantool.Header{RequestId: 123}
+	buf := []byte{'v', '3'}
+	req := tarantool.NewSelectRequest(nil)
+
+	resp, err := req.Response(header, encodeResponseData(t, buf))
+	require.NoError(t, err)
+
+	data, err := resp.Decode()
+	require.NoError(t, err)
+	require.Equal(t, resp.Header(), header)
+	require.Equal(t, []interface{}{buf}, data)
+
+	resp.Release()
+
+	selResp, ok := resp.(*tarantool.SelectResponse)
+	require.True(t, ok)
+	require.Equal(t, tarantool.SelectResponse{}, *selResp)
+}