internal: add Release method for Execute and Prepare Responses

Now, ExecuteResponse.Release() PrepareResponse.Release() are freeing
Response resources by using sync.Pool.

Closes #558

Author: ermyar

prepared.go

@@ -90,8 +90,10 @@ func (req *PrepareRequest) Response(header Header, body io.Reader) (Response, er
 	if err != nil {
 		return nil, err
 	}
+	resp := preparesPool.Get().(*PrepareResponse)
+	resp.baseResponse = baseResp
 
-	return &PrepareResponse{baseResponse: baseResp}, nil
+	return resp, nil
 }
 
 // UnprepareRequest helps you to create an unprepare request object for
@@ -205,6 +207,8 @@ func (req *ExecutePreparedRequest) Response(header Header, body io.Reader) (Resp
 	if err != nil {
 		return nil, err
 	}
+	exec := executesPool.Get().(*ExecuteResponse)
+	exec.baseResponse = baseResp
 
-	return &ExecuteResponse{baseResponse: baseResp}, nil
+	return exec, nil
 }

request.go

@@ -1160,14 +1160,23 @@ func (req *ExecuteRequest) Context(ctx context.Context) *ExecuteRequest {
 	return req
 }
 
+var executesPool = sync.Pool{
+	New: func() interface{} {
+		return &ExecuteResponse{}
+	},
+}
+
 // Response creates a response for the ExecuteRequest.
 func (req *ExecuteRequest) Response(header Header, body io.Reader) (Response, error) {
 	baseResp, err := createBaseResponse(header, body)
 	if err != nil {
 		return nil, err
 	}
 
-	return &ExecuteResponse{baseResponse: baseResp}, nil
+	exec := executesPool.Get().(*ExecuteResponse)
+	exec.baseResponse = baseResp
+
+	return exec, 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"
@@ -79,6 +80,12 @@ type SelectResponse struct {
 // You need to cast to PrepareResponse a response from PrepareRequest.
 type PrepareResponse ExecuteResponse
 
+var preparesPool = sync.Pool{
+	New: func() interface{} {
+		return &PrepareResponse{}
+	},
+}
+
 // ExecuteResponse is used for the execute requests.
 // It might contain meta-data and sql info.
 //
@@ -699,6 +706,20 @@ func (resp *SelectResponse) Pos() ([]byte, error) {
 	return resp.pos, resp.err
 }
 
+func (resp *ExecuteResponse) Release() {
+	resp.baseResponse.Release()
+	*resp = ExecuteResponse{}
+
+	executesPool.Put(resp)
+}
+
+func (resp *PrepareResponse) Release() {
+	resp.baseResponse.Release()
+	*resp = PrepareResponse{}
+
+	preparesPool.Put(resp)
+}
+
 // MetaData returns ExecuteResponse meta-data.
 // If the response was not decoded, this method will call Decode().
 func (resp *ExecuteResponse) MetaData() ([]ColumnMetaData, error) {

response_test.go

@@ -154,3 +154,89 @@ func TestSelectResponseReleaseMultipleObjects(t *testing.T) {
 	require.Equal(t, []interface{}{buf3}, data3)
 	require.Equal(t, header3, resp3.Header())
 }
+
+func TestExecuteResponseRelease(t *testing.T) {
+	req := tarantool.NewExecuteRequest(insertQuery)
+
+	header := tarantool.Header{RequestId: 123}
+	buf := []byte{'v', '3'}
+	resp, err := req.Response(header, encodeResponseData(t, buf))
+
+	require.NoError(t, err)
+	require.NotNil(t, resp)
+
+	resp.Release()
+
+	execResp, ok := resp.(*tarantool.ExecuteResponse)
+	require.True(t, ok)
+	require.Equal(t, tarantool.ExecuteResponse{}, *execResp)
+}
+
+func TestExecuteResponseReleaseReuse(t *testing.T) {
+	req := tarantool.NewExecuteRequest(insertQuery)
+
+	header1 := tarantool.Header{RequestId: 100}
+	buf1 := []byte{'d', 'a', 't', 'a', '1'}
+	resp1, err := req.Response(header1, encodeResponseData(t, buf1))
+	require.NoError(t, err)
+
+	data1, err := resp1.Decode()
+	require.NoError(t, err)
+	require.Equal(t, []interface{}{buf1}, data1)
+
+	resp1.Release()
+
+	header2 := tarantool.Header{RequestId: 200}
+	buf2 := []byte{'d', 'a', 't', 'a', '2'}
+	resp2, err := req.Response(header2, encodeResponseData(t, buf2))
+	require.NoError(t, err)
+
+	data2, err := resp2.Decode()
+	require.NoError(t, err)
+	require.Equal(t, []interface{}{buf2}, data2)
+	require.Equal(t, header2, resp2.Header())
+	require.Equal(t, []interface{}{buf2}, data2)
+}
+
+func TestPrepareResponseRelease(t *testing.T) {
+	req := tarantool.NewPrepareRequest(insertQuery)
+
+	header := tarantool.Header{RequestId: 123}
+	buf := []byte{'v', '3'}
+	resp, err := req.Response(header, encodeResponseData(t, buf))
+
+	require.NoError(t, err)
+	require.NotNil(t, resp)
+
+	resp.Release()
+
+	prepResp, ok := resp.(*tarantool.PrepareResponse)
+	require.True(t, ok)
+	require.Equal(t, tarantool.PrepareResponse{}, *prepResp)
+}
+
+func TestPrepareResponseReleaseReuse(t *testing.T) {
+	req := tarantool.NewPrepareRequest(insertQuery)
+
+	header1 := tarantool.Header{RequestId: 100}
+	buf1 := []byte{'d', 'a', 't', 'a', '1'}
+	resp1, err := req.Response(header1, encodeResponseData(t, buf1))
+	require.NoError(t, err)
+
+	data1, err := resp1.Decode()
+	require.NoError(t, err)
+	require.Equal(t, []interface{}{buf1}, data1)
+
+	resp1.Release()
+
+	header2 := tarantool.Header{RequestId: 200}
+	buf2 := []byte{'d', 'a', 't', 'a', '2'}
+	resp2, err := req.Response(header2, encodeResponseData(t, buf2))
+	require.NoError(t, err)
+
+	data2, err := resp2.Decode()
+	require.NoError(t, err)
+	require.Equal(t, []interface{}{buf2}, data2)
+	require.Equal(t, header2, resp2.Header())
+	require.Equal(t, []interface{}{buf2}, data2)
+}

tarantool_test.go

@@ -271,6 +271,33 @@ func BenchmarkSync_naive_with_custom_type_with_Release(b *testing.B) {
 	}
 }
 
+func BenchmarkSync_execute_with_Release(b *testing.B) {
+	test_helpers.SkipIfSQLUnsupported(b)
+
+	conn := test_helpers.ConnectWithValidation(b, dialer, opts)
+	defer func() { _ = conn.Close() }()
+
+	var mem []Member
+
+	req := NewExecuteRequest(selectTypedQuery).Args(
+		[]interface{}{1},
+	)
+
+	b.ResetTimer()
+
+	for b.Loop() {
+		fut := conn.Do(req)
+		if err := fut.GetTyped(&mem); err != nil {
+			b.Errorf("request error: %s", err)
+		}
+
+		if len(mem) != 1 || mem[0].Name != "test" {
+			b.Errorf("invalid result, got: %v", mem)
+		}
+		fut.Release()
+	}
+}
+
 func BenchmarkSync_multithread(b *testing.B) {
 	var err error
 
@@ -1536,6 +1563,30 @@ func TestNewPrepared(t *testing.T) {
 	}
 }
 
+func TestExecutePrepareResponseRelease(t *testing.T) {
+	test_helpers.SkipIfSQLUnsupported(t)
+
+	conn := test_helpers.ConnectWithValidation(t, dialer, opts)
+
+	stmt, err := conn.NewPrepared(selectNamedQuery2)
+	if err != nil {
+		t.Errorf("failed to prepare: %v", err)
+	}
+
+	req := NewExecutePreparedRequest(stmt)
+
+	resp, err := conn.Do(req.Args([]interface{}{1, "test"})).GetResponse()
+	if err != nil {
+		t.Errorf("failed to execute prepared: %v", err)
+	}
+
+	resp.Release()
+
+	exPrepResp, ok := resp.(*ExecuteResponse)
+	require.True(t, ok, "got wrong response type")
+	require.Equal(t, ExecuteResponse{}, *exPrepResp)
+}
+
 func TestConnection_DoWithStrangerConn(t *testing.T) {
 	expectedErr := fmt.Errorf("the passed connected request doesn't belong to the current" +
 		" connection or connection pool")