Repository: grpc/grpc-go
File: balancer/pickfirst/pickfirstleaf/pickfirstleaf.go (Go)

Code before:
	// this SubConn.
	if activeSD, found := b.subConns.Get(sd.addr); !found || activeSD != sd {
		return
	}
	switch state.ConnectivityState {
	case connectivity.Ready:
		b.updateConcludedStateLocked(balancer.State{
			ConnectivityState: connectivity.Ready,
			Picker:            &picker{result: balancer.PickResult{SubConn: sd.subConn}},
		})
	case connectivity.TransientFailure:
		b.updateConcludedStateLocked(balancer.State{
			ConnectivityState: connectivity.TransientFailure,
			Picker:            &picker{err: fmt.Errorf("health check failure: %v", state.ConnectionError)},
		})
	case connectivity.Connecting:
		b.updateConcludedStateLocked(balancer.State{
			ConnectivityState: connectivity.Connecting,
			Picker:            &picker{err: balancer.ErrNoSubConnAvailable},
		})
	default:
		b.logger.Errorf("Got unexpected health update for SubConn %p: %v", state)
	}
}

func (b *pickfirstBalancer) updateConcludedStateLocked(newState balancer.State) {
	// Optimization to not send duplicate CONNECTING and IDLE updates.
	if newState.ConnectivityState == b.concludedState && b.concludedState == connectivity.Connecting {
		return
	}
	b.forceUpdateConcludedStateLocked(newState)
}

// A separate function to force update the ClientConn state is required to send
// the first CONNECTING update since the channel doesn't correctly assume that
// LB policies start in CONNECTING and relies on LB policy to send an initial
// CONNECTING update.
func (b *pickfirstBalancer) forceUpdateConcludedStateLocked(newState balancer.State) {
	b.concludedState = newState.ConnectivityState
	b.cc.UpdateState(newState)
}

type picker struct {
	result balancer.PickResult
	err    error
}

func (p *picker) Pick(balancer.PickInfo) (balancer.PickResult, error) {
	return p.result, p.err
}


Review comment (suggestion):
Docstring for this one please. And here and in the next method, also please clearly indicate that a new picker is sent. Because "update concluded state" could simply also mean updating the `concludedState` field. Thanks.

Code after:
		// TODO: #7534 - Move re-resolution requests from SubConn into
		// pick_first.
	case connectivity.Idle:
		sd.subConn.Connect()
	}
}

// endFirstPassIfPossibleLocked ends the first happy-eyeballs pass if all the
// addresses are tried and their SubConns have reported a failure.
func (b *pickfirstBalancer) endFirstPassIfPossibleLocked(lastErr error) {
	// An optimization to avoid iterating over the entire SubConn map.
	if b.addressList.isValid() {
		return
	}
	// Connect() has been called on all the SubConns. The first pass can be
	// ended if all the SubConns have reported a failure.
	for _, v := range b.subConns.Values() {
		sd := v.(*scData)
		if !sd.connectionFailedInFirstPass {
			return
		}
	}
	b.firstPass = false
	b.updateBalancerState(balancer.State{
		ConnectivityState: connectivity.TransientFailure,
		Picker:            &picker{err: lastErr},
	})
	// Start re-connecting all the SubConns that are already in IDLE.
	for _, v := range b.subConns.Values() {
		sd := v.(*scData)
		if sd.rawConnectivityState == connectivity.Idle {
			sd.subConn.Connect()
		}
	}
}

func (b *pickfirstBalancer) isActiveSCData(sd *scData) bool {
	activeSD, found := b.subConns.Get(sd.addr)
	return found && activeSD == sd
}

func (b *pickfirstBalancer) updateSubConnHealthState(sd *scData, state balancer.SubConnState) {
	b.mu.Lock()
	defer b.mu.Unlock()
	// Previously relevant SubConns can still callback with state updates.
	// To prevent pickers from returning these obsolete SubConns, this logic
	// is included to check if the current list of active SubConns includes
	// this SubConn.
	if !b.isActiveSCData(sd) {
		return
	}