#20160

Committed 28 Jan 2026 12:21AM UTC coverage: 88.71% (+0.04%) from 88.666%

Build # #20160

Build Type

push

github

Committed by

ejona86

Commit Message

xds: Normalize weights before combining endpoint and locality weights

Previously, the number of endpoints in a locality would skew how much
traffic was sent to that locality. Also, if endpoints in localities had
wildly different weights, that would impact cross-locality weighting.

For example, consider:
  LocalityA weight=1 endpointWeights=[100, 100, 100, 100]
  LocalityB weight=1 endpointWeights=[1]

The endpoint in LocalityB should have an endpoint weight that is half
the total sum of endpoint weights, in order to receive half the traffic.
But the multiple endpoints in LocalityA would cause it to get 4x the
traffic and the endpoint weights in LocalityA causes them to get 100x
the traffic.

See gRFC A113

Run Details

35415 of 39922 relevant lines covered (88.71%)

0.89 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

98.73

/../core/src/main/java/io/grpc/internal/PickFirstLoadBalancer.java

/*
 * Copyright 2015 The 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 io.grpc.internal;

import static com.google.common.base.Preconditions.checkNotNull;
import static io.grpc.ConnectivityState.CONNECTING;
import static io.grpc.ConnectivityState.IDLE;
import static io.grpc.ConnectivityState.SHUTDOWN;
import static io.grpc.ConnectivityState.TRANSIENT_FAILURE;

import io.grpc.ConnectivityState;
import io.grpc.ConnectivityStateInfo;
import io.grpc.EquivalentAddressGroup;
import io.grpc.LoadBalancer;
import io.grpc.Status;
import java.util.List;
import java.util.Random;
import java.util.concurrent.atomic.AtomicBoolean;
import javax.annotation.Nullable;

/**
 * A {@link LoadBalancer} that provides no load-balancing over the addresses from the {@link
 * io.grpc.NameResolver}.  The channel's default behavior is used, which is walking down the address
 * list and sticking to the first that works.
 */
final class PickFirstLoadBalancer extends LoadBalancer {
  private final Helper helper;
  private Subchannel subchannel;
  private ConnectivityState currentState = IDLE;

  PickFirstLoadBalancer(Helper helper) {
    this.helper = checkNotNull(helper, "helper");
  }

  @Override
  public Status acceptResolvedAddresses(ResolvedAddresses resolvedAddresses) {
    List<EquivalentAddressGroup> servers = resolvedAddresses.getAddresses();
    if (servers.isEmpty()) {
      Status unavailableStatus = Status.UNAVAILABLE.withDescription(
              "NameResolver returned no usable address. addrs=" + resolvedAddresses.getAddresses()
                      + ", attrs=" + resolvedAddresses.getAttributes());
      handleNameResolutionError(unavailableStatus);
      return unavailableStatus;
    }

    // We can optionally be configured to shuffle the address list. This can help better distribute
    // the load.
    if (resolvedAddresses.getLoadBalancingPolicyConfig() instanceof PickFirstLoadBalancerConfig) {
      PickFirstLoadBalancerConfig config
          = (PickFirstLoadBalancerConfig) resolvedAddresses.getLoadBalancingPolicyConfig();
      if (config.shuffleAddressList != null && config.shuffleAddressList) {
        servers = PickFirstLeafLoadBalancer.shuffle(
            servers, config.randomSeed != null ? new Random(config.randomSeed) : new Random());
      }
    }

    if (subchannel == null) {
      final Subchannel subchannel = helper.createSubchannel(
          CreateSubchannelArgs.newBuilder()
              .setAddresses(servers)
              .build());
      subchannel.start(new SubchannelStateListener() {
          @Override
          public void onSubchannelState(ConnectivityStateInfo stateInfo) {
            processSubchannelState(subchannel, stateInfo);
          }
        });
      this.subchannel = subchannel;

      // The channel state does not get updated when doing name resolving today, so for the moment
      // let LB report CONNECTION and call subchannel.requestConnection() immediately.
      updateBalancingState(CONNECTING, new FixedResultPicker(PickResult.withNoResult()));
      subchannel.requestConnection();
    } else {
      subchannel.updateAddresses(servers);
    }

    return Status.OK;
  }

  @Override
  public void handleNameResolutionError(Status error) {
    if (subchannel != null) {
      subchannel.shutdown();
      subchannel = null;
    }

    // NB(lukaszx0) Whether we should propagate the error unconditionally is arguable. It's fine
    // for time being.
    updateBalancingState(TRANSIENT_FAILURE, new FixedResultPicker(PickResult.withError(error)));
  }

  private void processSubchannelState(Subchannel subchannel, ConnectivityStateInfo stateInfo) {
    ConnectivityState newState = stateInfo.getState();
    if (newState == SHUTDOWN) {
      return;
    }
    if (newState == TRANSIENT_FAILURE || newState == IDLE) {
      helper.refreshNameResolution();
    }

    // If we are transitioning from a TRANSIENT_FAILURE to CONNECTING or IDLE we ignore this state
    // transition and still keep the LB in TRANSIENT_FAILURE state. This is referred to as "sticky
    // transient failure". Only a subchannel state change to READY will get the LB out of
    // TRANSIENT_FAILURE. If the state is IDLE we additionally request a new connection so that we
    // keep retrying for a connection.
    if (currentState == TRANSIENT_FAILURE) {
      if (newState == CONNECTING) {
        return;
      } else if (newState == IDLE) {
        requestConnection();
        return;
      }
    }

    SubchannelPicker picker;
    switch (newState) {
      case IDLE:
        picker = new RequestConnectionPicker();
        break;
      case CONNECTING:
        // It's safe to use RequestConnectionPicker here, so when coming from IDLE we could leave
        // the current picker in-place. But ignoring the potential optimization is simpler.
        picker = new FixedResultPicker(PickResult.withNoResult());
        break;
      case READY:
        picker = new FixedResultPicker(PickResult.withSubchannel(subchannel));
        break;
      case TRANSIENT_FAILURE:
        picker = new FixedResultPicker(PickResult.withError(stateInfo.getStatus()));
        break;
      default:
        throw new IllegalArgumentException("Unsupported state:" + newState);
    }

    updateBalancingState(newState, picker);
  }

  private void updateBalancingState(ConnectivityState state, SubchannelPicker picker) {
    currentState = state;
    helper.updateBalancingState(state, picker);
  }

  @Override
  public void shutdown() {
    if (subchannel != null) {
      subchannel.shutdown();
    }
  }

  @Override
  public void requestConnection() {
    if (subchannel != null) {
      subchannel.requestConnection();
    }
  }

  /** Picker that requests connection during the first pick, and returns noResult. */
  private final class RequestConnectionPicker extends SubchannelPicker {
    private final AtomicBoolean connectionRequested = new AtomicBoolean(false);

    @Override
    public PickResult pickSubchannel(PickSubchannelArgs args) {
      if (connectionRequested.compareAndSet(false, true)) {
        helper.getSynchronizationContext().execute(PickFirstLoadBalancer.this::requestConnection);
      }
      return PickResult.withNoResult();
    }
  }

  public static final class PickFirstLoadBalancerConfig {

    @Nullable
    public final Boolean shuffleAddressList;

    // For testing purposes only, not meant to be parsed from a real config.
    @Nullable final Long randomSeed;

    public PickFirstLoadBalancerConfig(@Nullable Boolean shuffleAddressList) {
      this(shuffleAddressList, null);
    }

    PickFirstLoadBalancerConfig(@Nullable Boolean shuffleAddressList, @Nullable Long randomSeed) {
      this.shuffleAddressList = shuffleAddressList;
      this.randomSeed = randomSeed;
    }
  }
}

1	/*
2	* Copyright 2015 The gRPC Authors
3	*
4	* Licensed under the Apache License, Version 2.0 (the "License");
5	* you may not use this file except in compliance with the License.
6	* You may obtain a copy of the License at
7	*
8	* http://www.apache.org/licenses/LICENSE-2.0
9	*
10	* Unless required by applicable law or agreed to in writing, software
11	* distributed under the License is distributed on an "AS IS" BASIS,
12	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	* See the License for the specific language governing permissions and
14	* limitations under the License.
15	*/
16
17	package io.grpc.internal;
18
19	import static com.google.common.base.Preconditions.checkNotNull;
20	import static io.grpc.ConnectivityState.CONNECTING;
21	import static io.grpc.ConnectivityState.IDLE;
22	import static io.grpc.ConnectivityState.SHUTDOWN;
23	import static io.grpc.ConnectivityState.TRANSIENT_FAILURE;
24
25	import io.grpc.ConnectivityState;
26	import io.grpc.ConnectivityStateInfo;
27	import io.grpc.EquivalentAddressGroup;
28	import io.grpc.LoadBalancer;
29	import io.grpc.Status;
30	import java.util.List;
31	import java.util.Random;
32	import java.util.concurrent.atomic.AtomicBoolean;
33	import javax.annotation.Nullable;
34
35	/**
36	* A {@link LoadBalancer} that provides no load-balancing over the addresses from the {@link
37	* io.grpc.NameResolver}. The channel's default behavior is used, which is walking down the address
38	* list and sticking to the first that works.
39	*/
40	final class PickFirstLoadBalancer extends LoadBalancer {
41	private final Helper helper;
42	private Subchannel subchannel;
43	private ConnectivityState currentState = IDLE;	1✔
44
45	PickFirstLoadBalancer(Helper helper) {	1✔
46	this.helper = checkNotNull(helper, "helper");	1✔
47	}	1✔
48
49	@Override
50	public Status acceptResolvedAddresses(ResolvedAddresses resolvedAddresses) {
51	List<EquivalentAddressGroup> servers = resolvedAddresses.getAddresses();	1✔
52	if (servers.isEmpty()) {	1✔
53	Status unavailableStatus = Status.UNAVAILABLE.withDescription(	1✔
54	"NameResolver returned no usable address. addrs=" + resolvedAddresses.getAddresses()	1✔
55	+ ", attrs=" + resolvedAddresses.getAttributes());	1✔
56	handleNameResolutionError(unavailableStatus);	1✔
57	return unavailableStatus;	1✔
58	}
59
60	// We can optionally be configured to shuffle the address list. This can help better distribute
61	// the load.
62	if (resolvedAddresses.getLoadBalancingPolicyConfig() instanceof PickFirstLoadBalancerConfig) {	1✔
63	PickFirstLoadBalancerConfig config	1✔
64	= (PickFirstLoadBalancerConfig) resolvedAddresses.getLoadBalancingPolicyConfig();	1✔
65	if (config.shuffleAddressList != null && config.shuffleAddressList) {	1✔
66	servers = PickFirstLeafLoadBalancer.shuffle(	1✔
67	servers, config.randomSeed != null ? new Random(config.randomSeed) : new Random());	1✔
68	}
69	}
70
71	if (subchannel == null) {	1✔
72	final Subchannel subchannel = helper.createSubchannel(	1✔
73	CreateSubchannelArgs.newBuilder()	1✔
74	.setAddresses(servers)	1✔
75	.build());	1✔
76	subchannel.start(new SubchannelStateListener() {	1✔
77	@Override
78	public void onSubchannelState(ConnectivityStateInfo stateInfo) {
79	processSubchannelState(subchannel, stateInfo);	1✔
80	}	1✔
81	});
82	this.subchannel = subchannel;	1✔
83
84	// The channel state does not get updated when doing name resolving today, so for the moment
85	// let LB report CONNECTION and call subchannel.requestConnection() immediately.
86	updateBalancingState(CONNECTING, new FixedResultPicker(PickResult.withNoResult()));	1✔
87	subchannel.requestConnection();	1✔
88	} else {	1✔
89	subchannel.updateAddresses(servers);	1✔
90	}
91
92	return Status.OK;	1✔
93	}
94
95	@Override
96	public void handleNameResolutionError(Status error) {
97	if (subchannel != null) {	1✔
98	subchannel.shutdown();	1✔
99	subchannel = null;	1✔
100	}
101
102	// NB(lukaszx0) Whether we should propagate the error unconditionally is arguable. It's fine
103	// for time being.
104	updateBalancingState(TRANSIENT_FAILURE, new FixedResultPicker(PickResult.withError(error)));	1✔
105	}	1✔
106
107	private void processSubchannelState(Subchannel subchannel, ConnectivityStateInfo stateInfo) {
108	ConnectivityState newState = stateInfo.getState();	1✔
109	if (newState == SHUTDOWN) {	1✔
110	return;	1✔
111	}
112	if (newState == TRANSIENT_FAILURE \|\| newState == IDLE) {	1✔
113	helper.refreshNameResolution();	1✔
114	}
115
116	// If we are transitioning from a TRANSIENT_FAILURE to CONNECTING or IDLE we ignore this state
117	// transition and still keep the LB in TRANSIENT_FAILURE state. This is referred to as "sticky
118	// transient failure". Only a subchannel state change to READY will get the LB out of
119	// TRANSIENT_FAILURE. If the state is IDLE we additionally request a new connection so that we
120	// keep retrying for a connection.
121	if (currentState == TRANSIENT_FAILURE) {	1✔
122	if (newState == CONNECTING) {	1✔
123	return;	1✔
124	} else if (newState == IDLE) {	1✔
125	requestConnection();	1✔
126	return;	1✔
127	}
128	}
129
130	SubchannelPicker picker;
131	switch (newState) {	1✔
132	case IDLE:
133	picker = new RequestConnectionPicker();	1✔
134	break;	1✔
135	case CONNECTING:
136	// It's safe to use RequestConnectionPicker here, so when coming from IDLE we could leave
137	// the current picker in-place. But ignoring the potential optimization is simpler.
138	picker = new FixedResultPicker(PickResult.withNoResult());	1✔
139	break;	1✔
140	case READY:
141	picker = new FixedResultPicker(PickResult.withSubchannel(subchannel));	1✔
142	break;	1✔
143	case TRANSIENT_FAILURE:
144	picker = new FixedResultPicker(PickResult.withError(stateInfo.getStatus()));	1✔
145	break;	1✔
146	default:
147	throw new IllegalArgumentException("Unsupported state:" + newState);	×
148	}
149
150	updateBalancingState(newState, picker);	1✔
151	}	1✔
152
153	private void updateBalancingState(ConnectivityState state, SubchannelPicker picker) {
154	currentState = state;	1✔
155	helper.updateBalancingState(state, picker);	1✔
156	}	1✔
157
158	@Override
159	public void shutdown() {
160	if (subchannel != null) {	1✔
161	subchannel.shutdown();	1✔
162	}
163	}	1✔
164
165	@Override
166	public void requestConnection() {
167	if (subchannel != null) {	1✔
168	subchannel.requestConnection();	1✔
169	}
170	}	1✔
171
172	/** Picker that requests connection during the first pick, and returns noResult. */
173	private final class RequestConnectionPicker extends SubchannelPicker {	1✔
174	private final AtomicBoolean connectionRequested = new AtomicBoolean(false);	1✔
175
176	@Override
177	public PickResult pickSubchannel(PickSubchannelArgs args) {
178	if (connectionRequested.compareAndSet(false, true)) {	1✔
179	helper.getSynchronizationContext().execute(PickFirstLoadBalancer.this::requestConnection);	1✔
180	}
181	return PickResult.withNoResult();	1✔
182	}
183	}
184
185	public static final class PickFirstLoadBalancerConfig {
186
187	@Nullable
188	public final Boolean shuffleAddressList;
189
190	// For testing purposes only, not meant to be parsed from a real config.
191	@Nullable final Long randomSeed;
192
193	public PickFirstLoadBalancerConfig(@Nullable Boolean shuffleAddressList) {
194	this(shuffleAddressList, null);	1✔
195	}	1✔
196
197	PickFirstLoadBalancerConfig(@Nullable Boolean shuffleAddressList, @Nullable Long randomSeed) {	1✔
198	this.shuffleAddressList = shuffleAddressList;	1✔
199	this.randomSeed = randomSeed;	1✔
200	}	1✔
201	}
202	}

grpc / grpc-java / #20160

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous