Change PickFirstLeafLoadBalancer to only have 1 subchannel at a time …

…if environment variable GRPC_SERIALIZE_RETRIES == true.

Cache serializingRetries value so that it doesn't have to look up the flag every time.

Clear the correct task when READY in processSubchannelState and move the logic to cancelScheduledTasks

Cleanup based on PR review

remove unneeded checks for shutdown.

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

@@ -64,17 +64,26 @@ final class PickFirstLeafLoadBalancer extends LoadBalancer {
   private int numTf = 0;
   private boolean firstPass = true;
   @Nullable
-  private ScheduledHandle scheduleConnectionTask;
+  private ScheduledHandle scheduleConnectionTask = null;
   private ConnectivityState rawConnectivityState = IDLE;
   private ConnectivityState concludedState = IDLE;
-  private final boolean enableHappyEyeballs =
-      PickFirstLoadBalancerProvider.isEnabledHappyEyeballs();
+  private final boolean enableHappyEyeballs = !isSerializingRetries()
+      && PickFirstLoadBalancerProvider.isEnabledHappyEyeballs();
   private boolean notAPetiolePolicy = true; // means not under a petiole policy
+  private final BackoffPolicy.Provider bkoffPolProvider = new ExponentialBackoffPolicy.Provider();
+  private BackoffPolicy reconnectPolicy;
+  @Nullable
+  private ScheduledHandle reconnectTask = null;
+  private boolean serializingRetries = isSerializingRetries();
 
   PickFirstLeafLoadBalancer(Helper helper) {
     this.helper = checkNotNull(helper, "helper");
   }
 
+  static boolean isSerializingRetries() {
+    return GrpcUtil.getFlag("GRPC_SERIALIZE_RETRIES", false);
+  }
+
   @Override
   public Status acceptResolvedAddresses(ResolvedAddresses resolvedAddresses) {
     if (rawConnectivityState == SHUTDOWN) {
@@ -277,10 +286,12 @@ void processSubchannelState(SubchannelData subchannelData, ConnectivityStateInfo
           if (addressIndex.increment()) {
             cancelScheduleTask();
             requestConnection(); // is recursive so might hit the end of the addresses
+          } else {
+            scheduleBackoff();
           }
         }
 
-        if (isPassComplete()) {
+        if ( isPassComplete()) {
           rawConnectivityState = TRANSIENT_FAILURE;
           updateBalancingState(TRANSIENT_FAILURE,
               new Picker(PickResult.withError(stateInfo.getStatus())));
@@ -304,6 +315,42 @@ void processSubchannelState(SubchannelData subchannelData, ConnectivityStateInfo
     }
   }
 
+  /**
+   * Only called after all addresses attempted and failed (TRANSIENT_FAILURE).
+   */
+  private void scheduleBackoff() {
+    if (!serializingRetries) {
+      return;
+    }
+
+    class EndOfCurrentBackoff implements Runnable {
+      @Override
+      public void run() {
+        reconnectTask = null;
+        if (rawConnectivityState == SHUTDOWN) {
+          return;
+        }
+        addressIndex.reset();
+        requestConnection();
+      }
+    }
+
+    // Just allow the previous one to trigger when ready if we're already in backoff
+    if (reconnectTask != null) {
+      return;
+    }
+
+    if (reconnectPolicy == null) {
+      reconnectPolicy = bkoffPolProvider.get();
+    }
+    long delayNanos = reconnectPolicy.nextBackoffNanos();
+    reconnectTask = helper.getSynchronizationContext().schedule(
+        new EndOfCurrentBackoff(),
+        delayNanos,
+        TimeUnit.NANOSECONDS,
+        helper.getScheduledExecutorService());
+  }
+
   private void updateHealthCheckedState(SubchannelData subchannelData) {
     if (subchannelData.state != READY) {
       return;
@@ -337,6 +384,11 @@ public void shutdown() {
     rawConnectivityState = SHUTDOWN;
     concludedState = SHUTDOWN;
     cancelScheduleTask();
+    if (reconnectTask != null) {
+      reconnectTask.cancel();
+      reconnectTask = null;
+    }
+    reconnectPolicy = null;
 
     for (SubchannelData subchannelData : subchannels.values()) {
       subchannelData.getSubchannel().shutdown();
@@ -350,6 +402,12 @@ public void shutdown() {
   * that all other subchannels must be shutdown.
   */
   private void shutdownRemaining(SubchannelData activeSubchannelData) {
+    if (reconnectTask != null) {
+      reconnectTask.cancel();
+      reconnectTask = null;
+    }
+    reconnectPolicy = null;
+
     cancelScheduleTask();
     for (SubchannelData subchannelData : subchannels.values()) {
       if (!subchannelData.getSubchannel().equals(activeSubchannelData.subchannel)) {
@@ -370,7 +428,12 @@ private void shutdownRemaining(SubchannelData activeSubchannelData) {
    */
   @Override
   public void requestConnection() {
-    if (!addressIndex.isValid() || rawConnectivityState == SHUTDOWN) {
+    if (rawConnectivityState == SHUTDOWN) {
+      return;
+    }
+
+    if (!addressIndex.isValid()) {
+      scheduleBackoff();
       return;
     }
 
@@ -391,8 +454,21 @@ public void requestConnection() {
         scheduleNextConnection();
         break;
       case TRANSIENT_FAILURE:
+        if (scheduleConnectionTask != null) {
+          break; // let the already scheduled task do its job
+        }
         addressIndex.increment();
-        requestConnection();
+        if (!serializingRetries) {
+          requestConnection();
+        } else {
+          if (!addressIndex.isValid()) {
+            scheduleBackoff();
+          } else {
+            subchannelData.subchannel.shutdown(); // shutdown the previous subchannel
+            subchannels.remove(currentAddress);
+            requestConnection();
+          }
+        }
         break;
       default:
         // Wait for current subchannel to change state
@@ -458,7 +534,8 @@ private SubchannelData createNewSubchannel(SocketAddress addr, Attributes attrs)
   }
 
   private boolean isPassComplete() {
-    if (addressIndex.isValid() || subchannels.size() < addressIndex.size()) {
+    if ((!serializingRetries && addressIndex.isValid())
+        || subchannels.size() < addressIndex.size()) {
       return false;
     }
     for (SubchannelData sc : subchannels.values()) {

core/src/test/java/io/grpc/internal/PickFirstLeafLoadBalancerTest.java

@@ -27,6 +27,7 @@
 import static io.grpc.LoadBalancer.HEALTH_CONSUMER_LISTENER_ARG_KEY;
 import static io.grpc.LoadBalancer.IS_PETIOLE_POLICY;
 import static io.grpc.internal.PickFirstLeafLoadBalancer.CONNECTION_DELAY_INTERVAL_MS;
+import static io.grpc.internal.PickFirstLeafLoadBalancer.isSerializingRetries;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
@@ -95,7 +96,11 @@ public class PickFirstLeafLoadBalancerTest {
 
   @Parameterized.Parameters(name = "{0}")
   public static List<Boolean> enableHappyEyeballs() {
-    return Arrays.asList(true, false);
+    if (PickFirstLeafLoadBalancer.isSerializingRetries()) {
+      return Arrays.asList(false);
+    } else {
+      return Arrays.asList(false, true);
+    }
   }
 
   @Parameterized.Parameter
@@ -143,7 +148,8 @@ public void setUp() {
     originalHappyEyeballsEnabledValue =
         System.getProperty(PickFirstLoadBalancerProvider.GRPC_PF_USE_HAPPY_EYEBALLS);
     System.setProperty(PickFirstLoadBalancerProvider.GRPC_PF_USE_HAPPY_EYEBALLS,
-        enableHappyEyeballs ? "true" : "false");
+         !PickFirstLeafLoadBalancer.isSerializingRetries() && enableHappyEyeballs
+         ? "true" : "false");
 
     for (int i = 1; i <= 5; i++) {
       SocketAddress addr = new FakeSocketAddress("server" + i);
@@ -498,6 +504,9 @@ public void healthCheckFlow() {
     inOrder.verify(mockHelper).updateBalancingState(eq(READY), pickerCaptor.capture());
     assertThat(pickerCaptor.getValue().pickSubchannel(mockArgs)
         .getSubchannel()).isSameInstanceAs(mockSubchannel1);
+    verify(mockHelper, atLeast(0)).getSynchronizationContext();
+    verify(mockHelper, atLeast(0)).getScheduledExecutorService();
+    verifyNoMoreInteractions(mockHelper);
 
     healthListener2.onSubchannelState(ConnectivityStateInfo.forNonError(READY));
     verifyNoMoreInteractions(mockHelper);
@@ -589,6 +598,8 @@ public void pickAfterResolutionAfterTransientValue() {
 
     // Transition from TRANSIENT_ERROR to CONNECTING should also be ignored.
     stateListener.onSubchannelState(ConnectivityStateInfo.forNonError(CONNECTING));
+    verify(mockHelper, atLeast(0)).getSynchronizationContext();
+    verify(mockHelper, atLeast(0)).getScheduledExecutorService();
     verifyNoMoreInteractions(mockHelper);
     assertEquals(error, pickerCaptor.getValue().pickSubchannel(mockArgs).getStatus());
   }
@@ -619,6 +630,8 @@ public void pickWithDupAddressesUpDownUp() {
 
     // Transition from TRANSIENT_ERROR to CONNECTING should also be ignored.
     stateListener.onSubchannelState(ConnectivityStateInfo.forNonError(CONNECTING));
+    verify(mockHelper, atLeast(0)).getSynchronizationContext();
+    verify(mockHelper, atLeast(0)).getScheduledExecutorService();
     verifyNoMoreInteractions(mockHelper);
     assertEquals(error, pickerCaptor.getValue().pickSubchannel(mockArgs).getStatus());
 
@@ -651,6 +664,8 @@ public void pickWithDupEagsUpDownUp() {
 
     // Transition from TRANSIENT_ERROR to CONNECTING should also be ignored.
     stateListener.onSubchannelState(ConnectivityStateInfo.forNonError(CONNECTING));
+    verify(mockHelper, atLeast(0)).getSynchronizationContext();
+    verify(mockHelper, atLeast(0)).getScheduledExecutorService();
     verifyNoMoreInteractions(mockHelper);
     assertEquals(error, pickerCaptor.getValue().pickSubchannel(mockArgs).getStatus());
 
@@ -1518,6 +1533,8 @@ public void updateAddresses_intersecting_ready() {
 
   @Test
   public void updateAddresses_intersecting_transient_failure() {
+    Assume.assumeTrue(!isSerializingRetries());
+
     // Starting first connection attempt
     InOrder inOrder = inOrder(mockHelper, mockSubchannel1, mockSubchannel2,
         mockSubchannel3, mockSubchannel4); // captor: captures
@@ -1782,6 +1799,8 @@ public void updateAddresses_identical_ready() {
 
   @Test
   public void updateAddresses_identical_transient_failure() {
+    Assume.assumeTrue(!isSerializingRetries());
+
     InOrder inOrder = inOrder(mockHelper, mockSubchannel1, mockSubchannel2,
         mockSubchannel3, mockSubchannel4);
     // Creating first set of endpoints/addresses