golang源码分析：grpc 链接池（2）

func Dial(target string, opts ...DialOption) (*ClientConn, error) {
  return DialContext(context.Background(), target, opts...)
}

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),
    firstResolveEvent: grpcsync.NewEvent(),
  }
  resolverBuilder, err := cc.parseTargetAndFindResolver()
  cc.balancerWrapper = newCCBalancerWrapper(cc, balancer.BuildOptions{
    DialCreds:        credsClone,
    CredsBundle:      cc.dopts.copts.CredsBundle,
    Dialer:           cc.dopts.copts.Dialer,
    Authority:        cc.authority,
    CustomUserAgent:  cc.dopts.copts.UserAgent,
    ChannelzParentID: cc.channelzID,
    Target:           cc.parsedTarget,
  })
  rWrapper, err := newCCResolverWrapper(cc, resolverBuilder)
  if cc.dopts.block {
    for {
      cc.Connect()
      }
  }

func newCCResolverWrapper(cc *ClientConn, rb resolver.Builder) (*ccResolverWrapper, error)
        ccr := &ccResolverWrapper{
    cc:   cc,
    done: grpcsync.NewEvent(),
  }
      ccr.resolver, err = rb.Build(cc.parsedTarget, ccr, rbo)

func (cc *ClientConn) Connect() {
  cc.balancerWrapper.exitIdle()
}

func (s State) 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:
    logger.Errorf("unknown connectivity state: %d", s)
    return "INVALID_STATE"
  }
}

func newCCBalancerWrapper(cc *ClientConn, bopts balancer.BuildOptions) *ccBalancerWrapper
      ccb := &ccBalancerWrapper{
    cc:       cc,
    updateCh: buffer.NewUnbounded(),
    resultCh: buffer.NewUnbounded(),
    closed:   grpcsync.NewEvent(),
    done:     grpcsync.NewEvent(),
  }
    go ccb.watcher()
    ccb.balancer = gracefulswitch.NewBalancer(ccb, bopts)

type ccBalancerWrapper struct {
  cc *ClientConn
  // Since these fields are accessed only from handleXxx() methods which are
  // synchronized by the watcher goroutine, we do not need a mutex to protect
  // these fields.
  balancer        *gracefulswitch.Balancer
  curBalancerName string
  
  updateCh *buffer.Unbounded // Updates written on this channel are processed by watcher().
  resultCh *buffer.Unbounded // Results of calls to UpdateClientConnState() are pushed here.
  closed   *grpcsync.Event   // Indicates if close has been called.
  done     *grpcsync.Event   // Indicates if close has completed its work.
}

func (ccb *ccBalancerWrapper) watcher() {
          case u := <-ccb.updateCh.Get():
      ccb.updateCh.Load()
      if ccb.closed.HasFired() {
        break
      }
      switch update := u.(type) {
      case *ccStateUpdate:
        ccb.handleClientConnStateChange(update.ccs)

func (ccb *ccBalancerWrapper) handleSubConnStateChange(update *scStateUpdate) {
  ccb.balancer.UpdateSubConnState(update.sc, balancer.SubConnState{ConnectivityState: update.state, ConnectionError: update.err})
}

func (ccb *ccBalancerWrapper) updateSubConnState(sc balancer.SubConn, s connectivity.State, err error) {
      ccb.updateCh.Put(&scStateUpdate{

func (ccb *ccBalancerWrapper) exitIdle() {
  ccb.updateCh.Put(&exitIdleUpdate{})
}

func (ccb *ccBalancerWrapper) handleExitIdle() {
  if ccb.cc.GetState() != connectivity.Idle {
    return
  }
  ccb.balancer.ExitIdle()
}

type Unbounded struct {
  c       chan interface{}
  mu      sync.Mutex
  backlog []interface{}
}

func (b *Unbounded) Put(t interface{}) {
        if len(b.backlog) == 0 {
    select {
    case b.c <- t:
      b.backlog = append(b.backlog, t)

func (b *Unbounded) Load() {
        if len(b.backlog) > 0 {
    select {
    case b.c <- b.backlog[0]:
      b.backlog[0] = nil
      b.backlog = b.backlog[1:]

func (b *Unbounded) Get() <-chan interface{} {

func (gsb *Balancer) ExitIdle() {
        for sc := range balToUpdate.subconns {
          sc.Connect()
      }

type balancerWrapper struct {
  balancer.Balancer
  gsb *Balancer

  lastState balancer.State
  subconns  map[balancer.SubConn]bool // subconns created by this balancer
}

func (bw *balancerWrapper) UpdateSubConnState(sc balancer.SubConn, state balancer.SubConnState) {
      if state.ConnectivityState == connectivity.Shutdown {
        delete(bw.subconns, sc)
      bw.Balancer.UpdateSubConnState(sc, state)

func (gsb *Balancer) UpdateSubConnState(sc balancer.SubConn, state balancer.SubConnState) {
        if gsb.balancerCurrent != nil && gsb.balancerCurrent.subconns[sc] {
    balToUpdate = gsb.balancerCurrent
  } else if gsb.balancerPending != nil && gsb.balancerPending.subconns[sc] {
    balToUpdate = gsb.balancerPending
  }
      balToUpdate.UpdateSubConnState(sc, state)

type SubConn interface {
  // UpdateAddresses updates the addresses used in this SubConn.
  // gRPC checks if currently-connected address is still in the new list.
  // If it's in the list, the connection will be kept.
  // If it's not in the list, the connection will gracefully closed, and
  // a new connection will be created.
  //
  // This will trigger a state transition for the SubConn.
  //
  // Deprecated: This method is now part of the ClientConn interface and will
  // eventually be removed from here.
  UpdateAddresses([]resolver.Address)
  // Connect starts the connecting for this SubConn.
  Connect()
}

type Picker interface {
  // Pick returns the connection to use for this RPC and related information.
  //
  // Pick should not block.  If the balancer needs to do I/O or any blocking
  // or time-consuming work to service this call, it should return
  // ErrNoSubConnAvailable, and the Pick call will be repeated by gRPC when
  // the Picker is updated (using ClientConn.UpdateState).
  //
  // If an error is returned:
  //
  // - If the error is ErrNoSubConnAvailable, gRPC will block until a new
  //   Picker is provided by the balancer (using ClientConn.UpdateState).
  //
  // - If the error is a status error (implemented by the grpc/status
  //   package), gRPC will terminate the RPC with the code and message
  //   provided.
  //
  // - For all other errors, wait for ready RPCs will wait, but non-wait for
  //   ready RPCs will be terminated with this error's Error() string and
  //   status code Unavailable.
  Pick(info PickInfo) (PickResult, error)
}

type Balancer interface {
  // UpdateClientConnState is called by gRPC when the state of the ClientConn
  // changes.  If the error returned is ErrBadResolverState, the ClientConn
  // will begin calling ResolveNow on the active name resolver with
  // exponential backoff until a subsequent call to UpdateClientConnState
  // returns a nil error.  Any other errors are currently ignored.
  UpdateClientConnState(ClientConnState) error
  // ResolverError is called by gRPC when the name resolver reports an error.
  ResolverError(error)
  // UpdateSubConnState is called by gRPC when the state of a SubConn
  // changes.
  UpdateSubConnState(SubConn, SubConnState)
  // Close closes the balancer. The balancer is not required to call
  // ClientConn.RemoveSubConn for its existing SubConns.
  Close()
}

func (bb) Build(cc balancer.ClientConn, bOpts balancer.BuildOptions) balancer.Balancer {
  b := &ringhashBalancer{
    cc:       cc,
    subConns: make(map[resolver.Address]*subConn),
    scStates: make(map[balancer.SubConn]*subConn),
    csEvltr:  &connectivityStateEvaluator{},
  }

func (cc *ClientConn) Invoke(ctx context.Context, method string, args, reply interface{}, opts ...CallOption) error {

      return invoke(ctx, method, args, reply, cc, opts...)

func invoke(ctx context.Context, method string, req, reply interface{}, cc *ClientConn, opts ...CallOption) error {
      cs, err := newClientStream(ctx, unaryStreamDesc, cc, method, opts...)

func newClientStream(ctx context.Context, desc *StreamDesc, cc *ClientConn, method string, opts ...CallOption) (_ ClientStream, err error) {
        var newStream = func(ctx context.Context, done func()) (iresolver.ClientStream, error) {
         return newClientStreamWithParams(ctx, desc, cc, method, mc, onCommit, done, opts...)
  }

func newClientStreamWithParams(ctx context.Context, desc *StreamDesc, cc *ClientConn, method string, mc serviceconfig.MethodConfig, onCommit, doneFunc func(), opts ...CallOption) (_ iresolver.ClientStream, err error) {
        if err := cs.newAttemptLocked(false /* isTransparent */); err != nil {
    cs.finish(err)
    return nil, err
  }

func (cs *clientStream) newAttemptLocked(isTransparent bool) (retErr error) {
      t, done, err := cs.cc.getTransport(ctx, cs.callInfo.failFast, cs.callHdr.Method)

func (cc *ClientConn) getTransport(ctx context.Context, failfast bool, method string) (transport.ClientTransport, func(balancer.DoneInfo), error)
        t, done, err := cc.blockingpicker.pick(ctx, failfast, balancer.PickInfo{
    Ctx:            ctx,
    FullMethodName: method,
  })

func (pw *pickerWrapper) pick(ctx context.Context, failfast bool, info balancer.PickInfo) (transport.ClientTransport, func(balancer.DoneInfo), error) {
      pickResult, err := p.Pick(info)
      acw, ok := pickResult.SubConn.(*acBalancerWrapper)
      if t := acw.getAddrConn().getReadyTransport(); t != nil {

func (cc *ClientConn) Close() error {
      cc.csMgr.updateState(connectivity.Shutdown)
   for ac := range conns {
    ac.tearDown(ErrClientConnClosing)
  }

func (ac *addrConn) tearDown(err error) {

      tearDown doesn't remove ac from ac.cc.conns, so the addrConn struct

// will leak. In most cases, call cc.removeAddrConn() instead.

      ac.updateConnectivityState(connectivity.Shutdown, nil)

func (ac *addrConn) updateConnectivityState(s connectivity.State, lastErr error) {
      ac.state = s
      ac.cc.handleSubConnStateChange(ac.acbw, s, lastErr)

func (cc *ClientConn) handleSubConnStateChange(sc balancer.SubConn, s connectivity.State, err error) {
  cc.balancerWrapper.updateSubConnState(sc, s, err)
}