| /* |
| * 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); |
| } |
| }); |
| } |
| } |
| } |
| } |