Fresh dep ensure

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

@@ -21,6 +21,8 @@ package grpc
 import (
 	"errors"
 	"io"
+	"math"
+	"strconv"
 	"sync"
 	"time"
 
@@ -29,11 +31,13 @@ import (
 	"google.golang.org/grpc/balancer"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/encoding"
+	"google.golang.org/grpc/grpclog"
 	"google.golang.org/grpc/internal/channelz"
+	"google.golang.org/grpc/internal/grpcrand"
+	"google.golang.org/grpc/internal/transport"
 	"google.golang.org/grpc/metadata"
 	"google.golang.org/grpc/stats"
 	"google.golang.org/grpc/status"
-	"google.golang.org/grpc/transport"
 )
 
 // StreamHandler defines the handler called by gRPC server to complete the
@@ -55,31 +59,20 @@ type StreamDesc struct {
 
 // Stream defines the common interface a client or server stream has to satisfy.
 //
-// All errors returned from Stream are compatible with the status package.
+// Deprecated: See ClientStream and ServerStream documentation instead.
 type Stream interface {
-	// Context returns the context for this stream.
+	// Deprecated: See ClientStream and ServerStream documentation instead.
 	Context() context.Context
-	// SendMsg blocks until it sends m, the stream is done or the stream
-	// breaks.
-	// On error, it aborts the stream and returns an RPC status on client
-	// side. On server side, it simply returns the error to the caller.
-	// SendMsg is called by generated code. Also Users can call SendMsg
-	// directly when it is really needed in their use cases.
-	// It's safe to have a goroutine calling SendMsg and another goroutine calling
-	// recvMsg on the same stream at the same time.
-	// But it is not safe to call SendMsg on the same stream in different goroutines.
+	// Deprecated: See ClientStream and ServerStream documentation instead.
 	SendMsg(m interface{}) error
-	// RecvMsg blocks until it receives a message or the stream is
-	// done. On client side, it returns io.EOF when the stream is done. On
-	// any other error, it aborts the stream and returns an RPC status. On
-	// server side, it simply returns the error to the caller.
-	// It's safe to have a goroutine calling SendMsg and another goroutine calling
-	// recvMsg on the same stream at the same time.
-	// But it is not safe to call RecvMsg on the same stream in different goroutines.
+	// Deprecated: See ClientStream and ServerStream documentation instead.
 	RecvMsg(m interface{}) error
 }
 
-// ClientStream defines the interface a client stream has to satisfy.
+// ClientStream defines the client-side behavior of a streaming RPC.
+//
+// All errors returned from ClientStream methods are compatible with the
+// status package.
 type ClientStream interface {
 	// Header returns the header metadata received from the server if there
 	// is any. It blocks if the metadata is not ready to read.
@@ -91,13 +84,38 @@ type ClientStream interface {
 	// CloseSend closes the send direction of the stream. It closes the stream
 	// when non-nil error is met.
 	CloseSend() error
-	// Stream.SendMsg() may return a non-nil error when something wrong happens sending
-	// the request. The returned error indicates the status of this sending, not the final
-	// status of the RPC.
+	// Context returns the context for this stream.
 	//
-	// Always call Stream.RecvMsg() to drain the stream and get the final
-	// status, otherwise there could be leaked resources.
-	Stream
+	// It should not be called until after Header or RecvMsg has returned. Once
+	// called, subsequent client-side retries are disabled.
+	Context() context.Context
+	// SendMsg is generally called by generated code. On error, SendMsg aborts
+	// the stream. If the error was generated by the client, the status is
+	// returned directly; otherwise, io.EOF is returned and the status of
+	// the stream may be discovered using RecvMsg.
+	//
+	// SendMsg blocks until:
+	//   - There is sufficient flow control to schedule m with the transport, or
+	//   - The stream is done, or
+	//   - The stream breaks.
+	//
+	// SendMsg does not wait until the message is received by the server. An
+	// untimely stream closure may result in lost messages. To ensure delivery,
+	// users should ensure the RPC completed successfully using RecvMsg.
+	//
+	// It is safe to have a goroutine calling SendMsg and another goroutine
+	// calling RecvMsg on the same stream at the same time, but it is not safe
+	// to call SendMsg on the same stream in different goroutines.
+	SendMsg(m interface{}) error
+	// RecvMsg blocks until it receives a message into m or the stream is
+	// done. It returns io.EOF when the stream completes successfully. On
+	// any other error, the stream is aborted and the error contains the RPC
+	// status.
+	//
+	// It is safe to have a goroutine calling SendMsg and another goroutine
+	// calling RecvMsg on the same stream at the same time, but it is not
+	// safe to call RecvMsg on the same stream in different goroutines.
+	RecvMsg(m interface{}) error
 }
 
 // NewStream creates a new Stream for the client side. This is typically
@@ -128,8 +146,6 @@ func (cc *ClientConn) NewStream(ctx context.Context, desc *StreamDesc, method st
 }
 
 // NewClientStream is a wrapper for ClientConn.NewStream.
-//
-// DEPRECATED: Use ClientConn.NewStream instead.
 func NewClientStream(ctx context.Context, desc *StreamDesc, cc *ClientConn, method string, opts ...CallOption) (ClientStream, error) {
 	return cc.NewStream(ctx, desc, method, opts...)
 }
@@ -178,13 +194,8 @@ func newClientStream(ctx context.Context, desc *StreamDesc, cc *ClientConn, meth
 	}
 
 	callHdr := &transport.CallHdr{
-		Host:   cc.authority,
-		Method: method,
-		// If it's not client streaming, we should already have the request to be sent,
-		// so we don't flush the header.
-		// If it's client streaming, the user may never send a request or send it any
-		// time soon, so we ask the transport to flush the header.
-		Flush:          desc.ClientStreams,
+		Host:           cc.authority,
+		Method:         method,
 		ContentSubtype: c.contentSubtype,
 	}
 
@@ -218,15 +229,6 @@ func newClientStream(ctx context.Context, desc *StreamDesc, cc *ClientConn, meth
 		}
 		trInfo.tr.LazyLog(&trInfo.firstLine, false)
 		ctx = trace.NewContext(ctx, trInfo.tr)
-		defer func() {
-			if err != nil {
-				// Need to call tr.finish() if error is returned.
-				// Because tr will not be returned to caller.
-				trInfo.tr.LazyPrintf("RPC: [%v]", err)
-				trInfo.tr.SetError()
-				trInfo.tr.Finish()
-			}
-		}()
 	}
 	ctx = newContextWithRPCInfo(ctx, c.failFast)
 	sh := cc.dopts.copts.StatsHandler
@@ -240,80 +242,41 @@ func newClientStream(ctx context.Context, desc *StreamDesc, cc *ClientConn, meth
 			FailFast:  c.failFast,
 		}
 		sh.HandleRPC(ctx, begin)
-		defer func() {
-			if err != nil {
-				// Only handle end stats if err != nil.
-				end := &stats.End{
-					Client:    true,
-					Error:     err,
-					BeginTime: beginTime,
-					EndTime:   time.Now(),
-				}
-				sh.HandleRPC(ctx, end)
-			}
-		}()
-	}
-
-	var (
-		t    transport.ClientTransport
-		s    *transport.Stream
-		done func(balancer.DoneInfo)
-	)
-	for {
-		// Check to make sure the context has expired.  This will prevent us from
-		// looping forever if an error occurs for wait-for-ready RPCs where no data
-		// is sent on the wire.
-		select {
-		case <-ctx.Done():
-			return nil, toRPCErr(ctx.Err())
-		default:
-		}
-
-		t, done, err = cc.getTransport(ctx, c.failFast)
-		if err != nil {
-			return nil, err
-		}
-
-		s, err = t.NewStream(ctx, callHdr)
-		if err != nil {
-			if done != nil {
-				done(balancer.DoneInfo{Err: err})
-				done = nil
-			}
-			// In the event of any error from NewStream, we never attempted to write
-			// anything to the wire, so we can retry indefinitely for non-fail-fast
-			// RPCs.
-			if !c.failFast {
-				continue
-			}
-			return nil, toRPCErr(err)
-		}
-		break
 	}
 
 	cs := &clientStream{
-		opts:   opts,
-		c:      c,
-		cc:     cc,
-		desc:   desc,
-		codec:  c.codec,
-		cp:     cp,
-		comp:   comp,
-		cancel: cancel,
-		attempt: &csAttempt{
-			t:            t,
-			s:            s,
-			p:            &parser{r: s},
-			done:         done,
-			dc:           cc.dopts.dc,
-			ctx:          ctx,
-			trInfo:       trInfo,
-			statsHandler: sh,
-			beginTime:    beginTime,
-		},
+		callHdr:      callHdr,
+		ctx:          ctx,
+		methodConfig: &mc,
+		opts:         opts,
+		callInfo:     c,
+		cc:           cc,
+		desc:         desc,
+		codec:        c.codec,
+		cp:           cp,
+		comp:         comp,
+		cancel:       cancel,
+		beginTime:    beginTime,
+		firstAttempt: true,
 	}
-	cs.c.stream = cs
-	cs.attempt.cs = cs
+	if !cc.dopts.disableRetry {
+		cs.retryThrottler = cc.retryThrottler.Load().(*retryThrottler)
+	}
+
+	cs.callInfo.stream = cs
+	// Only this initial attempt has stats/tracing.
+	// TODO(dfawley): move to newAttempt when per-attempt stats are implemented.
+	if err := cs.newAttemptLocked(sh, trInfo); err != nil {
+		cs.finish(err)
+		return nil, err
+	}
+
+	op := func(a *csAttempt) error { return a.newStream() }
+	if err := cs.withRetry(op, func() { cs.bufferForRetryLocked(0, op) }); err != nil {
+		cs.finish(err)
+		return nil, err
+	}
+
 	if desc != unaryStreamDesc {
 		// Listen on cc and stream contexts to cleanup when the user closes the
 		// ClientConn or cancels the stream context.  In all other cases, an error
@@ -332,12 +295,45 @@ func newClientStream(ctx context.Context, desc *StreamDesc, cc *ClientConn, meth
 	return cs, nil
 }
 
+func (cs *clientStream) newAttemptLocked(sh stats.Handler, trInfo traceInfo) error {
+	cs.attempt = &csAttempt{
+		cs:           cs,
+		dc:           cs.cc.dopts.dc,
+		statsHandler: sh,
+		trInfo:       trInfo,
+	}
+
+	if err := cs.ctx.Err(); err != nil {
+		return toRPCErr(err)
+	}
+	t, done, err := cs.cc.getTransport(cs.ctx, cs.callInfo.failFast, cs.callHdr.Method)
+	if err != nil {
+		return err
+	}
+	cs.attempt.t = t
+	cs.attempt.done = done
+	return nil
+}
+
+func (a *csAttempt) newStream() error {
+	cs := a.cs
+	cs.callHdr.PreviousAttempts = cs.numRetries
+	s, err := a.t.NewStream(cs.ctx, cs.callHdr)
+	if err != nil {
+		return toRPCErr(err)
+	}
+	cs.attempt.s = s
+	cs.attempt.p = &parser{r: s}
+	return nil
+}
+
 // clientStream implements a client side Stream.
 type clientStream struct {
-	opts []CallOption
-	c    *callInfo
-	cc   *ClientConn
-	desc *StreamDesc
+	callHdr  *transport.CallHdr
+	opts     []CallOption
+	callInfo *callInfo
+	cc       *ClientConn
+	desc     *StreamDesc
 
 	codec baseCodec
 	cp    Compressor
@@ -345,13 +341,25 @@ type clientStream struct {
 
 	cancel context.CancelFunc // cancels all attempts
 
-	sentLast bool // sent an end stream
+	sentLast  bool // sent an end stream
+	beginTime time.Time
 
-	mu       sync.Mutex // guards finished
-	finished bool       // TODO: replace with atomic cmpxchg or sync.Once?
+	methodConfig *MethodConfig
 
-	attempt *csAttempt // the active client stream attempt
+	ctx context.Context // the application's context, wrapped by stats/tracing
+
+	retryThrottler *retryThrottler // The throttler active when the RPC began.
+
+	mu                      sync.Mutex
+	firstAttempt            bool       // if true, transparent retry is valid
+	numRetries              int        // exclusive of transparent retry attempt(s)
+	numRetriesSincePushback int        // retries since pushback; to reset backoff
+	finished                bool       // TODO: replace with atomic cmpxchg or sync.Once?
+	attempt                 *csAttempt // the active client stream attempt
 	// TODO(hedging): hedging will have multiple attempts simultaneously.
+	committed  bool                       // active attempt committed for retry?
+	buffer     []func(a *csAttempt) error // operations to replay on retry
+	bufferSize int                        // current size of buffer
 }
 
 // csAttempt implements a single transport stream attempt within a
@@ -363,53 +371,305 @@ type csAttempt struct {
 	p    *parser
 	done func(balancer.DoneInfo)
 
+	finished  bool
 	dc        Decompressor
 	decomp    encoding.Compressor
 	decompSet bool
 
-	ctx context.Context // the application's context, wrapped by stats/tracing
-
 	mu sync.Mutex // guards trInfo.tr
 	// trInfo.tr is set when created (if EnableTracing is true),
 	// and cleared when the finish method is called.
 	trInfo traceInfo
 
 	statsHandler stats.Handler
-	beginTime    time.Time
+}
+
+func (cs *clientStream) commitAttemptLocked() {
+	cs.committed = true
+	cs.buffer = nil
+}
+
+func (cs *clientStream) commitAttempt() {
+	cs.mu.Lock()
+	cs.commitAttemptLocked()
+	cs.mu.Unlock()
+}
+
+// shouldRetry returns nil if the RPC should be retried; otherwise it returns
+// the error that should be returned by the operation.
+func (cs *clientStream) shouldRetry(err error) error {
+	if cs.attempt.s == nil && !cs.callInfo.failFast {
+		// In the event of any error from NewStream (attempt.s == nil), we
+		// never attempted to write anything to the wire, so we can retry
+		// indefinitely for non-fail-fast RPCs.
+		return nil
+	}
+	if cs.finished || cs.committed {
+		// RPC is finished or committed; cannot retry.
+		return err
+	}
+	// Wait for the trailers.
+	if cs.attempt.s != nil {
+		<-cs.attempt.s.Done()
+	}
+	if cs.firstAttempt && !cs.callInfo.failFast && (cs.attempt.s == nil || cs.attempt.s.Unprocessed()) {
+		// First attempt, wait-for-ready, stream unprocessed: transparently retry.
+		cs.firstAttempt = false
+		return nil
+	}
+	cs.firstAttempt = false
+	if cs.cc.dopts.disableRetry {
+		return err
+	}
+
+	pushback := 0
+	hasPushback := false
+	if cs.attempt.s != nil {
+		if to, toErr := cs.attempt.s.TrailersOnly(); toErr != nil {
+			// Context error; stop now.
+			return toErr
+		} else if !to {
+			return err
+		}
+
+		// TODO(retry): Move down if the spec changes to not check server pushback
+		// before considering this a failure for throttling.
+		sps := cs.attempt.s.Trailer()["grpc-retry-pushback-ms"]
+		if len(sps) == 1 {
+			var e error
+			if pushback, e = strconv.Atoi(sps[0]); e != nil || pushback < 0 {
+				grpclog.Infof("Server retry pushback specified to abort (%q).", sps[0])
+				cs.retryThrottler.throttle() // This counts as a failure for throttling.
+				return err
+			}
+			hasPushback = true
+		} else if len(sps) > 1 {
+			grpclog.Warningf("Server retry pushback specified multiple values (%q); not retrying.", sps)
+			cs.retryThrottler.throttle() // This counts as a failure for throttling.
+			return err
+		}
+	}
+
+	var code codes.Code
+	if cs.attempt.s != nil {
+		code = cs.attempt.s.Status().Code()
+	} else {
+		code = status.Convert(err).Code()
+	}
+
+	rp := cs.methodConfig.retryPolicy
+	if rp == nil || !rp.retryableStatusCodes[code] {
+		return err
+	}
+
+	// Note: the ordering here is important; we count this as a failure
+	// only if the code matched a retryable code.
+	if cs.retryThrottler.throttle() {
+		return err
+	}
+	if cs.numRetries+1 >= rp.maxAttempts {
+		return err
+	}
+
+	var dur time.Duration
+	if hasPushback {
+		dur = time.Millisecond * time.Duration(pushback)
+		cs.numRetriesSincePushback = 0
+	} else {
+		fact := math.Pow(rp.backoffMultiplier, float64(cs.numRetriesSincePushback))
+		cur := float64(rp.initialBackoff) * fact
+		if max := float64(rp.maxBackoff); cur > max {
+			cur = max
+		}
+		dur = time.Duration(grpcrand.Int63n(int64(cur)))
+		cs.numRetriesSincePushback++
+	}
+
+	// TODO(dfawley): we could eagerly fail here if dur puts us past the
+	// deadline, but unsure if it is worth doing.
+	t := time.NewTimer(dur)
+	select {
+	case <-t.C:
+		cs.numRetries++
+		return nil
+	case <-cs.ctx.Done():
+		t.Stop()
+		return status.FromContextError(cs.ctx.Err()).Err()
+	}
+}
+
+// Returns nil if a retry was performed and succeeded; error otherwise.
+func (cs *clientStream) retryLocked(lastErr error) error {
+	for {
+		cs.attempt.finish(lastErr)
+		if err := cs.shouldRetry(lastErr); err != nil {
+			cs.commitAttemptLocked()
+			return err
+		}
+		if err := cs.newAttemptLocked(nil, traceInfo{}); err != nil {
+			return err
+		}
+		if lastErr = cs.replayBufferLocked(); lastErr == nil {
+			return nil
+		}
+	}
 }
 
 func (cs *clientStream) Context() context.Context {
-	// TODO(retry): commit the current attempt (the context has peer-aware data).
-	return cs.attempt.context()
+	cs.commitAttempt()
+	// No need to lock before using attempt, since we know it is committed and
+	// cannot change.
+	return cs.attempt.s.Context()
+}
+
+func (cs *clientStream) withRetry(op func(a *csAttempt) error, onSuccess func()) error {
+	cs.mu.Lock()
+	for {
+		if cs.committed {
+			cs.mu.Unlock()
+			return op(cs.attempt)
+		}
+		a := cs.attempt
+		cs.mu.Unlock()
+		err := op(a)
+		cs.mu.Lock()
+		if a != cs.attempt {
+			// We started another attempt already.
+			continue
+		}
+		if err == io.EOF {
+			<-a.s.Done()
+		}
+		if err == nil || (err == io.EOF && a.s.Status().Code() == codes.OK) {
+			onSuccess()
+			cs.mu.Unlock()
+			return err
+		}
+		if err := cs.retryLocked(err); err != nil {
+			cs.mu.Unlock()
+			return err
+		}
+	}
 }
 
 func (cs *clientStream) Header() (metadata.MD, error) {
-	m, err := cs.attempt.header()
+	var m metadata.MD
+	err := cs.withRetry(func(a *csAttempt) error {
+		var err error
+		m, err = a.s.Header()
+		return toRPCErr(err)
+	}, cs.commitAttemptLocked)
 	if err != nil {
-		// TODO(retry): maybe retry on error or commit attempt on success.
-		err = toRPCErr(err)
 		cs.finish(err)
 	}
 	return m, err
 }
 
 func (cs *clientStream) Trailer() metadata.MD {
-	// TODO(retry): on error, maybe retry (trailers-only).
-	return cs.attempt.trailer()
+	// On RPC failure, we never need to retry, because usage requires that
+	// RecvMsg() returned a non-nil error before calling this function is valid.
+	// We would have retried earlier if necessary.
+	//
+	// Commit the attempt anyway, just in case users are not following those
+	// directions -- it will prevent races and should not meaningfully impact
+	// performance.
+	cs.commitAttempt()
+	if cs.attempt.s == nil {
+		return nil
+	}
+	return cs.attempt.s.Trailer()
+}
+
+func (cs *clientStream) replayBufferLocked() error {
+	a := cs.attempt
+	for _, f := range cs.buffer {
+		if err := f(a); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (cs *clientStream) bufferForRetryLocked(sz int, op func(a *csAttempt) error) {
+	// Note: we still will buffer if retry is disabled (for transparent retries).
+	if cs.committed {
+		return
+	}
+	cs.bufferSize += sz
+	if cs.bufferSize > cs.callInfo.maxRetryRPCBufferSize {
+		cs.commitAttemptLocked()
+		return
+	}
+	cs.buffer = append(cs.buffer, op)
 }
 
 func (cs *clientStream) SendMsg(m interface{}) (err error) {
-	// TODO(retry): buffer message for replaying if not committed.
-	return cs.attempt.sendMsg(m)
+	defer func() {
+		if err != nil && err != io.EOF {
+			// Call finish on the client stream for errors generated by this SendMsg
+			// call, as these indicate problems created by this client.  (Transport
+			// errors are converted to an io.EOF error in csAttempt.sendMsg; the real
+			// error will be returned from RecvMsg eventually in that case, or be
+			// retried.)
+			cs.finish(err)
+		}
+	}()
+	if cs.sentLast {
+		return status.Errorf(codes.Internal, "SendMsg called after CloseSend")
+	}
+	if !cs.desc.ClientStreams {
+		cs.sentLast = true
+	}
+	data, err := encode(cs.codec, m)
+	if err != nil {
+		return err
+	}
+	compData, err := compress(data, cs.cp, cs.comp)
+	if err != nil {
+		return err
+	}
+	hdr, payload := msgHeader(data, compData)
+	// TODO(dfawley): should we be checking len(data) instead?
+	if len(payload) > *cs.callInfo.maxSendMessageSize {
+		return status.Errorf(codes.ResourceExhausted, "trying to send message larger than max (%d vs. %d)", len(payload), *cs.callInfo.maxSendMessageSize)
+	}
+	op := func(a *csAttempt) error {
+		err := a.sendMsg(m, hdr, payload, data)
+		// nil out the message and uncomp when replaying; they are only needed for
+		// stats which is disabled for subsequent attempts.
+		m, data = nil, nil
+		return err
+	}
+	return cs.withRetry(op, func() { cs.bufferForRetryLocked(len(hdr)+len(payload), op) })
 }
 
-func (cs *clientStream) RecvMsg(m interface{}) (err error) {
-	// TODO(retry): maybe retry on error or commit attempt on success.
-	return cs.attempt.recvMsg(m)
+func (cs *clientStream) RecvMsg(m interface{}) error {
+	err := cs.withRetry(func(a *csAttempt) error {
+		return a.recvMsg(m)
+	}, cs.commitAttemptLocked)
+	if err != nil || !cs.desc.ServerStreams {
+		// err != nil or non-server-streaming indicates end of stream.
+		cs.finish(err)
+	}
+	return err
 }
 
 func (cs *clientStream) CloseSend() error {
-	cs.attempt.closeSend()
+	if cs.sentLast {
+		// TODO: return an error and finish the stream instead, due to API misuse?
+		return nil
+	}
+	cs.sentLast = true
+	op := func(a *csAttempt) error {
+		a.t.Write(a.s, nil, nil, &transport.Options{Last: true})
+		// Always return nil; io.EOF is the only error that might make sense
+		// instead, but there is no need to signal the client to call RecvMsg
+		// as the only use left for the stream after CloseSend is to call
+		// RecvMsg.  This also matches historical behavior.
+		return nil
+	}
+	cs.withRetry(op, func() { cs.bufferForRetryLocked(0, op) })
+	// We never returned an error here for reasons.
 	return nil
 }
 
@@ -424,7 +684,11 @@ func (cs *clientStream) finish(err error) {
 		return
 	}
 	cs.finished = true
+	cs.commitAttemptLocked()
 	cs.mu.Unlock()
+	if err == nil {
+		cs.retryThrottler.successfulRPC()
+	}
 	if channelz.IsOn() {
 		if err != nil {
 			cs.cc.incrCallsFailed()
@@ -432,46 +696,20 @@ func (cs *clientStream) finish(err error) {
 			cs.cc.incrCallsSucceeded()
 		}
 	}
-	// TODO(retry): commit current attempt if necessary.
-	cs.attempt.finish(err)
-	for _, o := range cs.opts {
-		o.after(cs.c)
+	if cs.attempt != nil {
+		cs.attempt.finish(err)
+	}
+	// after functions all rely upon having a stream.
+	if cs.attempt.s != nil {
+		for _, o := range cs.opts {
+			o.after(cs.callInfo)
+		}
 	}
 	cs.cancel()
 }
 
-func (a *csAttempt) context() context.Context {
-	return a.s.Context()
-}
-
-func (a *csAttempt) header() (metadata.MD, error) {
-	return a.s.Header()
-}
-
-func (a *csAttempt) trailer() metadata.MD {
-	return a.s.Trailer()
-}
-
-func (a *csAttempt) sendMsg(m interface{}) (err error) {
-	// TODO Investigate how to signal the stats handling party.
-	// generate error stats if err != nil && err != io.EOF?
+func (a *csAttempt) sendMsg(m interface{}, hdr, payld, data []byte) error {
 	cs := a.cs
-	defer func() {
-		// For non-client-streaming RPCs, we return nil instead of EOF on success
-		// because the generated code requires it.  finish is not called; RecvMsg()
-		// will call it with the stream's status independently.
-		if err == io.EOF && !cs.desc.ClientStreams {
-			err = nil
-		}
-		if err != nil && err != io.EOF {
-			// Call finish on the client stream for errors generated by this SendMsg
-			// call, as these indicate problems created by this client.  (Transport
-			// errors are converted to an io.EOF error below; the real error will be
-			// returned from RecvMsg eventually in that case, or be retried.)
-			cs.finish(err)
-		}
-	}()
-	// TODO: Check cs.sentLast and error if we already ended the stream.
 	if EnableTracing {
 		a.mu.Lock()
 		if a.trInfo.tr != nil {
@@ -479,44 +717,26 @@ func (a *csAttempt) sendMsg(m interface{}) (err error) {
 		}
 		a.mu.Unlock()
 	}
-	data, err := encode(cs.codec, m)
-	if err != nil {
-		return err
-	}
-	compData, err := compress(data, cs.cp, cs.comp)
-	if err != nil {
-		return err
-	}
-	hdr, payload := msgHeader(data, compData)
-	// TODO(dfawley): should we be checking len(data) instead?
-	if len(payload) > *cs.c.maxSendMessageSize {
-		return status.Errorf(codes.ResourceExhausted, "trying to send message larger than max (%d vs. %d)", len(payload), *cs.c.maxSendMessageSize)
-	}
-
-	if !cs.desc.ClientStreams {
-		cs.sentLast = true
-	}
-	err = a.t.Write(a.s, hdr, payload, &transport.Options{Last: !cs.desc.ClientStreams})
-	if err == nil {
-		if a.statsHandler != nil {
-			a.statsHandler.HandleRPC(a.ctx, outPayload(true, m, data, payload, time.Now()))
+	if err := a.t.Write(a.s, hdr, payld, &transport.Options{Last: !cs.desc.ClientStreams}); err != nil {
+		if !cs.desc.ClientStreams {
+			// For non-client-streaming RPCs, we return nil instead of EOF on error
+			// because the generated code requires it.  finish is not called; RecvMsg()
+			// will call it with the stream's status independently.
+			return nil
 		}
-		if channelz.IsOn() {
-			a.t.IncrMsgSent()
-		}
-		return nil
+		return io.EOF
 	}
-	return io.EOF
+	if a.statsHandler != nil {
+		a.statsHandler.HandleRPC(cs.ctx, outPayload(true, m, data, payld, time.Now()))
+	}
+	if channelz.IsOn() {
+		a.t.IncrMsgSent()
+	}
+	return nil
 }
 
 func (a *csAttempt) recvMsg(m interface{}) (err error) {
 	cs := a.cs
-	defer func() {
-		if err != nil || !cs.desc.ServerStreams {
-			// err != nil or non-server-streaming indicates end of stream.
-			cs.finish(err)
-		}
-	}()
 	var inPayload *stats.InPayload
 	if a.statsHandler != nil {
 		inPayload = &stats.InPayload{
@@ -539,7 +759,7 @@ func (a *csAttempt) recvMsg(m interface{}) (err error) {
 		// Only initialize this state once per stream.
 		a.decompSet = true
 	}
-	err = recv(a.p, cs.codec, a.s, a.dc, m, *cs.c.maxReceiveMessageSize, inPayload, a.decomp)
+	err = recv(a.p, cs.codec, a.s, a.dc, m, *cs.callInfo.maxReceiveMessageSize, inPayload, a.decomp)
 	if err != nil {
 		if err == io.EOF {
 			if statusErr := a.s.Status().Err(); statusErr != nil {
@@ -557,7 +777,7 @@ func (a *csAttempt) recvMsg(m interface{}) (err error) {
 		a.mu.Unlock()
 	}
 	if inPayload != nil {
-		a.statsHandler.HandleRPC(a.ctx, inPayload)
+		a.statsHandler.HandleRPC(cs.ctx, inPayload)
 	}
 	if channelz.IsOn() {
 		a.t.IncrMsgRecv()
@@ -569,7 +789,7 @@ func (a *csAttempt) recvMsg(m interface{}) (err error) {
 
 	// Special handling for non-server-stream rpcs.
 	// This recv expects EOF or errors, so we don't collect inPayload.
-	err = recv(a.p, cs.codec, a.s, a.dc, m, *cs.c.maxReceiveMessageSize, nil, a.decomp)
+	err = recv(a.p, cs.codec, a.s, a.dc, m, *cs.callInfo.maxReceiveMessageSize, nil, a.decomp)
 	if err == nil {
 		return toRPCErr(errors.New("grpc: client streaming protocol violation: get <nil>, want <EOF>"))
 	}
@@ -579,37 +799,43 @@ func (a *csAttempt) recvMsg(m interface{}) (err error) {
 	return toRPCErr(err)
 }
 
-func (a *csAttempt) closeSend() {
-	cs := a.cs
-	if cs.sentLast {
-		return
-	}
-	cs.sentLast = true
-	cs.attempt.t.Write(cs.attempt.s, nil, nil, &transport.Options{Last: true})
-	// We ignore errors from Write.  Any error it would return would also be
-	// returned by a subsequent RecvMsg call, and the user is supposed to always
-	// finish the stream by calling RecvMsg until it returns err != nil.
-}
-
 func (a *csAttempt) finish(err error) {
 	a.mu.Lock()
-	a.t.CloseStream(a.s, err)
+	if a.finished {
+		a.mu.Unlock()
+		return
+	}
+	a.finished = true
+	if err == io.EOF {
+		// Ending a stream with EOF indicates a success.
+		err = nil
+	}
+	if a.s != nil {
+		a.t.CloseStream(a.s, err)
+	}
 
 	if a.done != nil {
+		br := false
+		var tr metadata.MD
+		if a.s != nil {
+			br = a.s.BytesReceived()
+			tr = a.s.Trailer()
+		}
 		a.done(balancer.DoneInfo{
 			Err:           err,
-			BytesSent:     true,
-			BytesReceived: a.s.BytesReceived(),
+			Trailer:       tr,
+			BytesSent:     a.s != nil,
+			BytesReceived: br,
 		})
 	}
 	if a.statsHandler != nil {
 		end := &stats.End{
 			Client:    true,
-			BeginTime: a.beginTime,
+			BeginTime: a.cs.beginTime,
 			EndTime:   time.Now(),
 			Error:     err,
 		}
-		a.statsHandler.HandleRPC(a.ctx, end)
+		a.statsHandler.HandleRPC(a.cs.ctx, end)
 	}
 	if a.trInfo.tr != nil {
 		if err == nil {
@@ -624,7 +850,10 @@ func (a *csAttempt) finish(err error) {
 	a.mu.Unlock()
 }
 
-// ServerStream defines the interface a server stream has to satisfy.
+// ServerStream defines the server-side behavior of a streaming RPC.
+//
+// All errors returned from ServerStream methods are compatible with the
+// status package.
 type ServerStream interface {
 	// SetHeader sets the header metadata. It may be called multiple times.
 	// When call multiple times, all the provided metadata will be merged.
@@ -640,7 +869,32 @@ type ServerStream interface {
 	// SetTrailer sets the trailer metadata which will be sent with the RPC status.
 	// When called more than once, all the provided metadata will be merged.
 	SetTrailer(metadata.MD)
-	Stream
+	// Context returns the context for this stream.
+	Context() context.Context
+	// SendMsg sends a message. On error, SendMsg aborts the stream and the
+	// error is returned directly.
+	//
+	// SendMsg blocks until:
+	//   - There is sufficient flow control to schedule m with the transport, or
+	//   - The stream is done, or
+	//   - The stream breaks.
+	//
+	// SendMsg does not wait until the message is received by the client. An
+	// untimely stream closure may result in lost messages.
+	//
+	// It is safe to have a goroutine calling SendMsg and another goroutine
+	// calling RecvMsg on the same stream at the same time, but it is not safe
+	// to call SendMsg on the same stream in different goroutines.
+	SendMsg(m interface{}) error
+	// RecvMsg blocks until it receives a message into m or the stream is
+	// done. It returns io.EOF when the client has performed a CloseSend. On
+	// any non-EOF error, the stream is aborted and the error contains the
+	// RPC status.
+	//
+	// It is safe to have a goroutine calling SendMsg and another goroutine
+	// calling RecvMsg on the same stream at the same time, but it is not
+	// safe to call RecvMsg on the same stream in different goroutines.
+	RecvMsg(m interface{}) error
 }
 
 // serverStream implements a server side Stream.