core/src/main/java/io/grpc/internal/TransportSet.java - platform/external/grpc-grpc-java - Gitiles

 /*
  * Copyright 2015, Google Inc. All rights reserved.
  *
  * Redistribution and use in source and binary forms, with or without
  * modification, are permitted provided that the following conditions are
  * met:
  *
  *    * Redistributions of source code must retain the above copyright
  * notice, this list of conditions and the following disclaimer.
  *    * Redistributions in binary form must reproduce the above
  * copyright notice, this list of conditions and the following disclaimer
  * in the documentation and/or other materials provided with the
  * distribution.
  *
  *    * Neither the name of Google Inc. nor the names of its
  * contributors may be used to endorse or promote products derived from
  * this software without specific prior written permission.
  *
  * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
  * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
  * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
  * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
  * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
  * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
  * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
  * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  */

 package io.grpc.internal;

 import com.google.common.base.Preconditions;
 import com.google.common.base.Stopwatch;
 import com.google.common.base.Supplier;

 import io.grpc.CallOptions;
 import io.grpc.ClientCall;
 import io.grpc.ConnectivityState;
 import io.grpc.EquivalentAddressGroup;
 import io.grpc.LoadBalancer;
 import io.grpc.ManagedChannel;
 import io.grpc.Metadata;
 import io.grpc.MethodDescriptor;
 import io.grpc.Status;
 import io.grpc.internal.ClientCallImpl.ClientTransportProvider;

 import java.net.SocketAddress;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Executor;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.logging.Level;
 import java.util.logging.Logger;

 import javax.annotation.CheckReturnValue;
 import javax.annotation.Nullable;
 import javax.annotation.concurrent.GuardedBy;
 import javax.annotation.concurrent.ThreadSafe;

 /**
  * Transports for a single {@link SocketAddress}.
  */
 @ThreadSafe
 final class TransportSet extends ManagedChannel implements WithLogId {
   private static final Logger log = Logger.getLogger(TransportSet.class.getName());
   private static final ClientTransport SHUTDOWN_TRANSPORT =
       new FailingClientTransport(Status.UNAVAILABLE.withDescription("TransportSet is shutdown"));

   private final CountDownLatch terminatedLatch = new CountDownLatch(1);
   private final Object lock = new Object();
   private final LogId logId = LogId.allocate(getClass().getName());
   private final EquivalentAddressGroup addressGroup;
   private final String authority;
   private final String userAgent;
   private final BackoffPolicy.Provider backoffPolicyProvider;
   private final Callback callback;
   private final ClientTransportFactory transportFactory;
   private final ScheduledExecutorService scheduledExecutor;
   private final Executor appExecutor;

   @GuardedBy("lock")
   private int nextAddressIndex;

   /**
    * The policy to control back off between reconnects. Non-{@code null} when last connect failed.
    */
   @GuardedBy("lock")
   private BackoffPolicy reconnectPolicy;

   /**
    * Timer monitoring duration since entering CONNECTING state.
    */
   @GuardedBy("lock")
   private final Stopwatch connectingTimer;

   @GuardedBy("lock")
   @Nullable
   private ScheduledFuture<?> reconnectTask;

   /**
    * All transports that are not terminated. At the very least the value of {@link #activeTransport}
    * will be present, but previously used transports that still have streams or are stopping may
    * also be present.
    */
   @GuardedBy("lock")
   private final Collection<ManagedClientTransport> transports =
       new ArrayList<ManagedClientTransport>();

   private final InUseStateAggregator<ManagedClientTransport> inUseStateAggregator =
       new InUseStateAggregator<ManagedClientTransport>() {
         @Override
         Object getLock() {
           return lock;
         }

         @Override
         Runnable handleInUse() {
           return callback.onInUse(TransportSet.this);
         }

         @Override
         void handleNotInUse() {
           callback.onNotInUse(TransportSet.this);
         }
       };

   /**
    * The to-be active transport, which is not ready yet.
    */
   @GuardedBy("lock")
   @Nullable
   private ConnectionClientTransport pendingTransport;

   private final LoadBalancer<ClientTransport> loadBalancer;

   @GuardedBy("lock")
   private boolean shutdown;

   /**
    * The transport for new outgoing requests. 'lock' must be held when assigning to it.
    *
    * <pre><code>
    * State             Value
    * -----             ------
    * IDLE              null, shutdown == false
    * CONNECTING        instanceof DelayedTransport, reconnectTask == null
    * READY             connected transport
    * TRANSIENT_FAILURE instanceof DelayedTransport, reconnectTask != null
    * SHUTDOWN          null, shutdown == true
    * </code></pre>
    */
   @Nullable
   private volatile ManagedClientTransport activeTransport;

   @GuardedBy("lock")
   private final ConnectivityStateManager stateManager =
       new ConnectivityStateManager(ConnectivityState.IDLE);

   TransportSet(EquivalentAddressGroup addressGroup, String authority, String userAgent,
       LoadBalancer<ClientTransport> loadBalancer, BackoffPolicy.Provider backoffPolicyProvider,
       ClientTransportFactory transportFactory, ScheduledExecutorService scheduledExecutor,
       Supplier<Stopwatch> stopwatchSupplier, Executor appExecutor, Callback callback) {
     this.addressGroup = Preconditions.checkNotNull(addressGroup, "addressGroup");
     this.authority = authority;
     this.userAgent = userAgent;
     this.loadBalancer = loadBalancer;
     this.backoffPolicyProvider = backoffPolicyProvider;
     this.transportFactory = transportFactory;
     this.scheduledExecutor = scheduledExecutor;
     this.connectingTimer = stopwatchSupplier.get();
     this.appExecutor = appExecutor;
     this.callback = callback;
   }

   /**
    * Returns the active transport that will be used to create new streams.
    *
    * <p>Never returns {@code null}.
    */
   final ClientTransport obtainActiveTransport() {
     ClientTransport savedTransport = activeTransport;
     if (savedTransport != null) {
       return savedTransport;
     }
     Runnable runnable;
     synchronized (lock) {
       // Check again, since it could have changed before acquiring the lock
       savedTransport = activeTransport;
       if (savedTransport != null) {
         return savedTransport;
       }
       if (shutdown) {
         return SHUTDOWN_TRANSPORT;
       }
       stateManager.gotoState(ConnectivityState.CONNECTING);
       DelayedClientTransport delayedTransport = new DelayedClientTransport(appExecutor);
       transports.add(delayedTransport);
       delayedTransport.start(new BaseTransportListener(delayedTransport));
       savedTransport = activeTransport = delayedTransport;
       runnable = startNewTransport(delayedTransport);
     }
     if (runnable != null) {
       runnable.run();
     }
     return savedTransport;
   }

   @CheckReturnValue
   @GuardedBy("lock")
   private Runnable startNewTransport(DelayedClientTransport delayedTransport) {
     Preconditions.checkState(reconnectTask == null, "Should have no reconnectTask scheduled");

     if (nextAddressIndex == 0) {
       connectingTimer.reset().start();
     }
     List<SocketAddress> addrs = addressGroup.getAddresses();
     final SocketAddress address = addrs.get(nextAddressIndex++);
     if (nextAddressIndex >= addrs.size()) {
       nextAddressIndex = 0;
     }

     ConnectionClientTransport transport =
         transportFactory.newClientTransport(address, authority, userAgent);
     if (log.isLoggable(Level.FINE)) {
       log.log(Level.FINE, "[{0}] Created {1} for {2}",
           new Object[] {getLogId(), transport.getLogId(), address});
     }
     pendingTransport = transport;
     transports.add(transport);
     return transport.start(new TransportListener(transport, delayedTransport, address));
   }

   /**
    * Only called after all addresses attempted and failed (TRANSIENT_FAILURE).
    * @param status the causal status when the channel begins transition to
    *     TRANSIENT_FAILURE.
    */
   private void scheduleBackoff(
       final DelayedClientTransport delayedTransport, final Status status) {
     // This must be run outside of lock. The TransportSet lock is a channel level lock.
     // startBackoff() will acquire the delayed transport lock, which is a transport level
     // lock. Our lock ordering mandates transport lock > channel lock.  Otherwise a deadlock
     // could happen (https://github.com/grpc/grpc-java/issues/2152).
     delayedTransport.startBackoff(status);

     class EndOfCurrentBackoff implements Runnable {
       @Override
       public void run() {
         try {
           delayedTransport.endBackoff();
           Runnable runnable = null;
           synchronized (lock) {
             reconnectTask = null;
             if (!shutdown) {
               stateManager.gotoState(ConnectivityState.CONNECTING);
             }
             runnable = startNewTransport(delayedTransport);
           }
           if (runnable != null) {
             runnable.run();
           }
         } catch (Throwable t) {
           log.log(Level.WARNING, "Exception handling end of backoff", t);
         }
       }
     }

     synchronized (lock) {
       if (shutdown) {
         return;
       }
       stateManager.gotoState(ConnectivityState.TRANSIENT_FAILURE);
       if (reconnectPolicy == null) {
         reconnectPolicy = backoffPolicyProvider.get();
       }
       long delayMillis =
           reconnectPolicy.nextBackoffMillis() - connectingTimer.elapsed(TimeUnit.MILLISECONDS);
       if (log.isLoggable(Level.FINE)) {
         log.log(Level.FINE, "[{0}] Scheduling backoff for {1} ms",
             new Object[]{getLogId(), delayMillis});
       }
       Preconditions.checkState(reconnectTask == null, "previous reconnectTask is not done");
       reconnectTask = scheduledExecutor.schedule(
           new LogExceptionRunnable(new EndOfCurrentBackoff()),
           delayMillis,
           TimeUnit.MILLISECONDS);
     }
   }

   @Override
   public ManagedChannel shutdown() {
     ManagedClientTransport savedActiveTransport;
     ConnectionClientTransport savedPendingTransport;
     boolean runCallback = false;
     synchronized (lock) {
       if (shutdown) {
         return this;
       }
       stateManager.gotoState(ConnectivityState.SHUTDOWN);
       shutdown = true;
       savedActiveTransport = activeTransport;
       savedPendingTransport = pendingTransport;
       activeTransport = null;
       if (transports.isEmpty()) {
         runCallback = true;
         terminatedLatch.countDown();
         if (log.isLoggable(Level.FINE)) {
           log.log(Level.FINE, "[{0}] Terminated in shutdown()", getLogId());
         }
         Preconditions.checkState(reconnectTask == null, "Should have no reconnectTask scheduled");
       }  // else: the callback will be run once all transports have been terminated
     }
     if (savedActiveTransport != null) {
       savedActiveTransport.shutdown();
     }
     if (savedPendingTransport != null) {
       savedPendingTransport.shutdown();
     }
     if (runCallback) {
       callback.onTerminated(this);
     }
     return this;
   }

   void shutdownNow(Status reason) {
     shutdown();
     Collection<ManagedClientTransport> transportsCopy;
     synchronized (lock) {
       transportsCopy = new ArrayList<ManagedClientTransport>(transports);
     }
     for (ManagedClientTransport transport : transportsCopy) {
       transport.shutdownNow(reason);
     }
   }

   @Override
   public ManagedChannel shutdownNow() {
     shutdownNow(Status.UNAVAILABLE.withDescription("TransportSet shutdown as ManagedChannel"));
     return this;
   }

   @GuardedBy("lock")
   private void cancelReconnectTask() {
     if (reconnectTask != null) {
       reconnectTask.cancel(false);
       reconnectTask = null;
     }
   }

   @Override
   public LogId getLogId() {
     return logId;
   }

   @Override
   public final <RequestT, ResponseT> ClientCall<RequestT, ResponseT> newCall(
       MethodDescriptor<RequestT, ResponseT> methodDescriptor, CallOptions callOptions) {
     return new ClientCallImpl<RequestT, ResponseT>(methodDescriptor,
         new SerializingExecutor(appExecutor), callOptions, StatsTraceContext.NOOP,
         new ClientTransportProvider() {
           @Override
           public ClientTransport get(CallOptions callOptions, Metadata headers) {
             return obtainActiveTransport();
           }
         },
         scheduledExecutor);
   }

   @Override
   public boolean isShutdown() {
     synchronized (lock) {
       return shutdown;
     }
   }

   @Override
   public boolean isTerminated() {
     return terminatedLatch.getCount() == 0;
   }

   @Override
   public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
     return terminatedLatch.await(timeout, unit);
   }

   @Override
   public String authority() {
     return authority;
   }

   @Override
   public ConnectivityState getState(boolean requestConnection) {
     if (requestConnection) {
       boolean connect = false;
       synchronized (lock) {
         connect = stateManager.getState() == ConnectivityState.IDLE;
       }
       if (connect) {
         obtainActiveTransport();
       }
     }
     synchronized (lock) {
       return stateManager.getState();
     }
   }

   @Override
   public void notifyWhenStateChanged(ConnectivityState source, Runnable callback) {
     synchronized (lock) {
       stateManager.notifyWhenStateChanged(callback, appExecutor, source);
     }
   }

   /** Shared base for both delayed and real transports. */
   private class BaseTransportListener implements ManagedClientTransport.Listener {
     protected final ManagedClientTransport transport;

     public BaseTransportListener(ManagedClientTransport transport) {
       this.transport = transport;
     }

     @Override
     public void transportReady() {}

     @Override
     public void transportInUse(boolean inUse) {
       Runnable r = inUseStateAggregator.updateObjectInUse(transport, inUse);
       if (r != null) {
         r.run();
       }
     }

     @Override
     public void transportShutdown(Status status) {}

     @Override
     public void transportTerminated() {
       boolean runCallback = false;
       Runnable r = inUseStateAggregator.updateObjectInUse(transport, false);
       assert r == null;
       synchronized (lock) {
         transports.remove(transport);
         if (shutdown && transports.isEmpty()) {
           if (log.isLoggable(Level.FINE)) {
             log.log(Level.FINE, "[{0}] Terminated in transportTerminated()", getLogId());
           }
           terminatedLatch.countDown();
           runCallback = true;
           cancelReconnectTask();
         }
       }
       if (runCallback) {
         callback.onTerminated(TransportSet.this);
       }
     }
   }

   /** Listener for real transports. */
   private class TransportListener extends BaseTransportListener {
     private final SocketAddress address;
     private final DelayedClientTransport delayedTransport;

     public TransportListener(ManagedClientTransport transport,
         DelayedClientTransport delayedTransport, SocketAddress address) {
       super(transport);
       this.address = address;
       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});
       }
       super.transportReady();
       boolean savedShutdown;
       synchronized (lock) {
         savedShutdown = shutdown;
         reconnectPolicy = null;
         nextAddressIndex = 0;
         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) {
           stateManager.gotoState(ConnectivityState.READY);
           Preconditions.checkState(pendingTransport == transport, "transport mismatch");
           activeTransport = transport;
           pendingTransport = null;
         }
       }
       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) {
       boolean allAddressesFailed = false;
       boolean closedByServer = false;
       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});
       }
       super.transportShutdown(s);
       Runnable runnable = null;
       synchronized (lock) {
         if (activeTransport == transport) {
           // This is true only if the transport was ready.
           // shutdown() should have set activeTransport to null
           Preconditions.checkState(!shutdown, "unexpected shutdown state");
           stateManager.gotoState(ConnectivityState.IDLE);
           activeTransport = null;
           closedByServer = true;
         } else if (activeTransport == delayedTransport) {
           // shutdown() should have set activeTransport to null
           Preconditions.checkState(!shutdown, "unexpected shutdown state");
           // Continue reconnect if there are still addresses to try.
           if (nextAddressIndex == 0) {
             allAddressesFailed = true;
           } else {
             Preconditions.checkState(stateManager.getState() == ConnectivityState.CONNECTING,
                 "Expected state is CONNECTING, actual state is %s", stateManager.getState());
             runnable = startNewTransport(delayedTransport);
           }
         }
       }
       if (allAddressesFailed) {
         // Initiate backoff
         // Transition to TRANSIENT_FAILURE
         scheduleBackoff(delayedTransport, s);
       }
       if (runnable != null) {
         runnable.run();
       }
       loadBalancer.handleTransportShutdown(addressGroup, s);
       if (allAddressesFailed) {
         callback.onAllAddressesFailed();
       }
       if (closedByServer) {
         callback.onConnectionClosedByServer(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});
       }
       super.transportTerminated();
       Preconditions.checkState(activeTransport != transport,
           "activeTransport still points to the delayedTransport. "
           + "Seems transportShutdown() was not called.");
     }
   }

   abstract static class Callback {
     /**
      * Called when the TransportSet is terminated, which means it's shut down and all transports
      * have been terminated.
      */
     public void onTerminated(TransportSet ts) { }

     /**
      * Called when all addresses have failed to connect.
      */
     public void onAllAddressesFailed() { }

     /**
      * Called when a once-live connection is shut down by server-side.
      */
     public void onConnectionClosedByServer(Status status) { }

     /**
      * Called when the TransportSet's in-use state has changed to true, which means at least one
      * transport is in use. This method is called under a lock thus externally synchronized. If the
      * return value is non-{@code null}, the runnable will be executed after releasing the lock.
      */
     @CheckReturnValue
     public Runnable onInUse(TransportSet ts) {
       return null;
     }

     /**
      * Called when the TransportSet's in-use state has changed to false, which means no transport is
      * in use. This method is called under a lock thus externally synchronized.
      */
     public void onNotInUse(TransportSet ts) { }
   }
 }
	/*
	* Copyright 2015, Google Inc. All rights reserved.
	*
	* Redistribution and use in source and binary forms, with or without
	* modification, are permitted provided that the following conditions are
	* met:
	*
	* * Redistributions of source code must retain the above copyright
	* notice, this list of conditions and the following disclaimer.
	* * Redistributions in binary form must reproduce the above
	* copyright notice, this list of conditions and the following disclaimer
	* in the documentation and/or other materials provided with the
	* distribution.
	*
	* * Neither the name of Google Inc. nor the names of its
	* contributors may be used to endorse or promote products derived from
	* this software without specific prior written permission.
	*
	* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
	* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
	* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
	* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
	* OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
	* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
	* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
	* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
	* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
	* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
	* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
	*/

	package io.grpc.internal;

	import com.google.common.base.Preconditions;
	import com.google.common.base.Stopwatch;
	import com.google.common.base.Supplier;

	import io.grpc.CallOptions;
	import io.grpc.ClientCall;
	import io.grpc.ConnectivityState;
	import io.grpc.EquivalentAddressGroup;
	import io.grpc.LoadBalancer;
	import io.grpc.ManagedChannel;
	import io.grpc.Metadata;
	import io.grpc.MethodDescriptor;
	import io.grpc.Status;
	import io.grpc.internal.ClientCallImpl.ClientTransportProvider;

	import java.net.SocketAddress;
	import java.util.ArrayList;
	import java.util.Collection;
	import java.util.List;
	import java.util.concurrent.CountDownLatch;
	import java.util.concurrent.Executor;
	import java.util.concurrent.ScheduledExecutorService;
	import java.util.concurrent.ScheduledFuture;
	import java.util.concurrent.TimeUnit;
	import java.util.logging.Level;
	import java.util.logging.Logger;

	import javax.annotation.CheckReturnValue;
	import javax.annotation.Nullable;
	import javax.annotation.concurrent.GuardedBy;
	import javax.annotation.concurrent.ThreadSafe;

	/**
	* Transports for a single {@link SocketAddress}.
	*/
	@ThreadSafe
	final class TransportSet extends ManagedChannel implements WithLogId {
	private static final Logger log = Logger.getLogger(TransportSet.class.getName());
	private static final ClientTransport SHUTDOWN_TRANSPORT =
	new FailingClientTransport(Status.UNAVAILABLE.withDescription("TransportSet is shutdown"));

	private final CountDownLatch terminatedLatch = new CountDownLatch(1);
	private final Object lock = new Object();
	private final LogId logId = LogId.allocate(getClass().getName());
	private final EquivalentAddressGroup addressGroup;
	private final String authority;
	private final String userAgent;
	private final BackoffPolicy.Provider backoffPolicyProvider;
	private final Callback callback;
	private final ClientTransportFactory transportFactory;
	private final ScheduledExecutorService scheduledExecutor;
	private final Executor appExecutor;

	@GuardedBy("lock")
	private int nextAddressIndex;

	/**
	* The policy to control back off between reconnects. Non-{@code null} when last connect failed.
	*/
	@GuardedBy("lock")
	private BackoffPolicy reconnectPolicy;

	/**
	* Timer monitoring duration since entering CONNECTING state.
	*/
	@GuardedBy("lock")
	private final Stopwatch connectingTimer;

	@GuardedBy("lock")
	@Nullable
	private ScheduledFuture<?> reconnectTask;

	/**
	* All transports that are not terminated. At the very least the value of {@link #activeTransport}
	* will be present, but previously used transports that still have streams or are stopping may
	* also be present.
	*/
	@GuardedBy("lock")
	private final Collection<ManagedClientTransport> transports =
	new ArrayList<ManagedClientTransport>();

	private final InUseStateAggregator<ManagedClientTransport> inUseStateAggregator =
	new InUseStateAggregator<ManagedClientTransport>() {
	@Override
	Object getLock() {
	return lock;
	}

	@Override
	Runnable handleInUse() {
	return callback.onInUse(TransportSet.this);
	}

	@Override
	void handleNotInUse() {
	callback.onNotInUse(TransportSet.this);
	}
	};

	/**
	* The to-be active transport, which is not ready yet.
	*/
	@GuardedBy("lock")
	@Nullable
	private ConnectionClientTransport pendingTransport;

	private final LoadBalancer<ClientTransport> loadBalancer;

	@GuardedBy("lock")
	private boolean shutdown;

	/**
	* The transport for new outgoing requests. 'lock' must be held when assigning to it.
	*
	* <pre><code>
	* State Value
	* ----- ------
	* IDLE null, shutdown == false
	* CONNECTING instanceof DelayedTransport, reconnectTask == null
	* READY connected transport
	* TRANSIENT_FAILURE instanceof DelayedTransport, reconnectTask != null
	* SHUTDOWN null, shutdown == true
	* </code></pre>
	*/
	@Nullable
	private volatile ManagedClientTransport activeTransport;

	@GuardedBy("lock")
	private final ConnectivityStateManager stateManager =
	new ConnectivityStateManager(ConnectivityState.IDLE);

	TransportSet(EquivalentAddressGroup addressGroup, String authority, String userAgent,
	LoadBalancer<ClientTransport> loadBalancer, BackoffPolicy.Provider backoffPolicyProvider,
	ClientTransportFactory transportFactory, ScheduledExecutorService scheduledExecutor,
	Supplier<Stopwatch> stopwatchSupplier, Executor appExecutor, Callback callback) {
	this.addressGroup = Preconditions.checkNotNull(addressGroup, "addressGroup");
	this.authority = authority;
	this.userAgent = userAgent;
	this.loadBalancer = loadBalancer;
	this.backoffPolicyProvider = backoffPolicyProvider;
	this.transportFactory = transportFactory;
	this.scheduledExecutor = scheduledExecutor;
	this.connectingTimer = stopwatchSupplier.get();
	this.appExecutor = appExecutor;
	this.callback = callback;
	}

	/**
	* Returns the active transport that will be used to create new streams.
	*
	* <p>Never returns {@code null}.
	*/
	final ClientTransport obtainActiveTransport() {
	ClientTransport savedTransport = activeTransport;
	if (savedTransport != null) {
	return savedTransport;
	}
	Runnable runnable;
	synchronized (lock) {
	// Check again, since it could have changed before acquiring the lock
	savedTransport = activeTransport;
	if (savedTransport != null) {
	return savedTransport;
	}
	if (shutdown) {
	return SHUTDOWN_TRANSPORT;
	}
	stateManager.gotoState(ConnectivityState.CONNECTING);
	DelayedClientTransport delayedTransport = new DelayedClientTransport(appExecutor);
	transports.add(delayedTransport);
	delayedTransport.start(new BaseTransportListener(delayedTransport));
	savedTransport = activeTransport = delayedTransport;
	runnable = startNewTransport(delayedTransport);
	}
	if (runnable != null) {
	runnable.run();
	}
	return savedTransport;
	}

	@CheckReturnValue
	@GuardedBy("lock")
	private Runnable startNewTransport(DelayedClientTransport delayedTransport) {
	Preconditions.checkState(reconnectTask == null, "Should have no reconnectTask scheduled");

	if (nextAddressIndex == 0) {
	connectingTimer.reset().start();
	}
	List<SocketAddress> addrs = addressGroup.getAddresses();
	final SocketAddress address = addrs.get(nextAddressIndex++);
	if (nextAddressIndex >= addrs.size()) {
	nextAddressIndex = 0;
	}

	ConnectionClientTransport transport =
	transportFactory.newClientTransport(address, authority, userAgent);
	if (log.isLoggable(Level.FINE)) {
	log.log(Level.FINE, "[{0}] Created {1} for {2}",
	new Object[] {getLogId(), transport.getLogId(), address});
	}
	pendingTransport = transport;
	transports.add(transport);
	return transport.start(new TransportListener(transport, delayedTransport, address));
	}

	/**
	* Only called after all addresses attempted and failed (TRANSIENT_FAILURE).
	* @param status the causal status when the channel begins transition to
	* TRANSIENT_FAILURE.
	*/
	private void scheduleBackoff(
	final DelayedClientTransport delayedTransport, final Status status) {
	// This must be run outside of lock. The TransportSet lock is a channel level lock.
	// startBackoff() will acquire the delayed transport lock, which is a transport level
	// lock. Our lock ordering mandates transport lock > channel lock. Otherwise a deadlock
	// could happen (https://github.com/grpc/grpc-java/issues/2152).
	delayedTransport.startBackoff(status);

	class EndOfCurrentBackoff implements Runnable {
	@Override
	public void run() {
	try {
	delayedTransport.endBackoff();
	Runnable runnable = null;
	synchronized (lock) {
	reconnectTask = null;
	if (!shutdown) {
	stateManager.gotoState(ConnectivityState.CONNECTING);
	}
	runnable = startNewTransport(delayedTransport);
	}
	if (runnable != null) {
	runnable.run();
	}
	} catch (Throwable t) {
	log.log(Level.WARNING, "Exception handling end of backoff", t);
	}
	}
	}

	synchronized (lock) {
	if (shutdown) {
	return;
	}
	stateManager.gotoState(ConnectivityState.TRANSIENT_FAILURE);
	if (reconnectPolicy == null) {
	reconnectPolicy = backoffPolicyProvider.get();
	}
	long delayMillis =
	reconnectPolicy.nextBackoffMillis() - connectingTimer.elapsed(TimeUnit.MILLISECONDS);
	if (log.isLoggable(Level.FINE)) {
	log.log(Level.FINE, "[{0}] Scheduling backoff for {1} ms",
	new Object[]{getLogId(), delayMillis});
	}
	Preconditions.checkState(reconnectTask == null, "previous reconnectTask is not done");
	reconnectTask = scheduledExecutor.schedule(
	new LogExceptionRunnable(new EndOfCurrentBackoff()),
	delayMillis,
	TimeUnit.MILLISECONDS);
	}
	}

	@Override
	public ManagedChannel shutdown() {
	ManagedClientTransport savedActiveTransport;
	ConnectionClientTransport savedPendingTransport;
	boolean runCallback = false;
	synchronized (lock) {
	if (shutdown) {
	return this;
	}
	stateManager.gotoState(ConnectivityState.SHUTDOWN);
	shutdown = true;
	savedActiveTransport = activeTransport;
	savedPendingTransport = pendingTransport;
	activeTransport = null;
	if (transports.isEmpty()) {
	runCallback = true;
	terminatedLatch.countDown();
	if (log.isLoggable(Level.FINE)) {
	log.log(Level.FINE, "[{0}] Terminated in shutdown()", getLogId());
	}
	Preconditions.checkState(reconnectTask == null, "Should have no reconnectTask scheduled");
	} // else: the callback will be run once all transports have been terminated
	}
	if (savedActiveTransport != null) {
	savedActiveTransport.shutdown();
	}
	if (savedPendingTransport != null) {
	savedPendingTransport.shutdown();
	}
	if (runCallback) {
	callback.onTerminated(this);
	}
	return this;
	}

	void shutdownNow(Status reason) {
	shutdown();
	Collection<ManagedClientTransport> transportsCopy;
	synchronized (lock) {
	transportsCopy = new ArrayList<ManagedClientTransport>(transports);
	}
	for (ManagedClientTransport transport : transportsCopy) {
	transport.shutdownNow(reason);
	}
	}

	@Override
	public ManagedChannel shutdownNow() {
	shutdownNow(Status.UNAVAILABLE.withDescription("TransportSet shutdown as ManagedChannel"));
	return this;
	}

	@GuardedBy("lock")
	private void cancelReconnectTask() {
	if (reconnectTask != null) {
	reconnectTask.cancel(false);
	reconnectTask = null;
	}
	}

	@Override
	public LogId getLogId() {
	return logId;
	}

	@Override
	public final <RequestT, ResponseT> ClientCall<RequestT, ResponseT> newCall(
	MethodDescriptor<RequestT, ResponseT> methodDescriptor, CallOptions callOptions) {
	return new ClientCallImpl<RequestT, ResponseT>(methodDescriptor,
	new SerializingExecutor(appExecutor), callOptions, StatsTraceContext.NOOP,
	new ClientTransportProvider() {
	@Override
	public ClientTransport get(CallOptions callOptions, Metadata headers) {
	return obtainActiveTransport();
	}
	},
	scheduledExecutor);
	}

	@Override
	public boolean isShutdown() {
	synchronized (lock) {
	return shutdown;
	}
	}

	@Override
	public boolean isTerminated() {
	return terminatedLatch.getCount() == 0;
	}

	@Override
	public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
	return terminatedLatch.await(timeout, unit);
	}

	@Override
	public String authority() {
	return authority;
	}

	@Override
	public ConnectivityState getState(boolean requestConnection) {
	if (requestConnection) {
	boolean connect = false;
	synchronized (lock) {
	connect = stateManager.getState() == ConnectivityState.IDLE;
	}
	if (connect) {
	obtainActiveTransport();
	}
	}
	synchronized (lock) {
	return stateManager.getState();
	}
	}

	@Override
	public void notifyWhenStateChanged(ConnectivityState source, Runnable callback) {
	synchronized (lock) {
	stateManager.notifyWhenStateChanged(callback, appExecutor, source);
	}
	}

	/** Shared base for both delayed and real transports. */
	private class BaseTransportListener implements ManagedClientTransport.Listener {
	protected final ManagedClientTransport transport;

	public BaseTransportListener(ManagedClientTransport transport) {
	this.transport = transport;
	}

	@Override
	public void transportReady() {}

	@Override
	public void transportInUse(boolean inUse) {
	Runnable r = inUseStateAggregator.updateObjectInUse(transport, inUse);
	if (r != null) {
	r.run();
	}
	}

	@Override
	public void transportShutdown(Status status) {}

	@Override
	public void transportTerminated() {
	boolean runCallback = false;
	Runnable r = inUseStateAggregator.updateObjectInUse(transport, false);
	assert r == null;
	synchronized (lock) {
	transports.remove(transport);
	if (shutdown && transports.isEmpty()) {
	if (log.isLoggable(Level.FINE)) {
	log.log(Level.FINE, "[{0}] Terminated in transportTerminated()", getLogId());
	}
	terminatedLatch.countDown();
	runCallback = true;
	cancelReconnectTask();
	}
	}
	if (runCallback) {
	callback.onTerminated(TransportSet.this);
	}
	}
	}

	/** Listener for real transports. */
	private class TransportListener extends BaseTransportListener {
	private final SocketAddress address;
	private final DelayedClientTransport delayedTransport;

	public TransportListener(ManagedClientTransport transport,
	DelayedClientTransport delayedTransport, SocketAddress address) {
	super(transport);
	this.address = address;
	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});
	}
	super.transportReady();
	boolean savedShutdown;
	synchronized (lock) {
	savedShutdown = shutdown;
	reconnectPolicy = null;
	nextAddressIndex = 0;
	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) {
	stateManager.gotoState(ConnectivityState.READY);
	Preconditions.checkState(pendingTransport == transport, "transport mismatch");
	activeTransport = transport;
	pendingTransport = null;
	}
	}
	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) {
	boolean allAddressesFailed = false;
	boolean closedByServer = false;
	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});
	}
	super.transportShutdown(s);
	Runnable runnable = null;
	synchronized (lock) {
	if (activeTransport == transport) {
	// This is true only if the transport was ready.
	// shutdown() should have set activeTransport to null
	Preconditions.checkState(!shutdown, "unexpected shutdown state");
	stateManager.gotoState(ConnectivityState.IDLE);
	activeTransport = null;
	closedByServer = true;
	} else if (activeTransport == delayedTransport) {
	// shutdown() should have set activeTransport to null
	Preconditions.checkState(!shutdown, "unexpected shutdown state");
	// Continue reconnect if there are still addresses to try.
	if (nextAddressIndex == 0) {
	allAddressesFailed = true;
	} else {
	Preconditions.checkState(stateManager.getState() == ConnectivityState.CONNECTING,
	"Expected state is CONNECTING, actual state is %s", stateManager.getState());
	runnable = startNewTransport(delayedTransport);
	}
	}
	}
	if (allAddressesFailed) {
	// Initiate backoff
	// Transition to TRANSIENT_FAILURE
	scheduleBackoff(delayedTransport, s);
	}
	if (runnable != null) {
	runnable.run();
	}
	loadBalancer.handleTransportShutdown(addressGroup, s);
	if (allAddressesFailed) {
	callback.onAllAddressesFailed();
	}
	if (closedByServer) {
	callback.onConnectionClosedByServer(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});
	}
	super.transportTerminated();
	Preconditions.checkState(activeTransport != transport,
	"activeTransport still points to the delayedTransport. "
	+ "Seems transportShutdown() was not called.");
	}
	}

	abstract static class Callback {
	/**
	* Called when the TransportSet is terminated, which means it's shut down and all transports
	* have been terminated.
	*/
	public void onTerminated(TransportSet ts) { }

	/**
	* Called when all addresses have failed to connect.
	*/
	public void onAllAddressesFailed() { }

	/**
	* Called when a once-live connection is shut down by server-side.
	*/
	public void onConnectionClosedByServer(Status status) { }

	/**
	* Called when the TransportSet's in-use state has changed to true, which means at least one
	* transport is in use. This method is called under a lock thus externally synchronized. If the
	* return value is non-{@code null}, the runnable will be executed after releasing the lock.
	*/
	@CheckReturnValue
	public Runnable onInUse(TransportSet ts) {
	return null;
	}

	/**
	* Called when the TransportSet's in-use state has changed to false, which means no transport is
	* in use. This method is called under a lock thus externally synchronized.
	*/
	public void onNotInUse(TransportSet ts) { }
	}
	}