api: proposal to add the context support

This patch adds the support of using context in API.
The proposed API is based on using request objects.
Added tests that cover almost all cases of using the context
in a query. Added benchamrk tests are equivalent to other,
that use the same query but without any context.

Closes #48

Author: vr009

CHANGELOG.md

@@ -13,6 +13,7 @@ Versioning](http://semver.org/spec/v2.0.0.html) except to the first release.
 - SSL support (#155)
 - IPROTO_PUSH messages support (#67)
 - Public API with request object types (#126)
+- Context support for request objects (#48)
 
 ### Changed
 

connection.go

@@ -5,6 +5,7 @@ package tarantool
 import (
 	"bufio"
 	"bytes"
+	"context"
 	"errors"
 	"fmt"
 	"io"
@@ -785,8 +786,10 @@ func (conn *Connection) newFuture() (fut *Future) {
 	return
 }
 
-func (conn *Connection) send(req Request) *Future {
+func (conn *Connection) send(ctx context.Context, req Request) *Future {
 	fut := conn.newFuture()
+	fut.ctx = ctx
+	fut.conn = conn
 	if fut.ready == nil {
 		return fut
 	}
@@ -992,26 +995,34 @@ func (conn *Connection) nextRequestId() (requestId uint32) {
 //
 // An error is returned if the request was formed incorrectly, or failure to
 // communicate by the connection, or unable to decode the response.
-func (conn *Connection) Do(req Request) (*Response, error) {
-	fut := conn.DoAsync(req)
+func (conn *Connection) Do(ctx context.Context, req Request) (*Response, error) {
+	fut := conn.DoAsync(ctx, req)
 	return fut.Get()
 }
 
 // DoTyped verifies, sends the request and fills the typed result.
 //
 // An error is returned if the request was formed incorrectly, or failure to
 // communicate by the connection, or unable to decode the response.
-func (conn *Connection) DoTyped(req Request, result interface{}) error {
-	fut := conn.DoAsync(req)
+func (conn *Connection) DoTyped(ctx context.Context, req Request, result interface{}) error {
+	fut := conn.DoAsync(ctx, req)
 	return fut.GetTyped(result)
 }
 
 // DoAsync verifies, sends the request and returns a future.
 //
 // An error is returned if the request was formed incorrectly, or failure to
 // create the future.
-func (conn *Connection) DoAsync(req Request) *Future {
-	return conn.send(req)
+func (conn *Connection) DoAsync(ctx context.Context, req Request) *Future {
+	if ctx == nil {
+		return NewErrorFuture(errors.New("passed nil context"))
+	}
+	select {
+	case <-ctx.Done():
+		return NewErrorFuture(errors.New("context is done"))
+	default:
+	}
+	return conn.send(ctx, req)
 }
 
 // ConfiguredTimeout returns a timeout from connection config.

connection_pool/connection_pool.go

@@ -11,6 +11,7 @@
 package connection_pool
 
 import (
+	"context"
 	"errors"
 	"log"
 	"sync/atomic"
@@ -483,33 +484,33 @@ func (connPool *ConnectionPool) EvalAsync(expr string, args interface{}, userMod
 }
 
 // Do sends the request and returns a response.
-func (connPool *ConnectionPool) Do(req tarantool.Request, userMode Mode) (*tarantool.Response, error) {
+func (connPool *ConnectionPool) Do(ctx context.Context, req tarantool.Request, userMode Mode) (*tarantool.Response, error) {
 	conn, err := connPool.getNextConnection(userMode)
 	if err != nil {
 		return nil, err
 	}
 
-	return conn.Do(req)
+	return conn.Do(ctx, req)
 }
 
 // DoTyped sends the request and fills the typed result.
-func (connPool *ConnectionPool) DoTyped(req tarantool.Request, result interface{}, userMode Mode) error {
+func (connPool *ConnectionPool) DoTyped(ctx context.Context, req tarantool.Request, result interface{}, userMode Mode) error {
 	conn, err := connPool.getNextConnection(userMode)
 	if err != nil {
 		return err
 	}
 
-	return conn.DoTyped(req, result)
+	return conn.DoTyped(ctx, req, result)
 }
 
 // DoAsync sends the request and returns a future.
-func (connPool *ConnectionPool) DoAsync(req tarantool.Request, userMode Mode) *tarantool.Future {
+func (connPool *ConnectionPool) DoAsync(ctx context.Context, req tarantool.Request, userMode Mode) *tarantool.Future {
 	conn, err := connPool.getNextConnection(userMode)
 	if err != nil {
 		return tarantool.NewErrorFuture(err)
 	}
 
-	return conn.DoAsync(req)
+	return conn.DoAsync(ctx, req)
 }
 
 //

connection_pool/connection_pool_test.go

@@ -1,6 +1,7 @@
 package connection_pool_test
 
 import (
+	"context"
 	"log"
 	"os"
 	"strings"
@@ -1266,27 +1267,27 @@ func TestDo(t *testing.T) {
 
 	req := tarantool.NewPingRequest()
 	// ANY
-	resp, err := connPool.Do(req, connection_pool.ANY)
+	resp, err := connPool.Do(context.Background(), req, connection_pool.ANY)
 	require.Nilf(t, err, "failed to Ping")
 	require.NotNilf(t, resp, "response is nil after Ping")
 
 	// RW
-	resp, err = connPool.Do(req, connection_pool.RW)
+	resp, err = connPool.Do(context.Background(), req, connection_pool.RW)
 	require.Nilf(t, err, "failed to Ping")
 	require.NotNilf(t, resp, "response is nil after Ping")
 
 	// RO
-	resp, err = connPool.Do(req, connection_pool.RO)
+	resp, err = connPool.Do(context.Background(), req, connection_pool.RO)
 	require.Nilf(t, err, "failed to Ping")
 	require.NotNilf(t, resp, "response is nil after Ping")
 
 	// PreferRW
-	resp, err = connPool.Do(req, connection_pool.PreferRW)
+	resp, err = connPool.Do(context.Background(), req, connection_pool.PreferRW)
 	require.Nilf(t, err, "failed to Ping")
 	require.NotNilf(t, resp, "response is nil after Ping")
 
 	// PreferRO
-	resp, err = connPool.Do(req, connection_pool.PreferRO)
+	resp, err = connPool.Do(context.Background(), req, connection_pool.PreferRO)
 	require.Nilf(t, err, "failed to Ping")
 	require.NotNilf(t, resp, "response is nil after Ping")
 }

connection_pool/example_test.go

@@ -1,6 +1,7 @@
 package connection_pool_test
 
 import (
+	"context"
 	"fmt"
 
 	"github.com/tarantool/go-tarantool"
@@ -539,7 +540,7 @@ func ExampleConnectionPool_Do() {
 
 	// Ping a Tarantool instance to check connection.
 	req := tarantool.NewPingRequest()
-	resp, err := pool.Do(req, connection_pool.ANY)
+	resp, err := pool.Do(context.Background(), req, connection_pool.ANY)
 	fmt.Println("Ping Code", resp.Code)
 	fmt.Println("Ping Data", resp.Data)
 	fmt.Println("Ping Error", err)

connector.go

@@ -1,6 +1,9 @@
 package tarantool
 
-import "time"
+import (
+	"context"
+	"time"
+)
 
 type Connector interface {
 	ConnectedNow() bool
@@ -39,7 +42,7 @@ type Connector interface {
 	Call17Async(functionName string, args interface{}) *Future
 	EvalAsync(expr string, args interface{}) *Future
 
-	Do(req Request) (resp *Response, err error)
-	DoTyped(req Request, result interface{}) (err error)
-	DoAsync(req Request) (fut *Future)
+	Do(ctx context.Context, req Request) (resp *Response, err error)
+	DoTyped(ctx context.Context, req Request, result interface{}) (err error)
+	DoAsync(ctx context.Context, req Request) (fut *Future)
 }

example_test.go

@@ -1,6 +1,7 @@
 package tarantool_test
 
 import (
+	"context"
 	"fmt"
 	"time"
 
@@ -133,7 +134,7 @@ func ExampleSelectRequest() {
 	req := tarantool.NewSelectRequest(517).
 		Limit(100).
 		Key(tarantool.IntKey{1111})
-	resp, err := conn.Do(req)
+	resp, err := conn.Do(context.Background(), req)
 	if err != nil {
 		fmt.Printf("error in do select request is %v", err)
 		return
@@ -144,7 +145,7 @@ func ExampleSelectRequest() {
 		Index("primary").
 		Limit(100).
 		Key(tarantool.IntKey{1111})
-	fut := conn.DoAsync(req)
+	fut := conn.DoAsync(context.Background(), req)
 	resp, err = fut.Get()
 	if err != nil {
 		fmt.Printf("error in do async select request is %v", err)
@@ -163,7 +164,7 @@ func ExampleUpdateRequest() {
 	req := tarantool.NewUpdateRequest(517).
 		Key(tarantool.IntKey{1111}).
 		Operations(tarantool.NewOperations().Assign(1, "bye"))
-	resp, err := conn.Do(req)
+	resp, err := conn.Do(context.Background(), req)
 	if err != nil {
 		fmt.Printf("error in do update request is %v", err)
 		return
@@ -174,7 +175,7 @@ func ExampleUpdateRequest() {
 		Index("primary").
 		Key(tarantool.IntKey{1111}).
 		Operations(tarantool.NewOperations().Assign(1, "hello"))
-	fut := conn.DoAsync(req)
+	fut := conn.DoAsync(context.Background(), req)
 	resp, err = fut.Get()
 	if err != nil {
 		fmt.Printf("error in do async update request is %v", err)
@@ -194,7 +195,7 @@ func ExampleUpsertRequest() {
 	req = tarantool.NewUpsertRequest(517).
 		Tuple([]interface{}{uint(1113), "first", "first"}).
 		Operations(tarantool.NewOperations().Assign(1, "updated"))
-	resp, err := conn.Do(req)
+	resp, err := conn.Do(context.Background(), req)
 	if err != nil {
 		fmt.Printf("error in do select upsert is %v", err)
 		return
@@ -204,7 +205,7 @@ func ExampleUpsertRequest() {
 	req = tarantool.NewUpsertRequest("test").
 		Tuple([]interface{}{uint(1113), "second", "second"}).
 		Operations(tarantool.NewOperations().Assign(2, "updated"))
-	fut := conn.DoAsync(req)
+	fut := conn.DoAsync(context.Background(), req)
 	resp, err = fut.Get()
 	if err != nil {
 		fmt.Printf("error in do async upsert request is %v", err)
@@ -215,7 +216,7 @@ func ExampleUpsertRequest() {
 	req = tarantool.NewSelectRequest(517).
 		Limit(100).
 		Key(tarantool.IntKey{1113})
-	resp, err = conn.Do(req)
+	resp, err = conn.Do(context.Background(), req)
 	if err != nil {
 		fmt.Printf("error in do select request is %v", err)
 		return

future.go

@@ -1,6 +1,8 @@
 package tarantool
 
 import (
+	"context"
+	"fmt"
 	"sync"
 	"time"
 )
@@ -16,13 +18,34 @@ type Future struct {
 	err       error
 	ready     chan struct{}
 	done      chan struct{}
+	ctx       context.Context
+	conn      *Connection
 }
 
 func (fut *Future) wait() {
 	if fut.done == nil {
 		return
 	}
-	<-fut.done
+	if fut.ctx == nil {
+		<-fut.done
+		return
+	}
+	select {
+	case <-fut.done:
+	default:
+		select {
+		case <-fut.ctx.Done():
+			fut.conn.fetchFuture(fut.requestId)
+			fut.SetError(fmt.Errorf("context is done"))
+		default:
+			select {
+			case <-fut.done:
+			case <-fut.ctx.Done():
+				fut.conn.fetchFuture(fut.requestId)
+				fut.SetError(fmt.Errorf("context is done"))
+			}
+		}
+	}
 }
 
 func (fut *Future) isDone() bool {

multi/multi.go

@@ -12,6 +12,7 @@
 package multi
 
 import (
+	"context"
 	"errors"
 	"sync"
 	"sync/atomic"
@@ -456,16 +457,16 @@ func (connMulti *ConnectionMulti) EvalAsync(expr string, args interface{}) *tara
 }
 
 // Do sends the request and returns a response.
-func (connMulti *ConnectionMulti) Do(req tarantool.Request) (*tarantool.Response, error) {
-	return connMulti.getCurrentConnection().Do(req)
+func (connMulti *ConnectionMulti) Do(ctx context.Context, req tarantool.Request) (*tarantool.Response, error) {
+	return connMulti.getCurrentConnection().Do(ctx, req)
 }
 
 // DoTyped sends the request and fills the typed result.
-func (connMulti *ConnectionMulti) DoTyped(req tarantool.Request, result interface{}) error {
-	return connMulti.getCurrentConnection().DoTyped(req, result)
+func (connMulti *ConnectionMulti) DoTyped(ctx context.Context, req tarantool.Request, result interface{}) error {
+	return connMulti.getCurrentConnection().DoTyped(ctx, req, result)
 }
 
 // DoAsync sends the request and returns a future.
-func (connMulti *ConnectionMulti) DoAsync(req tarantool.Request) *tarantool.Future {
-	return connMulti.getCurrentConnection().DoAsync(req)
+func (connMulti *ConnectionMulti) DoAsync(ctx context.Context, req tarantool.Request) *tarantool.Future {
+	return connMulti.getCurrentConnection().DoAsync(ctx, req)
 }