vendor/google.golang.org/grpc/rpc_util.go

   1 /*
   2  *
   3  * Copyright 2014, Google Inc.
   4  * All rights reserved.
   5  *
   6  * Redistribution and use in source and binary forms, with or without
   7  * modification, are permitted provided that the following conditions are
   8  * met:
   9  *
  10  *     * Redistributions of source code must retain the above copyright
  11  * notice, this list of conditions and the following disclaimer.
  12  *     * Redistributions in binary form must reproduce the above
  13  * copyright notice, this list of conditions and the following disclaimer
  14  * in the documentation and/or other materials provided with the
  15  * distribution.
  16  *     * Neither the name of Google Inc. nor the names of its
  17  * contributors may be used to endorse or promote products derived from
  18  * this software without specific prior written permission.
  19  *
  20  * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
  21  * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
  22  * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
  23  * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
  24  * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  25  * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  26  * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
  27  * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
  28  * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  29  * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
  30  * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  31  *
  32  */
  33
  34 package grpc
  35
  36 import (
  37         "bytes"
  38         "compress/gzip"
  39         "encoding/binary"
  40         "io"
  41         "io/ioutil"
  42         "math"
  43         "os"
  44         "time"
  45
  46         "golang.org/x/net/context"
  47         "google.golang.org/grpc/codes"
  48         "google.golang.org/grpc/metadata"
  49         "google.golang.org/grpc/peer"
  50         "google.golang.org/grpc/stats"
  51         "google.golang.org/grpc/status"
  52         "google.golang.org/grpc/transport"
  53 )
  54
  55 // Compressor defines the interface gRPC uses to compress a message.
  56 type Compressor interface {
  57         // Do compresses p into w.
  58         Do(w io.Writer, p []byte) error
  59         // Type returns the compression algorithm the Compressor uses.
  60         Type() string
  61 }
  62
  63 // NewGZIPCompressor creates a Compressor based on GZIP.
  64 func NewGZIPCompressor() Compressor {
  65         return &gzipCompressor{}
  66 }
  67
  68 type gzipCompressor struct {
  69 }
  70
  71 func (c *gzipCompressor) Do(w io.Writer, p []byte) error {
  72         z := gzip.NewWriter(w)
  73         if _, err := z.Write(p); err != nil {
  74                 return err
  75         }
  76         return z.Close()
  77 }
  78
  79 func (c *gzipCompressor) Type() string {
  80         return "gzip"
  81 }
  82
  83 // Decompressor defines the interface gRPC uses to decompress a message.
  84 type Decompressor interface {
  85         // Do reads the data from r and uncompress them.
  86         Do(r io.Reader) ([]byte, error)
  87         // Type returns the compression algorithm the Decompressor uses.
  88         Type() string
  89 }
  90
  91 type gzipDecompressor struct {
  92 }
  93
  94 // NewGZIPDecompressor creates a Decompressor based on GZIP.
  95 func NewGZIPDecompressor() Decompressor {
  96         return &gzipDecompressor{}
  97 }
  98
  99 func (d *gzipDecompressor) Do(r io.Reader) ([]byte, error) {
 100         z, err := gzip.NewReader(r)
 101         if err != nil {
 102                 return nil, err
 103         }
 104         defer z.Close()
 105         return ioutil.ReadAll(z)
 106 }
 107
 108 func (d *gzipDecompressor) Type() string {
 109         return "gzip"
 110 }
 111
 112 // callInfo contains all related configuration and information about an RPC.
 113 type callInfo struct {
 114         failFast  bool
 115         headerMD  metadata.MD
 116         trailerMD metadata.MD
 117         peer      *peer.Peer
 118         traceInfo traceInfo // in trace.go
 119 }
 120
 121 var defaultCallInfo = callInfo{failFast: true}
 122
 123 // CallOption configures a Call before it starts or extracts information from
 124 // a Call after it completes.
 125 type CallOption interface {
 126         // before is called before the call is sent to any server.  If before
 127         // returns a non-nil error, the RPC fails with that error.
 128         before(*callInfo) error
 129
 130         // after is called after the call has completed.  after cannot return an
 131         // error, so any failures should be reported via output parameters.
 132         after(*callInfo)
 133 }
 134
 135 type beforeCall func(c *callInfo) error
 136
 137 func (o beforeCall) before(c *callInfo) error { return o(c) }
 138 func (o beforeCall) after(c *callInfo)        {}
 139
 140 type afterCall func(c *callInfo)
 141
 142 func (o afterCall) before(c *callInfo) error { return nil }
 143 func (o afterCall) after(c *callInfo)        { o(c) }
 144
 145 // Header returns a CallOptions that retrieves the header metadata
 146 // for a unary RPC.
 147 func Header(md *metadata.MD) CallOption {
 148         return afterCall(func(c *callInfo) {
 149                 *md = c.headerMD
 150         })
 151 }
 152
 153 // Trailer returns a CallOptions that retrieves the trailer metadata
 154 // for a unary RPC.
 155 func Trailer(md *metadata.MD) CallOption {
 156         return afterCall(func(c *callInfo) {
 157                 *md = c.trailerMD
 158         })
 159 }
 160
 161 // Peer returns a CallOption that retrieves peer information for a
 162 // unary RPC.
 163 func Peer(peer *peer.Peer) CallOption {
 164         return afterCall(func(c *callInfo) {
 165                 if c.peer != nil {
 166                         *peer = *c.peer
 167                 }
 168         })
 169 }
 170
 171 // FailFast configures the action to take when an RPC is attempted on broken
 172 // connections or unreachable servers. If failfast is true, the RPC will fail
 173 // immediately. Otherwise, the RPC client will block the call until a
 174 // connection is available (or the call is canceled or times out) and will retry
 175 // the call if it fails due to a transient error. Please refer to
 176 // https://github.com/grpc/grpc/blob/master/doc/fail_fast.md. Note: failFast is default to true.
 177 func FailFast(failFast bool) CallOption {
 178         return beforeCall(func(c *callInfo) error {
 179                 c.failFast = failFast
 180                 return nil
 181         })
 182 }
 183
 184 // The format of the payload: compressed or not?
 185 type payloadFormat uint8
 186
 187 const (
 188         compressionNone payloadFormat = iota // no compression
 189         compressionMade
 190 )
 191
 192 // parser reads complete gRPC messages from the underlying reader.
 193 type parser struct {
 194         // r is the underlying reader.
 195         // See the comment on recvMsg for the permissible
 196         // error types.
 197         r io.Reader
 198
 199         // The header of a gRPC message. Find more detail
 200         // at http://www.grpc.io/docs/guides/wire.html.
 201         header [5]byte
 202 }
 203
 204 // recvMsg reads a complete gRPC message from the stream.
 205 //
 206 // It returns the message and its payload (compression/encoding)
 207 // format. The caller owns the returned msg memory.
 208 //
 209 // If there is an error, possible values are:
 210 //   * io.EOF, when no messages remain
 211 //   * io.ErrUnexpectedEOF
 212 //   * of type transport.ConnectionError
 213 //   * of type transport.StreamError
 214 // No other error values or types must be returned, which also means
 215 // that the underlying io.Reader must not return an incompatible
 216 // error.
 217 func (p *parser) recvMsg(maxMsgSize int) (pf payloadFormat, msg []byte, err error) {
 218         if _, err := io.ReadFull(p.r, p.header[:]); err != nil {
 219                 return 0, nil, err
 220         }
 221
 222         pf = payloadFormat(p.header[0])
 223         length := binary.BigEndian.Uint32(p.header[1:])
 224
 225         if length == 0 {
 226                 return pf, nil, nil
 227         }
 228         if length > uint32(maxMsgSize) {
 229                 return 0, nil, Errorf(codes.Internal, "grpc: received message length %d exceeding the max size %d", length, maxMsgSize)
 230         }
 231         // TODO(bradfitz,zhaoq): garbage. reuse buffer after proto decoding instead
 232         // of making it for each message:
 233         msg = make([]byte, int(length))
 234         if _, err := io.ReadFull(p.r, msg); err != nil {
 235                 if err == io.EOF {
 236                         err = io.ErrUnexpectedEOF
 237                 }
 238                 return 0, nil, err
 239         }
 240         return pf, msg, nil
 241 }
 242
 243 // encode serializes msg and prepends the message header. If msg is nil, it
 244 // generates the message header of 0 message length.
 245 func encode(c Codec, msg interface{}, cp Compressor, cbuf *bytes.Buffer, outPayload *stats.OutPayload) ([]byte, error) {
 246         var (
 247                 b      []byte
 248                 length uint
 249         )
 250         if msg != nil {
 251                 var err error
 252                 // TODO(zhaoq): optimize to reduce memory alloc and copying.
 253                 b, err = c.Marshal(msg)
 254                 if err != nil {
 255                         return nil, err
 256                 }
 257                 if outPayload != nil {
 258                         outPayload.Payload = msg
 259                         // TODO truncate large payload.
 260                         outPayload.Data = b
 261                         outPayload.Length = len(b)
 262                 }
 263                 if cp != nil {
 264                         if err := cp.Do(cbuf, b); err != nil {
 265                                 return nil, err
 266                         }
 267                         b = cbuf.Bytes()
 268                 }
 269                 length = uint(len(b))
 270         }
 271         if length > math.MaxUint32 {
 272                 return nil, Errorf(codes.InvalidArgument, "grpc: message too large (%d bytes)", length)
 273         }
 274
 275         const (
 276                 payloadLen = 1
 277                 sizeLen    = 4
 278         )
 279
 280         var buf = make([]byte, payloadLen+sizeLen+len(b))
 281
 282         // Write payload format
 283         if cp == nil {
 284                 buf[0] = byte(compressionNone)
 285         } else {
 286                 buf[0] = byte(compressionMade)
 287         }
 288         // Write length of b into buf
 289         binary.BigEndian.PutUint32(buf[1:], uint32(length))
 290         // Copy encoded msg to buf
 291         copy(buf[5:], b)
 292
 293         if outPayload != nil {
 294                 outPayload.WireLength = len(buf)
 295         }
 296
 297         return buf, nil
 298 }
 299
 300 func checkRecvPayload(pf payloadFormat, recvCompress string, dc Decompressor) error {
 301         switch pf {
 302         case compressionNone:
 303         case compressionMade:
 304                 if dc == nil || recvCompress != dc.Type() {
 305                         return Errorf(codes.Unimplemented, "grpc: Decompressor is not installed for grpc-encoding %q", recvCompress)
 306                 }
 307         default:
 308                 return Errorf(codes.Internal, "grpc: received unexpected payload format %d", pf)
 309         }
 310         return nil
 311 }
 312
 313 func recv(p *parser, c Codec, s *transport.Stream, dc Decompressor, m interface{}, maxMsgSize int, inPayload *stats.InPayload) error {
 314         pf, d, err := p.recvMsg(maxMsgSize)
 315         if err != nil {
 316                 return err
 317         }
 318         if inPayload != nil {
 319                 inPayload.WireLength = len(d)
 320         }
 321         if err := checkRecvPayload(pf, s.RecvCompress(), dc); err != nil {
 322                 return err
 323         }
 324         if pf == compressionMade {
 325                 d, err = dc.Do(bytes.NewReader(d))
 326                 if err != nil {
 327                         return Errorf(codes.Internal, "grpc: failed to decompress the received message %v", err)
 328                 }
 329         }
 330         if len(d) > maxMsgSize {
 331                 // TODO: Revisit the error code. Currently keep it consistent with java
 332                 // implementation.
 333                 return Errorf(codes.Internal, "grpc: received a message of %d bytes exceeding %d limit", len(d), maxMsgSize)
 334         }
 335         if err := c.Unmarshal(d, m); err != nil {
 336                 return Errorf(codes.Internal, "grpc: failed to unmarshal the received message %v", err)
 337         }
 338         if inPayload != nil {
 339                 inPayload.RecvTime = time.Now()
 340                 inPayload.Payload = m
 341                 // TODO truncate large payload.
 342                 inPayload.Data = d
 343                 inPayload.Length = len(d)
 344         }
 345         return nil
 346 }
 347
 348 type rpcInfo struct {
 349         bytesSent     bool
 350         bytesReceived bool
 351 }
 352
 353 type rpcInfoContextKey struct{}
 354
 355 func newContextWithRPCInfo(ctx context.Context) context.Context {
 356         return context.WithValue(ctx, rpcInfoContextKey{}, &rpcInfo{})
 357 }
 358
 359 func rpcInfoFromContext(ctx context.Context) (s *rpcInfo, ok bool) {
 360         s, ok = ctx.Value(rpcInfoContextKey{}).(*rpcInfo)
 361         return
 362 }
 363
 364 func updateRPCInfoInContext(ctx context.Context, s rpcInfo) {
 365         if ss, ok := rpcInfoFromContext(ctx); ok {
 366                 *ss = s
 367         }
 368         return
 369 }
 370
 371 // Code returns the error code for err if it was produced by the rpc system.
 372 // Otherwise, it returns codes.Unknown.
 373 //
 374 // Deprecated; use status.FromError and Code method instead.
 375 func Code(err error) codes.Code {
 376         if s, ok := status.FromError(err); ok {
 377                 return s.Code()
 378         }
 379         return codes.Unknown
 380 }
 381
 382 // ErrorDesc returns the error description of err if it was produced by the rpc system.
 383 // Otherwise, it returns err.Error() or empty string when err is nil.
 384 //
 385 // Deprecated; use status.FromError and Message method instead.
 386 func ErrorDesc(err error) string {
 387         if s, ok := status.FromError(err); ok {
 388                 return s.Message()
 389         }
 390         return err.Error()
 391 }
 392
 393 // Errorf returns an error containing an error code and a description;
 394 // Errorf returns nil if c is OK.
 395 //
 396 // Deprecated; use status.Errorf instead.
 397 func Errorf(c codes.Code, format string, a ...interface{}) error {
 398         return status.Errorf(c, format, a...)
 399 }
 400
 401 // toRPCErr converts an error into an error from the status package.
 402 func toRPCErr(err error) error {
 403         if _, ok := status.FromError(err); ok {
 404                 return err
 405         }
 406         switch e := err.(type) {
 407         case transport.StreamError:
 408                 return status.Error(e.Code, e.Desc)
 409         case transport.ConnectionError:
 410                 return status.Error(codes.Internal, e.Desc)
 411         default:
 412                 switch err {
 413                 case context.DeadlineExceeded:
 414                         return status.Error(codes.DeadlineExceeded, err.Error())
 415                 case context.Canceled:
 416                         return status.Error(codes.Canceled, err.Error())
 417                 case ErrClientConnClosing:
 418                         return status.Error(codes.FailedPrecondition, err.Error())
 419                 }
 420         }
 421         return status.Error(codes.Unknown, err.Error())
 422 }
 423
 424 // convertCode converts a standard Go error into its canonical code. Note that
 425 // this is only used to translate the error returned by the server applications.
 426 func convertCode(err error) codes.Code {
 427         switch err {
 428         case nil:
 429                 return codes.OK
 430         case io.EOF:
 431                 return codes.OutOfRange
 432         case io.ErrClosedPipe, io.ErrNoProgress, io.ErrShortBuffer, io.ErrShortWrite, io.ErrUnexpectedEOF:
 433                 return codes.FailedPrecondition
 434         case os.ErrInvalid:
 435                 return codes.InvalidArgument
 436         case context.Canceled:
 437                 return codes.Canceled
 438         case context.DeadlineExceeded:
 439                 return codes.DeadlineExceeded
 440         }
 441         switch {
 442         case os.IsExist(err):
 443                 return codes.AlreadyExists
 444         case os.IsNotExist(err):
 445                 return codes.NotFound
 446         case os.IsPermission(err):
 447                 return codes.PermissionDenied
 448         }
 449         return codes.Unknown
 450 }
 451
 452 // MethodConfig defines the configuration recommended by the service providers for a
 453 // particular method.
 454 // This is EXPERIMENTAL and subject to change.
 455 type MethodConfig struct {
 456         // WaitForReady indicates whether RPCs sent to this method should wait until
 457         // the connection is ready by default (!failfast). The value specified via the
 458         // gRPC client API will override the value set here.
 459         WaitForReady bool
 460         // Timeout is the default timeout for RPCs sent to this method. The actual
 461         // deadline used will be the minimum of the value specified here and the value
 462         // set by the application via the gRPC client API.  If either one is not set,
 463         // then the other will be used.  If neither is set, then the RPC has no deadline.
 464         Timeout time.Duration
 465         // MaxReqSize is the maximum allowed payload size for an individual request in a
 466         // stream (client->server) in bytes. The size which is measured is the serialized
 467         // payload after per-message compression (but before stream compression) in bytes.
 468         // The actual value used is the minumum of the value specified here and the value set
 469         // by the application via the gRPC client API. If either one is not set, then the other
 470         // will be used.  If neither is set, then the built-in default is used.
 471         // TODO: support this.
 472         MaxReqSize uint32
 473         // MaxRespSize is the maximum allowed payload size for an individual response in a
 474         // stream (server->client) in bytes.
 475         // TODO: support this.
 476         MaxRespSize uint32
 477 }
 478
 479 // ServiceConfig is provided by the service provider and contains parameters for how
 480 // clients that connect to the service should behave.
 481 // This is EXPERIMENTAL and subject to change.
 482 type ServiceConfig struct {
 483         // LB is the load balancer the service providers recommends. The balancer specified
 484         // via grpc.WithBalancer will override this.
 485         LB Balancer
 486         // Methods contains a map for the methods in this service.
 487         Methods map[string]MethodConfig
 488 }
 489
 490 // SupportPackageIsVersion4 is referenced from generated protocol buffer files
 491 // to assert that that code is compatible with this version of the grpc package.
 492 //
 493 // This constant may be renamed in the future if a change in the generated code
 494 // requires a synchronised update of grpc-go and protoc-gen-go. This constant
 495 // should not be referenced from any other code.
 496 const SupportPackageIsVersion4 = true
 497
 498 // Version is the current grpc version.
 499 const Version = "1.3.0"
 500
 501 const grpcUA = "grpc-go/" + Version