summaryrefslogtreecommitdiff
path: root/vendor/google.golang.org/grpc/internal/transport/http2_client.go
diff options
context:
space:
mode:
Diffstat (limited to 'vendor/google.golang.org/grpc/internal/transport/http2_client.go')
-rw-r--r--vendor/google.golang.org/grpc/internal/transport/http2_client.go45
1 files changed, 22 insertions, 23 deletions
diff --git a/vendor/google.golang.org/grpc/internal/transport/http2_client.go b/vendor/google.golang.org/grpc/internal/transport/http2_client.go
index 326bf0848..badab8acf 100644
--- a/vendor/google.golang.org/grpc/internal/transport/http2_client.go
+++ b/vendor/google.golang.org/grpc/internal/transport/http2_client.go
@@ -330,7 +330,7 @@ func newHTTP2Client(connectCtx, ctx context.Context, addr resolver.Address, opts
readerDone: make(chan struct{}),
writerDone: make(chan struct{}),
goAway: make(chan struct{}),
- framer: newFramer(conn, writeBufSize, readBufSize, maxHeaderListSize),
+ framer: newFramer(conn, writeBufSize, readBufSize, opts.SharedWriteBuffer, maxHeaderListSize),
fc: &trInFlow{limit: uint32(icwz)},
scheme: scheme,
activeStreams: make(map[uint32]*Stream),
@@ -762,7 +762,7 @@ func (t *http2Client) NewStream(ctx context.Context, callHdr *CallHdr) (*Stream,
firstTry := true
var ch chan struct{}
transportDrainRequired := false
- checkForStreamQuota := func(it interface{}) bool {
+ checkForStreamQuota := func(it any) bool {
if t.streamQuota <= 0 { // Can go negative if server decreases it.
if firstTry {
t.waitingStreams++
@@ -800,7 +800,7 @@ func (t *http2Client) NewStream(ctx context.Context, callHdr *CallHdr) (*Stream,
return true
}
var hdrListSizeErr error
- checkForHeaderListSize := func(it interface{}) bool {
+ checkForHeaderListSize := func(it any) bool {
if t.maxSendHeaderListSize == nil {
return true
}
@@ -815,7 +815,7 @@ func (t *http2Client) NewStream(ctx context.Context, callHdr *CallHdr) (*Stream,
return true
}
for {
- success, err := t.controlBuf.executeAndPut(func(it interface{}) bool {
+ success, err := t.controlBuf.executeAndPut(func(it any) bool {
return checkForHeaderListSize(it) && checkForStreamQuota(it)
}, hdr)
if err != nil {
@@ -927,7 +927,7 @@ func (t *http2Client) closeStream(s *Stream, err error, rst bool, rstCode http2.
rst: rst,
rstCode: rstCode,
}
- addBackStreamQuota := func(interface{}) bool {
+ addBackStreamQuota := func(any) bool {
t.streamQuota++
if t.streamQuota > 0 && t.waitingStreams > 0 {
select {
@@ -1080,7 +1080,7 @@ func (t *http2Client) updateWindow(s *Stream, n uint32) {
// for the transport and the stream based on the current bdp
// estimation.
func (t *http2Client) updateFlowControl(n uint32) {
- updateIWS := func(interface{}) bool {
+ updateIWS := func(any) bool {
t.initialWindowSize = int32(n)
t.mu.Lock()
for _, s := range t.activeStreams {
@@ -1233,7 +1233,7 @@ func (t *http2Client) handleSettings(f *http2.SettingsFrame, isFirst bool) {
}
updateFuncs = append(updateFuncs, updateStreamQuota)
}
- t.controlBuf.executeAndPut(func(interface{}) bool {
+ t.controlBuf.executeAndPut(func(any) bool {
for _, f := range updateFuncs {
f()
}
@@ -1505,14 +1505,15 @@ func (t *http2Client) operateHeaders(frame *http2.MetaHeadersFrame) {
return
}
- isHeader := false
-
- // If headerChan hasn't been closed yet
- if atomic.CompareAndSwapUint32(&s.headerChanClosed, 0, 1) {
- s.headerValid = true
- if !endStream {
- // HEADERS frame block carries a Response-Headers.
- isHeader = true
+ // For headers, set them in s.header and close headerChan. For trailers or
+ // trailers-only, closeStream will set the trailers and close headerChan as
+ // needed.
+ if !endStream {
+ // If headerChan hasn't been closed yet (expected, given we checked it
+ // above, but something else could have potentially closed the whole
+ // stream).
+ if atomic.CompareAndSwapUint32(&s.headerChanClosed, 0, 1) {
+ s.headerValid = true
// These values can be set without any synchronization because
// stream goroutine will read it only after seeing a closed
// headerChan which we'll close after setting this.
@@ -1520,15 +1521,12 @@ func (t *http2Client) operateHeaders(frame *http2.MetaHeadersFrame) {
if len(mdata) > 0 {
s.header = mdata
}
- } else {
- // HEADERS frame block carries a Trailers-Only.
- s.noHeaders = true
+ close(s.headerChan)
}
- close(s.headerChan)
}
for _, sh := range t.statsHandlers {
- if isHeader {
+ if !endStream {
inHeader := &stats.InHeader{
Client: true,
WireLength: int(frame.Header().Length),
@@ -1554,9 +1552,10 @@ func (t *http2Client) operateHeaders(frame *http2.MetaHeadersFrame) {
statusGen = status.New(rawStatusCode, grpcMessage)
}
- // if client received END_STREAM from server while stream was still active, send RST_STREAM
- rst := s.getState() == streamActive
- t.closeStream(s, io.EOF, rst, http2.ErrCodeNo, statusGen, mdata, true)
+ // If client received END_STREAM from server while stream was still active,
+ // send RST_STREAM.
+ rstStream := s.getState() == streamActive
+ t.closeStream(s, io.EOF, rstStream, http2.ErrCodeNo, statusGen, mdata, true)
}
// readServerPreface reads and handles the initial settings frame from the