#19102

Committed 14 Mar 2024 03:53AM UTC coverage: 88.259% (-0.05%) from 88.311%

Build # #19102

Build Type

push

github

Committed by

web-flow

Commit Message

core: Eliminate NPE seen in PickFirstLeafLoadBalancer (#11013)

ref b/329420531

Run Details

31150 of 35294 relevant lines covered (88.26%)

0.88 hits per line

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

93.26

/../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 com.google.common.base.MoreObjects;
import io.grpc.ConnectivityState;
import io.grpc.ConnectivityStateInfo;
import io.grpc.EquivalentAddressGroup;
import io.grpc.LoadBalancer;
import io.grpc.Status;
import java.util.ArrayList;
import java.util.Collections;
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 = new ArrayList<EquivalentAddressGroup>(servers);
        Collections.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 Picker(PickResult.withSubchannel(subchannel)));
      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 Picker(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(subchannel);
        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 Picker(PickResult.withNoResult());
        break;
      case READY:
        picker = new Picker(PickResult.withSubchannel(subchannel));
        break;
      case TRANSIENT_FAILURE:
        picker = new Picker(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();
    }
  }

  /**
   * No-op picker which doesn't add any custom picking logic. It just passes already known result
   * received in constructor.
   */
  private static final class Picker extends SubchannelPicker {
    private final PickResult result;

    Picker(PickResult result) {
      this.result = checkNotNull(result, "result");
    }

    @Override
    public PickResult pickSubchannel(PickSubchannelArgs args) {
      return result;
    }

    @Override
    public String toString() {
      return MoreObjects.toStringHelper(Picker.class).add("result", result).toString();
    }
  }

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

    RequestConnectionPicker(Subchannel subchannel) {
      this.subchannel = checkNotNull(subchannel, "subchannel");
    }

    @Override
    public PickResult pickSubchannel(PickSubchannelArgs args) {
      if (connectionRequested.compareAndSet(false, true)) {
        helper.getSynchronizationContext().execute(new Runnable() {
            @Override
            public void run() {
              subchannel.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 com.google.common.base.MoreObjects;
26	import io.grpc.ConnectivityState;
27	import io.grpc.ConnectivityStateInfo;
28	import io.grpc.EquivalentAddressGroup;
29	import io.grpc.LoadBalancer;
30	import io.grpc.Status;
31	import java.util.ArrayList;
32	import java.util.Collections;
33	import java.util.List;
34	import java.util.Random;
35	import java.util.concurrent.atomic.AtomicBoolean;
36	import javax.annotation.Nullable;
37
38	/**
39	* A {@link LoadBalancer} that provides no load-balancing over the addresses from the {@link
40	* io.grpc.NameResolver}. The channel's default behavior is used, which is walking down the address
41	* list and sticking to the first that works.
42	*/
43	final class PickFirstLoadBalancer extends LoadBalancer {
44	private final Helper helper;
45	private Subchannel subchannel;
46	private ConnectivityState currentState = IDLE;	1✔
47
48	PickFirstLoadBalancer(Helper helper) {	1✔
49	this.helper = checkNotNull(helper, "helper");	1✔
50	}	1✔
51
52	@Override
53	public Status acceptResolvedAddresses(ResolvedAddresses resolvedAddresses) {
54	List<EquivalentAddressGroup> servers = resolvedAddresses.getAddresses();	1✔
55	if (servers.isEmpty()) {	1✔
56	Status unavailableStatus = Status.UNAVAILABLE.withDescription(	1✔
57	"NameResolver returned no usable address. addrs=" + resolvedAddresses.getAddresses()	1✔
58	+ ", attrs=" + resolvedAddresses.getAttributes());	1✔
59	handleNameResolutionError(unavailableStatus);	1✔
60	return unavailableStatus;	1✔
61	}
62
63	// We can optionally be configured to shuffle the address list. This can help better distribute
64	// the load.
65	if (resolvedAddresses.getLoadBalancingPolicyConfig() instanceof PickFirstLoadBalancerConfig) {	1✔
66	PickFirstLoadBalancerConfig config	1✔
67	= (PickFirstLoadBalancerConfig) resolvedAddresses.getLoadBalancingPolicyConfig();	1✔
68	if (config.shuffleAddressList != null && config.shuffleAddressList) {	1✔
69	servers = new ArrayList<EquivalentAddressGroup>(servers);	1✔
70	Collections.shuffle(servers,	1✔
71	config.randomSeed != null ? new Random(config.randomSeed) : new Random());	1✔
72	}
73	}
74
75	if (subchannel == null) {	1✔
76	final Subchannel subchannel = helper.createSubchannel(	1✔
77	CreateSubchannelArgs.newBuilder()	1✔
78	.setAddresses(servers)	1✔
79	.build());	1✔
80	subchannel.start(new SubchannelStateListener() {	1✔
81	@Override
82	public void onSubchannelState(ConnectivityStateInfo stateInfo) {
83	processSubchannelState(subchannel, stateInfo);	1✔
84	}	1✔
85	});
86	this.subchannel = subchannel;	1✔
87
88	// The channel state does not get updated when doing name resolving today, so for the moment
89	// let LB report CONNECTION and call subchannel.requestConnection() immediately.
90	updateBalancingState(CONNECTING, new Picker(PickResult.withSubchannel(subchannel)));	1✔
91	subchannel.requestConnection();	1✔
92	} else {	1✔
93	subchannel.updateAddresses(servers);	1✔
94	}
95
96	return Status.OK;	1✔
97	}
98
99	@Override
100	public void handleNameResolutionError(Status error) {
101	if (subchannel != null) {	1✔
102	subchannel.shutdown();	1✔
103	subchannel = null;	1✔
104	}
105
106	// NB(lukaszx0) Whether we should propagate the error unconditionally is arguable. It's fine
107	// for time being.
108	updateBalancingState(TRANSIENT_FAILURE, new Picker(PickResult.withError(error)));	1✔
109	}	1✔
110
111	private void processSubchannelState(Subchannel subchannel, ConnectivityStateInfo stateInfo) {
112	ConnectivityState newState = stateInfo.getState();	1✔
113	if (newState == SHUTDOWN) {	1✔
114	return;	×
115	}
116	if (newState == TRANSIENT_FAILURE \|\| newState == IDLE) {	1✔
117	helper.refreshNameResolution();	1✔
118	}
119
120	// If we are transitioning from a TRANSIENT_FAILURE to CONNECTING or IDLE we ignore this state
121	// transition and still keep the LB in TRANSIENT_FAILURE state. This is referred to as "sticky
122	// transient failure". Only a subchannel state change to READY will get the LB out of
123	// TRANSIENT_FAILURE. If the state is IDLE we additionally request a new connection so that we
124	// keep retrying for a connection.
125	if (currentState == TRANSIENT_FAILURE) {	1✔
126	if (newState == CONNECTING) {	1✔
127	return;	1✔
128	} else if (newState == IDLE) {	1✔
129	requestConnection();	1✔
130	return;	1✔
131	}
132	}
133
134	SubchannelPicker picker;
135	switch (newState) {	1✔
136	case IDLE:
137	picker = new RequestConnectionPicker(subchannel);	1✔
138	break;	1✔
139	case CONNECTING:
140	// It's safe to use RequestConnectionPicker here, so when coming from IDLE we could leave
141	// the current picker in-place. But ignoring the potential optimization is simpler.
142	picker = new Picker(PickResult.withNoResult());	1✔
143	break;	1✔
144	case READY:
145	picker = new Picker(PickResult.withSubchannel(subchannel));	1✔
146	break;	1✔
147	case TRANSIENT_FAILURE:
148	picker = new Picker(PickResult.withError(stateInfo.getStatus()));	1✔
149	break;	1✔
150	default:
151	throw new IllegalArgumentException("Unsupported state:" + newState);	×
152	}
153
154	updateBalancingState(newState, picker);	1✔
155	}	1✔
156
157	private void updateBalancingState(ConnectivityState state, SubchannelPicker picker) {
158	currentState = state;	1✔
159	helper.updateBalancingState(state, picker);	1✔
160	}	1✔
161
162	@Override
163	public void shutdown() {
164	if (subchannel != null) {	×
165	subchannel.shutdown();	×
166	}
167	}	×
168
169	@Override
170	public void requestConnection() {
171	if (subchannel != null) {	1✔
172	subchannel.requestConnection();	1✔
173	}
174	}	1✔
175
176	/**
177	* No-op picker which doesn't add any custom picking logic. It just passes already known result
178	* received in constructor.
179	*/
180	private static final class Picker extends SubchannelPicker {
181	private final PickResult result;
182
183	Picker(PickResult result) {	1✔
184	this.result = checkNotNull(result, "result");	1✔
185	}	1✔
186
187	@Override
188	public PickResult pickSubchannel(PickSubchannelArgs args) {
189	return result;	1✔
190	}
191
192	@Override
193	public String toString() {
194	return MoreObjects.toStringHelper(Picker.class).add("result", result).toString();	×
195	}
196	}
197
198	/** Picker that requests connection during the first pick, and returns noResult. */
199	private final class RequestConnectionPicker extends SubchannelPicker {
200	private final Subchannel subchannel;
201	private final AtomicBoolean connectionRequested = new AtomicBoolean(false);	1✔
202
203	RequestConnectionPicker(Subchannel subchannel) {	1✔
204	this.subchannel = checkNotNull(subchannel, "subchannel");	1✔
205	}	1✔
206
207	@Override
208	public PickResult pickSubchannel(PickSubchannelArgs args) {
209	if (connectionRequested.compareAndSet(false, true)) {	1✔
210	helper.getSynchronizationContext().execute(new Runnable() {	1✔
211	@Override
212	public void run() {
213	subchannel.requestConnection();	1✔
214	}	1✔
215	});
216	}
217	return PickResult.withNoResult();	1✔
218	}
219	}
220
221	public static final class PickFirstLoadBalancerConfig {
222
223	@Nullable
224	public final Boolean shuffleAddressList;
225
226	// For testing purposes only, not meant to be parsed from a real config.
227	@Nullable final Long randomSeed;
228
229	public PickFirstLoadBalancerConfig(@Nullable Boolean shuffleAddressList) {
230	this(shuffleAddressList, null);	1✔
231	}	1✔
232
233	PickFirstLoadBalancerConfig(@Nullable Boolean shuffleAddressList, @Nullable Long randomSeed) {	1✔
234	this.shuffleAddressList = shuffleAddressList;	1✔
235	this.randomSeed = randomSeed;	1✔
236	}	1✔
237	}
238	}

grpc / grpc-java / #19102

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