cubefs

1
/*
2
 *
3
 * Copyright 2014 gRPC authors.
4
 *
5
 * Licensed under the Apache License, Version 2.0 (the "License");
6
 * you may not use this file except in compliance with the License.
7
 * You may obtain a copy of the License at
8
 *
9
 *     http://www.apache.org/licenses/LICENSE-2.0
10
 *
11
 * Unless required by applicable law or agreed to in writing, software
12
 * distributed under the License is distributed on an "AS IS" BASIS,
13
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14
 * See the License for the specific language governing permissions and
15
 * limitations under the License.
16
 *
17
 */
18

19
package grpc
20

21
import (
22
	"bytes"
23
	"compress/gzip"
24
	"context"
25
	"encoding/binary"
26
	"fmt"
27
	"io"
28
	"io/ioutil"
29
	"math"
30
	"strings"
31
	"sync"
32
	"time"
33

34
	"google.golang.org/grpc/codes"
35
	"google.golang.org/grpc/credentials"
36
	"google.golang.org/grpc/encoding"
37
	"google.golang.org/grpc/encoding/proto"
38
	"google.golang.org/grpc/internal/transport"
39
	"google.golang.org/grpc/metadata"
40
	"google.golang.org/grpc/peer"
41
	"google.golang.org/grpc/stats"
42
	"google.golang.org/grpc/status"
43
)
44

45
// Compressor defines the interface gRPC uses to compress a message.
46
//
47
// Deprecated: use package encoding.
48
type Compressor interface {
49
	// Do compresses p into w.
50
	Do(w io.Writer, p []byte) error
51
	// Type returns the compression algorithm the Compressor uses.
52
	Type() string
53
}
54

55
type gzipCompressor struct {
56
	pool sync.Pool
57
}
58

59
// NewGZIPCompressor creates a Compressor based on GZIP.
60
//
61
// Deprecated: use package encoding/gzip.
62
func NewGZIPCompressor() Compressor {
63
	c, _ := NewGZIPCompressorWithLevel(gzip.DefaultCompression)
64
	return c
65
}
66

67
// NewGZIPCompressorWithLevel is like NewGZIPCompressor but specifies the gzip compression level instead
68
// of assuming DefaultCompression.
69
//
70
// The error returned will be nil if the level is valid.
71
//
72
// Deprecated: use package encoding/gzip.
73
func NewGZIPCompressorWithLevel(level int) (Compressor, error) {
74
	if level < gzip.DefaultCompression || level > gzip.BestCompression {
75
		return nil, fmt.Errorf("grpc: invalid compression level: %d", level)
76
	}
77
	return &gzipCompressor{
78
		pool: sync.Pool{
79
			New: func() interface{} {
80
				w, err := gzip.NewWriterLevel(ioutil.Discard, level)
81
				if err != nil {
82
					panic(err)
83
				}
84
				return w
85
			},
86
		},
87
	}, nil
88
}
89

90
func (c *gzipCompressor) Do(w io.Writer, p []byte) error {
91
	z := c.pool.Get().(*gzip.Writer)
92
	defer c.pool.Put(z)
93
	z.Reset(w)
94
	if _, err := z.Write(p); err != nil {
95
		return err
96
	}
97
	return z.Close()
98
}
99

100
func (c *gzipCompressor) Type() string {
101
	return "gzip"
102
}
103

104
// Decompressor defines the interface gRPC uses to decompress a message.
105
//
106
// Deprecated: use package encoding.
107
type Decompressor interface {
108
	// Do reads the data from r and uncompress them.
109
	Do(r io.Reader) ([]byte, error)
110
	// Type returns the compression algorithm the Decompressor uses.
111
	Type() string
112
}
113

114
type gzipDecompressor struct {
115
	pool sync.Pool
116
}
117

118
// NewGZIPDecompressor creates a Decompressor based on GZIP.
119
//
120
// Deprecated: use package encoding/gzip.
121
func NewGZIPDecompressor() Decompressor {
122
	return &gzipDecompressor{}
123
}
124

125
func (d *gzipDecompressor) Do(r io.Reader) ([]byte, error) {
126
	var z *gzip.Reader
127
	switch maybeZ := d.pool.Get().(type) {
128
	case nil:
129
		newZ, err := gzip.NewReader(r)
130
		if err != nil {
131
			return nil, err
132
		}
133
		z = newZ
134
	case *gzip.Reader:
135
		z = maybeZ
136
		if err := z.Reset(r); err != nil {
137
			d.pool.Put(z)
138
			return nil, err
139
		}
140
	}
141

142
	defer func() {
143
		z.Close()
144
		d.pool.Put(z)
145
	}()
146
	return ioutil.ReadAll(z)
147
}
148

149
func (d *gzipDecompressor) Type() string {
150
	return "gzip"
151
}
152

153
// callInfo contains all related configuration and information about an RPC.
154
type callInfo struct {
155
	compressorType        string
156
	failFast              bool
157
	maxReceiveMessageSize *int
158
	maxSendMessageSize    *int
159
	creds                 credentials.PerRPCCredentials
160
	contentSubtype        string
161
	codec                 baseCodec
162
	maxRetryRPCBufferSize int
163
}
164

165
func defaultCallInfo() *callInfo {
166
	return &callInfo{
167
		failFast:              true,
168
		maxRetryRPCBufferSize: 256 * 1024, // 256KB
169
	}
170
}
171

172
// CallOption configures a Call before it starts or extracts information from
173
// a Call after it completes.
174
type CallOption interface {
175
	// before is called before the call is sent to any server.  If before
176
	// returns a non-nil error, the RPC fails with that error.
177
	before(*callInfo) error
178

179
	// after is called after the call has completed.  after cannot return an
180
	// error, so any failures should be reported via output parameters.
181
	after(*callInfo, *csAttempt)
182
}
183

184
// EmptyCallOption does not alter the Call configuration.
185
// It can be embedded in another structure to carry satellite data for use
186
// by interceptors.
187
type EmptyCallOption struct{}
188

189
func (EmptyCallOption) before(*callInfo) error      { return nil }
190
func (EmptyCallOption) after(*callInfo, *csAttempt) {}
191

192
// Header returns a CallOptions that retrieves the header metadata
193
// for a unary RPC.
194
func Header(md *metadata.MD) CallOption {
195
	return HeaderCallOption{HeaderAddr: md}
196
}
197

198
// HeaderCallOption is a CallOption for collecting response header metadata.
199
// The metadata field will be populated *after* the RPC completes.
200
//
201
// Experimental
202
//
203
// Notice: This type is EXPERIMENTAL and may be changed or removed in a
204
// later release.
205
type HeaderCallOption struct {
206
	HeaderAddr *metadata.MD
207
}
208

209
func (o HeaderCallOption) before(c *callInfo) error { return nil }
210
func (o HeaderCallOption) after(c *callInfo, attempt *csAttempt) {
211
	*o.HeaderAddr, _ = attempt.s.Header()
212
}
213

214
// Trailer returns a CallOptions that retrieves the trailer metadata
215
// for a unary RPC.
216
func Trailer(md *metadata.MD) CallOption {
217
	return TrailerCallOption{TrailerAddr: md}
218
}
219

220
// TrailerCallOption is a CallOption for collecting response trailer metadata.
221
// The metadata field will be populated *after* the RPC completes.
222
//
223
// Experimental
224
//
225
// Notice: This type is EXPERIMENTAL and may be changed or removed in a
226
// later release.
227
type TrailerCallOption struct {
228
	TrailerAddr *metadata.MD
229
}
230

231
func (o TrailerCallOption) before(c *callInfo) error { return nil }
232
func (o TrailerCallOption) after(c *callInfo, attempt *csAttempt) {
233
	*o.TrailerAddr = attempt.s.Trailer()
234
}
235

236
// Peer returns a CallOption that retrieves peer information for a unary RPC.
237
// The peer field will be populated *after* the RPC completes.
238
func Peer(p *peer.Peer) CallOption {
239
	return PeerCallOption{PeerAddr: p}
240
}
241

242
// PeerCallOption is a CallOption for collecting the identity of the remote
243
// peer. The peer field will be populated *after* the RPC completes.
244
//
245
// Experimental
246
//
247
// Notice: This type is EXPERIMENTAL and may be changed or removed in a
248
// later release.
249
type PeerCallOption struct {
250
	PeerAddr *peer.Peer
251
}
252

253
func (o PeerCallOption) before(c *callInfo) error { return nil }
254
func (o PeerCallOption) after(c *callInfo, attempt *csAttempt) {
255
	if x, ok := peer.FromContext(attempt.s.Context()); ok {
256
		*o.PeerAddr = *x
257
	}
258
}
259

260
// WaitForReady configures the action to take when an RPC is attempted on broken
261
// connections or unreachable servers. If waitForReady is false, the RPC will fail
262
// immediately. Otherwise, the RPC client will block the call until a
263
// connection is available (or the call is canceled or times out) and will
264
// retry the call if it fails due to a transient error.  gRPC will not retry if
265
// data was written to the wire unless the server indicates it did not process
266
// the data.  Please refer to
267
// https://github.com/grpc/grpc/blob/master/doc/wait-for-ready.md.
268
//
269
// By default, RPCs don't "wait for ready".
270
func WaitForReady(waitForReady bool) CallOption {
271
	return FailFastCallOption{FailFast: !waitForReady}
272
}
273

274
// FailFast is the opposite of WaitForReady.
275
//
276
// Deprecated: use WaitForReady.
277
func FailFast(failFast bool) CallOption {
278
	return FailFastCallOption{FailFast: failFast}
279
}
280

281
// FailFastCallOption is a CallOption for indicating whether an RPC should fail
282
// fast or not.
283
//
284
// Experimental
285
//
286
// Notice: This type is EXPERIMENTAL and may be changed or removed in a
287
// later release.
288
type FailFastCallOption struct {
289
	FailFast bool
290
}
291

292
func (o FailFastCallOption) before(c *callInfo) error {
293
	c.failFast = o.FailFast
294
	return nil
295
}
296
func (o FailFastCallOption) after(c *callInfo, attempt *csAttempt) {}
297

298
// MaxCallRecvMsgSize returns a CallOption which sets the maximum message size
299
// in bytes the client can receive.
300
func MaxCallRecvMsgSize(bytes int) CallOption {
301
	return MaxRecvMsgSizeCallOption{MaxRecvMsgSize: bytes}
302
}
303

304
// MaxRecvMsgSizeCallOption is a CallOption that indicates the maximum message
305
// size in bytes the client can receive.
306
//
307
// Experimental
308
//
309
// Notice: This type is EXPERIMENTAL and may be changed or removed in a
310
// later release.
311
type MaxRecvMsgSizeCallOption struct {
312
	MaxRecvMsgSize int
313
}
314

315
func (o MaxRecvMsgSizeCallOption) before(c *callInfo) error {
316
	c.maxReceiveMessageSize = &o.MaxRecvMsgSize
317
	return nil
318
}
319
func (o MaxRecvMsgSizeCallOption) after(c *callInfo, attempt *csAttempt) {}
320

321
// MaxCallSendMsgSize returns a CallOption which sets the maximum message size
322
// in bytes the client can send.
323
func MaxCallSendMsgSize(bytes int) CallOption {
324
	return MaxSendMsgSizeCallOption{MaxSendMsgSize: bytes}
325
}
326

327
// MaxSendMsgSizeCallOption is a CallOption that indicates the maximum message
328
// size in bytes the client can send.
329
//
330
// Experimental
331
//
332
// Notice: This type is EXPERIMENTAL and may be changed or removed in a
333
// later release.
334
type MaxSendMsgSizeCallOption struct {
335
	MaxSendMsgSize int
336
}
337

338
func (o MaxSendMsgSizeCallOption) before(c *callInfo) error {
339
	c.maxSendMessageSize = &o.MaxSendMsgSize
340
	return nil
341
}
342
func (o MaxSendMsgSizeCallOption) after(c *callInfo, attempt *csAttempt) {}
343

344
// PerRPCCredentials returns a CallOption that sets credentials.PerRPCCredentials
345
// for a call.
346
func PerRPCCredentials(creds credentials.PerRPCCredentials) CallOption {
347
	return PerRPCCredsCallOption{Creds: creds}
348
}
349

350
// PerRPCCredsCallOption is a CallOption that indicates the per-RPC
351
// credentials to use for the call.
352
//
353
// Experimental
354
//
355
// Notice: This type is EXPERIMENTAL and may be changed or removed in a
356
// later release.
357
type PerRPCCredsCallOption struct {
358
	Creds credentials.PerRPCCredentials
359
}
360

361
func (o PerRPCCredsCallOption) before(c *callInfo) error {
362
	c.creds = o.Creds
363
	return nil
364
}
365
func (o PerRPCCredsCallOption) after(c *callInfo, attempt *csAttempt) {}
366

367
// UseCompressor returns a CallOption which sets the compressor used when
368
// sending the request.  If WithCompressor is also set, UseCompressor has
369
// higher priority.
370
//
371
// Experimental
372
//
373
// Notice: This API is EXPERIMENTAL and may be changed or removed in a
374
// later release.
375
func UseCompressor(name string) CallOption {
376
	return CompressorCallOption{CompressorType: name}
377
}
378

379
// CompressorCallOption is a CallOption that indicates the compressor to use.
380
//
381
// Experimental
382
//
383
// Notice: This type is EXPERIMENTAL and may be changed or removed in a
384
// later release.
385
type CompressorCallOption struct {
386
	CompressorType string
387
}
388

389
func (o CompressorCallOption) before(c *callInfo) error {
390
	c.compressorType = o.CompressorType
391
	return nil
392
}
393
func (o CompressorCallOption) after(c *callInfo, attempt *csAttempt) {}
394

395
// CallContentSubtype returns a CallOption that will set the content-subtype
396
// for a call. For example, if content-subtype is "json", the Content-Type over
397
// the wire will be "application/grpc+json". The content-subtype is converted
398
// to lowercase before being included in Content-Type. See Content-Type on
399
// https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-HTTP2.md#requests for
400
// more details.
401
//
402
// If ForceCodec is not also used, the content-subtype will be used to look up
403
// the Codec to use in the registry controlled by RegisterCodec. See the
404
// documentation on RegisterCodec for details on registration. The lookup of
405
// content-subtype is case-insensitive. If no such Codec is found, the call
406
// will result in an error with code codes.Internal.
407
//
408
// If ForceCodec is also used, that Codec will be used for all request and
409
// response messages, with the content-subtype set to the given contentSubtype
410
// here for requests.
411
func CallContentSubtype(contentSubtype string) CallOption {
412
	return ContentSubtypeCallOption{ContentSubtype: strings.ToLower(contentSubtype)}
413
}
414

415
// ContentSubtypeCallOption is a CallOption that indicates the content-subtype
416
// used for marshaling messages.
417
//
418
// Experimental
419
//
420
// Notice: This type is EXPERIMENTAL and may be changed or removed in a
421
// later release.
422
type ContentSubtypeCallOption struct {
423
	ContentSubtype string
424
}
425

426
func (o ContentSubtypeCallOption) before(c *callInfo) error {
427
	c.contentSubtype = o.ContentSubtype
428
	return nil
429
}
430
func (o ContentSubtypeCallOption) after(c *callInfo, attempt *csAttempt) {}
431

432
// ForceCodec returns a CallOption that will set the given Codec to be
433
// used for all request and response messages for a call. The result of calling
434
// String() will be used as the content-subtype in a case-insensitive manner.
435
//
436
// See Content-Type on
437
// https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-HTTP2.md#requests for
438
// more details. Also see the documentation on RegisterCodec and
439
// CallContentSubtype for more details on the interaction between Codec and
440
// content-subtype.
441
//
442
// This function is provided for advanced users; prefer to use only
443
// CallContentSubtype to select a registered codec instead.
444
//
445
// Experimental
446
//
447
// Notice: This API is EXPERIMENTAL and may be changed or removed in a
448
// later release.
449
func ForceCodec(codec encoding.Codec) CallOption {
450
	return ForceCodecCallOption{Codec: codec}
451
}
452

453
// ForceCodecCallOption is a CallOption that indicates the codec used for
454
// marshaling messages.
455
//
456
// Experimental
457
//
458
// Notice: This type is EXPERIMENTAL and may be changed or removed in a
459
// later release.
460
type ForceCodecCallOption struct {
461
	Codec encoding.Codec
462
}
463

464
func (o ForceCodecCallOption) before(c *callInfo) error {
465
	c.codec = o.Codec
466
	return nil
467
}
468
func (o ForceCodecCallOption) after(c *callInfo, attempt *csAttempt) {}
469

470
// CallCustomCodec behaves like ForceCodec, but accepts a grpc.Codec instead of
471
// an encoding.Codec.
472
//
473
// Deprecated: use ForceCodec instead.
474
func CallCustomCodec(codec Codec) CallOption {
475
	return CustomCodecCallOption{Codec: codec}
476
}
477

478
// CustomCodecCallOption is a CallOption that indicates the codec used for
479
// marshaling messages.
480
//
481
// Experimental
482
//
483
// Notice: This type is EXPERIMENTAL and may be changed or removed in a
484
// later release.
485
type CustomCodecCallOption struct {
486
	Codec Codec
487
}
488

489
func (o CustomCodecCallOption) before(c *callInfo) error {
490
	c.codec = o.Codec
491
	return nil
492
}
493
func (o CustomCodecCallOption) after(c *callInfo, attempt *csAttempt) {}
494

495
// MaxRetryRPCBufferSize returns a CallOption that limits the amount of memory
496
// used for buffering this RPC's requests for retry purposes.
497
//
498
// Experimental
499
//
500
// Notice: This API is EXPERIMENTAL and may be changed or removed in a
501
// later release.
502
func MaxRetryRPCBufferSize(bytes int) CallOption {
503
	return MaxRetryRPCBufferSizeCallOption{bytes}
504
}
505

506
// MaxRetryRPCBufferSizeCallOption is a CallOption indicating the amount of
507
// memory to be used for caching this RPC for retry purposes.
508
//
509
// Experimental
510
//
511
// Notice: This type is EXPERIMENTAL and may be changed or removed in a
512
// later release.
513
type MaxRetryRPCBufferSizeCallOption struct {
514
	MaxRetryRPCBufferSize int
515
}
516

517
func (o MaxRetryRPCBufferSizeCallOption) before(c *callInfo) error {
518
	c.maxRetryRPCBufferSize = o.MaxRetryRPCBufferSize
519
	return nil
520
}
521
func (o MaxRetryRPCBufferSizeCallOption) after(c *callInfo, attempt *csAttempt) {}
522

523
// The format of the payload: compressed or not?
524
type payloadFormat uint8
525

526
const (
527
	compressionNone payloadFormat = 0 // no compression
528
	compressionMade payloadFormat = 1 // compressed
529
)
530

531
// parser reads complete gRPC messages from the underlying reader.
532
type parser struct {
533
	// r is the underlying reader.
534
	// See the comment on recvMsg for the permissible
535
	// error types.
536
	r io.Reader
537

538
	// The header of a gRPC message. Find more detail at
539
	// https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-HTTP2.md
540
	header [5]byte
541
}
542

543
// recvMsg reads a complete gRPC message from the stream.
544
//
545
// It returns the message and its payload (compression/encoding)
546
// format. The caller owns the returned msg memory.
547
//
548
// If there is an error, possible values are:
549
//   * io.EOF, when no messages remain
550
//   * io.ErrUnexpectedEOF
551
//   * of type transport.ConnectionError
552
//   * an error from the status package
553
// No other error values or types must be returned, which also means
554
// that the underlying io.Reader must not return an incompatible
555
// error.
556
func (p *parser) recvMsg(maxReceiveMessageSize int) (pf payloadFormat, msg []byte, err error) {
557
	if _, err := p.r.Read(p.header[:]); err != nil {
558
		return 0, nil, err
559
	}
560

561
	pf = payloadFormat(p.header[0])
562
	length := binary.BigEndian.Uint32(p.header[1:])
563

564
	if length == 0 {
565
		return pf, nil, nil
566
	}
567
	if int64(length) > int64(maxInt) {
568
		return 0, nil, status.Errorf(codes.ResourceExhausted, "grpc: received message larger than max length allowed on current machine (%d vs. %d)", length, maxInt)
569
	}
570
	if int(length) > maxReceiveMessageSize {
571
		return 0, nil, status.Errorf(codes.ResourceExhausted, "grpc: received message larger than max (%d vs. %d)", length, maxReceiveMessageSize)
572
	}
573
	// TODO(bradfitz,zhaoq): garbage. reuse buffer after proto decoding instead
574
	// of making it for each message:
575
	msg = make([]byte, int(length))
576
	if _, err := p.r.Read(msg); err != nil {
577
		if err == io.EOF {
578
			err = io.ErrUnexpectedEOF
579
		}
580
		return 0, nil, err
581
	}
582
	return pf, msg, nil
583
}
584

585
// encode serializes msg and returns a buffer containing the message, or an
586
// error if it is too large to be transmitted by grpc.  If msg is nil, it
587
// generates an empty message.
588
func encode(c baseCodec, msg interface{}) ([]byte, error) {
589
	if msg == nil { // NOTE: typed nils will not be caught by this check
590
		return nil, nil
591
	}
592
	b, err := c.Marshal(msg)
593
	if err != nil {
594
		return nil, status.Errorf(codes.Internal, "grpc: error while marshaling: %v", err.Error())
595
	}
596
	if uint(len(b)) > math.MaxUint32 {
597
		return nil, status.Errorf(codes.ResourceExhausted, "grpc: message too large (%d bytes)", len(b))
598
	}
599
	return b, nil
600
}
601

602
// compress returns the input bytes compressed by compressor or cp.  If both
603
// compressors are nil, returns nil.
604
//
605
// TODO(dfawley): eliminate cp parameter by wrapping Compressor in an encoding.Compressor.
606
func compress(in []byte, cp Compressor, compressor encoding.Compressor) ([]byte, error) {
607
	if compressor == nil && cp == nil {
608
		return nil, nil
609
	}
610
	wrapErr := func(err error) error {
611
		return status.Errorf(codes.Internal, "grpc: error while compressing: %v", err.Error())
612
	}
613
	cbuf := &bytes.Buffer{}
614
	if compressor != nil {
615
		z, err := compressor.Compress(cbuf)
616
		if err != nil {
617
			return nil, wrapErr(err)
618
		}
619
		if _, err := z.Write(in); err != nil {
620
			return nil, wrapErr(err)
621
		}
622
		if err := z.Close(); err != nil {
623
			return nil, wrapErr(err)
624
		}
625
	} else {
626
		if err := cp.Do(cbuf, in); err != nil {
627
			return nil, wrapErr(err)
628
		}
629
	}
630
	return cbuf.Bytes(), nil
631
}
632

633
const (
634
	payloadLen = 1
635
	sizeLen    = 4
636
	headerLen  = payloadLen + sizeLen
637
)
638

639
// msgHeader returns a 5-byte header for the message being transmitted and the
640
// payload, which is compData if non-nil or data otherwise.
641
func msgHeader(data, compData []byte) (hdr []byte, payload []byte) {
642
	hdr = make([]byte, headerLen)
643
	if compData != nil {
644
		hdr[0] = byte(compressionMade)
645
		data = compData
646
	} else {
647
		hdr[0] = byte(compressionNone)
648
	}
649

650
	// Write length of payload into buf
651
	binary.BigEndian.PutUint32(hdr[payloadLen:], uint32(len(data)))
652
	return hdr, data
653
}
654

655
func outPayload(client bool, msg interface{}, data, payload []byte, t time.Time) *stats.OutPayload {
656
	return &stats.OutPayload{
657
		Client:     client,
658
		Payload:    msg,
659
		Data:       data,
660
		Length:     len(data),
661
		WireLength: len(payload) + headerLen,
662
		SentTime:   t,
663
	}
664
}
665

666
func checkRecvPayload(pf payloadFormat, recvCompress string, haveCompressor bool) *status.Status {
667
	switch pf {
668
	case compressionNone:
669
	case compressionMade:
670
		if recvCompress == "" || recvCompress == encoding.Identity {
671
			return status.New(codes.Internal, "grpc: compressed flag set with identity or empty encoding")
672
		}
673
		if !haveCompressor {
674
			return status.Newf(codes.Unimplemented, "grpc: Decompressor is not installed for grpc-encoding %q", recvCompress)
675
		}
676
	default:
677
		return status.Newf(codes.Internal, "grpc: received unexpected payload format %d", pf)
678
	}
679
	return nil
680
}
681

682
type payloadInfo struct {
683
	wireLength        int // The compressed length got from wire.
684
	uncompressedBytes []byte
685
}
686

687
func recvAndDecompress(p *parser, s *transport.Stream, dc Decompressor, maxReceiveMessageSize int, payInfo *payloadInfo, compressor encoding.Compressor) ([]byte, error) {
688
	pf, d, err := p.recvMsg(maxReceiveMessageSize)
689
	if err != nil {
690
		return nil, err
691
	}
692
	if payInfo != nil {
693
		payInfo.wireLength = len(d)
694
	}
695

696
	if st := checkRecvPayload(pf, s.RecvCompress(), compressor != nil || dc != nil); st != nil {
697
		return nil, st.Err()
698
	}
699

700
	var size int
701
	if pf == compressionMade {
702
		// To match legacy behavior, if the decompressor is set by WithDecompressor or RPCDecompressor,
703
		// use this decompressor as the default.
704
		if dc != nil {
705
			d, err = dc.Do(bytes.NewReader(d))
706
			size = len(d)
707
		} else {
708
			d, size, err = decompress(compressor, d, maxReceiveMessageSize)
709
		}
710
		if err != nil {
711
			return nil, status.Errorf(codes.Internal, "grpc: failed to decompress the received message %v", err)
712
		}
713
	} else {
714
		size = len(d)
715
	}
716
	if size > maxReceiveMessageSize {
717
		// TODO: Revisit the error code. Currently keep it consistent with java
718
		// implementation.
719
		return nil, status.Errorf(codes.ResourceExhausted, "grpc: received message larger than max (%d vs. %d)", size, maxReceiveMessageSize)
720
	}
721
	return d, nil
722
}
723

724
// Using compressor, decompress d, returning data and size.
725
// Optionally, if data will be over maxReceiveMessageSize, just return the size.
726
func decompress(compressor encoding.Compressor, d []byte, maxReceiveMessageSize int) ([]byte, int, error) {
727
	dcReader, err := compressor.Decompress(bytes.NewReader(d))
728
	if err != nil {
729
		return nil, 0, err
730
	}
731
	if sizer, ok := compressor.(interface {
732
		DecompressedSize(compressedBytes []byte) int
733
	}); ok {
734
		if size := sizer.DecompressedSize(d); size >= 0 {
735
			if size > maxReceiveMessageSize {
736
				return nil, size, nil
737
			}
738
			// size is used as an estimate to size the buffer, but we
739
			// will read more data if available.
740
			// +MinRead so ReadFrom will not reallocate if size is correct.
741
			buf := bytes.NewBuffer(make([]byte, 0, size+bytes.MinRead))
742
			bytesRead, err := buf.ReadFrom(io.LimitReader(dcReader, int64(maxReceiveMessageSize)+1))
743
			return buf.Bytes(), int(bytesRead), err
744
		}
745
	}
746
	// Read from LimitReader with limit max+1. So if the underlying
747
	// reader is over limit, the result will be bigger than max.
748
	d, err = ioutil.ReadAll(io.LimitReader(dcReader, int64(maxReceiveMessageSize)+1))
749
	return d, len(d), err
750
}
751

752
// For the two compressor parameters, both should not be set, but if they are,
753
// dc takes precedence over compressor.
754
// TODO(dfawley): wrap the old compressor/decompressor using the new API?
755
func recv(p *parser, c baseCodec, s *transport.Stream, dc Decompressor, m interface{}, maxReceiveMessageSize int, payInfo *payloadInfo, compressor encoding.Compressor) error {
756
	d, err := recvAndDecompress(p, s, dc, maxReceiveMessageSize, payInfo, compressor)
757
	if err != nil {
758
		return err
759
	}
760
	if err := c.Unmarshal(d, m); err != nil {
761
		return status.Errorf(codes.Internal, "grpc: failed to unmarshal the received message %v", err)
762
	}
763
	if payInfo != nil {
764
		payInfo.uncompressedBytes = d
765
	}
766
	return nil
767
}
768

769
// Information about RPC
770
type rpcInfo struct {
771
	failfast      bool
772
	preloaderInfo *compressorInfo
773
}
774

775
// Information about Preloader
776
// Responsible for storing codec, and compressors
777
// If stream (s) has  context s.Context which stores rpcInfo that has non nil
778
// pointers to codec, and compressors, then we can use preparedMsg for Async message prep
779
// and reuse marshalled bytes
780
type compressorInfo struct {
781
	codec baseCodec
782
	cp    Compressor
783
	comp  encoding.Compressor
784
}
785

786
type rpcInfoContextKey struct{}
787

788
func newContextWithRPCInfo(ctx context.Context, failfast bool, codec baseCodec, cp Compressor, comp encoding.Compressor) context.Context {
789
	return context.WithValue(ctx, rpcInfoContextKey{}, &rpcInfo{
790
		failfast: failfast,
791
		preloaderInfo: &compressorInfo{
792
			codec: codec,
793
			cp:    cp,
794
			comp:  comp,
795
		},
796
	})
797
}
798

799
func rpcInfoFromContext(ctx context.Context) (s *rpcInfo, ok bool) {
800
	s, ok = ctx.Value(rpcInfoContextKey{}).(*rpcInfo)
801
	return
802
}
803

804
// Code returns the error code for err if it was produced by the rpc system.
805
// Otherwise, it returns codes.Unknown.
806
//
807
// Deprecated: use status.Code instead.
808
func Code(err error) codes.Code {
809
	return status.Code(err)
810
}
811

812
// ErrorDesc returns the error description of err if it was produced by the rpc system.
813
// Otherwise, it returns err.Error() or empty string when err is nil.
814
//
815
// Deprecated: use status.Convert and Message method instead.
816
func ErrorDesc(err error) string {
817
	return status.Convert(err).Message()
818
}
819

820
// Errorf returns an error containing an error code and a description;
821
// Errorf returns nil if c is OK.
822
//
823
// Deprecated: use status.Errorf instead.
824
func Errorf(c codes.Code, format string, a ...interface{}) error {
825
	return status.Errorf(c, format, a...)
826
}
827

828
// toRPCErr converts an error into an error from the status package.
829
func toRPCErr(err error) error {
830
	if err == nil || err == io.EOF {
831
		return err
832
	}
833
	if err == io.ErrUnexpectedEOF {
834
		return status.Error(codes.Internal, err.Error())
835
	}
836
	if _, ok := status.FromError(err); ok {
837
		return err
838
	}
839
	switch e := err.(type) {
840
	case transport.ConnectionError:
841
		return status.Error(codes.Unavailable, e.Desc)
842
	default:
843
		switch err {
844
		case context.DeadlineExceeded:
845
			return status.Error(codes.DeadlineExceeded, err.Error())
846
		case context.Canceled:
847
			return status.Error(codes.Canceled, err.Error())
848
		}
849
	}
850
	return status.Error(codes.Unknown, err.Error())
851
}
852

853
// setCallInfoCodec should only be called after CallOptions have been applied.
854
func setCallInfoCodec(c *callInfo) error {
855
	if c.codec != nil {
856
		// codec was already set by a CallOption; use it.
857
		return nil
858
	}
859

860
	if c.contentSubtype == "" {
861
		// No codec specified in CallOptions; use proto by default.
862
		c.codec = encoding.GetCodec(proto.Name)
863
		return nil
864
	}
865

866
	// c.contentSubtype is already lowercased in CallContentSubtype
867
	c.codec = encoding.GetCodec(c.contentSubtype)
868
	if c.codec == nil {
869
		return status.Errorf(codes.Internal, "no codec registered for content-subtype %s", c.contentSubtype)
870
	}
871
	return nil
872
}
873

874
// channelzData is used to store channelz related data for ClientConn, addrConn and Server.
875
// These fields cannot be embedded in the original structs (e.g. ClientConn), since to do atomic
876
// operation on int64 variable on 32-bit machine, user is responsible to enforce memory alignment.
877
// Here, by grouping those int64 fields inside a struct, we are enforcing the alignment.
878
type channelzData struct {
879
	callsStarted   int64
880
	callsFailed    int64
881
	callsSucceeded int64
882
	// lastCallStartedTime stores the timestamp that last call starts. It is of int64 type instead of
883
	// time.Time since it's more costly to atomically update time.Time variable than int64 variable.
884
	lastCallStartedTime int64
885
}
886

887
// The SupportPackageIsVersion variables are referenced from generated protocol
888
// buffer files to ensure compatibility with the gRPC version used.  The latest
889
// support package version is 7.
890
//
891
// Older versions are kept for compatibility. They may be removed if
892
// compatibility cannot be maintained.
893
//
894
// These constants should not be referenced from any other code.
895
const (
896
	SupportPackageIsVersion3 = true
897
	SupportPackageIsVersion4 = true
898
	SupportPackageIsVersion5 = true
899
	SupportPackageIsVersion6 = true
900
	SupportPackageIsVersion7 = true
901
)
902

903
const grpcUA = "grpc-go/" + Version
904