{"action":"created","comment":{"url":"https://api.github.com/repos/grpc/grpc-java/pulls/comments/94174176","pull_request_review_id":14705294,"id":94174176,"diff_hunk":"@@ -0,0 +1,873 @@\n+/*\n+ * Copyright 2016, Google Inc. All rights reserved.\n+ *\n+ * Redistribution and use in source and binary forms, with or without\n+ * modification, are permitted provided that the following conditions are\n+ * met:\n+ *\n+ *    * Redistributions of source code must retain the above copyright\n+ * notice, this list of conditions and the following disclaimer.\n+ *    * Redistributions in binary form must reproduce the above\n+ * copyright notice, this list of conditions and the following disclaimer\n+ * in the documentation and/or other materials provided with the\n+ * distribution.\n+ *\n+ *    * Neither the name of Google Inc. nor the names of its\n+ * contributors may be used to endorse or promote products derived from\n+ * this software without specific prior written permission.\n+ *\n+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS\n+ * \"AS IS\" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT\n+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR\n+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT\n+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,\n+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT\n+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,\n+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY\n+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT\n+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE\n+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.\n+ */\n+\n+package io.grpc.internal;\n+\n+import static com.google.common.base.Preconditions.checkArgument;\n+import static com.google.common.base.Preconditions.checkNotNull;\n+import static com.google.common.base.Preconditions.checkState;\n+import static io.grpc.ConnectivityState.IDLE;\n+import static io.grpc.ConnectivityState.TRANSIENT_FAILURE;\n+\n+import com.google.census.CensusContextFactory;\n+import com.google.common.annotations.VisibleForTesting;\n+import com.google.common.base.Stopwatch;\n+import com.google.common.base.Supplier;\n+import com.google.common.util.concurrent.MoreExecutors;\n+\n+import io.grpc.Attributes;\n+import io.grpc.CallOptions;\n+import io.grpc.Channel;\n+import io.grpc.ClientCall;\n+import io.grpc.ClientInterceptor;\n+import io.grpc.ClientInterceptors;\n+import io.grpc.CompressorRegistry;\n+import io.grpc.ConnectivityStateInfo;\n+import io.grpc.DecompressorRegistry;\n+import io.grpc.EquivalentAddressGroup;\n+import io.grpc.LoadBalancer2.PickResult;\n+import io.grpc.LoadBalancer2.SubchannelPicker;\n+import io.grpc.LoadBalancer2;\n+import io.grpc.ManagedChannel;\n+import io.grpc.Metadata;\n+import io.grpc.MethodDescriptor;\n+import io.grpc.NameResolver;\n+import io.grpc.ResolvedServerInfoGroup;\n+import io.grpc.Status;\n+import io.grpc.internal.ClientCallImpl.ClientTransportProvider;\n+\n+import java.net.URI;\n+import java.net.URISyntaxException;\n+import java.util.HashSet;\n+import java.util.List;\n+import java.util.Set;\n+import java.util.concurrent.CountDownLatch;\n+import java.util.concurrent.Executor;\n+import java.util.concurrent.ScheduledExecutorService;\n+import java.util.concurrent.ScheduledFuture;\n+import java.util.concurrent.TimeUnit;\n+import java.util.concurrent.atomic.AtomicBoolean;\n+import java.util.logging.Level;\n+import java.util.logging.Logger;\n+import java.util.regex.Pattern;\n+\n+import javax.annotation.Nullable;\n+import javax.annotation.concurrent.GuardedBy;\n+import javax.annotation.concurrent.ThreadSafe;\n+\n+/** A communication channel for making outgoing RPCs. */\n+@ThreadSafe\n+public final class ManagedChannelImpl2 extends ManagedChannel implements WithLogId {\n+  private static final Logger log = Logger.getLogger(ManagedChannelImpl2.class.getName());\n+\n+  // Matching this pattern means the target string is a URI target or at least intended to be one.\n+  // A URI target must be an absolute hierarchical URI.\n+  // From RFC 2396: scheme = alpha *( alpha | digit | \"+\" | \"-\" | \".\" )\n+  @VisibleForTesting\n+  static final Pattern URI_PATTERN = Pattern.compile(\"[a-zA-Z][a-zA-Z0-9+.-]*:/.*\");\n+\n+  static final long IDLE_TIMEOUT_MILLIS_DISABLE = -1;\n+\n+  @VisibleForTesting\n+  static final long SUBCHANNEL_SHUTDOWN_DELAY_SECONDS = 5;\n+\n+  private static final ClientTransport SHUTDOWN_TRANSPORT =\n+      new FailingClientTransport(Status.UNAVAILABLE.withDescription(\"Channel is shutdown\"));\n+\n+  @VisibleForTesting\n+  static final Status SHUTDOWN_NOW_STATUS =\n+      Status.UNAVAILABLE.withDescription(\"Channel shutdownNow invoked\");\n+\n+  private final String target;\n+  private final NameResolver.Factory nameResolverFactory;\n+  private final Attributes nameResolverParams;\n+  private final LoadBalancer2.Factory loadBalancerFactory;\n+  private final ClientTransportFactory transportFactory;\n+  private volatile Executor executor;\n+  private final ObjectPool<? extends Executor> executorPool;\n+  private final ObjectPool<? extends Executor> oobExecutorPool;\n+  private final LogId logId = LogId.allocate(getClass().getName());\n+\n+  private final ChannelExecutor channelExecutor = new ChannelExecutor();\n+\n+  private final DecompressorRegistry decompressorRegistry;\n+  private final CompressorRegistry compressorRegistry;\n+\n+  private final ObjectPool<ScheduledExecutorService> timerServicePool;\n+  private final Supplier<Stopwatch> stopwatchSupplier;\n+  /** The timout before entering idle mode. */\n+  private final long idleTimeoutMillis;\n+  private final CensusContextFactory censusFactory;\n+\n+  /**\n+   * Executor that runs deadline timers for requests.\n+   */\n+  // Must be assigned from channelExecutor\n+  private volatile ScheduledExecutorService scheduledExecutor;\n+\n+  private final BackoffPolicy.Provider backoffPolicyProvider;\n+\n+  /**\n+   * We delegate to this channel, so that we can have interceptors as necessary. If there aren't\n+   * any interceptors this will just be {@link RealChannel}.\n+   */\n+  private final Channel interceptorChannel;\n+  @Nullable private final String userAgent;\n+\n+  // Only null after channel is terminated. Must be assigned from the channelExecutor.\n+  private NameResolver nameResolver;\n+\n+  // null when channel is in idle mode.  Must be assigned from channelExecutor.\n+  @Nullable\n+  private LoadBalancer2 loadBalancer;\n+\n+  // Must be assigned from channelExecutor.  null if channel is in idle mode.\n+  @Nullable\n+  private volatile SubchannelPicker subchannelPicker;\n+\n+  // Must be mutated from channelExecutor\n+  // If any monitoring hook to be added later needs to get a snapshot of this Set, we could\n+  // switch to a ConcurrentHashMap.\n+  private final Set<InternalSubchannel> subchannels = new HashSet<InternalSubchannel>(16, .75f);\n+\n+  // Must be mutated from channelExecutor\n+  private final Set<InternalSubchannel> oobChannels = new HashSet<InternalSubchannel>(1, .75f);\n+\n+  // reprocess() must be run from channelExecutor\n+  private final DelayedClientTransport2 delayedTransport;\n+\n+  // Shutdown states.\n+  //\n+  // Channel's shutdown process:\n+  // 1. shutdown(): stop accepting new calls from applications\n+  //   1a shutdown <- true\n+  //   1b subchannelPicker <- null\n+  //   1c delayedTransport.shutdown()\n+  // 2. delayedTransport terminated: stop stream-creation functionality\n+  //   2a terminating <- true\n+  //   2b loadBalancer.shutdown()\n+  //     * LoadBalancer will shutdown subchannels and OOB channels\n+  //   2c loadBalancer <- null\n+  //   2d nameResolver.shutdown()\n+  //   2e nameResolver <- null\n+  // 3. All subchannels and OOB channels terminated: Channel considered terminated\n+\n+  private final AtomicBoolean shutdown = new AtomicBoolean(false);\n+  // Must only be mutated and read from channelExecutor\n+  private boolean shutdownNowed;\n+  // Must be mutated from channelExecutor\n+  private volatile boolean terminating;\n+  // Must be mutated from channelExecutor\n+  private volatile boolean terminated;\n+  private final CountDownLatch terminatedLatch = new CountDownLatch(1);\n+\n+  // Called from channelExecutor\n+  private final ManagedClientTransport.Listener delayedTransportListener =\n+      new ManagedClientTransport.Listener() {\n+        @Override\n+        public void transportShutdown(Status s) {\n+          checkState(shutdown.get(), \"Channel must have been shut down\");\n+        }\n+\n+        @Override\n+        public void transportReady() {\n+          // Don't care\n+        }\n+\n+        @Override\n+        public void transportInUse(final boolean inUse) {\n+          inUseStateAggregator.updateObjectInUse(delayedTransport, inUse);\n+        }\n+\n+        @Override\n+        public void transportTerminated() {\n+          checkState(shutdown.get(), \"Channel must have been shut down\");\n+          terminating = true;\n+          if (loadBalancer != null) {\n+            loadBalancer.shutdown();\n+            loadBalancer = null;\n+          }\n+          if (nameResolver != null) {\n+            nameResolver.shutdown();\n+            nameResolver = null;\n+          }\n+\n+          // Until LoadBalancer is shutdown, it may still create new subchannels.  We catch them\n+          // here.\n+          maybeShutdownNowSubchannels();\n+          maybeTerminateChannel();\n+        }\n+      };\n+\n+  // Must be called from channelExecutor\n+  private void maybeShutdownNowSubchannels() {\n+    if (shutdownNowed) {\n+      for (InternalSubchannel subchannel : subchannels) {\n+        subchannel.shutdownNow(SHUTDOWN_NOW_STATUS);\n+      }\n+      subchannels.clear();\n+      for (InternalSubchannel oobChannel : oobChannels) {\n+        oobChannel.shutdownNow(SHUTDOWN_NOW_STATUS);\n+      }\n+      oobChannels.clear();\n+    }\n+  }\n+\n+  // Must be accessed from channelExecutor\n+  @VisibleForTesting\n+  final InUseStateAggregator2<Object> inUseStateAggregator =\n+      new InUseStateAggregator2<Object>() {\n+        @Override\n+        void handleInUse() {\n+          exitIdleMode();\n+        }\n+\n+        @Override\n+        void handleNotInUse() {\n+          if (shutdown.get()) {\n+            return;\n+          }\n+          rescheduleIdleTimer();\n+        }\n+      };\n+\n+  // Run from channelExecutor\n+  private class IdleModeTimer implements Runnable {\n+    // Only mutated from channelExecutor\n+    boolean cancelled;\n+\n+    @Override\n+    public void run() {\n+      if (cancelled) {\n+        // Race detected: this task was scheduled on channelExecutor before cancelIdleTimer()\n+        // could cancel the timer.\n+        return;\n+      }\n+      log.log(Level.FINE, \"[{0}] Entering idle mode\", getLogId());\n+      // nameResolver and loadBalancer are guaranteed to be non-null.  If any of them were null,\n+      // either the idleModeTimer ran twice without exiting the idle mode, or the task in shutdown()\n+      // did not cancel idleModeTimer, both of which are bugs.\n+      nameResolver.shutdown();\n+      nameResolver = getNameResolver(target, nameResolverFactory, nameResolverParams);\n+      loadBalancer.shutdown();\n+      loadBalancer = null;\n+      subchannelPicker = null;\n+    }\n+  }\n+\n+  // Must be used from channelExecutor\n+  @Nullable\n+  private ScheduledFuture<?> idleModeTimerFuture;\n+  // Must be used from channelExecutor\n+  @Nullable\n+  private IdleModeTimer idleModeTimer;\n+\n+  /**\n+   * Make the channel exit idle mode, if it's in it.\n+   *\n+   * <p>Must be called from channelExecutor\n+   */\n+  @VisibleForTesting\n+  void exitIdleMode() {\n+    if (shutdown.get()) {\n+      return;\n+    }\n+    // Cancel the timer now, so that a racing due timer will not put Channel on idleness\n+    // when the caller of exitIdleMode() is about to use the returned loadBalancer.\n+    cancelIdleTimer();\n+    // exitIdleMode() may be called outside of inUseStateAggregator.handleNotInUse() while\n+    // isInUse() == false, in which case we still need to schedule the timer.\n+    if (!inUseStateAggregator.isInUse()) {\n+      rescheduleIdleTimer();\n+    }\n+    if (loadBalancer != null) {\n+      return;\n+    }\n+    log.log(Level.FINE, \"[{0}] Exiting idle mode\", getLogId());\n+    LbHelperImpl helper = new LbHelperImpl(nameResolver);\n+    helper.lb = loadBalancerFactory.newLoadBalancer(helper);\n+    this.loadBalancer = helper.lb;\n+\n+    NameResolverListenerImpl listener = new NameResolverListenerImpl(helper.lb);\n+    try {\n+      nameResolver.start(listener);\n+    } catch (Throwable t) {\n+      listener.onError(Status.fromThrowable(t));\n+    }\n+  }\n+\n+  // Must be run from channelExecutor\n+  private void cancelIdleTimer() {\n+    if (idleModeTimerFuture != null) {\n+      idleModeTimerFuture.cancel(false);\n+      idleModeTimer.cancelled = true;\n+      idleModeTimerFuture = null;\n+      idleModeTimer = null;\n+    }\n+  }\n+\n+  // Always run from channelExecutor\n+  private void rescheduleIdleTimer() {\n+    if (idleTimeoutMillis == IDLE_TIMEOUT_MILLIS_DISABLE) {\n+      return;\n+    }\n+    cancelIdleTimer();\n+    idleModeTimer = new IdleModeTimer();\n+    idleModeTimerFuture = scheduledExecutor.schedule(\n+        new LogExceptionRunnable(new Runnable() {\n+            @Override\n+            public void run() {\n+              channelExecutor.executeLater(idleModeTimer).drain();\n+            }\n+          }),\n+        idleTimeoutMillis, TimeUnit.MILLISECONDS);\n+  }\n+\n+  private final ClientTransportProvider transportProvider = new ClientTransportProvider() {\n+    @Override\n+    public ClientTransport get(CallOptions callOptions, Metadata headers) {\n+      SubchannelPicker pickerCopy = subchannelPicker;\n+      if (shutdown.get()) {\n+        // If channel is shut down, delayedTransport is also shut down which will fail the stream\n+        // properly.\n+        return delayedTransport;\n+      }\n+      if (pickerCopy == null) {\n+        channelExecutor.executeLater(new Runnable() {\n+            @Override\n+            public void run() {\n+              exitIdleMode();\n+            }\n+          }).drain();\n+        return delayedTransport;\n+      }\n+      // There is no need to reschedule the idle timer here.\n+      //\n+      // pickerCopy != null, which means idle timer has not expired when this method starts.\n+      // Even if idle timer expires right after we grab pickerCopy, and it shuts down LoadBalancer\n+      // which calls Subchannel.shutdown(), the InternalSubchannel will be actually shutdown after\n+      // SUBCHANNEL_SHUTDOWN_DELAY_SECONDS, which gives the caller time to start RPC on it.\n+      //\n+      // In most cases the idle timer is scheduled to fire after the transport has created the\n+      // stream, which would have reported in-use state to the channel that would have cancelled\n+      // the idle timer.\n+      PickResult pickResult = pickerCopy.pickSubchannel(callOptions.getAffinity(), headers);\n+      ClientTransport transport = GrpcUtil.getTransportFromPickResult(\n+          pickResult, callOptions.isWaitForReady());\n+      if (transport != null) {\n+        return transport;\n+      }\n+      return delayedTransport;\n+    }\n+  };\n+\n+  ManagedChannelImpl2(String target, BackoffPolicy.Provider backoffPolicyProvider,\n+      NameResolver.Factory nameResolverFactory, Attributes nameResolverParams,\n+      LoadBalancer2.Factory loadBalancerFactory, ClientTransportFactory transportFactory,\n+      DecompressorRegistry decompressorRegistry, CompressorRegistry compressorRegistry,\n+      ObjectPool<ScheduledExecutorService> timerServicePool,\n+      ObjectPool<? extends Executor> executorPool, ObjectPool<? extends Executor> oobExecutorPool,\n+      Supplier<Stopwatch> stopwatchSupplier, long idleTimeoutMillis,\n+      @Nullable String userAgent,\n+      List<ClientInterceptor> interceptors, CensusContextFactory censusFactory) {\n+    this.target = checkNotNull(target, \"target\");\n+    this.nameResolverFactory = checkNotNull(nameResolverFactory, \"nameResolverFactory\");\n+    this.nameResolverParams = checkNotNull(nameResolverParams, \"nameResolverParams\");\n+    this.nameResolver = getNameResolver(target, nameResolverFactory, nameResolverParams);\n+    this.loadBalancerFactory = checkNotNull(loadBalancerFactory, \"loadBalancerFactory\");\n+    this.executorPool = checkNotNull(executorPool, \"executorPool\");\n+    this.oobExecutorPool = checkNotNull(oobExecutorPool, \"oobExecutorPool\");\n+    this.executor = checkNotNull(executorPool.getObject(), \"executor\");\n+    this.delayedTransport = new DelayedClientTransport2(this.executor, this.channelExecutor);\n+    this.delayedTransport.start(delayedTransportListener);\n+    this.backoffPolicyProvider = backoffPolicyProvider;\n+    this.transportFactory =\n+        new CallCredentialsApplyingTransportFactory(transportFactory, this.executor);\n+    this.interceptorChannel = ClientInterceptors.intercept(new RealChannel(), interceptors);\n+    this.timerServicePool = timerServicePool;\n+    this.scheduledExecutor = timerServicePool.getObject();\n+    this.stopwatchSupplier = checkNotNull(stopwatchSupplier, \"stopwatchSupplier\");\n+    if (idleTimeoutMillis == IDLE_TIMEOUT_MILLIS_DISABLE) {\n+      this.idleTimeoutMillis = idleTimeoutMillis;\n+    } else {\n+      checkArgument(\n+          idleTimeoutMillis >= AbstractManagedChannelImplBuilder.IDLE_MODE_MIN_TIMEOUT_MILLIS,\n+          \"invalid idleTimeoutMillis %s\", idleTimeoutMillis);\n+      this.idleTimeoutMillis = idleTimeoutMillis;\n+    }\n+    this.decompressorRegistry = decompressorRegistry;\n+    this.compressorRegistry = compressorRegistry;\n+    this.userAgent = userAgent;\n+    this.censusFactory = checkNotNull(censusFactory, \"censusFactory\");\n+\n+    log.log(Level.FINE, \"[{0}] Created with target {1}\", new Object[] {getLogId(), target});\n+  }\n+\n+  @VisibleForTesting\n+  static NameResolver getNameResolver(String target, NameResolver.Factory nameResolverFactory,\n+      Attributes nameResolverParams) {\n+    // Finding a NameResolver. Try using the target string as the URI. If that fails, try prepending\n+    // \"dns:///\".\n+    URI targetUri = null;\n+    StringBuilder uriSyntaxErrors = new StringBuilder();\n+    try {\n+      targetUri = new URI(target);\n+      // For \"localhost:8080\" this would likely cause newNameResolver to return null, because\n+      // \"localhost\" is parsed as the scheme. Will fall into the next branch and try\n+      // \"dns:///localhost:8080\".\n+    } catch (URISyntaxException e) {\n+      // Can happen with ip addresses like \"[::1]:1234\" or 127.0.0.1:1234.\n+      uriSyntaxErrors.append(e.getMessage());\n+    }\n+    if (targetUri != null) {\n+      NameResolver resolver = nameResolverFactory.newNameResolver(targetUri, nameResolverParams);\n+      if (resolver != null) {\n+        return resolver;\n+      }\n+      // \"foo.googleapis.com:8080\" cause resolver to be null, because \"foo.googleapis.com\" is an\n+      // unmapped scheme. Just fall through and will try \"dns:///foo.googleapis.com:8080\"\n+    }\n+\n+    // If we reached here, the targetUri couldn't be used.\n+    if (!URI_PATTERN.matcher(target).matches()) {\n+      // It doesn't look like a URI target. Maybe it's an authority string. Try with the default\n+      // scheme from the factory.\n+      try {\n+        targetUri = new URI(nameResolverFactory.getDefaultScheme(), \"\", \"/\" + target, null);\n+      } catch (URISyntaxException e) {\n+        // Should not be possible.\n+        throw new IllegalArgumentException(e);\n+      }\n+      if (targetUri != null) {\n+        NameResolver resolver = nameResolverFactory.newNameResolver(targetUri, nameResolverParams);\n+        if (resolver != null) {\n+          return resolver;\n+        }\n+      }\n+    }\n+    throw new IllegalArgumentException(String.format(\n+        \"cannot find a NameResolver for %s%s\",\n+        target, uriSyntaxErrors.length() > 0 ? \" (\" + uriSyntaxErrors + \")\" : \"\"));\n+  }\n+\n+  /**\n+   * Initiates an orderly shutdown in which preexisting calls continue but new calls are immediately\n+   * cancelled.\n+   */\n+  @Override\n+  public ManagedChannelImpl2 shutdown() {\n+    log.log(Level.FINE, \"[{0}] shutdown() called\", getLogId());\n+    if (!shutdown.compareAndSet(false, true)) {\n+      return this;\n+    }\n+    delayedTransport.shutdown();\n+    channelExecutor.executeLater(new Runnable() {\n+        @Override\n+        public void run() {\n+          maybeTerminateChannel();\n+          cancelIdleTimer();\n+        }\n+      }).drain();\n+    log.log(Level.FINE, \"[{0}] Shutting down\", getLogId());\n+    return this;\n+  }\n+\n+  /**\n+   * Initiates a forceful shutdown in which preexisting and new calls are cancelled. Although\n+   * forceful, the shutdown process is still not instantaneous; {@link #isTerminated()} will likely\n+   * return {@code false} immediately after this method returns.\n+   */\n+  @Override\n+  public ManagedChannelImpl2 shutdownNow() {\n+    log.log(Level.FINE, \"[{0}] shutdownNow() called\", getLogId());\n+    shutdown();\n+    delayedTransport.shutdownNow(SHUTDOWN_NOW_STATUS);\n+    channelExecutor.executeLater(new Runnable() {\n+        @Override\n+        public void run() {\n+          if (shutdownNowed) {\n+            return;\n+          }\n+          shutdownNowed = true;\n+          maybeShutdownNowSubchannels();\n+        }\n+      }).drain();\n+    return this;\n+  }\n+\n+  @Override\n+  public boolean isShutdown() {\n+    return shutdown.get();\n+  }\n+\n+  @Override\n+  public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {\n+    return terminatedLatch.await(timeout, unit);\n+  }\n+\n+  @Override\n+  public boolean isTerminated() {\n+    return terminated;\n+  }\n+\n+  /*\n+   * Creates a new outgoing call on the channel.\n+   */\n+  @Override\n+  public <ReqT, RespT> ClientCall<ReqT, RespT> newCall(MethodDescriptor<ReqT, RespT> method,\n+      CallOptions callOptions) {\n+    return interceptorChannel.newCall(method, callOptions);\n+  }\n+\n+  @Override\n+  public String authority() {\n+    return interceptorChannel.authority();\n+  }\n+\n+  private class RealChannel extends Channel {\n+    @Override\n+    public <ReqT, RespT> ClientCall<ReqT, RespT> newCall(MethodDescriptor<ReqT, RespT> method,\n+        CallOptions callOptions) {\n+      Executor executor = callOptions.getExecutor();\n+      if (executor == null) {\n+        executor = ManagedChannelImpl2.this.executor;\n+      }\n+      StatsTraceContext statsTraceCtx = StatsTraceContext.newClientContext(\n+          method.getFullMethodName(), censusFactory, stopwatchSupplier);\n+      return new ClientCallImpl<ReqT, RespT>(\n+          method,\n+          executor,\n+          callOptions,\n+          statsTraceCtx,\n+          transportProvider,\n+          scheduledExecutor)\n+              .setDecompressorRegistry(decompressorRegistry)\n+              .setCompressorRegistry(compressorRegistry);\n+    }\n+\n+    @Override\n+    public String authority() {\n+      String authority = nameResolver.getServiceAuthority();\n+      return checkNotNull(authority, \"authority\");\n+    }\n+  }\n+\n+  /**\n+   * Terminate the channel if termination conditions are met.\n+   */\n+  // Must be run from channelExecutor\n+  private void maybeTerminateChannel() {\n+    if (terminated) {\n+      return;\n+    }\n+    if (shutdown.get() && subchannels.isEmpty() && oobChannels.isEmpty()) {\n+      log.log(Level.FINE, \"[{0}] Terminated\", getLogId());\n+      terminated = true;\n+      terminatedLatch.countDown();\n+      executorPool.returnObject(executor);\n+      // Needed for delivering rejections to new calls after OobChannel is terminated.\n+      executor = MoreExecutors.directExecutor();\n+      scheduledExecutor = timerServicePool.returnObject(scheduledExecutor);\n+      // Release the transport factory so that it can deallocate any resources.\n+      transportFactory.close();\n+    }\n+  }\n+\n+  private class LbHelperImpl extends LoadBalancer2.Helper {\n+    LoadBalancer2 lb;\n+    final NameResolver nr;\n+\n+    LbHelperImpl(NameResolver nr) {\n+      this.nr = checkNotNull(nr, \"NameResolver\");\n+    }\n+\n+    @Override\n+    public SubchannelImpl createSubchannel(EquivalentAddressGroup addressGroup, Attributes attrs) {\n+      checkNotNull(addressGroup, \"addressGroup\");\n+      checkNotNull(attrs, \"attrs\");\n+      ScheduledExecutorService scheduledExecutorCopy = scheduledExecutor;\n+      checkState(scheduledExecutorCopy != null,\n+          \"scheduledExecutor is already cleared. Looks like you are calling this method after \"\n+          + \"you've already shut down\");\n+      final SubchannelImplImpl subchannel = new SubchannelImplImpl(attrs);\n+      final InternalSubchannel internalSubchannel = new InternalSubchannel(\n+            addressGroup, authority(), userAgent, backoffPolicyProvider, transportFactory,\n+            scheduledExecutorCopy, stopwatchSupplier, channelExecutor,\n+            new InternalSubchannel.Callback() {\n+              // All callbacks are run in channelExecutor\n+              @Override\n+              void onTerminated(InternalSubchannel is) {\n+                subchannels.remove(is);\n+                maybeTerminateChannel();\n+              }\n+\n+              @Override\n+              void onStateChange(InternalSubchannel is, ConnectivityStateInfo newState) {\n+                if ((newState.getState() == TRANSIENT_FAILURE || newState.getState() == IDLE)) {\n+                  nr.refresh();\n+                }\n+                lb.handleSubchannelState(subchannel, newState);\n+              }\n+\n+              @Override\n+              public void onInUse(InternalSubchannel is) {\n+                inUseStateAggregator.updateObjectInUse(is, true);\n+              }\n+\n+              @Override\n+              public void onNotInUse(InternalSubchannel is) {\n+                inUseStateAggregator.updateObjectInUse(is, false);\n+              }\n+            });\n+      subchannel.subchannel = internalSubchannel;\n+      log.log(Level.FINE, \"[{0}] {1} created for {2}\",\n+          new Object[] {getLogId(), internalSubchannel.getLogId(), addressGroup});\n+      runSerialized(new Runnable() {\n+          @Override\n+          public void run() {\n+            if (terminating) {\n+              internalSubchannel.shutdown();","path":"core/src/main/java/io/grpc/internal/ManagedChannelImpl2.java","position":657,"original_position":657,"commit_id":"014ffc2154586217461b40d2198a26c74b929746","original_commit_id":"014ffc2154586217461b40d2198a26c74b929746","user":{"login":"ejona86","id":2811396,"avatar_url":"https://avatars.githubusercontent.com/u/2811396?v=3","gravatar_id":"","url":"https://api.github.com/users/ejona86","html_url":"https://github.com/ejona86","followers_url":"https://api.github.com/users/ejona86/followers","following_url":"https://api.github.com/users/ejona86/following{/other_user}","gists_url":"https://api.github.com/users/ejona86/gists{/gist_id}","starred_url":"https://api.github.com/users/ejona86/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/ejona86/subscriptions","organizations_url":"https://api.github.com/users/ejona86/orgs","repos_url":"https://api.github.com/users/ejona86/repos","events_url":"https://api.github.com/users/ejona86/events{/privacy}","received_events_url":"https://api.github.com/users/ejona86/received_events","type":"User","site_admin":false},"body":"This `runSerialized()` block looks subtle. The `runSerialized()` does not guarantee that the `Runnable` has executed before it returns. So we could return a subchannel here that is non-yet-shutdown even when terminating. But that subchannel won't (can't?) be used because the delayed transport is already terminated and so has no RPCs and is not accepting any more RPCs (or passing-through any more RPCs).\r\n\r\nIt seems a comment would go a long way, if nothing else to prevent future bugs.","created_at":"2016-12-29T19:57:50Z","updated_at":"2016-12-29T20:03:16Z","html_url":"https://github.com/grpc/grpc-java/pull/2530#discussion_r94174176","pull_request_url":"https://api.github.com/repos/grpc/grpc-java/pulls/2530","_links":{"self":{"href":"https://api.github.com/repos/grpc/grpc-java/pulls/comments/94174176"},"html":{"href":"https://github.com/grpc/grpc-java/pull/2530#discussion_r94174176"},"pull_request":{"href":"https://api.github.com/repos/grpc/grpc-java/pulls/2530"}}},"pull_request":{"url":"https://api.github.com/repos/grpc/grpc-java/pulls/2530","id":98423199,"html_url":"https://github.com/grpc/grpc-java/pull/2530","diff_url":"https://github.com/grpc/grpc-java/pull/2530.diff","patch_url":"https://github.com/grpc/grpc-java/pull/2530.patch","issue_url":"https://api.github.com/repos/grpc/grpc-java/issues/2530","number":2530,"state":"open","locked":false,"title":"core: ManagedChannelImpl2.","user":{"login":"zhangkun83","id":9053690,"avatar_url":"https://avatars.githubusercontent.com/u/9053690?v=3","gravatar_id":"","url":"https://api.github.com/users/zhangkun83","html_url":"https://github.com/zhangkun83","followers_url":"https://api.github.com/users/zhangkun83/followers","following_url":"https://api.github.com/users/zhangkun83/following{/other_user}","gists_url":"https://api.github.com/users/zhangkun83/gists{/gist_id}","starred_url":"https://api.github.com/users/zhangkun83/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/zhangkun83/subscriptions","organizations_url":"https://api.github.com/users/zhangkun83/orgs","repos_url":"https://api.github.com/users/zhangkun83/repos","events_url":"https://api.github.com/users/zhangkun83/events{/privacy}","received_events_url":"https://api.github.com/users/zhangkun83/received_events","type":"User","site_admin":false},"body":"1. Adapt to LoadBalancer2 interface.  Channel holds on to a single\r\n   DelayedClientTransport2.\r\n\r\n2. Lock-free: every channel state mutation, including Subchannel\r\n   mutations, calling into LoadBalancer, idleness and shutdown, is made\r\n   from channelExecutor.\r\n\r\n3. Idleness grace period is no longer needed.\r\n\r\nAs usual, the first commit is a pure fork of ManagedChannelImpl and its tests, although the change may be so drastic so that reviewing the whole PR may be easier.\r\n\r\n/cc @lukaszx0 @kkaneda","created_at":"2016-12-17T00:10:05Z","updated_at":"2016-12-29T20:03:16Z","closed_at":null,"merged_at":null,"merge_commit_sha":"30c7b9fb2dca3463f4b97cebe4b5cb85c43e8fcc","assignee":{"login":"ejona86","id":2811396,"avatar_url":"https://avatars.githubusercontent.com/u/2811396?v=3","gravatar_id":"","url":"https://api.github.com/users/ejona86","html_url":"https://github.com/ejona86","followers_url":"https://api.github.com/users/ejona86/followers","following_url":"https://api.github.com/users/ejona86/following{/other_user}","gists_url":"https://api.github.com/users/ejona86/gists{/gist_id}","starred_url":"https://api.github.com/users/ejona86/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/ejona86/subscriptions","organizations_url":"https://api.github.com/users/ejona86/orgs","repos_url":"https://api.github.com/users/ejona86/repos","events_url":"https://api.github.com/users/ejona86/events{/privacy}","received_events_url":"https://api.github.com/users/ejona86/received_events","type":"User","site_admin":false},"assignees":[{"login":"ejona86","id":2811396,"avatar_url":"https://avatars.githubusercontent.com/u/2811396?v=3","gravatar_id":"","url":"https://api.github.com/users/ejona86","html_url":"https://github.com/ejona86","followers_url":"https://api.github.com/users/ejona86/followers","following_url":"https://api.github.com/users/ejona86/following{/other_user}","gists_url":"https://api.github.com/users/ejona86/gists{/gist_id}","starred_url":"https://api.github.com/users/ejona86/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/ejona86/subscriptions","organizations_url":"https://api.github.com/users/ejona86/orgs","repos_url":"https://api.github.com/users/ejona86/repos","events_url":"https://api.github.com/users/ejona86/events{/privacy}","received_events_url":"https://api.github.com/users/ejona86/received_events","type":"User","site_admin":false}],"milestone":null,"commits_url":"https://api.github.com/repos/grpc/grpc-java/pulls/2530/commits","review_comments_url":"https://api.github.com/repos/grpc/grpc-java/pulls/2530/comments","review_comment_url":"https://api.github.com/repos/grpc/grpc-java/pulls/comments{/number}","comments_url":"https://api.github.com/repos/grpc/grpc-java/issues/2530/comments","statuses_url":"https://api.github.com/repos/grpc/grpc-java/statuses/014ffc2154586217461b40d2198a26c74b929746","head":{"label":"zhangkun83:lbv2_channel_review","ref":"lbv2_channel_review","sha":"014ffc2154586217461b40d2198a26c74b929746","user":{"login":"zhangkun83","id":9053690,"avatar_url":"https://avatars.githubusercontent.com/u/9053690?v=3","gravatar_id":"","url":"https://api.github.com/users/zhangkun83","html_url":"https://github.com/zhangkun83","followers_url":"https://api.github.com/users/zhangkun83/followers","following_url":"https://api.github.com/users/zhangkun83/following{/other_user}","gists_url":"https://api.github.com/users/zhangkun83/gists{/gist_id}","starred_url":"https://api.github.com/users/zhangkun83/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/zhangkun83/subscriptions","organizations_url":"https://api.github.com/users/zhangkun83/orgs","repos_url":"https://api.github.com/users/zhangkun83/repos","events_url":"https://api.github.com/users/zhangkun83/events{/privacy}","received_events_url":"https://api.github.com/users/zhangkun83/received_events","type":"User","site_admin":false},"repo":{"id":31399597,"name":"grpc-java","full_name":"zhangkun83/grpc-java","owner":{"login":"zhangkun83","id":9053690,"avatar_url":"https://avatars.githubusercontent.com/u/9053690?v=3","gravatar_id":"","url":"https://api.github.com/users/zhangkun83","html_url":"https://github.com/zhangkun83","followers_url":"https://api.github.com/users/zhangkun83/followers","following_url":"https://api.github.com/users/zhangkun83/following{/other_user}","gists_url":"https://api.github.com/users/zhangkun83/gists{/gist_id}","starred_url":"https://api.github.com/users/zhangkun83/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/zhangkun83/subscriptions","organizations_url":"https://api.github.com/users/zhangkun83/orgs","repos_url":"https://api.github.com/users/zhangkun83/repos","events_url":"https://api.github.com/users/zhangkun83/events{/privacy}","received_events_url":"https://api.github.com/users/zhangkun83/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/zhangkun83/grpc-java","description":"The Java gRPC implementation. HTTP/2 based RPC","fork":true,"url":"https://api.github.com/repos/zhangkun83/grpc-java","forks_url":"https://api.github.com/repos/zhangkun83/grpc-java/forks","keys_url":"https://api.github.com/repos/zhangkun83/grpc-java/keys{/key_id}","collaborators_url":"https://api.github.com/repos/zhangkun83/grpc-java/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/zhangkun83/grpc-java/teams","hooks_url":"https://api.github.com/repos/zhangkun83/grpc-java/hooks","issue_events_url":"https://api.github.com/repos/zhangkun83/grpc-java/issues/events{/number}","events_url":"https://api.github.com/repos/zhangkun83/grpc-java/events","assignees_url":"https://api.github.com/repos/zhangkun83/grpc-java/assignees{/user}","branches_url":"https://api.github.com/repos/zhangkun83/grpc-java/branches{/branch}","tags_url":"https://api.github.com/repos/zhangkun83/grpc-java/tags","blobs_url":"https://api.github.com/repos/zhangkun83/grpc-java/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/zhangkun83/grpc-java/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/zhangkun83/grpc-java/git/refs{/sha}","trees_url":"https://api.github.com/repos/zhangkun83/grpc-java/git/trees{/sha}","statuses_url":"https://api.github.com/repos/zhangkun83/grpc-java/statuses/{sha}","languages_url":"https://api.github.com/repos/zhangkun83/grpc-java/languages","stargazers_url":"https://api.github.com/repos/zhangkun83/grpc-java/stargazers","contributors_url":"https://api.github.com/repos/zhangkun83/grpc-java/contributors","subscribers_url":"https://api.github.com/repos/zhangkun83/grpc-java/subscribers","subscription_url":"https://api.github.com/repos/zhangkun83/grpc-java/subscription","commits_url":"https://api.github.com/repos/zhangkun83/grpc-java/commits{/sha}","git_commits_url":"https://api.github.com/repos/zhangkun83/grpc-java/git/commits{/sha}","comments_url":"https://api.github.com/repos/zhangkun83/grpc-java/comments{/number}","issue_comment_url":"https://api.github.com/repos/zhangkun83/grpc-java/issues/comments{/number}","contents_url":"https://api.github.com/repos/zhangkun83/grpc-java/contents/{+path}","compare_url":"https://api.github.com/repos/zhangkun83/grpc-java/compare/{base}...{head}","merges_url":"https://api.github.com/repos/zhangkun83/grpc-java/merges","archive_url":"https://api.github.com/repos/zhangkun83/grpc-java/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/zhangkun83/grpc-java/downloads","issues_url":"https://api.github.com/repos/zhangkun83/grpc-java/issues{/number}","pulls_url":"https://api.github.com/repos/zhangkun83/grpc-java/pulls{/number}","milestones_url":"https://api.github.com/repos/zhangkun83/grpc-java/milestones{/number}","notifications_url":"https://api.github.com/repos/zhangkun83/grpc-java/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/zhangkun83/grpc-java/labels{/name}","releases_url":"https://api.github.com/repos/zhangkun83/grpc-java/releases{/id}","deployments_url":"https://api.github.com/repos/zhangkun83/grpc-java/deployments","created_at":"2015-02-27T02:25:11Z","updated_at":"2016-01-22T23:33:39Z","pushed_at":"2016-12-29T19:20:56Z","git_url":"git://github.com/zhangkun83/grpc-java.git","ssh_url":"git@github.com:zhangkun83/grpc-java.git","clone_url":"https://github.com/zhangkun83/grpc-java.git","svn_url":"https://github.com/zhangkun83/grpc-java","homepage":"","size":9637,"stargazers_count":0,"watchers_count":0,"language":"Java","has_issues":false,"has_downloads":false,"has_wiki":false,"has_pages":false,"forks_count":1,"mirror_url":null,"open_issues_count":0,"forks":1,"open_issues":0,"watchers":0,"default_branch":"master"}},"base":{"label":"grpc:master","ref":"master","sha":"a6ecece6f5be27e8688b3cb7cc5e13ca20c5c7bd","user":{"login":"grpc","id":7802525,"avatar_url":"https://avatars.githubusercontent.com/u/7802525?v=3","gravatar_id":"","url":"https://api.github.com/users/grpc","html_url":"https://github.com/grpc","followers_url":"https://api.github.com/users/grpc/followers","following_url":"https://api.github.com/users/grpc/following{/other_user}","gists_url":"https://api.github.com/users/grpc/gists{/gist_id}","starred_url":"https://api.github.com/users/grpc/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/grpc/subscriptions","organizations_url":"https://api.github.com/users/grpc/orgs","repos_url":"https://api.github.com/users/grpc/repos","events_url":"https://api.github.com/users/grpc/events{/privacy}","received_events_url":"https://api.github.com/users/grpc/received_events","type":"Organization","site_admin":false},"repo":{"id":27729926,"name":"grpc-java","full_name":"grpc/grpc-java","owner":{"login":"grpc","id":7802525,"avatar_url":"https://avatars.githubusercontent.com/u/7802525?v=3","gravatar_id":"","url":"https://api.github.com/users/grpc","html_url":"https://github.com/grpc","followers_url":"https://api.github.com/users/grpc/followers","following_url":"https://api.github.com/users/grpc/following{/other_user}","gists_url":"https://api.github.com/users/grpc/gists{/gist_id}","starred_url":"https://api.github.com/users/grpc/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/grpc/subscriptions","organizations_url":"https://api.github.com/users/grpc/orgs","repos_url":"https://api.github.com/users/grpc/repos","events_url":"https://api.github.com/users/grpc/events{/privacy}","received_events_url":"https://api.github.com/users/grpc/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/grpc/grpc-java","description":"The Java gRPC implementation. HTTP/2 based RPC","fork":false,"url":"https://api.github.com/repos/grpc/grpc-java","forks_url":"https://api.github.com/repos/grpc/grpc-java/forks","keys_url":"https://api.github.com/repos/grpc/grpc-java/keys{/key_id}","collaborators_url":"https://api.github.com/repos/grpc/grpc-java/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/grpc/grpc-java/teams","hooks_url":"https://api.github.com/repos/grpc/grpc-java/hooks","issue_events_url":"https://api.github.com/repos/grpc/grpc-java/issues/events{/number}","events_url":"https://api.github.com/repos/grpc/grpc-java/events","assignees_url":"https://api.github.com/repos/grpc/grpc-java/assignees{/user}","branches_url":"https://api.github.com/repos/grpc/grpc-java/branches{/branch}","tags_url":"https://api.github.com/repos/grpc/grpc-java/tags","blobs_url":"https://api.github.com/repos/grpc/grpc-java/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/grpc/grpc-java/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/grpc/grpc-java/git/refs{/sha}","trees_url":"https://api.github.com/repos/grpc/grpc-java/git/trees{/sha}","statuses_url":"https://api.github.com/repos/grpc/grpc-java/statuses/{sha}","languages_url":"https://api.github.com/repos/grpc/grpc-java/languages","stargazers_url":"https://api.github.com/repos/grpc/grpc-java/stargazers","contributors_url":"https://api.github.com/repos/grpc/grpc-java/contributors","subscribers_url":"https://api.github.com/repos/grpc/grpc-java/subscribers","subscription_url":"https://api.github.com/repos/grpc/grpc-java/subscription","commits_url":"https://api.github.com/repos/grpc/grpc-java/commits{/sha}","git_commits_url":"https://api.github.com/repos/grpc/grpc-java/git/commits{/sha}","comments_url":"https://api.github.com/repos/grpc/grpc-java/comments{/number}","issue_comment_url":"https://api.github.com/repos/grpc/grpc-java/issues/comments{/number}","contents_url":"https://api.github.com/repos/grpc/grpc-java/contents/{+path}","compare_url":"https://api.github.com/repos/grpc/grpc-java/compare/{base}...{head}","merges_url":"https://api.github.com/repos/grpc/grpc-java/merges","archive_url":"https://api.github.com/repos/grpc/grpc-java/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/grpc/grpc-java/downloads","issues_url":"https://api.github.com/repos/grpc/grpc-java/issues{/number}","pulls_url":"https://api.github.com/repos/grpc/grpc-java/pulls{/number}","milestones_url":"https://api.github.com/repos/grpc/grpc-java/milestones{/number}","notifications_url":"https://api.github.com/repos/grpc/grpc-java/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/grpc/grpc-java/labels{/name}","releases_url":"https://api.github.com/repos/grpc/grpc-java/releases{/id}","deployments_url":"https://api.github.com/repos/grpc/grpc-java/deployments","created_at":"2014-12-08T19:00:00Z","updated_at":"2016-12-29T01:06:48Z","pushed_at":"2016-12-29T19:20:50Z","git_url":"git://github.com/grpc/grpc-java.git","ssh_url":"git@github.com:grpc/grpc-java.git","clone_url":"https://github.com/grpc/grpc-java.git","svn_url":"https://github.com/grpc/grpc-java","homepage":"http://grpc.io","size":11474,"stargazers_count":2308,"watchers_count":2308,"language":"Java","has_issues":true,"has_downloads":false,"has_wiki":false,"has_pages":true,"forks_count":701,"mirror_url":null,"open_issues_count":295,"forks":701,"open_issues":295,"watchers":2308,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/grpc/grpc-java/pulls/2530"},"html":{"href":"https://github.com/grpc/grpc-java/pull/2530"},"issue":{"href":"https://api.github.com/repos/grpc/grpc-java/issues/2530"},"comments":{"href":"https://api.github.com/repos/grpc/grpc-java/issues/2530/comments"},"review_comments":{"href":"https://api.github.com/repos/grpc/grpc-java/pulls/2530/comments"},"review_comment":{"href":"https://api.github.com/repos/grpc/grpc-java/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/grpc/grpc-java/pulls/2530/commits"},"statuses":{"href":"https://api.github.com/repos/grpc/grpc-java/statuses/014ffc2154586217461b40d2198a26c74b929746"}}}}
{"actor":{"display_login":"ejona86"}}