vendor: update buildkit

Signed-off-by: CrazyMax <crazy-max@users.noreply.github.com>

vendor/github.com/containerd/ttrpc/client.go

@@ -19,30 +19,30 @@ package ttrpc
 import (
 	"context"
 	"errors"
+	"fmt"
 	"io"
 	"net"
-	"os"
 	"strings"
 	"sync"
 	"syscall"
 	"time"
 
-	"github.com/gogo/protobuf/proto"
 	"github.com/sirupsen/logrus"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/proto"
 )
 
-// ErrClosed is returned by client methods when the underlying connection is
-// closed.
-var ErrClosed = errors.New("ttrpc: closed")
-
 // Client for a ttrpc server
 type Client struct {
 	codec   codec
 	conn    net.Conn
 	channel *channel
-	calls   chan *callRequest
+
+	streamLock   sync.RWMutex
+	streams      map[streamID]*stream
+	nextStreamID streamID
+	sendLock     sync.Mutex
 
 	ctx    context.Context
 	closed func()
@@ -51,8 +51,6 @@ type Client struct {
 	userCloseFunc   func()
 	userCloseWaitCh chan struct{}
 
-	errOnce     sync.Once
-	err         error
 	interceptor UnaryClientInterceptor
 }
 
@@ -73,13 +71,16 @@ func WithUnaryClientInterceptor(i UnaryClientInterceptor) ClientOpts {
 	}
 }
 
+// NewClient creates a new ttrpc client using the given connection
 func NewClient(conn net.Conn, opts ...ClientOpts) *Client {
 	ctx, cancel := context.WithCancel(context.Background())
+	channel := newChannel(conn)
 	c := &Client{
 		codec:           codec{},
 		conn:            conn,
-		channel:         newChannel(conn),
-		calls:           make(chan *callRequest),
+		channel:         channel,
+		streams:         make(map[streamID]*stream),
+		nextStreamID:    1,
 		closed:          cancel,
 		ctx:             ctx,
 		userCloseFunc:   func() {},
@@ -95,13 +96,13 @@ func NewClient(conn net.Conn, opts ...ClientOpts) *Client {
 	return c
 }
 
-type callRequest struct {
-	ctx  context.Context
-	req  *Request
-	resp *Response  // response will be written back here
-	errs chan error // error written here on completion
+func (c *Client) send(sid uint32, mt messageType, flags uint8, b []byte) error {
+	c.sendLock.Lock()
+	defer c.sendLock.Unlock()
+	return c.channel.send(sid, mt, flags, b)
 }
 
+// Call makes a unary request and returns with response
 func (c *Client) Call(ctx context.Context, service, method string, req, resp interface{}) error {
 	payload, err := c.codec.Marshal(req)
 	if err != nil {
@@ -113,6 +114,7 @@ func (c *Client) Call(ctx context.Context, service, method string, req, resp int
 			Service: service,
 			Method:  method,
 			Payload: payload,
+			// TODO: metadata from context
 		}
 
 		cresp = &Response{}
@@ -123,7 +125,7 @@ func (c *Client) Call(ctx context.Context, service, method string, req, resp int
 	}
 
 	if dl, ok := ctx.Deadline(); ok {
-		creq.TimeoutNano = dl.Sub(time.Now()).Nanoseconds()
+		creq.TimeoutNano = time.Until(dl).Nanoseconds()
 	}
 
 	info := &UnaryClientInfo{
@@ -143,36 +145,137 @@ func (c *Client) Call(ctx context.Context, service, method string, req, resp int
 	return nil
 }
 
-func (c *Client) dispatch(ctx context.Context, req *Request, resp *Response) error {
-	errs := make(chan error, 1)
-	call := &callRequest{
-		ctx:  ctx,
-		req:  req,
-		resp: resp,
-		errs: errs,
-	}
+// StreamDesc describes the stream properties, whether the stream has
+// a streaming client, a streaming server, or both
+type StreamDesc struct {
+	StreamingClient bool
+	StreamingServer bool
+}
 
-	select {
-	case <-ctx.Done():
-		return ctx.Err()
-	case c.calls <- call:
-	case <-c.ctx.Done():
-		return c.error()
-	}
+// ClientStream is used to send or recv messages on the underlying stream
+type ClientStream interface {
+	CloseSend() error
+	SendMsg(m interface{}) error
+	RecvMsg(m interface{}) error
+}
 
-	select {
-	case <-ctx.Done():
-		return ctx.Err()
-	case err := <-errs:
+type clientStream struct {
+	ctx          context.Context
+	s            *stream
+	c            *Client
+	desc         *StreamDesc
+	localClosed  bool
+	remoteClosed bool
+}
+
+func (cs *clientStream) CloseSend() error {
+	if !cs.desc.StreamingClient {
+		return fmt.Errorf("%w: cannot close non-streaming client", ErrProtocol)
+	}
+	if cs.localClosed {
+		return ErrStreamClosed
+	}
+	err := cs.s.send(messageTypeData, flagRemoteClosed|flagNoData, nil)
+	if err != nil {
 		return filterCloseErr(err)
-	case <-c.ctx.Done():
-		return c.error()
+	}
+	cs.localClosed = true
+	return nil
+}
+
+func (cs *clientStream) SendMsg(m interface{}) error {
+	if !cs.desc.StreamingClient {
+		return fmt.Errorf("%w: cannot send data from non-streaming client", ErrProtocol)
+	}
+	if cs.localClosed {
+		return ErrStreamClosed
+	}
+
+	var (
+		payload []byte
+		err     error
+	)
+	if m != nil {
+		payload, err = cs.c.codec.Marshal(m)
+		if err != nil {
+			return err
+		}
+	}
+
+	err = cs.s.send(messageTypeData, 0, payload)
+	if err != nil {
+		return filterCloseErr(err)
+	}
+
+	return nil
+}
+
+func (cs *clientStream) RecvMsg(m interface{}) error {
+	if cs.remoteClosed {
+		return io.EOF
+	}
+	select {
+	case <-cs.ctx.Done():
+		return cs.ctx.Err()
+	case msg, ok := <-cs.s.recv:
+		if !ok {
+			return cs.s.recvErr
+		}
+
+		if msg.header.Type == messageTypeResponse {
+			resp := &Response{}
+			err := proto.Unmarshal(msg.payload[:msg.header.Length], resp)
+			// return the payload buffer for reuse
+			cs.c.channel.putmbuf(msg.payload)
+			if err != nil {
+				return err
+			}
+
+			if err := cs.c.codec.Unmarshal(resp.Payload, m); err != nil {
+				return err
+			}
+
+			if resp.Status != nil && resp.Status.Code != int32(codes.OK) {
+				return status.ErrorProto(resp.Status)
+			}
+
+			cs.c.deleteStream(cs.s)
+			cs.remoteClosed = true
+
+			return nil
+		} else if msg.header.Type == messageTypeData {
+			if !cs.desc.StreamingServer {
+				cs.c.deleteStream(cs.s)
+				cs.remoteClosed = true
+				return fmt.Errorf("received data from non-streaming server: %w", ErrProtocol)
+			}
+			if msg.header.Flags&flagRemoteClosed == flagRemoteClosed {
+				cs.c.deleteStream(cs.s)
+				cs.remoteClosed = true
+
+				if msg.header.Flags&flagNoData == flagNoData {
+					return io.EOF
+				}
+			}
+
+			err := cs.c.codec.Unmarshal(msg.payload[:msg.header.Length], m)
+			cs.c.channel.putmbuf(msg.payload)
+			if err != nil {
+				return err
+			}
+			return nil
+		}
+
+		return fmt.Errorf("unexpected %q message received: %w", msg.header.Type, ErrProtocol)
 	}
 }
 
+// Close closes the ttrpc connection and underlying connection
 func (c *Client) Close() error {
 	c.closeOnce.Do(func() {
 		c.closed()
+
+		c.conn.Close()
 	})
 	return nil
 }
@@ -188,194 +291,105 @@ func (c *Client) UserOnCloseWait(ctx context.Context) error {
 	}
 }
 
-type message struct {
-	messageHeader
-	p   []byte
-	err error
-}
-
-// callMap provides access to a map of active calls, guarded by a mutex.
-type callMap struct {
-	m           sync.Mutex
-	activeCalls map[uint32]*callRequest
-	closeErr    error
-}
-
-// newCallMap returns a new callMap with an empty set of active calls.
-func newCallMap() *callMap {
-	return &callMap{
-		activeCalls: make(map[uint32]*callRequest),
-	}
-}
-
-// set adds a call entry to the map with the given streamID key.
-func (cm *callMap) set(streamID uint32, cr *callRequest) error {
-	cm.m.Lock()
-	defer cm.m.Unlock()
-	if cm.closeErr != nil {
-		return cm.closeErr
-	}
-	cm.activeCalls[streamID] = cr
-	return nil
-}
-
-// get looks up the call entry for the given streamID key, then removes it
-// from the map and returns it.
-func (cm *callMap) get(streamID uint32) (cr *callRequest, ok bool, err error) {
-	cm.m.Lock()
-	defer cm.m.Unlock()
-	if cm.closeErr != nil {
-		return nil, false, cm.closeErr
-	}
-	cr, ok = cm.activeCalls[streamID]
-	if ok {
-		delete(cm.activeCalls, streamID)
-	}
-	return
-}
-
-// abort sends the given error to each active call, and clears the map.
-// Once abort has been called, any subsequent calls to the callMap will return the error passed to abort.
-func (cm *callMap) abort(err error) error {
-	cm.m.Lock()
-	defer cm.m.Unlock()
-	if cm.closeErr != nil {
-		return cm.closeErr
-	}
-	for streamID, call := range cm.activeCalls {
-		call.errs <- err
-		delete(cm.activeCalls, streamID)
-	}
-	cm.closeErr = err
-	return nil
-}
-
 func (c *Client) run() {
-	var (
-		waiters      = newCallMap()
-		receiverDone = make(chan struct{})
-	)
+	err := c.receiveLoop()
+	c.Close()
+	c.cleanupStreams(err)
 
-	// Sender goroutine
-	// Receives calls from dispatch, adds them to the set of active calls, and sends them
-	// to the server.
-	go func() {
-		var streamID uint32 = 1
-		for {
-			select {
-			case <-c.ctx.Done():
-				return
-			case call := <-c.calls:
-				id := streamID
-				streamID += 2 // enforce odd client initiated request ids
-				if err := waiters.set(id, call); err != nil {
-					call.errs <- err // errs is buffered so should not block.
-					continue
-				}
-				if err := c.send(id, messageTypeRequest, call.req); err != nil {
-					call.errs <- err // errs is buffered so should not block.
-					waiters.get(id)  // remove from waiters set
-				}
-			}
-		}
-	}()
-
-	// Receiver goroutine
-	// Receives responses from the server, looks up the call info in the set of active calls,
-	// and notifies the caller of the response.
-	go func() {
-		defer close(receiverDone)
-		for {
-			select {
-			case <-c.ctx.Done():
-				c.setError(c.ctx.Err())
-				return
-			default:
-				mh, p, err := c.channel.recv()
-				if err != nil {
-					_, ok := status.FromError(err)
-					if !ok {
-						// treat all errors that are not an rpc status as terminal.
-						// all others poison the connection.
-						c.setError(filterCloseErr(err))
-						return
-					}
-				}
-				msg := &message{
-					messageHeader: mh,
-					p:             p[:mh.Length],
-					err:           err,
-				}
-				call, ok, err := waiters.get(mh.StreamID)
-				if err != nil {
-					logrus.Errorf("ttrpc: failed to look up active call: %s", err)
-					continue
-				}
-				if !ok {
-					logrus.Errorf("ttrpc: received message for unknown channel %v", mh.StreamID)
-					continue
-				}
-				call.errs <- c.recv(call.resp, msg)
-			}
-		}
-	}()
-
-	defer func() {
-		c.conn.Close()
-		c.userCloseFunc()
-		close(c.userCloseWaitCh)
-	}()
+	c.userCloseFunc()
+	close(c.userCloseWaitCh)
+}
 
+func (c *Client) receiveLoop() error {
 	for {
 		select {
-		case <-receiverDone:
-			// The receiver has exited.
-			// don't return out, let the close of the context trigger the abort of waiters
-			c.Close()
 		case <-c.ctx.Done():
-			// Abort all active calls. This will also prevent any new calls from being added
-			// to waiters.
-			waiters.abort(c.error())
-			return
+			return ErrClosed
+		default:
+			var (
+				msg = &streamMessage{}
+				err error
+			)
+
+			msg.header, msg.payload, err = c.channel.recv()
+			if err != nil {
+				_, ok := status.FromError(err)
+				if !ok {
+					// treat all errors that are not an rpc status as terminal.
+					// all others poison the connection.
+					return filterCloseErr(err)
+				}
+			}
+			sid := streamID(msg.header.StreamID)
+			s := c.getStream(sid)
+			if s == nil {
+				logrus.WithField("stream", sid).Errorf("ttrpc: received message on inactive stream")
+				continue
+			}
+
+			if err != nil {
+				s.closeWithError(err)
+			} else {
+				if err := s.receive(c.ctx, msg); err != nil {
+					logrus.WithError(err).WithField("stream", sid).Errorf("ttrpc: failed to handle message")
+				}
+			}
 		}
 	}
 }
 
-func (c *Client) error() error {
-	c.errOnce.Do(func() {
-		if c.err == nil {
-			c.err = ErrClosed
-		}
-	})
-	return c.err
-}
+// createStream creates a new stream and registers it with the client
+// Introduce stream types for multiple or single response
+func (c *Client) createStream(flags uint8, b []byte) (*stream, error) {
+	c.streamLock.Lock()
 
-func (c *Client) setError(err error) {
-	c.errOnce.Do(func() {
-		c.err = err
-	})
-}
-
-func (c *Client) send(streamID uint32, mtype messageType, msg interface{}) error {
-	p, err := c.codec.Marshal(msg)
-	if err != nil {
-		return err
+	// Check if closed since lock acquired to prevent adding
+	// anything after cleanup completes
+	select {
+	case <-c.ctx.Done():
+		c.streamLock.Unlock()
+		return nil, ErrClosed
+	default:
 	}
 
-	return c.channel.send(streamID, mtype, p)
+	// Stream ID should be allocated at same time
+	s := newStream(c.nextStreamID, c)
+	c.streams[s.id] = s
+	c.nextStreamID = c.nextStreamID + 2
+
+	c.sendLock.Lock()
+	defer c.sendLock.Unlock()
+	c.streamLock.Unlock()
+
+	if err := c.channel.send(uint32(s.id), messageTypeRequest, flags, b); err != nil {
+		return s, filterCloseErr(err)
+	}
+
+	return s, nil
 }
 
-func (c *Client) recv(resp *Response, msg *message) error {
-	if msg.err != nil {
-		return msg.err
-	}
+func (c *Client) deleteStream(s *stream) {
+	c.streamLock.Lock()
+	delete(c.streams, s.id)
+	c.streamLock.Unlock()
+	s.closeWithError(nil)
+}
 
-	if msg.Type != messageTypeResponse {
-		return errors.New("unknown message type received")
-	}
+func (c *Client) getStream(sid streamID) *stream {
+	c.streamLock.RLock()
+	s := c.streams[sid]
+	c.streamLock.RUnlock()
+	return s
+}
 
-	defer c.channel.putmbuf(msg.p)
-	return proto.Unmarshal(msg.p, resp)
+func (c *Client) cleanupStreams(err error) {
+	c.streamLock.Lock()
+	defer c.streamLock.Unlock()
+
+	for sid, s := range c.streams {
+		s.closeWithError(err)
+		delete(c.streams, sid)
+	}
 }
 
 // filterCloseErr rewrites EOF and EPIPE errors to ErrClosed. Use when
@@ -388,6 +402,8 @@ func filterCloseErr(err error) error {
 		return nil
 	case err == io.EOF:
 		return ErrClosed
+	case errors.Is(err, io.ErrClosedPipe):
+		return ErrClosed
 	case errors.Is(err, io.EOF):
 		return ErrClosed
 	case strings.Contains(err.Error(), "use of closed network connection"):
@@ -395,11 +411,9 @@ func filterCloseErr(err error) error {
 	default:
 		// if we have an epipe on a write or econnreset on a read , we cast to errclosed
 		var oerr *net.OpError
-		if errors.As(err, &oerr) && (oerr.Op == "write" || oerr.Op == "read") {
-			serr, sok := oerr.Err.(*os.SyscallError)
-			if sok && ((serr.Err == syscall.EPIPE && oerr.Op == "write") ||
-				(serr.Err == syscall.ECONNRESET && oerr.Op == "read")) {
-
+		if errors.As(err, &oerr) {
+			if (oerr.Op == "write" && errors.Is(err, syscall.EPIPE)) ||
+				(oerr.Op == "read" && errors.Is(err, syscall.ECONNRESET)) {
 				return ErrClosed
 			}
 		}
@@ -407,3 +421,81 @@ func filterCloseErr(err error) error {
 
 	return err
 }
+
+// NewStream creates a new stream with the given stream descriptor to the
+// specified service and method. If not a streaming client, the request object
+// may be provided.
+func (c *Client) NewStream(ctx context.Context, desc *StreamDesc, service, method string, req interface{}) (ClientStream, error) {
+	var payload []byte
+	if req != nil {
+		var err error
+		payload, err = c.codec.Marshal(req)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	request := &Request{
+		Service: service,
+		Method:  method,
+		Payload: payload,
+		// TODO: metadata from context
+	}
+	p, err := c.codec.Marshal(request)
+	if err != nil {
+		return nil, err
+	}
+
+	var flags uint8
+	if desc.StreamingClient {
+		flags = flagRemoteOpen
+	} else {
+		flags = flagRemoteClosed
+	}
+	s, err := c.createStream(flags, p)
+	if err != nil {
+		return nil, err
+	}
+
+	return &clientStream{
+		ctx:  ctx,
+		s:    s,
+		c:    c,
+		desc: desc,
+	}, nil
+}
+
+func (c *Client) dispatch(ctx context.Context, req *Request, resp *Response) error {
+	p, err := c.codec.Marshal(req)
+	if err != nil {
+		return err
+	}
+
+	s, err := c.createStream(0, p)
+	if err != nil {
+		return err
+	}
+	defer c.deleteStream(s)
+
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-c.ctx.Done():
+		return ErrClosed
+	case msg, ok := <-s.recv:
+		if !ok {
+			return s.recvErr
+		}
+
+		if msg.header.Type == messageTypeResponse {
+			err = proto.Unmarshal(msg.payload[:msg.header.Length], resp)
+		} else {
+			err = fmt.Errorf("unexpected %q message received: %w", msg.header.Type, ErrProtocol)
+		}
+
+		// return the payload buffer for reuse
+		c.channel.putmbuf(msg.payload)
+
+		return err
+	}
+}