Only link delayed transport AFTER real transport has called transportReady().

If TransportSet fails to connect a transport (i.e., transportShutdown()
called without transportReady()), TransportSet will automatically
schedule reconnection for the next address, unless it has reached the end
of the address list, in which case it will fail the delayed transport.

This will reduce stream errors caused by bad addresses appearing before
good addresses in the resolved address list.

TODO: add unit test for such logic in TransportSetTest.

Author: zhangkun83

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

@@ -45,6 +45,7 @@
 import java.util.LinkedHashSet;
 import java.util.concurrent.Executor;
 
+import javax.annotation.Nullable;
 import javax.annotation.concurrent.GuardedBy;
 
 /**
@@ -232,6 +233,12 @@ public String getLogId() {
     return GrpcUtil.getLogId(this);
   }
 
+  @VisibleForTesting
+  @Nullable
+  Supplier<ClientTransport> getTransportSupplier() {
+    return transportSupplier;
+  }
+
   private class PendingStream extends DelayedStream {
     private final MethodDescriptor<?, ?> method;
     private final Metadata headers;

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

@@ -34,7 +34,6 @@
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import com.google.common.base.Stopwatch;
-import com.google.common.base.Throwables;
 
 import io.grpc.EquivalentAddressGroup;
 import io.grpc.LoadBalancer;
@@ -44,7 +43,6 @@
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
-import java.util.concurrent.Callable;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
@@ -59,8 +57,9 @@
  * Transports for a single {@link SocketAddress}.
  */
 @ThreadSafe
-final class TransportSet implements WithLogId {
+final class TransportSet {
   private static final Logger log = Logger.getLogger(TransportSet.class.getName());
+
   private static final ClientTransport SHUTDOWN_TRANSPORT =
       new FailingClientTransport(Status.UNAVAILABLE.withDescription("TransportSet is shutdown"));
 
@@ -108,20 +107,16 @@ final class TransportSet implements WithLogId {
   /*
    * The transport for new outgoing requests.
    * - If shutdown == true, activeTransport is null (shutdown)
-   * - Otherwise, if delayedTransport != null,
-   *   activeTransport is delayedTransport (waiting to connect)
+   * - Otherwise, if a connection is pending or connecting,
+   *   activeTransport is a DelayedClientTransport
    * - Otherwise, activeTransport is either null (initially or when idle)
-   *   or points to a real transport (when connecting or connected).
+   *   or points to a real transport (when ready).
    *
    * 'lock' must be held when assigning to it.
    */
   @Nullable
   private volatile ManagedClientTransport activeTransport;
 
-  @GuardedBy("lock")
-  @Nullable
-  private DelayedClientTransport delayedTransport;
-
   TransportSet(EquivalentAddressGroup addressGroup, String authority,
       LoadBalancer<ClientTransport> loadBalancer, BackoffPolicy.Provider backoffPolicyProvider,
       ClientTransportFactory transportFactory, ScheduledExecutorService scheduledExecutor,
@@ -155,40 +150,24 @@ final ClientTransport obtainActiveTransport() {
     if (savedTransport != null) {
       return savedTransport;
     }
-    Callable<ClientTransport> immediateConnectionTask = null;
     synchronized (lock) {
       // Check again, since it could have changed before acquiring the lock
       if (activeTransport == null) {
         if (shutdown) {
           return SHUTDOWN_TRANSPORT;
         }
-        delayedTransport = new DelayedClientTransport();
+        DelayedClientTransport delayedTransport = new DelayedClientTransport();
         transports.add(delayedTransport);
         delayedTransport.start(new BaseTransportListener(delayedTransport));
         activeTransport = delayedTransport;
-        immediateConnectionTask = scheduleConnection();
+        scheduleConnection(delayedTransport);
       }
-      savedTransport = activeTransport;
+      return activeTransport;
     }
-    if (immediateConnectionTask != null) {
-      try {
-        return immediateConnectionTask.call();
-      } catch (Exception e) {
-        throw Throwables.propagate(e);
-      }
-    }
-    return savedTransport;
   }
 
-  /**
-   * Schedule a task that creates a new transport.
-   *
-   * @return if not {@code null}, caller should run the returned callable outside of lock. The
-   *         callable returns the real transport that has been created.
-   */
-  @Nullable
   @GuardedBy("lock")
-  private Callable<ClientTransport> scheduleConnection() {
+  private void scheduleConnection(final DelayedClientTransport delayedTransport) {
     Preconditions.checkState(reconnectTask == null || reconnectTask.isDone(),
         "previous reconnectTask is not done");
 
@@ -203,47 +182,20 @@ private Callable<ClientTransport> scheduleConnection() {
       nextAddressIndex = 0;
     }
 
-    final Callable<ClientTransport> createTransportCallable = new Callable<ClientTransport>() {
+    Runnable createTransportRunnable = new Runnable() {
       @Override
-      public ClientTransport call() {
-        DelayedClientTransport savedDelayedTransport;
-        ManagedClientTransport newActiveTransport;
-        boolean savedShutdown;
+      public void run() {
         synchronized (lock) {
-          savedShutdown = shutdown;
           reconnectTask = null;
           if (currentAddressIndex == 0) {
             backoffWatch.reset().start();
           }
-          newActiveTransport = transportFactory.newClientTransport(address, authority);
-          if (log.isLoggable(Level.FINE)) {
-            log.log(Level.FINE, "[{0}] Created {1} for {2}",
-                new Object[] {getLogId(), newActiveTransport.getLogId(), address});
-          }
-          transports.add(newActiveTransport);
-          newActiveTransport.start(
-              new TransportListener(newActiveTransport, address));
-          if (shutdown) {
-            // If TransportSet already shutdown, newActiveTransport is only to take care of pending
-            // streams in delayedTransport, but will not serve new streams, and it will be shutdown
-            // as soon as it's set to the delayedTransport.
-            // activeTransport should have already been set to null by shutdown(). We keep it null.
-            Preconditions.checkState(activeTransport == null,
-                "Unexpected non-null activeTransport");
-          } else {
-            activeTransport = newActiveTransport;
-          }
-          savedDelayedTransport = delayedTransport;
-          delayedTransport = null;
-        }
-        savedDelayedTransport.setTransport(newActiveTransport);
-        // This delayed transport will terminate and be removed from transports.
-        savedDelayedTransport.shutdown();
-        if (savedShutdown) {
-          // See comments in the synchronized block above on why we shutdown here.
-          newActiveTransport.shutdown();
+          ManagedClientTransport transport =
+              transportFactory.newClientTransport(address, authority);
+          log.log(Level.FINE, "Created transport {0} for {1}", new Object[] {transport, address});
+          transports.add(transport);
+          transport.start(new TransportListener(transport, delayedTransport, address));
         }
-        return newActiveTransport;
       }
     };
 
@@ -259,27 +211,13 @@ public ClientTransport call() {
       }
     }
     firstAttempt = false;
-    if (log.isLoggable(Level.FINE)) {
-      log.log(Level.FINE, "[{0}] Scheduling connection after {1} ms for {2}",
-          new Object[]{getLogId(), delayMillis, address});
-    }
     if (delayMillis <= 0) {
       reconnectTask = null;
       // No back-off this time.
-      // Note createTransportRunnable is not supposed to run under the lock.
-      return createTransportCallable;
+      createTransportRunnable.run();
     } else {
       reconnectTask = scheduledExecutor.schedule(
-          new Runnable() {
-            @Override public void run() {
-              try {
-                createTransportCallable.call();
-              } catch (Exception e) {
-                throw Throwables.propagate(e);
-              }
-            }
-          }, delayMillis, TimeUnit.MILLISECONDS);
-      return null;
+          createTransportRunnable, delayMillis, TimeUnit.MILLISECONDS);
     }
   }
 
@@ -301,7 +239,6 @@ final void shutdown() {
       if (transports.isEmpty()) {
         runCallback = true;
         Preconditions.checkState(reconnectTask == null, "Should have no reconnectTask scheduled");
-        Preconditions.checkState(delayedTransport == null, "Should have no delayedTransport");
       }  // else: the callback will be run once all transports have been terminated
     }
     if (savedActiveTransport != null) {
@@ -320,11 +257,6 @@ private void cancelReconnectTask() {
     }
   }
 
-  @Override
-  public String getLogId() {
-    return GrpcUtil.getLogId(this);
-  }
-
   /** Shared base for both delayed and real transports. */
   private class BaseTransportListener implements ManagedClientTransport.Listener {
     protected final ManagedClientTransport transport;
@@ -345,9 +277,6 @@ public void transportTerminated() {
       synchronized (lock) {
         transports.remove(transport);
         if (shutdown && transports.isEmpty()) {
-          if (log.isLoggable(Level.FINE)) {
-            log.log(Level.FINE, "[{0}] Terminated", getLogId());
-          }
           runCallback = true;
           cancelReconnectTask();
         }
@@ -361,56 +290,75 @@ public void transportTerminated() {
   /** Listener for real transports. */
   private class TransportListener extends BaseTransportListener {
     private final SocketAddress address;
+    private final DelayedClientTransport delayedTransport;
 
-    public TransportListener(ManagedClientTransport transport, SocketAddress address) {
+    public TransportListener(ManagedClientTransport transport,
+        DelayedClientTransport delayedTransport, SocketAddress address) {
       super(transport);
       this.address = address;
-    }
-
-    private boolean isAttachedToActiveTransport() {
-      return activeTransport == transport;
+      this.delayedTransport = delayedTransport;
     }
 
     @Override
     public void transportReady() {
-      if (log.isLoggable(Level.FINE)) {
-        log.log(Level.FINE, "[{0}] {1} for {2} is ready",
-            new Object[] {getLogId(), transport.getLogId(), address});
-      }
+      log.log(Level.FINE, "Transport {0} for {1} is ready", new Object[] {transport, address});
       super.transportReady();
+      boolean savedShutdown;
       synchronized (lock) {
-        if (isAttachedToActiveTransport()) {
-          firstAttempt = true;
+        savedShutdown = shutdown;
+        firstAttempt = true;
+        if (shutdown) {
+          // If TransportSet already shutdown, transport is only to take care of pending
+          // streams in delayedTransport, but will not serve new streams, and it will be shutdown
+          // as soon as it's set to the delayedTransport.
+          // activeTransport should have already been set to null by shutdown(). We keep it null.
+          Preconditions.checkState(activeTransport == null,
+              "Unexpected non-null activeTransport");
+        } else if (activeTransport == delayedTransport) {
+          activeTransport = transport;
         }
       }
+      delayedTransport.setTransport(transport);
+      // This delayed transport will terminate and be removed from transports.
+      delayedTransport.shutdown();
+      if (savedShutdown) {
+        // See comments in the synchronized block above on why we shutdown here.
+        transport.shutdown();
+      }
       loadBalancer.handleTransportReady(addressGroup);
     }
 
     @Override
     public void transportShutdown(Status s) {
-      if (log.isLoggable(Level.FINE)) {
-        log.log(Level.FINE, "[{0}] {1} for {2} is being shutdown with status {3}",
-            new Object[] {getLogId(), transport.getLogId(), address, s});
-      }
+      log.log(Level.FINE, "Transport {0} for {1} is being shutdown with {2}",
+          new Object[] {transport, address, s});
       super.transportShutdown(s);
       synchronized (lock) {
-        if (isAttachedToActiveTransport()) {
+        if (activeTransport == transport) {
           activeTransport = null;
+        } else if (activeTransport == delayedTransport) {
+          // Continue reconnect if there are still addresses to try.
+          // Fail if all addresses have been tried and failed in a row.
+          if (nextAddressIndex == 0) {
+            delayedTransport.setTransport(new FailingClientTransport(s));
+            delayedTransport.shutdown();
+            activeTransport = null;
+          } else {
+            scheduleConnection(delayedTransport);
+          }
         }
       }
       loadBalancer.handleTransportShutdown(addressGroup, s);
     }
 
     @Override
     public void transportTerminated() {
-      if (log.isLoggable(Level.FINE)) {
-        log.log(Level.FINE, "[{0}] {1} for {2} is terminated",
-            new Object[] {getLogId(), transport.getLogId(), address});
-      }
+      log.log(Level.FINE, "Transport {0} for {1} is terminated",
+          new Object[] {transport, address});
       super.transportTerminated();
-      Preconditions.checkState(!isAttachedToActiveTransport(),
-          "Listener is still attached to activeTransport. "
-          + "Seems transportTerminated was not called.");
+      Preconditions.checkState(activeTransport != transport,
+          "activeTransport still points to the delayedTransport. "
+          + "Seems transportShutdown() was not called.");
     }
   }