func (ac *addrConn) tearDown(err error) {
	ac.cancel()
	ac.mu.Lock()
	defer ac.mu.Unlock()
	if ac.down != nil {
		ac.down(downErrorf(false, false, "%v", err))
		ac.down = nil
	}
	if err == errConnDrain && ac.transport != nil {
		ac.transport.GracefulClose()
	}
	if ac.state == Shutdown {
		return
	}
	ac.state = Shutdown
	ac.tearDownErr = err
	ac.stateCV.Broadcast()
	if ac.events != nil {
		ac.events.Finish()
		ac.events = nil
	}
	if ac.ready != nil {
		close(ac.ready)
		ac.ready = nil
	}
	if ac.transport != nil && err != errConnDrain {
		ac.transport.Close()
	}
	return
}
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.
package grpc
import (
	"errors"
	"fmt"
	"net"
	"strings"
	"sync"
	"time"
	"golang.org/x/net/context"
	"golang.org/x/net/trace"
	"google.golang.org/grpc/credentials"
	"google.golang.org/grpc/grpclog"
	"google.golang.org/grpc/transport"
)
var (
	ErrClientConnClosing = errors.New("grpc: the client connection is closing")
	ErrClientConnTimeout = errors.New("grpc: timed out when dialing")
	errNoTransportSecurity = errors.New("grpc: no transport security set (use grpc.WithInsecure() explicitly or set credentials)")
	errTransportCredentialsMissing = errors.New("grpc: the credentials require transport level security (use grpc.WithTransportCredentials() to set)")
	errCredentialsConflict = errors.New("grpc: transport credentials are set for an insecure connection (grpc.WithTransportCredentials() and grpc.WithInsecure() are both called)")
	errNetworkIO = errors.New("grpc: failed with network I/O error")
	errConnDrain = errors.New("grpc: the connection is drained")
	errConnClosing = errors.New("grpc: the connection is closing")
	errConnUnavailable = errors.New("grpc: the connection is unavailable")
	errNoAddr          = errors.New("grpc: there is no address available to dial")
	minConnectTimeout = 20 * time.Second
)
type dialOptions struct {
	codec    Codec
	cp       Compressor
	dc       Decompressor
	bs       backoffStrategy
	balancer Balancer
	block    bool
	insecure bool
	timeout  time.Duration
	copts    transport.ConnectOptions
}
type DialOption func(*dialOptions)
func WithCodec(c Codec) DialOption {
	return func(o *dialOptions) {
		o.codec = c
	}
}
func WithCompressor(cp Compressor) DialOption {
	return func(o *dialOptions) {
		o.cp = cp
	}
}
func WithDecompressor(dc Decompressor) DialOption {
	return func(o *dialOptions) {
		o.dc = dc
	}
}
func WithBalancer(b Balancer) DialOption {
	return func(o *dialOptions) {
		o.balancer = b
	}
}
func WithBackoffMaxDelay(md time.Duration) DialOption {
	return WithBackoffConfig(BackoffConfig{MaxDelay: md})
}
func WithBackoffConfig(b BackoffConfig) DialOption {
	setDefaults(&b)
	return withBackoff(b)
}
func withBackoff(bs backoffStrategy) DialOption {
	return func(o *dialOptions) {
		o.bs = bs
	}
}
func WithBlock() DialOption {
	return func(o *dialOptions) {
		o.block = true
	}
}
func WithInsecure() DialOption {
	return func(o *dialOptions) {
		o.insecure = true
	}
}
func WithTransportCredentials(creds credentials.TransportCredentials) DialOption {
	return func(o *dialOptions) {
		o.copts.TransportCredentials = creds
	}
}
func WithPerRPCCredentials(creds credentials.PerRPCCredentials) DialOption {
	return func(o *dialOptions) {
		o.copts.PerRPCCredentials = append(o.copts.PerRPCCredentials, creds)
	}
}
func WithTimeout(d time.Duration) DialOption {
	return func(o *dialOptions) {
		o.timeout = d
	}
}
func WithDialer(f func(string, time.Duration) (net.Conn, error)) DialOption {
	return func(o *dialOptions) {
		o.copts.Dialer = func(ctx context.Context, addr string) (net.Conn, error) {
			if deadline, ok := ctx.Deadline(); ok {
				return f(addr, deadline.Sub(time.Now()))
			}
			return f(addr, 0)
		}
	}
}
func WithUserAgent(s string) DialOption {
	return func(o *dialOptions) {
		o.copts.UserAgent = s
	}
}
func Dial(target string, opts ...DialOption) (*ClientConn, error) {
	return DialContext(context.Background(), target, opts...)
}
func DialContext(ctx context.Context, target string, opts ...DialOption) (*ClientConn, error) {
	cc := &ClientConn{
		target: target,
		conns:  make(map[Address]*addrConn),
	}
	cc.ctx, cc.cancel = context.WithCancel(ctx)
	for _, opt := range opts {
		opt(&cc.dopts)
	}
	if cc.dopts.codec == nil {
		cc.dopts.codec = protoCodec{}
	}
	if cc.dopts.bs == nil {
		cc.dopts.bs = DefaultBackoffConfig
	}
	var (
		ok    bool
		addrs []Address
	)
	if cc.dopts.balancer == nil {
		addrs = append(addrs, Address{Addr: target})
	} else {
		if err := cc.dopts.balancer.Start(target); err != nil {
			return nil, err
		}
		ch := cc.dopts.balancer.Notify()
		if ch == nil {
			addrs = append(addrs, Address{Addr: target})
		} else {
			addrs, ok = <-ch
			if !ok || len(addrs) == 0 {
				return nil, errNoAddr
			}
		}
	}
	waitC := make(chan error, 1)
	go func() {
		for _, a := range addrs {
			if err := cc.resetAddrConn(a, false, nil); err != nil {
				waitC <- err
				return
			}
		}
		close(waitC)
	}()
	var timeoutCh <-chan time.Time
	if cc.dopts.timeout > 0 {
		timeoutCh = time.After(cc.dopts.timeout)
	}
	select {
	case err := <-waitC:
		if err != nil {
			cc.Close()
			return nil, err
		}
	case <-cc.ctx.Done():
		cc.Close()
		return nil, cc.ctx.Err()
	case <-timeoutCh:
		cc.Close()
		return nil, ErrClientConnTimeout
	}
	if ok {
		go cc.lbWatcher()
	}
	colonPos := strings.LastIndex(target, ":")
	if colonPos == -1 {
		colonPos = len(target)
	}
	cc.authority = target[:colonPos]
	return cc, nil
}
type ConnectivityState int
const (
	Idle ConnectivityState = iota
	Connecting
	Ready
	TransientFailure
	Shutdown
)
func (s ConnectivityState) String() string {
	switch s {
	case Idle:
		return "IDLE"
	case Connecting:
		return "CONNECTING"
	case Ready:
		return "READY"
	case TransientFailure:
		return "TRANSIENT_FAILURE"
	case Shutdown:
		return "SHUTDOWN"
	default:
		panic(fmt.Sprintf("unknown connectivity state: %d", s))
	}
}
type ClientConn struct {
	ctx    context.Context
	cancel context.CancelFunc
	target    string
	authority string
	dopts     dialOptions
	mu    sync.RWMutex
	conns map[Address]*addrConn
}
func (cc *ClientConn) lbWatcher() {
	for addrs := range cc.dopts.balancer.Notify() {
		var (
			add []Address    
			del []*addrConn  
		)
		cc.mu.Lock()
		for _, a := range addrs {
			if _, ok := cc.conns[a]; !ok {
				add = append(add, a)
			}
		}
		for k, c := range cc.conns {
			var keep bool
			for _, a := range addrs {
				if k == a {
					keep = true
					break
				}
			}
			if !keep {
				del = append(del, c)
				delete(cc.conns, c.addr)
			}
		}
		cc.mu.Unlock()
		for _, a := range add {
			cc.resetAddrConn(a, true, nil)
		}
		for _, c := range del {
			c.tearDown(errConnDrain)
		}
	}
}
func (cc *ClientConn) resetAddrConn(addr Address, skipWait bool, tearDownErr error) error {
	ac := &addrConn{
		cc:    cc,
		addr:  addr,
		dopts: cc.dopts,
	}
	ac.ctx, ac.cancel = context.WithCancel(cc.ctx)
	ac.stateCV = sync.NewCond(&ac.mu)
	if EnableTracing {
		ac.events = trace.NewEventLog("grpc.ClientConn", ac.addr.Addr)
	}
	if !ac.dopts.insecure {
		if ac.dopts.copts.TransportCredentials == nil {
			return errNoTransportSecurity
		}
	} else {
		if ac.dopts.copts.TransportCredentials != nil {
			return errCredentialsConflict
		}
		for _, cd := range ac.dopts.copts.PerRPCCredentials {
			if cd.RequireTransportSecurity() {
				return errTransportCredentialsMissing
			}
		}
	}
	cc.mu.Lock()
	if cc.conns == nil {
		cc.mu.Unlock()
		return ErrClientConnClosing
	}
	stale := cc.conns[ac.addr]
	cc.conns[ac.addr] = ac
	cc.mu.Unlock()
	if stale != nil {
		if tearDownErr == nil {
			stale.tearDown(errConnDrain)
		} else {
			stale.tearDown(tearDownErr)
		}
	}
	if ac.dopts.block && !skipWait {
		if err := ac.resetTransport(false); err != nil {
			if err != errConnClosing {
				cc.mu.Lock()
				delete(cc.conns, ac.addr)
				cc.mu.Unlock()
				ac.tearDown(err)
			}
			if e, ok := err.(transport.ConnectionError); ok && !e.Temporary() {
				return e.Origin()
			}
			return err
		}
		go ac.transportMonitor()
	} else {
		go func() {
			if err := ac.resetTransport(false); err != nil {
				grpclog.Printf("Failed to dial %s: %v; please retry.", ac.addr.Addr, err)
				if err != errConnClosing {
					ac.tearDown(err)
				}
				return
			}
			ac.transportMonitor()
		}()
	}
	return nil
}
func (cc *ClientConn) getTransport(ctx context.Context, opts BalancerGetOptions) (transport.ClientTransport, func(), error) {
	var (
		ac  *addrConn
		ok  bool
		put func()
	)
	if cc.dopts.balancer == nil {
		cc.mu.RLock()
		if cc.conns == nil {
			cc.mu.RUnlock()
			return nil, nil, toRPCErr(ErrClientConnClosing)
		}
		for _, ac = range cc.conns {
			ok = true
			break
		}
		cc.mu.RUnlock()
	} else {
		var (
			addr Address
			err  error
		)
		addr, put, err = cc.dopts.balancer.Get(ctx, opts)
		if err != nil {
			return nil, nil, toRPCErr(err)
		}
		cc.mu.RLock()
		if cc.conns == nil {
			cc.mu.RUnlock()
			return nil, nil, toRPCErr(ErrClientConnClosing)
		}
		ac, ok = cc.conns[addr]
		cc.mu.RUnlock()
	}
	if !ok {
		if put != nil {
			put()
		}
		return nil, nil, errConnClosing
	}
	t, err := ac.wait(ctx, cc.dopts.balancer != nil, !opts.BlockingWait)
	if err != nil {
		if put != nil {
			put()
		}
		return nil, nil, err
	}
	return t, put, nil
}
func (cc *ClientConn) Close() error {
	cc.cancel()
	cc.mu.Lock()
	if cc.conns == nil {
		cc.mu.Unlock()
		return ErrClientConnClosing
	}
	conns := cc.conns
	cc.conns = nil
	cc.mu.Unlock()
	if cc.dopts.balancer != nil {
		cc.dopts.balancer.Close()
	}
	for _, ac := range conns {
		ac.tearDown(ErrClientConnClosing)
	}
	return nil
}
type addrConn struct {
	ctx    context.Context
	cancel context.CancelFunc
	cc     *ClientConn
	addr   Address
	dopts  dialOptions
	events trace.EventLog
	mu      sync.Mutex
	state   ConnectivityState
	stateCV *sync.Cond
	down    func(error)  
	ready     chan struct{}
	transport transport.ClientTransport
	tearDownErr error
}
func (ac *addrConn) printf(format string, a ...interface{}) {
	if ac.events != nil {
		ac.events.Printf(format, a...)
	}
}
func (ac *addrConn) errorf(format string, a ...interface{}) {
	if ac.events != nil {
		ac.events.Errorf(format, a...)
	}
}
func (ac *addrConn) getState() ConnectivityState {
	ac.mu.Lock()
	defer ac.mu.Unlock()
	return ac.state
}
func (ac *addrConn) waitForStateChange(ctx context.Context, sourceState ConnectivityState) (ConnectivityState, error) {
	ac.mu.Lock()
	defer ac.mu.Unlock()
	if sourceState != ac.state {
		return ac.state, nil
	}
	done := make(chan struct{})
	var err error
	go func() {
		select {
		case <-ctx.Done():
			ac.mu.Lock()
			err = ctx.Err()
			ac.stateCV.Broadcast()
			ac.mu.Unlock()
		case <-done:
		}
	}()
	defer close(done)
	for sourceState == ac.state {
		ac.stateCV.Wait()
		if err != nil {
			return ac.state, err
		}
	}
	return ac.state, nil
}
func (ac *addrConn) resetTransport(closeTransport bool) error {
	for retries := 0; ; retries++ {
		ac.mu.Lock()
		ac.printf("connecting")
		if ac.state == Shutdown {
			ac.mu.Unlock()
			return errConnClosing
		}
		if ac.down != nil {
			ac.down(downErrorf(false, true, "%v", errNetworkIO))
			ac.down = nil
		}
		ac.state = Connecting
		ac.stateCV.Broadcast()
		t := ac.transport
		ac.mu.Unlock()
		if closeTransport && t != nil {
			t.Close()
		}
		sleepTime := ac.dopts.bs.backoff(retries)
		timeout := minConnectTimeout
		if timeout < sleepTime {
			timeout = sleepTime
		}
		ctx, cancel := context.WithTimeout(ac.ctx, timeout)
		connectTime := time.Now()
		newTransport, err := transport.NewClientTransport(ctx, ac.addr.Addr, ac.dopts.copts)
		if err != nil {
			cancel()
			if e, ok := err.(transport.ConnectionError); ok && !e.Temporary() {
				return err
			}
			grpclog.Printf("grpc: addrConn.resetTransport failed to create client transport: %v; Reconnecting to %q", err, ac.addr)
			ac.mu.Lock()
			if ac.state == Shutdown {
				ac.mu.Unlock()
				return errConnClosing
			}
			ac.errorf("transient failure: %v", err)
			ac.state = TransientFailure
			ac.stateCV.Broadcast()
			if ac.ready != nil {
				close(ac.ready)
				ac.ready = nil
			}
			ac.mu.Unlock()
			closeTransport = false
			select {
			case <-time.After(sleepTime - time.Since(connectTime)):
			case <-ac.ctx.Done():
				return ac.ctx.Err()
			}
			continue
		}
		ac.mu.Lock()
		ac.printf("ready")
		if ac.state == Shutdown {
			ac.mu.Unlock()
			newTransport.Close()
			return errConnClosing
		}
		ac.state = Ready
		ac.stateCV.Broadcast()
		ac.transport = newTransport
		if ac.ready != nil {
			close(ac.ready)
			ac.ready = nil
		}
		if ac.cc.dopts.balancer != nil {
			ac.down = ac.cc.dopts.balancer.Up(ac.addr)
		}
		ac.mu.Unlock()
		return nil
	}
}
func (ac *addrConn) transportMonitor() {
	for {
		ac.mu.Lock()
		t := ac.transport
		ac.mu.Unlock()
		select {
		case <-ac.ctx.Done():
			select {
			case <-t.Error():
				t.Close()
			default:
			}
			return
		case <-t.GoAway():
			select {
			case <-t.Error():
				ac.cc.resetAddrConn(ac.addr, true, errNetworkIO)
			default:
				ac.cc.resetAddrConn(ac.addr, true, errConnDrain)
			}
			return
		case <-t.Error():
			select {
			case <-ac.ctx.Done():
				t.Close()
				return
			case <-t.GoAway():
				ac.cc.resetAddrConn(ac.addr, true, errNetworkIO)
				return
			default:
			}
			ac.mu.Lock()
			if ac.state == Shutdown {
				ac.mu.Unlock()
				return
			}
			ac.state = TransientFailure
			ac.stateCV.Broadcast()
			ac.mu.Unlock()
			if err := ac.resetTransport(true); err != nil {
				ac.mu.Lock()
				ac.printf("transport exiting: %v", err)
				ac.mu.Unlock()
				grpclog.Printf("grpc: addrConn.transportMonitor exits due to: %v", err)
				if err != errConnClosing {
					ac.tearDown(err)
				}
				return
			}
		}
	}
}
func (ac *addrConn) wait(ctx context.Context, hasBalancer, failfast bool) (transport.ClientTransport, error) {
	for {
		ac.mu.Lock()
		switch {
		case ac.state == Shutdown:
			if failfast || !hasBalancer {
				err := ac.tearDownErr
				ac.mu.Unlock()
				return nil, err
			}
			ac.mu.Unlock()
			return nil, errConnClosing
		case ac.state == Ready:
			ct := ac.transport
			ac.mu.Unlock()
			return ct, nil
		case ac.state == TransientFailure:
			if failfast || hasBalancer {
				ac.mu.Unlock()
				return nil, errConnUnavailable
			}
		}
		ready := ac.ready
		if ready == nil {
			ready = make(chan struct{})
			ac.ready = ready
		}
		ac.mu.Unlock()
		select {
		case <-ctx.Done():
			return nil, toRPCErr(ctx.Err())
		case <-ready:
		}
	}
}