Skip to content
Open
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
39 changes: 23 additions & 16 deletions stream.go
Original file line number Diff line number Diff line change
Expand Up @@ -681,30 +681,17 @@ type clientStream struct {

cancel context.CancelFunc // cancels all attempts

sentLast bool // sent an end stream

receivedFirstMsg bool // set after the first message is received

methodConfig *MethodConfig

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

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

binlogs []binarylog.MethodLogger
// serverHeaderBinlogged is a boolean for whether server header has been
// logged. Server header will be logged when the first time one of those
// happens: stream.Header(), stream.Recv().
//
// It's only read and used by Recv() and Header(), so it doesn't need to be
// synchronized.
serverHeaderBinlogged bool

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?
numRetries int // exclusive of transparent retry attempt(s)
numRetriesSincePushback int // retries since pushback; to reset backoff
// attempt is the active client stream attempt.
// The only place where it is written is the newAttemptLocked method and this method never writes nil.
// So, attempt can be nil only inside newClientStream function when clientStream is first created.
Expand All @@ -714,13 +701,33 @@ type clientStream struct {
// place where we need to check if the attempt is nil.
attempt *csAttempt
// TODO(hedging): hedging will have multiple attempts simultaneously.
committed bool // active attempt committed for retry?
onCommit func()
replayBuffer []replayOp // operations to replay on retry
replayBufferSize int // current size of replayBuffer

// Bool fields are grouped at the tail to eliminate the alignment padding
// that would otherwise follow each bool when the next field is pointer- or
// int-sized. See https://github.com/grpc/grpc-go/issues/9280 for benchmarks.
// Add new bool fields here, not inline above.

// Not guarded by mu.
sentLast bool // sent an end stream
receivedFirstMsg bool // set after the first message is received
// serverHeaderBinlogged is a boolean for whether server header has been
// logged. Server header will be logged when the first time one of those
// happens: stream.Header(), stream.Recv().
//
// It's only read and used by Recv() and Header(), so it doesn't need to be
// synchronized.
serverHeaderBinlogged bool
// nameResolutionDelay indicates if there was a delay in the name resolution.
// This field is only valid on client side, it's always false on server side.
nameResolutionDelay bool

// Guarded by mu.
firstAttempt bool // if true, transparent retry is valid
finished bool // TODO: replace with atomic cmpxchg or sync.Once?
committed bool // active attempt committed for retry?
}

type replayOp struct {
Expand Down
Loading