rebase: bump google.golang.org/grpc from 1.48.0 to 1.49.0

Bumps [google.golang.org/grpc](https://github.com/grpc/grpc-go) from 1.48.0 to 1.49.0.
- [Release notes](https://github.com/grpc/grpc-go/releases)
- [Commits](https://github.com/grpc/grpc-go/compare/v1.48.0...v1.49.0)

---
updated-dependencies:
- dependency-name: google.golang.org/grpc
  dependency-type: direct:production
  update-type: version-update:semver-minor
...

Signed-off-by: dependabot[bot] <support@github.com>
This commit is contained in:
dependabot[bot] 2022-08-29 20:15:30 +00:00 committed by mergify[bot]
parent 586a9cc8ee
commit 9a54f05e66
17 changed files with 175 additions and 145 deletions

2
go.mod
View File

@ -28,7 +28,7 @@ require (
golang.org/x/crypto v0.0.0-20220315160706-3147a52a75dd golang.org/x/crypto v0.0.0-20220315160706-3147a52a75dd
golang.org/x/net v0.0.0-20220722155237-a158d28d115b golang.org/x/net v0.0.0-20220722155237-a158d28d115b
golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f
google.golang.org/grpc v1.48.0 google.golang.org/grpc v1.49.0
google.golang.org/protobuf v1.28.0 google.golang.org/protobuf v1.28.0
k8s.io/api v0.25.0 k8s.io/api v0.25.0
k8s.io/apimachinery v0.25.0 k8s.io/apimachinery v0.25.0

4
go.sum
View File

@ -1761,8 +1761,8 @@ google.golang.org/grpc v1.41.0/go.mod h1:U3l9uK9J0sini8mHphKoXyaqDA/8VyGnDee1zzI
google.golang.org/grpc v1.44.0/go.mod h1:k+4IHHFw41K8+bbowsex27ge2rCb65oeWqe4jJ590SU= google.golang.org/grpc v1.44.0/go.mod h1:k+4IHHFw41K8+bbowsex27ge2rCb65oeWqe4jJ590SU=
google.golang.org/grpc v1.46.0/go.mod h1:vN9eftEi1UMyUsIF80+uQXhHjbXYbm0uXoFCACuMGWk= google.golang.org/grpc v1.46.0/go.mod h1:vN9eftEi1UMyUsIF80+uQXhHjbXYbm0uXoFCACuMGWk=
google.golang.org/grpc v1.47.0/go.mod h1:vN9eftEi1UMyUsIF80+uQXhHjbXYbm0uXoFCACuMGWk= google.golang.org/grpc v1.47.0/go.mod h1:vN9eftEi1UMyUsIF80+uQXhHjbXYbm0uXoFCACuMGWk=
google.golang.org/grpc v1.48.0 h1:rQOsyJ/8+ufEDJd/Gdsz7HG220Mh9HAhFHRGnIjda0w= google.golang.org/grpc v1.49.0 h1:WTLtQzmQori5FUH25Pq4WT22oCsv8USpQ+F6rqtsmxw=
google.golang.org/grpc v1.48.0/go.mod h1:vN9eftEi1UMyUsIF80+uQXhHjbXYbm0uXoFCACuMGWk= google.golang.org/grpc v1.49.0/go.mod h1:ZgQEeidpAuNRZ8iRrlBKXZQP1ghovWIVhdJRyCDK+GI=
google.golang.org/grpc/cmd/protoc-gen-go-grpc v1.1.0/go.mod h1:6Kw0yEErY5E/yWrBtf03jp27GLLJujG4z/JK95pnjjw= google.golang.org/grpc/cmd/protoc-gen-go-grpc v1.1.0/go.mod h1:6Kw0yEErY5E/yWrBtf03jp27GLLJujG4z/JK95pnjjw=
google.golang.org/protobuf v1.22.0/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU= google.golang.org/protobuf v1.22.0/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU=
google.golang.org/protobuf v1.23.0/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU= google.golang.org/protobuf v1.23.0/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU=

View File

@ -371,56 +371,3 @@ type ClientConnState struct {
// ErrBadResolverState may be returned by UpdateClientConnState to indicate a // ErrBadResolverState may be returned by UpdateClientConnState to indicate a
// problem with the provided name resolver data. // problem with the provided name resolver data.
var ErrBadResolverState = errors.New("bad resolver state") var ErrBadResolverState = errors.New("bad resolver state")
// ConnectivityStateEvaluator takes the connectivity states of multiple SubConns
// and returns one aggregated connectivity state.
//
// It's not thread safe.
type ConnectivityStateEvaluator struct {
numReady uint64 // Number of addrConns in ready state.
numConnecting uint64 // Number of addrConns in connecting state.
numTransientFailure uint64 // Number of addrConns in transient failure state.
numIdle uint64 // Number of addrConns in idle state.
}
// RecordTransition records state change happening in subConn and based on that
// it evaluates what aggregated state should be.
//
// - If at least one SubConn in Ready, the aggregated state is Ready;
// - Else if at least one SubConn in Connecting, the aggregated state is Connecting;
// - Else if at least one SubConn is TransientFailure, the aggregated state is Transient Failure;
// - Else if at least one SubConn is Idle, the aggregated state is Idle;
// - Else there are no subconns and the aggregated state is Transient Failure
//
// Shutdown is not considered.
func (cse *ConnectivityStateEvaluator) RecordTransition(oldState, newState connectivity.State) connectivity.State {
// Update counters.
for idx, state := range []connectivity.State{oldState, newState} {
updateVal := 2*uint64(idx) - 1 // -1 for oldState and +1 for new.
switch state {
case connectivity.Ready:
cse.numReady += updateVal
case connectivity.Connecting:
cse.numConnecting += updateVal
case connectivity.TransientFailure:
cse.numTransientFailure += updateVal
case connectivity.Idle:
cse.numIdle += updateVal
}
}
// Evaluate.
if cse.numReady > 0 {
return connectivity.Ready
}
if cse.numConnecting > 0 {
return connectivity.Connecting
}
if cse.numTransientFailure > 0 {
return connectivity.TransientFailure
}
if cse.numIdle > 0 {
return connectivity.Idle
}
return connectivity.TransientFailure
}

View File

@ -0,0 +1,70 @@
/*
*
* Copyright 2022 gRPC authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
package balancer
import "google.golang.org/grpc/connectivity"
// ConnectivityStateEvaluator takes the connectivity states of multiple SubConns
// and returns one aggregated connectivity state.
//
// It's not thread safe.
type ConnectivityStateEvaluator struct {
numReady uint64 // Number of addrConns in ready state.
numConnecting uint64 // Number of addrConns in connecting state.
numTransientFailure uint64 // Number of addrConns in transient failure state.
numIdle uint64 // Number of addrConns in idle state.
}
// RecordTransition records state change happening in subConn and based on that
// it evaluates what aggregated state should be.
//
// - If at least one SubConn in Ready, the aggregated state is Ready;
// - Else if at least one SubConn in Connecting, the aggregated state is Connecting;
// - Else if at least one SubConn is Idle, the aggregated state is Idle;
// - Else if at least one SubConn is TransientFailure (or there are no SubConns), the aggregated state is Transient Failure.
//
// Shutdown is not considered.
func (cse *ConnectivityStateEvaluator) RecordTransition(oldState, newState connectivity.State) connectivity.State {
// Update counters.
for idx, state := range []connectivity.State{oldState, newState} {
updateVal := 2*uint64(idx) - 1 // -1 for oldState and +1 for new.
switch state {
case connectivity.Ready:
cse.numReady += updateVal
case connectivity.Connecting:
cse.numConnecting += updateVal
case connectivity.TransientFailure:
cse.numTransientFailure += updateVal
case connectivity.Idle:
cse.numIdle += updateVal
}
}
// Evaluate.
if cse.numReady > 0 {
return connectivity.Ready
}
if cse.numConnecting > 0 {
return connectivity.Connecting
}
if cse.numIdle > 0 {
return connectivity.Idle
}
return connectivity.TransientFailure
}

View File

@ -712,8 +712,8 @@ func (cc *ClientConn) newAddrConn(addrs []resolver.Address, opts balancer.NewSub
ac.ctx, ac.cancel = context.WithCancel(cc.ctx) ac.ctx, ac.cancel = context.WithCancel(cc.ctx)
// Track ac in cc. This needs to be done before any getTransport(...) is called. // Track ac in cc. This needs to be done before any getTransport(...) is called.
cc.mu.Lock() cc.mu.Lock()
defer cc.mu.Unlock()
if cc.conns == nil { if cc.conns == nil {
cc.mu.Unlock()
return nil, ErrClientConnClosing return nil, ErrClientConnClosing
} }
@ -732,7 +732,6 @@ func (cc *ClientConn) newAddrConn(addrs []resolver.Address, opts balancer.NewSub
}) })
cc.conns[ac] = struct{}{} cc.conns[ac] = struct{}{}
cc.mu.Unlock()
return ac, nil return ac, nil
} }

View File

@ -84,7 +84,7 @@ var extraDialOptions []DialOption
// EmptyDialOption does not alter the dial configuration. It can be embedded in // EmptyDialOption does not alter the dial configuration. It can be embedded in
// another structure to build custom dial options. // another structure to build custom dial options.
// //
// Experimental // # Experimental
// //
// Notice: This type is EXPERIMENTAL and may be changed or removed in a // Notice: This type is EXPERIMENTAL and may be changed or removed in a
// later release. // later release.
@ -275,7 +275,7 @@ func WithBlock() DialOption {
// the context.DeadlineExceeded error. // the context.DeadlineExceeded error.
// Implies WithBlock() // Implies WithBlock()
// //
// Experimental // # Experimental
// //
// Notice: This API is EXPERIMENTAL and may be changed or removed in a // Notice: This API is EXPERIMENTAL and may be changed or removed in a
// later release. // later release.
@ -304,7 +304,7 @@ func WithInsecure() DialOption {
// WithNoProxy returns a DialOption which disables the use of proxies for this // WithNoProxy returns a DialOption which disables the use of proxies for this
// ClientConn. This is ignored if WithDialer or WithContextDialer are used. // ClientConn. This is ignored if WithDialer or WithContextDialer are used.
// //
// Experimental // # Experimental
// //
// Notice: This API is EXPERIMENTAL and may be changed or removed in a // Notice: This API is EXPERIMENTAL and may be changed or removed in a
// later release. // later release.
@ -335,7 +335,7 @@ func WithPerRPCCredentials(creds credentials.PerRPCCredentials) DialOption {
// the ClientConn.WithCreds. This should not be used together with // the ClientConn.WithCreds. This should not be used together with
// WithTransportCredentials. // WithTransportCredentials.
// //
// Experimental // # Experimental
// //
// Notice: This API is EXPERIMENTAL and may be changed or removed in a // Notice: This API is EXPERIMENTAL and may be changed or removed in a
// later release. // later release.
@ -391,6 +391,12 @@ func WithDialer(f func(string, time.Duration) (net.Conn, error)) DialOption {
// all the RPCs and underlying network connections in this ClientConn. // all the RPCs and underlying network connections in this ClientConn.
func WithStatsHandler(h stats.Handler) DialOption { func WithStatsHandler(h stats.Handler) DialOption {
return newFuncDialOption(func(o *dialOptions) { return newFuncDialOption(func(o *dialOptions) {
if h == nil {
logger.Error("ignoring nil parameter in grpc.WithStatsHandler ClientOption")
// Do not allow a nil stats handler, which would otherwise cause
// panics.
return
}
o.copts.StatsHandlers = append(o.copts.StatsHandlers, h) o.copts.StatsHandlers = append(o.copts.StatsHandlers, h)
}) })
} }
@ -403,7 +409,7 @@ func WithStatsHandler(h stats.Handler) DialOption {
// FailOnNonTempDialError only affects the initial dial, and does not do // FailOnNonTempDialError only affects the initial dial, and does not do
// anything useful unless you are also using WithBlock(). // anything useful unless you are also using WithBlock().
// //
// Experimental // # Experimental
// //
// Notice: This API is EXPERIMENTAL and may be changed or removed in a // Notice: This API is EXPERIMENTAL and may be changed or removed in a
// later release. // later release.
@ -483,7 +489,7 @@ func WithAuthority(a string) DialOption {
// current ClientConn's parent. This function is used in nested channel creation // current ClientConn's parent. This function is used in nested channel creation
// (e.g. grpclb dial). // (e.g. grpclb dial).
// //
// Experimental // # Experimental
// //
// Notice: This API is EXPERIMENTAL and may be changed or removed in a // Notice: This API is EXPERIMENTAL and may be changed or removed in a
// later release. // later release.
@ -528,9 +534,6 @@ func WithDefaultServiceConfig(s string) DialOption {
// service config enables them. This does not impact transparent retries, which // service config enables them. This does not impact transparent retries, which
// will happen automatically if no data is written to the wire or if the RPC is // will happen automatically if no data is written to the wire or if the RPC is
// unprocessed by the remote server. // unprocessed by the remote server.
//
// Retry support is currently enabled by default, but may be disabled by
// setting the environment variable "GRPC_GO_RETRY" to "off".
func WithDisableRetry() DialOption { func WithDisableRetry() DialOption {
return newFuncDialOption(func(o *dialOptions) { return newFuncDialOption(func(o *dialOptions) {
o.disableRetry = true o.disableRetry = true
@ -548,7 +551,7 @@ func WithMaxHeaderListSize(s uint32) DialOption {
// WithDisableHealthCheck disables the LB channel health checking for all // WithDisableHealthCheck disables the LB channel health checking for all
// SubConns of this ClientConn. // SubConns of this ClientConn.
// //
// Experimental // # Experimental
// //
// Notice: This API is EXPERIMENTAL and may be changed or removed in a // Notice: This API is EXPERIMENTAL and may be changed or removed in a
// later release. // later release.
@ -595,7 +598,7 @@ func withMinConnectDeadline(f func() time.Duration) DialOption {
// resolver.Register. They will be matched against the scheme used for the // resolver.Register. They will be matched against the scheme used for the
// current Dial only, and will take precedence over the global registry. // current Dial only, and will take precedence over the global registry.
// //
// Experimental // # Experimental
// //
// Notice: This API is EXPERIMENTAL and may be changed or removed in a // Notice: This API is EXPERIMENTAL and may be changed or removed in a
// later release. // later release.

View File

@ -41,7 +41,6 @@ const (
clientSideSecuritySupportEnv = "GRPC_XDS_EXPERIMENTAL_SECURITY_SUPPORT" clientSideSecuritySupportEnv = "GRPC_XDS_EXPERIMENTAL_SECURITY_SUPPORT"
aggregateAndDNSSupportEnv = "GRPC_XDS_EXPERIMENTAL_ENABLE_AGGREGATE_AND_LOGICAL_DNS_CLUSTER" aggregateAndDNSSupportEnv = "GRPC_XDS_EXPERIMENTAL_ENABLE_AGGREGATE_AND_LOGICAL_DNS_CLUSTER"
rbacSupportEnv = "GRPC_XDS_EXPERIMENTAL_RBAC" rbacSupportEnv = "GRPC_XDS_EXPERIMENTAL_RBAC"
outlierDetectionSupportEnv = "GRPC_EXPERIMENTAL_ENABLE_OUTLIER_DETECTION"
federationEnv = "GRPC_EXPERIMENTAL_XDS_FEDERATION" federationEnv = "GRPC_EXPERIMENTAL_XDS_FEDERATION"
rlsInXDSEnv = "GRPC_EXPERIMENTAL_XDS_RLS_LB" rlsInXDSEnv = "GRPC_EXPERIMENTAL_XDS_RLS_LB"
@ -86,7 +85,7 @@ var (
// XDSOutlierDetection indicates whether outlier detection support is // XDSOutlierDetection indicates whether outlier detection support is
// enabled, which can be enabled by setting the environment variable // enabled, which can be enabled by setting the environment variable
// "GRPC_EXPERIMENTAL_ENABLE_OUTLIER_DETECTION" to "true". // "GRPC_EXPERIMENTAL_ENABLE_OUTLIER_DETECTION" to "true".
XDSOutlierDetection = strings.EqualFold(os.Getenv(outlierDetectionSupportEnv), "true") XDSOutlierDetection = false
// XDSFederation indicates whether federation support is enabled. // XDSFederation indicates whether federation support is enabled.
XDSFederation = strings.EqualFold(os.Getenv(federationEnv), "true") XDSFederation = strings.EqualFold(os.Getenv(federationEnv), "true")

View File

@ -39,6 +39,11 @@ func ParseMethod(methodName string) (service, method string, _ error) {
return methodName[:pos], methodName[pos+1:], nil return methodName[:pos], methodName[pos+1:], nil
} }
// baseContentType is the base content-type for gRPC. This is a valid
// content-type on it's own, but can also include a content-subtype such as
// "proto" as a suffix after "+" or ";". See
// https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-HTTP2.md#requests
// for more details.
const baseContentType = "application/grpc" const baseContentType = "application/grpc"
// ContentSubtype returns the content-subtype for the given content-type. The // ContentSubtype returns the content-subtype for the given content-type. The

View File

@ -78,6 +78,7 @@ type http2Client struct {
framer *framer framer *framer
// controlBuf delivers all the control related tasks (e.g., window // controlBuf delivers all the control related tasks (e.g., window
// updates, reset streams, and various settings) to the controller. // updates, reset streams, and various settings) to the controller.
// Do not access controlBuf with mu held.
controlBuf *controlBuffer controlBuf *controlBuffer
fc *trInFlow fc *trInFlow
// The scheme used: https if TLS is on, http otherwise. // The scheme used: https if TLS is on, http otherwise.
@ -109,6 +110,7 @@ type http2Client struct {
waitingStreams uint32 waitingStreams uint32
nextID uint32 nextID uint32
// Do not access controlBuf with mu held.
mu sync.Mutex // guard the following variables mu sync.Mutex // guard the following variables
state transportState state transportState
activeStreams map[uint32]*Stream activeStreams map[uint32]*Stream
@ -685,7 +687,6 @@ func (t *http2Client) NewStream(ctx context.Context, callHdr *CallHdr) (*Stream,
cleanup(err) cleanup(err)
return err return err
} }
t.activeStreams[id] = s
if channelz.IsOn() { if channelz.IsOn() {
atomic.AddInt64(&t.czData.streamsStarted, 1) atomic.AddInt64(&t.czData.streamsStarted, 1)
atomic.StoreInt64(&t.czData.lastStreamCreatedTime, time.Now().UnixNano()) atomic.StoreInt64(&t.czData.lastStreamCreatedTime, time.Now().UnixNano())
@ -719,6 +720,13 @@ func (t *http2Client) NewStream(ctx context.Context, callHdr *CallHdr) (*Stream,
t.nextID += 2 t.nextID += 2
s.id = h.streamID s.id = h.streamID
s.fc = &inFlow{limit: uint32(t.initialWindowSize)} s.fc = &inFlow{limit: uint32(t.initialWindowSize)}
t.mu.Lock()
if t.activeStreams == nil { // Can be niled from Close().
t.mu.Unlock()
return false // Don't create a stream if the transport is already closed.
}
t.activeStreams[s.id] = s
t.mu.Unlock()
if t.streamQuota > 0 && t.waitingStreams > 0 { if t.streamQuota > 0 && t.waitingStreams > 0 {
select { select {
case t.streamsQuotaAvailable <- struct{}{}: case t.streamsQuotaAvailable <- struct{}{}:
@ -744,13 +752,7 @@ func (t *http2Client) NewStream(ctx context.Context, callHdr *CallHdr) (*Stream,
} }
for { for {
success, err := t.controlBuf.executeAndPut(func(it interface{}) bool { success, err := t.controlBuf.executeAndPut(func(it interface{}) bool {
if !checkForStreamQuota(it) { return checkForHeaderListSize(it) && checkForStreamQuota(it)
return false
}
if !checkForHeaderListSize(it) {
return false
}
return true
}, hdr) }, hdr)
if err != nil { if err != nil {
// Connection closed. // Connection closed.
@ -1003,13 +1005,13 @@ func (t *http2Client) updateWindow(s *Stream, n uint32) {
// for the transport and the stream based on the current bdp // for the transport and the stream based on the current bdp
// estimation. // estimation.
func (t *http2Client) updateFlowControl(n uint32) { func (t *http2Client) updateFlowControl(n uint32) {
t.mu.Lock()
for _, s := range t.activeStreams {
s.fc.newLimit(n)
}
t.mu.Unlock()
updateIWS := func(interface{}) bool { updateIWS := func(interface{}) bool {
t.initialWindowSize = int32(n) t.initialWindowSize = int32(n)
t.mu.Lock()
for _, s := range t.activeStreams {
s.fc.newLimit(n)
}
t.mu.Unlock()
return true return true
} }
t.controlBuf.executeAndPut(updateIWS, &outgoingWindowUpdate{streamID: 0, increment: t.fc.newLimit(n)}) t.controlBuf.executeAndPut(updateIWS, &outgoingWindowUpdate{streamID: 0, increment: t.fc.newLimit(n)})
@ -1215,7 +1217,7 @@ func (t *http2Client) handleGoAway(f *http2.GoAwayFrame) {
default: default:
t.setGoAwayReason(f) t.setGoAwayReason(f)
close(t.goAway) close(t.goAway)
t.controlBuf.put(&incomingGoAway{}) defer t.controlBuf.put(&incomingGoAway{}) // Defer as t.mu is currently held.
// Notify the clientconn about the GOAWAY before we set the state to // Notify the clientconn about the GOAWAY before we set the state to
// draining, to allow the client to stop attempting to create streams // draining, to allow the client to stop attempting to create streams
// before disallowing new streams on this connection. // before disallowing new streams on this connection.

View File

@ -945,15 +945,16 @@ func (t *http2Server) streamContextErr(s *Stream) error {
// WriteHeader sends the header metadata md back to the client. // WriteHeader sends the header metadata md back to the client.
func (t *http2Server) WriteHeader(s *Stream, md metadata.MD) error { func (t *http2Server) WriteHeader(s *Stream, md metadata.MD) error {
if s.updateHeaderSent() { s.hdrMu.Lock()
return ErrIllegalHeaderWrite defer s.hdrMu.Unlock()
}
if s.getState() == streamDone { if s.getState() == streamDone {
return t.streamContextErr(s) return t.streamContextErr(s)
} }
s.hdrMu.Lock() if s.updateHeaderSent() {
return ErrIllegalHeaderWrite
}
if md.Len() > 0 { if md.Len() > 0 {
if s.header.Len() > 0 { if s.header.Len() > 0 {
s.header = metadata.Join(s.header, md) s.header = metadata.Join(s.header, md)
@ -962,10 +963,8 @@ func (t *http2Server) WriteHeader(s *Stream, md metadata.MD) error {
} }
} }
if err := t.writeHeaderLocked(s); err != nil { if err := t.writeHeaderLocked(s); err != nil {
s.hdrMu.Unlock()
return status.Convert(err).Err() return status.Convert(err).Err()
} }
s.hdrMu.Unlock()
return nil return nil
} }
@ -1013,17 +1012,19 @@ func (t *http2Server) writeHeaderLocked(s *Stream) error {
// TODO(zhaoq): Now it indicates the end of entire stream. Revisit if early // TODO(zhaoq): Now it indicates the end of entire stream. Revisit if early
// OK is adopted. // OK is adopted.
func (t *http2Server) WriteStatus(s *Stream, st *status.Status) error { func (t *http2Server) WriteStatus(s *Stream, st *status.Status) error {
s.hdrMu.Lock()
defer s.hdrMu.Unlock()
if s.getState() == streamDone { if s.getState() == streamDone {
return nil return nil
} }
s.hdrMu.Lock()
// TODO(mmukhi): Benchmark if the performance gets better if count the metadata and other header fields // TODO(mmukhi): Benchmark if the performance gets better if count the metadata and other header fields
// first and create a slice of that exact size. // first and create a slice of that exact size.
headerFields := make([]hpack.HeaderField, 0, 2) // grpc-status and grpc-message will be there if none else. headerFields := make([]hpack.HeaderField, 0, 2) // grpc-status and grpc-message will be there if none else.
if !s.updateHeaderSent() { // No headers have been sent. if !s.updateHeaderSent() { // No headers have been sent.
if len(s.header) > 0 { // Send a separate header frame. if len(s.header) > 0 { // Send a separate header frame.
if err := t.writeHeaderLocked(s); err != nil { if err := t.writeHeaderLocked(s); err != nil {
s.hdrMu.Unlock()
return err return err
} }
} else { // Send a trailer only response. } else { // Send a trailer only response.
@ -1052,7 +1053,7 @@ func (t *http2Server) WriteStatus(s *Stream, st *status.Status) error {
endStream: true, endStream: true,
onWrite: t.setResetPingStrikes, onWrite: t.setResetPingStrikes,
} }
s.hdrMu.Unlock()
success, err := t.controlBuf.execute(t.checkForHeaderListSize, trailingHeader) success, err := t.controlBuf.execute(t.checkForHeaderListSize, trailingHeader)
if !success { if !success {
if err != nil { if err != nil {

View File

@ -47,12 +47,6 @@ const (
http2MaxFrameLen = 16384 // 16KB frame http2MaxFrameLen = 16384 // 16KB frame
// http://http2.github.io/http2-spec/#SettingValues // http://http2.github.io/http2-spec/#SettingValues
http2InitHeaderTableSize = 4096 http2InitHeaderTableSize = 4096
// baseContentType is the base content-type for gRPC. This is a valid
// content-type on it's own, but can also include a content-subtype such as
// "proto" as a suffix after "+" or ";". See
// https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-HTTP2.md#requests
// for more details.
) )
var ( var (

View File

@ -190,7 +190,7 @@ type ServerOption interface {
// EmptyServerOption does not alter the server configuration. It can be embedded // EmptyServerOption does not alter the server configuration. It can be embedded
// in another structure to build custom server options. // in another structure to build custom server options.
// //
// Experimental // # Experimental
// //
// Notice: This type is EXPERIMENTAL and may be changed or removed in a // Notice: This type is EXPERIMENTAL and may be changed or removed in a
// later release. // later release.
@ -305,7 +305,7 @@ func CustomCodec(codec Codec) ServerOption {
// https://github.com/grpc/grpc-go/blob/master/Documentation/encoding.md#using-a-codec. // https://github.com/grpc/grpc-go/blob/master/Documentation/encoding.md#using-a-codec.
// Will be supported throughout 1.x. // Will be supported throughout 1.x.
// //
// Experimental // # Experimental
// //
// Notice: This API is EXPERIMENTAL and may be changed or removed in a // Notice: This API is EXPERIMENTAL and may be changed or removed in a
// later release. // later release.
@ -426,7 +426,7 @@ func ChainStreamInterceptor(interceptors ...StreamServerInterceptor) ServerOptio
// InTapHandle returns a ServerOption that sets the tap handle for all the server // InTapHandle returns a ServerOption that sets the tap handle for all the server
// transport to be created. Only one can be installed. // transport to be created. Only one can be installed.
// //
// Experimental // # Experimental
// //
// Notice: This API is EXPERIMENTAL and may be changed or removed in a // Notice: This API is EXPERIMENTAL and may be changed or removed in a
// later release. // later release.
@ -442,6 +442,12 @@ func InTapHandle(h tap.ServerInHandle) ServerOption {
// StatsHandler returns a ServerOption that sets the stats handler for the server. // StatsHandler returns a ServerOption that sets the stats handler for the server.
func StatsHandler(h stats.Handler) ServerOption { func StatsHandler(h stats.Handler) ServerOption {
return newFuncServerOption(func(o *serverOptions) { return newFuncServerOption(func(o *serverOptions) {
if h == nil {
logger.Error("ignoring nil parameter in grpc.StatsHandler ServerOption")
// Do not allow a nil stats handler, which would otherwise cause
// panics.
return
}
o.statsHandlers = append(o.statsHandlers, h) o.statsHandlers = append(o.statsHandlers, h)
}) })
} }
@ -469,7 +475,7 @@ func UnknownServiceHandler(streamHandler StreamHandler) ServerOption {
// new connections. If this is not set, the default is 120 seconds. A zero or // new connections. If this is not set, the default is 120 seconds. A zero or
// negative value will result in an immediate timeout. // negative value will result in an immediate timeout.
// //
// Experimental // # Experimental
// //
// Notice: This API is EXPERIMENTAL and may be changed or removed in a // Notice: This API is EXPERIMENTAL and may be changed or removed in a
// later release. // later release.
@ -490,7 +496,7 @@ func MaxHeaderListSize(s uint32) ServerOption {
// HeaderTableSize returns a ServerOption that sets the size of dynamic // HeaderTableSize returns a ServerOption that sets the size of dynamic
// header table for stream. // header table for stream.
// //
// Experimental // # Experimental
// //
// Notice: This API is EXPERIMENTAL and may be changed or removed in a // Notice: This API is EXPERIMENTAL and may be changed or removed in a
// later release. // later release.
@ -505,7 +511,7 @@ func HeaderTableSize(s uint32) ServerOption {
// zero (default) will disable workers and spawn a new goroutine for each // zero (default) will disable workers and spawn a new goroutine for each
// stream. // stream.
// //
// Experimental // # Experimental
// //
// Notice: This API is EXPERIMENTAL and may be changed or removed in a // Notice: This API is EXPERIMENTAL and may be changed or removed in a
// later release. // later release.
@ -898,7 +904,7 @@ func (s *Server) newHTTP2Transport(c net.Conn) transport.ServerTransport {
if err != credentials.ErrConnDispatched { if err != credentials.ErrConnDispatched {
// Don't log on ErrConnDispatched and io.EOF to prevent log spam. // Don't log on ErrConnDispatched and io.EOF to prevent log spam.
if err != io.EOF { if err != io.EOF {
channelz.Warning(logger, s.channelzID, "grpc: Server.Serve failed to create ServerTransport: ", err) channelz.Info(logger, s.channelzID, "grpc: Server.Serve failed to create ServerTransport: ", err)
} }
c.Close() c.Close()
} }
@ -956,19 +962,19 @@ var _ http.Handler = (*Server)(nil)
// To share one port (such as 443 for https) between gRPC and an // To share one port (such as 443 for https) between gRPC and an
// existing http.Handler, use a root http.Handler such as: // existing http.Handler, use a root http.Handler such as:
// //
// if r.ProtoMajor == 2 && strings.HasPrefix( // if r.ProtoMajor == 2 && strings.HasPrefix(
// r.Header.Get("Content-Type"), "application/grpc") { // r.Header.Get("Content-Type"), "application/grpc") {
// grpcServer.ServeHTTP(w, r) // grpcServer.ServeHTTP(w, r)
// } else { // } else {
// yourMux.ServeHTTP(w, r) // yourMux.ServeHTTP(w, r)
// } // }
// //
// Note that ServeHTTP uses Go's HTTP/2 server implementation which is totally // Note that ServeHTTP uses Go's HTTP/2 server implementation which is totally
// separate from grpc-go's HTTP/2 server. Performance and features may vary // separate from grpc-go's HTTP/2 server. Performance and features may vary
// between the two paths. ServeHTTP does not support some gRPC features // between the two paths. ServeHTTP does not support some gRPC features
// available through grpc-go's HTTP/2 server. // available through grpc-go's HTTP/2 server.
// //
// Experimental // # Experimental
// //
// Notice: This API is EXPERIMENTAL and may be changed or removed in a // Notice: This API is EXPERIMENTAL and may be changed or removed in a
// later release. // later release.
@ -1674,7 +1680,7 @@ type streamKey struct{}
// NewContextWithServerTransportStream creates a new context from ctx and // NewContextWithServerTransportStream creates a new context from ctx and
// attaches stream to it. // attaches stream to it.
// //
// Experimental // # Experimental
// //
// Notice: This API is EXPERIMENTAL and may be changed or removed in a // Notice: This API is EXPERIMENTAL and may be changed or removed in a
// later release. // later release.
@ -1689,7 +1695,7 @@ func NewContextWithServerTransportStream(ctx context.Context, stream ServerTrans
// //
// See also NewContextWithServerTransportStream. // See also NewContextWithServerTransportStream.
// //
// Experimental // # Experimental
// //
// Notice: This type is EXPERIMENTAL and may be changed or removed in a // Notice: This type is EXPERIMENTAL and may be changed or removed in a
// later release. // later release.
@ -1704,7 +1710,7 @@ type ServerTransportStream interface {
// ctx. Returns nil if the given context has no stream associated with it // ctx. Returns nil if the given context has no stream associated with it
// (which implies it is not an RPC invocation context). // (which implies it is not an RPC invocation context).
// //
// Experimental // # Experimental
// //
// Notice: This API is EXPERIMENTAL and may be changed or removed in a // Notice: This API is EXPERIMENTAL and may be changed or removed in a
// later release. // later release.
@ -1825,12 +1831,12 @@ func (s *Server) getCodec(contentSubtype string) baseCodec {
// When called multiple times, all the provided metadata will be merged. All // When called multiple times, all the provided metadata will be merged. All
// the metadata will be sent out when one of the following happens: // the metadata will be sent out when one of the following happens:
// //
// - grpc.SendHeader is called, or for streaming handlers, stream.SendHeader. // - grpc.SendHeader is called, or for streaming handlers, stream.SendHeader.
// - The first response message is sent. For unary handlers, this occurs when // - The first response message is sent. For unary handlers, this occurs when
// the handler returns; for streaming handlers, this can happen when stream's // the handler returns; for streaming handlers, this can happen when stream's
// SendMsg method is called. // SendMsg method is called.
// - An RPC status is sent out (error or success). This occurs when the handler // - An RPC status is sent out (error or success). This occurs when the handler
// returns. // returns.
// //
// SetHeader will fail if called after any of the events above. // SetHeader will fail if called after any of the events above.
// //

View File

@ -57,10 +57,9 @@ type lbConfig struct {
type ServiceConfig struct { type ServiceConfig struct {
serviceconfig.Config serviceconfig.Config
// LB is the load balancer the service providers recommends. The balancer // LB is the load balancer the service providers recommends. This is
// specified via grpc.WithBalancerName will override this. This is deprecated; // deprecated; lbConfigs is preferred. If lbConfig and LB are both present,
// lbConfigs is preferred. If lbConfig and LB are both present, lbConfig // lbConfig will be used.
// will be used.
LB *string LB *string
// lbConfig is the service config's load balancing configuration. If // lbConfig is the service config's load balancing configuration. If

View File

@ -140,13 +140,13 @@ type ClientStream interface {
// To ensure resources are not leaked due to the stream returned, one of the following // To ensure resources are not leaked due to the stream returned, one of the following
// actions must be performed: // actions must be performed:
// //
// 1. Call Close on the ClientConn. // 1. Call Close on the ClientConn.
// 2. Cancel the context provided. // 2. Cancel the context provided.
// 3. Call RecvMsg until a non-nil error is returned. A protobuf-generated // 3. Call RecvMsg until a non-nil error is returned. A protobuf-generated
// client-streaming RPC, for instance, might use the helper function // client-streaming RPC, for instance, might use the helper function
// CloseAndRecv (note that CloseSend does not Recv, therefore is not // CloseAndRecv (note that CloseSend does not Recv, therefore is not
// guaranteed to release all resources). // guaranteed to release all resources).
// 4. Receive a non-nil, non-io.EOF error from Header or SendMsg. // 4. Receive a non-nil, non-io.EOF error from Header or SendMsg.
// //
// If none of the above happen, a goroutine and a context will be leaked, and grpc // If none of the above happen, a goroutine and a context will be leaked, and grpc
// will not call the optionally-configured stats handler with a stats.End message. // will not call the optionally-configured stats handler with a stats.End message.
@ -303,12 +303,6 @@ func newClientStreamWithParams(ctx context.Context, desc *StreamDesc, cc *Client
} }
cs.binlog = binarylog.GetMethodLogger(method) cs.binlog = binarylog.GetMethodLogger(method)
cs.attempt, err = cs.newAttemptLocked(false /* isTransparent */)
if err != nil {
cs.finish(err)
return nil, err
}
// Pick the transport to use and create a new stream on the transport. // Pick the transport to use and create a new stream on the transport.
// Assign cs.attempt upon success. // Assign cs.attempt upon success.
op := func(a *csAttempt) error { op := func(a *csAttempt) error {
@ -704,6 +698,18 @@ func (cs *clientStream) withRetry(op func(a *csAttempt) error, onSuccess func())
// already be status errors. // already be status errors.
return toRPCErr(op(cs.attempt)) return toRPCErr(op(cs.attempt))
} }
if len(cs.buffer) == 0 {
// For the first op, which controls creation of the stream and
// assigns cs.attempt, we need to create a new attempt inline
// before executing the first op. On subsequent ops, the attempt
// is created immediately before replaying the ops.
var err error
if cs.attempt, err = cs.newAttemptLocked(false /* isTransparent */); err != nil {
cs.mu.Unlock()
cs.finish(err)
return err
}
}
a := cs.attempt a := cs.attempt
cs.mu.Unlock() cs.mu.Unlock()
err := op(a) err := op(a)

View File

@ -19,4 +19,4 @@
package grpc package grpc
// Version is the current grpc version. // Version is the current grpc version.
const Version = "1.48.0" const Version = "1.49.0"

View File

@ -147,7 +147,6 @@ grpc.NewGZIPDecompressor
grpc.RPCCompressor grpc.RPCCompressor
grpc.RPCDecompressor grpc.RPCDecompressor
grpc.ServiceConfig grpc.ServiceConfig
grpc.WithBalancerName
grpc.WithCompressor grpc.WithCompressor
grpc.WithDecompressor grpc.WithDecompressor
grpc.WithDialer grpc.WithDialer

4
vendor/modules.txt vendored
View File

@ -679,8 +679,8 @@ google.golang.org/appengine/urlfetch
google.golang.org/genproto/googleapis/api/httpbody google.golang.org/genproto/googleapis/api/httpbody
google.golang.org/genproto/googleapis/rpc/status google.golang.org/genproto/googleapis/rpc/status
google.golang.org/genproto/protobuf/field_mask google.golang.org/genproto/protobuf/field_mask
# google.golang.org/grpc v1.48.0 # google.golang.org/grpc v1.49.0
## explicit; go 1.14 ## explicit; go 1.17
google.golang.org/grpc google.golang.org/grpc
google.golang.org/grpc/attributes google.golang.org/grpc/attributes
google.golang.org/grpc/backoff google.golang.org/grpc/backoff