DelayedClientTransport.java

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

package io.grpc.internal;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.base.Supplier;
import com.google.common.base.Suppliers;
import io.grpc.CallOptions;
import io.grpc.Context;
import io.grpc.LoadBalancer2;
import io.grpc.LoadBalancer2.PickResult;
import io.grpc.LoadBalancer2.Subchannel;
import io.grpc.Metadata;
import io.grpc.MethodDescriptor;
import io.grpc.Status;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Iterator;
import java.util.LinkedHashSet;
import java.util.concurrent.Executor;
import javax.annotation.Nullable;
import javax.annotation.concurrent.GuardedBy;

/**
 * A client transport that queues requests before a real transport is available. When a backing
 * transport supplier is later provided, this class delegates to the transports from it.
 *
 * <p>This transport owns the streams that it has created before {@link #setTransport} is
 * called. When {@link #setTransport} is called, the ownership of pending streams and subsequent new
 * streams are transferred to the given transport, thus this transport won't own any stream.
 */
class DelayedClientTransport implements ManagedClientTransport {

  private final LogId lodId = LogId.allocate(getClass().getName());

  private final Object lock = new Object();

  private final Executor streamCreationExecutor;

  private Listener listener;
  /** 'lock' must be held when assigning to transportSupplier. */
  private volatile Supplier<ClientTransport> transportSupplier;

  @GuardedBy("lock")
  private Collection<PendingStream> pendingStreams = new LinkedHashSet<PendingStream>();

  // TODO(zhangkun83): remove it once LBv2 refactor is done.  In practice ping() is only called on
  // real transports.
  @GuardedBy("lock")
  private Collection<PendingPing> pendingPings = new ArrayList<PendingPing>();

  /**
   * When shutdown == true and pendingStreams == null, then the transport is considered terminated.
   */
  @GuardedBy("lock")
  private boolean shutdown;

  /**
   * The delayed client transport will come into a back-off interval if it fails to establish a real
   * transport for all addresses, namely the channel is in TRANSIENT_FAILURE. When in a back-off
   * interval, {@code backoffStatus != null}.
   *
   * <p>If the transport is in a back-off interval, then all fail fast streams (including the
   * pending as well as new ones) will fail immediately. New non-fail fast streams can be created as
   * {@link PendingStream} and will keep pending during this back-off period.
   */
  @GuardedBy("lock")
  @Nullable
  private Status backoffStatus;

  DelayedClientTransport(Executor streamCreationExecutor) {
    this.streamCreationExecutor = streamCreationExecutor;
  }

  @Override
  public final Runnable start(Listener listener) {
    this.listener = Preconditions.checkNotNull(listener, "listener");
    return null;
  }

  /**
   * If the transport has acquired a transport {@link Supplier}, then returned stream is delegated
   * from its supplier.
   *
   * <p>If the new stream to be created is with fail fast call option and the delayed transport is
   * in a back-off interval, then a {@link FailingClientStream} is returned.
   *
   * <p>If it is not the above cases and the delayed transport is not shutdown, then a
   * {@link PendingStream} is returned; if the transport is shutdown, then a
   * {@link FailingClientStream} is returned.
   */
  @Override
  public final ClientStream newStream(MethodDescriptor<?, ?> method, Metadata headers,
      CallOptions callOptions, StatsTraceContext statsTraceCtx) {
    Supplier<ClientTransport> supplier = transportSupplier;
    if (supplier == null) {
      synchronized (lock) {
        // Check again, since it may have changed while waiting for lock
        supplier = transportSupplier;
        if (supplier == null && !shutdown) {
          if (backoffStatus != null && !callOptions.isWaitForReady()) {
            return new FailingClientStream(backoffStatus);
          }
          PendingStream pendingStream = new PendingStream(method, headers, callOptions,
              statsTraceCtx);
          pendingStreams.add(pendingStream);
          if (pendingStreams.size() == 1) {
            listener.transportInUse(true);
          }
          return pendingStream;
        }
      }
    }
    if (supplier != null) {
      return supplier.get().newStream(method, headers, callOptions, statsTraceCtx);
    }
    return new FailingClientStream(Status.UNAVAILABLE.withDescription("transport shutdown"));
  }

  @Override
  public final ClientStream newStream(MethodDescriptor<?, ?> method, Metadata headers) {
    return newStream(method, headers, CallOptions.DEFAULT, StatsTraceContext.NOOP);
  }

  @Override
  public final void ping(final PingCallback callback, Executor executor) {
    Supplier<ClientTransport> supplier = transportSupplier;
    if (supplier == null) {
      synchronized (lock) {
        // Check again, since it may have changed while waiting for lock
        supplier = transportSupplier;
        if (supplier == null && !shutdown) {
          PendingPing pendingPing = new PendingPing(callback, executor);
          pendingPings.add(pendingPing);
          return;
        }
      }
    }
    if (supplier != null) {
      supplier.get().ping(callback, executor);
      return;
    }
    executor.execute(new Runnable() {
        @Override public void run() {
          callback.onFailure(
              Status.UNAVAILABLE.withDescription("transport shutdown").asException());
        }
      });
  }

  /**
   * Prevents creating any new streams until {@link #setTransport} is called. Buffered streams are
   * not failed, so if {@link #shutdown} is called when {@link #setTransport} has not been called,
   * you still need to call {@link #setTransport} to make this transport terminated.
   */
  @Override
  public final void shutdown() {
    synchronized (lock) {
      if (shutdown) {
        return;
      }
      shutdown = true;
      listener.transportShutdown(
          Status.UNAVAILABLE.withDescription("Channel requested transport to shut down"));
      if (pendingStreams == null || pendingStreams.isEmpty()) {
        pendingStreams = null;
        listener.transportTerminated();
      }
    }
  }

  /**
   * Shuts down this transport and cancels all streams that it owns, hence immediately terminates
   * this transport.
   */
  @Override
  public final void shutdownNow(Status status) {
    shutdown();
    Collection<PendingStream> savedPendingStreams = null;
    synchronized (lock) {
      if (pendingStreams != null) {
        savedPendingStreams = pendingStreams;
        pendingStreams = null;
      }
    }
    if (savedPendingStreams != null) {
      for (PendingStream stream : savedPendingStreams) {
        stream.cancel(status);
      }
      listener.transportTerminated();
    }
    // If savedPendingStreams == null, transportTerminated() has already been called in shutdown().
  }

  /**
   * Transfers all the pending and future streams and pings to the given transport.
   *
   * <p>May only be called after {@link #start(Listener)}.
   *
   * <p>{@code transport} will be used for all future calls to {@link #newStream}, even if this
   * transport is {@link #shutdown}.
   */
  public final void setTransport(ClientTransport transport) {
    Preconditions.checkArgument(this != transport,
        "delayed transport calling setTransport on itself");
    setTransportSupplier(Suppliers.ofInstance(transport));
  }

  /**
   * Transfers all the pending and future streams and pings to the transports from the given {@link
   * Supplier}.
   *
   * <p>May only be called after {@link #start}. No effect if already called.
   *
   * <p>Each stream or ping will result in an invocation to {@link Supplier#get} once. The supplier
   * will be used for all future calls to {@link #newStream}, even if this transport is {@link
   * #shutdown}.
   */
  public final void setTransportSupplier(final Supplier<ClientTransport> supplier) {
    synchronized (lock) {
      if (transportSupplier != null) {
        return;
      }
      Preconditions.checkState(listener != null, "start() not called");
      transportSupplier = Preconditions.checkNotNull(supplier, "supplier");
      for (PendingPing ping : pendingPings) {
        ping.createRealPing(supplier.get());
      }
      pendingPings = null;
      if (shutdown && pendingStreams != null) {
        listener.transportTerminated();
      }
      if (pendingStreams != null && !pendingStreams.isEmpty()) {
        final Collection<PendingStream> savedPendingStreams = pendingStreams;
        // createRealStream may be expensive. It will start real streams on the transport. If there
        // are pending requests, they will be serialized too, which may be expensive. Since we are
        // now on transport thread, we need to offload the work to an executor.
        streamCreationExecutor.execute(new Runnable() {
            @Override public void run() {
              for (final PendingStream stream : savedPendingStreams) {
                stream.createRealStream(supplier.get());
              }
              // TODO(zhangkun83): some transports (e.g., netty) may have a short delay between
              // stream.start() and transportInUse(true). If netty's transportInUse(true) is called
              // after the delayed transport's transportInUse(false), the channel may have a brief
              // period where all transports are not in-use, and may go to IDLE mode unexpectedly if
              // the IDLE timeout is shorter (e.g., 0) than that brief period. Maybe we should
              // have a minimum IDLE timeout?
              synchronized (lock) {
                listener.transportInUse(false);
              }
            }
        });
      }
      pendingStreams = null;
      if (!shutdown) {
        listener.transportReady();
      }
    }
  }

  public final boolean hasPendingStreams() {
    synchronized (lock) {
      return pendingStreams != null && !pendingStreams.isEmpty();
    }
  }

  @VisibleForTesting
  final int getPendingStreamsCount() {
    synchronized (lock) {
      return pendingStreams == null ? 0 : pendingStreams.size();
    }
  }

  /**
   * True return value indicates that the delayed transport is in a back-off interval (in
   * TRANSIENT_FAILURE), that all fail fast streams (including pending as well as new ones) should
   * fail immediately, and that non-fail fast streams can be created as {@link PendingStream} and
   * should keep pending during this back-off period.
   */
  // TODO(zhangkun83): remove it once the LBv2 refactor is done.
  @VisibleForTesting
  final boolean isInBackoffPeriod() {
    synchronized (lock) {
      return backoffStatus != null;
    }
  }

  /**
   * Is only called at the beginning of {@link TransportSet#scheduleBackoff}.
   *
   * <p>Does jobs at the beginning of the back-off:
   *
   * <p>sets {@link #backoffStatus};
   *
   * <p>sets all pending streams with a fail fast call option of the delayed transport as
   * {@link FailingClientStream}s, and removes them from the list of pending streams of the
   * transport.
   *
   * @param status the causal status for triggering back-off.
   */
  // TODO(zhangkun83): remove it once the LBv2 refactor is done.
  final void startBackoff(final Status status) {
    synchronized (lock) {
      if (shutdown) {
        return;
      }
      Preconditions.checkState(backoffStatus == null,
          "Error when calling startBackoff: transport is already in backoff period");
      backoffStatus = Status.UNAVAILABLE.withDescription("Channel in TRANSIENT_FAILURE state")
          .withCause(status.asRuntimeException());
      final ArrayList<PendingStream> failFastPendingStreams = new ArrayList<PendingStream>();
      if (pendingStreams != null && !pendingStreams.isEmpty()) {
        final Iterator<PendingStream> it = pendingStreams.iterator();
        while (it.hasNext()) {
          PendingStream stream = it.next();
          if (!stream.callOptions.isWaitForReady()) {
            failFastPendingStreams.add(stream);
            it.remove();
          }
        }

        class FailTheFailFastPendingStreams implements Runnable {
          @Override
          public void run() {
            for (PendingStream stream : failFastPendingStreams) {
              stream.setStream(new FailingClientStream(status));
            }
          }
        }

        streamCreationExecutor.execute(new FailTheFailFastPendingStreams());
      }
    }
  }

  /**
   * Is only called at the beginning of the callback function of {@code endOfCurrentBackoff} in the
   * {@link TransportSet#scheduleBackoff} method.
   */
  // TODO(zhangkun83): remove it once the LBv2 refactor is done.
  final void endBackoff() {
    synchronized (lock) {
      Preconditions.checkState(backoffStatus != null,
          "Error when calling endBackoff: transport is not in backoff period");
      backoffStatus = null;
    }
  }

  /**
   * Use the picker to try picking a transport for every pending stream and pending ping, proceed
   * the stream or ping if the pick is successful, otherwise keep it pending.
   *
   * <p>If new pending streams are created while reprocess() is in progress, there is no guarantee
   * that the pending streams will or will not be processed by this picker.
   *
   * <p>This method <strong>must not</strong> be called concurrently, either with itself or with
   * {@link #setTransportSupplier}/{@link #setTransport}.
   */
  final void reprocess(LoadBalancer2.SubchannelPicker picker) {
    ArrayList<PendingStream> toProcess;
    ArrayList<PendingStream> toRemove = new ArrayList<PendingStream>();
    synchronized (lock) {
      if (pendingStreams == null || pendingStreams.isEmpty()) {
        return;
      }
      toProcess = new ArrayList<PendingStream>(pendingStreams);
    }

    for (final PendingStream stream : toProcess) {
      PickResult pickResult = picker.pickSubchannel(
          stream.callOptions.getAffinity(), stream.headers);
      final ClientTransport realTransport;
      Subchannel subchannel = pickResult.getSubchannel();
      if (subchannel != null) {
        realTransport = ((SubchannelImpl) subchannel).obtainActiveTransport();
      } else {
        realTransport = null;
      }
      if (realTransport != null) {
        Executor executor = streamCreationExecutor;
        // createRealStream may be expensive. It will start real streams on the transport. If
        // there are pending requests, they will be serialized too, which may be expensive. Since
        // we are now on transport thread, we need to offload the work to an executor.
        if (stream.callOptions.getExecutor() != null) {
          executor = stream.callOptions.getExecutor();
        }
        executor.execute(new Runnable() {
            @Override
            public void run() {
              stream.createRealStream(realTransport);
            }
          });
        toRemove.add(stream);
      } else if (!pickResult.getStatus().isOk() && !stream.callOptions.isWaitForReady()) {
        stream.setStream(new FailingClientStream(pickResult.getStatus()));
        toRemove.add(stream);
      }  // other cases: stay pending
    }

    synchronized (lock) {
      // Between this synchronized and the previous one:
      //   - Streams may have been cancelled, which may turn pendingStreams into emptiness.
      //   - shutdown() may be called, which may turn pendingStreams into null.
      if (pendingStreams == null || pendingStreams.isEmpty()) {
        return;
      }
      pendingStreams.removeAll(toRemove);
      if (pendingStreams.isEmpty()) {
        // There may be a brief gap between delayed transport clearing in-use state, and first real
        // transport starting streams and setting in-use state.  During the gap the whole channel's
        // in-use state may be false. However, it shouldn't cause spurious switching to idleness
        // (which would shutdown the transports and LoadBalancer) because the gap should be shorter
        // than IDLE_MODE_DEFAULT_TIMEOUT_MILLIS (1 second).
        listener.transportInUse(false);
        if (shutdown) {
          pendingStreams = null;
          listener.transportTerminated();
        } else {
          // Because delayed transport is long-lived, we take this opportunity to down-size the
          // hashmap.
          pendingStreams = new LinkedHashSet<PendingStream>();
        }
      }
    }
  }

  // TODO(carl-mastrangelo): remove this once the Subchannel change is in.
  @Override
  public LogId getLogId() {
    return lodId;
  }

  @VisibleForTesting
  @Nullable
  final Supplier<ClientTransport> getTransportSupplier() {
    return transportSupplier;
  }

  private class PendingStream extends DelayedStream {
    private final MethodDescriptor<?, ?> method;
    private final Metadata headers;
    private final CallOptions callOptions;
    private final Context context;
    private final StatsTraceContext statsTraceCtx;

    private PendingStream(MethodDescriptor<?, ?> method, Metadata headers,
        CallOptions callOptions, StatsTraceContext statsTraceCtx) {
      this.method = method;
      this.headers = headers;
      this.callOptions = callOptions;
      this.context = Context.current();
      this.statsTraceCtx = statsTraceCtx;
    }

    private void createRealStream(ClientTransport transport) {
      ClientStream realStream;
      Context origContext = context.attach();
      try {
        realStream = transport.newStream(method, headers, callOptions, statsTraceCtx);
      } finally {
        context.detach(origContext);
      }
      setStream(realStream);
    }

    @Override
    public void cancel(Status reason) {
      super.cancel(reason);
      synchronized (lock) {
        if (pendingStreams != null) {
          boolean justRemovedAnElement = pendingStreams.remove(this);
          if (pendingStreams.isEmpty() && justRemovedAnElement) {
            listener.transportInUse(false);
            if (shutdown) {
              pendingStreams = null;
              listener.transportTerminated();
            }
          }
        }
      }
    }
  }

  private static class PendingPing {
    private final PingCallback callback;
    private final Executor executor;

    public PendingPing(PingCallback callback, Executor executor) {
      this.callback = callback;
      this.executor = executor;
    }

    public void createRealPing(ClientTransport transport) {
      try {
        transport.ping(callback, executor);
      } catch (final UnsupportedOperationException ex) {
        executor.execute(new Runnable() {
          @Override
          public void run() {
            callback.onFailure(ex);
          }
        });
      }
    }
  }
}