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

@@ -36,6 +36,7 @@ CI and documentation.
 - Master discovery (#113)
 - SQL support (#62)
 - Add public API with a request object for Select/Update/Upstream (#126)
+- Context support for request objects
 
 ### Changed
 

config.lua

@@ -110,6 +110,12 @@ local function simple_incr(a)
 end
 rawset(_G, 'simple_incr', simple_incr)
 
+local function simple_sleep(a)
+    require('fiber').sleep(15)
+    return a + 1
+end
+rawset(_G, 'simple_sleep', simple_sleep)
+
 box.space.test:truncate()
 
 --box.schema.user.revoke('guest', 'read,write,execute', 'universe')

connection.go

@@ -5,6 +5,7 @@ package tarantool
 import (
 	"bufio"
 	"bytes"
+	"context"
 	"errors"
 	"fmt"
 	"io"
@@ -916,9 +917,14 @@ func (conn *Connection) nextRequestId() (requestId uint32) {
 // Do verifies, sends the request and returns a response.
 //
 // 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, err := conn.DoAsync(req)
+// communicate by the connection, or unable to decode the response, or nil context is passed
+// or context is canceled or done.
+func (conn *Connection) Do(ctx context.Context, req Request) (*Response, error) {
+	if ctx == nil {
+		return nil, errors.New("passed nil context")
+	}
+
+	fut, err := conn.DoAsync(ctx, req)
 	if err != nil {
 		return nil, err
 	}
@@ -929,8 +935,12 @@ func (conn *Connection) Do(req Request) (*Response, error) {
 //
 // 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, err := conn.DoAsync(req)
+func (conn *Connection) DoTyped(ctx context.Context, req Request, result interface{}) error {
+	if ctx == nil {
+		return errors.New("passed nil context")
+	}
+
+	fut, err := conn.DoAsync(ctx, req)
 	if err != nil {
 		return err
 	}
@@ -941,12 +951,20 @@ func (conn *Connection) DoTyped(req Request, result interface{}) error {
 //
 // An error is returned if the request was formed incorrectly, or failure to
 // create the future.
-func (conn *Connection) DoAsync(req Request) (*Future, error) {
+func (conn *Connection) DoAsync(ctx context.Context, req Request) (*Future, error) {
 	bodyFunc, err := req.BodyFunc(conn.Schema)
 	if err != nil {
 		return nil, err
 	}
 	future := conn.newFuture(req.Code())
+	if ctx != nil {
+		select {
+		case <-ctx.Done():
+			return nil, fmt.Errorf("context is done")
+		default:
+			future.WithCtx(ctx)
+		}
+	}
 	return future.send(conn, bodyFunc), nil
 }
 

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, error) {
+func (connPool *ConnectionPool) DoAsync(ctx context.Context, req tarantool.Request, userMode Mode) (*tarantool.Future, error) {
 	conn, err := connPool.getNextConnection(userMode)
 	if err != nil {
 		return nil, 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, err error)
+	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, err error)
 }

example_test.go

@@ -1,6 +1,7 @@
 package tarantool_test
 
 import (
+	"context"
 	"fmt"
 	"time"
 
@@ -129,11 +130,13 @@ func ExampleConnection_SelectAsync() {
 func ExampleSelectRequest() {
 	conn := example_connect()
 	defer conn.Close()
+	var ctx, cancel = context.WithCancel(context.Background())
+	defer cancel()
 
 	req := tarantool.NewSelectRequest(517).
 		Limit(100).
 		Key(tarantool.IntKey{1111})
-	resp, err := conn.Do(req)
+	resp, err := conn.Do(ctx, req)
 	if err != nil {
 		fmt.Printf("error in do select request is %v", err)
 		return
@@ -144,7 +147,7 @@ func ExampleSelectRequest() {
 		Index("primary").
 		Limit(100).
 		Key(tarantool.IntKey{1111})
-	fut, err := conn.DoAsync(req)
+	fut, err := conn.DoAsync(ctx, req)
 	if err != nil {
 		fmt.Printf("error in do async select request is %v", err)
 	}
@@ -162,11 +165,13 @@ func ExampleSelectRequest() {
 func ExampleUpdateRequest() {
 	conn := example_connect()
 	defer conn.Close()
+	var ctx, cancel = context.WithCancel(context.Background())
+	defer cancel()
 
 	req := tarantool.NewUpdateRequest(517).
 		Key(tarantool.IntKey{1111}).
 		Operations(tarantool.NewOperations().Assign(1, "bye"))
-	resp, err := conn.Do(req)
+	resp, err := conn.Do(ctx, req)
 	if err != nil {
 		fmt.Printf("error in do update request is %v", err)
 		return
@@ -177,7 +182,7 @@ func ExampleUpdateRequest() {
 		Index("primary").
 		Key(tarantool.IntKey{1111}).
 		Operations(tarantool.NewOperations().Assign(1, "hello"))
-	fut, err := conn.DoAsync(req)
+	fut, err := conn.DoAsync(ctx, req)
 	if err != nil {
 		fmt.Printf("error in do async update request is %v", err)
 	}
@@ -195,12 +200,14 @@ func ExampleUpdateRequest() {
 func ExampleUpsertRequest() {
 	conn := example_connect()
 	defer conn.Close()
+	var ctx, cancel = context.WithCancel(context.Background())
+	defer cancel()
 
 	var req tarantool.Request
 	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(ctx, req)
 	if err != nil {
 		fmt.Printf("error in do select upsert is %v", err)
 		return
@@ -210,7 +217,7 @@ func ExampleUpsertRequest() {
 	req = tarantool.NewUpsertRequest("test").
 		Tuple([]interface{}{uint(1113), "second", "second"}).
 		Operations(tarantool.NewOperations().Assign(2, "updated"))
-	fut, err := conn.DoAsync(req)
+	fut, err := conn.DoAsync(ctx, req)
 	if err != nil {
 		fmt.Printf("error in do async upsert request is %v", err)
 	}
@@ -224,7 +231,7 @@ func ExampleUpsertRequest() {
 	req = tarantool.NewSelectRequest(517).
 		Limit(100).
 		Key(tarantool.IntKey{1113})
-	resp, err = conn.Do(req)
+	resp, err = conn.Do(ctx, 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"
 	"time"
 
 	"gopkg.in/vmihailenco/msgpack.v2"
@@ -15,6 +17,14 @@ type Future struct {
 	err         error
 	ready       chan struct{}
 	next        *Future
+	ctx         context.Context
+}
+
+// WithCtx sets a passed context to a Future object and
+// returns the same object with that context.
+func (fut *Future) WithCtx(ctx context.Context) *Future {
+	fut.ctx = ctx
+	return fut
 }
 
 // NewErrorFuture returns new set empty Future with filled error field.
@@ -127,5 +137,22 @@ func (fut *Future) wait() {
 	if fut.ready == nil {
 		return
 	}
-	<-fut.ready
+	if fut.ctx == nil {
+		<-fut.ready
+		return
+	}
+	select {
+	case <-fut.ready:
+	default:
+		select {
+		case <-fut.ctx.Done():
+			fut.err = fmt.Errorf("context is done")
+		default:
+			select {
+			case <-fut.ready:
+			case <-fut.ctx.Done():
+				fut.err = fmt.Errorf("context is done")
+			}
+		}
+	}
 }

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, error) {
-	return connMulti.getCurrentConnection().DoAsync(req)
+func (connMulti *ConnectionMulti) DoAsync(ctx context.Context, req tarantool.Request) (*tarantool.Future, error) {
+	return connMulti.getCurrentConnection().DoAsync(ctx, req)
 }