mirror of
https://github.com/rclone/rclone.git
synced 2024-12-11 17:51:12 +01:00
8ec57d145e
Re-added cobra patch 499475bb41
612 lines
18 KiB
Go
612 lines
18 KiB
Go
/*
|
|
*
|
|
* Copyright 2014, Google Inc.
|
|
* All rights reserved.
|
|
*
|
|
* Redistribution and use in source and binary forms, with or without
|
|
* modification, are permitted provided that the following conditions are
|
|
* met:
|
|
*
|
|
* * Redistributions of source code must retain the above copyright
|
|
* notice, this list of conditions and the following disclaimer.
|
|
* * Redistributions in binary form must reproduce the above
|
|
* copyright notice, this list of conditions and the following disclaimer
|
|
* in the documentation and/or other materials provided with the
|
|
* distribution.
|
|
* * Neither the name of Google Inc. nor the names of its
|
|
* contributors may be used to endorse or promote products derived from
|
|
* this software without specific prior written permission.
|
|
*
|
|
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
|
|
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
|
|
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
|
|
* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
|
|
* OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
|
|
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
|
|
* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
|
|
* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
|
|
* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
|
|
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
|
|
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
|
|
*
|
|
*/
|
|
|
|
/*
|
|
Package transport defines and implements message oriented communication channel
|
|
to complete various transactions (e.g., an RPC).
|
|
*/
|
|
package transport
|
|
|
|
import (
|
|
"bytes"
|
|
"fmt"
|
|
"io"
|
|
"net"
|
|
"sync"
|
|
|
|
"golang.org/x/net/context"
|
|
"google.golang.org/grpc/codes"
|
|
"google.golang.org/grpc/credentials"
|
|
"google.golang.org/grpc/metadata"
|
|
"google.golang.org/grpc/stats"
|
|
"google.golang.org/grpc/tap"
|
|
)
|
|
|
|
// recvMsg represents the received msg from the transport. All transport
|
|
// protocol specific info has been removed.
|
|
type recvMsg struct {
|
|
data []byte
|
|
// nil: received some data
|
|
// io.EOF: stream is completed. data is nil.
|
|
// other non-nil error: transport failure. data is nil.
|
|
err error
|
|
}
|
|
|
|
func (*recvMsg) item() {}
|
|
|
|
// All items in an out of a recvBuffer should be the same type.
|
|
type item interface {
|
|
item()
|
|
}
|
|
|
|
// recvBuffer is an unbounded channel of item.
|
|
type recvBuffer struct {
|
|
c chan item
|
|
mu sync.Mutex
|
|
backlog []item
|
|
}
|
|
|
|
func newRecvBuffer() *recvBuffer {
|
|
b := &recvBuffer{
|
|
c: make(chan item, 1),
|
|
}
|
|
return b
|
|
}
|
|
|
|
func (b *recvBuffer) put(r item) {
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
if len(b.backlog) == 0 {
|
|
select {
|
|
case b.c <- r:
|
|
return
|
|
default:
|
|
}
|
|
}
|
|
b.backlog = append(b.backlog, r)
|
|
}
|
|
|
|
func (b *recvBuffer) load() {
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
if len(b.backlog) > 0 {
|
|
select {
|
|
case b.c <- b.backlog[0]:
|
|
b.backlog = b.backlog[1:]
|
|
default:
|
|
}
|
|
}
|
|
}
|
|
|
|
// get returns the channel that receives an item in the buffer.
|
|
//
|
|
// Upon receipt of an item, the caller should call load to send another
|
|
// item onto the channel if there is any.
|
|
func (b *recvBuffer) get() <-chan item {
|
|
return b.c
|
|
}
|
|
|
|
// recvBufferReader implements io.Reader interface to read the data from
|
|
// recvBuffer.
|
|
type recvBufferReader struct {
|
|
ctx context.Context
|
|
goAway chan struct{}
|
|
recv *recvBuffer
|
|
last *bytes.Reader // Stores the remaining data in the previous calls.
|
|
err error
|
|
}
|
|
|
|
// Read reads the next len(p) bytes from last. If last is drained, it tries to
|
|
// read additional data from recv. It blocks if there no additional data available
|
|
// in recv. If Read returns any non-nil error, it will continue to return that error.
|
|
func (r *recvBufferReader) Read(p []byte) (n int, err error) {
|
|
if r.err != nil {
|
|
return 0, r.err
|
|
}
|
|
defer func() { r.err = err }()
|
|
if r.last != nil && r.last.Len() > 0 {
|
|
// Read remaining data left in last call.
|
|
return r.last.Read(p)
|
|
}
|
|
select {
|
|
case <-r.ctx.Done():
|
|
return 0, ContextErr(r.ctx.Err())
|
|
case <-r.goAway:
|
|
return 0, ErrStreamDrain
|
|
case i := <-r.recv.get():
|
|
r.recv.load()
|
|
m := i.(*recvMsg)
|
|
if m.err != nil {
|
|
return 0, m.err
|
|
}
|
|
r.last = bytes.NewReader(m.data)
|
|
return r.last.Read(p)
|
|
}
|
|
}
|
|
|
|
type streamState uint8
|
|
|
|
const (
|
|
streamActive streamState = iota
|
|
streamWriteDone // EndStream sent
|
|
streamReadDone // EndStream received
|
|
streamDone // the entire stream is finished.
|
|
)
|
|
|
|
// Stream represents an RPC in the transport layer.
|
|
type Stream struct {
|
|
id uint32
|
|
// nil for client side Stream.
|
|
st ServerTransport
|
|
// clientStatsCtx keeps the user context for stats handling.
|
|
// It's only valid on client side. Server side stats context is same as s.ctx.
|
|
// All client side stats collection should use the clientStatsCtx (instead of the stream context)
|
|
// so that all the generated stats for a particular RPC can be associated in the processing phase.
|
|
clientStatsCtx context.Context
|
|
// ctx is the associated context of the stream.
|
|
ctx context.Context
|
|
// cancel is always nil for client side Stream.
|
|
cancel context.CancelFunc
|
|
// done is closed when the final status arrives.
|
|
done chan struct{}
|
|
// goAway is closed when the server sent GoAways signal before this stream was initiated.
|
|
goAway chan struct{}
|
|
// method records the associated RPC method of the stream.
|
|
method string
|
|
recvCompress string
|
|
sendCompress string
|
|
buf *recvBuffer
|
|
dec io.Reader
|
|
fc *inFlow
|
|
recvQuota uint32
|
|
// The accumulated inbound quota pending for window update.
|
|
updateQuota uint32
|
|
// The handler to control the window update procedure for both this
|
|
// particular stream and the associated transport.
|
|
windowHandler func(int)
|
|
|
|
sendQuotaPool *quotaPool
|
|
// Close headerChan to indicate the end of reception of header metadata.
|
|
headerChan chan struct{}
|
|
// header caches the received header metadata.
|
|
header metadata.MD
|
|
// The key-value map of trailer metadata.
|
|
trailer metadata.MD
|
|
|
|
mu sync.RWMutex // guard the following
|
|
// headerOK becomes true from the first header is about to send.
|
|
headerOk bool
|
|
state streamState
|
|
// true iff headerChan is closed. Used to avoid closing headerChan
|
|
// multiple times.
|
|
headerDone bool
|
|
// the status received from the server.
|
|
statusCode codes.Code
|
|
statusDesc string
|
|
}
|
|
|
|
// RecvCompress returns the compression algorithm applied to the inbound
|
|
// message. It is empty string if there is no compression applied.
|
|
func (s *Stream) RecvCompress() string {
|
|
return s.recvCompress
|
|
}
|
|
|
|
// SetSendCompress sets the compression algorithm to the stream.
|
|
func (s *Stream) SetSendCompress(str string) {
|
|
s.sendCompress = str
|
|
}
|
|
|
|
// Done returns a chanel which is closed when it receives the final status
|
|
// from the server.
|
|
func (s *Stream) Done() <-chan struct{} {
|
|
return s.done
|
|
}
|
|
|
|
// GoAway returns a channel which is closed when the server sent GoAways signal
|
|
// before this stream was initiated.
|
|
func (s *Stream) GoAway() <-chan struct{} {
|
|
return s.goAway
|
|
}
|
|
|
|
// Header acquires the key-value pairs of header metadata once it
|
|
// is available. It blocks until i) the metadata is ready or ii) there is no
|
|
// header metadata or iii) the stream is cancelled/expired.
|
|
func (s *Stream) Header() (metadata.MD, error) {
|
|
select {
|
|
case <-s.ctx.Done():
|
|
return nil, ContextErr(s.ctx.Err())
|
|
case <-s.goAway:
|
|
return nil, ErrStreamDrain
|
|
case <-s.headerChan:
|
|
return s.header.Copy(), nil
|
|
}
|
|
}
|
|
|
|
// Trailer returns the cached trailer metedata. Note that if it is not called
|
|
// after the entire stream is done, it could return an empty MD. Client
|
|
// side only.
|
|
func (s *Stream) Trailer() metadata.MD {
|
|
s.mu.RLock()
|
|
defer s.mu.RUnlock()
|
|
return s.trailer.Copy()
|
|
}
|
|
|
|
// ServerTransport returns the underlying ServerTransport for the stream.
|
|
// The client side stream always returns nil.
|
|
func (s *Stream) ServerTransport() ServerTransport {
|
|
return s.st
|
|
}
|
|
|
|
// Context returns the context of the stream.
|
|
func (s *Stream) Context() context.Context {
|
|
return s.ctx
|
|
}
|
|
|
|
// Method returns the method for the stream.
|
|
func (s *Stream) Method() string {
|
|
return s.method
|
|
}
|
|
|
|
// StatusCode returns statusCode received from the server.
|
|
func (s *Stream) StatusCode() codes.Code {
|
|
return s.statusCode
|
|
}
|
|
|
|
// StatusDesc returns statusDesc received from the server.
|
|
func (s *Stream) StatusDesc() string {
|
|
return s.statusDesc
|
|
}
|
|
|
|
// SetHeader sets the header metadata. This can be called multiple times.
|
|
// Server side only.
|
|
func (s *Stream) SetHeader(md metadata.MD) error {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
if s.headerOk || s.state == streamDone {
|
|
return ErrIllegalHeaderWrite
|
|
}
|
|
if md.Len() == 0 {
|
|
return nil
|
|
}
|
|
s.header = metadata.Join(s.header, md)
|
|
return nil
|
|
}
|
|
|
|
// SetTrailer sets the trailer metadata which will be sent with the RPC status
|
|
// by the server. This can be called multiple times. Server side only.
|
|
func (s *Stream) SetTrailer(md metadata.MD) error {
|
|
if md.Len() == 0 {
|
|
return nil
|
|
}
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
s.trailer = metadata.Join(s.trailer, md)
|
|
return nil
|
|
}
|
|
|
|
func (s *Stream) write(m recvMsg) {
|
|
s.buf.put(&m)
|
|
}
|
|
|
|
// Read reads all the data available for this Stream from the transport and
|
|
// passes them into the decoder, which converts them into a gRPC message stream.
|
|
// The error is io.EOF when the stream is done or another non-nil error if
|
|
// the stream broke.
|
|
func (s *Stream) Read(p []byte) (n int, err error) {
|
|
n, err = s.dec.Read(p)
|
|
if err != nil {
|
|
return
|
|
}
|
|
s.windowHandler(n)
|
|
return
|
|
}
|
|
|
|
// The key to save transport.Stream in the context.
|
|
type streamKey struct{}
|
|
|
|
// newContextWithStream creates a new context from ctx and attaches stream
|
|
// to it.
|
|
func newContextWithStream(ctx context.Context, stream *Stream) context.Context {
|
|
return context.WithValue(ctx, streamKey{}, stream)
|
|
}
|
|
|
|
// StreamFromContext returns the stream saved in ctx.
|
|
func StreamFromContext(ctx context.Context) (s *Stream, ok bool) {
|
|
s, ok = ctx.Value(streamKey{}).(*Stream)
|
|
return
|
|
}
|
|
|
|
// state of transport
|
|
type transportState int
|
|
|
|
const (
|
|
reachable transportState = iota
|
|
unreachable
|
|
closing
|
|
draining
|
|
)
|
|
|
|
// ServerConfig consists of all the configurations to establish a server transport.
|
|
type ServerConfig struct {
|
|
MaxStreams uint32
|
|
AuthInfo credentials.AuthInfo
|
|
InTapHandle tap.ServerInHandle
|
|
StatsHandler stats.Handler
|
|
}
|
|
|
|
// NewServerTransport creates a ServerTransport with conn or non-nil error
|
|
// if it fails.
|
|
func NewServerTransport(protocol string, conn net.Conn, config *ServerConfig) (ServerTransport, error) {
|
|
return newHTTP2Server(conn, config)
|
|
}
|
|
|
|
// ConnectOptions covers all relevant options for communicating with the server.
|
|
type ConnectOptions struct {
|
|
// UserAgent is the application user agent.
|
|
UserAgent string
|
|
// Authority is the :authority pseudo-header to use. This field has no effect if
|
|
// TransportCredentials is set.
|
|
Authority string
|
|
// Dialer specifies how to dial a network address.
|
|
Dialer func(context.Context, string) (net.Conn, error)
|
|
// FailOnNonTempDialError specifies if gRPC fails on non-temporary dial errors.
|
|
FailOnNonTempDialError bool
|
|
// PerRPCCredentials stores the PerRPCCredentials required to issue RPCs.
|
|
PerRPCCredentials []credentials.PerRPCCredentials
|
|
// TransportCredentials stores the Authenticator required to setup a client connection.
|
|
TransportCredentials credentials.TransportCredentials
|
|
// StatsHandler stores the handler for stats.
|
|
StatsHandler stats.Handler
|
|
}
|
|
|
|
// TargetInfo contains the information of the target such as network address and metadata.
|
|
type TargetInfo struct {
|
|
Addr string
|
|
Metadata interface{}
|
|
}
|
|
|
|
// NewClientTransport establishes the transport with the required ConnectOptions
|
|
// and returns it to the caller.
|
|
func NewClientTransport(ctx context.Context, target TargetInfo, opts ConnectOptions) (ClientTransport, error) {
|
|
return newHTTP2Client(ctx, target, opts)
|
|
}
|
|
|
|
// Options provides additional hints and information for message
|
|
// transmission.
|
|
type Options struct {
|
|
// Last indicates whether this write is the last piece for
|
|
// this stream.
|
|
Last bool
|
|
|
|
// Delay is a hint to the transport implementation for whether
|
|
// the data could be buffered for a batching write. The
|
|
// Transport implementation may ignore the hint.
|
|
Delay bool
|
|
}
|
|
|
|
// CallHdr carries the information of a particular RPC.
|
|
type CallHdr struct {
|
|
// Host specifies the peer's host.
|
|
Host string
|
|
|
|
// Method specifies the operation to perform.
|
|
Method string
|
|
|
|
// RecvCompress specifies the compression algorithm applied on
|
|
// inbound messages.
|
|
RecvCompress string
|
|
|
|
// SendCompress specifies the compression algorithm applied on
|
|
// outbound message.
|
|
SendCompress string
|
|
|
|
// Flush indicates whether a new stream command should be sent
|
|
// to the peer without waiting for the first data. This is
|
|
// only a hint. The transport may modify the flush decision
|
|
// for performance purposes.
|
|
Flush bool
|
|
}
|
|
|
|
// ClientTransport is the common interface for all gRPC client-side transport
|
|
// implementations.
|
|
type ClientTransport interface {
|
|
// Close tears down this transport. Once it returns, the transport
|
|
// should not be accessed any more. The caller must make sure this
|
|
// is called only once.
|
|
Close() error
|
|
|
|
// GracefulClose starts to tear down the transport. It stops accepting
|
|
// new RPCs and wait the completion of the pending RPCs.
|
|
GracefulClose() error
|
|
|
|
// Write sends the data for the given stream. A nil stream indicates
|
|
// the write is to be performed on the transport as a whole.
|
|
Write(s *Stream, data []byte, opts *Options) error
|
|
|
|
// NewStream creates a Stream for an RPC.
|
|
NewStream(ctx context.Context, callHdr *CallHdr) (*Stream, error)
|
|
|
|
// CloseStream clears the footprint of a stream when the stream is
|
|
// not needed any more. The err indicates the error incurred when
|
|
// CloseStream is called. Must be called when a stream is finished
|
|
// unless the associated transport is closing.
|
|
CloseStream(stream *Stream, err error)
|
|
|
|
// Error returns a channel that is closed when some I/O error
|
|
// happens. Typically the caller should have a goroutine to monitor
|
|
// this in order to take action (e.g., close the current transport
|
|
// and create a new one) in error case. It should not return nil
|
|
// once the transport is initiated.
|
|
Error() <-chan struct{}
|
|
|
|
// GoAway returns a channel that is closed when ClientTranspor
|
|
// receives the draining signal from the server (e.g., GOAWAY frame in
|
|
// HTTP/2).
|
|
GoAway() <-chan struct{}
|
|
}
|
|
|
|
// ServerTransport is the common interface for all gRPC server-side transport
|
|
// implementations.
|
|
//
|
|
// Methods may be called concurrently from multiple goroutines, but
|
|
// Write methods for a given Stream will be called serially.
|
|
type ServerTransport interface {
|
|
// HandleStreams receives incoming streams using the given handler.
|
|
HandleStreams(func(*Stream), func(context.Context, string) context.Context)
|
|
|
|
// WriteHeader sends the header metadata for the given stream.
|
|
// WriteHeader may not be called on all streams.
|
|
WriteHeader(s *Stream, md metadata.MD) error
|
|
|
|
// Write sends the data for the given stream.
|
|
// Write may not be called on all streams.
|
|
Write(s *Stream, data []byte, opts *Options) error
|
|
|
|
// WriteStatus sends the status of a stream to the client.
|
|
// WriteStatus is the final call made on a stream and always
|
|
// occurs.
|
|
WriteStatus(s *Stream, statusCode codes.Code, statusDesc string) error
|
|
|
|
// Close tears down the transport. Once it is called, the transport
|
|
// should not be accessed any more. All the pending streams and their
|
|
// handlers will be terminated asynchronously.
|
|
Close() error
|
|
|
|
// RemoteAddr returns the remote network address.
|
|
RemoteAddr() net.Addr
|
|
|
|
// Drain notifies the client this ServerTransport stops accepting new RPCs.
|
|
Drain()
|
|
}
|
|
|
|
// streamErrorf creates an StreamError with the specified error code and description.
|
|
func streamErrorf(c codes.Code, format string, a ...interface{}) StreamError {
|
|
return StreamError{
|
|
Code: c,
|
|
Desc: fmt.Sprintf(format, a...),
|
|
}
|
|
}
|
|
|
|
// connectionErrorf creates an ConnectionError with the specified error description.
|
|
func connectionErrorf(temp bool, e error, format string, a ...interface{}) ConnectionError {
|
|
return ConnectionError{
|
|
Desc: fmt.Sprintf(format, a...),
|
|
temp: temp,
|
|
err: e,
|
|
}
|
|
}
|
|
|
|
// ConnectionError is an error that results in the termination of the
|
|
// entire connection and the retry of all the active streams.
|
|
type ConnectionError struct {
|
|
Desc string
|
|
temp bool
|
|
err error
|
|
}
|
|
|
|
func (e ConnectionError) Error() string {
|
|
return fmt.Sprintf("connection error: desc = %q", e.Desc)
|
|
}
|
|
|
|
// Temporary indicates if this connection error is temporary or fatal.
|
|
func (e ConnectionError) Temporary() bool {
|
|
return e.temp
|
|
}
|
|
|
|
// Origin returns the original error of this connection error.
|
|
func (e ConnectionError) Origin() error {
|
|
// Never return nil error here.
|
|
// If the original error is nil, return itself.
|
|
if e.err == nil {
|
|
return e
|
|
}
|
|
return e.err
|
|
}
|
|
|
|
var (
|
|
// ErrConnClosing indicates that the transport is closing.
|
|
ErrConnClosing = connectionErrorf(true, nil, "transport is closing")
|
|
// ErrStreamDrain indicates that the stream is rejected by the server because
|
|
// the server stops accepting new RPCs.
|
|
ErrStreamDrain = streamErrorf(codes.Unavailable, "the server stops accepting new RPCs")
|
|
)
|
|
|
|
// StreamError is an error that only affects one stream within a connection.
|
|
type StreamError struct {
|
|
Code codes.Code
|
|
Desc string
|
|
}
|
|
|
|
func (e StreamError) Error() string {
|
|
return fmt.Sprintf("stream error: code = %d desc = %q", e.Code, e.Desc)
|
|
}
|
|
|
|
// ContextErr converts the error from context package into a StreamError.
|
|
func ContextErr(err error) StreamError {
|
|
switch err {
|
|
case context.DeadlineExceeded:
|
|
return streamErrorf(codes.DeadlineExceeded, "%v", err)
|
|
case context.Canceled:
|
|
return streamErrorf(codes.Canceled, "%v", err)
|
|
}
|
|
panic(fmt.Sprintf("Unexpected error from context packet: %v", err))
|
|
}
|
|
|
|
// wait blocks until it can receive from ctx.Done, closing, or proceed.
|
|
// If it receives from ctx.Done, it returns 0, the StreamError for ctx.Err.
|
|
// If it receives from done, it returns 0, io.EOF if ctx is not done; otherwise
|
|
// it return the StreamError for ctx.Err.
|
|
// If it receives from goAway, it returns 0, ErrStreamDrain.
|
|
// If it receives from closing, it returns 0, ErrConnClosing.
|
|
// If it receives from proceed, it returns the received integer, nil.
|
|
func wait(ctx context.Context, done, goAway, closing <-chan struct{}, proceed <-chan int) (int, error) {
|
|
select {
|
|
case <-ctx.Done():
|
|
return 0, ContextErr(ctx.Err())
|
|
case <-done:
|
|
// User cancellation has precedence.
|
|
select {
|
|
case <-ctx.Done():
|
|
return 0, ContextErr(ctx.Err())
|
|
default:
|
|
}
|
|
return 0, io.EOF
|
|
case <-goAway:
|
|
return 0, ErrStreamDrain
|
|
case <-closing:
|
|
return 0, ErrConnClosing
|
|
case i := <-proceed:
|
|
return i, nil
|
|
}
|
|
}
|