summaryrefslogtreecommitdiff
path: root/go/vendor/google.golang.org/grpc/clientconn.go
diff options
context:
space:
mode:
authorAsh McKenzie <amckenzie@gitlab.com>2019-08-19 05:02:05 +0000
committerAsh McKenzie <amckenzie@gitlab.com>2019-08-19 05:02:05 +0000
commit256c97066bb7838d43798b6760d3acbb82c44263 (patch)
treea11de1a5bb9a3c27bcd6c9b31c0663bd91c5c8c8 /go/vendor/google.golang.org/grpc/clientconn.go
parent81f5854513a527c3a00398e91930a12ebad70272 (diff)
parent4d330cb1dd0fa2d6470d459fc33a987e8f550887 (diff)
downloadgitlab-shell-256c97066bb7838d43798b6760d3acbb82c44263.tar.gz
Merge branch 'jv-go-mod' into 'master'
Use go mod See merge request gitlab-org/gitlab-shell!323
Diffstat (limited to 'go/vendor/google.golang.org/grpc/clientconn.go')
-rw-r--r--go/vendor/google.golang.org/grpc/clientconn.go1398
1 files changed, 0 insertions, 1398 deletions
diff --git a/go/vendor/google.golang.org/grpc/clientconn.go b/go/vendor/google.golang.org/grpc/clientconn.go
deleted file mode 100644
index f49ac3f..0000000
--- a/go/vendor/google.golang.org/grpc/clientconn.go
+++ /dev/null
@@ -1,1398 +0,0 @@
-/*
- *
- * Copyright 2014 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 grpc
-
-import (
- "errors"
- "fmt"
- "math"
- "net"
- "reflect"
- "strings"
- "sync"
- "sync/atomic"
- "time"
-
- "golang.org/x/net/context"
- "google.golang.org/grpc/balancer"
- _ "google.golang.org/grpc/balancer/roundrobin" // To register roundrobin.
- "google.golang.org/grpc/codes"
- "google.golang.org/grpc/connectivity"
- "google.golang.org/grpc/credentials"
- "google.golang.org/grpc/grpclog"
- "google.golang.org/grpc/internal/backoff"
- "google.golang.org/grpc/internal/channelz"
- "google.golang.org/grpc/internal/transport"
- "google.golang.org/grpc/keepalive"
- "google.golang.org/grpc/metadata"
- "google.golang.org/grpc/resolver"
- _ "google.golang.org/grpc/resolver/dns" // To register dns resolver.
- _ "google.golang.org/grpc/resolver/passthrough" // To register passthrough resolver.
- "google.golang.org/grpc/status"
-)
-
-const (
- // minimum time to give a connection to complete
- minConnectTimeout = 20 * time.Second
- // must match grpclbName in grpclb/grpclb.go
- grpclbName = "grpclb"
-)
-
-var (
- // ErrClientConnClosing indicates that the operation is illegal because
- // the ClientConn is closing.
- //
- // Deprecated: this error should not be relied upon by users; use the status
- // code of Canceled instead.
- ErrClientConnClosing = status.Error(codes.Canceled, "grpc: the client connection is closing")
- // errConnDrain indicates that the connection starts to be drained and does not accept any new RPCs.
- errConnDrain = errors.New("grpc: the connection is drained")
- // errConnClosing indicates that the connection is closing.
- errConnClosing = errors.New("grpc: the connection is closing")
- // errBalancerClosed indicates that the balancer is closed.
- errBalancerClosed = errors.New("grpc: balancer is closed")
- // We use an accessor so that minConnectTimeout can be
- // atomically read and updated while testing.
- getMinConnectTimeout = func() time.Duration {
- return minConnectTimeout
- }
-)
-
-// The following errors are returned from Dial and DialContext
-var (
- // errNoTransportSecurity indicates that there is no transport security
- // being set for ClientConn. Users should either set one or explicitly
- // call WithInsecure DialOption to disable security.
- errNoTransportSecurity = errors.New("grpc: no transport security set (use grpc.WithInsecure() explicitly or set credentials)")
- // errTransportCredsAndBundle indicates that creds bundle is used together
- // with other individual Transport Credentials.
- errTransportCredsAndBundle = errors.New("grpc: credentials.Bundle may not be used with individual TransportCredentials")
- // errTransportCredentialsMissing indicates that users want to transmit security
- // information (e.g., oauth2 token) which requires secure connection on an insecure
- // connection.
- errTransportCredentialsMissing = errors.New("grpc: the credentials require transport level security (use grpc.WithTransportCredentials() to set)")
- // errCredentialsConflict indicates that grpc.WithTransportCredentials()
- // and grpc.WithInsecure() are both called for a connection.
- errCredentialsConflict = errors.New("grpc: transport credentials are set for an insecure connection (grpc.WithTransportCredentials() and grpc.WithInsecure() are both called)")
-)
-
-const (
- defaultClientMaxReceiveMessageSize = 1024 * 1024 * 4
- defaultClientMaxSendMessageSize = math.MaxInt32
- // http2IOBufSize specifies the buffer size for sending frames.
- defaultWriteBufSize = 32 * 1024
- defaultReadBufSize = 32 * 1024
-)
-
-// Dial creates a client connection to the given target.
-func Dial(target string, opts ...DialOption) (*ClientConn, error) {
- return DialContext(context.Background(), target, opts...)
-}
-
-// DialContext creates a client connection to the given target. By default, it's
-// a non-blocking dial (the function won't wait for connections to be
-// established, and connecting happens in the background). To make it a blocking
-// dial, use WithBlock() dial option.
-//
-// In the non-blocking case, the ctx does not act against the connection. It
-// only controls the setup steps.
-//
-// In the blocking case, ctx can be used to cancel or expire the pending
-// connection. Once this function returns, the cancellation and expiration of
-// ctx will be noop. Users should call ClientConn.Close to terminate all the
-// pending operations after this function returns.
-//
-// The target name syntax is defined in
-// https://github.com/grpc/grpc/blob/master/doc/naming.md.
-// e.g. to use dns resolver, a "dns:///" prefix should be applied to the target.
-func DialContext(ctx context.Context, target string, opts ...DialOption) (conn *ClientConn, err error) {
- cc := &ClientConn{
- target: target,
- csMgr: &connectivityStateManager{},
- conns: make(map[*addrConn]struct{}),
- dopts: defaultDialOptions(),
- blockingpicker: newPickerWrapper(),
- czData: new(channelzData),
- }
- cc.retryThrottler.Store((*retryThrottler)(nil))
- cc.ctx, cc.cancel = context.WithCancel(context.Background())
-
- for _, opt := range opts {
- opt.apply(&cc.dopts)
- }
-
- if channelz.IsOn() {
- if cc.dopts.channelzParentID != 0 {
- cc.channelzID = channelz.RegisterChannel(&channelzChannel{cc}, cc.dopts.channelzParentID, target)
- channelz.AddTraceEvent(cc.channelzID, &channelz.TraceEventDesc{
- Desc: "Channel Created",
- Severity: channelz.CtINFO,
- Parent: &channelz.TraceEventDesc{
- Desc: fmt.Sprintf("Nested Channel(id:%d) created", cc.channelzID),
- Severity: channelz.CtINFO,
- },
- })
- } else {
- cc.channelzID = channelz.RegisterChannel(&channelzChannel{cc}, 0, target)
- channelz.AddTraceEvent(cc.channelzID, &channelz.TraceEventDesc{
- Desc: "Channel Created",
- Severity: channelz.CtINFO,
- })
- }
- cc.csMgr.channelzID = cc.channelzID
- }
-
- if !cc.dopts.insecure {
- if cc.dopts.copts.TransportCredentials == nil && cc.dopts.copts.CredsBundle == nil {
- return nil, errNoTransportSecurity
- }
- if cc.dopts.copts.TransportCredentials != nil && cc.dopts.copts.CredsBundle != nil {
- return nil, errTransportCredsAndBundle
- }
- } else {
- if cc.dopts.copts.TransportCredentials != nil || cc.dopts.copts.CredsBundle != nil {
- return nil, errCredentialsConflict
- }
- for _, cd := range cc.dopts.copts.PerRPCCredentials {
- if cd.RequireTransportSecurity() {
- return nil, errTransportCredentialsMissing
- }
- }
- }
-
- cc.mkp = cc.dopts.copts.KeepaliveParams
-
- if cc.dopts.copts.Dialer == nil {
- cc.dopts.copts.Dialer = newProxyDialer(
- func(ctx context.Context, addr string) (net.Conn, error) {
- network, addr := parseDialTarget(addr)
- return dialContext(ctx, network, addr)
- },
- )
- }
-
- if cc.dopts.copts.UserAgent != "" {
- cc.dopts.copts.UserAgent += " " + grpcUA
- } else {
- cc.dopts.copts.UserAgent = grpcUA
- }
-
- if cc.dopts.timeout > 0 {
- var cancel context.CancelFunc
- ctx, cancel = context.WithTimeout(ctx, cc.dopts.timeout)
- defer cancel()
- }
-
- defer func() {
- select {
- case <-ctx.Done():
- conn, err = nil, ctx.Err()
- default:
- }
-
- if err != nil {
- cc.Close()
- }
- }()
-
- scSet := false
- if cc.dopts.scChan != nil {
- // Try to get an initial service config.
- select {
- case sc, ok := <-cc.dopts.scChan:
- if ok {
- cc.sc = sc
- scSet = true
- }
- default:
- }
- }
- if cc.dopts.bs == nil {
- cc.dopts.bs = backoff.Exponential{
- MaxDelay: DefaultBackoffConfig.MaxDelay,
- }
- }
- if cc.dopts.resolverBuilder == nil {
- // Only try to parse target when resolver builder is not already set.
- cc.parsedTarget = parseTarget(cc.target)
- grpclog.Infof("parsed scheme: %q", cc.parsedTarget.Scheme)
- cc.dopts.resolverBuilder = resolver.Get(cc.parsedTarget.Scheme)
- if cc.dopts.resolverBuilder == nil {
- // If resolver builder is still nil, the parse target's scheme is
- // not registered. Fallback to default resolver and set Endpoint to
- // the original unparsed target.
- grpclog.Infof("scheme %q not registered, fallback to default scheme", cc.parsedTarget.Scheme)
- cc.parsedTarget = resolver.Target{
- Scheme: resolver.GetDefaultScheme(),
- Endpoint: target,
- }
- cc.dopts.resolverBuilder = resolver.Get(cc.parsedTarget.Scheme)
- }
- } else {
- cc.parsedTarget = resolver.Target{Endpoint: target}
- }
- creds := cc.dopts.copts.TransportCredentials
- if creds != nil && creds.Info().ServerName != "" {
- cc.authority = creds.Info().ServerName
- } else if cc.dopts.insecure && cc.dopts.authority != "" {
- cc.authority = cc.dopts.authority
- } else {
- // Use endpoint from "scheme://authority/endpoint" as the default
- // authority for ClientConn.
- cc.authority = cc.parsedTarget.Endpoint
- }
-
- if cc.dopts.scChan != nil && !scSet {
- // Blocking wait for the initial service config.
- select {
- case sc, ok := <-cc.dopts.scChan:
- if ok {
- cc.sc = sc
- }
- case <-ctx.Done():
- return nil, ctx.Err()
- }
- }
- if cc.dopts.scChan != nil {
- go cc.scWatcher()
- }
-
- var credsClone credentials.TransportCredentials
- if creds := cc.dopts.copts.TransportCredentials; creds != nil {
- credsClone = creds.Clone()
- }
- cc.balancerBuildOpts = balancer.BuildOptions{
- DialCreds: credsClone,
- CredsBundle: cc.dopts.copts.CredsBundle,
- Dialer: cc.dopts.copts.Dialer,
- ChannelzParentID: cc.channelzID,
- }
-
- // Build the resolver.
- cc.resolverWrapper, err = newCCResolverWrapper(cc)
- if err != nil {
- return nil, fmt.Errorf("failed to build resolver: %v", err)
- }
- // Start the resolver wrapper goroutine after resolverWrapper is created.
- //
- // If the goroutine is started before resolverWrapper is ready, the
- // following may happen: The goroutine sends updates to cc. cc forwards
- // those to balancer. Balancer creates new addrConn. addrConn fails to
- // connect, and calls resolveNow(). resolveNow() tries to use the non-ready
- // resolverWrapper.
- cc.resolverWrapper.start()
-
- // A blocking dial blocks until the clientConn is ready.
- if cc.dopts.block {
- for {
- s := cc.GetState()
- if s == connectivity.Ready {
- break
- } else if cc.dopts.copts.FailOnNonTempDialError && s == connectivity.TransientFailure {
- if err = cc.blockingpicker.connectionError(); err != nil {
- terr, ok := err.(interface{ Temporary() bool })
- if ok && !terr.Temporary() {
- return nil, err
- }
- }
- }
- if !cc.WaitForStateChange(ctx, s) {
- // ctx got timeout or canceled.
- return nil, ctx.Err()
- }
- }
- }
-
- return cc, nil
-}
-
-// connectivityStateManager keeps the connectivity.State of ClientConn.
-// This struct will eventually be exported so the balancers can access it.
-type connectivityStateManager struct {
- mu sync.Mutex
- state connectivity.State
- notifyChan chan struct{}
- channelzID int64
-}
-
-// updateState updates the connectivity.State of ClientConn.
-// If there's a change it notifies goroutines waiting on state change to
-// happen.
-func (csm *connectivityStateManager) updateState(state connectivity.State) {
- csm.mu.Lock()
- defer csm.mu.Unlock()
- if csm.state == connectivity.Shutdown {
- return
- }
- if csm.state == state {
- return
- }
- csm.state = state
- if channelz.IsOn() {
- channelz.AddTraceEvent(csm.channelzID, &channelz.TraceEventDesc{
- Desc: fmt.Sprintf("Channel Connectivity change to %v", state),
- Severity: channelz.CtINFO,
- })
- }
- if csm.notifyChan != nil {
- // There are other goroutines waiting on this channel.
- close(csm.notifyChan)
- csm.notifyChan = nil
- }
-}
-
-func (csm *connectivityStateManager) getState() connectivity.State {
- csm.mu.Lock()
- defer csm.mu.Unlock()
- return csm.state
-}
-
-func (csm *connectivityStateManager) getNotifyChan() <-chan struct{} {
- csm.mu.Lock()
- defer csm.mu.Unlock()
- if csm.notifyChan == nil {
- csm.notifyChan = make(chan struct{})
- }
- return csm.notifyChan
-}
-
-// ClientConn represents a client connection to an RPC server.
-type ClientConn struct {
- ctx context.Context
- cancel context.CancelFunc
-
- target string
- parsedTarget resolver.Target
- authority string
- dopts dialOptions
- csMgr *connectivityStateManager
-
- balancerBuildOpts balancer.BuildOptions
- resolverWrapper *ccResolverWrapper
- blockingpicker *pickerWrapper
-
- mu sync.RWMutex
- sc ServiceConfig
- scRaw string
- conns map[*addrConn]struct{}
- // Keepalive parameter can be updated if a GoAway is received.
- mkp keepalive.ClientParameters
- curBalancerName string
- preBalancerName string // previous balancer name.
- curAddresses []resolver.Address
- balancerWrapper *ccBalancerWrapper
- retryThrottler atomic.Value
-
- channelzID int64 // channelz unique identification number
- czData *channelzData
-}
-
-// WaitForStateChange waits until the connectivity.State of ClientConn changes from sourceState or
-// ctx expires. A true value is returned in former case and false in latter.
-// This is an EXPERIMENTAL API.
-func (cc *ClientConn) WaitForStateChange(ctx context.Context, sourceState connectivity.State) bool {
- ch := cc.csMgr.getNotifyChan()
- if cc.csMgr.getState() != sourceState {
- return true
- }
- select {
- case <-ctx.Done():
- return false
- case <-ch:
- return true
- }
-}
-
-// GetState returns the connectivity.State of ClientConn.
-// This is an EXPERIMENTAL API.
-func (cc *ClientConn) GetState() connectivity.State {
- return cc.csMgr.getState()
-}
-
-func (cc *ClientConn) scWatcher() {
- for {
- select {
- case sc, ok := <-cc.dopts.scChan:
- if !ok {
- return
- }
- cc.mu.Lock()
- // TODO: load balance policy runtime change is ignored.
- // We may revist this decision in the future.
- cc.sc = sc
- cc.scRaw = ""
- cc.mu.Unlock()
- case <-cc.ctx.Done():
- return
- }
- }
-}
-
-func (cc *ClientConn) handleResolvedAddrs(addrs []resolver.Address, err error) {
- cc.mu.Lock()
- defer cc.mu.Unlock()
- if cc.conns == nil {
- // cc was closed.
- return
- }
-
- if reflect.DeepEqual(cc.curAddresses, addrs) {
- return
- }
-
- cc.curAddresses = addrs
-
- if cc.dopts.balancerBuilder == nil {
- // Only look at balancer types and switch balancer if balancer dial
- // option is not set.
- var isGRPCLB bool
- for _, a := range addrs {
- if a.Type == resolver.GRPCLB {
- isGRPCLB = true
- break
- }
- }
- var newBalancerName string
- if isGRPCLB {
- newBalancerName = grpclbName
- } else {
- // Address list doesn't contain grpclb address. Try to pick a
- // non-grpclb balancer.
- newBalancerName = cc.curBalancerName
- // If current balancer is grpclb, switch to the previous one.
- if newBalancerName == grpclbName {
- newBalancerName = cc.preBalancerName
- }
- // The following could be true in two cases:
- // - the first time handling resolved addresses
- // (curBalancerName="")
- // - the first time handling non-grpclb addresses
- // (curBalancerName="grpclb", preBalancerName="")
- if newBalancerName == "" {
- newBalancerName = PickFirstBalancerName
- }
- }
- cc.switchBalancer(newBalancerName)
- } else if cc.balancerWrapper == nil {
- // Balancer dial option was set, and this is the first time handling
- // resolved addresses. Build a balancer with dopts.balancerBuilder.
- cc.balancerWrapper = newCCBalancerWrapper(cc, cc.dopts.balancerBuilder, cc.balancerBuildOpts)
- }
-
- cc.balancerWrapper.handleResolvedAddrs(addrs, nil)
-}
-
-// switchBalancer starts the switching from current balancer to the balancer
-// with the given name.
-//
-// It will NOT send the current address list to the new balancer. If needed,
-// caller of this function should send address list to the new balancer after
-// this function returns.
-//
-// Caller must hold cc.mu.
-func (cc *ClientConn) switchBalancer(name string) {
- if cc.conns == nil {
- return
- }
-
- if strings.ToLower(cc.curBalancerName) == strings.ToLower(name) {
- return
- }
-
- grpclog.Infof("ClientConn switching balancer to %q", name)
- if cc.dopts.balancerBuilder != nil {
- grpclog.Infoln("ignoring balancer switching: Balancer DialOption used instead")
- return
- }
- // TODO(bar switching) change this to two steps: drain and close.
- // Keep track of sc in wrapper.
- if cc.balancerWrapper != nil {
- cc.balancerWrapper.close()
- }
-
- builder := balancer.Get(name)
- // TODO(yuxuanli): If user send a service config that does not contain a valid balancer name, should
- // we reuse previous one?
- if channelz.IsOn() {
- if builder == nil {
- channelz.AddTraceEvent(cc.channelzID, &channelz.TraceEventDesc{
- Desc: fmt.Sprintf("Channel switches to new LB policy %q due to fallback from invalid balancer name", PickFirstBalancerName),
- Severity: channelz.CtWarning,
- })
- } else {
- channelz.AddTraceEvent(cc.channelzID, &channelz.TraceEventDesc{
- Desc: fmt.Sprintf("Channel switches to new LB policy %q", name),
- Severity: channelz.CtINFO,
- })
- }
- }
- if builder == nil {
- grpclog.Infof("failed to get balancer builder for: %v, using pick_first instead", name)
- builder = newPickfirstBuilder()
- }
-
- cc.preBalancerName = cc.curBalancerName
- cc.curBalancerName = builder.Name()
- cc.balancerWrapper = newCCBalancerWrapper(cc, builder, cc.balancerBuildOpts)
-}
-
-func (cc *ClientConn) handleSubConnStateChange(sc balancer.SubConn, s connectivity.State) {
- cc.mu.Lock()
- if cc.conns == nil {
- cc.mu.Unlock()
- return
- }
- // TODO(bar switching) send updates to all balancer wrappers when balancer
- // gracefully switching is supported.
- cc.balancerWrapper.handleSubConnStateChange(sc, s)
- cc.mu.Unlock()
-}
-
-// newAddrConn creates an addrConn for addrs and adds it to cc.conns.
-//
-// Caller needs to make sure len(addrs) > 0.
-func (cc *ClientConn) newAddrConn(addrs []resolver.Address, opts balancer.NewSubConnOptions) (*addrConn, error) {
- ac := &addrConn{
- cc: cc,
- addrs: addrs,
- scopts: opts,
- dopts: cc.dopts,
- czData: new(channelzData),
- successfulHandshake: true, // make the first nextAddr() call _not_ move addrIdx up by 1
- resetBackoff: make(chan struct{}),
- }
- ac.ctx, ac.cancel = context.WithCancel(cc.ctx)
- // Track ac in cc. This needs to be done before any getTransport(...) is called.
- cc.mu.Lock()
- if cc.conns == nil {
- cc.mu.Unlock()
- return nil, ErrClientConnClosing
- }
- if channelz.IsOn() {
- ac.channelzID = channelz.RegisterSubChannel(ac, cc.channelzID, "")
- channelz.AddTraceEvent(ac.channelzID, &channelz.TraceEventDesc{
- Desc: "Subchannel Created",
- Severity: channelz.CtINFO,
- Parent: &channelz.TraceEventDesc{
- Desc: fmt.Sprintf("Subchannel(id:%d) created", ac.channelzID),
- Severity: channelz.CtINFO,
- },
- })
- }
- cc.conns[ac] = struct{}{}
- cc.mu.Unlock()
- return ac, nil
-}
-
-// removeAddrConn removes the addrConn in the subConn from clientConn.
-// It also tears down the ac with the given error.
-func (cc *ClientConn) removeAddrConn(ac *addrConn, err error) {
- cc.mu.Lock()
- if cc.conns == nil {
- cc.mu.Unlock()
- return
- }
- delete(cc.conns, ac)
- cc.mu.Unlock()
- ac.tearDown(err)
-}
-
-func (cc *ClientConn) channelzMetric() *channelz.ChannelInternalMetric {
- return &channelz.ChannelInternalMetric{
- State: cc.GetState(),
- Target: cc.target,
- CallsStarted: atomic.LoadInt64(&cc.czData.callsStarted),
- CallsSucceeded: atomic.LoadInt64(&cc.czData.callsSucceeded),
- CallsFailed: atomic.LoadInt64(&cc.czData.callsFailed),
- LastCallStartedTimestamp: time.Unix(0, atomic.LoadInt64(&cc.czData.lastCallStartedTime)),
- }
-}
-
-// Target returns the target string of the ClientConn.
-// This is an EXPERIMENTAL API.
-func (cc *ClientConn) Target() string {
- return cc.target
-}
-
-func (cc *ClientConn) incrCallsStarted() {
- atomic.AddInt64(&cc.czData.callsStarted, 1)
- atomic.StoreInt64(&cc.czData.lastCallStartedTime, time.Now().UnixNano())
-}
-
-func (cc *ClientConn) incrCallsSucceeded() {
- atomic.AddInt64(&cc.czData.callsSucceeded, 1)
-}
-
-func (cc *ClientConn) incrCallsFailed() {
- atomic.AddInt64(&cc.czData.callsFailed, 1)
-}
-
-// connect starts creating a transport.
-// It does nothing if the ac is not IDLE.
-// TODO(bar) Move this to the addrConn section.
-func (ac *addrConn) connect() error {
- ac.mu.Lock()
- if ac.state == connectivity.Shutdown {
- ac.mu.Unlock()
- return errConnClosing
- }
- if ac.state != connectivity.Idle {
- ac.mu.Unlock()
- return nil
- }
- ac.updateConnectivityState(connectivity.Connecting)
- ac.cc.handleSubConnStateChange(ac.acbw, ac.state)
- ac.mu.Unlock()
-
- // Start a goroutine connecting to the server asynchronously.
- go ac.resetTransport(false)
- return nil
-}
-
-// tryUpdateAddrs tries to update ac.addrs with the new addresses list.
-//
-// It checks whether current connected address of ac is in the new addrs list.
-// - If true, it updates ac.addrs and returns true. The ac will keep using
-// the existing connection.
-// - If false, it does nothing and returns false.
-func (ac *addrConn) tryUpdateAddrs(addrs []resolver.Address) bool {
- ac.mu.Lock()
- defer ac.mu.Unlock()
- grpclog.Infof("addrConn: tryUpdateAddrs curAddr: %v, addrs: %v", ac.curAddr, addrs)
- if ac.state == connectivity.Shutdown {
- ac.addrs = addrs
- return true
- }
-
- var curAddrFound bool
- for _, a := range addrs {
- if reflect.DeepEqual(ac.curAddr, a) {
- curAddrFound = true
- break
- }
- }
- grpclog.Infof("addrConn: tryUpdateAddrs curAddrFound: %v", curAddrFound)
- if curAddrFound {
- ac.addrs = addrs
- ac.addrIdx = 0 // Start reconnecting from beginning in the new list.
- }
-
- return curAddrFound
-}
-
-// GetMethodConfig gets the method config of the input method.
-// If there's an exact match for input method (i.e. /service/method), we return
-// the corresponding MethodConfig.
-// If there isn't an exact match for the input method, we look for the default config
-// under the service (i.e /service/). If there is a default MethodConfig for
-// the service, we return it.
-// Otherwise, we return an empty MethodConfig.
-func (cc *ClientConn) GetMethodConfig(method string) MethodConfig {
- // TODO: Avoid the locking here.
- cc.mu.RLock()
- defer cc.mu.RUnlock()
- m, ok := cc.sc.Methods[method]
- if !ok {
- i := strings.LastIndex(method, "/")
- m = cc.sc.Methods[method[:i+1]]
- }
- return m
-}
-
-func (cc *ClientConn) getTransport(ctx context.Context, failfast bool, method string) (transport.ClientTransport, func(balancer.DoneInfo), error) {
- hdr, _ := metadata.FromOutgoingContext(ctx)
- t, done, err := cc.blockingpicker.pick(ctx, failfast, balancer.PickOptions{
- FullMethodName: method,
- Header: hdr,
- })
- if err != nil {
- return nil, nil, toRPCErr(err)
- }
- return t, done, nil
-}
-
-// handleServiceConfig parses the service config string in JSON format to Go native
-// struct ServiceConfig, and store both the struct and the JSON string in ClientConn.
-func (cc *ClientConn) handleServiceConfig(js string) error {
- if cc.dopts.disableServiceConfig {
- return nil
- }
- if cc.scRaw == js {
- return nil
- }
- if channelz.IsOn() {
- channelz.AddTraceEvent(cc.channelzID, &channelz.TraceEventDesc{
- // The special formatting of \"%s\" instead of %q is to provide nice printing of service config
- // for human consumption.
- Desc: fmt.Sprintf("Channel has a new service config \"%s\"", js),
- Severity: channelz.CtINFO,
- })
- }
- sc, err := parseServiceConfig(js)
- if err != nil {
- return err
- }
- cc.mu.Lock()
- // Check if the ClientConn is already closed. Some fields (e.g.
- // balancerWrapper) are set to nil when closing the ClientConn, and could
- // cause nil pointer panic if we don't have this check.
- if cc.conns == nil {
- cc.mu.Unlock()
- return nil
- }
- cc.scRaw = js
- cc.sc = sc
-
- if sc.retryThrottling != nil {
- newThrottler := &retryThrottler{
- tokens: sc.retryThrottling.MaxTokens,
- max: sc.retryThrottling.MaxTokens,
- thresh: sc.retryThrottling.MaxTokens / 2,
- ratio: sc.retryThrottling.TokenRatio,
- }
- cc.retryThrottler.Store(newThrottler)
- } else {
- cc.retryThrottler.Store((*retryThrottler)(nil))
- }
-
- if sc.LB != nil && *sc.LB != grpclbName { // "grpclb" is not a valid balancer option in service config.
- if cc.curBalancerName == grpclbName {
- // If current balancer is grpclb, there's at least one grpclb
- // balancer address in the resolved list. Don't switch the balancer,
- // but change the previous balancer name, so if a new resolved
- // address list doesn't contain grpclb address, balancer will be
- // switched to *sc.LB.
- cc.preBalancerName = *sc.LB
- } else {
- cc.switchBalancer(*sc.LB)
- cc.balancerWrapper.handleResolvedAddrs(cc.curAddresses, nil)
- }
- }
-
- cc.mu.Unlock()
- return nil
-}
-
-func (cc *ClientConn) resolveNow(o resolver.ResolveNowOption) {
- cc.mu.RLock()
- r := cc.resolverWrapper
- cc.mu.RUnlock()
- if r == nil {
- return
- }
- go r.resolveNow(o)
-}
-
-// ResetConnectBackoff wakes up all subchannels in transient failure and causes
-// them to attempt another connection immediately. It also resets the backoff
-// times used for subsequent attempts regardless of the current state.
-//
-// In general, this function should not be used. Typical service or network
-// outages result in a reasonable client reconnection strategy by default.
-// However, if a previously unavailable network becomes available, this may be
-// used to trigger an immediate reconnect.
-//
-// This API is EXPERIMENTAL.
-func (cc *ClientConn) ResetConnectBackoff() {
- cc.mu.Lock()
- defer cc.mu.Unlock()
- for ac := range cc.conns {
- ac.resetConnectBackoff()
- }
-}
-
-// Close tears down the ClientConn and all underlying connections.
-func (cc *ClientConn) Close() error {
- defer cc.cancel()
-
- cc.mu.Lock()
- if cc.conns == nil {
- cc.mu.Unlock()
- return ErrClientConnClosing
- }
- conns := cc.conns
- cc.conns = nil
- cc.csMgr.updateState(connectivity.Shutdown)
-
- rWrapper := cc.resolverWrapper
- cc.resolverWrapper = nil
- bWrapper := cc.balancerWrapper
- cc.balancerWrapper = nil
- cc.mu.Unlock()
-
- cc.blockingpicker.close()
-
- if rWrapper != nil {
- rWrapper.close()
- }
- if bWrapper != nil {
- bWrapper.close()
- }
-
- for ac := range conns {
- ac.tearDown(ErrClientConnClosing)
- }
- if channelz.IsOn() {
- ted := &channelz.TraceEventDesc{
- Desc: "Channel Deleted",
- Severity: channelz.CtINFO,
- }
- if cc.dopts.channelzParentID != 0 {
- ted.Parent = &channelz.TraceEventDesc{
- Desc: fmt.Sprintf("Nested channel(id:%d) deleted", cc.channelzID),
- Severity: channelz.CtINFO,
- }
- }
- channelz.AddTraceEvent(cc.channelzID, ted)
- // TraceEvent needs to be called before RemoveEntry, as TraceEvent may add trace reference to
- // the entity beng deleted, and thus prevent it from being deleted right away.
- channelz.RemoveEntry(cc.channelzID)
- }
- return nil
-}
-
-// addrConn is a network connection to a given address.
-type addrConn struct {
- ctx context.Context
- cancel context.CancelFunc
-
- cc *ClientConn
- dopts dialOptions
- acbw balancer.SubConn
- scopts balancer.NewSubConnOptions
-
- transport transport.ClientTransport // The current transport.
-
- mu sync.Mutex
- addrIdx int // The index in addrs list to start reconnecting from.
- curAddr resolver.Address // The current address.
- addrs []resolver.Address // All addresses that the resolver resolved to.
-
- // Use updateConnectivityState for updating addrConn's connectivity state.
- state connectivity.State
-
- tearDownErr error // The reason this addrConn is torn down.
-
- backoffIdx int
- // backoffDeadline is the time until which resetTransport needs to
- // wait before increasing backoffIdx count.
- backoffDeadline time.Time
- // connectDeadline is the time by which all connection
- // negotiations must complete.
- connectDeadline time.Time
-
- resetBackoff chan struct{}
-
- channelzID int64 // channelz unique identification number
- czData *channelzData
-
- successfulHandshake bool
-}
-
-// Note: this requires a lock on ac.mu.
-func (ac *addrConn) updateConnectivityState(s connectivity.State) {
- ac.state = s
- if channelz.IsOn() {
- channelz.AddTraceEvent(ac.channelzID, &channelz.TraceEventDesc{
- Desc: fmt.Sprintf("Subchannel Connectivity change to %v", s),
- Severity: channelz.CtINFO,
- })
- }
-}
-
-// adjustParams updates parameters used to create transports upon
-// receiving a GoAway.
-func (ac *addrConn) adjustParams(r transport.GoAwayReason) {
- switch r {
- case transport.GoAwayTooManyPings:
- v := 2 * ac.dopts.copts.KeepaliveParams.Time
- ac.cc.mu.Lock()
- if v > ac.cc.mkp.Time {
- ac.cc.mkp.Time = v
- }
- ac.cc.mu.Unlock()
- }
-}
-
-// resetTransport makes sure that a healthy ac.transport exists.
-//
-// The transport will close itself when it encounters an error, or on GOAWAY, or on deadline waiting for handshake, or
-// when the clientconn is closed. Each iteration creating a new transport will try a different address that the balancer
-// assigned to the addrConn, until it has tried all addresses. Once it has tried all addresses, it will re-resolve to
-// get a new address list. If an error is received, the list is re-resolved and the next reset attempt will try from the
-// beginning. This method has backoff built in. The backoff amount starts at 0 and increases each time resolution occurs
-// (addresses are exhausted). The backoff amount is reset to 0 each time a handshake is received.
-//
-// If the DialOption WithWaitForHandshake was set, resetTransport returns successfully only after handshake is received.
-func (ac *addrConn) resetTransport(resolveNow bool) {
- for {
- // If this is the first in a line of resets, we want to resolve immediately. The only other time we
- // want to reset is if we have tried all the addresses handed to us.
- if resolveNow {
- ac.mu.Lock()
- ac.cc.resolveNow(resolver.ResolveNowOption{})
- ac.mu.Unlock()
- }
-
- ac.mu.Lock()
- if ac.state == connectivity.Shutdown {
- ac.mu.Unlock()
- return
- }
-
- // If the connection is READY, a failure must have occurred.
- // Otherwise, we'll consider this is a transient failure when:
- // We've exhausted all addresses
- // We're in CONNECTING
- // And it's not the very first addr to try TODO(deklerk) find a better way to do this than checking ac.successfulHandshake
- if ac.state == connectivity.Ready || (ac.addrIdx == len(ac.addrs)-1 && ac.state == connectivity.Connecting && !ac.successfulHandshake) {
- ac.updateConnectivityState(connectivity.TransientFailure)
- ac.cc.handleSubConnStateChange(ac.acbw, ac.state)
- }
- ac.transport = nil
- ac.mu.Unlock()
-
- if err := ac.nextAddr(); err != nil {
- return
- }
-
- ac.mu.Lock()
- if ac.state == connectivity.Shutdown {
- ac.mu.Unlock()
- return
- }
-
- backoffIdx := ac.backoffIdx
- backoffFor := ac.dopts.bs.Backoff(backoffIdx)
-
- // This will be the duration that dial gets to finish.
- dialDuration := getMinConnectTimeout()
- if backoffFor > dialDuration {
- // Give dial more time as we keep failing to connect.
- dialDuration = backoffFor
- }
- start := time.Now()
- connectDeadline := start.Add(dialDuration)
- ac.backoffDeadline = start.Add(backoffFor)
- ac.connectDeadline = connectDeadline
-
- ac.mu.Unlock()
-
- ac.cc.mu.RLock()
- ac.dopts.copts.KeepaliveParams = ac.cc.mkp
- ac.cc.mu.RUnlock()
-
- ac.mu.Lock()
-
- if ac.state == connectivity.Shutdown {
- ac.mu.Unlock()
- return
- }
-
- if ac.state != connectivity.Connecting {
- ac.updateConnectivityState(connectivity.Connecting)
- ac.cc.handleSubConnStateChange(ac.acbw, ac.state)
- }
-
- addr := ac.addrs[ac.addrIdx]
- copts := ac.dopts.copts
- if ac.scopts.CredsBundle != nil {
- copts.CredsBundle = ac.scopts.CredsBundle
- }
- ac.mu.Unlock()
-
- if channelz.IsOn() {
- channelz.AddTraceEvent(ac.channelzID, &channelz.TraceEventDesc{
- Desc: fmt.Sprintf("Subchannel picks a new address %q to connect", addr.Addr),
- Severity: channelz.CtINFO,
- })
- }
-
- if err := ac.createTransport(backoffIdx, addr, copts, connectDeadline); err != nil {
- continue
- }
-
- return
- }
-}
-
-// createTransport creates a connection to one of the backends in addrs.
-func (ac *addrConn) createTransport(backoffNum int, addr resolver.Address, copts transport.ConnectOptions, connectDeadline time.Time) error {
- oneReset := sync.Once{}
- skipReset := make(chan struct{})
- allowedToReset := make(chan struct{})
- prefaceReceived := make(chan struct{})
- onCloseCalled := make(chan struct{})
-
- var prefaceMu sync.Mutex
- var serverPrefaceReceived bool
- var clientPrefaceWrote bool
-
- onGoAway := func(r transport.GoAwayReason) {
- ac.mu.Lock()
- ac.adjustParams(r)
- ac.mu.Unlock()
- select {
- case <-skipReset: // The outer resetTransport loop will handle reconnection.
- return
- case <-allowedToReset: // We're in the clear to reset.
- go oneReset.Do(func() { ac.resetTransport(false) })
- }
- }
-
- prefaceTimer := time.NewTimer(connectDeadline.Sub(time.Now()))
-
- onClose := func() {
- close(onCloseCalled)
- prefaceTimer.Stop()
-
- select {
- case <-skipReset: // The outer resetTransport loop will handle reconnection.
- return
- case <-allowedToReset: // We're in the clear to reset.
- oneReset.Do(func() { ac.resetTransport(false) })
- }
- }
-
- target := transport.TargetInfo{
- Addr: addr.Addr,
- Metadata: addr.Metadata,
- Authority: ac.cc.authority,
- }
-
- onPrefaceReceipt := func() {
- close(prefaceReceived)
- prefaceTimer.Stop()
-
- // TODO(deklerk): optimization; does anyone else actually use this lock? maybe we can just remove it for this scope
- ac.mu.Lock()
-
- prefaceMu.Lock()
- serverPrefaceReceived = true
- if clientPrefaceWrote {
- ac.successfulHandshake = true
- ac.backoffDeadline = time.Time{}
- ac.connectDeadline = time.Time{}
- ac.addrIdx = 0
- ac.backoffIdx = 0
- }
- prefaceMu.Unlock()
-
- ac.mu.Unlock()
- }
-
- // Do not cancel in the success path because of this issue in Go1.6: https://github.com/golang/go/issues/15078.
- connectCtx, cancel := context.WithDeadline(ac.ctx, connectDeadline)
- if channelz.IsOn() {
- copts.ChannelzParentID = ac.channelzID
- }
-
- newTr, err := transport.NewClientTransport(connectCtx, ac.cc.ctx, target, copts, onPrefaceReceipt, onGoAway, onClose)
-
- if err == nil {
- prefaceMu.Lock()
- clientPrefaceWrote = true
- if serverPrefaceReceived {
- ac.successfulHandshake = true
- }
- prefaceMu.Unlock()
-
- if ac.dopts.waitForHandshake {
- select {
- case <-prefaceTimer.C:
- // We didn't get the preface in time.
- newTr.Close()
- err = errors.New("timed out waiting for server handshake")
- case <-prefaceReceived:
- // We got the preface - huzzah! things are good.
- case <-onCloseCalled:
- // The transport has already closed - noop.
- close(allowedToReset)
- return nil
- }
- } else {
- go func() {
- select {
- case <-prefaceTimer.C:
- // We didn't get the preface in time.
- newTr.Close()
- case <-prefaceReceived:
- // We got the preface just in the nick of time - huzzah!
- case <-onCloseCalled:
- // The transport has already closed - noop.
- }
- }()
- }
- }
-
- if err != nil {
- // newTr is either nil, or closed.
- cancel()
- ac.cc.blockingpicker.updateConnectionError(err)
- ac.mu.Lock()
- if ac.state == connectivity.Shutdown {
- // ac.tearDown(...) has been invoked.
- ac.mu.Unlock()
-
- // We don't want to reset during this close because we prefer to kick out of this function and let the loop
- // in resetTransport take care of reconnecting.
- close(skipReset)
-
- return errConnClosing
- }
- ac.mu.Unlock()
- grpclog.Warningf("grpc: addrConn.createTransport failed to connect to %v. Err :%v. Reconnecting...", addr, err)
-
- // We don't want to reset during this close because we prefer to kick out of this function and let the loop
- // in resetTransport take care of reconnecting.
- close(skipReset)
-
- return err
- }
-
- ac.mu.Lock()
-
- if ac.state == connectivity.Shutdown {
- ac.mu.Unlock()
-
- // We don't want to reset during this close because we prefer to kick out of this function and let the loop
- // in resetTransport take care of reconnecting.
- close(skipReset)
-
- newTr.Close()
- return errConnClosing
- }
-
- ac.updateConnectivityState(connectivity.Ready)
- ac.cc.handleSubConnStateChange(ac.acbw, ac.state)
- ac.transport = newTr
- ac.curAddr = addr
-
- ac.mu.Unlock()
-
- // Ok, _now_ we will finally let the transport reset if it encounters a closable error. Without this, the reader
- // goroutine failing races with all the code in this method that sets the connection to "ready".
- close(allowedToReset)
- return nil
-}
-
-// nextAddr increments the addrIdx if there are more addresses to try. If
-// there are no more addrs to try it will re-resolve, set addrIdx to 0, and
-// increment the backoffIdx.
-//
-// nextAddr must be called without ac.mu being held.
-func (ac *addrConn) nextAddr() error {
- ac.mu.Lock()
-
- // If a handshake has been observed, we expect the counters to have manually
- // been reset so we'll just return, since we want the next usage to start
- // at index 0.
- if ac.successfulHandshake {
- ac.successfulHandshake = false
- ac.mu.Unlock()
- return nil
- }
-
- if ac.addrIdx < len(ac.addrs)-1 {
- ac.addrIdx++
- ac.mu.Unlock()
- return nil
- }
-
- ac.addrIdx = 0
- ac.backoffIdx++
-
- if ac.state == connectivity.Shutdown {
- ac.mu.Unlock()
- return errConnClosing
- }
- ac.cc.resolveNow(resolver.ResolveNowOption{})
- backoffDeadline := ac.backoffDeadline
- b := ac.resetBackoff
- ac.mu.Unlock()
- timer := time.NewTimer(backoffDeadline.Sub(time.Now()))
- select {
- case <-timer.C:
- case <-b:
- timer.Stop()
- case <-ac.ctx.Done():
- timer.Stop()
- return ac.ctx.Err()
- }
- return nil
-}
-
-func (ac *addrConn) resetConnectBackoff() {
- ac.mu.Lock()
- close(ac.resetBackoff)
- ac.backoffIdx = 0
- ac.resetBackoff = make(chan struct{})
- ac.mu.Unlock()
-}
-
-// getReadyTransport returns the transport if ac's state is READY.
-// Otherwise it returns nil, false.
-// If ac's state is IDLE, it will trigger ac to connect.
-func (ac *addrConn) getReadyTransport() (transport.ClientTransport, bool) {
- ac.mu.Lock()
- if ac.state == connectivity.Ready && ac.transport != nil {
- t := ac.transport
- ac.mu.Unlock()
- return t, true
- }
- var idle bool
- if ac.state == connectivity.Idle {
- idle = true
- }
- ac.mu.Unlock()
- // Trigger idle ac to connect.
- if idle {
- ac.connect()
- }
- return nil, false
-}
-
-// tearDown starts to tear down the addrConn.
-// TODO(zhaoq): Make this synchronous to avoid unbounded memory consumption in
-// some edge cases (e.g., the caller opens and closes many addrConn's in a
-// tight loop.
-// tearDown doesn't remove ac from ac.cc.conns.
-func (ac *addrConn) tearDown(err error) {
- ac.mu.Lock()
- if ac.state == connectivity.Shutdown {
- ac.mu.Unlock()
- return
- }
- // We have to set the state to Shutdown before anything else to prevent races
- // between setting the state and logic that waits on context cancelation / etc.
- ac.updateConnectivityState(connectivity.Shutdown)
- ac.cancel()
- ac.tearDownErr = err
- ac.cc.handleSubConnStateChange(ac.acbw, ac.state)
- ac.curAddr = resolver.Address{}
- if err == errConnDrain && ac.transport != nil {
- // GracefulClose(...) may be executed multiple times when
- // i) receiving multiple GoAway frames from the server; or
- // ii) there are concurrent name resolver/Balancer triggered
- // address removal and GoAway.
- // We have to unlock and re-lock here because GracefulClose => Close => onClose, which requires locking ac.mu.
- ac.mu.Unlock()
- ac.transport.GracefulClose()
- ac.mu.Lock()
- }
- if channelz.IsOn() {
- channelz.AddTraceEvent(ac.channelzID, &channelz.TraceEventDesc{
- Desc: "Subchannel Deleted",
- Severity: channelz.CtINFO,
- Parent: &channelz.TraceEventDesc{
- Desc: fmt.Sprintf("Subchanel(id:%d) deleted", ac.channelzID),
- Severity: channelz.CtINFO,
- },
- })
- // TraceEvent needs to be called before RemoveEntry, as TraceEvent may add trace reference to
- // the entity beng deleted, and thus prevent it from being deleted right away.
- channelz.RemoveEntry(ac.channelzID)
- }
- ac.mu.Unlock()
-}
-
-func (ac *addrConn) getState() connectivity.State {
- ac.mu.Lock()
- defer ac.mu.Unlock()
- return ac.state
-}
-
-func (ac *addrConn) ChannelzMetric() *channelz.ChannelInternalMetric {
- ac.mu.Lock()
- addr := ac.curAddr.Addr
- ac.mu.Unlock()
- return &channelz.ChannelInternalMetric{
- State: ac.getState(),
- Target: addr,
- CallsStarted: atomic.LoadInt64(&ac.czData.callsStarted),
- CallsSucceeded: atomic.LoadInt64(&ac.czData.callsSucceeded),
- CallsFailed: atomic.LoadInt64(&ac.czData.callsFailed),
- LastCallStartedTimestamp: time.Unix(0, atomic.LoadInt64(&ac.czData.lastCallStartedTime)),
- }
-}
-
-func (ac *addrConn) incrCallsStarted() {
- atomic.AddInt64(&ac.czData.callsStarted, 1)
- atomic.StoreInt64(&ac.czData.lastCallStartedTime, time.Now().UnixNano())
-}
-
-func (ac *addrConn) incrCallsSucceeded() {
- atomic.AddInt64(&ac.czData.callsSucceeded, 1)
-}
-
-func (ac *addrConn) incrCallsFailed() {
- atomic.AddInt64(&ac.czData.callsFailed, 1)
-}
-
-type retryThrottler struct {
- max float64
- thresh float64
- ratio float64
-
- mu sync.Mutex
- tokens float64 // TODO(dfawley): replace with atomic and remove lock.
-}
-
-// throttle subtracts a retry token from the pool and returns whether a retry
-// should be throttled (disallowed) based upon the retry throttling policy in
-// the service config.
-func (rt *retryThrottler) throttle() bool {
- if rt == nil {
- return false
- }
- rt.mu.Lock()
- defer rt.mu.Unlock()
- rt.tokens--
- if rt.tokens < 0 {
- rt.tokens = 0
- }
- return rt.tokens <= rt.thresh
-}
-
-func (rt *retryThrottler) successfulRPC() {
- if rt == nil {
- return
- }
- rt.mu.Lock()
- defer rt.mu.Unlock()
- rt.tokens += rt.ratio
- if rt.tokens > rt.max {
- rt.tokens = rt.max
- }
-}
-
-type channelzChannel struct {
- cc *ClientConn
-}
-
-func (c *channelzChannel) ChannelzMetric() *channelz.ChannelInternalMetric {
- return c.cc.channelzMetric()
-}
-
-// ErrClientConnTimeout indicates that the ClientConn cannot establish the
-// underlying connections within the specified timeout.
-//
-// Deprecated: This error is never returned by grpc and should not be
-// referenced by users.
-var ErrClientConnTimeout = errors.New("grpc: timed out when dialing")