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
	"context"
23
	"errors"
24
	"io"
25
	"math"
26
	"strconv"
27
	"sync"
28
	"time"
29

30
	"golang.org/x/net/trace"
31
	"google.golang.org/grpc/balancer"
32
	"google.golang.org/grpc/codes"
33
	"google.golang.org/grpc/encoding"
34
	"google.golang.org/grpc/internal/balancerload"
35
	"google.golang.org/grpc/internal/binarylog"
36
	"google.golang.org/grpc/internal/channelz"
37
	"google.golang.org/grpc/internal/grpcrand"
38
	"google.golang.org/grpc/internal/grpcutil"
39
	iresolver "google.golang.org/grpc/internal/resolver"
40
	"google.golang.org/grpc/internal/serviceconfig"
41
	"google.golang.org/grpc/internal/transport"
42
	"google.golang.org/grpc/metadata"
43
	"google.golang.org/grpc/peer"
44
	"google.golang.org/grpc/stats"
45
	"google.golang.org/grpc/status"
46
)
47

48
// StreamHandler defines the handler called by gRPC server to complete the
49
// execution of a streaming RPC. If a StreamHandler returns an error, it
50
// should be produced by the status package, or else gRPC will use
51
// codes.Unknown as the status code and err.Error() as the status message
52
// of the RPC.
53
type StreamHandler func(srv interface{}, stream ServerStream) error
54

55
// StreamDesc represents a streaming RPC service's method specification.
56
type StreamDesc struct {
57
	StreamName string
58
	Handler    StreamHandler
59

60
	// At least one of these is true.
61
	ServerStreams bool
62
	ClientStreams bool
63
}
64

65
// Stream defines the common interface a client or server stream has to satisfy.
66
//
67
// Deprecated: See ClientStream and ServerStream documentation instead.
68
type Stream interface {
69
	// Deprecated: See ClientStream and ServerStream documentation instead.
70
	Context() context.Context
71
	// Deprecated: See ClientStream and ServerStream documentation instead.
72
	SendMsg(m interface{}) error
73
	// Deprecated: See ClientStream and ServerStream documentation instead.
74
	RecvMsg(m interface{}) error
75
}
76

77
// ClientStream defines the client-side behavior of a streaming RPC.
78
//
79
// All errors returned from ClientStream methods are compatible with the
80
// status package.
81
type ClientStream interface {
82
	// Header returns the header metadata received from the server if there
83
	// is any. It blocks if the metadata is not ready to read.
84
	Header() (metadata.MD, error)
85
	// Trailer returns the trailer metadata from the server, if there is any.
86
	// It must only be called after stream.CloseAndRecv has returned, or
87
	// stream.Recv has returned a non-nil error (including io.EOF).
88
	Trailer() metadata.MD
89
	// CloseSend closes the send direction of the stream. It closes the stream
90
	// when non-nil error is met. It is also not safe to call CloseSend
91
	// concurrently with SendMsg.
92
	CloseSend() error
93
	// Context returns the context for this stream.
94
	//
95
	// It should not be called until after Header or RecvMsg has returned. Once
96
	// called, subsequent client-side retries are disabled.
97
	Context() context.Context
98
	// SendMsg is generally called by generated code. On error, SendMsg aborts
99
	// the stream. If the error was generated by the client, the status is
100
	// returned directly; otherwise, io.EOF is returned and the status of
101
	// the stream may be discovered using RecvMsg.
102
	//
103
	// SendMsg blocks until:
104
	//   - There is sufficient flow control to schedule m with the transport, or
105
	//   - The stream is done, or
106
	//   - The stream breaks.
107
	//
108
	// SendMsg does not wait until the message is received by the server. An
109
	// untimely stream closure may result in lost messages. To ensure delivery,
110
	// users should ensure the RPC completed successfully using RecvMsg.
111
	//
112
	// It is safe to have a goroutine calling SendMsg and another goroutine
113
	// calling RecvMsg on the same stream at the same time, but it is not safe
114
	// to call SendMsg on the same stream in different goroutines. It is also
115
	// not safe to call CloseSend concurrently with SendMsg.
116
	SendMsg(m interface{}) error
117
	// RecvMsg blocks until it receives a message into m or the stream is
118
	// done. It returns io.EOF when the stream completes successfully. On
119
	// any other error, the stream is aborted and the error contains the RPC
120
	// status.
121
	//
122
	// It is safe to have a goroutine calling SendMsg and another goroutine
123
	// calling RecvMsg on the same stream at the same time, but it is not
124
	// safe to call RecvMsg on the same stream in different goroutines.
125
	RecvMsg(m interface{}) error
126
}
127

128
// NewStream creates a new Stream for the client side. This is typically
129
// called by generated code. ctx is used for the lifetime of the stream.
130
//
131
// To ensure resources are not leaked due to the stream returned, one of the following
132
// actions must be performed:
133
//
134
//      1. Call Close on the ClientConn.
135
//      2. Cancel the context provided.
136
//      3. Call RecvMsg until a non-nil error is returned. A protobuf-generated
137
//         client-streaming RPC, for instance, might use the helper function
138
//         CloseAndRecv (note that CloseSend does not Recv, therefore is not
139
//         guaranteed to release all resources).
140
//      4. Receive a non-nil, non-io.EOF error from Header or SendMsg.
141
//
142
// If none of the above happen, a goroutine and a context will be leaked, and grpc
143
// will not call the optionally-configured stats handler with a stats.End message.
144
func (cc *ClientConn) NewStream(ctx context.Context, desc *StreamDesc, method string, opts ...CallOption) (ClientStream, error) {
145
	// allow interceptor to see all applicable call options, which means those
146
	// configured as defaults from dial option as well as per-call options
147
	opts = combine(cc.dopts.callOptions, opts)
148

149
	if cc.dopts.streamInt != nil {
150
		return cc.dopts.streamInt(ctx, desc, cc, method, newClientStream, opts...)
151
	}
152
	return newClientStream(ctx, desc, cc, method, opts...)
153
}
154

155
// NewClientStream is a wrapper for ClientConn.NewStream.
156
func NewClientStream(ctx context.Context, desc *StreamDesc, cc *ClientConn, method string, opts ...CallOption) (ClientStream, error) {
157
	return cc.NewStream(ctx, desc, method, opts...)
158
}
159

160
func newClientStream(ctx context.Context, desc *StreamDesc, cc *ClientConn, method string, opts ...CallOption) (_ ClientStream, err error) {
161
	if channelz.IsOn() {
162
		cc.incrCallsStarted()
163
		defer func() {
164
			if err != nil {
165
				cc.incrCallsFailed()
166
			}
167
		}()
168
	}
169
	c := defaultCallInfo()
170
	// Provide an opportunity for the first RPC to see the first service config
171
	// provided by the resolver.
172
	if err := cc.waitForResolvedAddrs(ctx); err != nil {
173
		return nil, err
174
	}
175

176
	var mc serviceconfig.MethodConfig
177
	var onCommit func()
178
	rpcConfig, err := cc.safeConfigSelector.SelectConfig(iresolver.RPCInfo{Context: ctx, Method: method})
179
	if err != nil {
180
		return nil, status.Convert(err).Err()
181
	}
182
	if rpcConfig != nil {
183
		if rpcConfig.Context != nil {
184
			ctx = rpcConfig.Context
185
		}
186
		mc = rpcConfig.MethodConfig
187
		onCommit = rpcConfig.OnCommitted
188
	}
189

190
	if mc.WaitForReady != nil {
191
		c.failFast = !*mc.WaitForReady
192
	}
193

194
	// Possible context leak:
195
	// The cancel function for the child context we create will only be called
196
	// when RecvMsg returns a non-nil error, if the ClientConn is closed, or if
197
	// an error is generated by SendMsg.
198
	// https://github.com/grpc/grpc-go/issues/1818.
199
	var cancel context.CancelFunc
200
	if mc.Timeout != nil && *mc.Timeout >= 0 {
201
		ctx, cancel = context.WithTimeout(ctx, *mc.Timeout)
202
	} else {
203
		ctx, cancel = context.WithCancel(ctx)
204
	}
205
	defer func() {
206
		if err != nil {
207
			cancel()
208
		}
209
	}()
210

211
	for _, o := range opts {
212
		if err := o.before(c); err != nil {
213
			return nil, toRPCErr(err)
214
		}
215
	}
216
	c.maxSendMessageSize = getMaxSize(mc.MaxReqSize, c.maxSendMessageSize, defaultClientMaxSendMessageSize)
217
	c.maxReceiveMessageSize = getMaxSize(mc.MaxRespSize, c.maxReceiveMessageSize, defaultClientMaxReceiveMessageSize)
218
	if err := setCallInfoCodec(c); err != nil {
219
		return nil, err
220
	}
221

222
	callHdr := &transport.CallHdr{
223
		Host:           cc.authority,
224
		Method:         method,
225
		ContentSubtype: c.contentSubtype,
226
	}
227

228
	// Set our outgoing compression according to the UseCompressor CallOption, if
229
	// set.  In that case, also find the compressor from the encoding package.
230
	// Otherwise, use the compressor configured by the WithCompressor DialOption,
231
	// if set.
232
	var cp Compressor
233
	var comp encoding.Compressor
234
	if ct := c.compressorType; ct != "" {
235
		callHdr.SendCompress = ct
236
		if ct != encoding.Identity {
237
			comp = encoding.GetCompressor(ct)
238
			if comp == nil {
239
				return nil, status.Errorf(codes.Internal, "grpc: Compressor is not installed for requested grpc-encoding %q", ct)
240
			}
241
		}
242
	} else if cc.dopts.cp != nil {
243
		callHdr.SendCompress = cc.dopts.cp.Type()
244
		cp = cc.dopts.cp
245
	}
246
	if c.creds != nil {
247
		callHdr.Creds = c.creds
248
	}
249
	var trInfo *traceInfo
250
	if EnableTracing {
251
		trInfo = &traceInfo{
252
			tr: trace.New("grpc.Sent."+methodFamily(method), method),
253
			firstLine: firstLine{
254
				client: true,
255
			},
256
		}
257
		if deadline, ok := ctx.Deadline(); ok {
258
			trInfo.firstLine.deadline = time.Until(deadline)
259
		}
260
		trInfo.tr.LazyLog(&trInfo.firstLine, false)
261
		ctx = trace.NewContext(ctx, trInfo.tr)
262
	}
263
	ctx = newContextWithRPCInfo(ctx, c.failFast, c.codec, cp, comp)
264
	sh := cc.dopts.copts.StatsHandler
265
	var beginTime time.Time
266
	if sh != nil {
267
		ctx = sh.TagRPC(ctx, &stats.RPCTagInfo{FullMethodName: method, FailFast: c.failFast})
268
		beginTime = time.Now()
269
		begin := &stats.Begin{
270
			Client:    true,
271
			BeginTime: beginTime,
272
			FailFast:  c.failFast,
273
		}
274
		sh.HandleRPC(ctx, begin)
275
	}
276

277
	cs := &clientStream{
278
		callHdr:      callHdr,
279
		ctx:          ctx,
280
		methodConfig: &mc,
281
		opts:         opts,
282
		callInfo:     c,
283
		cc:           cc,
284
		desc:         desc,
285
		codec:        c.codec,
286
		cp:           cp,
287
		comp:         comp,
288
		cancel:       cancel,
289
		beginTime:    beginTime,
290
		firstAttempt: true,
291
		onCommit:     onCommit,
292
	}
293
	if !cc.dopts.disableRetry {
294
		cs.retryThrottler = cc.retryThrottler.Load().(*retryThrottler)
295
	}
296
	cs.binlog = binarylog.GetMethodLogger(method)
297

298
	// Only this initial attempt has stats/tracing.
299
	// TODO(dfawley): move to newAttempt when per-attempt stats are implemented.
300
	if err := cs.newAttemptLocked(sh, trInfo); err != nil {
301
		cs.finish(err)
302
		return nil, err
303
	}
304

305
	op := func(a *csAttempt) error { return a.newStream() }
306
	if err := cs.withRetry(op, func() { cs.bufferForRetryLocked(0, op) }); err != nil {
307
		cs.finish(err)
308
		return nil, err
309
	}
310

311
	if cs.binlog != nil {
312
		md, _ := metadata.FromOutgoingContext(ctx)
313
		logEntry := &binarylog.ClientHeader{
314
			OnClientSide: true,
315
			Header:       md,
316
			MethodName:   method,
317
			Authority:    cs.cc.authority,
318
		}
319
		if deadline, ok := ctx.Deadline(); ok {
320
			logEntry.Timeout = time.Until(deadline)
321
			if logEntry.Timeout < 0 {
322
				logEntry.Timeout = 0
323
			}
324
		}
325
		cs.binlog.Log(logEntry)
326
	}
327

328
	if desc != unaryStreamDesc {
329
		// Listen on cc and stream contexts to cleanup when the user closes the
330
		// ClientConn or cancels the stream context.  In all other cases, an error
331
		// should already be injected into the recv buffer by the transport, which
332
		// the client will eventually receive, and then we will cancel the stream's
333
		// context in clientStream.finish.
334
		go func() {
335
			select {
336
			case <-cc.ctx.Done():
337
				cs.finish(ErrClientConnClosing)
338
			case <-ctx.Done():
339
				cs.finish(toRPCErr(ctx.Err()))
340
			}
341
		}()
342
	}
343
	return cs, nil
344
}
345

346
// newAttemptLocked creates a new attempt with a transport.
347
// If it succeeds, then it replaces clientStream's attempt with this new attempt.
348
func (cs *clientStream) newAttemptLocked(sh stats.Handler, trInfo *traceInfo) (retErr error) {
349
	newAttempt := &csAttempt{
350
		cs:           cs,
351
		dc:           cs.cc.dopts.dc,
352
		statsHandler: sh,
353
		trInfo:       trInfo,
354
	}
355
	defer func() {
356
		if retErr != nil {
357
			// This attempt is not set in the clientStream, so it's finish won't
358
			// be called. Call it here for stats and trace in case they are not
359
			// nil.
360
			newAttempt.finish(retErr)
361
		}
362
	}()
363

364
	if err := cs.ctx.Err(); err != nil {
365
		return toRPCErr(err)
366
	}
367

368
	ctx := cs.ctx
369
	if cs.cc.parsedTarget.Scheme == "xds" {
370
		// Add extra metadata (metadata that will be added by transport) to context
371
		// so the balancer can see them.
372
		ctx = grpcutil.WithExtraMetadata(cs.ctx, metadata.Pairs(
373
			"content-type", grpcutil.ContentType(cs.callHdr.ContentSubtype),
374
		))
375
	}
376
	t, done, err := cs.cc.getTransport(ctx, cs.callInfo.failFast, cs.callHdr.Method)
377
	if err != nil {
378
		return err
379
	}
380
	if trInfo != nil {
381
		trInfo.firstLine.SetRemoteAddr(t.RemoteAddr())
382
	}
383
	newAttempt.t = t
384
	newAttempt.done = done
385
	cs.attempt = newAttempt
386
	return nil
387
}
388

389
func (a *csAttempt) newStream() error {
390
	cs := a.cs
391
	cs.callHdr.PreviousAttempts = cs.numRetries
392
	s, err := a.t.NewStream(cs.ctx, cs.callHdr)
393
	if err != nil {
394
		if _, ok := err.(transport.PerformedIOError); ok {
395
			// Return without converting to an RPC error so retry code can
396
			// inspect.
397
			return err
398
		}
399
		return toRPCErr(err)
400
	}
401
	cs.attempt.s = s
402
	cs.attempt.p = &parser{r: s}
403
	return nil
404
}
405

406
// clientStream implements a client side Stream.
407
type clientStream struct {
408
	callHdr  *transport.CallHdr
409
	opts     []CallOption
410
	callInfo *callInfo
411
	cc       *ClientConn
412
	desc     *StreamDesc
413

414
	codec baseCodec
415
	cp    Compressor
416
	comp  encoding.Compressor
417

418
	cancel context.CancelFunc // cancels all attempts
419

420
	sentLast  bool // sent an end stream
421
	beginTime time.Time
422

423
	methodConfig *MethodConfig
424

425
	ctx context.Context // the application's context, wrapped by stats/tracing
426

427
	retryThrottler *retryThrottler // The throttler active when the RPC began.
428

429
	binlog *binarylog.MethodLogger // Binary logger, can be nil.
430
	// serverHeaderBinlogged is a boolean for whether server header has been
431
	// logged. Server header will be logged when the first time one of those
432
	// happens: stream.Header(), stream.Recv().
433
	//
434
	// It's only read and used by Recv() and Header(), so it doesn't need to be
435
	// synchronized.
436
	serverHeaderBinlogged bool
437

438
	mu                      sync.Mutex
439
	firstAttempt            bool // if true, transparent retry is valid
440
	numRetries              int  // exclusive of transparent retry attempt(s)
441
	numRetriesSincePushback int  // retries since pushback; to reset backoff
442
	finished                bool // TODO: replace with atomic cmpxchg or sync.Once?
443
	// attempt is the active client stream attempt.
444
	// The only place where it is written is the newAttemptLocked method and this method never writes nil.
445
	// So, attempt can be nil only inside newClientStream function when clientStream is first created.
446
	// One of the first things done after clientStream's creation, is to call newAttemptLocked which either
447
	// assigns a non nil value to the attempt or returns an error. If an error is returned from newAttemptLocked,
448
	// then newClientStream calls finish on the clientStream and returns. So, finish method is the only
449
	// place where we need to check if the attempt is nil.
450
	attempt *csAttempt
451
	// TODO(hedging): hedging will have multiple attempts simultaneously.
452
	committed  bool // active attempt committed for retry?
453
	onCommit   func()
454
	buffer     []func(a *csAttempt) error // operations to replay on retry
455
	bufferSize int                        // current size of buffer
456
}
457

458
// csAttempt implements a single transport stream attempt within a
459
// clientStream.
460
type csAttempt struct {
461
	cs   *clientStream
462
	t    transport.ClientTransport
463
	s    *transport.Stream
464
	p    *parser
465
	done func(balancer.DoneInfo)
466

467
	finished  bool
468
	dc        Decompressor
469
	decomp    encoding.Compressor
470
	decompSet bool
471

472
	mu sync.Mutex // guards trInfo.tr
473
	// trInfo may be nil (if EnableTracing is false).
474
	// trInfo.tr is set when created (if EnableTracing is true),
475
	// and cleared when the finish method is called.
476
	trInfo *traceInfo
477

478
	statsHandler stats.Handler
479
}
480

481
func (cs *clientStream) commitAttemptLocked() {
482
	if !cs.committed && cs.onCommit != nil {
483
		cs.onCommit()
484
	}
485
	cs.committed = true
486
	cs.buffer = nil
487
}
488

489
func (cs *clientStream) commitAttempt() {
490
	cs.mu.Lock()
491
	cs.commitAttemptLocked()
492
	cs.mu.Unlock()
493
}
494

495
// shouldRetry returns nil if the RPC should be retried; otherwise it returns
496
// the error that should be returned by the operation.
497
func (cs *clientStream) shouldRetry(err error) error {
498
	unprocessed := false
499
	if cs.attempt.s == nil {
500
		pioErr, ok := err.(transport.PerformedIOError)
501
		if ok {
502
			// Unwrap error.
503
			err = toRPCErr(pioErr.Err)
504
		} else {
505
			unprocessed = true
506
		}
507
		if !ok && !cs.callInfo.failFast {
508
			// In the event of a non-IO operation error from NewStream, we
509
			// never attempted to write anything to the wire, so we can retry
510
			// indefinitely for non-fail-fast RPCs.
511
			return nil
512
		}
513
	}
514
	if cs.finished || cs.committed {
515
		// RPC is finished or committed; cannot retry.
516
		return err
517
	}
518
	// Wait for the trailers.
519
	if cs.attempt.s != nil {
520
		<-cs.attempt.s.Done()
521
		unprocessed = cs.attempt.s.Unprocessed()
522
	}
523
	if cs.firstAttempt && unprocessed {
524
		// First attempt, stream unprocessed: transparently retry.
525
		return nil
526
	}
527
	if cs.cc.dopts.disableRetry {
528
		return err
529
	}
530

531
	pushback := 0
532
	hasPushback := false
533
	if cs.attempt.s != nil {
534
		if !cs.attempt.s.TrailersOnly() {
535
			return err
536
		}
537

538
		// TODO(retry): Move down if the spec changes to not check server pushback
539
		// before considering this a failure for throttling.
540
		sps := cs.attempt.s.Trailer()["grpc-retry-pushback-ms"]
541
		if len(sps) == 1 {
542
			var e error
543
			if pushback, e = strconv.Atoi(sps[0]); e != nil || pushback < 0 {
544
				channelz.Infof(logger, cs.cc.channelzID, "Server retry pushback specified to abort (%q).", sps[0])
545
				cs.retryThrottler.throttle() // This counts as a failure for throttling.
546
				return err
547
			}
548
			hasPushback = true
549
		} else if len(sps) > 1 {
550
			channelz.Warningf(logger, cs.cc.channelzID, "Server retry pushback specified multiple values (%q); not retrying.", sps)
551
			cs.retryThrottler.throttle() // This counts as a failure for throttling.
552
			return err
553
		}
554
	}
555

556
	var code codes.Code
557
	if cs.attempt.s != nil {
558
		code = cs.attempt.s.Status().Code()
559
	} else {
560
		code = status.Convert(err).Code()
561
	}
562

563
	rp := cs.methodConfig.RetryPolicy
564
	if rp == nil || !rp.RetryableStatusCodes[code] {
565
		return err
566
	}
567

568
	// Note: the ordering here is important; we count this as a failure
569
	// only if the code matched a retryable code.
570
	if cs.retryThrottler.throttle() {
571
		return err
572
	}
573
	if cs.numRetries+1 >= rp.MaxAttempts {
574
		return err
575
	}
576

577
	var dur time.Duration
578
	if hasPushback {
579
		dur = time.Millisecond * time.Duration(pushback)
580
		cs.numRetriesSincePushback = 0
581
	} else {
582
		fact := math.Pow(rp.BackoffMultiplier, float64(cs.numRetriesSincePushback))
583
		cur := float64(rp.InitialBackoff) * fact
584
		if max := float64(rp.MaxBackoff); cur > max {
585
			cur = max
586
		}
587
		dur = time.Duration(grpcrand.Int63n(int64(cur)))
588
		cs.numRetriesSincePushback++
589
	}
590

591
	// TODO(dfawley): we could eagerly fail here if dur puts us past the
592
	// deadline, but unsure if it is worth doing.
593
	t := time.NewTimer(dur)
594
	select {
595
	case <-t.C:
596
		cs.numRetries++
597
		return nil
598
	case <-cs.ctx.Done():
599
		t.Stop()
600
		return status.FromContextError(cs.ctx.Err()).Err()
601
	}
602
}
603

604
// Returns nil if a retry was performed and succeeded; error otherwise.
605
func (cs *clientStream) retryLocked(lastErr error) error {
606
	for {
607
		cs.attempt.finish(lastErr)
608
		if err := cs.shouldRetry(lastErr); err != nil {
609
			cs.commitAttemptLocked()
610
			return err
611
		}
612
		cs.firstAttempt = false
613
		if err := cs.newAttemptLocked(nil, nil); err != nil {
614
			return err
615
		}
616
		if lastErr = cs.replayBufferLocked(); lastErr == nil {
617
			return nil
618
		}
619
	}
620
}
621

622
func (cs *clientStream) Context() context.Context {
623
	cs.commitAttempt()
624
	// No need to lock before using attempt, since we know it is committed and
625
	// cannot change.
626
	return cs.attempt.s.Context()
627
}
628

629
func (cs *clientStream) withRetry(op func(a *csAttempt) error, onSuccess func()) error {
630
	cs.mu.Lock()
631
	for {
632
		if cs.committed {
633
			cs.mu.Unlock()
634
			return op(cs.attempt)
635
		}
636
		a := cs.attempt
637
		cs.mu.Unlock()
638
		err := op(a)
639
		cs.mu.Lock()
640
		if a != cs.attempt {
641
			// We started another attempt already.
642
			continue
643
		}
644
		if err == io.EOF {
645
			<-a.s.Done()
646
		}
647
		if err == nil || (err == io.EOF && a.s.Status().Code() == codes.OK) {
648
			onSuccess()
649
			cs.mu.Unlock()
650
			return err
651
		}
652
		if err := cs.retryLocked(err); err != nil {
653
			cs.mu.Unlock()
654
			return err
655
		}
656
	}
657
}
658

659
func (cs *clientStream) Header() (metadata.MD, error) {
660
	var m metadata.MD
661
	err := cs.withRetry(func(a *csAttempt) error {
662
		var err error
663
		m, err = a.s.Header()
664
		return toRPCErr(err)
665
	}, cs.commitAttemptLocked)
666
	if err != nil {
667
		cs.finish(err)
668
		return nil, err
669
	}
670
	if cs.binlog != nil && !cs.serverHeaderBinlogged {
671
		// Only log if binary log is on and header has not been logged.
672
		logEntry := &binarylog.ServerHeader{
673
			OnClientSide: true,
674
			Header:       m,
675
			PeerAddr:     nil,
676
		}
677
		if peer, ok := peer.FromContext(cs.Context()); ok {
678
			logEntry.PeerAddr = peer.Addr
679
		}
680
		cs.binlog.Log(logEntry)
681
		cs.serverHeaderBinlogged = true
682
	}
683
	return m, err
684
}
685

686
func (cs *clientStream) Trailer() metadata.MD {
687
	// On RPC failure, we never need to retry, because usage requires that
688
	// RecvMsg() returned a non-nil error before calling this function is valid.
689
	// We would have retried earlier if necessary.
690
	//
691
	// Commit the attempt anyway, just in case users are not following those
692
	// directions -- it will prevent races and should not meaningfully impact
693
	// performance.
694
	cs.commitAttempt()
695
	if cs.attempt.s == nil {
696
		return nil
697
	}
698
	return cs.attempt.s.Trailer()
699
}
700

701
func (cs *clientStream) replayBufferLocked() error {
702
	a := cs.attempt
703
	for _, f := range cs.buffer {
704
		if err := f(a); err != nil {
705
			return err
706
		}
707
	}
708
	return nil
709
}
710

711
func (cs *clientStream) bufferForRetryLocked(sz int, op func(a *csAttempt) error) {
712
	// Note: we still will buffer if retry is disabled (for transparent retries).
713
	if cs.committed {
714
		return
715
	}
716
	cs.bufferSize += sz
717
	if cs.bufferSize > cs.callInfo.maxRetryRPCBufferSize {
718
		cs.commitAttemptLocked()
719
		return
720
	}
721
	cs.buffer = append(cs.buffer, op)
722
}
723

724
func (cs *clientStream) SendMsg(m interface{}) (err error) {
725
	defer func() {
726
		if err != nil && err != io.EOF {
727
			// Call finish on the client stream for errors generated by this SendMsg
728
			// call, as these indicate problems created by this client.  (Transport
729
			// errors are converted to an io.EOF error in csAttempt.sendMsg; the real
730
			// error will be returned from RecvMsg eventually in that case, or be
731
			// retried.)
732
			cs.finish(err)
733
		}
734
	}()
735
	if cs.sentLast {
736
		return status.Errorf(codes.Internal, "SendMsg called after CloseSend")
737
	}
738
	if !cs.desc.ClientStreams {
739
		cs.sentLast = true
740
	}
741

742
	// load hdr, payload, data
743
	hdr, payload, data, err := prepareMsg(m, cs.codec, cs.cp, cs.comp)
744
	if err != nil {
745
		return err
746
	}
747

748
	// TODO(dfawley): should we be checking len(data) instead?
749
	if len(payload) > *cs.callInfo.maxSendMessageSize {
750
		return status.Errorf(codes.ResourceExhausted, "trying to send message larger than max (%d vs. %d)", len(payload), *cs.callInfo.maxSendMessageSize)
751
	}
752
	msgBytes := data // Store the pointer before setting to nil. For binary logging.
753
	op := func(a *csAttempt) error {
754
		err := a.sendMsg(m, hdr, payload, data)
755
		// nil out the message and uncomp when replaying; they are only needed for
756
		// stats which is disabled for subsequent attempts.
757
		m, data = nil, nil
758
		return err
759
	}
760
	err = cs.withRetry(op, func() { cs.bufferForRetryLocked(len(hdr)+len(payload), op) })
761
	if cs.binlog != nil && err == nil {
762
		cs.binlog.Log(&binarylog.ClientMessage{
763
			OnClientSide: true,
764
			Message:      msgBytes,
765
		})
766
	}
767
	return
768
}
769

770
func (cs *clientStream) RecvMsg(m interface{}) error {
771
	if cs.binlog != nil && !cs.serverHeaderBinlogged {
772
		// Call Header() to binary log header if it's not already logged.
773
		cs.Header()
774
	}
775
	var recvInfo *payloadInfo
776
	if cs.binlog != nil {
777
		recvInfo = &payloadInfo{}
778
	}
779
	err := cs.withRetry(func(a *csAttempt) error {
780
		return a.recvMsg(m, recvInfo)
781
	}, cs.commitAttemptLocked)
782
	if cs.binlog != nil && err == nil {
783
		cs.binlog.Log(&binarylog.ServerMessage{
784
			OnClientSide: true,
785
			Message:      recvInfo.uncompressedBytes,
786
		})
787
	}
788
	if err != nil || !cs.desc.ServerStreams {
789
		// err != nil or non-server-streaming indicates end of stream.
790
		cs.finish(err)
791

792
		if cs.binlog != nil {
793
			// finish will not log Trailer. Log Trailer here.
794
			logEntry := &binarylog.ServerTrailer{
795
				OnClientSide: true,
796
				Trailer:      cs.Trailer(),
797
				Err:          err,
798
			}
799
			if logEntry.Err == io.EOF {
800
				logEntry.Err = nil
801
			}
802
			if peer, ok := peer.FromContext(cs.Context()); ok {
803
				logEntry.PeerAddr = peer.Addr
804
			}
805
			cs.binlog.Log(logEntry)
806
		}
807
	}
808
	return err
809
}
810

811
func (cs *clientStream) CloseSend() error {
812
	if cs.sentLast {
813
		// TODO: return an error and finish the stream instead, due to API misuse?
814
		return nil
815
	}
816
	cs.sentLast = true
817
	op := func(a *csAttempt) error {
818
		a.t.Write(a.s, nil, nil, &transport.Options{Last: true})
819
		// Always return nil; io.EOF is the only error that might make sense
820
		// instead, but there is no need to signal the client to call RecvMsg
821
		// as the only use left for the stream after CloseSend is to call
822
		// RecvMsg.  This also matches historical behavior.
823
		return nil
824
	}
825
	cs.withRetry(op, func() { cs.bufferForRetryLocked(0, op) })
826
	if cs.binlog != nil {
827
		cs.binlog.Log(&binarylog.ClientHalfClose{
828
			OnClientSide: true,
829
		})
830
	}
831
	// We never returned an error here for reasons.
832
	return nil
833
}
834

835
func (cs *clientStream) finish(err error) {
836
	if err == io.EOF {
837
		// Ending a stream with EOF indicates a success.
838
		err = nil
839
	}
840
	cs.mu.Lock()
841
	if cs.finished {
842
		cs.mu.Unlock()
843
		return
844
	}
845
	cs.finished = true
846
	cs.commitAttemptLocked()
847
	if cs.attempt != nil {
848
		cs.attempt.finish(err)
849
		// after functions all rely upon having a stream.
850
		if cs.attempt.s != nil {
851
			for _, o := range cs.opts {
852
				o.after(cs.callInfo, cs.attempt)
853
			}
854
		}
855
	}
856
	cs.mu.Unlock()
857
	// For binary logging. only log cancel in finish (could be caused by RPC ctx
858
	// canceled or ClientConn closed). Trailer will be logged in RecvMsg.
859
	//
860
	// Only one of cancel or trailer needs to be logged. In the cases where
861
	// users don't call RecvMsg, users must have already canceled the RPC.
862
	if cs.binlog != nil && status.Code(err) == codes.Canceled {
863
		cs.binlog.Log(&binarylog.Cancel{
864
			OnClientSide: true,
865
		})
866
	}
867
	if err == nil {
868
		cs.retryThrottler.successfulRPC()
869
	}
870
	if channelz.IsOn() {
871
		if err != nil {
872
			cs.cc.incrCallsFailed()
873
		} else {
874
			cs.cc.incrCallsSucceeded()
875
		}
876
	}
877
	cs.cancel()
878
}
879

880
func (a *csAttempt) sendMsg(m interface{}, hdr, payld, data []byte) error {
881
	cs := a.cs
882
	if a.trInfo != nil {
883
		a.mu.Lock()
884
		if a.trInfo.tr != nil {
885
			a.trInfo.tr.LazyLog(&payload{sent: true, msg: m}, true)
886
		}
887
		a.mu.Unlock()
888
	}
889
	if err := a.t.Write(a.s, hdr, payld, &transport.Options{Last: !cs.desc.ClientStreams}); err != nil {
890
		if !cs.desc.ClientStreams {
891
			// For non-client-streaming RPCs, we return nil instead of EOF on error
892
			// because the generated code requires it.  finish is not called; RecvMsg()
893
			// will call it with the stream's status independently.
894
			return nil
895
		}
896
		return io.EOF
897
	}
898
	if a.statsHandler != nil {
899
		a.statsHandler.HandleRPC(cs.ctx, outPayload(true, m, data, payld, time.Now()))
900
	}
901
	if channelz.IsOn() {
902
		a.t.IncrMsgSent()
903
	}
904
	return nil
905
}
906

907
func (a *csAttempt) recvMsg(m interface{}, payInfo *payloadInfo) (err error) {
908
	cs := a.cs
909
	if a.statsHandler != nil && payInfo == nil {
910
		payInfo = &payloadInfo{}
911
	}
912

913
	if !a.decompSet {
914
		// Block until we receive headers containing received message encoding.
915
		if ct := a.s.RecvCompress(); ct != "" && ct != encoding.Identity {
916
			if a.dc == nil || a.dc.Type() != ct {
917
				// No configured decompressor, or it does not match the incoming
918
				// message encoding; attempt to find a registered compressor that does.
919
				a.dc = nil
920
				a.decomp = encoding.GetCompressor(ct)
921
			}
922
		} else {
923
			// No compression is used; disable our decompressor.
924
			a.dc = nil
925
		}
926
		// Only initialize this state once per stream.
927
		a.decompSet = true
928
	}
929
	err = recv(a.p, cs.codec, a.s, a.dc, m, *cs.callInfo.maxReceiveMessageSize, payInfo, a.decomp)
930
	if err != nil {
931
		if err == io.EOF {
932
			if statusErr := a.s.Status().Err(); statusErr != nil {
933
				return statusErr
934
			}
935
			return io.EOF // indicates successful end of stream.
936
		}
937
		return toRPCErr(err)
938
	}
939
	if a.trInfo != nil {
940
		a.mu.Lock()
941
		if a.trInfo.tr != nil {
942
			a.trInfo.tr.LazyLog(&payload{sent: false, msg: m}, true)
943
		}
944
		a.mu.Unlock()
945
	}
946
	if a.statsHandler != nil {
947
		a.statsHandler.HandleRPC(cs.ctx, &stats.InPayload{
948
			Client:   true,
949
			RecvTime: time.Now(),
950
			Payload:  m,
951
			// TODO truncate large payload.
952
			Data:       payInfo.uncompressedBytes,
953
			WireLength: payInfo.wireLength + headerLen,
954
			Length:     len(payInfo.uncompressedBytes),
955
		})
956
	}
957
	if channelz.IsOn() {
958
		a.t.IncrMsgRecv()
959
	}
960
	if cs.desc.ServerStreams {
961
		// Subsequent messages should be received by subsequent RecvMsg calls.
962
		return nil
963
	}
964
	// Special handling for non-server-stream rpcs.
965
	// This recv expects EOF or errors, so we don't collect inPayload.
966
	err = recv(a.p, cs.codec, a.s, a.dc, m, *cs.callInfo.maxReceiveMessageSize, nil, a.decomp)
967
	if err == nil {
968
		return toRPCErr(errors.New("grpc: client streaming protocol violation: get <nil>, want <EOF>"))
969
	}
970
	if err == io.EOF {
971
		return a.s.Status().Err() // non-server streaming Recv returns nil on success
972
	}
973
	return toRPCErr(err)
974
}
975

976
func (a *csAttempt) finish(err error) {
977
	a.mu.Lock()
978
	if a.finished {
979
		a.mu.Unlock()
980
		return
981
	}
982
	a.finished = true
983
	if err == io.EOF {
984
		// Ending a stream with EOF indicates a success.
985
		err = nil
986
	}
987
	var tr metadata.MD
988
	if a.s != nil {
989
		a.t.CloseStream(a.s, err)
990
		tr = a.s.Trailer()
991
	}
992

993
	if a.done != nil {
994
		br := false
995
		if a.s != nil {
996
			br = a.s.BytesReceived()
997
		}
998
		a.done(balancer.DoneInfo{
999
			Err:           err,
1000
			Trailer:       tr,
1001
			BytesSent:     a.s != nil,
1002
			BytesReceived: br,
1003
			ServerLoad:    balancerload.Parse(tr),
1004
		})
1005
	}
1006
	if a.statsHandler != nil {
1007
		end := &stats.End{
1008
			Client:    true,
1009
			BeginTime: a.cs.beginTime,
1010
			EndTime:   time.Now(),
1011
			Trailer:   tr,
1012
			Error:     err,
1013
		}
1014
		a.statsHandler.HandleRPC(a.cs.ctx, end)
1015
	}
1016
	if a.trInfo != nil && a.trInfo.tr != nil {
1017
		if err == nil {
1018
			a.trInfo.tr.LazyPrintf("RPC: [OK]")
1019
		} else {
1020
			a.trInfo.tr.LazyPrintf("RPC: [%v]", err)
1021
			a.trInfo.tr.SetError()
1022
		}
1023
		a.trInfo.tr.Finish()
1024
		a.trInfo.tr = nil
1025
	}
1026
	a.mu.Unlock()
1027
}
1028

1029
// newClientStream creates a ClientStream with the specified transport, on the
1030
// given addrConn.
1031
//
1032
// It's expected that the given transport is either the same one in addrConn, or
1033
// is already closed. To avoid race, transport is specified separately, instead
1034
// of using ac.transpot.
1035
//
1036
// Main difference between this and ClientConn.NewStream:
1037
// - no retry
1038
// - no service config (or wait for service config)
1039
// - no tracing or stats
1040
func newNonRetryClientStream(ctx context.Context, desc *StreamDesc, method string, t transport.ClientTransport, ac *addrConn, opts ...CallOption) (_ ClientStream, err error) {
1041
	if t == nil {
1042
		// TODO: return RPC error here?
1043
		return nil, errors.New("transport provided is nil")
1044
	}
1045
	// defaultCallInfo contains unnecessary info(i.e. failfast, maxRetryRPCBufferSize), so we just initialize an empty struct.
1046
	c := &callInfo{}
1047

1048
	// Possible context leak:
1049
	// The cancel function for the child context we create will only be called
1050
	// when RecvMsg returns a non-nil error, if the ClientConn is closed, or if
1051
	// an error is generated by SendMsg.
1052
	// https://github.com/grpc/grpc-go/issues/1818.
1053
	ctx, cancel := context.WithCancel(ctx)
1054
	defer func() {
1055
		if err != nil {
1056
			cancel()
1057
		}
1058
	}()
1059

1060
	for _, o := range opts {
1061
		if err := o.before(c); err != nil {
1062
			return nil, toRPCErr(err)
1063
		}
1064
	}
1065
	c.maxReceiveMessageSize = getMaxSize(nil, c.maxReceiveMessageSize, defaultClientMaxReceiveMessageSize)
1066
	c.maxSendMessageSize = getMaxSize(nil, c.maxSendMessageSize, defaultServerMaxSendMessageSize)
1067
	if err := setCallInfoCodec(c); err != nil {
1068
		return nil, err
1069
	}
1070

1071
	callHdr := &transport.CallHdr{
1072
		Host:           ac.cc.authority,
1073
		Method:         method,
1074
		ContentSubtype: c.contentSubtype,
1075
	}
1076

1077
	// Set our outgoing compression according to the UseCompressor CallOption, if
1078
	// set.  In that case, also find the compressor from the encoding package.
1079
	// Otherwise, use the compressor configured by the WithCompressor DialOption,
1080
	// if set.
1081
	var cp Compressor
1082
	var comp encoding.Compressor
1083
	if ct := c.compressorType; ct != "" {
1084
		callHdr.SendCompress = ct
1085
		if ct != encoding.Identity {
1086
			comp = encoding.GetCompressor(ct)
1087
			if comp == nil {
1088
				return nil, status.Errorf(codes.Internal, "grpc: Compressor is not installed for requested grpc-encoding %q", ct)
1089
			}
1090
		}
1091
	} else if ac.cc.dopts.cp != nil {
1092
		callHdr.SendCompress = ac.cc.dopts.cp.Type()
1093
		cp = ac.cc.dopts.cp
1094
	}
1095
	if c.creds != nil {
1096
		callHdr.Creds = c.creds
1097
	}
1098

1099
	// Use a special addrConnStream to avoid retry.
1100
	as := &addrConnStream{
1101
		callHdr:  callHdr,
1102
		ac:       ac,
1103
		ctx:      ctx,
1104
		cancel:   cancel,
1105
		opts:     opts,
1106
		callInfo: c,
1107
		desc:     desc,
1108
		codec:    c.codec,
1109
		cp:       cp,
1110
		comp:     comp,
1111
		t:        t,
1112
	}
1113

1114
	s, err := as.t.NewStream(as.ctx, as.callHdr)
1115
	if err != nil {
1116
		err = toRPCErr(err)
1117
		return nil, err
1118
	}
1119
	as.s = s
1120
	as.p = &parser{r: s}
1121
	ac.incrCallsStarted()
1122
	if desc != unaryStreamDesc {
1123
		// Listen on cc and stream contexts to cleanup when the user closes the
1124
		// ClientConn or cancels the stream context.  In all other cases, an error
1125
		// should already be injected into the recv buffer by the transport, which
1126
		// the client will eventually receive, and then we will cancel the stream's
1127
		// context in clientStream.finish.
1128
		go func() {
1129
			select {
1130
			case <-ac.ctx.Done():
1131
				as.finish(status.Error(codes.Canceled, "grpc: the SubConn is closing"))
1132
			case <-ctx.Done():
1133
				as.finish(toRPCErr(ctx.Err()))
1134
			}
1135
		}()
1136
	}
1137
	return as, nil
1138
}
1139

1140
type addrConnStream struct {
1141
	s         *transport.Stream
1142
	ac        *addrConn
1143
	callHdr   *transport.CallHdr
1144
	cancel    context.CancelFunc
1145
	opts      []CallOption
1146
	callInfo  *callInfo
1147
	t         transport.ClientTransport
1148
	ctx       context.Context
1149
	sentLast  bool
1150
	desc      *StreamDesc
1151
	codec     baseCodec
1152
	cp        Compressor
1153
	comp      encoding.Compressor
1154
	decompSet bool
1155
	dc        Decompressor
1156
	decomp    encoding.Compressor
1157
	p         *parser
1158
	mu        sync.Mutex
1159
	finished  bool
1160
}
1161

1162
func (as *addrConnStream) Header() (metadata.MD, error) {
1163
	m, err := as.s.Header()
1164
	if err != nil {
1165
		as.finish(toRPCErr(err))
1166
	}
1167
	return m, err
1168
}
1169

1170
func (as *addrConnStream) Trailer() metadata.MD {
1171
	return as.s.Trailer()
1172
}
1173

1174
func (as *addrConnStream) CloseSend() error {
1175
	if as.sentLast {
1176
		// TODO: return an error and finish the stream instead, due to API misuse?
1177
		return nil
1178
	}
1179
	as.sentLast = true
1180

1181
	as.t.Write(as.s, nil, nil, &transport.Options{Last: true})
1182
	// Always return nil; io.EOF is the only error that might make sense
1183
	// instead, but there is no need to signal the client to call RecvMsg
1184
	// as the only use left for the stream after CloseSend is to call
1185
	// RecvMsg.  This also matches historical behavior.
1186
	return nil
1187
}
1188

1189
func (as *addrConnStream) Context() context.Context {
1190
	return as.s.Context()
1191
}
1192

1193
func (as *addrConnStream) SendMsg(m interface{}) (err error) {
1194
	defer func() {
1195
		if err != nil && err != io.EOF {
1196
			// Call finish on the client stream for errors generated by this SendMsg
1197
			// call, as these indicate problems created by this client.  (Transport
1198
			// errors are converted to an io.EOF error in csAttempt.sendMsg; the real
1199
			// error will be returned from RecvMsg eventually in that case, or be
1200
			// retried.)
1201
			as.finish(err)
1202
		}
1203
	}()
1204
	if as.sentLast {
1205
		return status.Errorf(codes.Internal, "SendMsg called after CloseSend")
1206
	}
1207
	if !as.desc.ClientStreams {
1208
		as.sentLast = true
1209
	}
1210

1211
	// load hdr, payload, data
1212
	hdr, payld, _, err := prepareMsg(m, as.codec, as.cp, as.comp)
1213
	if err != nil {
1214
		return err
1215
	}
1216

1217
	// TODO(dfawley): should we be checking len(data) instead?
1218
	if len(payld) > *as.callInfo.maxSendMessageSize {
1219
		return status.Errorf(codes.ResourceExhausted, "trying to send message larger than max (%d vs. %d)", len(payld), *as.callInfo.maxSendMessageSize)
1220
	}
1221

1222
	if err := as.t.Write(as.s, hdr, payld, &transport.Options{Last: !as.desc.ClientStreams}); err != nil {
1223
		if !as.desc.ClientStreams {
1224
			// For non-client-streaming RPCs, we return nil instead of EOF on error
1225
			// because the generated code requires it.  finish is not called; RecvMsg()
1226
			// will call it with the stream's status independently.
1227
			return nil
1228
		}
1229
		return io.EOF
1230
	}
1231

1232
	if channelz.IsOn() {
1233
		as.t.IncrMsgSent()
1234
	}
1235
	return nil
1236
}
1237

1238
func (as *addrConnStream) RecvMsg(m interface{}) (err error) {
1239
	defer func() {
1240
		if err != nil || !as.desc.ServerStreams {
1241
			// err != nil or non-server-streaming indicates end of stream.
1242
			as.finish(err)
1243
		}
1244
	}()
1245

1246
	if !as.decompSet {
1247
		// Block until we receive headers containing received message encoding.
1248
		if ct := as.s.RecvCompress(); ct != "" && ct != encoding.Identity {
1249
			if as.dc == nil || as.dc.Type() != ct {
1250
				// No configured decompressor, or it does not match the incoming
1251
				// message encoding; attempt to find a registered compressor that does.
1252
				as.dc = nil
1253
				as.decomp = encoding.GetCompressor(ct)
1254
			}
1255
		} else {
1256
			// No compression is used; disable our decompressor.
1257
			as.dc = nil
1258
		}
1259
		// Only initialize this state once per stream.
1260
		as.decompSet = true
1261
	}
1262
	err = recv(as.p, as.codec, as.s, as.dc, m, *as.callInfo.maxReceiveMessageSize, nil, as.decomp)
1263
	if err != nil {
1264
		if err == io.EOF {
1265
			if statusErr := as.s.Status().Err(); statusErr != nil {
1266
				return statusErr
1267
			}
1268
			return io.EOF // indicates successful end of stream.
1269
		}
1270
		return toRPCErr(err)
1271
	}
1272

1273
	if channelz.IsOn() {
1274
		as.t.IncrMsgRecv()
1275
	}
1276
	if as.desc.ServerStreams {
1277
		// Subsequent messages should be received by subsequent RecvMsg calls.
1278
		return nil
1279
	}
1280

1281
	// Special handling for non-server-stream rpcs.
1282
	// This recv expects EOF or errors, so we don't collect inPayload.
1283
	err = recv(as.p, as.codec, as.s, as.dc, m, *as.callInfo.maxReceiveMessageSize, nil, as.decomp)
1284
	if err == nil {
1285
		return toRPCErr(errors.New("grpc: client streaming protocol violation: get <nil>, want <EOF>"))
1286
	}
1287
	if err == io.EOF {
1288
		return as.s.Status().Err() // non-server streaming Recv returns nil on success
1289
	}
1290
	return toRPCErr(err)
1291
}
1292

1293
func (as *addrConnStream) finish(err error) {
1294
	as.mu.Lock()
1295
	if as.finished {
1296
		as.mu.Unlock()
1297
		return
1298
	}
1299
	as.finished = true
1300
	if err == io.EOF {
1301
		// Ending a stream with EOF indicates a success.
1302
		err = nil
1303
	}
1304
	if as.s != nil {
1305
		as.t.CloseStream(as.s, err)
1306
	}
1307

1308
	if err != nil {
1309
		as.ac.incrCallsFailed()
1310
	} else {
1311
		as.ac.incrCallsSucceeded()
1312
	}
1313
	as.cancel()
1314
	as.mu.Unlock()
1315
}
1316

1317
// ServerStream defines the server-side behavior of a streaming RPC.
1318
//
1319
// All errors returned from ServerStream methods are compatible with the
1320
// status package.
1321
type ServerStream interface {
1322
	// SetHeader sets the header metadata. It may be called multiple times.
1323
	// When call multiple times, all the provided metadata will be merged.
1324
	// All the metadata will be sent out when one of the following happens:
1325
	//  - ServerStream.SendHeader() is called;
1326
	//  - The first response is sent out;
1327
	//  - An RPC status is sent out (error or success).
1328
	SetHeader(metadata.MD) error
1329
	// SendHeader sends the header metadata.
1330
	// The provided md and headers set by SetHeader() will be sent.
1331
	// It fails if called multiple times.
1332
	SendHeader(metadata.MD) error
1333
	// SetTrailer sets the trailer metadata which will be sent with the RPC status.
1334
	// When called more than once, all the provided metadata will be merged.
1335
	SetTrailer(metadata.MD)
1336
	// Context returns the context for this stream.
1337
	Context() context.Context
1338
	// SendMsg sends a message. On error, SendMsg aborts the stream and the
1339
	// error is returned directly.
1340
	//
1341
	// SendMsg blocks until:
1342
	//   - There is sufficient flow control to schedule m with the transport, or
1343
	//   - The stream is done, or
1344
	//   - The stream breaks.
1345
	//
1346
	// SendMsg does not wait until the message is received by the client. An
1347
	// untimely stream closure may result in lost messages.
1348
	//
1349
	// It is safe to have a goroutine calling SendMsg and another goroutine
1350
	// calling RecvMsg on the same stream at the same time, but it is not safe
1351
	// to call SendMsg on the same stream in different goroutines.
1352
	SendMsg(m interface{}) error
1353
	// RecvMsg blocks until it receives a message into m or the stream is
1354
	// done. It returns io.EOF when the client has performed a CloseSend. On
1355
	// any non-EOF error, the stream is aborted and the error contains the
1356
	// RPC status.
1357
	//
1358
	// It is safe to have a goroutine calling SendMsg and another goroutine
1359
	// calling RecvMsg on the same stream at the same time, but it is not
1360
	// safe to call RecvMsg on the same stream in different goroutines.
1361
	RecvMsg(m interface{}) error
1362
}
1363

1364
// serverStream implements a server side Stream.
1365
type serverStream struct {
1366
	ctx   context.Context
1367
	t     transport.ServerTransport
1368
	s     *transport.Stream
1369
	p     *parser
1370
	codec baseCodec
1371

1372
	cp     Compressor
1373
	dc     Decompressor
1374
	comp   encoding.Compressor
1375
	decomp encoding.Compressor
1376

1377
	maxReceiveMessageSize int
1378
	maxSendMessageSize    int
1379
	trInfo                *traceInfo
1380

1381
	statsHandler stats.Handler
1382

1383
	binlog *binarylog.MethodLogger
1384
	// serverHeaderBinlogged indicates whether server header has been logged. It
1385
	// will happen when one of the following two happens: stream.SendHeader(),
1386
	// stream.Send().
1387
	//
1388
	// It's only checked in send and sendHeader, doesn't need to be
1389
	// synchronized.
1390
	serverHeaderBinlogged bool
1391

1392
	mu sync.Mutex // protects trInfo.tr after the service handler runs.
1393
}
1394

1395
func (ss *serverStream) Context() context.Context {
1396
	return ss.ctx
1397
}
1398

1399
func (ss *serverStream) SetHeader(md metadata.MD) error {
1400
	if md.Len() == 0 {
1401
		return nil
1402
	}
1403
	return ss.s.SetHeader(md)
1404
}
1405

1406
func (ss *serverStream) SendHeader(md metadata.MD) error {
1407
	err := ss.t.WriteHeader(ss.s, md)
1408
	if ss.binlog != nil && !ss.serverHeaderBinlogged {
1409
		h, _ := ss.s.Header()
1410
		ss.binlog.Log(&binarylog.ServerHeader{
1411
			Header: h,
1412
		})
1413
		ss.serverHeaderBinlogged = true
1414
	}
1415
	return err
1416
}
1417

1418
func (ss *serverStream) SetTrailer(md metadata.MD) {
1419
	if md.Len() == 0 {
1420
		return
1421
	}
1422
	ss.s.SetTrailer(md)
1423
}
1424

1425
func (ss *serverStream) SendMsg(m interface{}) (err error) {
1426
	defer func() {
1427
		if ss.trInfo != nil {
1428
			ss.mu.Lock()
1429
			if ss.trInfo.tr != nil {
1430
				if err == nil {
1431
					ss.trInfo.tr.LazyLog(&payload{sent: true, msg: m}, true)
1432
				} else {
1433
					ss.trInfo.tr.LazyLog(&fmtStringer{"%v", []interface{}{err}}, true)
1434
					ss.trInfo.tr.SetError()
1435
				}
1436
			}
1437
			ss.mu.Unlock()
1438
		}
1439
		if err != nil && err != io.EOF {
1440
			st, _ := status.FromError(toRPCErr(err))
1441
			ss.t.WriteStatus(ss.s, st)
1442
			// Non-user specified status was sent out. This should be an error
1443
			// case (as a server side Cancel maybe).
1444
			//
1445
			// This is not handled specifically now. User will return a final
1446
			// status from the service handler, we will log that error instead.
1447
			// This behavior is similar to an interceptor.
1448
		}
1449
		if channelz.IsOn() && err == nil {
1450
			ss.t.IncrMsgSent()
1451
		}
1452
	}()
1453

1454
	// load hdr, payload, data
1455
	hdr, payload, data, err := prepareMsg(m, ss.codec, ss.cp, ss.comp)
1456
	if err != nil {
1457
		return err
1458
	}
1459

1460
	// TODO(dfawley): should we be checking len(data) instead?
1461
	if len(payload) > ss.maxSendMessageSize {
1462
		return status.Errorf(codes.ResourceExhausted, "trying to send message larger than max (%d vs. %d)", len(payload), ss.maxSendMessageSize)
1463
	}
1464
	if err := ss.t.Write(ss.s, hdr, payload, &transport.Options{Last: false}); err != nil {
1465
		return toRPCErr(err)
1466
	}
1467
	if ss.binlog != nil {
1468
		if !ss.serverHeaderBinlogged {
1469
			h, _ := ss.s.Header()
1470
			ss.binlog.Log(&binarylog.ServerHeader{
1471
				Header: h,
1472
			})
1473
			ss.serverHeaderBinlogged = true
1474
		}
1475
		ss.binlog.Log(&binarylog.ServerMessage{
1476
			Message: data,
1477
		})
1478
	}
1479
	if ss.statsHandler != nil {
1480
		ss.statsHandler.HandleRPC(ss.s.Context(), outPayload(false, m, data, payload, time.Now()))
1481
	}
1482
	return nil
1483
}
1484

1485
func (ss *serverStream) RecvMsg(m interface{}) (err error) {
1486
	defer func() {
1487
		if ss.trInfo != nil {
1488
			ss.mu.Lock()
1489
			if ss.trInfo.tr != nil {
1490
				if err == nil {
1491
					ss.trInfo.tr.LazyLog(&payload{sent: false, msg: m}, true)
1492
				} else if err != io.EOF {
1493
					ss.trInfo.tr.LazyLog(&fmtStringer{"%v", []interface{}{err}}, true)
1494
					ss.trInfo.tr.SetError()
1495
				}
1496
			}
1497
			ss.mu.Unlock()
1498
		}
1499
		if err != nil && err != io.EOF {
1500
			st, _ := status.FromError(toRPCErr(err))
1501
			ss.t.WriteStatus(ss.s, st)
1502
			// Non-user specified status was sent out. This should be an error
1503
			// case (as a server side Cancel maybe).
1504
			//
1505
			// This is not handled specifically now. User will return a final
1506
			// status from the service handler, we will log that error instead.
1507
			// This behavior is similar to an interceptor.
1508
		}
1509
		if channelz.IsOn() && err == nil {
1510
			ss.t.IncrMsgRecv()
1511
		}
1512
	}()
1513
	var payInfo *payloadInfo
1514
	if ss.statsHandler != nil || ss.binlog != nil {
1515
		payInfo = &payloadInfo{}
1516
	}
1517
	if err := recv(ss.p, ss.codec, ss.s, ss.dc, m, ss.maxReceiveMessageSize, payInfo, ss.decomp); err != nil {
1518
		if err == io.EOF {
1519
			if ss.binlog != nil {
1520
				ss.binlog.Log(&binarylog.ClientHalfClose{})
1521
			}
1522
			return err
1523
		}
1524
		if err == io.ErrUnexpectedEOF {
1525
			err = status.Errorf(codes.Internal, io.ErrUnexpectedEOF.Error())
1526
		}
1527
		return toRPCErr(err)
1528
	}
1529
	if ss.statsHandler != nil {
1530
		ss.statsHandler.HandleRPC(ss.s.Context(), &stats.InPayload{
1531
			RecvTime: time.Now(),
1532
			Payload:  m,
1533
			// TODO truncate large payload.
1534
			Data:       payInfo.uncompressedBytes,
1535
			WireLength: payInfo.wireLength + headerLen,
1536
			Length:     len(payInfo.uncompressedBytes),
1537
		})
1538
	}
1539
	if ss.binlog != nil {
1540
		ss.binlog.Log(&binarylog.ClientMessage{
1541
			Message: payInfo.uncompressedBytes,
1542
		})
1543
	}
1544
	return nil
1545
}
1546

1547
// MethodFromServerStream returns the method string for the input stream.
1548
// The returned string is in the format of "/service/method".
1549
func MethodFromServerStream(stream ServerStream) (string, bool) {
1550
	return Method(stream.Context())
1551
}
1552

1553
// prepareMsg returns the hdr, payload and data
1554
// using the compressors passed or using the
1555
// passed preparedmsg
1556
func prepareMsg(m interface{}, codec baseCodec, cp Compressor, comp encoding.Compressor) (hdr, payload, data []byte, err error) {
1557
	if preparedMsg, ok := m.(*PreparedMsg); ok {
1558
		return preparedMsg.hdr, preparedMsg.payload, preparedMsg.encodedData, nil
1559
	}
1560
	// The input interface is not a prepared msg.
1561
	// Marshal and Compress the data at this point
1562
	data, err = encode(codec, m)
1563
	if err != nil {
1564
		return nil, nil, nil, err
1565
	}
1566
	compData, err := compress(data, cp, comp)
1567
	if err != nil {
1568
		return nil, nil, nil, err
1569
	}
1570
	hdr, payload = msgHeader(data, compData)
1571
	return hdr, payload, data, nil
1572
}
1573