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.

to fix and squash

Author: vr009

CHANGELOG.md

@@ -19,6 +19,7 @@ Versioning](http://semver.org/spec/v2.0.0.html) except to the first release.
 - Master discovery (#113)
 - SQL support (#62)
 - Add public API with a request object for Select/Update/Upstream (#126)
+- Context support for request objects
 
 ### Fixed
 

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"
@@ -877,9 +878,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
 	}
@@ -890,8 +896,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
 	}
@@ -902,12 +912,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
 }
 

example_test.go

@@ -1,6 +1,7 @@
 package tarantool_test
 
 import (
+	"context"
 	"fmt"
 	"github.com/tarantool/go-tarantool/test_helpers"
 	"time"
@@ -111,11 +112,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
@@ -126,7 +129,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)
 	}
@@ -144,11 +147,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
@@ -159,7 +164,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)
 	}
@@ -177,12 +182,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
@@ -192,7 +199,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)
 	}
@@ -206,7 +213,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 Future object and
+// returns the same object with that context.
+func (fut *Future) WithCtx(ctx context.Context) *Future {
+	fut.ctx = ctx
+	return fut
 }
 
 // Get waits for Future to be filled and returns Response and error.
@@ -122,5 +132,20 @@ 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:
+		case <-fut.ready:
+		case <-fut.ctx.Done():
+			fut.err = fmt.Errorf("context is done")
+		}
+	}
 }