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

grpc / grpc-java / #20358

21 Jul 2026 03:26PM UTC coverage: 89.113% (+0.002%) from 89.111%
#20358

push

github

web-flow
Revert "enable child channel plugins (#12578)" (v1.83.x backport) (#12913)

Backport of #12912 to v1.83.x.
---
This reverts commit 89aef90d5.
reverted for preventing behaviour change across releases, will re raise
after adding in
https://github.com/grpc/proposal/pull/529#discussion_r3547838948

37995 of 42637 relevant lines covered (89.11%)

0.89 hits per line

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

93.24
/../core/src/main/java/io/grpc/internal/ManagedChannelImpl.java
1
/*
2
 * Copyright 2016 The gRPC Authors
3
 *
4
 * Licensed under the Apache License, Version 2.0 (the "License");
5
 * you may not use this file except in compliance with the License.
6
 * You may obtain a copy of the License at
7
 *
8
 *     http://www.apache.org/licenses/LICENSE-2.0
9
 *
10
 * Unless required by applicable law or agreed to in writing, software
11
 * distributed under the License is distributed on an "AS IS" BASIS,
12
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13
 * See the License for the specific language governing permissions and
14
 * limitations under the License.
15
 */
16

17
package io.grpc.internal;
18

19
import static com.google.common.base.Preconditions.checkArgument;
20
import static com.google.common.base.Preconditions.checkNotNull;
21
import static com.google.common.base.Preconditions.checkState;
22
import static io.grpc.ClientStreamTracer.NAME_RESOLUTION_DELAYED;
23
import static io.grpc.ConnectivityState.CONNECTING;
24
import static io.grpc.ConnectivityState.IDLE;
25
import static io.grpc.ConnectivityState.SHUTDOWN;
26
import static io.grpc.ConnectivityState.TRANSIENT_FAILURE;
27
import static io.grpc.EquivalentAddressGroup.ATTR_AUTHORITY_OVERRIDE;
28

29
import com.google.common.annotations.VisibleForTesting;
30
import com.google.common.base.MoreObjects;
31
import com.google.common.base.Stopwatch;
32
import com.google.common.base.Supplier;
33
import com.google.common.util.concurrent.ListenableFuture;
34
import com.google.common.util.concurrent.SettableFuture;
35
import com.google.errorprone.annotations.concurrent.GuardedBy;
36
import io.grpc.Attributes;
37
import io.grpc.CallCredentials;
38
import io.grpc.CallOptions;
39
import io.grpc.Channel;
40
import io.grpc.ChannelCredentials;
41
import io.grpc.ChannelLogger;
42
import io.grpc.ChannelLogger.ChannelLogLevel;
43
import io.grpc.ClientCall;
44
import io.grpc.ClientInterceptor;
45
import io.grpc.ClientInterceptors;
46
import io.grpc.ClientStreamTracer;
47
import io.grpc.ClientTransportFilter;
48
import io.grpc.CompressorRegistry;
49
import io.grpc.ConnectivityState;
50
import io.grpc.ConnectivityStateInfo;
51
import io.grpc.Context;
52
import io.grpc.Deadline;
53
import io.grpc.DecompressorRegistry;
54
import io.grpc.EquivalentAddressGroup;
55
import io.grpc.ForwardingChannelBuilder2;
56
import io.grpc.ForwardingClientCall;
57
import io.grpc.Grpc;
58
import io.grpc.InternalChannelz;
59
import io.grpc.InternalChannelz.ChannelStats;
60
import io.grpc.InternalChannelz.ChannelTrace;
61
import io.grpc.InternalConfigSelector;
62
import io.grpc.InternalInstrumented;
63
import io.grpc.InternalLogId;
64
import io.grpc.InternalWithLogId;
65
import io.grpc.LoadBalancer;
66
import io.grpc.LoadBalancer.CreateSubchannelArgs;
67
import io.grpc.LoadBalancer.PickResult;
68
import io.grpc.LoadBalancer.PickSubchannelArgs;
69
import io.grpc.LoadBalancer.ResolvedAddresses;
70
import io.grpc.LoadBalancer.SubchannelPicker;
71
import io.grpc.LoadBalancer.SubchannelStateListener;
72
import io.grpc.LoadBalancerProvider;
73
import io.grpc.ManagedChannel;
74
import io.grpc.ManagedChannelBuilder;
75
import io.grpc.Metadata;
76
import io.grpc.MethodDescriptor;
77
import io.grpc.MetricInstrumentRegistry;
78
import io.grpc.MetricRecorder;
79
import io.grpc.NameResolver;
80
import io.grpc.NameResolver.ConfigOrError;
81
import io.grpc.NameResolver.ResolutionResult;
82
import io.grpc.NameResolverProvider;
83
import io.grpc.NameResolverRegistry;
84
import io.grpc.ProxyDetector;
85
import io.grpc.Status;
86
import io.grpc.StatusOr;
87
import io.grpc.SynchronizationContext;
88
import io.grpc.SynchronizationContext.ScheduledHandle;
89
import io.grpc.internal.ClientCallImpl.ClientStreamProvider;
90
import io.grpc.internal.ClientTransportFactory.SwapChannelCredentialsResult;
91
import io.grpc.internal.ManagedChannelImplBuilder.ClientTransportFactoryBuilder;
92
import io.grpc.internal.ManagedChannelImplBuilder.FixedPortProvider;
93
import io.grpc.internal.ManagedChannelServiceConfig.MethodInfo;
94
import io.grpc.internal.ManagedChannelServiceConfig.ServiceConfigConvertedSelector;
95
import io.grpc.internal.RetriableStream.ChannelBufferMeter;
96
import io.grpc.internal.RetriableStream.Throttle;
97
import java.net.URI;
98
import java.net.URISyntaxException;
99
import java.util.ArrayList;
100
import java.util.Collection;
101
import java.util.Collections;
102
import java.util.HashSet;
103
import java.util.LinkedHashSet;
104
import java.util.List;
105
import java.util.Map;
106
import java.util.Set;
107
import java.util.concurrent.Callable;
108
import java.util.concurrent.CountDownLatch;
109
import java.util.concurrent.ExecutionException;
110
import java.util.concurrent.Executor;
111
import java.util.concurrent.Future;
112
import java.util.concurrent.ScheduledExecutorService;
113
import java.util.concurrent.ScheduledFuture;
114
import java.util.concurrent.TimeUnit;
115
import java.util.concurrent.TimeoutException;
116
import java.util.concurrent.atomic.AtomicBoolean;
117
import java.util.concurrent.atomic.AtomicReference;
118
import java.util.logging.Level;
119
import java.util.logging.Logger;
120
import javax.annotation.Nullable;
121
import javax.annotation.concurrent.ThreadSafe;
122

123
/** A communication channel for making outgoing RPCs. */
124
@ThreadSafe
125
final class ManagedChannelImpl extends ManagedChannel implements
126
    InternalInstrumented<ChannelStats> {
127
  @VisibleForTesting
128
  static final Logger logger = Logger.getLogger(ManagedChannelImpl.class.getName());
1✔
129

130
  static final long IDLE_TIMEOUT_MILLIS_DISABLE = -1;
131

132
  static final long SUBCHANNEL_SHUTDOWN_DELAY_SECONDS = 5;
133

134
  @VisibleForTesting
135
  static final Status SHUTDOWN_NOW_STATUS =
1✔
136
      Status.UNAVAILABLE.withDescription("Channel shutdownNow invoked");
1✔
137

138
  @VisibleForTesting
139
  static final Status SHUTDOWN_STATUS =
1✔
140
      Status.UNAVAILABLE.withDescription("Channel shutdown invoked");
1✔
141

142
  @VisibleForTesting
143
  static final Status SUBCHANNEL_SHUTDOWN_STATUS =
1✔
144
      Status.UNAVAILABLE.withDescription("Subchannel shutdown invoked");
1✔
145

146
  private static final ManagedChannelServiceConfig EMPTY_SERVICE_CONFIG =
147
      ManagedChannelServiceConfig.empty();
1✔
148
  private static final InternalConfigSelector INITIAL_PENDING_SELECTOR =
1✔
149
      new InternalConfigSelector() {
1✔
150
        @Override
151
        public Result selectConfig(PickSubchannelArgs args) {
152
          throw new IllegalStateException("Resolution is pending");
×
153
        }
154
      };
155
  private static final LoadBalancer.PickDetailsConsumer NOOP_PICK_DETAILS_CONSUMER =
1✔
156
      new LoadBalancer.PickDetailsConsumer() {};
1✔
157

158
  private final InternalLogId logId;
159
  private final String target;
160
  @Nullable
161
  private final String authorityOverride;
162
  private final NameResolverRegistry nameResolverRegistry;
163
  private final UriWrapper targetUri;
164
  private final NameResolverProvider nameResolverProvider;
165
  private final NameResolver.Args nameResolverArgs;
166
  private final LoadBalancerProvider loadBalancerFactory;
167
  private final ClientTransportFactory originalTransportFactory;
168
  @Nullable
169
  private final ChannelCredentials originalChannelCreds;
170
  private final ClientTransportFactory transportFactory;
171
  private final RestrictedScheduledExecutor scheduledExecutor;
172
  private final Executor executor;
173
  private final ObjectPool<? extends Executor> executorPool;
174
  private final ExecutorHolder balancerRpcExecutorHolder;
175
  private final ExecutorHolder offloadExecutorHolder;
176
  private final TimeProvider timeProvider;
177
  private final int maxTraceEvents;
178

179
  @VisibleForTesting
1✔
180
  final SynchronizationContext syncContext = new SynchronizationContext(
181
      new Thread.UncaughtExceptionHandler() {
1✔
182
        @Override
183
        public void uncaughtException(Thread t, Throwable e) {
184
          logger.log(
1✔
185
              Level.SEVERE,
186
              "[" + getLogId() + "] Uncaught exception in the SynchronizationContext. Panic!",
1✔
187
              e);
188
          try {
189
            panic(e);
1✔
190
          } catch (Throwable anotherT) {
×
191
            logger.log(
×
192
                Level.SEVERE, "[" + getLogId() + "] Uncaught exception while panicking", anotherT);
×
193
          }
1✔
194
        }
1✔
195
      });
196

197
  private boolean fullStreamDecompression;
198

199
  private final DecompressorRegistry decompressorRegistry;
200
  private final CompressorRegistry compressorRegistry;
201

202
  private final Supplier<Stopwatch> stopwatchSupplier;
203
  /** The timeout before entering idle mode. */
204
  private final long idleTimeoutMillis;
205

206
  private final ConnectivityStateManager channelStateManager = new ConnectivityStateManager();
1✔
207
  private final BackoffPolicy.Provider backoffPolicyProvider;
208

209
  /**
210
   * We delegate to this channel, so that we can have interceptors as necessary. If there aren't
211
   * any interceptors and the {@link io.grpc.BinaryLog} is {@code null} then this will just be a
212
   * {@link RealChannel}.
213
   */
214
  private final Channel interceptorChannel;
215

216
  private final List<ClientTransportFilter> transportFilters;
217
  @Nullable private final String userAgent;
218

219
  // Only null after channel is terminated. Must be assigned from the syncContext.
220
  private NameResolver nameResolver;
221

222
  // Must be accessed from the syncContext.
223
  private boolean nameResolverStarted;
224

225
  // null when channel is in idle mode.  Must be assigned from syncContext.
226
  @Nullable
227
  private LbHelperImpl lbHelper;
228

229
  // Must be accessed from the syncContext
230
  private boolean panicMode;
231

232
  // Must be mutated from syncContext
233
  // If any monitoring hook to be added later needs to get a snapshot of this Set, we could
234
  // switch to a ConcurrentHashMap.
235
  private final Set<InternalSubchannel> subchannels = new HashSet<>(16, .75f);
1✔
236

237
  // Must be accessed from syncContext
238
  @Nullable
239
  private Collection<RealChannel.PendingCall<?, ?>> pendingCalls;
240
  private final Object pendingCallsInUseObject = new Object();
1✔
241

242
  // reprocess() must be run from syncContext
243
  private final DelayedClientTransport delayedTransport;
244
  private final UncommittedRetriableStreamsRegistry uncommittedRetriableStreamsRegistry
1✔
245
      = new UncommittedRetriableStreamsRegistry();
246

247
  // Shutdown states.
248
  //
249
  // Channel's shutdown process:
250
  // 1. shutdown(): stop accepting new calls from applications
251
  //   1a shutdown <- true
252
  //   1b delayedTransport.shutdown()
253
  // 2. delayedTransport terminated: stop stream-creation functionality
254
  //   2a terminating <- true
255
  //   2b loadBalancer.shutdown()
256
  //     * LoadBalancer will shutdown subchannels and OOB channels
257
  //   2c loadBalancer <- null
258
  //   2d nameResolver.shutdown()
259
  //   2e nameResolver <- null
260
  // 3. All subchannels and OOB channels terminated: Channel considered terminated
261

262
  private final AtomicBoolean shutdown = new AtomicBoolean(false);
1✔
263
  // Must only be mutated and read from syncContext
264
  private boolean shutdownNowed;
265
  // Must only be mutated from syncContext
266
  private boolean terminating;
267
  // Must be mutated from syncContext
268
  private volatile boolean terminated;
269
  private final CountDownLatch terminatedLatch = new CountDownLatch(1);
1✔
270

271
  private final CallTracer.Factory callTracerFactory;
272
  private final CallTracer channelCallTracer;
273
  private final ChannelTracer channelTracer;
274
  private final ChannelLogger channelLogger;
275
  private final InternalChannelz channelz;
276
  private final RealChannel realChannel;
277
  // Must be mutated and read from syncContext
278
  // a flag for doing channel tracing when flipped
279
  private ResolutionState lastResolutionState = ResolutionState.NO_RESOLUTION;
1✔
280
  // Must be mutated and read from constructor or syncContext
281
  // used for channel tracing when value changed
282
  private ManagedChannelServiceConfig lastServiceConfig = EMPTY_SERVICE_CONFIG;
1✔
283

284
  @Nullable
285
  private final ManagedChannelServiceConfig defaultServiceConfig;
286
  // Must be mutated and read from constructor or syncContext
287
  private boolean serviceConfigUpdated = false;
1✔
288
  private final boolean lookUpServiceConfig;
289

290
  // One instance per channel.
291
  private final ChannelBufferMeter channelBufferUsed = new ChannelBufferMeter();
1✔
292

293
  private final long perRpcBufferLimit;
294
  private final long channelBufferLimit;
295

296
  // Temporary false flag that can skip the retry code path.
297
  private final boolean retryEnabled;
298

299
  private final Deadline.Ticker ticker = Deadline.getSystemTicker();
1✔
300

301
  // Called from syncContext
302
  private final ManagedClientTransport.Listener delayedTransportListener =
1✔
303
      new DelayedTransportListener();
304

305
  // Must be called from syncContext
306
  private void maybeShutdownNowSubchannels() {
307
    if (shutdownNowed) {
1✔
308
      for (InternalSubchannel subchannel : subchannels) {
1✔
309
        subchannel.shutdownNow(SHUTDOWN_NOW_STATUS);
1✔
310
      }
1✔
311
    }
312
  }
1✔
313

314
  // Must be accessed from syncContext
315
  @VisibleForTesting
1✔
316
  final InUseStateAggregator<Object> inUseStateAggregator = new IdleModeStateAggregator();
317

318
  @Override
319
  public ListenableFuture<ChannelStats> getStats() {
320
    final SettableFuture<ChannelStats> ret = SettableFuture.create();
1✔
321
    final class StatsFetcher implements Runnable {
1✔
322
      @Override
323
      public void run() {
324
        ChannelStats.Builder builder = new InternalChannelz.ChannelStats.Builder();
1✔
325
        channelCallTracer.updateBuilder(builder);
1✔
326
        channelTracer.updateBuilder(builder);
1✔
327
        builder.setTarget(target).setState(channelStateManager.getState());
1✔
328
        List<InternalWithLogId> children = new ArrayList<>();
1✔
329
        children.addAll(subchannels);
1✔
330
        builder.setSubchannels(children);
1✔
331
        ret.set(builder.build());
1✔
332
      }
1✔
333
    }
334

335
    // subchannels and oobchannels can only be accessed from syncContext
336
    syncContext.execute(new StatsFetcher());
1✔
337
    return ret;
1✔
338
  }
339

340
  @Override
341
  public InternalLogId getLogId() {
342
    return logId;
1✔
343
  }
344

345
  // Run from syncContext
346
  private class IdleModeTimer implements Runnable {
1✔
347

348
    @Override
349
    public void run() {
350
      // Workaround timer scheduled while in idle mode. This can happen from handleNotInUse() after
351
      // an explicit enterIdleMode() by the user. Protecting here as other locations are a bit too
352
      // subtle to change rapidly to resolve the channel panic. See #8714
353
      if (lbHelper == null) {
1✔
354
        return;
×
355
      }
356
      enterIdleMode();
1✔
357
    }
1✔
358
  }
359

360
  // Must be called from syncContext
361
  private void shutdownNameResolverAndLoadBalancer(boolean channelIsActive) {
362
    syncContext.throwIfNotInThisSynchronizationContext();
1✔
363
    if (channelIsActive) {
1✔
364
      checkState(nameResolverStarted, "nameResolver is not started");
1✔
365
      checkState(lbHelper != null, "lbHelper is null");
1✔
366
    }
367
    if (nameResolver != null) {
1✔
368
      nameResolver.shutdown();
1✔
369
      nameResolverStarted = false;
1✔
370
      if (channelIsActive) {
1✔
371
        nameResolver = getNameResolver(
1✔
372
            targetUri, authorityOverride, nameResolverProvider, nameResolverArgs);
373
      } else {
374
        nameResolver = null;
1✔
375
      }
376
    }
377
    if (lbHelper != null) {
1✔
378
      lbHelper.lb.shutdown();
1✔
379
      lbHelper = null;
1✔
380
    }
381
  }
1✔
382

383
  /**
384
   * Make the channel exit idle mode, if it's in it.
385
   *
386
   * <p>Must be called from syncContext
387
   */
388
  @VisibleForTesting
389
  void exitIdleMode() {
390
    syncContext.throwIfNotInThisSynchronizationContext();
1✔
391
    if (shutdown.get() || panicMode) {
1✔
392
      return;
1✔
393
    }
394
    if (inUseStateAggregator.isInUse()) {
1✔
395
      // Cancel the timer now, so that a racing due timer will not put Channel on idleness
396
      // when the caller of exitIdleMode() is about to use the returned loadBalancer.
397
      cancelIdleTimer(false);
1✔
398
    } else {
399
      // exitIdleMode() may be called outside of inUseStateAggregator.handleNotInUse() while
400
      // isInUse() == false, in which case we still need to schedule the timer.
401
      rescheduleIdleTimer();
1✔
402
    }
403
    if (lbHelper != null) {
1✔
404
      return;
1✔
405
    }
406
    channelLogger.log(ChannelLogLevel.INFO, "Exiting idle mode");
1✔
407
    LbHelperImpl lbHelper = new LbHelperImpl();
1✔
408
    lbHelper.lb = loadBalancerFactory.newLoadBalancer(lbHelper);
1✔
409
    // Delay setting lbHelper until fully initialized, since loadBalancerFactory is user code and
410
    // may throw. We don't want to confuse our state, even if we enter panic mode.
411
    this.lbHelper = lbHelper;
1✔
412

413
    channelStateManager.gotoState(CONNECTING);
1✔
414
    NameResolverListener listener = new NameResolverListener(lbHelper, nameResolver);
1✔
415
    nameResolver.start(listener);
1✔
416
    nameResolverStarted = true;
1✔
417
  }
1✔
418

419
  // Must be run from syncContext
420
  private void enterIdleMode() {
421
    // nameResolver and loadBalancer are guaranteed to be non-null.  If any of them were null,
422
    // either the idleModeTimer ran twice without exiting the idle mode, or the task in shutdown()
423
    // did not cancel idleModeTimer, or enterIdle() ran while shutdown or in idle, all of
424
    // which are bugs.
425
    shutdownNameResolverAndLoadBalancer(true);
1✔
426
    delayedTransport.reprocess(null);
1✔
427
    channelLogger.log(ChannelLogLevel.INFO, "Entering IDLE state");
1✔
428
    channelStateManager.gotoState(IDLE);
1✔
429
    // If the inUseStateAggregator still considers pending calls to be queued up or the delayed
430
    // transport to be holding some we need to exit idle mode to give these calls a chance to
431
    // be processed.
432
    if (inUseStateAggregator.anyObjectInUse(pendingCallsInUseObject, delayedTransport)) {
1✔
433
      exitIdleMode();
1✔
434
    }
435
  }
1✔
436

437
  // Must be run from syncContext
438
  private void cancelIdleTimer(boolean permanent) {
439
    idleTimer.cancel(permanent);
1✔
440
  }
1✔
441

442
  // Always run from syncContext
443
  private void rescheduleIdleTimer() {
444
    if (idleTimeoutMillis == IDLE_TIMEOUT_MILLIS_DISABLE) {
1✔
445
      return;
1✔
446
    }
447
    idleTimer.reschedule(idleTimeoutMillis, TimeUnit.MILLISECONDS);
1✔
448
  }
1✔
449

450
  /**
451
   * Force name resolution refresh to happen immediately. Must be run
452
   * from syncContext.
453
   */
454
  private void refreshNameResolution() {
455
    syncContext.throwIfNotInThisSynchronizationContext();
1✔
456
    if (nameResolverStarted) {
1✔
457
      nameResolver.refresh();
1✔
458
    }
459
  }
1✔
460

461
  private final class ChannelStreamProvider implements ClientStreamProvider {
1✔
462
    volatile Throttle throttle;
463

464
    @Override
465
    public ClientStream newStream(
466
        final MethodDescriptor<?, ?> method,
467
        final CallOptions callOptions,
468
        final Metadata headers,
469
        final Context context) {
470
      // There is no need to reschedule the idle timer here. If the channel isn't shut down, either
471
      // the delayed transport or a real transport will go in-use and cancel the idle timer.
472
      if (!retryEnabled) {
1✔
473
        ClientStreamTracer[] tracers = GrpcUtil.getClientStreamTracers(
1✔
474
            callOptions, headers, 0, /* isTransparentRetry= */ false,
475
            /* isHedging= */false);
476
        Context origContext = context.attach();
1✔
477
        try {
478
          return delayedTransport.newStream(method, headers, callOptions, tracers);
1✔
479
        } finally {
480
          context.detach(origContext);
1✔
481
        }
482
      } else {
483
        MethodInfo methodInfo = callOptions.getOption(MethodInfo.KEY);
1✔
484
        final RetryPolicy retryPolicy = methodInfo == null ? null : methodInfo.retryPolicy;
1✔
485
        final HedgingPolicy hedgingPolicy = methodInfo == null ? null : methodInfo.hedgingPolicy;
1✔
486
        final class RetryStream<ReqT> extends RetriableStream<ReqT> {
487
          @SuppressWarnings("unchecked")
488
          RetryStream() {
1✔
489
            super(
1✔
490
                (MethodDescriptor<ReqT, ?>) method,
491
                headers,
492
                channelBufferUsed,
1✔
493
                perRpcBufferLimit,
1✔
494
                channelBufferLimit,
1✔
495
                getCallExecutor(callOptions),
1✔
496
                transportFactory.getScheduledExecutorService(),
1✔
497
                retryPolicy,
498
                hedgingPolicy,
499
                throttle);
500
          }
1✔
501

502
          @Override
503
          Status prestart() {
504
            return uncommittedRetriableStreamsRegistry.add(this);
1✔
505
          }
506

507
          @Override
508
          void postCommit() {
509
            uncommittedRetriableStreamsRegistry.remove(this);
1✔
510
          }
1✔
511

512
          @Override
513
          ClientStream newSubstream(
514
              Metadata newHeaders, ClientStreamTracer.Factory factory, int previousAttempts,
515
              boolean isTransparentRetry, boolean isHedgedStream) {
516
            CallOptions newOptions = callOptions.withStreamTracerFactory(factory);
1✔
517
            ClientStreamTracer[] tracers = GrpcUtil.getClientStreamTracers(
1✔
518
                newOptions, newHeaders, previousAttempts, isTransparentRetry, isHedgedStream);
519
            Context origContext = context.attach();
1✔
520
            try {
521
              return delayedTransport.newStream(method, newHeaders, newOptions, tracers);
1✔
522
            } finally {
523
              context.detach(origContext);
1✔
524
            }
525
          }
526
        }
527

528
        return new RetryStream<>();
1✔
529
      }
530
    }
531
  }
532

533
  private final ChannelStreamProvider transportProvider = new ChannelStreamProvider();
1✔
534

535
  private final Rescheduler idleTimer;
536
  private final MetricRecorder metricRecorder;
537

538
  ManagedChannelImpl(
539
      ManagedChannelImplBuilder builder,
540
      ClientTransportFactory clientTransportFactory,
541
      UriWrapper targetUri,
542
      NameResolverProvider nameResolverProvider,
543
      BackoffPolicy.Provider backoffPolicyProvider,
544
      ObjectPool<? extends Executor> balancerRpcExecutorPool,
545
      Supplier<Stopwatch> stopwatchSupplier,
546
      List<ClientInterceptor> interceptors,
547
      final TimeProvider timeProvider) {
1✔
548
    this.target = checkNotNull(builder.target, "target");
1✔
549
    this.logId = InternalLogId.allocate("Channel", target);
1✔
550
    this.timeProvider = checkNotNull(timeProvider, "timeProvider");
1✔
551
    this.executorPool = checkNotNull(builder.executorPool, "executorPool");
1✔
552
    this.executor = checkNotNull(executorPool.getObject(), "executor");
1✔
553
    this.originalChannelCreds = builder.channelCredentials;
1✔
554
    this.originalTransportFactory = clientTransportFactory;
1✔
555
    this.offloadExecutorHolder =
1✔
556
        new ExecutorHolder(checkNotNull(builder.offloadExecutorPool, "offloadExecutorPool"));
1✔
557
    this.transportFactory = new CallCredentialsApplyingTransportFactory(
1✔
558
        clientTransportFactory, builder.callCredentials, this.offloadExecutorHolder);
559
    this.scheduledExecutor =
1✔
560
        new RestrictedScheduledExecutor(transportFactory.getScheduledExecutorService());
1✔
561
    maxTraceEvents = builder.maxTraceEvents;
1✔
562
    channelTracer = new ChannelTracer(
1✔
563
        logId, builder.maxTraceEvents, timeProvider.currentTimeNanos(),
1✔
564
        "Channel for '" + target + "'");
565
    channelLogger = new ChannelLoggerImpl(channelTracer, timeProvider);
1✔
566
    ProxyDetector proxyDetector =
567
        builder.proxyDetector != null ? builder.proxyDetector : GrpcUtil.DEFAULT_PROXY_DETECTOR;
1✔
568
    this.retryEnabled = builder.retryEnabled;
1✔
569
    this.loadBalancerFactory = new AutoConfiguredLoadBalancerFactory(builder.defaultLbPolicy);
1✔
570
    this.nameResolverRegistry = builder.nameResolverRegistry;
1✔
571
    this.targetUri = checkNotNull(targetUri, "targetUri");
1✔
572
    this.nameResolverProvider = checkNotNull(nameResolverProvider, "nameResolverProvider");
1✔
573
    ScParser serviceConfigParser =
1✔
574
        new ScParser(
575
            retryEnabled,
576
            builder.maxRetryAttempts,
577
            builder.maxHedgedAttempts,
578
            loadBalancerFactory);
579
    this.authorityOverride = builder.authorityOverride;
1✔
580
    this.metricRecorder = new MetricRecorderImpl(builder.metricSinks,
1✔
581
        MetricInstrumentRegistry.getDefaultRegistry());
1✔
582
    NameResolver.Args.Builder nameResolverArgsBuilder = NameResolver.Args.newBuilder()
1✔
583
            .setDefaultPort(builder.getDefaultPort())
1✔
584
            .setProxyDetector(proxyDetector)
1✔
585
            .setSynchronizationContext(syncContext)
1✔
586
            .setScheduledExecutorService(scheduledExecutor)
1✔
587
            .setServiceConfigParser(serviceConfigParser)
1✔
588
            .setChannelLogger(channelLogger)
1✔
589
            .setOffloadExecutor(this.offloadExecutorHolder)
1✔
590
            .setOverrideAuthority(this.authorityOverride)
1✔
591
            .setMetricRecorder(this.metricRecorder)
1✔
592
            .setNameResolverRegistry(builder.nameResolverRegistry);
1✔
593
    builder.copyAllNameResolverCustomArgsTo(nameResolverArgsBuilder);
1✔
594
    this.nameResolverArgs = nameResolverArgsBuilder.build();
1✔
595
    this.nameResolver = getNameResolver(
1✔
596
        targetUri, authorityOverride, nameResolverProvider, nameResolverArgs);
597
    this.balancerRpcExecutorHolder = new ExecutorHolder(
1✔
598
        checkNotNull(balancerRpcExecutorPool, "balancerRpcExecutorPool"));
1✔
599
    this.delayedTransport = new DelayedClientTransport(this.executor, this.syncContext);
1✔
600
    this.delayedTransport.start(delayedTransportListener);
1✔
601
    this.backoffPolicyProvider = backoffPolicyProvider;
1✔
602

603
    if (builder.defaultServiceConfig != null) {
1✔
604
      ConfigOrError parsedDefaultServiceConfig =
1✔
605
          serviceConfigParser.parseServiceConfig(builder.defaultServiceConfig);
1✔
606
      checkState(
1✔
607
          parsedDefaultServiceConfig.getError() == null,
1✔
608
          "Default config is invalid: %s",
609
          parsedDefaultServiceConfig.getError());
1✔
610
      this.defaultServiceConfig =
1✔
611
          (ManagedChannelServiceConfig) parsedDefaultServiceConfig.getConfig();
1✔
612
      this.transportProvider.throttle = this.defaultServiceConfig.getRetryThrottling();
1✔
613
    } else {
1✔
614
      this.defaultServiceConfig = null;
1✔
615
    }
616
    this.lookUpServiceConfig = builder.lookUpServiceConfig;
1✔
617
    realChannel = new RealChannel(nameResolver.getServiceAuthority());
1✔
618
    Channel channel = realChannel;
1✔
619
    if (builder.binlog != null) {
1✔
620
      channel = builder.binlog.wrapChannel(channel);
1✔
621
    }
622
    this.interceptorChannel = ClientInterceptors.intercept(channel, interceptors);
1✔
623
    this.transportFilters = new ArrayList<>(builder.transportFilters);
1✔
624
    this.stopwatchSupplier = checkNotNull(stopwatchSupplier, "stopwatchSupplier");
1✔
625
    if (builder.idleTimeoutMillis == IDLE_TIMEOUT_MILLIS_DISABLE) {
1✔
626
      this.idleTimeoutMillis = builder.idleTimeoutMillis;
1✔
627
    } else {
628
      checkArgument(
1✔
629
          builder.idleTimeoutMillis
630
              >= ManagedChannelImplBuilder.IDLE_MODE_MIN_TIMEOUT_MILLIS,
631
          "invalid idleTimeoutMillis %s", builder.idleTimeoutMillis);
632
      this.idleTimeoutMillis = builder.idleTimeoutMillis;
1✔
633
    }
634

635
    idleTimer = new Rescheduler(
1✔
636
        new IdleModeTimer(),
637
        syncContext,
638
        transportFactory.getScheduledExecutorService(),
1✔
639
        stopwatchSupplier.get());
1✔
640
    this.fullStreamDecompression = builder.fullStreamDecompression;
1✔
641
    this.decompressorRegistry = checkNotNull(builder.decompressorRegistry, "decompressorRegistry");
1✔
642
    this.compressorRegistry = checkNotNull(builder.compressorRegistry, "compressorRegistry");
1✔
643
    this.userAgent = builder.userAgent;
1✔
644

645
    this.channelBufferLimit = builder.retryBufferSize;
1✔
646
    this.perRpcBufferLimit = builder.perRpcBufferLimit;
1✔
647
    final class ChannelCallTracerFactory implements CallTracer.Factory {
1✔
648
      @Override
649
      public CallTracer create() {
650
        return new CallTracer(timeProvider);
1✔
651
      }
652
    }
653

654
    this.callTracerFactory = new ChannelCallTracerFactory();
1✔
655
    channelCallTracer = callTracerFactory.create();
1✔
656
    this.channelz = checkNotNull(builder.channelz);
1✔
657
    channelz.addRootChannel(this);
1✔
658

659
    if (!lookUpServiceConfig) {
1✔
660
      if (defaultServiceConfig != null) {
1✔
661
        channelLogger.log(
1✔
662
            ChannelLogLevel.INFO, "Service config look-up disabled, using default service config");
663
      }
664
      serviceConfigUpdated = true;
1✔
665
    }
666
  }
1✔
667

668
  @VisibleForTesting
669
  static NameResolver getNameResolver(
670
      UriWrapper targetUri, @Nullable final String overrideAuthority,
671
      NameResolverProvider provider, NameResolver.Args nameResolverArgs) {
672
    NameResolver resolver = targetUri.newNameResolver(provider, nameResolverArgs);
1✔
673
    if (resolver == null) {
1✔
674
      throw new IllegalArgumentException("cannot create a NameResolver for " + targetUri);
1✔
675
    }
676

677
    // We wrap the name resolver in a RetryingNameResolver to give it the ability to retry failures.
678
    // TODO: After a transition period, all NameResolver implementations that need retry should use
679
    //       RetryingNameResolver directly and this step can be removed.
680
    NameResolver usedNameResolver = RetryingNameResolver.wrap(resolver, nameResolverArgs);
1✔
681

682
    if (overrideAuthority == null) {
1✔
683
      return usedNameResolver;
1✔
684
    }
685

686
    return new ForwardingNameResolver(usedNameResolver) {
1✔
687
      @Override
688
      public String getServiceAuthority() {
689
        return overrideAuthority;
1✔
690
      }
691
    };
692
  }
693

694
  @VisibleForTesting
695
  InternalConfigSelector getConfigSelector() {
696
    return realChannel.configSelector.get();
1✔
697
  }
698
  
699
  @VisibleForTesting
700
  boolean hasThrottle() {
701
    return this.transportProvider.throttle != null;
1✔
702
  }
703

704
  /**
705
   * Initiates an orderly shutdown in which preexisting calls continue but new calls are immediately
706
   * cancelled.
707
   */
708
  @Override
709
  public ManagedChannelImpl shutdown() {
710
    channelLogger.log(ChannelLogLevel.DEBUG, "shutdown() called");
1✔
711
    if (!shutdown.compareAndSet(false, true)) {
1✔
712
      return this;
1✔
713
    }
714
    final class Shutdown implements Runnable {
1✔
715
      @Override
716
      public void run() {
717
        channelLogger.log(ChannelLogLevel.INFO, "Entering SHUTDOWN state");
1✔
718
        channelStateManager.gotoState(SHUTDOWN);
1✔
719
      }
1✔
720
    }
721

722
    syncContext.execute(new Shutdown());
1✔
723
    realChannel.shutdown();
1✔
724
    final class CancelIdleTimer implements Runnable {
1✔
725
      @Override
726
      public void run() {
727
        cancelIdleTimer(/* permanent= */ true);
1✔
728
      }
1✔
729
    }
730

731
    syncContext.execute(new CancelIdleTimer());
1✔
732
    return this;
1✔
733
  }
734

735
  /**
736
   * Initiates a forceful shutdown in which preexisting and new calls are cancelled. Although
737
   * forceful, the shutdown process is still not instantaneous; {@link #isTerminated()} will likely
738
   * return {@code false} immediately after this method returns.
739
   */
740
  @Override
741
  public ManagedChannelImpl shutdownNow() {
742
    channelLogger.log(ChannelLogLevel.DEBUG, "shutdownNow() called");
1✔
743
    shutdown();
1✔
744
    realChannel.shutdownNow();
1✔
745
    final class ShutdownNow implements Runnable {
1✔
746
      @Override
747
      public void run() {
748
        if (shutdownNowed) {
1✔
749
          return;
1✔
750
        }
751
        shutdownNowed = true;
1✔
752
        maybeShutdownNowSubchannels();
1✔
753
      }
1✔
754
    }
755

756
    syncContext.execute(new ShutdownNow());
1✔
757
    return this;
1✔
758
  }
759

760
  // Called from syncContext
761
  @VisibleForTesting
762
  void panic(final Throwable t) {
763
    if (panicMode) {
1✔
764
      // Preserve the first panic information
765
      return;
×
766
    }
767
    panicMode = true;
1✔
768
    try {
769
      cancelIdleTimer(/* permanent= */ true);
1✔
770
      shutdownNameResolverAndLoadBalancer(false);
1✔
771
    } finally {
772
      updateSubchannelPicker(new LoadBalancer.FixedResultPicker(PickResult.withDrop(
1✔
773
          Status.INTERNAL.withDescription("Panic! This is a bug!").withCause(t))));
1✔
774
      realChannel.updateConfigSelector(null);
1✔
775
      channelLogger.log(ChannelLogLevel.ERROR, "PANIC! Entering TRANSIENT_FAILURE");
1✔
776
      channelStateManager.gotoState(TRANSIENT_FAILURE);
1✔
777
    }
778
  }
1✔
779

780
  @VisibleForTesting
781
  boolean isInPanicMode() {
782
    return panicMode;
1✔
783
  }
784

785
  // Called from syncContext
786
  private void updateSubchannelPicker(SubchannelPicker newPicker) {
787
    delayedTransport.reprocess(newPicker);
1✔
788
  }
1✔
789

790
  @Override
791
  public boolean isShutdown() {
792
    return shutdown.get();
1✔
793
  }
794

795
  @Override
796
  public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
797
    return terminatedLatch.await(timeout, unit);
1✔
798
  }
799

800
  @Override
801
  public boolean isTerminated() {
802
    return terminated;
1✔
803
  }
804

805
  /*
806
   * Creates a new outgoing call on the channel.
807
   */
808
  @Override
809
  public <ReqT, RespT> ClientCall<ReqT, RespT> newCall(MethodDescriptor<ReqT, RespT> method,
810
      CallOptions callOptions) {
811
    return interceptorChannel.newCall(method, callOptions);
1✔
812
  }
813

814
  @Override
815
  public String authority() {
816
    return interceptorChannel.authority();
1✔
817
  }
818

819
  private Executor getCallExecutor(CallOptions callOptions) {
820
    Executor executor = callOptions.getExecutor();
1✔
821
    if (executor == null) {
1✔
822
      executor = this.executor;
1✔
823
    }
824
    return executor;
1✔
825
  }
826

827
  private class RealChannel extends Channel {
828
    // Reference to null if no config selector is available from resolution result
829
    // Reference must be set() from syncContext
830
    private final AtomicReference<InternalConfigSelector> configSelector =
1✔
831
        new AtomicReference<>(INITIAL_PENDING_SELECTOR);
1✔
832
    // Set when the NameResolver is initially created. When we create a new NameResolver for the
833
    // same target, the new instance must have the same value.
834
    private final String authority;
835

836
    private final Channel clientCallImplChannel = new Channel() {
1✔
837
      @Override
838
      public <RequestT, ResponseT> ClientCall<RequestT, ResponseT> newCall(
839
          MethodDescriptor<RequestT, ResponseT> method, CallOptions callOptions) {
840
        return new ClientCallImpl<>(
1✔
841
            method,
842
            getCallExecutor(callOptions),
1✔
843
            callOptions,
844
            transportProvider,
1✔
845
            terminated ? null : transportFactory.getScheduledExecutorService(),
1✔
846
            channelCallTracer,
1✔
847
            null)
848
            .setFullStreamDecompression(fullStreamDecompression)
1✔
849
            .setDecompressorRegistry(decompressorRegistry)
1✔
850
            .setCompressorRegistry(compressorRegistry);
1✔
851
      }
852

853
      @Override
854
      public String authority() {
855
        return authority;
×
856
      }
857
    };
858

859
    private RealChannel(String authority) {
1✔
860
      this.authority =  checkNotNull(authority, "authority");
1✔
861
    }
1✔
862

863
    @Override
864
    public <ReqT, RespT> ClientCall<ReqT, RespT> newCall(
865
        MethodDescriptor<ReqT, RespT> method, CallOptions callOptions) {
866
      if (configSelector.get() != INITIAL_PENDING_SELECTOR) {
1✔
867
        return newClientCall(method, callOptions);
1✔
868
      }
869
      syncContext.execute(new Runnable() {
1✔
870
        @Override
871
        public void run() {
872
          exitIdleMode();
1✔
873
        }
1✔
874
      });
875
      if (configSelector.get() != INITIAL_PENDING_SELECTOR) {
1✔
876
        // This is an optimization for the case (typically with InProcessTransport) when name
877
        // resolution result is immediately available at this point. Otherwise, some users'
878
        // tests might observe slight behavior difference from earlier grpc versions.
879
        return newClientCall(method, callOptions);
1✔
880
      }
881
      if (shutdown.get()) {
1✔
882
        // Return a failing ClientCall.
883
        return new ClientCall<ReqT, RespT>() {
×
884
          @Override
885
          public void start(Listener<RespT> responseListener, Metadata headers) {
886
            responseListener.onClose(SHUTDOWN_STATUS, new Metadata());
×
887
          }
×
888

889
          @Override public void request(int numMessages) {}
×
890

891
          @Override public void cancel(@Nullable String message, @Nullable Throwable cause) {}
×
892

893
          @Override public void halfClose() {}
×
894

895
          @Override public void sendMessage(ReqT message) {}
×
896
        };
897
      }
898
      Context context = Context.current();
1✔
899
      final PendingCall<ReqT, RespT> pendingCall = new PendingCall<>(context, method, callOptions);
1✔
900
      syncContext.execute(new Runnable() {
1✔
901
        @Override
902
        public void run() {
903
          if (configSelector.get() == INITIAL_PENDING_SELECTOR) {
1✔
904
            if (pendingCalls == null) {
1✔
905
              pendingCalls = new LinkedHashSet<>();
1✔
906
              inUseStateAggregator.updateObjectInUse(pendingCallsInUseObject, true);
1✔
907
            }
908
            pendingCalls.add(pendingCall);
1✔
909
          } else {
910
            pendingCall.reprocess();
1✔
911
          }
912
        }
1✔
913
      });
914
      return pendingCall;
1✔
915
    }
916

917
    // Must run in SynchronizationContext.
918
    void updateConfigSelector(@Nullable InternalConfigSelector config) {
919
      InternalConfigSelector prevConfig = configSelector.get();
1✔
920
      configSelector.set(config);
1✔
921
      if (prevConfig == INITIAL_PENDING_SELECTOR && pendingCalls != null) {
1✔
922
        for (RealChannel.PendingCall<?, ?> pendingCall : pendingCalls) {
1✔
923
          pendingCall.reprocess();
1✔
924
        }
1✔
925
      }
926
    }
1✔
927

928
    // Must run in SynchronizationContext.
929
    void onConfigError() {
930
      if (configSelector.get() == INITIAL_PENDING_SELECTOR) {
1✔
931
        // Apply Default Service Config if initial name resolution fails.
932
        if (defaultServiceConfig != null) {
1✔
933
          updateConfigSelector(defaultServiceConfig.getDefaultConfigSelector());
1✔
934
          lastServiceConfig = defaultServiceConfig;
1✔
935
          channelLogger.log(ChannelLogLevel.ERROR,
1✔
936
              "Initial Name Resolution error, using default service config");
937
        } else {
938
          updateConfigSelector(null);
1✔
939
        }
940
      }
941
    }
1✔
942

943
    void shutdown() {
944
      final class RealChannelShutdown implements Runnable {
1✔
945
        @Override
946
        public void run() {
947
          if (pendingCalls == null) {
1✔
948
            if (configSelector.get() == INITIAL_PENDING_SELECTOR) {
1✔
949
              configSelector.set(null);
1✔
950
            }
951
            uncommittedRetriableStreamsRegistry.onShutdown(SHUTDOWN_STATUS);
1✔
952
          }
953
        }
1✔
954
      }
955

956
      syncContext.execute(new RealChannelShutdown());
1✔
957
    }
1✔
958

959
    void shutdownNow() {
960
      final class RealChannelShutdownNow implements Runnable {
1✔
961
        @Override
962
        public void run() {
963
          if (configSelector.get() == INITIAL_PENDING_SELECTOR) {
1✔
964
            configSelector.set(null);
1✔
965
          }
966
          if (pendingCalls != null) {
1✔
967
            for (RealChannel.PendingCall<?, ?> pendingCall : pendingCalls) {
1✔
968
              pendingCall.cancel("Channel is forcefully shutdown", null);
1✔
969
            }
1✔
970
          }
971
          uncommittedRetriableStreamsRegistry.onShutdownNow(SHUTDOWN_NOW_STATUS);
1✔
972
        }
1✔
973
      }
974

975
      syncContext.execute(new RealChannelShutdownNow());
1✔
976
    }
1✔
977

978
    @Override
979
    public String authority() {
980
      return authority;
1✔
981
    }
982

983
    private final class PendingCall<ReqT, RespT> extends DelayedClientCall<ReqT, RespT> {
984
      final Context context;
985
      final MethodDescriptor<ReqT, RespT> method;
986
      final CallOptions callOptions;
987
      private final long callCreationTime;
988

989
      PendingCall(Context context, MethodDescriptor<ReqT, RespT> method, CallOptions callOptions) {
1✔
990
        super(
1✔
991
            "name_resolver",
992
            getCallExecutor(callOptions),
1✔
993
            scheduledExecutor,
1✔
994
            callOptions.getDeadline());
1✔
995
        this.context = context;
1✔
996
        this.method = method;
1✔
997
        this.callOptions = callOptions;
1✔
998
        this.callCreationTime = ticker.nanoTime();
1✔
999
      }
1✔
1000

1001
      /** Called when it's ready to create a real call and reprocess the pending call. */
1002
      void reprocess() {
1003
        ClientCall<ReqT, RespT> realCall;
1004
        Context previous = context.attach();
1✔
1005
        try {
1006
          CallOptions delayResolutionOption = callOptions.withOption(NAME_RESOLUTION_DELAYED,
1✔
1007
              ticker.nanoTime() - callCreationTime);
1✔
1008
          realCall = newClientCall(method, delayResolutionOption);
1✔
1009
        } finally {
1010
          context.detach(previous);
1✔
1011
        }
1012
        Runnable toRun = setCall(realCall);
1✔
1013
        if (toRun == null) {
1✔
1014
          syncContext.execute(new PendingCallRemoval());
1✔
1015
        } else {
1016
          getCallExecutor(callOptions).execute(new Runnable() {
1✔
1017
            @Override
1018
            public void run() {
1019
              toRun.run();
1✔
1020
              syncContext.execute(new PendingCallRemoval());
1✔
1021
            }
1✔
1022
          });
1023
        }
1024
      }
1✔
1025

1026
      @Override
1027
      protected void callCancelled() {
1028
        super.callCancelled();
1✔
1029
        syncContext.execute(new PendingCallRemoval());
1✔
1030
      }
1✔
1031

1032
      final class PendingCallRemoval implements Runnable {
1✔
1033
        @Override
1034
        public void run() {
1035
          if (pendingCalls != null) {
1✔
1036
            pendingCalls.remove(PendingCall.this);
1✔
1037
            if (pendingCalls.isEmpty()) {
1✔
1038
              inUseStateAggregator.updateObjectInUse(pendingCallsInUseObject, false);
1✔
1039
              pendingCalls = null;
1✔
1040
              if (shutdown.get()) {
1✔
1041
                uncommittedRetriableStreamsRegistry.onShutdown(SHUTDOWN_STATUS);
1✔
1042
              }
1043
            }
1044
          }
1045
        }
1✔
1046
      }
1047
    }
1048

1049
    private <ReqT, RespT> ClientCall<ReqT, RespT> newClientCall(
1050
        MethodDescriptor<ReqT, RespT> method, CallOptions callOptions) {
1051
      InternalConfigSelector selector = configSelector.get();
1✔
1052
      if (selector == null) {
1✔
1053
        return clientCallImplChannel.newCall(method, callOptions);
1✔
1054
      }
1055
      if (selector instanceof ServiceConfigConvertedSelector) {
1✔
1056
        MethodInfo methodInfo =
1✔
1057
            ((ServiceConfigConvertedSelector) selector).config.getMethodConfig(method);
1✔
1058
        if (methodInfo != null) {
1✔
1059
          callOptions = callOptions.withOption(MethodInfo.KEY, methodInfo);
1✔
1060
        }
1061
        return clientCallImplChannel.newCall(method, callOptions);
1✔
1062
      }
1063
      return new ConfigSelectingClientCall<>(
1✔
1064
          selector, clientCallImplChannel, executor, method, callOptions);
1✔
1065
    }
1066
  }
1067

1068
  /**
1069
   * A client call for a given channel that applies a given config selector when it starts.
1070
   */
1071
  static final class ConfigSelectingClientCall<ReqT, RespT>
1072
      extends ForwardingClientCall<ReqT, RespT> {
1073

1074
    private final InternalConfigSelector configSelector;
1075
    private final Channel channel;
1076
    private final Executor callExecutor;
1077
    private final MethodDescriptor<ReqT, RespT> method;
1078
    private final Context context;
1079
    private CallOptions callOptions;
1080

1081
    private ClientCall<ReqT, RespT> delegate;
1082

1083
    ConfigSelectingClientCall(
1084
        InternalConfigSelector configSelector, Channel channel, Executor channelExecutor,
1085
        MethodDescriptor<ReqT, RespT> method,
1086
        CallOptions callOptions) {
1✔
1087
      this.configSelector = configSelector;
1✔
1088
      this.channel = channel;
1✔
1089
      this.method = method;
1✔
1090
      this.callExecutor =
1✔
1091
          callOptions.getExecutor() == null ? channelExecutor : callOptions.getExecutor();
1✔
1092
      this.callOptions = callOptions.withExecutor(callExecutor);
1✔
1093
      this.context = Context.current();
1✔
1094
    }
1✔
1095

1096
    @Override
1097
    protected ClientCall<ReqT, RespT> delegate() {
1098
      return delegate;
1✔
1099
    }
1100

1101
    @SuppressWarnings("unchecked")
1102
    @Override
1103
    public void start(Listener<RespT> observer, Metadata headers) {
1104
      PickSubchannelArgs args =
1✔
1105
          new PickSubchannelArgsImpl(method, headers, callOptions, NOOP_PICK_DETAILS_CONSUMER);
1✔
1106
      InternalConfigSelector.Result result = configSelector.selectConfig(args);
1✔
1107
      Status status = result.getStatus();
1✔
1108
      if (!status.isOk()) {
1✔
1109
        executeCloseObserverInContext(observer,
1✔
1110
            GrpcUtil.replaceInappropriateControlPlaneStatus(status));
1✔
1111
        delegate = (ClientCall<ReqT, RespT>) NOOP_CALL;
1✔
1112
        return;
1✔
1113
      }
1114
      ClientInterceptor interceptor = result.getInterceptor();
1✔
1115
      ManagedChannelServiceConfig config = (ManagedChannelServiceConfig) result.getConfig();
1✔
1116
      MethodInfo methodInfo = config.getMethodConfig(method);
1✔
1117
      if (methodInfo != null) {
1✔
1118
        callOptions = callOptions.withOption(MethodInfo.KEY, methodInfo);
1✔
1119
      }
1120
      if (interceptor != null) {
1✔
1121
        delegate = interceptor.interceptCall(method, callOptions, channel);
1✔
1122
      } else {
1123
        delegate = channel.newCall(method, callOptions);
×
1124
      }
1125
      delegate.start(observer, headers);
1✔
1126
    }
1✔
1127

1128
    private void executeCloseObserverInContext(
1129
        final Listener<RespT> observer, final Status status) {
1130
      class CloseInContext extends ContextRunnable {
1131
        CloseInContext() {
1✔
1132
          super(context);
1✔
1133
        }
1✔
1134

1135
        @Override
1136
        public void runInContext() {
1137
          observer.onClose(status, new Metadata());
1✔
1138
        }
1✔
1139
      }
1140

1141
      callExecutor.execute(new CloseInContext());
1✔
1142
    }
1✔
1143

1144
    @Override
1145
    public void cancel(@Nullable String message, @Nullable Throwable cause) {
1146
      if (delegate != null) {
×
1147
        delegate.cancel(message, cause);
×
1148
      }
1149
    }
×
1150
  }
1151

1152
  private static final ClientCall<Object, Object> NOOP_CALL = new ClientCall<Object, Object>() {
1✔
1153
    @Override
1154
    public void start(Listener<Object> responseListener, Metadata headers) {}
×
1155

1156
    @Override
1157
    public void request(int numMessages) {}
1✔
1158

1159
    @Override
1160
    public void cancel(String message, Throwable cause) {}
×
1161

1162
    @Override
1163
    public void halfClose() {}
×
1164

1165
    @Override
1166
    public void sendMessage(Object message) {}
×
1167

1168
    // Always returns {@code false}, since this is only used when the startup of the call fails.
1169
    @Override
1170
    public boolean isReady() {
1171
      return false;
×
1172
    }
1173
  };
1174

1175
  /**
1176
   * Terminate the channel if termination conditions are met.
1177
   */
1178
  // Must be run from syncContext
1179
  private void maybeTerminateChannel() {
1180
    if (terminated) {
1✔
1181
      return;
×
1182
    }
1183
    if (shutdown.get() && subchannels.isEmpty()) {
1✔
1184
      channelLogger.log(ChannelLogLevel.INFO, "Terminated");
1✔
1185
      channelz.removeRootChannel(this);
1✔
1186
      executorPool.returnObject(executor);
1✔
1187
      balancerRpcExecutorHolder.release();
1✔
1188
      offloadExecutorHolder.release();
1✔
1189
      // Release the transport factory so that it can deallocate any resources.
1190
      transportFactory.close();
1✔
1191

1192
      terminated = true;
1✔
1193
      terminatedLatch.countDown();
1✔
1194
    }
1195
  }
1✔
1196

1197
  @Override
1198
  public ConnectivityState getState(boolean requestConnection) {
1199
    ConnectivityState savedChannelState = channelStateManager.getState();
1✔
1200
    if (requestConnection && savedChannelState == IDLE) {
1✔
1201
      final class RequestConnection implements Runnable {
1✔
1202
        @Override
1203
        public void run() {
1204
          exitIdleMode();
1✔
1205
          if (lbHelper != null) {
1✔
1206
            lbHelper.lb.requestConnection();
1✔
1207
          }
1208
        }
1✔
1209
      }
1210

1211
      syncContext.execute(new RequestConnection());
1✔
1212
    }
1213
    return savedChannelState;
1✔
1214
  }
1215

1216
  @Override
1217
  public void notifyWhenStateChanged(final ConnectivityState source, final Runnable callback) {
1218
    final class NotifyStateChanged implements Runnable {
1✔
1219
      @Override
1220
      public void run() {
1221
        channelStateManager.notifyWhenStateChanged(callback, executor, source);
1✔
1222
      }
1✔
1223
    }
1224

1225
    syncContext.execute(new NotifyStateChanged());
1✔
1226
  }
1✔
1227

1228
  @Override
1229
  public void resetConnectBackoff() {
1230
    final class ResetConnectBackoff implements Runnable {
1✔
1231
      @Override
1232
      public void run() {
1233
        if (shutdown.get()) {
1✔
1234
          return;
1✔
1235
        }
1236
        if (nameResolverStarted) {
1✔
1237
          refreshNameResolution();
1✔
1238
        }
1239
        for (InternalSubchannel subchannel : subchannels) {
1✔
1240
          subchannel.resetConnectBackoff();
1✔
1241
        }
1✔
1242
      }
1✔
1243
    }
1244

1245
    syncContext.execute(new ResetConnectBackoff());
1✔
1246
  }
1✔
1247

1248
  @Override
1249
  public void enterIdle() {
1250
    final class PrepareToLoseNetworkRunnable implements Runnable {
1✔
1251
      @Override
1252
      public void run() {
1253
        if (shutdown.get() || lbHelper == null) {
1✔
1254
          return;
1✔
1255
        }
1256
        cancelIdleTimer(/* permanent= */ false);
1✔
1257
        enterIdleMode();
1✔
1258
      }
1✔
1259
    }
1260

1261
    syncContext.execute(new PrepareToLoseNetworkRunnable());
1✔
1262
  }
1✔
1263

1264
  /**
1265
   * A registry that prevents channel shutdown from killing existing retry attempts that are in
1266
   * backoff.
1267
   */
1268
  private final class UncommittedRetriableStreamsRegistry {
1✔
1269
    // TODO(zdapeng): This means we would acquire a lock for each new retry-able stream,
1270
    // it's worthwhile to look for a lock-free approach.
1271
    final Object lock = new Object();
1✔
1272

1273
    @GuardedBy("lock")
1✔
1274
    Collection<ClientStream> uncommittedRetriableStreams = new HashSet<>();
1275

1276
    @GuardedBy("lock")
1277
    Status shutdownStatus;
1278

1279
    void onShutdown(Status reason) {
1280
      boolean shouldShutdownDelayedTransport = false;
1✔
1281
      synchronized (lock) {
1✔
1282
        if (shutdownStatus != null) {
1✔
1283
          return;
1✔
1284
        }
1285
        shutdownStatus = reason;
1✔
1286
        // Keep the delayedTransport open until there is no more uncommitted streams, b/c those
1287
        // retriable streams, which may be in backoff and not using any transport, are already
1288
        // started RPCs.
1289
        if (uncommittedRetriableStreams.isEmpty()) {
1✔
1290
          shouldShutdownDelayedTransport = true;
1✔
1291
        }
1292
      }
1✔
1293

1294
      if (shouldShutdownDelayedTransport) {
1✔
1295
        delayedTransport.shutdown(reason);
1✔
1296
      }
1297
    }
1✔
1298

1299
    void onShutdownNow(Status reason) {
1300
      onShutdown(reason);
1✔
1301
      Collection<ClientStream> streams;
1302

1303
      synchronized (lock) {
1✔
1304
        streams = new ArrayList<>(uncommittedRetriableStreams);
1✔
1305
      }
1✔
1306

1307
      for (ClientStream stream : streams) {
1✔
1308
        stream.cancel(reason);
1✔
1309
      }
1✔
1310
      delayedTransport.shutdownNow(reason);
1✔
1311
    }
1✔
1312

1313
    /**
1314
     * Registers a RetriableStream and return null if not shutdown, otherwise just returns the
1315
     * shutdown Status.
1316
     */
1317
    @Nullable
1318
    Status add(RetriableStream<?> retriableStream) {
1319
      synchronized (lock) {
1✔
1320
        if (shutdownStatus != null) {
1✔
1321
          return shutdownStatus;
1✔
1322
        }
1323
        uncommittedRetriableStreams.add(retriableStream);
1✔
1324
        return null;
1✔
1325
      }
1326
    }
1327

1328
    void remove(RetriableStream<?> retriableStream) {
1329
      Status shutdownStatusCopy = null;
1✔
1330

1331
      synchronized (lock) {
1✔
1332
        uncommittedRetriableStreams.remove(retriableStream);
1✔
1333
        if (uncommittedRetriableStreams.isEmpty()) {
1✔
1334
          shutdownStatusCopy = shutdownStatus;
1✔
1335
          // Because retriable transport is long-lived, we take this opportunity to down-size the
1336
          // hashmap.
1337
          uncommittedRetriableStreams = new HashSet<>();
1✔
1338
        }
1339
      }
1✔
1340

1341
      if (shutdownStatusCopy != null) {
1✔
1342
        delayedTransport.shutdown(shutdownStatusCopy);
1✔
1343
      }
1344
    }
1✔
1345
  }
1346

1347
  private final class LbHelperImpl extends LoadBalancer.Helper {
1✔
1348
    LoadBalancer lb;
1349

1350
    @Override
1351
    public AbstractSubchannel createSubchannel(CreateSubchannelArgs args) {
1352
      syncContext.throwIfNotInThisSynchronizationContext();
1✔
1353
      // No new subchannel should be created after load balancer has been shutdown.
1354
      checkState(!terminating, "Channel is being terminated");
1✔
1355
      return new SubchannelImpl(args);
1✔
1356
    }
1357

1358
    @Override
1359
    public void updateBalancingState(
1360
        final ConnectivityState newState, final SubchannelPicker newPicker) {
1361
      syncContext.throwIfNotInThisSynchronizationContext();
1✔
1362
      checkNotNull(newState, "newState");
1✔
1363
      checkNotNull(newPicker, "newPicker");
1✔
1364

1365
      if (LbHelperImpl.this != lbHelper || panicMode) {
1✔
1366
        return;
1✔
1367
      }
1368
      updateSubchannelPicker(newPicker);
1✔
1369
      // It's not appropriate to report SHUTDOWN state from lb.
1370
      // Ignore the case of newState == SHUTDOWN for now.
1371
      if (newState != SHUTDOWN) {
1✔
1372
        channelLogger.log(
1✔
1373
            ChannelLogLevel.INFO, "Entering {0} state with picker: {1}", newState, newPicker);
1374
        channelStateManager.gotoState(newState);
1✔
1375
      }
1376
    }
1✔
1377

1378
    @Override
1379
    public void refreshNameResolution() {
1380
      syncContext.throwIfNotInThisSynchronizationContext();
1✔
1381
      final class LoadBalancerRefreshNameResolution implements Runnable {
1✔
1382
        @Override
1383
        public void run() {
1384
          ManagedChannelImpl.this.refreshNameResolution();
1✔
1385
        }
1✔
1386
      }
1387

1388
      syncContext.execute(new LoadBalancerRefreshNameResolution());
1✔
1389
    }
1✔
1390

1391
    @Override
1392
    public ManagedChannel createOobChannel(EquivalentAddressGroup addressGroup, String authority) {
1393
      return createOobChannel(Collections.singletonList(addressGroup), authority);
×
1394
    }
1395

1396
    @Override
1397
    public ManagedChannel createOobChannel(List<EquivalentAddressGroup> addressGroup,
1398
        String authority) {
1399
      NameResolverRegistry nameResolverRegistry = new NameResolverRegistry();
1✔
1400
      OobNameResolverProvider resolverProvider =
1✔
1401
          new OobNameResolverProvider(authority, addressGroup, syncContext);
1402
      nameResolverRegistry.register(resolverProvider);
1✔
1403
      // We could use a hard-coded target, as the name resolver won't actually use this string.
1404
      // However, that would make debugging less clear, as we use the target to identify the
1405
      // channel.
1406
      String target;
1407
      try {
1408
        target = new URI("oob", "", "/" + authority, null, null).toString();
1✔
1409
      } catch (URISyntaxException ex) {
×
1410
        // Any special characters in the path will be percent encoded. So this should be impossible.
1411
        throw new AssertionError(ex);
×
1412
      }
1✔
1413
      ManagedChannel delegate = createResolvingOobChannelBuilder(
1✔
1414
          target, new DefaultChannelCreds(), nameResolverRegistry)
1415
          // TODO(zdapeng): executors should not outlive the parent channel.
1416
          .executor(balancerRpcExecutorHolder.getExecutor())
1✔
1417
          .idleTimeout(Integer.MAX_VALUE, TimeUnit.SECONDS)
1✔
1418
          .disableRetry()
1✔
1419
          .build();
1✔
1420
      return new OobChannel(delegate, resolverProvider);
1✔
1421
    }
1422

1423
    @Deprecated
1424
    @Override
1425
    public ManagedChannelBuilder<?> createResolvingOobChannelBuilder(String target) {
1426
      return createResolvingOobChannelBuilder(target, new DefaultChannelCreds())
1✔
1427
          // Override authority to keep the old behavior.
1428
          // createResolvingOobChannelBuilder(String target) will be deleted soon.
1429
          .overrideAuthority(getAuthority());
1✔
1430
    }
1431

1432
    @Override
1433
    public ManagedChannelBuilder<?> createResolvingOobChannelBuilder(
1434
        final String target, final ChannelCredentials channelCreds) {
1435
      return createResolvingOobChannelBuilder(target, channelCreds, nameResolverRegistry);
1✔
1436
    }
1437

1438
    // TODO(creamsoup) prevent main channel to shutdown if oob channel is not terminated
1439
    // TODO(zdapeng) register the channel as a subchannel of the parent channel in channelz.
1440
    private ManagedChannelBuilder<?> createResolvingOobChannelBuilder(
1441
        final String target, final ChannelCredentials channelCreds,
1442
        NameResolverRegistry nameResolverRegistry) {
1443
      checkNotNull(channelCreds, "channelCreds");
1✔
1444

1445
      final class ResolvingOobChannelBuilder
1446
          extends ForwardingChannelBuilder2<ResolvingOobChannelBuilder> {
1447
        final ManagedChannelBuilder<?> delegate;
1448

1449
        ResolvingOobChannelBuilder() {
1✔
1450
          final ClientTransportFactory transportFactory;
1451
          CallCredentials callCredentials;
1452
          if (channelCreds instanceof DefaultChannelCreds) {
1✔
1453
            transportFactory = originalTransportFactory;
1✔
1454
            callCredentials = null;
1✔
1455
          } else {
1456
            SwapChannelCredentialsResult swapResult =
1✔
1457
                originalTransportFactory.swapChannelCredentials(channelCreds);
1✔
1458
            if (swapResult == null) {
1✔
1459
              delegate = Grpc.newChannelBuilder(target, channelCreds);
×
1460
              return;
×
1461
            } else {
1462
              transportFactory = swapResult.transportFactory;
1✔
1463
              callCredentials = swapResult.callCredentials;
1✔
1464
            }
1465
          }
1466
          ClientTransportFactoryBuilder transportFactoryBuilder =
1✔
1467
              new ClientTransportFactoryBuilder() {
1✔
1468
                @Override
1469
                public ClientTransportFactory buildClientTransportFactory() {
1470
                  return transportFactory;
1✔
1471
                }
1472
              };
1473
          delegate = new ManagedChannelImplBuilder(
1✔
1474
              target,
1475
              channelCreds,
1476
              callCredentials,
1477
              transportFactoryBuilder,
1478
              new FixedPortProvider(nameResolverArgs.getDefaultPort()))
1✔
1479
              .nameResolverRegistry(nameResolverRegistry);
1✔
1480
        }
1✔
1481

1482
        @Override
1483
        protected ManagedChannelBuilder<?> delegate() {
1484
          return delegate;
1✔
1485
        }
1486
      }
1487

1488
      checkState(!terminated, "Channel is terminated");
1✔
1489

1490
      ResolvingOobChannelBuilder builder = new ResolvingOobChannelBuilder();
1✔
1491

1492
      return builder
1✔
1493
          // TODO(zdapeng): executors should not outlive the parent channel.
1494
          .executor(executor)
1✔
1495
          .offloadExecutor(offloadExecutorHolder.getExecutor())
1✔
1496
          .maxTraceEvents(maxTraceEvents)
1✔
1497
          .proxyDetector(nameResolverArgs.getProxyDetector())
1✔
1498
          .userAgent(userAgent);
1✔
1499
    }
1500

1501
    @Override
1502
    public ChannelCredentials getUnsafeChannelCredentials() {
1503
      if (originalChannelCreds == null) {
1✔
1504
        return new DefaultChannelCreds();
1✔
1505
      }
1506
      return originalChannelCreds;
×
1507
    }
1508

1509
    @Override
1510
    public void updateOobChannelAddresses(ManagedChannel channel, EquivalentAddressGroup eag) {
1511
      updateOobChannelAddresses(channel, Collections.singletonList(eag));
×
1512
    }
×
1513

1514
    @Override
1515
    public void updateOobChannelAddresses(ManagedChannel channel,
1516
        List<EquivalentAddressGroup> eag) {
1517
      checkArgument(channel instanceof OobChannel,
1✔
1518
          "channel must have been returned from createOobChannel");
1519
      ((OobChannel) channel).updateAddresses(eag);
1✔
1520
    }
1✔
1521

1522
    @Override
1523
    public String getAuthority() {
1524
      return ManagedChannelImpl.this.authority();
1✔
1525
    }
1526

1527
    @Override
1528
    public String getChannelTarget() {
1529
      return targetUri.toString();
1✔
1530
    }
1531

1532
    @Override
1533
    public SynchronizationContext getSynchronizationContext() {
1534
      return syncContext;
1✔
1535
    }
1536

1537
    @Override
1538
    public ScheduledExecutorService getScheduledExecutorService() {
1539
      return scheduledExecutor;
1✔
1540
    }
1541

1542
    @Override
1543
    public ChannelLogger getChannelLogger() {
1544
      return channelLogger;
1✔
1545
    }
1546

1547
    @Override
1548
    public NameResolver.Args getNameResolverArgs() {
1549
      return nameResolverArgs;
1✔
1550
    }
1551

1552
    @Override
1553
    public NameResolverRegistry getNameResolverRegistry() {
1554
      return nameResolverRegistry;
1✔
1555
    }
1556

1557
    @Override
1558
    public MetricRecorder getMetricRecorder() {
1559
      return metricRecorder;
1✔
1560
    }
1561

1562
    /**
1563
     * A placeholder for channel creds if user did not specify channel creds for the channel.
1564
     */
1565
    // TODO(zdapeng): get rid of this class and let all ChannelBuilders always provide a non-null
1566
    //     channel creds.
1567
    final class DefaultChannelCreds extends ChannelCredentials {
1✔
1568
      @Override
1569
      public ChannelCredentials withoutBearerTokens() {
1570
        return this;
×
1571
      }
1572
    }
1573
  }
1574

1575
  static final class OobChannel extends ForwardingManagedChannel {
1576
    private final OobNameResolverProvider resolverProvider;
1577

1578
    public OobChannel(ManagedChannel delegate, OobNameResolverProvider resolverProvider) {
1579
      super(delegate);
1✔
1580
      this.resolverProvider = checkNotNull(resolverProvider, "resolverProvider");
1✔
1581
    }
1✔
1582

1583
    public void updateAddresses(List<EquivalentAddressGroup> eags) {
1584
      resolverProvider.updateAddresses(eags);
1✔
1585
    }
1✔
1586
  }
1587

1588
  final class NameResolverListener extends NameResolver.Listener2 {
1589
    final LbHelperImpl helper;
1590
    final NameResolver resolver;
1591

1592
    NameResolverListener(LbHelperImpl helperImpl, NameResolver resolver) {
1✔
1593
      this.helper = checkNotNull(helperImpl, "helperImpl");
1✔
1594
      this.resolver = checkNotNull(resolver, "resolver");
1✔
1595
    }
1✔
1596

1597
    @Override
1598
    public void onResult(final ResolutionResult resolutionResult) {
1599
      syncContext.execute(() -> onResult2(resolutionResult));
×
1600
    }
×
1601

1602
    @SuppressWarnings("ReferenceEquality")
1603
    @Override
1604
    public Status onResult2(final ResolutionResult resolutionResult) {
1605
      syncContext.throwIfNotInThisSynchronizationContext();
1✔
1606
      if (ManagedChannelImpl.this.nameResolver != resolver) {
1✔
1607
        return Status.OK;
1✔
1608
      }
1609

1610
      StatusOr<List<EquivalentAddressGroup>> serversOrError =
1✔
1611
          resolutionResult.getAddressesOrError();
1✔
1612
      if (!serversOrError.hasValue()) {
1✔
1613
        handleErrorInSyncContext(serversOrError.getStatus());
1✔
1614
        return serversOrError.getStatus();
1✔
1615
      }
1616
      List<EquivalentAddressGroup> servers = serversOrError.getValue();
1✔
1617
      channelLogger.log(
1✔
1618
          ChannelLogLevel.DEBUG,
1619
          "Resolved address: {0}, config={1}",
1620
          servers,
1621
          resolutionResult.getAttributes());
1✔
1622

1623
      if (lastResolutionState != ResolutionState.SUCCESS) {
1✔
1624
        channelLogger.log(ChannelLogLevel.INFO, "Address resolved: {0}",
1✔
1625
            servers);
1626
        lastResolutionState = ResolutionState.SUCCESS;
1✔
1627
      }
1628
      ConfigOrError configOrError = resolutionResult.getServiceConfig();
1✔
1629
      InternalConfigSelector resolvedConfigSelector =
1✔
1630
          resolutionResult.getAttributes().get(InternalConfigSelector.KEY);
1✔
1631
      ManagedChannelServiceConfig validServiceConfig =
1632
          configOrError != null && configOrError.getConfig() != null
1✔
1633
              ? (ManagedChannelServiceConfig) configOrError.getConfig()
1✔
1634
              : null;
1✔
1635
      Status serviceConfigError = configOrError != null ? configOrError.getError() : null;
1✔
1636

1637
      ManagedChannelServiceConfig effectiveServiceConfig;
1638
      if (!lookUpServiceConfig) {
1✔
1639
        if (validServiceConfig != null) {
1✔
1640
          channelLogger.log(
1✔
1641
              ChannelLogLevel.INFO,
1642
              "Service config from name resolver discarded by channel settings");
1643
        }
1644
        effectiveServiceConfig =
1645
            defaultServiceConfig == null ? EMPTY_SERVICE_CONFIG : defaultServiceConfig;
1✔
1646
        if (resolvedConfigSelector != null) {
1✔
1647
          channelLogger.log(
1✔
1648
              ChannelLogLevel.INFO,
1649
              "Config selector from name resolver discarded by channel settings");
1650
        }
1651
        realChannel.updateConfigSelector(effectiveServiceConfig.getDefaultConfigSelector());
1✔
1652
      } else {
1653
        // Try to use config if returned from name resolver
1654
        // Otherwise, try to use the default config if available
1655
        if (validServiceConfig != null) {
1✔
1656
          effectiveServiceConfig = validServiceConfig;
1✔
1657
          if (resolvedConfigSelector != null) {
1✔
1658
            realChannel.updateConfigSelector(resolvedConfigSelector);
1✔
1659
            if (effectiveServiceConfig.getDefaultConfigSelector() != null) {
1✔
1660
              channelLogger.log(
×
1661
                  ChannelLogLevel.DEBUG,
1662
                  "Method configs in service config will be discarded due to presence of"
1663
                      + "config-selector");
1664
            }
1665
          } else {
1666
            realChannel.updateConfigSelector(effectiveServiceConfig.getDefaultConfigSelector());
1✔
1667
          }
1668
        } else if (defaultServiceConfig != null) {
1✔
1669
          effectiveServiceConfig = defaultServiceConfig;
1✔
1670
          realChannel.updateConfigSelector(effectiveServiceConfig.getDefaultConfigSelector());
1✔
1671
          channelLogger.log(
1✔
1672
              ChannelLogLevel.INFO,
1673
              "Received no service config, using default service config");
1674
        } else if (serviceConfigError != null) {
1✔
1675
          if (!serviceConfigUpdated) {
1✔
1676
            // First DNS lookup has invalid service config, and cannot fall back to default
1677
            channelLogger.log(
1✔
1678
                ChannelLogLevel.INFO,
1679
                "Fallback to error due to invalid first service config without default config");
1680
            // This error could be an "inappropriate" control plane error that should not bleed
1681
            // through to client code using gRPC. We let them flow through here to the LB as
1682
            // we later check for these error codes when investigating pick results in
1683
            // GrpcUtil.getTransportFromPickResult().
1684
            onError(configOrError.getError());
1✔
1685
            return configOrError.getError();
1✔
1686
          } else {
1687
            effectiveServiceConfig = lastServiceConfig;
1✔
1688
          }
1689
        } else {
1690
          effectiveServiceConfig = EMPTY_SERVICE_CONFIG;
1✔
1691
          realChannel.updateConfigSelector(null);
1✔
1692
        }
1693
        if (!effectiveServiceConfig.equals(lastServiceConfig)) {
1✔
1694
          channelLogger.log(
1✔
1695
              ChannelLogLevel.INFO,
1696
              "Service config changed{0}",
1697
              effectiveServiceConfig == EMPTY_SERVICE_CONFIG ? " to empty" : "");
1✔
1698
          lastServiceConfig = effectiveServiceConfig;
1✔
1699
          transportProvider.throttle = effectiveServiceConfig.getRetryThrottling();
1✔
1700
        }
1701

1702
        try {
1703
          // TODO(creamsoup): when `serversOrError` is empty and lastResolutionStateCopy == SUCCESS
1704
          //  and lbNeedAddress, it shouldn't call the handleServiceConfigUpdate. But,
1705
          //  lbNeedAddress is not deterministic
1706
          serviceConfigUpdated = true;
1✔
1707
        } catch (RuntimeException re) {
×
1708
          logger.log(
×
1709
              Level.WARNING,
1710
              "[" + getLogId() + "] Unexpected exception from parsing service config",
×
1711
              re);
1712
        }
1✔
1713
      }
1714

1715
      Attributes effectiveAttrs = resolutionResult.getAttributes();
1✔
1716
      // Call LB only if it's not shutdown.  If LB is shutdown, lbHelper won't match.
1717
      if (NameResolverListener.this.helper == ManagedChannelImpl.this.lbHelper) {
1✔
1718
        Attributes.Builder attrBuilder =
1✔
1719
            effectiveAttrs.toBuilder().discard(InternalConfigSelector.KEY);
1✔
1720
        Map<String, ?> healthCheckingConfig =
1✔
1721
            effectiveServiceConfig.getHealthCheckingConfig();
1✔
1722
        if (healthCheckingConfig != null) {
1✔
1723
          attrBuilder
1✔
1724
              .set(LoadBalancer.ATTR_HEALTH_CHECKING_CONFIG, healthCheckingConfig)
1✔
1725
              .build();
1✔
1726
        }
1727
        Attributes attributes = attrBuilder.build();
1✔
1728

1729
        ResolvedAddresses.Builder resolvedAddresses = ResolvedAddresses.newBuilder()
1✔
1730
            .setAddresses(serversOrError.getValue())
1✔
1731
            .setAttributes(attributes)
1✔
1732
            .setLoadBalancingPolicyConfig(effectiveServiceConfig.getLoadBalancingConfig());
1✔
1733
        Status addressAcceptanceStatus = helper.lb.acceptResolvedAddresses(
1✔
1734
            resolvedAddresses.build());
1✔
1735
        return addressAcceptanceStatus;
1✔
1736
      }
1737
      return Status.OK;
×
1738
    }
1739

1740
    @Override
1741
    public void onError(final Status error) {
1742
      checkArgument(!error.isOk(), "the error status must not be OK");
1✔
1743
      final class NameResolverErrorHandler implements Runnable {
1✔
1744
        @Override
1745
        public void run() {
1746
          handleErrorInSyncContext(error);
1✔
1747
        }
1✔
1748
      }
1749

1750
      syncContext.execute(new NameResolverErrorHandler());
1✔
1751
    }
1✔
1752

1753
    private void handleErrorInSyncContext(Status error) {
1754
      logger.log(Level.WARNING, "[{0}] Failed to resolve name. status={1}",
1✔
1755
          new Object[] {getLogId(), error});
1✔
1756
      realChannel.onConfigError();
1✔
1757
      if (lastResolutionState != ResolutionState.ERROR) {
1✔
1758
        channelLogger.log(ChannelLogLevel.WARNING, "Failed to resolve name: {0}", error);
1✔
1759
        lastResolutionState = ResolutionState.ERROR;
1✔
1760
      }
1761
      // Call LB only if it's not shutdown.  If LB is shutdown, lbHelper won't match.
1762
      if (NameResolverListener.this.helper != ManagedChannelImpl.this.lbHelper) {
1✔
1763
        return;
1✔
1764
      }
1765

1766
      helper.lb.handleNameResolutionError(error);
1✔
1767
    }
1✔
1768
  }
1769

1770
  private final class SubchannelImpl extends AbstractSubchannel {
1771
    final CreateSubchannelArgs args;
1772
    final InternalLogId subchannelLogId;
1773
    final ChannelLoggerImpl subchannelLogger;
1774
    final ChannelTracer subchannelTracer;
1775
    List<EquivalentAddressGroup> addressGroups;
1776
    InternalSubchannel subchannel;
1777
    boolean started;
1778
    boolean shutdown;
1779
    ScheduledHandle delayedShutdownTask;
1780

1781
    SubchannelImpl(CreateSubchannelArgs args) {
1✔
1782
      checkNotNull(args, "args");
1✔
1783
      addressGroups = args.getAddresses();
1✔
1784
      if (authorityOverride != null) {
1✔
1785
        List<EquivalentAddressGroup> eagsWithoutOverrideAttr =
1✔
1786
            stripOverrideAuthorityAttributes(args.getAddresses());
1✔
1787
        args = args.toBuilder().setAddresses(eagsWithoutOverrideAttr).build();
1✔
1788
      }
1789
      this.args = args;
1✔
1790
      subchannelLogId = InternalLogId.allocate("Subchannel", /*details=*/ authority());
1✔
1791
      subchannelTracer = new ChannelTracer(
1✔
1792
          subchannelLogId, maxTraceEvents, timeProvider.currentTimeNanos(),
1✔
1793
          "Subchannel for " + args.getAddresses());
1✔
1794
      subchannelLogger = new ChannelLoggerImpl(subchannelTracer, timeProvider);
1✔
1795
    }
1✔
1796

1797
    @Override
1798
    public void start(final SubchannelStateListener listener) {
1799
      syncContext.throwIfNotInThisSynchronizationContext();
1✔
1800
      checkState(!started, "already started");
1✔
1801
      checkState(!shutdown, "already shutdown");
1✔
1802
      checkState(!terminating, "Channel is being terminated");
1✔
1803
      started = true;
1✔
1804
      final class ManagedInternalSubchannelCallback extends InternalSubchannel.Callback {
1✔
1805
        // All callbacks are run in syncContext
1806
        @Override
1807
        void onTerminated(InternalSubchannel is) {
1808
          subchannels.remove(is);
1✔
1809
          channelz.removeSubchannel(is);
1✔
1810
          maybeTerminateChannel();
1✔
1811
        }
1✔
1812

1813
        @Override
1814
        void onStateChange(InternalSubchannel is, ConnectivityStateInfo newState) {
1815
          checkState(listener != null, "listener is null");
1✔
1816
          listener.onSubchannelState(newState);
1✔
1817
        }
1✔
1818

1819
        @Override
1820
        void onInUse(InternalSubchannel is) {
1821
          inUseStateAggregator.updateObjectInUse(is, true);
1✔
1822
        }
1✔
1823

1824
        @Override
1825
        void onNotInUse(InternalSubchannel is) {
1826
          inUseStateAggregator.updateObjectInUse(is, false);
1✔
1827
        }
1✔
1828
      }
1829

1830
      final InternalSubchannel internalSubchannel = new InternalSubchannel(
1✔
1831
          args,
1832
          authority(),
1✔
1833
          userAgent,
1✔
1834
          backoffPolicyProvider,
1✔
1835
          transportFactory,
1✔
1836
          transportFactory.getScheduledExecutorService(),
1✔
1837
          stopwatchSupplier,
1✔
1838
          syncContext,
1839
          new ManagedInternalSubchannelCallback(),
1840
          channelz,
1✔
1841
          callTracerFactory.create(),
1✔
1842
          subchannelTracer,
1843
          subchannelLogId,
1844
          subchannelLogger,
1845
          transportFilters, target,
1✔
1846
          lbHelper.getMetricRecorder());
1✔
1847

1848
      channelTracer.reportEvent(new ChannelTrace.Event.Builder()
1✔
1849
          .setDescription("Child Subchannel started")
1✔
1850
          .setSeverity(ChannelTrace.Event.Severity.CT_INFO)
1✔
1851
          .setTimestampNanos(timeProvider.currentTimeNanos())
1✔
1852
          .setSubchannelRef(internalSubchannel)
1✔
1853
          .build());
1✔
1854

1855
      this.subchannel = internalSubchannel;
1✔
1856
      channelz.addSubchannel(internalSubchannel);
1✔
1857
      subchannels.add(internalSubchannel);
1✔
1858
    }
1✔
1859

1860
    @Override
1861
    InternalInstrumented<ChannelStats> getInstrumentedInternalSubchannel() {
1862
      checkState(started, "not started");
1✔
1863
      return subchannel;
1✔
1864
    }
1865

1866
    @Override
1867
    public void shutdown() {
1868
      syncContext.throwIfNotInThisSynchronizationContext();
1✔
1869
      if (subchannel == null) {
1✔
1870
        // start() was not successful
1871
        shutdown = true;
×
1872
        return;
×
1873
      }
1874
      if (shutdown) {
1✔
1875
        if (terminating && delayedShutdownTask != null) {
1✔
1876
          // shutdown() was previously called when terminating == false, thus a delayed shutdown()
1877
          // was scheduled.  Now since terminating == true, We should expedite the shutdown.
1878
          delayedShutdownTask.cancel();
×
1879
          delayedShutdownTask = null;
×
1880
          // Will fall through to the subchannel.shutdown() at the end.
1881
        } else {
1882
          return;
1✔
1883
        }
1884
      } else {
1885
        shutdown = true;
1✔
1886
      }
1887
      // Add a delay to shutdown to deal with the race between 1) a transport being picked and
1888
      // newStream() being called on it, and 2) its Subchannel is shut down by LoadBalancer (e.g.,
1889
      // because of address change, or because LoadBalancer is shutdown by Channel entering idle
1890
      // mode). If (2) wins, the app will see a spurious error. We work around this by delaying
1891
      // shutdown of Subchannel for a few seconds here.
1892
      //
1893
      // TODO(zhangkun83): consider a better approach
1894
      // (https://github.com/grpc/grpc-java/issues/2562).
1895
      if (!terminating) {
1✔
1896
        final class ShutdownSubchannel implements Runnable {
1✔
1897
          @Override
1898
          public void run() {
1899
            subchannel.shutdown(SUBCHANNEL_SHUTDOWN_STATUS);
1✔
1900
          }
1✔
1901
        }
1902

1903
        delayedShutdownTask = syncContext.schedule(
1✔
1904
            new LogExceptionRunnable(new ShutdownSubchannel()),
1905
            SUBCHANNEL_SHUTDOWN_DELAY_SECONDS, TimeUnit.SECONDS,
1906
            transportFactory.getScheduledExecutorService());
1✔
1907
        return;
1✔
1908
      }
1909
      // When terminating == true, no more real streams will be created. It's safe and also
1910
      // desirable to shutdown timely.
1911
      subchannel.shutdown(SHUTDOWN_STATUS);
1✔
1912
    }
1✔
1913

1914
    @Override
1915
    public void requestConnection() {
1916
      syncContext.throwIfNotInThisSynchronizationContext();
1✔
1917
      checkState(started, "not started");
1✔
1918
      if (shutdown) {
1✔
1919
        return;
1✔
1920
      }
1921
      subchannel.obtainActiveTransport();
1✔
1922
    }
1✔
1923

1924
    @Override
1925
    public List<EquivalentAddressGroup> getAllAddresses() {
1926
      syncContext.throwIfNotInThisSynchronizationContext();
1✔
1927
      checkState(started, "not started");
1✔
1928
      return addressGroups;
1✔
1929
    }
1930

1931
    @Override
1932
    public Attributes getAttributes() {
1933
      return args.getAttributes();
1✔
1934
    }
1935

1936
    @Override
1937
    public String toString() {
1938
      return subchannelLogId.toString();
1✔
1939
    }
1940

1941
    @Override
1942
    public Channel asChannel() {
1943
      checkState(started, "not started");
1✔
1944
      return new SubchannelChannel(
1✔
1945
          subchannel, balancerRpcExecutorHolder.getExecutor(),
1✔
1946
          transportFactory.getScheduledExecutorService(),
1✔
1947
          callTracerFactory.create(),
1✔
1948
          new AtomicReference<InternalConfigSelector>(null));
1949
    }
1950

1951
    @Override
1952
    public Object getInternalSubchannel() {
1953
      checkState(started, "Subchannel is not started");
1✔
1954
      return subchannel;
1✔
1955
    }
1956

1957
    @Override
1958
    public ChannelLogger getChannelLogger() {
1959
      return subchannelLogger;
1✔
1960
    }
1961

1962
    @Override
1963
    public void updateAddresses(List<EquivalentAddressGroup> addrs) {
1964
      syncContext.throwIfNotInThisSynchronizationContext();
1✔
1965
      addressGroups = addrs;
1✔
1966
      if (authorityOverride != null) {
1✔
1967
        addrs = stripOverrideAuthorityAttributes(addrs);
1✔
1968
      }
1969
      subchannel.updateAddresses(addrs);
1✔
1970
    }
1✔
1971

1972
    @Override
1973
    public Attributes getConnectedAddressAttributes() {
1974
      return subchannel.getConnectedAddressAttributes();
1✔
1975
    }
1976

1977
    private List<EquivalentAddressGroup> stripOverrideAuthorityAttributes(
1978
        List<EquivalentAddressGroup> eags) {
1979
      List<EquivalentAddressGroup> eagsWithoutOverrideAttr = new ArrayList<>();
1✔
1980
      for (EquivalentAddressGroup eag : eags) {
1✔
1981
        EquivalentAddressGroup eagWithoutOverrideAttr = new EquivalentAddressGroup(
1✔
1982
            eag.getAddresses(),
1✔
1983
            eag.getAttributes().toBuilder().discard(ATTR_AUTHORITY_OVERRIDE).build());
1✔
1984
        eagsWithoutOverrideAttr.add(eagWithoutOverrideAttr);
1✔
1985
      }
1✔
1986
      return Collections.unmodifiableList(eagsWithoutOverrideAttr);
1✔
1987
    }
1988
  }
1989

1990
  @Override
1991
  public String toString() {
1992
    return MoreObjects.toStringHelper(this)
1✔
1993
        .add("logId", logId.getId())
1✔
1994
        .add("target", target)
1✔
1995
        .toString();
1✔
1996
  }
1997

1998
  /**
1999
   * Called from syncContext.
2000
   */
2001
  private final class DelayedTransportListener implements ManagedClientTransport.Listener {
1✔
2002
    @Override
2003
    public void transportShutdown(Status s, DisconnectError e) {
2004
      checkState(shutdown.get(), "Channel must have been shut down");
1✔
2005
    }
1✔
2006

2007
    @Override
2008
    public void transportReady() {
2009
      // Don't care
2010
    }
×
2011

2012
    @Override
2013
    public Attributes filterTransport(Attributes attributes) {
2014
      return attributes;
×
2015
    }
2016

2017
    @Override
2018
    public void transportInUse(final boolean inUse) {
2019
      inUseStateAggregator.updateObjectInUse(delayedTransport, inUse);
1✔
2020
      if (inUse) {
1✔
2021
        // It's possible to be in idle mode while inUseStateAggregator is in-use, if one of the
2022
        // subchannels is in use. But we should never be in idle mode when delayed transport is in
2023
        // use.
2024
        exitIdleMode();
1✔
2025
      }
2026
    }
1✔
2027

2028
    @Override
2029
    public void transportTerminated() {
2030
      checkState(shutdown.get(), "Channel must have been shut down");
1✔
2031
      terminating = true;
1✔
2032
      shutdownNameResolverAndLoadBalancer(false);
1✔
2033
      // No need to call channelStateManager since we are already in SHUTDOWN state.
2034
      // Until LoadBalancer is shutdown, it may still create new subchannels.  We catch them
2035
      // here.
2036
      maybeShutdownNowSubchannels();
1✔
2037
      maybeTerminateChannel();
1✔
2038
    }
1✔
2039
  }
2040

2041
  /**
2042
   * Must be accessed from syncContext.
2043
   */
2044
  private final class IdleModeStateAggregator extends InUseStateAggregator<Object> {
1✔
2045
    @Override
2046
    protected void handleInUse() {
2047
      exitIdleMode();
1✔
2048
    }
1✔
2049

2050
    @Override
2051
    protected void handleNotInUse() {
2052
      if (shutdown.get()) {
1✔
2053
        return;
1✔
2054
      }
2055
      rescheduleIdleTimer();
1✔
2056
    }
1✔
2057
  }
2058

2059
  /**
2060
   * Lazily request for Executor from an executor pool.
2061
   * Also act as an Executor directly to simply run a cmd
2062
   */
2063
  @VisibleForTesting
2064
  static final class ExecutorHolder implements Executor {
2065
    private final ObjectPool<? extends Executor> pool;
2066
    private Executor executor;
2067

2068
    ExecutorHolder(ObjectPool<? extends Executor> executorPool) {
1✔
2069
      this.pool = checkNotNull(executorPool, "executorPool");
1✔
2070
    }
1✔
2071

2072
    synchronized Executor getExecutor() {
2073
      if (executor == null) {
1✔
2074
        executor = checkNotNull(pool.getObject(), "%s.getObject()", executor);
1✔
2075
      }
2076
      return executor;
1✔
2077
    }
2078

2079
    synchronized void release() {
2080
      if (executor != null) {
1✔
2081
        executor = pool.returnObject(executor);
1✔
2082
      }
2083
    }
1✔
2084

2085
    @Override
2086
    public void execute(Runnable command) {
2087
      getExecutor().execute(command);
1✔
2088
    }
1✔
2089
  }
2090

2091
  private static final class RestrictedScheduledExecutor implements ScheduledExecutorService {
2092
    final ScheduledExecutorService delegate;
2093

2094
    private RestrictedScheduledExecutor(ScheduledExecutorService delegate) {
1✔
2095
      this.delegate = checkNotNull(delegate, "delegate");
1✔
2096
    }
1✔
2097

2098
    @Override
2099
    public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit) {
2100
      return delegate.schedule(callable, delay, unit);
×
2101
    }
2102

2103
    @Override
2104
    public ScheduledFuture<?> schedule(Runnable cmd, long delay, TimeUnit unit) {
2105
      return delegate.schedule(cmd, delay, unit);
1✔
2106
    }
2107

2108
    @Override
2109
    public ScheduledFuture<?> scheduleAtFixedRate(
2110
        Runnable command, long initialDelay, long period, TimeUnit unit) {
2111
      return delegate.scheduleAtFixedRate(command, initialDelay, period, unit);
1✔
2112
    }
2113

2114
    @Override
2115
    public ScheduledFuture<?> scheduleWithFixedDelay(
2116
        Runnable command, long initialDelay, long delay, TimeUnit unit) {
2117
      return delegate.scheduleWithFixedDelay(command, initialDelay, delay, unit);
×
2118
    }
2119

2120
    @Override
2121
    public boolean awaitTermination(long timeout, TimeUnit unit)
2122
        throws InterruptedException {
2123
      return delegate.awaitTermination(timeout, unit);
×
2124
    }
2125

2126
    @Override
2127
    public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
2128
        throws InterruptedException {
2129
      return delegate.invokeAll(tasks);
×
2130
    }
2131

2132
    @Override
2133
    public <T> List<Future<T>> invokeAll(
2134
        Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
2135
        throws InterruptedException {
2136
      return delegate.invokeAll(tasks, timeout, unit);
×
2137
    }
2138

2139
    @Override
2140
    public <T> T invokeAny(Collection<? extends Callable<T>> tasks)
2141
        throws InterruptedException, ExecutionException {
2142
      return delegate.invokeAny(tasks);
×
2143
    }
2144

2145
    @Override
2146
    public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
2147
        throws InterruptedException, ExecutionException, TimeoutException {
2148
      return delegate.invokeAny(tasks, timeout, unit);
×
2149
    }
2150

2151
    @Override
2152
    public boolean isShutdown() {
2153
      return delegate.isShutdown();
×
2154
    }
2155

2156
    @Override
2157
    public boolean isTerminated() {
2158
      return delegate.isTerminated();
×
2159
    }
2160

2161
    @Override
2162
    public void shutdown() {
2163
      throw new UnsupportedOperationException("Restricted: shutdown() is not allowed");
1✔
2164
    }
2165

2166
    @Override
2167
    public List<Runnable> shutdownNow() {
2168
      throw new UnsupportedOperationException("Restricted: shutdownNow() is not allowed");
1✔
2169
    }
2170

2171
    @Override
2172
    public <T> Future<T> submit(Callable<T> task) {
2173
      return delegate.submit(task);
×
2174
    }
2175

2176
    @Override
2177
    public Future<?> submit(Runnable task) {
2178
      return delegate.submit(task);
×
2179
    }
2180

2181
    @Override
2182
    public <T> Future<T> submit(Runnable task, T result) {
2183
      return delegate.submit(task, result);
×
2184
    }
2185

2186
    @Override
2187
    public void execute(Runnable command) {
2188
      delegate.execute(command);
×
2189
    }
×
2190
  }
2191

2192
  /**
2193
   * A ResolutionState indicates the status of last name resolution.
2194
   */
2195
  enum ResolutionState {
1✔
2196
    NO_RESOLUTION,
1✔
2197
    SUCCESS,
1✔
2198
    ERROR
1✔
2199
  }
2200
}
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE TRIAL · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2026 Coveralls, Inc