Method UpdateState

balancer/leastrequest/leastrequest.go:153–211

(state balancer.State)

}

func (lrb *leastRequestBalancer) UpdateState(state balancer.State) {
	var readyEndpoints []endpointsharding.ChildState
	for _, child := range endpointsharding.ChildStatesFromPicker(state.Picker) {
		if child.State.ConnectivityState == connectivity.Ready {
			readyEndpoints = append(readyEndpoints, child)
		}
	}

	// If no ready pickers are present, simply defer to the round robin picker
	// from endpoint sharding, which will round robin across the most relevant
	// pick first children in the highest precedence connectivity state.
	if len(readyEndpoints) == 0 {
		lrb.ClientConn.UpdateState(state)
		return
	}

	lrb.mu.Lock()
	defer lrb.mu.Unlock()

	if logger.V(2) {
		lrb.logger.Infof("UpdateState called with ready endpoints: %v", readyEndpoints)
	}

	// Reconcile endpoints.
	newEndpoints := resolver.NewEndpointMap[any]()
	for _, child := range readyEndpoints {
		newEndpoints.Set(child.Endpoint, nil)
	}

	// If endpoints are no longer ready, no need to count their active RPCs.
	for endpoint := range lrb.endpointRPCCounts.All() {
		if _, ok := newEndpoints.Get(endpoint); !ok {
			lrb.endpointRPCCounts.Delete(endpoint)
		}
	}

	// Copy refs to counters into picker.
	endpointStates := make([]endpointState, 0, len(readyEndpoints))
	for _, child := range readyEndpoints {
		counter, ok := lrb.endpointRPCCounts.Get(child.Endpoint)
		if !ok {
			// Create new counts if needed.
			counter = new(atomic.Int32)
			lrb.endpointRPCCounts.Set(child.Endpoint, counter)
		}
		endpointStates = append(endpointStates, endpointState{
			picker:  child.State.Picker,
			numRPCs: counter,
		})
	}

	lrb.ClientConn.UpdateState(balancer.State{
		Picker: &picker{
			choiceCount:    lrb.choiceCount,
			endpointStates: endpointStates,
		},
		ConnectivityState: connectivity.Ready,
	})