future: transformation into interface

Transformed `future` into public `Future` interface,
also `future` become inner implementation.
`setError` and `setResponse` became private methods of inner `future` type.
Added Future benchmarks.
Updated methods and functions with new interface.

Author: ermyar

CHANGELOG.md

@@ -15,6 +15,7 @@ Versioning](http://semver.org/spec/v2.0.0.html) except to the first release.
 * Added missing IPROTO feature flags to greeting negotiation
   (iproto.IPROTO_FEATURE_IS_SYNC, iproto.IPROTO_FEATURE_INSERT_ARROW) (#466).
 * Added Future.cond (sync.Cond) and Future.finished bool. Added Future.finish() marks Future as done (#496).
+* New `Future` interface (#470).
 
 ### Changed
 
@@ -30,6 +31,8 @@ Versioning](http://semver.org/spec/v2.0.0.html) except to the first release.
 * Removed deprecated `Connection` methods, related interfaces and tests are updated (#479).
 * Replaced the use of optional types in crud with go-option library (#492).
 * Future.done replaced with Future.cond (sync.Cond) + Future.finished bool (#496).
+* `Future` transform into `future` that implements interface `Future` and become private,
+  `SetError` and `SetResponse` become private (#470).
 
 ### Fixed
 

MIGRATION.md

@@ -32,6 +32,7 @@ TODO
   singleTpl := tpl[0]
   ```
 * Future.done replaced with Future.cond (sync.Cond) + Future.finished bool.
+* `Future` is an interface now.
 
 ## Migration from v1.x.x to v2.x.x
 

connection.go

@@ -214,11 +214,11 @@ type Connection struct {
 var _ = Connector(&Connection{}) // Check compatibility with connector interface.
 
 type futureList struct {
-	first *Future
-	last  **Future
+	first *future
+	last  **future
 }
 
-func (list *futureList) findFuture(reqid uint32, fetch bool) *Future {
+func (list *futureList) findFuture(reqid uint32, fetch bool) *future {
 	root := &list.first
 	for {
 		fut := *root
@@ -240,7 +240,7 @@ func (list *futureList) findFuture(reqid uint32, fetch bool) *Future {
 	}
 }
 
-func (list *futureList) addFuture(fut *Future) {
+func (list *futureList) addFuture(fut *future) {
 	*list.last = fut
 	list.last = &fut.next
 }
@@ -250,7 +250,7 @@ func (list *futureList) clear(err error, conn *Connection) {
 	list.first = nil
 	list.last = &list.first
 	for fut != nil {
-		fut.SetError(err)
+		fut.setError(err)
 		conn.markDone(fut)
 		fut, fut.next = fut.next, nil
 	}
@@ -446,9 +446,9 @@ func (conn *Connection) Handle() interface{} {
 	return conn.opts.Handle
 }
 
-func (conn *Connection) cancelFuture(fut *Future, err error) {
+func (conn *Connection) cancelFuture(fut *future, err error) {
 	if fut = conn.fetchFuture(fut.requestId); fut != nil {
-		fut.SetError(err)
+		fut.setError(err)
 		conn.markDone(fut)
 	}
 }
@@ -871,7 +871,7 @@ func (conn *Connection) reader(r io.Reader, c Conn) {
 			return
 		}
 
-		var fut *Future = nil
+		var fut *future = nil
 		if code == iproto.IPROTO_EVENT {
 			if event, err := readWatchEvent(&buf); err == nil {
 				events <- event
@@ -887,8 +887,8 @@ func (conn *Connection) reader(r io.Reader, c Conn) {
 			conn.opts.Logger.Report(LogBoxSessionPushUnsupported, conn, header)
 		} else {
 			if fut = conn.fetchFuture(header.RequestId); fut != nil {
-				if err := fut.SetResponse(header, &buf); err != nil {
-					fut.SetError(fmt.Errorf("failed to set response: %w", err))
+				if err := fut.setResponse(header, &buf); err != nil {
+					fut.setError(fmt.Errorf("failed to set response: %w", err))
 				}
 				conn.markDone(fut)
 			}
@@ -925,9 +925,9 @@ func (conn *Connection) eventer(events <-chan connWatchEvent) {
 	}
 }
 
-func (conn *Connection) newFuture(req Request) (fut *Future) {
+func (conn *Connection) newFuture(req Request) (fut *future) {
 	ctx := req.Ctx()
-	fut = NewFuture(req)
+	fut = newFuture(req)
 	if conn.rlimit != nil && conn.opts.RLimitAction == RLimitDrop {
 		select {
 		case conn.rlimit <- struct{}{}:
@@ -974,7 +974,7 @@ func (conn *Connection) newFuture(req Request) (fut *Future) {
 	if ctx != nil {
 		select {
 		case <-ctx.Done():
-			fut.SetError(fmt.Errorf("context is done (request ID %d): %w",
+			fut.setError(fmt.Errorf("context is done (request ID %d): %w",
 				fut.requestId, context.Cause(ctx)))
 			shard.rmut.Unlock()
 			return
@@ -1007,7 +1007,7 @@ func (conn *Connection) newFuture(req Request) (fut *Future) {
 
 // This method removes a future from the internal queue if the context
 // is "done" before the response is come.
-func (conn *Connection) contextWatchdog(fut *Future, ctx context.Context) {
+func (conn *Connection) contextWatchdog(fut *future, ctx context.Context) {
 	select {
 	case <-fut.WaitChan():
 	case <-ctx.Done():
@@ -1032,7 +1032,7 @@ func (conn *Connection) decrementRequestCnt() {
 	}
 }
 
-func (conn *Connection) send(req Request, streamId uint64) *Future {
+func (conn *Connection) send(req Request, streamId uint64) *future {
 	conn.incrementRequestCnt()
 
 	fut := conn.newFuture(req)
@@ -1056,7 +1056,7 @@ func (conn *Connection) send(req Request, streamId uint64) *Future {
 	return fut
 }
 
-func (conn *Connection) putFuture(fut *Future, req Request, streamId uint64) {
+func (conn *Connection) putFuture(fut *future, req Request, streamId uint64) {
 	shardn := fut.requestId & (conn.opts.Concurrency - 1)
 	shard := &conn.shard[shardn]
 	shard.bufmut.Lock()
@@ -1077,7 +1077,7 @@ func (conn *Connection) putFuture(fut *Future, req Request, streamId uint64) {
 		shard.buf.Trunc(blen)
 		shard.bufmut.Unlock()
 		if f := conn.fetchFuture(reqid); f == fut {
-			fut.SetError(err)
+			fut.setError(err)
 			conn.markDone(fut)
 		} else if f != nil {
 			/* in theory, it is possible. In practice, you have
@@ -1092,7 +1092,7 @@ func (conn *Connection) putFuture(fut *Future, req Request, streamId uint64) {
 				// packing error is more important than connection
 				// error, because it is indication of programmer's
 				// mistake.
-				fut.SetError(err)
+				fut.setError(err)
 			}
 		}
 		return
@@ -1109,28 +1109,28 @@ func (conn *Connection) putFuture(fut *Future, req Request, streamId uint64) {
 				RequestId: reqid,
 				Error:     ErrorNo,
 			}
-			_ = fut.SetResponse(header, nil)
+			_ = fut.setResponse(header, nil)
 			conn.markDone(fut)
 		}
 	}
 }
 
-func (conn *Connection) markDone(fut *Future) {
+func (conn *Connection) markDone(fut *future) {
 	if conn.rlimit != nil {
 		<-conn.rlimit
 	}
 	conn.decrementRequestCnt()
 }
 
-func (conn *Connection) fetchFuture(reqid uint32) (fut *Future) {
+func (conn *Connection) fetchFuture(reqid uint32) (fut *future) {
 	shard := &conn.shard[reqid&(conn.opts.Concurrency-1)]
 	shard.rmut.Lock()
 	fut = conn.getFutureImp(reqid, true)
 	shard.rmut.Unlock()
 	return fut
 }
 
-func (conn *Connection) getFutureImp(reqid uint32, fetch bool) *Future {
+func (conn *Connection) getFutureImp(reqid uint32, fetch bool) *future {
 	shard := &conn.shard[reqid&(conn.opts.Concurrency-1)]
 	pos := (reqid / conn.opts.Concurrency) & (requestsMap - 1)
 	// futures with even requests id belong to requests list with nil context
@@ -1168,7 +1168,7 @@ func (conn *Connection) timeouts() {
 					} else {
 						fut.next = nil
 					}
-					fut.SetError(ClientError{
+					fut.setError(ClientError{
 						Code: ErrTimeouted,
 						Msg:  fmt.Sprintf("client timeout for request %d", fut.requestId),
 					})
@@ -1232,11 +1232,10 @@ func (conn *Connection) nextRequestId(context bool) (requestId uint32) {
 //
 // An error is returned if the request was formed incorrectly, or failed to
 // create the future.
-func (conn *Connection) Do(req Request) *Future {
+func (conn *Connection) Do(req Request) Future {
 	if connectedReq, ok := req.(ConnectedRequest); ok {
 		if connectedReq.Conn() != conn {
-			fut := NewFuture(req)
-			fut.SetError(errUnknownRequest)
+			fut := NewFutureWithErr(req, errUnknownRequest)
 			return fut
 		}
 	}

connector.go

@@ -5,7 +5,7 @@ import "time"
 // Doer is an interface that performs requests asynchronously.
 type Doer interface {
 	// Do performs a request asynchronously.
-	Do(req Request) (fut *Future)
+	Do(req Request) Future
 }
 
 type Connector interface {

future.go

@@ -6,11 +6,19 @@ import (
 	"time"
 )
 
-// Future is a handle for asynchronous request.
-type Future struct {
+// Future is an interface that handle asynchronous request.
+type Future interface {
+	Get() ([]interface{}, error)
+	GetTyped(result interface{}) error
+	GetResponse() (Response, error)
+	WaitChan() <-chan struct{}
+}
+
+// future is inner implementation of Future interface.
+type future struct {
 	requestId uint32
 	req       Request
-	next      *Future
+	next      *future
 	timeout   time.Duration
 	mutex     sync.Mutex
 	resp      Response
@@ -20,7 +28,9 @@ type Future struct {
 	done      chan struct{}
 }
 
-func (fut *Future) wait() {
+var _ = Future(&future{})
+
+func (fut *future) wait() {
 	fut.mutex.Lock()
 	defer fut.mutex.Unlock()
 
@@ -29,7 +39,7 @@ func (fut *Future) wait() {
 	}
 }
 
-func (fut *Future) finish() {
+func (fut *future) finish() {
 	fut.mutex.Lock()
 	defer fut.mutex.Unlock()
 
@@ -42,18 +52,34 @@ func (fut *Future) finish() {
 	fut.cond.Broadcast()
 }
 
-// NewFuture creates a new empty Future for a given Request.
-func NewFuture(req Request) (fut *Future) {
-	fut = &Future{}
+// NewFutureWithErr returns Future with given error.
+func NewFutureWithErr(req Request, err error) Future {
+	fut := newFuture(req)
+	fut.setError(err)
+	return fut
+}
+
+// NewFutureWithResponse returns Future with given Response.
+func NewFutureWithResponse(req Request, header Header, body io.Reader) (Future, error) {
+	fut := newFuture(req)
+	if err := fut.setResponse(header, body); err != nil {
+		return nil, err
+	}
+	return fut, nil
+}
+
+// newFuture creates a new empty future for a given Request.
+func newFuture(req Request) (fut *future) {
+	fut = &future{}
 	fut.done = nil
 	fut.finished = false
 	fut.cond.L = &fut.mutex
 	fut.req = req
 	return fut
 }
 
-// SetResponse sets a response for the future and finishes the future.
-func (fut *Future) SetResponse(header Header, body io.Reader) error {
+// setResponse sets a response for the future and finishes the future.
+func (fut *future) setResponse(header Header, body io.Reader) error {
 	fut.mutex.Lock()
 	defer fut.mutex.Unlock()
 
@@ -78,8 +104,8 @@ func (fut *Future) SetResponse(header Header, body io.Reader) error {
 	return nil
 }
 
-// SetError sets an error for the future and finishes the future.
-func (fut *Future) SetError(err error) {
+// setError sets an error for the future and finishes the future.
+func (fut *future) setError(err error) {
 	fut.mutex.Lock()
 	defer fut.mutex.Unlock()
 
@@ -103,7 +129,7 @@ func (fut *Future) SetError(err error) {
 //
 // "error" could be Error, if it is error returned by Tarantool,
 // or ClientError, if something bad happens in a client process.
-func (fut *Future) GetResponse() (Response, error) {
+func (fut *future) GetResponse() (Response, error) {
 	fut.wait()
 	return fut.resp, fut.err
 }
@@ -114,7 +140,7 @@ func (fut *Future) GetResponse() (Response, error) {
 //
 // "error" could be Error, if it is error returned by Tarantool,
 // or ClientError, if something bad happens in a client process.
-func (fut *Future) Get() ([]interface{}, error) {
+func (fut *future) Get() ([]interface{}, error) {
 	fut.wait()
 	if fut.err != nil {
 		return nil, fut.err
@@ -126,7 +152,7 @@ func (fut *Future) Get() ([]interface{}, error) {
 // It is could be much faster than Get() function.
 //
 // Note: Tarantool usually returns array of tuples (except for Eval and Call17 actions).
-func (fut *Future) GetTyped(result interface{}) error {
+func (fut *future) GetTyped(result interface{}) error {
 	fut.wait()
 	if fut.err != nil {
 		return fut.err
@@ -141,7 +167,7 @@ func init() {
 }
 
 // WaitChan returns channel which becomes closed when response arrived or error occurred.
-func (fut *Future) WaitChan() <-chan struct{} {
+func (fut *future) WaitChan() <-chan struct{} {
 	fut.mutex.Lock()
 	defer fut.mutex.Unlock()