* * 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 (
	
	
	
	
	
	
	
	
	
	

	
	
	
	
	
	
	
	
	
	
	
	
	
	

	_            // To register roundrobin.
	_          // To register dns resolver.
	_  // To register passthrough resolver.
)
minimum time to give a connection to complete
must match grpclbName in grpclb/grpclb.go
	grpclbName = "grpclb"
)
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.
errConnDrain indicates that the connection starts to be drained and does not accept any new RPCs.
errConnClosing indicates that the connection is closing.
invalidDefaultServiceConfigErrPrefix is used to prefix the json parsing error for the default service config.
	invalidDefaultServiceConfigErrPrefix = "grpc: the provided default service config is invalid"
)
The following errors are returned from Dial and DialContext
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.
errTransportCredsAndBundle indicates that creds bundle is used together with other individual Transport Credentials.
errTransportCredentialsMissing indicates that users want to transmit security information (e.g., OAuth2 token) which requires secure connection on an insecure connection.
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
http2IOBufSize specifies the buffer size for sending frames.
	defaultWriteBufSize = 32 * 1024
	defaultReadBufSize  = 32 * 1024
)
Dial creates a client connection to the given target.
func ( string,  ...DialOption) (*ClientConn, error) {
	return DialContext(context.Background(), , ...)
}
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 ( context.Context,  string,  ...DialOption) ( *ClientConn,  error) {
	 := &ClientConn{
		target:            ,
		csMgr:             &connectivityStateManager{},
		conns:             make(map[*addrConn]struct{}),
		dopts:             defaultDialOptions(),
		blockingpicker:    newPickerWrapper(),
		czData:            new(channelzData),
		firstResolveEvent: grpcsync.NewEvent(),
	}
	.retryThrottler.Store((*retryThrottler)(nil))
	.ctx, .cancel = context.WithCancel(context.Background())

	for ,  := range  {
		.apply(&.dopts)
	}

	chainUnaryClientInterceptors()
	chainStreamClientInterceptors()

	defer func() {
		if  != nil {
			.Close()
		}
	}()

	if channelz.IsOn() {
		if .dopts.channelzParentID != 0 {
			.channelzID = channelz.RegisterChannel(&channelzChannel{}, .dopts.channelzParentID, )
			channelz.AddTraceEvent(logger, .channelzID, 0, &channelz.TraceEventDesc{
				Desc:     "Channel Created",
				Severity: channelz.CtINFO,
				Parent: &channelz.TraceEventDesc{
					Desc:     fmt.Sprintf("Nested Channel(id:%d) created", .channelzID),
					Severity: channelz.CtINFO,
				},
			})
		} else {
			.channelzID = channelz.RegisterChannel(&channelzChannel{}, 0, )
			channelz.Info(logger, .channelzID, "Channel Created")
		}
		.csMgr.channelzID = .channelzID
	}

	if !.dopts.insecure {
		if .dopts.copts.TransportCredentials == nil && .dopts.copts.CredsBundle == nil {
			return nil, errNoTransportSecurity
		}
		if .dopts.copts.TransportCredentials != nil && .dopts.copts.CredsBundle != nil {
			return nil, errTransportCredsAndBundle
		}
	} else {
		if .dopts.copts.TransportCredentials != nil || .dopts.copts.CredsBundle != nil {
			return nil, errCredentialsConflict
		}
		for ,  := range .dopts.copts.PerRPCCredentials {
			if .RequireTransportSecurity() {
				return nil, errTransportCredentialsMissing
			}
		}
	}

	if .dopts.defaultServiceConfigRawJSON != nil {
		 := parseServiceConfig(*.dopts.defaultServiceConfigRawJSON)
		if .Err != nil {
			return nil, fmt.Errorf("%s: %v", invalidDefaultServiceConfigErrPrefix, .Err)
		}
		.dopts.defaultServiceConfig, _ = .Config.(*ServiceConfig)
	}
	.mkp = .dopts.copts.KeepaliveParams

	if .dopts.copts.Dialer == nil {
		.dopts.copts.Dialer = func( context.Context,  string) (net.Conn, error) {
			,  := parseDialTarget()
			return (&net.Dialer{}).DialContext(, , )
		}
		if .dopts.withProxy {
			.dopts.copts.Dialer = newProxyDialer(.dopts.copts.Dialer)
		}
	}

	if .dopts.copts.UserAgent != "" {
		.dopts.copts.UserAgent += " " + grpcUA
	} else {
		.dopts.copts.UserAgent = grpcUA
	}

	if .dopts.timeout > 0 {
		var  context.CancelFunc
		,  = context.WithTimeout(, .dopts.timeout)
		defer ()
	}
	defer func() {
		select {
		case <-.Done():
			switch {
			case .Err() == :
				 = nil
			case  == nil || !.dopts.returnLastError:
				,  = nil, .Err()
			default:
				,  = nil, fmt.Errorf("%v: %v", .Err(), )
			}
		default:
		}
	}()

	 := false
Try to get an initial service config.
		select {
		case ,  := <-.dopts.scChan:
			if  {
				.sc = &
				 = true
			}
		default:
		}
	}
	if .dopts.bs == nil {
		.dopts.bs = backoff.DefaultExponential
	}
Determine the resolver to use.
	.parsedTarget = grpcutil.ParseTarget(.target)
	 := strings.HasPrefix(.target, "unix:")
	channelz.Infof(logger, .channelzID, "parsed scheme: %q", .parsedTarget.Scheme)
	 := .getResolver(.parsedTarget.Scheme)
If resolver builder is still nil, the parsed target's scheme is not registered. Fallback to default resolver and set Endpoint to the original target.
		channelz.Infof(logger, .channelzID, "scheme %q not registered, fallback to default scheme", .parsedTarget.Scheme)
		.parsedTarget = resolver.Target{
			Scheme:   resolver.GetDefaultScheme(),
			Endpoint: ,
		}
		 = .getResolver(.parsedTarget.Scheme)
		if  == nil {
			return nil, fmt.Errorf("could not get resolver for default scheme: %q", .parsedTarget.Scheme)
		}
	}

	 := .dopts.copts.TransportCredentials
	if  != nil && .Info().ServerName != "" {
		.authority = .Info().ServerName
	} else if .dopts.insecure && .dopts.authority != "" {
		.authority = .dopts.authority
	} else if  {
		.authority = "localhost"
Use endpoint from "scheme://authority/endpoint" as the default authority for ClientConn.
		.authority = .parsedTarget.Endpoint
	}
Blocking wait for the initial service config.
		select {
		case ,  := <-.dopts.scChan:
			if  {
				.sc = &
			}
		case <-.Done():
			return nil, .Err()
		}
	}
	if .dopts.scChan != nil {
		go .scWatcher()
	}

	var  credentials.TransportCredentials
	if  := .dopts.copts.TransportCredentials;  != nil {
		 = .Clone()
	}
	.balancerBuildOpts = balancer.BuildOptions{
		DialCreds:        ,
		CredsBundle:      .dopts.copts.CredsBundle,
		Dialer:           .dopts.copts.Dialer,
		ChannelzParentID: .channelzID,
		Target:           .parsedTarget,
	}
Build the resolver.
	,  := newCCResolverWrapper(, )
	if  != nil {
		return nil, fmt.Errorf("failed to build resolver: %v", )
	}
	.mu.Lock()
	.resolverWrapper = 
	.mu.Unlock()
A blocking dial blocks until the clientConn is ready.
	if .dopts.block {
		for {
			 := .GetState()
			if  == connectivity.Ready {
				break
			} else if .dopts.copts.FailOnNonTempDialError &&  == connectivity.TransientFailure {
				if  = .connectionError();  != nil {
					,  := .(interface {
						() bool
					})
					if  && !.() {
						return nil, 
					}
				}
			}
ctx got timeout or canceled.
				if  = .connectionError();  != nil && .dopts.returnLastError {
					return nil, 
				}
				return nil, .Err()
			}
		}
	}

	return , nil
}
chainUnaryClientInterceptors chains all unary client interceptors into one.
func ( *ClientConn) {
Prepend dopts.unaryInt to the chaining interceptors if it exists, since unaryInt will be executed before any other chained interceptors.
	if .dopts.unaryInt != nil {
		 = append([]UnaryClientInterceptor{.dopts.unaryInt}, ...)
	}
	var  UnaryClientInterceptor
	if len() == 0 {
		 = nil
	} else if len() == 1 {
		 = [0]
	} else {
		 = func( context.Context,  string, ,  interface{},  *ClientConn,  UnaryInvoker,  ...CallOption) error {
			return [0](, , , , , getChainUnaryInvoker(, 0, ), ...)
		}
	}
	.dopts.unaryInt = 
}
getChainUnaryInvoker recursively generate the chained unary invoker.
func ( []UnaryClientInterceptor,  int,  UnaryInvoker) UnaryInvoker {
	if  == len()-1 {
		return 
	}
	return func( context.Context,  string, ,  interface{},  *ClientConn,  ...CallOption) error {
		return [+1](, , , , , (, +1, ), ...)
	}
}
chainStreamClientInterceptors chains all stream client interceptors into one.
func ( *ClientConn) {
Prepend dopts.streamInt to the chaining interceptors if it exists, since streamInt will be executed before any other chained interceptors.
	if .dopts.streamInt != nil {
		 = append([]StreamClientInterceptor{.dopts.streamInt}, ...)
	}
	var  StreamClientInterceptor
	if len() == 0 {
		 = nil
	} else if len() == 1 {
		 = [0]
	} else {
		 = func( context.Context,  *StreamDesc,  *ClientConn,  string,  Streamer,  ...CallOption) (ClientStream, error) {
			return [0](, , , , getChainStreamer(, 0, ), ...)
		}
	}
	.dopts.streamInt = 
}
getChainStreamer recursively generate the chained client stream constructor.
func ( []StreamClientInterceptor,  int,  Streamer) Streamer {
	if  == len()-1 {
		return 
	}
	return func( context.Context,  *StreamDesc,  *ClientConn,  string,  ...CallOption) (ClientStream, error) {
		return [+1](, , , , (, +1, ), ...)
	}
}
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 ( *connectivityStateManager) ( connectivity.State) {
	.mu.Lock()
	defer .mu.Unlock()
	if .state == connectivity.Shutdown {
		return
	}
	if .state ==  {
		return
	}
	.state = 
	channelz.Infof(logger, .channelzID, "Channel Connectivity change to %v", )
There are other goroutines waiting on this channel.
		close(.notifyChan)
		.notifyChan = nil
	}
}

func ( *connectivityStateManager) () connectivity.State {
	.mu.Lock()
	defer .mu.Unlock()
	return .state
}

func ( *connectivityStateManager) () <-chan struct{} {
	.mu.Lock()
	defer .mu.Unlock()
	if .notifyChan == nil {
		.notifyChan = make(chan struct{})
	}
	return .notifyChan
}
ClientConnInterface defines the functions clients need to perform unary and streaming RPCs. It is implemented by *ClientConn, and is only intended to be referenced by generated code.
Invoke performs a unary RPC and returns after the response is received into reply.
NewStream begins a streaming RPC.
	NewStream(ctx context.Context, desc *StreamDesc, method string, opts ...CallOption) (ClientStream, error)
}
Assert *ClientConn implements ClientConnInterface.
var _ ClientConnInterface = (*ClientConn)(nil)
ClientConn represents a virtual connection to a conceptual endpoint, to perform RPCs. A ClientConn is free to have zero or more actual connections to the endpoint based on configuration, load, etc. It is also free to determine which actual endpoints to use and may change it every RPC, permitting client-side load balancing. A ClientConn encapsulates a range of functionality including name resolution, TCP connection establishment (with retries and backoff) and TLS handshakes. It also handles errors on established connections by re-resolving the name and reconnecting.
type ClientConn struct {
	ctx    context.Context
	cancel context.CancelFunc

	target       string
	parsedTarget resolver.Target
	authority    string
	dopts        dialOptions
	csMgr        *connectivityStateManager

	balancerBuildOpts balancer.BuildOptions
	blockingpicker    *pickerWrapper

	mu              sync.RWMutex
	resolverWrapper *ccResolverWrapper
	sc              *ServiceConfig
Keepalive parameter can be updated if a GoAway is received.
	mkp             keepalive.ClientParameters
	curBalancerName string
	balancerWrapper *ccBalancerWrapper
	retryThrottler  atomic.Value

	firstResolveEvent *grpcsync.Event

	channelzID int64 // channelz unique identification number
	czData     *channelzData

	lceMu               sync.Mutex // protects lastConnectionError
	lastConnectionError error
}
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 ( *ClientConn) ( context.Context,  connectivity.State) bool {
	 := .csMgr.getNotifyChan()
	if .csMgr.getState() !=  {
		return true
	}
	select {
	case <-.Done():
		return false
	case <-:
		return true
	}
}
GetState returns the connectivity.State of ClientConn. This is an EXPERIMENTAL API.
func ( *ClientConn) () connectivity.State {
	return .csMgr.getState()
}

func ( *ClientConn) () {
	for {
		select {
		case ,  := <-.dopts.scChan:
			if ! {
				return
			}
TODO: load balance policy runtime change is ignored. We may revisit this decision in the future.
			.sc = &
			.mu.Unlock()
		case <-.ctx.Done():
			return
		}
	}
}
waitForResolvedAddrs blocks until the resolver has provided addresses or the context expires. Returns nil unless the context expires first; otherwise returns a status error based on the context.
This is on the RPC path, so we use a fast path to avoid the more-expensive "select" below after the resolver has returned once.
	if .firstResolveEvent.HasFired() {
		return nil
	}
	select {
	case <-.firstResolveEvent.Done():
		return nil
	case <-.Done():
		return status.FromContextError(.Err()).Err()
	case <-.ctx.Done():
		return ErrClientConnClosing
	}
}

var emptyServiceConfig *ServiceConfig

func () {
	 := parseServiceConfig("{}")
	if .Err != nil {
		panic(fmt.Sprintf("impossible error parsing empty service config: %v", .Err))
	}
	emptyServiceConfig = .Config.(*ServiceConfig)
}

func ( *ClientConn) ( []resolver.Address) {
	if .sc != nil {
		.applyServiceConfigAndBalancer(.sc, )
		return
	}
	if .dopts.defaultServiceConfig != nil {
		.applyServiceConfigAndBalancer(.dopts.defaultServiceConfig, )
	} else {
		.applyServiceConfigAndBalancer(emptyServiceConfig, )
	}
}

func ( *ClientConn) ( resolver.State,  error) error {
	defer .firstResolveEvent.Fire()
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 .conns == nil {
		.mu.Unlock()
		return nil
	}
May need to apply the initial service config in case the resolver doesn't support service configs, or doesn't provide a service config with the new addresses.
		.maybeApplyDefaultServiceConfig(nil)

		if .balancerWrapper != nil {
			.balancerWrapper.resolverError()
		}
No addresses are valid with err set; return early.
		.mu.Unlock()
		return balancer.ErrBadResolverState
	}

	var  error
	if .dopts.disableServiceConfig || .ServiceConfig == nil {
TODO: do we need to apply a failing LB policy if there is no default, per the error handling design?
	} else {
		if ,  := .ServiceConfig.Config.(*ServiceConfig); .ServiceConfig.Err == nil &&  {
			.applyServiceConfigAndBalancer(, .Addresses)
		} else {
			 = balancer.ErrBadResolverState
			if .balancerWrapper == nil {
				var  error
				if .ServiceConfig.Err != nil {
					 = status.Errorf(codes.Unavailable, "error parsing service config: %v", .ServiceConfig.Err)
				} else {
					 = status.Errorf(codes.Unavailable, "illegal service config type: %T", .ServiceConfig.Config)
				}
				.blockingpicker.updatePicker(base.NewErrPicker())
				.csMgr.updateState(connectivity.TransientFailure)
				.mu.Unlock()
				return 
			}
		}
	}

	var  serviceconfig.LoadBalancingConfig
	if .dopts.balancerBuilder == nil && .sc != nil && .sc.lbConfig != nil {
		 = .sc.lbConfig.cfg
	}

	 := .curBalancerName
	 := .balancerWrapper
	.mu.Unlock()
Filter any grpclb addresses since we don't have the grpclb balancer.
		for  := 0;  < len(.Addresses); {
			if .Addresses[].Type == resolver.GRPCLB {
				copy(.Addresses[:], .Addresses[+1:])
				.Addresses = .Addresses[:len(.Addresses)-1]
				continue
			}
			++
		}
	}
	 := .updateClientConnState(&balancer.ClientConnState{ResolverState: , BalancerConfig: })
	if  == nil {
currently meaningless to the caller.
	}
	return 
}
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 ( *ClientConn) ( string) {
	if strings.EqualFold(.curBalancerName, ) {
		return
	}

	channelz.Infof(logger, .channelzID, "ClientConn switching balancer to %q", )
	if .dopts.balancerBuilder != nil {
		channelz.Info(logger, .channelzID, "ignoring balancer switching: Balancer DialOption used instead")
		return
	}
	if .balancerWrapper != nil {
		.balancerWrapper.close()
	}

	 := balancer.Get()
	if  == nil {
		channelz.Warningf(logger, .channelzID, "Channel switches to new LB policy %q due to fallback from invalid balancer name", PickFirstBalancerName)
		channelz.Infof(logger, .channelzID, "failed to get balancer builder for: %v, using pick_first instead", )
		 = newPickfirstBuilder()
	} else {
		channelz.Infof(logger, .channelzID, "Channel switches to new LB policy %q", )
	}

	.curBalancerName = .Name()
	.balancerWrapper = newCCBalancerWrapper(, , .balancerBuildOpts)
}

func ( *ClientConn) ( balancer.SubConn,  connectivity.State,  error) {
	.mu.Lock()
	if .conns == nil {
		.mu.Unlock()
		return
TODO(bar switching) send updates to all balancer wrappers when balancer gracefully switching is supported.
	.balancerWrapper.handleSubConnStateChange(, , )
	.mu.Unlock()
}
newAddrConn creates an addrConn for addrs and adds it to cc.conns. Caller needs to make sure len(addrs) > 0.
func ( *ClientConn) ( []resolver.Address,  balancer.NewSubConnOptions) (*addrConn, error) {
	 := &addrConn{
		state:        connectivity.Idle,
		cc:           ,
		addrs:        ,
		scopts:       ,
		dopts:        .dopts,
		czData:       new(channelzData),
		resetBackoff: make(chan struct{}),
	}
Track ac in cc. This needs to be done before any getTransport(...) is called.
	.mu.Lock()
	if .conns == nil {
		.mu.Unlock()
		return nil, ErrClientConnClosing
	}
	if channelz.IsOn() {
		.channelzID = channelz.RegisterSubChannel(, .channelzID, "")
		channelz.AddTraceEvent(logger, .channelzID, 0, &channelz.TraceEventDesc{
			Desc:     "Subchannel Created",
			Severity: channelz.CtINFO,
			Parent: &channelz.TraceEventDesc{
				Desc:     fmt.Sprintf("Subchannel(id:%d) created", .channelzID),
				Severity: channelz.CtINFO,
			},
		})
	}
	.conns[] = struct{}{}
	.mu.Unlock()
	return , nil
}
removeAddrConn removes the addrConn in the subConn from clientConn. It also tears down the ac with the given error.
func ( *ClientConn) ( *addrConn,  error) {
	.mu.Lock()
	if .conns == nil {
		.mu.Unlock()
		return
	}
	delete(.conns, )
	.mu.Unlock()
	.tearDown()
}

func ( *ClientConn) () *channelz.ChannelInternalMetric {
	return &channelz.ChannelInternalMetric{
		State:                    .GetState(),
		Target:                   .target,
		CallsStarted:             atomic.LoadInt64(&.czData.callsStarted),
		CallsSucceeded:           atomic.LoadInt64(&.czData.callsSucceeded),
		CallsFailed:              atomic.LoadInt64(&.czData.callsFailed),
		LastCallStartedTimestamp: time.Unix(0, atomic.LoadInt64(&.czData.lastCallStartedTime)),
	}
}
Target returns the target string of the ClientConn. This is an EXPERIMENTAL API.
func ( *ClientConn) () string {
	return .target
}

func ( *ClientConn) () {
	atomic.AddInt64(&.czData.callsStarted, 1)
	atomic.StoreInt64(&.czData.lastCallStartedTime, time.Now().UnixNano())
}

func ( *ClientConn) () {
	atomic.AddInt64(&.czData.callsSucceeded, 1)
}

func ( *ClientConn) () {
	atomic.AddInt64(&.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 ( *addrConn) () error {
	.mu.Lock()
	if .state == connectivity.Shutdown {
		.mu.Unlock()
		return errConnClosing
	}
	if .state != connectivity.Idle {
		.mu.Unlock()
		return nil
Update connectivity state within the lock to prevent subsequent or concurrent calls from resetting the transport more than once.
	.updateConnectivityState(connectivity.Connecting, nil)
	.mu.Unlock()
Start a goroutine connecting to the server asynchronously.
	go .resetTransport()
	return nil
}
tryUpdateAddrs tries to update ac.addrs with the new addresses list. If ac is Connecting, it returns false. The caller should tear down the ac and create a new one. Note that the backoff will be reset when this happens. If ac is TransientFailure, it updates ac.addrs and returns true. The updated addresses will be picked up by retry in the next iteration after backoff. If ac is Shutdown or Idle, it updates ac.addrs and returns true. If ac is Ready, 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 ( *addrConn) ( []resolver.Address) bool {
	.mu.Lock()
	defer .mu.Unlock()
	channelz.Infof(logger, .channelzID, "addrConn: tryUpdateAddrs curAddr: %v, addrs: %v", .curAddr, )
	if .state == connectivity.Shutdown ||
		.state == connectivity.TransientFailure ||
		.state == connectivity.Idle {
		.addrs = 
		return true
	}

	if .state == connectivity.Connecting {
		return false
	}
ac.state is Ready, try to find the connected address.
	var  bool
	for ,  := range  {
		if reflect.DeepEqual(.curAddr, ) {
			 = true
			break
		}
	}
	channelz.Infof(logger, .channelzID, "addrConn: tryUpdateAddrs curAddrFound: %v", )
	if  {
		.addrs = 
	}

	return 
}
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 service's default config under the service (i.e /service/) and then for the default for all services (empty string). If there is a default MethodConfig for the service, we return it. Otherwise, we return an empty MethodConfig.
TODO: Avoid the locking here.
	.mu.RLock()
	defer .mu.RUnlock()
	if .sc == nil {
		return MethodConfig{}
	}
	if ,  := .sc.Methods[];  {
		return 
	}
	 := strings.LastIndex(, "/")
	if ,  := .sc.Methods[[:+1]];  {
		return 
	}
	return .sc.Methods[""]
}

func ( *ClientConn) () *healthCheckConfig {
	.mu.RLock()
	defer .mu.RUnlock()
	if .sc == nil {
		return nil
	}
	return .sc.healthCheckConfig
}

func ( *ClientConn) ( context.Context,  bool,  string) (transport.ClientTransport, func(balancer.DoneInfo), error) {
	, ,  := .blockingpicker.pick(, , balancer.PickInfo{
		Ctx:            ,
		FullMethodName: ,
	})
	if  != nil {
		return nil, nil, toRPCErr()
	}
	return , , nil
}

func ( *ClientConn) ( *ServiceConfig,  []resolver.Address) {
should never reach here.
		return
	}
	.sc = 

	if .sc.retryThrottling != nil {
		 := &retryThrottler{
			tokens: .sc.retryThrottling.MaxTokens,
			max:    .sc.retryThrottling.MaxTokens,
			thresh: .sc.retryThrottling.MaxTokens / 2,
			ratio:  .sc.retryThrottling.TokenRatio,
		}
		.retryThrottler.Store()
	} else {
		.retryThrottler.Store((*retryThrottler)(nil))
	}
Only look at balancer types and switch balancer if balancer dial option is not set.
		var  string
		if .sc != nil && .sc.lbConfig != nil {
			 = .sc.lbConfig.name
		} else {
			var  bool
			for ,  := range  {
				if .Type == resolver.GRPCLB {
					 = true
					break
				}
			}
			if  {
				 = grpclbName
			} else if .sc != nil && .sc.LB != nil {
				 = *.sc.LB
			} else {
				 = PickFirstBalancerName
			}
		}
		.switchBalancer()
Balancer dial option was set, and this is the first time handling resolved addresses. Build a balancer with dopts.balancerBuilder.
		.curBalancerName = .dopts.balancerBuilder.Name()
		.balancerWrapper = newCCBalancerWrapper(, .dopts.balancerBuilder, .balancerBuildOpts)
	}
}

func ( *ClientConn) ( resolver.ResolveNowOptions) {
	.mu.RLock()
	 := .resolverWrapper
	.mu.RUnlock()
	if  == nil {
		return
	}
	go .resolveNow()
}
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 ( *ClientConn) () {
	.mu.Lock()
	 := .conns
	.mu.Unlock()
	for  := range  {
		.resetConnectBackoff()
	}
}
Close tears down the ClientConn and all underlying connections.
func ( *ClientConn) () error {
	defer .cancel()

	.mu.Lock()
	if .conns == nil {
		.mu.Unlock()
		return ErrClientConnClosing
	}
	 := .conns
	.conns = nil
	.csMgr.updateState(connectivity.Shutdown)

	 := .resolverWrapper
	.resolverWrapper = nil
	 := .balancerWrapper
	.balancerWrapper = nil
	.mu.Unlock()

	.blockingpicker.close()

	if  != nil {
		.close()
	}
	if  != nil {
		.close()
	}

	for  := range  {
		.tearDown(ErrClientConnClosing)
	}
	if channelz.IsOn() {
		 := &channelz.TraceEventDesc{
			Desc:     "Channel Deleted",
			Severity: channelz.CtINFO,
		}
		if .dopts.channelzParentID != 0 {
			.Parent = &channelz.TraceEventDesc{
				Desc:     fmt.Sprintf("Nested channel(id:%d) deleted", .channelzID),
				Severity: channelz.CtINFO,
			}
		}
TraceEvent needs to be called before RemoveEntry, as TraceEvent may add trace reference to the entity being deleted, and thus prevent it from being deleted right away.
		channelz.RemoveEntry(.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 is set when there's a viable transport (note: ac state may not be READY as LB channel health checking may require server to report healthy to set ac to READY), and is reset to nil when the current transport should no longer be used to create a stream (e.g. after GoAway is received, transport is closed, ac has been torn down).
	transport transport.ClientTransport // The current transport.

	mu      sync.Mutex
	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

	backoffIdx   int // Needs to be stateful for resetConnectBackoff.
	resetBackoff chan struct{}

	channelzID int64 // channelz unique identification number.
	czData     *channelzData
}
Note: this requires a lock on ac.mu.
func ( *addrConn) ( connectivity.State,  error) {
	if .state ==  {
		return
	}
	.state = 
	channelz.Infof(logger, .channelzID, "Subchannel Connectivity change to %v", )
	.cc.handleSubConnStateChange(.acbw, , )
}
adjustParams updates parameters used to create transports upon receiving a GoAway.
func ( *addrConn) ( transport.GoAwayReason) {
	switch  {
	case transport.GoAwayTooManyPings:
		 := 2 * .dopts.copts.KeepaliveParams.Time
		.cc.mu.Lock()
		if  > .cc.mkp.Time {
			.cc.mkp.Time = 
		}
		.cc.mu.Unlock()
	}
}

func ( *addrConn) () {
	for  := 0; ; ++ {
		if  > 0 {
			.cc.resolveNow(resolver.ResolveNowOptions{})
		}

		.mu.Lock()
		if .state == connectivity.Shutdown {
			.mu.Unlock()
			return
		}

		 := .addrs
This will be the duration that dial gets to finish.
		 := minConnectTimeout
		if .dopts.minConnectTimeout != nil {
			 = .dopts.minConnectTimeout()
		}
Give dial more time as we keep failing to connect.
			 =
We can potentially spend all the time trying the first address, and if the server accepts the connection and then hangs, the following addresses will never be tried. The spec doesn't mention what should be done for multiple addresses. https://github.com/grpc/grpc/blob/master/doc/connection-backoff.md#proposed-backoff-algorithm
		 := time.Now().Add()

		.updateConnectivityState(connectivity.Connecting, nil)
		.transport = nil
		.mu.Unlock()

		, , ,  := .tryAllAddrs(, )
After exhausting all addresses, the addrConn enters TRANSIENT_FAILURE.
			.mu.Lock()
			if .state == connectivity.Shutdown {
				.mu.Unlock()
				return
			}
			.updateConnectivityState(connectivity.TransientFailure, )
Backoff.
			 := .resetBackoff
			.mu.Unlock()

			 := time.NewTimer()
			select {
			case <-.C:
				.mu.Lock()
				.backoffIdx++
				.mu.Unlock()
			case <-:
				.Stop()
			case <-.ctx.Done():
				.Stop()
				return
			}
			continue
		}

		.mu.Lock()
		if .state == connectivity.Shutdown {
			.mu.Unlock()
			.Close()
			return
		}
		.curAddr = 
		.transport = 
		.backoffIdx = 0

		,  := context.WithCancel(.ctx)
		.startHealthCheck()
		.mu.Unlock()
Block until the created transport is down. And when this happens, we restart from the top of the addr list.
		<-.Done()
restart connecting - the top of the loop will set state to CONNECTING. This is against the current connectivity semantics doc, however it allows for graceful behavior for RPCs not yet dispatched - unfortunate timing would otherwise lead to the RPC failing even though the TRANSIENT_FAILURE state (called for by the doc) would be instantaneous. Ideally we should transition to Idle here and block until there is RPC activity that leads to the balancer requesting a reconnect of the associated SubConn.
	}
}
tryAllAddrs tries to creates a connection to the addresses, and stop when at the first successful one. It returns the transport, the address and a Event in the successful case. The Event fires when the returned transport disconnects.
func ( *addrConn) ( []resolver.Address,  time.Time) (transport.ClientTransport, resolver.Address, *grpcsync.Event, error) {
	var  error
	for ,  := range  {
		.mu.Lock()
		if .state == connectivity.Shutdown {
			.mu.Unlock()
			return nil, resolver.Address{}, nil, errConnClosing
		}

		.cc.mu.RLock()
		.dopts.copts.KeepaliveParams = .cc.mkp
		.cc.mu.RUnlock()

		 := .dopts.copts
		if .scopts.CredsBundle != nil {
			.CredsBundle = .scopts.CredsBundle
		}
		.mu.Unlock()

		channelz.Infof(logger, .channelzID, "Subchannel picks a new address %q to connect", .Addr)

		, ,  := .createTransport(, , )
		if  == nil {
			return , , , nil
		}
		if  == nil {
			 = 
		}
		.cc.updateConnectionError()
	}
Couldn't connect to any address.
	return nil, resolver.Address{}, nil, 
}
createTransport creates a connection to addr. It returns the transport and a Event in the successful case. The Event fires when the returned transport disconnects.
func ( *addrConn) ( resolver.Address,  transport.ConnectOptions,  time.Time) (transport.ClientTransport, *grpcsync.Event, error) {
	 := make(chan struct{})
	 := make(chan struct{})
	 := grpcsync.NewEvent()
addr.ServerName takes precedent over ClientConn authority, if present.
	if .ServerName == "" {
		.ServerName = .cc.authority
	}

	 := sync.Once{}
	 := func( transport.GoAwayReason) {
		.mu.Lock()
		.adjustParams()
		.Do(func() {
Prevent this SubConn from being used for new RPCs by setting its state to Connecting. TODO: this should be Idle when grpc-go properly supports it.
				.updateConnectivityState(connectivity.Connecting, nil)
			}
		})
		.mu.Unlock()
		.Fire()
	}

	 := func() {
		.mu.Lock()
		.Do(func() {
Prevent this SubConn from being used for new RPCs by setting its state to Connecting. TODO: this should be Idle when grpc-go properly supports it.
				.updateConnectivityState(connectivity.Connecting, nil)
			}
		})
		.mu.Unlock()
		close()
		.Fire()
	}

	 := func() {
		close()
	}

	,  := context.WithDeadline(.ctx, )
	defer ()
	if channelz.IsOn() {
		.ChannelzParentID = .channelzID
	}

	,  := transport.NewClientTransport(, .cc.ctx, , , , , )
newTr is either nil, or closed.
		channelz.Warningf(logger, .channelzID, "grpc: addrConn.createTransport failed to connect to %v. Err: %v. Reconnecting...", , )
		return nil, nil, 
	}

	select {
We didn't get the preface in time.
		.Close()
		channelz.Warningf(logger, .channelzID, "grpc: addrConn.createTransport failed to connect to %v: didn't receive server preface in time. Reconnecting...", )
		return nil, nil, errors.New("timed out waiting for server handshake")
We got the preface - huzzah! things are good.
The transport has already closed - noop.
TODO(deklerk) this should bail on ac.ctx.Done(). Add a test and fix.
	}
	return , , nil
}
startHealthCheck starts the health checking stream (RPC) to watch the health stats of this connection if health checking is requested and configured. LB channel health checking is enabled when all requirements below are met: 1. it is not disabled by the user with the WithDisableHealthCheck DialOption 2. internal.HealthCheckFunc is set by importing the grpc/health package 3. a service config with non-empty healthCheckConfig field is provided 4. the load balancer requests it It sets addrConn to READY if the health checking stream is not started. Caller must hold ac.mu.
func ( *addrConn) ( context.Context) {
	var  bool
	defer func() {
		if ! {
			.updateConnectivityState(connectivity.Ready, nil)
		}
	}()

	if .cc.dopts.disableHealthCheck {
		return
	}
	 := .cc.healthCheckConfig()
	if  == nil {
		return
	}
	if !.scopts.HealthCheckEnabled {
		return
	}
	 := .cc.dopts.healthCheckFunc
The health package is not imported to set health check function. TODO: add a link to the health check doc in the error message.
		channelz.Error(logger, .channelzID, "Health check is requested but health check function is not set.")
		return
	}

	 = true
Set up the health check helper functions.
	 := .transport
	 := func( string) (interface{}, error) {
		.mu.Lock()
		if .transport !=  {
			.mu.Unlock()
			return nil, status.Error(codes.Canceled, "the provided transport is no longer valid to use")
		}
		.mu.Unlock()
		return newNonRetryClientStream(, &StreamDesc{ServerStreams: true}, , , )
	}
	 := func( connectivity.State,  error) {
		.mu.Lock()
		defer .mu.Unlock()
		if .transport !=  {
			return
		}
		.updateConnectivityState(, )
Start the health checking stream.
	go func() {
		 := .cc.dopts.healthCheckFunc(, , , .ServiceName)
		if  != nil {
			if status.Code() == codes.Unimplemented {
				channelz.Error(logger, .channelzID, "Subchannel health check is unimplemented at server side, thus health check is disabled")
			} else {
				channelz.Errorf(logger, .channelzID, "HealthCheckFunc exits with unexpected error %v", )
			}
		}
	}()
}

func ( *addrConn) () {
	.mu.Lock()
	close(.resetBackoff)
	.backoffIdx = 0
	.resetBackoff = make(chan struct{})
	.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 ( *addrConn) () (transport.ClientTransport, bool) {
	.mu.Lock()
	if .state == connectivity.Ready && .transport != nil {
		 := .transport
		.mu.Unlock()
		return , true
	}
	var  bool
	if .state == connectivity.Idle {
		 = true
	}
Trigger idle ac to connect.
	if  {
		.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 ( *addrConn) ( error) {
	.mu.Lock()
	if .state == connectivity.Shutdown {
		.mu.Unlock()
		return
	}
	 := .transport
We have to set the state to Shutdown before anything else to prevent races between setting the state and logic that waits on context cancellation / etc.
	.updateConnectivityState(connectivity.Shutdown, nil)
	.cancel()
	.curAddr = resolver.Address{}
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.
		.mu.Unlock()
		.GracefulClose()
		.mu.Lock()
	}
	if channelz.IsOn() {
		channelz.AddTraceEvent(logger, .channelzID, 0, &channelz.TraceEventDesc{
			Desc:     "Subchannel Deleted",
			Severity: channelz.CtINFO,
			Parent: &channelz.TraceEventDesc{
				Desc:     fmt.Sprintf("Subchanel(id:%d) deleted", .channelzID),
				Severity: channelz.CtINFO,
			},
TraceEvent needs to be called before RemoveEntry, as TraceEvent may add trace reference to the entity being deleted, and thus prevent it from being deleted right away.
		channelz.RemoveEntry(.channelzID)
	}
	.mu.Unlock()
}

func ( *addrConn) () connectivity.State {
	.mu.Lock()
	defer .mu.Unlock()
	return .state
}

func ( *addrConn) () *channelz.ChannelInternalMetric {
	.mu.Lock()
	 := .curAddr.Addr
	.mu.Unlock()
	return &channelz.ChannelInternalMetric{
		State:                    .getState(),
		Target:                   ,
		CallsStarted:             atomic.LoadInt64(&.czData.callsStarted),
		CallsSucceeded:           atomic.LoadInt64(&.czData.callsSucceeded),
		CallsFailed:              atomic.LoadInt64(&.czData.callsFailed),
		LastCallStartedTimestamp: time.Unix(0, atomic.LoadInt64(&.czData.lastCallStartedTime)),
	}
}

func ( *addrConn) () {
	atomic.AddInt64(&.czData.callsStarted, 1)
	atomic.StoreInt64(&.czData.lastCallStartedTime, time.Now().UnixNano())
}

func ( *addrConn) () {
	atomic.AddInt64(&.czData.callsSucceeded, 1)
}

func ( *addrConn) () {
	atomic.AddInt64(&.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 ( *retryThrottler) () bool {
	if  == nil {
		return false
	}
	.mu.Lock()
	defer .mu.Unlock()
	.tokens--
	if .tokens < 0 {
		.tokens = 0
	}
	return .tokens <= .thresh
}

func ( *retryThrottler) () {
	if  == nil {
		return
	}
	.mu.Lock()
	defer .mu.Unlock()
	.tokens += .ratio
	if .tokens > .max {
		.tokens = .max
	}
}

type channelzChannel struct {
	cc *ClientConn
}

func ( *channelzChannel) () *channelz.ChannelInternalMetric {
	return .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")

func ( *ClientConn) ( string) resolver.Builder {
	for ,  := range .dopts.resolvers {
		if  == .Scheme() {
			return 
		}
	}
	return resolver.Get()
}

func ( *ClientConn) ( error) {
	.lceMu.Lock()
	.lastConnectionError = 
	.lceMu.Unlock()
}

func ( *ClientConn) () error {
	.lceMu.Lock()
	defer .lceMu.Unlock()
	return .lastConnectionError