• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

grpc / grpc-java / #19406

07 Aug 2024 12:12AM CUT coverage: 84.485% (+0.02%) from 84.464%
#19406

push

github

web-flow
Revert "Enable new PickFirst LB (#11348)" (#11425) (#11448)

This reverts commit ccfd351a2.

33249 of 39355 relevant lines covered (84.48%)

0.84 hits per line

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

95.71
/../core/src/main/java/io/grpc/internal/PickFirstLeafLoadBalancer.java
1
/*
2
 * Copyright 2023 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.READY;
23
import static io.grpc.ConnectivityState.SHUTDOWN;
24
import static io.grpc.ConnectivityState.TRANSIENT_FAILURE;
25

26
import com.google.common.annotations.VisibleForTesting;
27
import com.google.common.base.MoreObjects;
28
import com.google.common.collect.ImmutableList;
29
import com.google.common.collect.Lists;
30
import io.grpc.Attributes;
31
import io.grpc.ConnectivityState;
32
import io.grpc.ConnectivityStateInfo;
33
import io.grpc.EquivalentAddressGroup;
34
import io.grpc.LoadBalancer;
35
import io.grpc.Status;
36
import io.grpc.SynchronizationContext;
37
import io.grpc.SynchronizationContext.ScheduledHandle;
38
import java.net.SocketAddress;
39
import java.util.ArrayList;
40
import java.util.Collections;
41
import java.util.HashMap;
42
import java.util.HashSet;
43
import java.util.List;
44
import java.util.Map;
45
import java.util.Random;
46
import java.util.Set;
47
import java.util.concurrent.TimeUnit;
48
import java.util.concurrent.atomic.AtomicBoolean;
49
import java.util.logging.Level;
50
import java.util.logging.Logger;
51
import javax.annotation.Nullable;
52

53
/**
54
 * A {@link LoadBalancer} that provides no load-balancing over the addresses from the {@link
55
 * io.grpc.NameResolver}. The channel's default behavior is used, which is walking down the address
56
 * list and sticking to the first that works.
57
 */
58
final class PickFirstLeafLoadBalancer extends LoadBalancer {
59
  private static final Logger log = Logger.getLogger(PickFirstLeafLoadBalancer.class.getName());
1✔
60
  @VisibleForTesting
61
  static final int CONNECTION_DELAY_INTERVAL_MS = 250;
62
  private final Helper helper;
63
  private final Map<SocketAddress, SubchannelData> subchannels = new HashMap<>();
1✔
64
  private Index addressIndex;
65
  private int numTf = 0;
1✔
66
  private boolean firstPass = true;
1✔
67
  @Nullable
68
  private ScheduledHandle scheduleConnectionTask;
69
  private ConnectivityState rawConnectivityState = IDLE;
1✔
70
  private ConnectivityState concludedState = IDLE;
1✔
71
  private final boolean enableHappyEyeballs =
1✔
72
      PickFirstLoadBalancerProvider.isEnabledHappyEyeballs();
1✔
73

74
  PickFirstLeafLoadBalancer(Helper helper) {
1✔
75
    this.helper = checkNotNull(helper, "helper");
1✔
76
  }
1✔
77

78
  @Override
79
  public Status acceptResolvedAddresses(ResolvedAddresses resolvedAddresses) {
80
    if (rawConnectivityState == SHUTDOWN) {
1✔
81
      return Status.FAILED_PRECONDITION.withDescription("Already shut down");
1✔
82
    }
83

84
    List<EquivalentAddressGroup> servers = resolvedAddresses.getAddresses();
1✔
85

86
    // Validate the address list
87
    if (servers.isEmpty()) {
1✔
88
      Status unavailableStatus = Status.UNAVAILABLE.withDescription(
1✔
89
              "NameResolver returned no usable address. addrs=" + resolvedAddresses.getAddresses()
1✔
90
                      + ", attrs=" + resolvedAddresses.getAttributes());
1✔
91
      handleNameResolutionError(unavailableStatus);
1✔
92
      return unavailableStatus;
1✔
93
    }
94
    for (EquivalentAddressGroup eag : servers) {
1✔
95
      if (eag == null) {
1✔
96
        Status unavailableStatus = Status.UNAVAILABLE.withDescription(
1✔
97
            "NameResolver returned address list with null endpoint. addrs="
98
                + resolvedAddresses.getAddresses() + ", attrs="
1✔
99
                + resolvedAddresses.getAttributes());
1✔
100
        handleNameResolutionError(unavailableStatus);
1✔
101
        return unavailableStatus;
1✔
102
      }
103
    }
1✔
104

105
    // Since we have a new set of addresses, we are again at first pass
106
    firstPass = true;
1✔
107

108
    List<EquivalentAddressGroup> cleanServers = deDupAddresses(servers);
1✔
109

110
    // We can optionally be configured to shuffle the address list. This can help better distribute
111
    // the load.
112
    if (resolvedAddresses.getLoadBalancingPolicyConfig()
1✔
113
        instanceof PickFirstLeafLoadBalancerConfig) {
114
      PickFirstLeafLoadBalancerConfig config
1✔
115
          = (PickFirstLeafLoadBalancerConfig) resolvedAddresses.getLoadBalancingPolicyConfig();
1✔
116
      if (config.shuffleAddressList != null && config.shuffleAddressList) {
1✔
117
        Collections.shuffle(cleanServers,
1✔
118
            config.randomSeed != null ? new Random(config.randomSeed) : new Random());
1✔
119
      }
120
    }
121

122
    final ImmutableList<EquivalentAddressGroup> newImmutableAddressGroups =
123
        ImmutableList.<EquivalentAddressGroup>builder().addAll(cleanServers).build();
1✔
124

125
    if (addressIndex == null) {
1✔
126
      addressIndex = new Index(newImmutableAddressGroups);
1✔
127
    } else if (rawConnectivityState == READY) {
1✔
128
      // If the previous ready subchannel exists in new address list,
129
      // keep this connection and don't create new subchannels
130
      SocketAddress previousAddress = addressIndex.getCurrentAddress();
1✔
131
      addressIndex.updateGroups(newImmutableAddressGroups);
1✔
132
      if (addressIndex.seekTo(previousAddress)) {
1✔
133
        SubchannelData subchannelData = subchannels.get(previousAddress);
1✔
134
        subchannelData.getSubchannel().updateAddresses(addressIndex.getCurrentEagAsList());
1✔
135
        return Status.OK;
1✔
136
      } else {
137
        addressIndex.reset(); // Previous ready subchannel not in the new list of addresses
1✔
138
      }
139
    } else {
1✔
140
      addressIndex.updateGroups(newImmutableAddressGroups);
1✔
141
    }
142

143
    // remove old subchannels that were not in new address list
144
    Set<SocketAddress> oldAddrs = new HashSet<>(subchannels.keySet());
1✔
145

146
    // Flatten the new EAGs addresses
147
    Set<SocketAddress> newAddrs = new HashSet<>();
1✔
148
    for (EquivalentAddressGroup endpoint : newImmutableAddressGroups) {
1✔
149
      newAddrs.addAll(endpoint.getAddresses());
1✔
150
    }
1✔
151

152
    // Shut them down and remove them
153
    for (SocketAddress oldAddr : oldAddrs) {
1✔
154
      if (!newAddrs.contains(oldAddr)) {
1✔
155
        subchannels.remove(oldAddr).getSubchannel().shutdown();
1✔
156
      }
157
    }
1✔
158

159
    if (oldAddrs.size() == 0 || rawConnectivityState == CONNECTING
1✔
160
        || rawConnectivityState == READY) {
161
      // start connection attempt at first address
162
      rawConnectivityState = CONNECTING;
1✔
163
      updateBalancingState(CONNECTING, new Picker(PickResult.withNoResult()));
1✔
164
      cancelScheduleTask();
1✔
165
      requestConnection();
1✔
166

167
    } else if (rawConnectivityState == IDLE) {
1✔
168
      // start connection attempt at first address when requested
169
      SubchannelPicker picker = new RequestConnectionPicker(this);
1✔
170
      updateBalancingState(IDLE, picker);
1✔
171

172
    } else if (rawConnectivityState == TRANSIENT_FAILURE) {
1✔
173
      // start connection attempt at first address
174
      cancelScheduleTask();
1✔
175
      requestConnection();
1✔
176
    }
177

178
    return Status.OK;
1✔
179
  }
180

181
  private static List<EquivalentAddressGroup> deDupAddresses(List<EquivalentAddressGroup> groups) {
182
    Set<SocketAddress> seenAddresses = new HashSet<>();
1✔
183
    List<EquivalentAddressGroup> newGroups = new ArrayList<>();
1✔
184

185
    for (EquivalentAddressGroup group : groups) {
1✔
186
      List<SocketAddress> addrs = new ArrayList<>();
1✔
187
      for (SocketAddress addr : group.getAddresses()) {
1✔
188
        if (seenAddresses.add(addr)) {
1✔
189
          addrs.add(addr);
1✔
190
        }
191
      }
1✔
192
      if (!addrs.isEmpty()) {
1✔
193
        newGroups.add(new EquivalentAddressGroup(addrs, group.getAttributes()));
1✔
194
      }
195
    }
1✔
196

197
    return newGroups;
1✔
198
  }
199

200
  @Override
201
  public void handleNameResolutionError(Status error) {
202
    if (rawConnectivityState == SHUTDOWN) {
1✔
203
      return;
×
204
    }
205

206
    for (SubchannelData subchannelData : subchannels.values()) {
1✔
207
      subchannelData.getSubchannel().shutdown();
1✔
208
    }
1✔
209
    subchannels.clear();
1✔
210
    if (addressIndex != null) {
1✔
211
      addressIndex.updateGroups(null);
1✔
212
    }
213
    rawConnectivityState = TRANSIENT_FAILURE;
1✔
214
    updateBalancingState(TRANSIENT_FAILURE, new Picker(PickResult.withError(error)));
1✔
215
  }
1✔
216

217
  void processSubchannelState(Subchannel subchannel, ConnectivityStateInfo stateInfo) {
218
    ConnectivityState newState = stateInfo.getState();
1✔
219

220
    SubchannelData subchannelData = subchannels.get(getAddress(subchannel));
1✔
221
    // Shutdown channels/previously relevant subchannels can still callback with state updates.
222
    // To prevent pickers from returning these obsolete subchannels, this logic
223
    // is included to check if the current list of active subchannels includes this subchannel.
224
    if (subchannelData == null || subchannelData.getSubchannel() != subchannel) {
1✔
225
      return;
1✔
226
    }
227

228
    if (newState == SHUTDOWN) {
1✔
229
      return;
×
230
    }
231

232
    if (newState == IDLE) {
1✔
233
      helper.refreshNameResolution();
1✔
234
    }
235
    // If we are transitioning from a TRANSIENT_FAILURE to CONNECTING or IDLE we ignore this state
236
    // transition and still keep the LB in TRANSIENT_FAILURE state. This is referred to as "sticky
237
    // transient failure". Only a subchannel state change to READY will get the LB out of
238
    // TRANSIENT_FAILURE. If the state is IDLE we additionally request a new connection so that we
239
    // keep retrying for a connection.
240

241
    // With the new pick first implementation, individual subchannels will have their own backoff
242
    // on a per-address basis. Thus, iterative requests for connections will not be requested
243
    // once the first pass through is complete.
244
    // However, every time there is an address update, we will perform a pass through for the new
245
    // addresses in the updated list.
246
    subchannelData.updateState(newState);
1✔
247
    if (rawConnectivityState == TRANSIENT_FAILURE || concludedState == TRANSIENT_FAILURE)  {
1✔
248
      if (newState == CONNECTING) {
1✔
249
        // each subchannel is responsible for its own backoff
250
        return;
1✔
251
      } else if (newState == IDLE) {
1✔
252
        requestConnection();
1✔
253
        return;
1✔
254
      }
255
    }
256

257
    switch (newState) {
1✔
258
      case IDLE:
259
        // Shutdown when ready: connect from beginning when prompted
260
        addressIndex.reset();
1✔
261
        rawConnectivityState = IDLE;
1✔
262
        updateBalancingState(IDLE, new RequestConnectionPicker(this));
1✔
263
        break;
1✔
264

265
      case CONNECTING:
266
        rawConnectivityState = CONNECTING;
1✔
267
        updateBalancingState(CONNECTING, new Picker(PickResult.withNoResult()));
1✔
268
        break;
1✔
269

270
      case READY:
271
        shutdownRemaining(subchannelData);
1✔
272
        addressIndex.seekTo(getAddress(subchannel));
1✔
273
        rawConnectivityState = READY;
1✔
274
        updateHealthCheckedState(subchannelData);
1✔
275
        break;
1✔
276

277
      case TRANSIENT_FAILURE:
278
        // If we are looking at current channel, request a connection if possible
279
        if (addressIndex.isValid()
1✔
280
            && subchannels.get(addressIndex.getCurrentAddress()).getSubchannel() == subchannel) {
1✔
281
          if (addressIndex.increment()) {
1✔
282
            cancelScheduleTask();
1✔
283
            requestConnection(); // is recursive so might hit the end of the addresses
1✔
284
          }
285
        }
286

287
        if (isPassComplete()) {
1✔
288
          rawConnectivityState = TRANSIENT_FAILURE;
1✔
289
          updateBalancingState(TRANSIENT_FAILURE,
1✔
290
              new Picker(PickResult.withError(stateInfo.getStatus())));
1✔
291

292
          // Refresh Name Resolution, but only when all 3 conditions are met
293
          // * We are at the end of addressIndex
294
          // * have had status reported for all subchannels.
295
          // * And one of the following conditions:
296
          //    * Have had enough TF reported since we completed first pass
297
          //    * Just completed the first pass
298
          if (++numTf >= addressIndex.size() || firstPass) {
1✔
299
            firstPass = false;
1✔
300
            numTf = 0;
1✔
301
            helper.refreshNameResolution();
1✔
302
          }
303
        }
304
        break;
305

306
      default:
307
        throw new IllegalArgumentException("Unsupported state:" + newState);
×
308
    }
309
  }
1✔
310

311
  private void updateHealthCheckedState(SubchannelData subchannelData) {
312
    if (subchannelData.state != READY) {
1✔
313
      return;
1✔
314
    }
315
    if (subchannelData.getHealthState() == READY) {
1✔
316
      updateBalancingState(READY,
1✔
317
          new FixedResultPicker(PickResult.withSubchannel(subchannelData.subchannel)));
1✔
318
    } else if (subchannelData.getHealthState() == TRANSIENT_FAILURE) {
1✔
319
      updateBalancingState(TRANSIENT_FAILURE, new Picker(PickResult.withError(
1✔
320
          subchannelData.healthListener.healthStateInfo.getStatus())));
1✔
321
    } else if (concludedState != TRANSIENT_FAILURE) {
1✔
322
      updateBalancingState(subchannelData.getHealthState(),
1✔
323
          new Picker(PickResult.withNoResult()));
1✔
324
    }
325
  }
1✔
326

327
  private void updateBalancingState(ConnectivityState state, SubchannelPicker picker) {
328
    // an optimization: de-dup IDLE or CONNECTING notification.
329
    if (state == concludedState && (state == IDLE || state == CONNECTING)) {
1✔
330
      return;
1✔
331
    }
332
    concludedState = state;
1✔
333
    helper.updateBalancingState(state, picker);
1✔
334
  }
1✔
335

336
  @Override
337
  public void shutdown() {
338
    log.log(Level.FINE,
1✔
339
        "Shutting down, currently have {} subchannels created", subchannels.size());
1✔
340
    rawConnectivityState = SHUTDOWN;
1✔
341
    concludedState = SHUTDOWN;
1✔
342
    cancelScheduleTask();
1✔
343

344
    for (SubchannelData subchannelData : subchannels.values()) {
1✔
345
      subchannelData.getSubchannel().shutdown();
1✔
346
    }
1✔
347

348
    subchannels.clear();
1✔
349
  }
1✔
350

351
  /**
352
  * Shuts down remaining subchannels. Called when a subchannel becomes ready, which means
353
  * that all other subchannels must be shutdown.
354
  */
355
  private void shutdownRemaining(SubchannelData activeSubchannelData) {
356
    cancelScheduleTask();
1✔
357
    for (SubchannelData subchannelData : subchannels.values()) {
1✔
358
      if (!subchannelData.getSubchannel().equals(activeSubchannelData.subchannel)) {
1✔
359
        subchannelData.getSubchannel().shutdown();
1✔
360
      }
361
    }
1✔
362
    subchannels.clear();
1✔
363
    activeSubchannelData.updateState(READY);
1✔
364
    subchannels.put(getAddress(activeSubchannelData.subchannel), activeSubchannelData);
1✔
365
  }
1✔
366

367
  /**
368
   * Requests a connection to the next applicable address' subchannel, creating one if necessary.
369
   * Schedules a connection to next address in list as well.
370
   * If the current channel has already attempted a connection, we attempt a connection
371
   * to the next address/subchannel in our list.  We assume that createNewSubchannel will never
372
   * return null.
373
   */
374
  @Override
375
  public void requestConnection() {
376
    if (addressIndex == null || !addressIndex.isValid() || rawConnectivityState == SHUTDOWN ) {
1✔
377
      return;
1✔
378
    }
379

380
    Subchannel subchannel;
381
    SocketAddress currentAddress;
382
    currentAddress = addressIndex.getCurrentAddress();
1✔
383
    subchannel = subchannels.containsKey(currentAddress)
1✔
384
        ? subchannels.get(currentAddress).getSubchannel()
1✔
385
        : createNewSubchannel(currentAddress, addressIndex.getCurrentEagAttributes());
1✔
386

387
    ConnectivityState subchannelState = subchannels.get(currentAddress).getState();
1✔
388
    switch (subchannelState) {
1✔
389
      case IDLE:
390
        subchannel.requestConnection();
1✔
391
        subchannels.get(currentAddress).updateState(CONNECTING);
1✔
392
        scheduleNextConnection();
1✔
393
        break;
1✔
394
      case CONNECTING:
395
        if (enableHappyEyeballs) {
1✔
396
          scheduleNextConnection();
1✔
397
        } else {
398
          subchannel.requestConnection();
1✔
399
        }
400
        break;
1✔
401
      case TRANSIENT_FAILURE:
402
        addressIndex.increment();
1✔
403
        requestConnection();
1✔
404
        break;
1✔
405
      case READY: // Shouldn't ever happen
406
        log.warning("Requesting a connection even though we have a READY subchannel");
×
407
        break;
×
408
      case SHUTDOWN:
409
      default:
410
        // Makes checkstyle happy
411
    }
412
  }
1✔
413

414

415
  /**
416
  * Happy Eyeballs
417
  * Schedules connection attempt to happen after a delay to the next available address.
418
  */
419
  private void scheduleNextConnection() {
420
    if (!enableHappyEyeballs
1✔
421
        || (scheduleConnectionTask != null && scheduleConnectionTask.isPending())) {
1✔
422
      return;
1✔
423
    }
424

425
    class StartNextConnection implements Runnable {
1✔
426
      @Override
427
      public void run() {
428
        scheduleConnectionTask = null;
1✔
429
        if (addressIndex.increment()) {
1✔
430
          requestConnection();
1✔
431
        }
432
      }
1✔
433
    }
434

435
    SynchronizationContext synchronizationContext = null;
1✔
436
    try {
437
      synchronizationContext = helper.getSynchronizationContext();
1✔
438
    } catch (NullPointerException e) {
×
439
      // All helpers should have a sync context, but if one doesn't (ex. user had a custom test)
440
      // we don't want to break previously working functionality.
441
      return;
×
442
    }
1✔
443

444
    scheduleConnectionTask = synchronizationContext.schedule(
1✔
445
        new StartNextConnection(),
446
        CONNECTION_DELAY_INTERVAL_MS,
447
        TimeUnit.MILLISECONDS,
448
        helper.getScheduledExecutorService());
1✔
449
  }
1✔
450

451
  private void cancelScheduleTask() {
452
    if (scheduleConnectionTask != null) {
1✔
453
      scheduleConnectionTask.cancel();
1✔
454
      scheduleConnectionTask = null;
1✔
455
    }
456
  }
1✔
457

458
  private Subchannel createNewSubchannel(SocketAddress addr, Attributes attrs) {
459
    HealthListener hcListener = new HealthListener();
1✔
460
    final Subchannel subchannel = helper.createSubchannel(
1✔
461
        CreateSubchannelArgs.newBuilder()
1✔
462
        .setAddresses(Lists.newArrayList(
1✔
463
            new EquivalentAddressGroup(addr, attrs)))
464
        .addOption(HEALTH_CONSUMER_LISTENER_ARG_KEY, hcListener)
1✔
465
            .build());
1✔
466
    if (subchannel == null) {
1✔
467
      log.warning("Was not able to create subchannel for " + addr);
×
468
      throw new IllegalStateException("Can't create subchannel");
×
469
    }
470
    SubchannelData subchannelData = new SubchannelData(subchannel, IDLE, hcListener);
1✔
471
    hcListener.subchannelData = subchannelData;
1✔
472
    subchannels.put(addr, subchannelData);
1✔
473
    Attributes scAttrs = subchannel.getAttributes();
1✔
474
    if (scAttrs.get(LoadBalancer.HAS_HEALTH_PRODUCER_LISTENER_KEY) == null) {
1✔
475
      hcListener.healthStateInfo = ConnectivityStateInfo.forNonError(READY);
1✔
476
    }
477
    subchannel.start(stateInfo -> processSubchannelState(subchannel, stateInfo));
1✔
478
    return subchannel;
1✔
479
  }
480

481
  private boolean isPassComplete() {
482
    if (addressIndex == null || addressIndex.isValid()
1✔
483
        || subchannels.size() < addressIndex.size()) {
1✔
484
      return false;
1✔
485
    }
486
    for (SubchannelData sc : subchannels.values()) {
1✔
487
      if (!sc.isCompletedConnectivityAttempt() ) {
1✔
488
        return false;
1✔
489
      }
490
    }
1✔
491
    return true;
1✔
492
  }
493

494
  private final class HealthListener implements SubchannelStateListener {
1✔
495
    private ConnectivityStateInfo healthStateInfo = ConnectivityStateInfo.forNonError(IDLE);
1✔
496
    private SubchannelData subchannelData;
497

498
    @Override
499
    public void onSubchannelState(ConnectivityStateInfo newState) {
500
      log.log(Level.FINE, "Received health status {0} for subchannel {1}",
1✔
501
          new Object[]{newState, subchannelData.subchannel});
1✔
502
      healthStateInfo = newState;
1✔
503
      try {
504
        SubchannelData curSubChanData = subchannels.get(addressIndex.getCurrentAddress());
1✔
505
        if (curSubChanData != null && curSubChanData.healthListener == this) {
1✔
506
          updateHealthCheckedState(subchannelData);
1✔
507
        }
508
      } catch (IllegalStateException e) {
×
509
        log.fine("Health listener received state change after subchannel was removed");
×
510
      }
1✔
511
    }
1✔
512
  }
513

514
  private SocketAddress getAddress(Subchannel subchannel) {
515
    return subchannel.getAddresses().getAddresses().get(0);
1✔
516
  }
517

518
  @VisibleForTesting
519
  ConnectivityState getConcludedConnectivityState() {
520
    return this.concludedState;
1✔
521
  }
522

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

530
    Picker(PickResult result) {
1✔
531
      this.result = checkNotNull(result, "result");
1✔
532
    }
1✔
533

534
    @Override
535
    public PickResult pickSubchannel(PickSubchannelArgs args) {
536
      return result;
1✔
537
    }
538

539
    @Override
540
    public String toString() {
541
      return MoreObjects.toStringHelper(Picker.class).add("result", result).toString();
×
542
    }
543
  }
544

545
  /**
546
   * Picker that requests connection during the first pick, and returns noResult.
547
   */
548
  private final class RequestConnectionPicker extends SubchannelPicker {
549
    private final PickFirstLeafLoadBalancer pickFirstLeafLoadBalancer;
550
    private final AtomicBoolean connectionRequested = new AtomicBoolean(false);
1✔
551

552
    RequestConnectionPicker(PickFirstLeafLoadBalancer pickFirstLeafLoadBalancer) {
1✔
553
      this.pickFirstLeafLoadBalancer =
1✔
554
          checkNotNull(pickFirstLeafLoadBalancer, "pickFirstLeafLoadBalancer");
1✔
555
    }
1✔
556

557
    @Override
558
    public PickResult pickSubchannel(PickSubchannelArgs args) {
559
      if (connectionRequested.compareAndSet(false, true)) {
1✔
560
        helper.getSynchronizationContext().execute(pickFirstLeafLoadBalancer::requestConnection);
1✔
561
      }
562
      return PickResult.withNoResult();
1✔
563
    }
564
  }
565

566
  /**
567
   * Index as in 'i', the pointer to an entry. Not a "search index."
568
   * All updates should be done in a synchronization context.
569
   */
570
  @VisibleForTesting
571
  static final class Index {
572
    private List<EquivalentAddressGroup> addressGroups;
573
    private int groupIndex;
574
    private int addressIndex;
575

576
    public Index(List<EquivalentAddressGroup> groups) {
1✔
577
      this.addressGroups = groups != null ? groups : Collections.emptyList();
1✔
578
    }
1✔
579

580
    public boolean isValid() {
581
      // Is invalid if empty or has incremented off the end
582
      return groupIndex < addressGroups.size();
1✔
583
    }
584

585
    public boolean isAtBeginning() {
586
      return groupIndex == 0 && addressIndex == 0;
1✔
587
    }
588

589
    /**
590
     * Move to next address in group.  If last address in group move to first address of next group.
591
     * @return false if went off end of the list, otherwise true
592
     */
593
    public boolean increment() {
594
      if (!isValid()) {
1✔
595
        return false;
1✔
596
      }
597

598
      EquivalentAddressGroup group = addressGroups.get(groupIndex);
1✔
599
      addressIndex++;
1✔
600
      if (addressIndex >= group.getAddresses().size()) {
1✔
601
        groupIndex++;
1✔
602
        addressIndex = 0;
1✔
603
        return groupIndex < addressGroups.size();
1✔
604
      }
605

606
      return true;
1✔
607
    }
608

609
    public void reset() {
610
      groupIndex = 0;
1✔
611
      addressIndex = 0;
1✔
612
    }
1✔
613

614
    public SocketAddress getCurrentAddress() {
615
      if (!isValid()) {
1✔
616
        throw new IllegalStateException("Index is past the end of the address group list");
×
617
      }
618
      return addressGroups.get(groupIndex).getAddresses().get(addressIndex);
1✔
619
    }
620

621
    public Attributes getCurrentEagAttributes() {
622
      if (!isValid()) {
1✔
623
        throw new IllegalStateException("Index is off the end of the address group list");
×
624
      }
625
      return addressGroups.get(groupIndex).getAttributes();
1✔
626
    }
627

628
    public List<EquivalentAddressGroup> getCurrentEagAsList() {
629
      return Collections.singletonList(
1✔
630
          new EquivalentAddressGroup(getCurrentAddress(), getCurrentEagAttributes()));
1✔
631
    }
632

633
    /**
634
     * Update to new groups, resetting the current index.
635
     */
636
    public void updateGroups(ImmutableList<EquivalentAddressGroup> newGroups) {
637
      addressGroups = newGroups != null ? newGroups : Collections.emptyList();
1✔
638
      reset();
1✔
639
    }
1✔
640

641
    /**
642
     * Returns false if the needle was not found and the current index was left unchanged.
643
     */
644
    public boolean seekTo(SocketAddress needle) {
645
      for (int i = 0; i < addressGroups.size(); i++) {
1✔
646
        EquivalentAddressGroup group = addressGroups.get(i);
1✔
647
        int j = group.getAddresses().indexOf(needle);
1✔
648
        if (j == -1) {
1✔
649
          continue;
1✔
650
        }
651
        this.groupIndex = i;
1✔
652
        this.addressIndex = j;
1✔
653
        return true;
1✔
654
      }
655
      return false;
1✔
656
    }
657

658
    public int size() {
659
      return (addressGroups != null) ? addressGroups.size() : 0;
1✔
660
    }
661
  }
662

663
  private static final class SubchannelData {
664
    private final Subchannel subchannel;
665
    private ConnectivityState state;
666
    private final HealthListener healthListener;
667
    private boolean completedConnectivityAttempt = false;
1✔
668

669
    public SubchannelData(Subchannel subchannel, ConnectivityState state,
670
                          HealthListener subchannelHealthListener) {
1✔
671
      this.subchannel = subchannel;
1✔
672
      this.state = state;
1✔
673
      this.healthListener = subchannelHealthListener;
1✔
674
    }
1✔
675

676
    public Subchannel getSubchannel() {
677
      return this.subchannel;
1✔
678
    }
679

680
    public ConnectivityState getState() {
681
      return this.state;
1✔
682
    }
683

684
    public boolean isCompletedConnectivityAttempt() {
685
      return completedConnectivityAttempt;
1✔
686
    }
687

688
    private void updateState(ConnectivityState newState) {
689
      this.state = newState;
1✔
690
      if (newState == READY || newState == TRANSIENT_FAILURE) {
1✔
691
        completedConnectivityAttempt = true;
1✔
692
      } else if (newState == IDLE) {
1✔
693
        completedConnectivityAttempt = false;
1✔
694
      }
695
    }
1✔
696

697
    private ConnectivityState getHealthState() {
698
      return healthListener.healthStateInfo.getState();
1✔
699
    }
700
  }
701

702
  public static final class PickFirstLeafLoadBalancerConfig {
703

704
    @Nullable
705
    public final Boolean shuffleAddressList;
706

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

711
    public PickFirstLeafLoadBalancerConfig(@Nullable Boolean shuffleAddressList) {
712
      this(shuffleAddressList, null);
1✔
713
    }
1✔
714

715
    PickFirstLeafLoadBalancerConfig(@Nullable Boolean shuffleAddressList,
716
        @Nullable Long randomSeed) {
1✔
717
      this.shuffleAddressList = shuffleAddressList;
1✔
718
      this.randomSeed = randomSeed;
1✔
719
    }
1✔
720
  }
721
}
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2025 Coveralls, Inc