#20136

Committed 06 Jan 2026 05:27AM UTC coverage: 88.693% (+0.01%) from 88.681%

Build # #20136

Build Type

push

github

Committed by

web-flow

Commit Message

core: Implement oobChannel with resolvingOobChannel

The most important part of this change is to ensure that CallCredentials
are not propagated to the OOB channel. Because the authority of the OOB
channel doesn't match the parent channel, we must ensure that any bearer
tokens are not sent to the different server. However, this was not a
problem because resolvingOobChannel has the same constraint. (RLS has a
different constraint, but we were able to let RLS manage that itself.)

This commit does change the behavior of channelz, shutdown, and metrics
for the OOB channel. Previously the OOB channel was registered with
channelz, but it is only a TODO for resolving channel. Channel shutdown
no longer shuts down the OOB channel and it no longer waits for the OOB
channel to terminate before becoming terminated itself. That is also a
pre-existing TODO. Since ManagedChannelImplBuilder is now being used,
global configurators and census are enabled. The proper behavior here is
still being determined, but we would want it to be the same for
resolving OOB channel and OOB channel.

The OOB channel used to refresh the name resolution when the subchannel
went IDLE or TF. That is an older behavior from back when regular
subchannels would also cause the name resolver to refresh. Now-a-days
that goes though the LB tree. gRPC-LB already refreshes name resolution
when its RPC closes, so no longer doing it automatically should be fine.

balancerRpcExecutorPool no longer has its lifetime managed by the child.
It'd be easiest to not use it at all from OOB channel, which wouldn't
actually change the regular behavior, as channels already use the same
executor by default. However, the tests are making use of the executor
being injected, so some propagation needs to be preserved.

Lots of OOB channel tests were deleted, but these were either testing
OobChannel, which is now gone, or things like channelz, which are known
to no longer work like before.

Run Details

35361 of 39869 relevant lines covered (88.69%)

0.89 hits per line

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

96.25

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

/*
 * Copyright 2015 The gRPC Authors
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package io.grpc.internal;

import static io.grpc.ConnectivityState.CONNECTING;
import static io.grpc.ConnectivityState.IDLE;
import static io.grpc.ConnectivityState.READY;
import static io.grpc.ConnectivityState.SHUTDOWN;
import static io.grpc.ConnectivityState.TRANSIENT_FAILURE;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.MoreObjects;
import com.google.common.base.Preconditions;
import com.google.common.base.Stopwatch;
import com.google.common.base.Supplier;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.SettableFuture;
import com.google.errorprone.annotations.ForOverride;
import io.grpc.Attributes;
import io.grpc.CallOptions;
import io.grpc.ChannelLogger;
import io.grpc.ChannelLogger.ChannelLogLevel;
import io.grpc.ClientStreamTracer;
import io.grpc.ClientTransportFilter;
import io.grpc.ConnectivityState;
import io.grpc.ConnectivityStateInfo;
import io.grpc.EquivalentAddressGroup;
import io.grpc.HttpConnectProxiedSocketAddress;
import io.grpc.InternalChannelz;
import io.grpc.InternalChannelz.ChannelStats;
import io.grpc.InternalInstrumented;
import io.grpc.InternalLogId;
import io.grpc.InternalWithLogId;
import io.grpc.LoadBalancer;
import io.grpc.Metadata;
import io.grpc.MethodDescriptor;
import io.grpc.MetricRecorder;
import io.grpc.NameResolver;
import io.grpc.SecurityLevel;
import io.grpc.Status;
import io.grpc.SynchronizationContext;
import io.grpc.SynchronizationContext.ScheduledHandle;
import java.net.SocketAddress;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import javax.annotation.Nullable;
import javax.annotation.concurrent.ThreadSafe;

/**
 * Transports for a single {@link SocketAddress}.
 */
@ThreadSafe
final class InternalSubchannel implements InternalInstrumented<ChannelStats>, TransportProvider {

  private final InternalLogId logId;
  private final String authority;
  private final String userAgent;
  private final BackoffPolicy.Provider backoffPolicyProvider;
  private final Callback callback;
  private final ClientTransportFactory transportFactory;
  private final ScheduledExecutorService scheduledExecutor;
  private final InternalChannelz channelz;
  private final CallTracer callsTracer;
  private final ChannelTracer channelTracer;
  private final ChannelLogger channelLogger;
  private final boolean reconnectDisabled;

  private final List<ClientTransportFilter> transportFilters;

  /**
   * All field must be mutated in the syncContext.
   */
  private final SynchronizationContext syncContext;

  /**
   * The index of the address corresponding to pendingTransport/activeTransport, or at beginning if
   * both are null.
   *
   * <p>Note: any {@link Index#updateAddresses(List)} should also update {@link #addressGroups}.
   */
  private final Index addressIndex;

  /**
   * A volatile accessor to {@link Index#getAddressGroups()}. There are few methods ({@link
   * #getAddressGroups()} and {@link #toString()} access this value where they supposed to access
   * in the {@link #syncContext}. Ideally {@link Index#getAddressGroups()} can be volatile, so we
   * don't need to maintain this volatile accessor. Although, having this accessor can reduce
   * unnecessary volatile reads while it delivers clearer intention of why .
   */
  private volatile List<EquivalentAddressGroup> addressGroups;

  /**
   * The policy to control back off between reconnects. Non-{@code null} when a reconnect task is
   * scheduled.
   */
  private BackoffPolicy reconnectPolicy;

  /**
   * Timer monitoring duration since entering CONNECTING state.
   */
  private final Stopwatch connectingTimer;

  @Nullable
  private ScheduledHandle reconnectTask;
  @Nullable
  private ScheduledHandle shutdownDueToUpdateTask;
  @Nullable
  private ManagedClientTransport shutdownDueToUpdateTransport;

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

  // Must only be used from syncContext
  private final InUseStateAggregator<ConnectionClientTransport> inUseStateAggregator =
      new InUseStateAggregator<ConnectionClientTransport>() {
        @Override
        protected void handleInUse() {
          callback.onInUse(InternalSubchannel.this);
        }

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

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

  /**
   * The transport for new outgoing requests. Non-null only in READY state.
   */
  @Nullable
  private volatile ManagedClientTransport activeTransport;

  private volatile ConnectivityStateInfo state = ConnectivityStateInfo.forNonError(IDLE);

  private Status shutdownReason;

  private volatile Attributes connectedAddressAttributes;
  private final SubchannelMetrics subchannelMetrics;
  private final String target;

  InternalSubchannel(LoadBalancer.CreateSubchannelArgs args, String authority, String userAgent,
                     BackoffPolicy.Provider backoffPolicyProvider,
                     ClientTransportFactory transportFactory,
                     ScheduledExecutorService scheduledExecutor,
                     Supplier<Stopwatch> stopwatchSupplier, SynchronizationContext syncContext,
                     Callback callback, InternalChannelz channelz, CallTracer callsTracer,
                     ChannelTracer channelTracer, InternalLogId logId,
                     ChannelLogger channelLogger, List<ClientTransportFilter> transportFilters,
                     String target,
                     MetricRecorder metricRecorder) {
    List<EquivalentAddressGroup> addressGroups = args.getAddresses();
    Preconditions.checkNotNull(addressGroups, "addressGroups");
    Preconditions.checkArgument(!addressGroups.isEmpty(), "addressGroups is empty");
    checkListHasNoNulls(addressGroups, "addressGroups contains null entry");
    List<EquivalentAddressGroup> unmodifiableAddressGroups =
        Collections.unmodifiableList(new ArrayList<>(addressGroups));
    this.addressGroups = unmodifiableAddressGroups;
    this.addressIndex = new Index(unmodifiableAddressGroups);
    this.authority = authority;
    this.userAgent = userAgent;
    this.backoffPolicyProvider = backoffPolicyProvider;
    this.transportFactory = transportFactory;
    this.scheduledExecutor = scheduledExecutor;
    this.connectingTimer = stopwatchSupplier.get();
    this.syncContext = syncContext;
    this.callback = callback;
    this.channelz = channelz;
    this.callsTracer = callsTracer;
    this.channelTracer = Preconditions.checkNotNull(channelTracer, "channelTracer");
    this.logId = Preconditions.checkNotNull(logId, "logId");
    this.channelLogger = Preconditions.checkNotNull(channelLogger, "channelLogger");
    this.transportFilters = transportFilters;
    this.reconnectDisabled = args.getOption(LoadBalancer.DISABLE_SUBCHANNEL_RECONNECT_KEY);
    this.target = target;
    this.subchannelMetrics = new SubchannelMetrics(metricRecorder);
  }

  ChannelLogger getChannelLogger() {
    return channelLogger;
  }

  @Override
  public ClientTransport obtainActiveTransport() {
    ClientTransport savedTransport = activeTransport;
    if (savedTransport != null) {
      return savedTransport;
    }
    syncContext.execute(new Runnable() {
      @Override
      public void run() {
        if (state.getState() == IDLE) {
          channelLogger.log(ChannelLogLevel.INFO, "CONNECTING as requested");
          gotoNonErrorState(CONNECTING);
          startNewTransport();
        }
      }
    });
    return null;
  }

  /**
   * Returns a READY transport if there is any, without trying to connect.
   */
  @Nullable
  ClientTransport getTransport() {
    return activeTransport;
  }

  /**
   * Returns the authority string associated with this Subchannel.
   */
  String getAuthority() {
    return authority;
  }

  private void startNewTransport() {
    syncContext.throwIfNotInThisSynchronizationContext();

    Preconditions.checkState(reconnectTask == null, "Should have no reconnectTask scheduled");

    if (addressIndex.isAtBeginning()) {
      connectingTimer.reset().start();
    }
    SocketAddress address = addressIndex.getCurrentAddress();

    HttpConnectProxiedSocketAddress proxiedAddr = null;
    if (address instanceof HttpConnectProxiedSocketAddress) {
      proxiedAddr = (HttpConnectProxiedSocketAddress) address;
      address = proxiedAddr.getTargetAddress();
    }

    Attributes currentEagAttributes = addressIndex.getCurrentEagAttributes();
    String eagChannelAuthority = currentEagAttributes
            .get(EquivalentAddressGroup.ATTR_AUTHORITY_OVERRIDE);
    ClientTransportFactory.ClientTransportOptions options =
        new ClientTransportFactory.ClientTransportOptions()
          .setAuthority(eagChannelAuthority != null ? eagChannelAuthority : authority)
          .setEagAttributes(currentEagAttributes)
          .setUserAgent(userAgent)
          .setHttpConnectProxiedSocketAddress(proxiedAddr);
    TransportLogger transportLogger = new TransportLogger();
    // In case the transport logs in the constructor, use the subchannel logId
    transportLogger.logId = getLogId();
    ConnectionClientTransport transport =
        new CallTracingTransport(
            transportFactory
                .newClientTransport(address, options, transportLogger), callsTracer);
    transportLogger.logId = transport.getLogId();
    channelz.addClientSocket(transport);
    pendingTransport = transport;
    transports.add(transport);
    Runnable runnable = transport.start(new TransportListener(transport));
    if (runnable != null) {
      syncContext.executeLater(runnable);
    }
    channelLogger.log(ChannelLogLevel.INFO, "Started transport {0}", transportLogger.logId);
  }

  /**
   * Only called after all addresses attempted and failed (TRANSIENT_FAILURE).
   * @param status the causal status when the channel begins transition to
   *     TRANSIENT_FAILURE.
   */
  private void scheduleBackoff(final Status status) {
    syncContext.throwIfNotInThisSynchronizationContext();

    class EndOfCurrentBackoff implements Runnable {
      @Override
      public void run() {
        reconnectTask = null;
        channelLogger.log(ChannelLogLevel.INFO, "CONNECTING after backoff");
        gotoNonErrorState(CONNECTING);
        startNewTransport();
      }
    }

    gotoState(ConnectivityStateInfo.forTransientFailure(status));

    if (reconnectDisabled) {
      return;
    }

    if (reconnectPolicy == null) {
      reconnectPolicy = backoffPolicyProvider.get();
    }
    long delayNanos =
        reconnectPolicy.nextBackoffNanos() - connectingTimer.elapsed(TimeUnit.NANOSECONDS);
    channelLogger.log(
        ChannelLogLevel.INFO,
        "TRANSIENT_FAILURE ({0}). Will reconnect after {1} ns",
        printShortStatus(status), delayNanos);
    Preconditions.checkState(reconnectTask == null, "previous reconnectTask is not done");
    reconnectTask = syncContext.schedule(
        new EndOfCurrentBackoff(),
        delayNanos,
        TimeUnit.NANOSECONDS,
        scheduledExecutor);
  }

  /**
   * Immediately attempt to reconnect if the current state is TRANSIENT_FAILURE. Otherwise, this
   * method has no effect.
   */
  void resetConnectBackoff() {
    syncContext.execute(new Runnable() {
      @Override
      public void run() {
        if (state.getState() != TRANSIENT_FAILURE) {
          return;
        }
        cancelReconnectTask();
        channelLogger.log(ChannelLogLevel.INFO, "CONNECTING; backoff interrupted");
        gotoNonErrorState(CONNECTING);
        startNewTransport();
      }
    });
  }

  private void gotoNonErrorState(final ConnectivityState newState) {
    syncContext.throwIfNotInThisSynchronizationContext();

    gotoState(ConnectivityStateInfo.forNonError(newState));
  }

  private void gotoState(final ConnectivityStateInfo newState) {
    syncContext.throwIfNotInThisSynchronizationContext();

    if (state.getState() != newState.getState()) {
      Preconditions.checkState(state.getState() != SHUTDOWN,
          "Cannot transition out of SHUTDOWN to %s", newState.getState());
      if (reconnectDisabled && newState.getState() == TRANSIENT_FAILURE) {
        state = ConnectivityStateInfo.forNonError(IDLE);
      } else {
        state = newState;
      }
      callback.onStateChange(InternalSubchannel.this, newState);
    }
  }

  /** Replaces the existing addresses, avoiding unnecessary reconnects. */
  public void updateAddresses(final List<EquivalentAddressGroup> newAddressGroups) {
    Preconditions.checkNotNull(newAddressGroups, "newAddressGroups");
    checkListHasNoNulls(newAddressGroups, "newAddressGroups contains null entry");
    Preconditions.checkArgument(!newAddressGroups.isEmpty(), "newAddressGroups is empty");
    final List<EquivalentAddressGroup> newImmutableAddressGroups =
        Collections.unmodifiableList(new ArrayList<>(newAddressGroups));

    syncContext.execute(new Runnable() {
      @Override
      public void run() {
        ManagedClientTransport savedTransport = null;
        SocketAddress previousAddress = addressIndex.getCurrentAddress();
        addressIndex.updateGroups(newImmutableAddressGroups);
        addressGroups = newImmutableAddressGroups;
        if (state.getState() == READY || state.getState() == CONNECTING) {
          if (!addressIndex.seekTo(previousAddress)) {
            // Forced to drop the connection
            if (state.getState() == READY) {
              savedTransport = activeTransport;
              activeTransport = null;
              addressIndex.reset();
              gotoNonErrorState(IDLE);
            } else {
              pendingTransport.shutdown(
                  Status.UNAVAILABLE.withDescription(
                    "InternalSubchannel closed pending transport due to address change"));
              pendingTransport = null;
              addressIndex.reset();
              startNewTransport();
            }
          }
        }
        if (savedTransport != null) {
          if (shutdownDueToUpdateTask != null) {
            // Keeping track of multiple shutdown tasks adds complexity, and shouldn't generally be
            // necessary. This transport has probably already had plenty of time.
            shutdownDueToUpdateTransport.shutdown(
                Status.UNAVAILABLE.withDescription(
                    "InternalSubchannel closed transport early due to address change"));
            shutdownDueToUpdateTask.cancel();
            shutdownDueToUpdateTask = null;
            shutdownDueToUpdateTransport = null;
          }
          // Avoid needless RPC failures by delaying the shutdown. See
          // https://github.com/grpc/grpc-java/issues/2562
          shutdownDueToUpdateTransport = savedTransport;
          shutdownDueToUpdateTask = syncContext.schedule(
              new Runnable() {
                @Override public void run() {
                  ManagedClientTransport transport = shutdownDueToUpdateTransport;
                  shutdownDueToUpdateTask = null;
                  shutdownDueToUpdateTransport = null;
                  transport.shutdown(
                      Status.UNAVAILABLE.withDescription(
                          "InternalSubchannel closed transport due to address change"));
                }
              },
              ManagedChannelImpl.SUBCHANNEL_SHUTDOWN_DELAY_SECONDS,
              TimeUnit.SECONDS,
              scheduledExecutor);
        }
      }
    });
  }

  public void shutdown(final Status reason) {
    syncContext.execute(new Runnable() {
      @Override
      public void run() {
        ManagedClientTransport savedActiveTransport;
        ConnectionClientTransport savedPendingTransport;
        if (state.getState() == SHUTDOWN) {
          return;
        }
        shutdownReason = reason;
        savedActiveTransport = activeTransport;
        savedPendingTransport = pendingTransport;
        activeTransport = null;
        pendingTransport = null;
        gotoNonErrorState(SHUTDOWN);
        addressIndex.reset();
        if (transports.isEmpty()) {
          handleTermination();
        }  // else: the callback will be run once all transports have been terminated
        cancelReconnectTask();
        if (shutdownDueToUpdateTask != null) {
          shutdownDueToUpdateTask.cancel();
          shutdownDueToUpdateTransport.shutdown(reason);
          shutdownDueToUpdateTask = null;
          shutdownDueToUpdateTransport = null;
        }
        if (savedActiveTransport != null) {
          savedActiveTransport.shutdown(reason);
        }
        if (savedPendingTransport != null) {
          savedPendingTransport.shutdown(reason);
        }
      }
    });
  }

  @Override
  public String toString() {
    // addressGroupsCopy being a little stale is fine, just avoid calling toString with the lock
    // since there may be many addresses.
    return MoreObjects.toStringHelper(this)
        .add("logId", logId.getId())
        .add("addressGroups", addressGroups)
        .toString();
  }

  private void handleTermination() {
    syncContext.execute(new Runnable() {
      @Override
      public void run() {
        channelLogger.log(ChannelLogLevel.INFO, "Terminated");
        callback.onTerminated(InternalSubchannel.this);
      }
    });
  }

  private void handleTransportInUseState(
      final ConnectionClientTransport transport, final boolean inUse) {
    syncContext.execute(new Runnable() {
      @Override
      public void run() {
        inUseStateAggregator.updateObjectInUse(transport, inUse);
      }
    });
  }

  void shutdownNow(final Status reason) {
    shutdown(reason);
    syncContext.execute(new Runnable() {
      @Override
      public void run() {
        Collection<ManagedClientTransport> transportsCopy =
            new ArrayList<ManagedClientTransport>(transports);

        for (ManagedClientTransport transport : transportsCopy) {
          transport.shutdownNow(reason);
        }
      }
    });
  }

  List<EquivalentAddressGroup> getAddressGroups() {
    return addressGroups;
  }

  private void cancelReconnectTask() {
    syncContext.throwIfNotInThisSynchronizationContext();

    if (reconnectTask != null) {
      reconnectTask.cancel();
      reconnectTask = null;
      reconnectPolicy = null;
    }
  }

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

  @Override
  public ListenableFuture<ChannelStats> getStats() {
    final SettableFuture<ChannelStats> channelStatsFuture = SettableFuture.create();
    syncContext.execute(new Runnable() {
      @Override
      public void run() {
        ChannelStats.Builder builder = new ChannelStats.Builder();
        List<EquivalentAddressGroup> addressGroupsSnapshot = addressIndex.getGroups();
        List<InternalWithLogId> transportsSnapshot = new ArrayList<InternalWithLogId>(transports);
        builder.setTarget(addressGroupsSnapshot.toString()).setState(getState());
        builder.setSockets(transportsSnapshot);
        callsTracer.updateBuilder(builder);
        channelTracer.updateBuilder(builder);
        channelStatsFuture.set(builder.build());
      }
    });
    return channelStatsFuture;
  }

  /**
   * Return attributes for server address connected by sub channel.
   */
  public Attributes getConnectedAddressAttributes() {
    return connectedAddressAttributes;
  }

  ConnectivityState getState() {
    return state.getState();
  }

  private static void checkListHasNoNulls(List<?> list, String msg) {
    for (Object item : list) {
      Preconditions.checkNotNull(item, msg);
    }
  }

  /** Listener for real transports. */
  private class TransportListener implements ManagedClientTransport.Listener {
    final ConnectionClientTransport transport;
    boolean shutdownInitiated = false;

    TransportListener(ConnectionClientTransport transport) {
      this.transport = transport;
    }

    @Override
    public Attributes filterTransport(Attributes attributes) {
      for (ClientTransportFilter filter : transportFilters) {
        attributes = Preconditions.checkNotNull(filter.transportReady(attributes),
            "Filter %s returned null", filter);
      }
      return attributes;
    }

    @Override
    public void transportReady() {
      channelLogger.log(ChannelLogLevel.INFO, "READY");
      syncContext.execute(new Runnable() {
        @Override
        public void run() {
          reconnectPolicy = null;
          if (shutdownReason != null) {
            // activeTransport should have already been set to null by shutdown(). We keep it null.
            Preconditions.checkState(activeTransport == null,
                "Unexpected non-null activeTransport");
            transport.shutdown(shutdownReason);
          } else if (pendingTransport == transport) {
            activeTransport = transport;
            pendingTransport = null;
            connectedAddressAttributes = addressIndex.getCurrentEagAttributes();
            gotoNonErrorState(READY);
            subchannelMetrics.recordConnectionAttemptSucceeded(/* target= */ target,
                /* backendService= */ getAttributeOrDefault(
                    addressIndex.getCurrentEagAttributes(), NameResolver.ATTR_BACKEND_SERVICE),
                /* locality= */ getAttributeOrDefault(addressIndex.getCurrentEagAttributes(),
                    EquivalentAddressGroup.ATTR_LOCALITY_NAME),
                /* securityLevel= */ extractSecurityLevel(addressIndex.getCurrentEagAttributes()
                    .get(GrpcAttributes.ATTR_SECURITY_LEVEL)));
          }
        }
      });
    }

    @Override
    public void transportInUse(boolean inUse) {
      handleTransportInUseState(transport, inUse);
    }

    @Override
    public void transportShutdown(final Status s, final DisconnectError disconnectError) {
      channelLogger.log(
          ChannelLogLevel.INFO, "{0} SHUTDOWN with {1}", transport.getLogId(), printShortStatus(s));
      shutdownInitiated = true;
      syncContext.execute(new Runnable() {
        @Override
        public void run() {
          if (state.getState() == SHUTDOWN) {
            return;
          }
          if (activeTransport == transport) {
            activeTransport = null;
            addressIndex.reset();
            gotoNonErrorState(IDLE);
            subchannelMetrics.recordDisconnection(/* target= */ target,
                /* backendService= */ getAttributeOrDefault(addressIndex.getCurrentEagAttributes(),
                    NameResolver.ATTR_BACKEND_SERVICE),
                /* locality= */ getAttributeOrDefault(addressIndex.getCurrentEagAttributes(),
                    EquivalentAddressGroup.ATTR_LOCALITY_NAME),
                /* disconnectError= */ disconnectError.toErrorString(),
                /* securityLevel= */ extractSecurityLevel(addressIndex.getCurrentEagAttributes()
                    .get(GrpcAttributes.ATTR_SECURITY_LEVEL)));
          } else if (pendingTransport == transport) {
            subchannelMetrics.recordConnectionAttemptFailed(/* target= */ target,
                /* backendService= */getAttributeOrDefault(addressIndex.getCurrentEagAttributes(),
                    NameResolver.ATTR_BACKEND_SERVICE),
                /* locality= */ getAttributeOrDefault(addressIndex.getCurrentEagAttributes(),
                    EquivalentAddressGroup.ATTR_LOCALITY_NAME));
            Preconditions.checkState(state.getState() == CONNECTING,
                "Expected state is CONNECTING, actual state is %s", state.getState());
            addressIndex.increment();
            // Continue to reconnect if there are still addresses to try.
            if (!addressIndex.isValid()) {
              pendingTransport = null;
              addressIndex.reset();
              // Initiate backoff
              // Transition to TRANSIENT_FAILURE
              scheduleBackoff(s);
            } else {
              startNewTransport();
            }
          }
        }
      });
    }

    @Override
    public void transportTerminated() {
      Preconditions.checkState(
          shutdownInitiated, "transportShutdown() must be called before transportTerminated().");

      channelLogger.log(ChannelLogLevel.INFO, "{0} Terminated", transport.getLogId());
      channelz.removeClientSocket(transport);
      handleTransportInUseState(transport, false);
      for (ClientTransportFilter filter : transportFilters) {
        filter.transportTerminated(transport.getAttributes());
      }
      syncContext.execute(new Runnable() {
        @Override
        public void run() {
          transports.remove(transport);
          if (state.getState() == SHUTDOWN && transports.isEmpty()) {
            handleTermination();
          }
        }
      });
    }

    private String extractSecurityLevel(SecurityLevel securityLevel) {
      if (securityLevel == null) {
        return "none";
      }
      switch (securityLevel) {
        case NONE:
          return "none";
        case INTEGRITY:
          return "integrity_only";
        case PRIVACY_AND_INTEGRITY:
          return "privacy_and_integrity";
        default:
          throw new IllegalArgumentException("Unknown SecurityLevel: " + securityLevel);
      }
    }

    private String getAttributeOrDefault(Attributes attributes, Attributes.Key<String> key) {
      String value = attributes.get(key);
      return value == null ? "" : value;
    }
  }

  // All methods are called in syncContext
  abstract static class Callback {
    /**
     * Called when the subchannel is terminated, which means it's shut down and all transports
     * have been terminated.
     */
    @ForOverride
    void onTerminated(InternalSubchannel is) { }

    /**
     * Called when the subchannel's connectivity state has changed.
     */
    @ForOverride
    void onStateChange(InternalSubchannel is, ConnectivityStateInfo newState) { }

    /**
     * Called when the subchannel's in-use state has changed to true, which means at least one
     * transport is in use.
     */
    @ForOverride
    void onInUse(InternalSubchannel is) { }

    /**
     * Called when the subchannel's in-use state has changed to false, which means no transport is
     * in use.
     */
    @ForOverride
    void onNotInUse(InternalSubchannel is) { }
  }

  @VisibleForTesting
  static final class CallTracingTransport extends ForwardingConnectionClientTransport {
    private final ConnectionClientTransport delegate;
    private final CallTracer callTracer;

    private CallTracingTransport(ConnectionClientTransport delegate, CallTracer callTracer) {
      this.delegate = delegate;
      this.callTracer = callTracer;
    }

    @Override
    protected ConnectionClientTransport delegate() {
      return delegate;
    }

    @Override
    public ClientStream newStream(
        MethodDescriptor<?, ?> method, Metadata headers, CallOptions callOptions,
        ClientStreamTracer[] tracers) {
      final ClientStream streamDelegate = super.newStream(method, headers, callOptions, tracers);
      return new ForwardingClientStream() {
        @Override
        protected ClientStream delegate() {
          return streamDelegate;
        }

        @Override
        public void start(final ClientStreamListener listener) {
          callTracer.reportCallStarted();
          super.start(new ForwardingClientStreamListener() {
            @Override
            protected ClientStreamListener delegate() {
              return listener;
            }

            @Override
            public void closed(
                Status status, RpcProgress rpcProgress, Metadata trailers) {
              callTracer.reportCallEnded(status.isOk());
              super.closed(status, rpcProgress, trailers);
            }
          });
        }
      };
    }
  }

  /** Index as in 'i', the pointer to an entry. Not a "search index." */
  @VisibleForTesting
  static final class Index {
    private List<EquivalentAddressGroup> addressGroups;
    private int groupIndex;
    private int addressIndex;

    public Index(List<EquivalentAddressGroup> groups) {
      this.addressGroups = groups;
    }

    public boolean isValid() {
      // addressIndex will never be invalid
      return groupIndex < addressGroups.size();
    }

    public boolean isAtBeginning() {
      return groupIndex == 0 && addressIndex == 0;
    }

    public void increment() {
      EquivalentAddressGroup group = addressGroups.get(groupIndex);
      addressIndex++;
      if (addressIndex >= group.getAddresses().size()) {
        groupIndex++;
        addressIndex = 0;
      }
    }

    public void reset() {
      groupIndex = 0;
      addressIndex = 0;
    }

    public SocketAddress getCurrentAddress() {
      return addressGroups.get(groupIndex).getAddresses().get(addressIndex);
    }

    public Attributes getCurrentEagAttributes() {
      return addressGroups.get(groupIndex).getAttributes();
    }

    public List<EquivalentAddressGroup> getGroups() {
      return addressGroups;
    }

    /** Update to new groups, resetting the current index. */
    public void updateGroups(List<EquivalentAddressGroup> newGroups) {
      addressGroups = newGroups;
      reset();
    }

    /** Returns false if the needle was not found and the current index was left unchanged. */
    public boolean seekTo(SocketAddress needle) {
      for (int i = 0; i < addressGroups.size(); i++) {
        EquivalentAddressGroup group = addressGroups.get(i);
        int j = group.getAddresses().indexOf(needle);
        if (j == -1) {
          continue;
        }
        this.groupIndex = i;
        this.addressIndex = j;
        return true;
      }
      return false;
    }
  }

  private String printShortStatus(Status status) {
    StringBuilder buffer = new StringBuilder();
    buffer.append(status.getCode());
    if (status.getDescription() != null) {
      buffer.append("(").append(status.getDescription()).append(")");
    }
    if (status.getCause() != null) {
      buffer.append("[").append(status.getCause()).append("]");
    }
    return buffer.toString();
  }

  @VisibleForTesting
  static final class TransportLogger extends ChannelLogger {
    // Changed just after construction to break a cyclic dependency.
    InternalLogId logId;

    @Override
    public void log(ChannelLogLevel level, String message) {
      ChannelLoggerImpl.logOnly(logId, level, message);
    }

    @Override
    public void log(ChannelLogLevel level, String messageFormat, Object... args) {
      ChannelLoggerImpl.logOnly(logId, level, messageFormat, args);
    }
  }
}

grpc / grpc-java / #20136

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

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