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.
Added a new interface for canceling Future objects from the
common queue.

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"
@@ -139,6 +140,8 @@ type Connection struct {
 	state   uint32
 	dec     *msgpack.Decoder
 	lenbuf  [PacketLengthBytes]byte
+
+	futCanceler futureCanceler
 }
 
 var _ = Connector(&Connection{}) // Check compatibility with connector interface.
@@ -334,6 +337,7 @@ func Connect(addr string, opts Opts) (conn *Connection, err error) {
 			return nil, err
 		}
 	}
+	conn.futCanceler = newFutCancelerImpl(conn.cancelFuture)
 
 	return conn, err
 }
@@ -387,6 +391,16 @@ func (conn *Connection) Handle() interface{} {
 	return conn.opts.Handle
 }
 
+// cancelFuture removes the passed future from the internal queue of Future objects.
+func (conn *Connection) cancelFuture(fut *Future, err error) error {
+	if fut == nil {
+		return fmt.Errorf("passed nil future")
+	}
+	fut.SetError(err)
+	conn.fetchFuture(fut.requestId)
+	return nil
+}
+
 func (conn *Connection) dial() (err error) {
 	var connection net.Conn
 	network := "tcp"
@@ -721,8 +735,8 @@ func (conn *Connection) reader(r *bufio.Reader, c net.Conn) {
 	}
 }
 
-func (conn *Connection) newFuture() (fut *Future) {
-	fut = NewFuture()
+func (conn *Connection) newFuture(ctx context.Context) (fut *Future) {
+	fut = NewFuture(ctx, conn.futCanceler)
 	if conn.rlimit != nil && conn.opts.RLimitAction == RLimitDrop {
 		select {
 		case conn.rlimit <- struct{}{}:
@@ -785,8 +799,8 @@ func (conn *Connection) newFuture() (fut *Future) {
 	return
 }
 
-func (conn *Connection) send(req Request) *Future {
-	fut := conn.newFuture()
+func (conn *Connection) send(ctx context.Context, req Request) *Future {
+	fut := conn.newFuture(ctx)
 	if fut.ready == nil {
 		return fut
 	}
@@ -992,26 +1006,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,32 @@ type Future struct {
 	err       error
 	ready     chan struct{}
 	done      chan struct{}
+	ctx       context.Context
+	canceler  futureCanceler
 }
 
 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.canceler.Cancel(fut, fmt.Errorf("context is done"))
+		default:
+			select {
+			case <-fut.done:
+			case <-fut.ctx.Done():
+				fut.canceler.Cancel(fut, fmt.Errorf("context is done"))
+			}
+		}
+	}
 }
 
 func (fut *Future) isDone() bool {
@@ -118,8 +139,10 @@ func (it *asyncResponseIterator) nextResponse() (resp *Response) {
 }
 
 // NewFuture creates a new empty Future.
-func NewFuture() (fut *Future) {
+func NewFuture(ctx context.Context, canceler futureCanceler) (fut *Future) {
 	fut = &Future{}
+	fut.ctx = ctx
+	fut.canceler = canceler
 	fut.ready = make(chan struct{}, 1000000000)
 	fut.done = make(chan struct{})
 	fut.pushes = make([]*Response, 0)
@@ -128,7 +151,7 @@ func NewFuture() (fut *Future) {
 
 // NewErrorFuture returns new set empty Future with filled error field.
 func NewErrorFuture(err error) *Future {
-	fut := NewFuture()
+	fut := NewFuture(context.Background(), nil)
 	fut.SetError(err)
 	return fut
 }

future_canceler.go

@@ -0,0 +1,17 @@
+package tarantool
+
+type futureCanceler interface {
+	Cancel(fut *Future, err error) error
+}
+
+type futCancelerImpl struct {
+	cancel func(fut *Future, err error) error
+}
+
+func newFutCancelerImpl(cancelFunc func(fut *Future, err error) error) *futCancelerImpl {
+	return &futCancelerImpl{cancel: cancelFunc}
+}
+
+func (fc *futCancelerImpl) Cancel(fut *Future, err error) error {
+	return fc.cancel(fut, err)
+}