#19691

Committed 14 Feb 2025 06:23PM CUT coverage: 88.603% (-0.02%) from 88.626%

Build # #19691

Build Type

push

github

Committed by

web-flow

Commit Message

xds:Cleanup to reduce test flakiness (#11895)

* don't process resourceDoesNotExist for watchers that have been cancelled.

* Change test to use an ArgumentMatcher instead of expecting that only the final result will be sent since depending on timing there may be configs sent for clusters being removed with their entries as errors.

Run Details

34261 of 38668 relevant lines covered (88.6%)

0.89 hits per line

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

96.51

/../core/src/main/java/io/grpc/internal/DelayedClientTransport.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 com.google.common.annotations.VisibleForTesting;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.SettableFuture;
import com.google.errorprone.annotations.concurrent.GuardedBy;
import io.grpc.CallOptions;
import io.grpc.ClientStreamTracer;
import io.grpc.Context;
import io.grpc.InternalChannelz.SocketStats;
import io.grpc.InternalLogId;
import io.grpc.LoadBalancer.PickResult;
import io.grpc.LoadBalancer.PickSubchannelArgs;
import io.grpc.LoadBalancer.SubchannelPicker;
import io.grpc.Metadata;
import io.grpc.MethodDescriptor;
import io.grpc.Status;
import io.grpc.SynchronizationContext;
import io.grpc.internal.ClientStreamListener.RpcProgress;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.LinkedHashSet;
import java.util.concurrent.Executor;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;

/**
 * A client transport that queues requests before a real transport is available. When {@link
 * #reprocess} is called, this class applies the provided {@link SubchannelPicker} to pick a
 * transport for each pending stream.
 *
 * <p>This transport owns every stream that it has created until a real transport has been picked
 * for that stream, at which point the ownership of the stream is transferred to the real transport,
 * thus the delayed transport stops owning the stream.
 */
final class DelayedClientTransport implements ManagedClientTransport {
  // lazily allocated, since it is infrequently used.
  private final InternalLogId logId =
      InternalLogId.allocate(DelayedClientTransport.class, /*details=*/ null);

  private final Object lock = new Object();

  private final Executor defaultAppExecutor;
  private final SynchronizationContext syncContext;

  private Runnable reportTransportInUse;
  private Runnable reportTransportNotInUse;
  private Runnable reportTransportTerminated;
  private Listener listener;

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

  /** Immutable state needed for picking. 'lock' must be held for writing. */
  private volatile PickerState pickerState = new PickerState(null, null);

  /**
   * Creates a new delayed transport.
   *
   * @param defaultAppExecutor pending streams will create real streams and run buffered operations
   *        in an application executor, which will be this executor, unless there is on provided in
   *        {@link CallOptions}.
   * @param syncContext all listener callbacks of the delayed transport will be run from this
   *        SynchronizationContext.
   */
  DelayedClientTransport(Executor defaultAppExecutor, SynchronizationContext syncContext) {
    this.defaultAppExecutor = defaultAppExecutor;
    this.syncContext = syncContext;
  }

  @Override
  public final Runnable start(final Listener listener) {
    this.listener = listener;
    reportTransportInUse = new Runnable() {
        @Override
        public void run() {
          listener.transportInUse(true);
        }
      };
    reportTransportNotInUse = new Runnable() {
        @Override
        public void run() {
          listener.transportInUse(false);
        }
      };
    reportTransportTerminated = new Runnable() {
        @Override
        public void run() {
          listener.transportTerminated();
        }
      };
    return null;
  }

  /**
   * If a {@link SubchannelPicker} is being, or has been provided via {@link #reprocess}, the last
   * picker will be consulted.
   *
   * <p>Otherwise, if 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,
      ClientStreamTracer[] tracers) {
    try {
      PickSubchannelArgs args = new PickSubchannelArgsImpl(
          method, headers, callOptions, new PickDetailsConsumerImpl(tracers));
      PickerState state = pickerState;
      while (true) {
        if (state.shutdownStatus != null) {
          return new FailingClientStream(state.shutdownStatus, tracers);
        }
        PickResult pickResult = null;
        if (state.lastPicker != null) {
          pickResult = state.lastPicker.pickSubchannel(args);
          callOptions = args.getCallOptions();
          // User code provided authority takes precedence over the LB provided one.
          if (callOptions.getAuthority() == null
              && pickResult.getAuthorityOverride() != null) {
            callOptions = callOptions.withAuthority(pickResult.getAuthorityOverride());
          }
          ClientTransport transport = GrpcUtil.getTransportFromPickResult(pickResult,
              callOptions.isWaitForReady());
          if (transport != null) {
            ClientStream stream = transport.newStream(
                args.getMethodDescriptor(), args.getHeaders(), callOptions,
                tracers);
            // User code provided authority takes precedence over the LB provided one; this will be
            // overwritten by ClientCallImpl if the application sets an authority override
            if (pickResult.getAuthorityOverride() != null) {
              stream.setAuthority(pickResult.getAuthorityOverride());
            }
            return stream;
          }
        }
        // This picker's conclusion is "buffer".  If there hasn't been a newer picker set (possible
        // race with reprocess()), we will buffer the RPC.  Otherwise, will try with the new picker.
        synchronized (lock) {
          PickerState newerState = pickerState;
          if (state == newerState) {
            return createPendingStream(args, tracers, pickResult);
          }
          state = newerState;
        }
      }
    } finally {
      syncContext.drain();
    }
  }

  /**
   * Caller must call {@code syncContext.drain()} outside of lock because this method may
   * schedule tasks on syncContext.
   */
  @GuardedBy("lock")
  private PendingStream createPendingStream(PickSubchannelArgs args, ClientStreamTracer[] tracers,
      PickResult pickResult) {
    PendingStream pendingStream = new PendingStream(args, tracers);
    if (args.getCallOptions().isWaitForReady() && pickResult != null && pickResult.hasResult()) {
      pendingStream.lastPickStatus = pickResult.getStatus();
    }
    pendingStreams.add(pendingStream);
    if (getPendingStreamsCount() == 1) {
      syncContext.executeLater(reportTransportInUse);
    }
    for (ClientStreamTracer streamTracer : tracers) {
      streamTracer.createPendingStream();
    }
    return pendingStream;
  }

  @Override
  public final void ping(final PingCallback callback, Executor executor) {
    throw new UnsupportedOperationException("This method is not expected to be called");
  }

  @Override
  public ListenableFuture<SocketStats> getStats() {
    SettableFuture<SocketStats> ret = SettableFuture.create();
    ret.set(null);
    return ret;
  }

  /**
   * Prevents creating any new streams.  Buffered streams are not failed and may still proceed
   * when {@link #reprocess} is called.  The delayed transport will be terminated when there is no
   * more buffered streams.
   */
  @Override
  public final void shutdown(final Status status) {
    synchronized (lock) {
      if (pickerState.shutdownStatus != null) {
        return;
      }
      pickerState = pickerState.withShutdownStatus(status);
      syncContext.executeLater(new Runnable() {
          @Override
          public void run() {
            listener.transportShutdown(status);
          }
        });
      if (!hasPendingStreams() && reportTransportTerminated != null) {
        syncContext.executeLater(reportTransportTerminated);
        reportTransportTerminated = null;
      }
    }
    syncContext.drain();
  }

  /**
   * Shuts down this transport and cancels all streams that it owns, hence immediately terminates
   * this transport.
   */
  @Override
  public final void shutdownNow(Status status) {
    shutdown(status);
    Collection<PendingStream> savedPendingStreams;
    Runnable savedReportTransportTerminated;
    synchronized (lock) {
      savedPendingStreams = pendingStreams;
      savedReportTransportTerminated = reportTransportTerminated;
      reportTransportTerminated = null;
      if (!pendingStreams.isEmpty()) {
        pendingStreams = Collections.emptyList();
      }
    }
    if (savedReportTransportTerminated != null) {
      for (PendingStream stream : savedPendingStreams) {
        Runnable runnable = stream.setStream(
            new FailingClientStream(status, RpcProgress.REFUSED, stream.tracers));
        if (runnable != null) {
          // Drain in-line instead of using an executor as failing stream just throws everything
          // away. This is essentially the same behavior as DelayedStream.cancel() but can be done
          // before stream.start().
          runnable.run();
        }
      }
      syncContext.execute(savedReportTransportTerminated);
    }
    // If savedReportTransportTerminated == null, transportTerminated() has already been called in
    // shutdown().
  }

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

  @VisibleForTesting
  final int getPendingStreamsCount() {
    synchronized (lock) {
      return pendingStreams.size();
    }
  }

  /**
   * Use the picker to try picking a transport for every pending stream, proceed the stream if the
   * pick is successful, otherwise keep it pending.
   *
   * <p>This method may be called concurrently with {@code newStream()}, and it's safe.  All pending
   * streams will be served by the latest picker (if a same picker is given more than once, they are
   * considered different pickers) as soon as possible.
   *
   * <p>This method <strong>must not</strong> be called concurrently with itself.
   */
  final void reprocess(@Nullable SubchannelPicker picker) {
    ArrayList<PendingStream> toProcess;
    synchronized (lock) {
      pickerState = pickerState.withPicker(picker);
      if (picker == null || !hasPendingStreams()) {
        return;
      }
      toProcess = new ArrayList<>(pendingStreams);
    }
    ArrayList<PendingStream> toRemove = new ArrayList<>();

    for (final PendingStream stream : toProcess) {
      PickResult pickResult = picker.pickSubchannel(stream.args);
      CallOptions callOptions = stream.args.getCallOptions();
      if (callOptions.isWaitForReady() && pickResult.hasResult()) {
        stream.lastPickStatus = pickResult.getStatus();
      }
      final ClientTransport transport = GrpcUtil.getTransportFromPickResult(pickResult,
          callOptions.isWaitForReady());
      if (transport != null) {
        Executor executor = defaultAppExecutor;
        // 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 (callOptions.getExecutor() != null) {
          executor = callOptions.getExecutor();
        }
        Runnable runnable = stream.createRealStream(transport, pickResult.getAuthorityOverride());
        if (runnable != null) {
          executor.execute(runnable);
        }
        toRemove.add(stream);
      }  // else: 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 (!hasPendingStreams()) {
        return;
      }
      pendingStreams.removeAll(toRemove);
      // Because delayed transport is long-lived, we take this opportunity to down-size the
      // hashmap.
      if (pendingStreams.isEmpty()) {
        pendingStreams = new LinkedHashSet<>();
      }
      if (!hasPendingStreams()) {
        // 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).
        syncContext.executeLater(reportTransportNotInUse);
        if (pickerState.shutdownStatus != null && reportTransportTerminated != null) {
          syncContext.executeLater(reportTransportTerminated);
          reportTransportTerminated = null;
        }
      }
    }
    syncContext.drain();
  }

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

  private class PendingStream extends DelayedStream {
    private final PickSubchannelArgs args;
    private final Context context = Context.current();
    private final ClientStreamTracer[] tracers;
    private volatile Status lastPickStatus;

    private PendingStream(PickSubchannelArgs args, ClientStreamTracer[] tracers) {
      this.args = args;
      this.tracers = tracers;
    }

    /** Runnable may be null. */
    private Runnable createRealStream(ClientTransport transport, String authorityOverride) {
      ClientStream realStream;
      Context origContext = context.attach();
      try {
        realStream = transport.newStream(
            args.getMethodDescriptor(), args.getHeaders(), args.getCallOptions(),
            tracers);
      } finally {
        context.detach(origContext);
      }
      if (authorityOverride != null) {
        // User code provided authority takes precedence over the LB provided one; this will be
        // overwritten by an enqueud call from ClientCallImpl if the application sets an authority
        // override. We must call the real stream directly because stream.start() has likely already
        // been called on the delayed stream.
        realStream.setAuthority(authorityOverride);
      }
      return setStream(realStream);
    }

    @Override
    public void cancel(Status reason) {
      super.cancel(reason);
      synchronized (lock) {
        if (reportTransportTerminated != null) {
          boolean justRemovedAnElement = pendingStreams.remove(this);
          if (!hasPendingStreams() && justRemovedAnElement) {
            syncContext.executeLater(reportTransportNotInUse);
            if (pickerState.shutdownStatus != null) {
              syncContext.executeLater(reportTransportTerminated);
              reportTransportTerminated = null;
            }
          }
        }
      }
      syncContext.drain();
    }

    @Override
    protected void onEarlyCancellation(Status reason) {
      for (ClientStreamTracer tracer : tracers) {
        tracer.streamClosed(reason);
      }
    }

    @Override
    public void appendTimeoutInsight(InsightBuilder insight) {
      if (args.getCallOptions().isWaitForReady()) {
        insight.append("wait_for_ready");
        Status status = lastPickStatus;
        if (status != null && !status.isOk()) {
          insight.appendKeyValue("Last Pick Failure", status);
        }
      }
      super.appendTimeoutInsight(insight);
    }
  }

  static final class PickerState {
    /**
     * The last picker that {@link #reprocess} has used. May be set to null when the channel has
     * moved to idle.
     */
    @Nullable
    final SubchannelPicker lastPicker;
    /**
     * When {@code shutdownStatus != null && !hasPendingStreams()}, then the transport is considered
     * terminated.
     */
    @Nullable
    final Status shutdownStatus;

    private PickerState(SubchannelPicker lastPicker, Status shutdownStatus) {
      this.lastPicker = lastPicker;
      this.shutdownStatus = shutdownStatus;
    }

    public PickerState withPicker(SubchannelPicker newPicker) {
      return new PickerState(newPicker, this.shutdownStatus);
    }

    public PickerState withShutdownStatus(Status newShutdownStatus) {
      return new PickerState(this.lastPicker, newShutdownStatus);
    }
  }
}

1	/*
2	* Copyright 2015 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 com.google.common.annotations.VisibleForTesting;
20	import com.google.common.util.concurrent.ListenableFuture;
21	import com.google.common.util.concurrent.SettableFuture;
22	import com.google.errorprone.annotations.concurrent.GuardedBy;
23	import io.grpc.CallOptions;
24	import io.grpc.ClientStreamTracer;
25	import io.grpc.Context;
26	import io.grpc.InternalChannelz.SocketStats;
27	import io.grpc.InternalLogId;
28	import io.grpc.LoadBalancer.PickResult;
29	import io.grpc.LoadBalancer.PickSubchannelArgs;
30	import io.grpc.LoadBalancer.SubchannelPicker;
31	import io.grpc.Metadata;
32	import io.grpc.MethodDescriptor;
33	import io.grpc.Status;
34	import io.grpc.SynchronizationContext;
35	import io.grpc.internal.ClientStreamListener.RpcProgress;
36	import java.util.ArrayList;
37	import java.util.Collection;
38	import java.util.Collections;
39	import java.util.LinkedHashSet;
40	import java.util.concurrent.Executor;
41	import javax.annotation.Nonnull;
42	import javax.annotation.Nullable;
43
44	/**
45	* A client transport that queues requests before a real transport is available. When {@link
46	* #reprocess} is called, this class applies the provided {@link SubchannelPicker} to pick a
47	* transport for each pending stream.
48	*
49	* <p>This transport owns every stream that it has created until a real transport has been picked
50	* for that stream, at which point the ownership of the stream is transferred to the real transport,
51	* thus the delayed transport stops owning the stream.
52	*/
53	final class DelayedClientTransport implements ManagedClientTransport {
54	// lazily allocated, since it is infrequently used.
55	private final InternalLogId logId =	1✔
56	InternalLogId.allocate(DelayedClientTransport.class, /details=/ null);	1✔
57
58	private final Object lock = new Object();	1✔
59
60	private final Executor defaultAppExecutor;
61	private final SynchronizationContext syncContext;
62
63	private Runnable reportTransportInUse;
64	private Runnable reportTransportNotInUse;
65	private Runnable reportTransportTerminated;
66	private Listener listener;
67
68	@Nonnull	1✔
69	@GuardedBy("lock")
70	private Collection<PendingStream> pendingStreams = new LinkedHashSet<>();
71
72	/** Immutable state needed for picking. 'lock' must be held for writing. */
73	private volatile PickerState pickerState = new PickerState(null, null);	1✔
74
75	/**
76	* Creates a new delayed transport.
77	*
78	* @param defaultAppExecutor pending streams will create real streams and run buffered operations
79	* in an application executor, which will be this executor, unless there is on provided in
80	* {@link CallOptions}.
81	* @param syncContext all listener callbacks of the delayed transport will be run from this
82	* SynchronizationContext.
83	*/
84	DelayedClientTransport(Executor defaultAppExecutor, SynchronizationContext syncContext) {	1✔
85	this.defaultAppExecutor = defaultAppExecutor;	1✔
86	this.syncContext = syncContext;	1✔
87	}	1✔
88
89	@Override
90	public final Runnable start(final Listener listener) {
91	this.listener = listener;	1✔
92	reportTransportInUse = new Runnable() {	1✔
93	@Override
94	public void run() {
95	listener.transportInUse(true);	1✔
96	}	1✔
97	};
98	reportTransportNotInUse = new Runnable() {	1✔
99	@Override
100	public void run() {
101	listener.transportInUse(false);	1✔
102	}	1✔
103	};
104	reportTransportTerminated = new Runnable() {	1✔
105	@Override
106	public void run() {
107	listener.transportTerminated();	1✔
108	}	1✔
109	};
110	return null;	1✔
111	}
112
113	/**
114	* If a {@link SubchannelPicker} is being, or has been provided via {@link #reprocess}, the last
115	* picker will be consulted.
116	*
117	* <p>Otherwise, if the delayed transport is not shutdown, then a {@link PendingStream} is
118	* returned; if the transport is shutdown, then a {@link FailingClientStream} is returned.
119	*/
120	@Override
121	public final ClientStream newStream(
122	MethodDescriptor<?, ?> method, Metadata headers, CallOptions callOptions,
123	ClientStreamTracer[] tracers) {
124	try {
125	PickSubchannelArgs args = new PickSubchannelArgsImpl(	1✔
126	method, headers, callOptions, new PickDetailsConsumerImpl(tracers));
127	PickerState state = pickerState;	1✔
128	while (true) {
129	if (state.shutdownStatus != null) {	1✔
130	return new FailingClientStream(state.shutdownStatus, tracers);	1✔
131	}
132	PickResult pickResult = null;	1✔
133	if (state.lastPicker != null) {	1✔
134	pickResult = state.lastPicker.pickSubchannel(args);	1✔
135	callOptions = args.getCallOptions();	1✔
136	// User code provided authority takes precedence over the LB provided one.
137	if (callOptions.getAuthority() == null	1✔
138	&& pickResult.getAuthorityOverride() != null) {	1✔
139	callOptions = callOptions.withAuthority(pickResult.getAuthorityOverride());	1✔
140	}
141	ClientTransport transport = GrpcUtil.getTransportFromPickResult(pickResult,	1✔
142	callOptions.isWaitForReady());	1✔
143	if (transport != null) {	1✔
144	ClientStream stream = transport.newStream(	1✔
145	args.getMethodDescriptor(), args.getHeaders(), callOptions,	1✔
146	tracers);
147	// User code provided authority takes precedence over the LB provided one; this will be
148	// overwritten by ClientCallImpl if the application sets an authority override
149	if (pickResult.getAuthorityOverride() != null) {	1✔
150	stream.setAuthority(pickResult.getAuthorityOverride());	1✔
151	}
152	return stream;	1✔
153	}
154	}
155	// This picker's conclusion is "buffer". If there hasn't been a newer picker set (possible
156	// race with reprocess()), we will buffer the RPC. Otherwise, will try with the new picker.
157	synchronized (lock) {	1✔
158	PickerState newerState = pickerState;	1✔
159	if (state == newerState) {	1✔
160	return createPendingStream(args, tracers, pickResult);	1✔
161	}
162	state = newerState;	1✔
163	}	1✔
164	}	1✔
165	} finally {
166	syncContext.drain();	1✔
167	}
168	}
169
170	/**
171	* Caller must call {@code syncContext.drain()} outside of lock because this method may
172	* schedule tasks on syncContext.
173	*/
174	@GuardedBy("lock")
175	private PendingStream createPendingStream(PickSubchannelArgs args, ClientStreamTracer[] tracers,
176	PickResult pickResult) {
177	PendingStream pendingStream = new PendingStream(args, tracers);	1✔
178	if (args.getCallOptions().isWaitForReady() && pickResult != null && pickResult.hasResult()) {	1✔
179	pendingStream.lastPickStatus = pickResult.getStatus();	1✔
180	}
181	pendingStreams.add(pendingStream);	1✔
182	if (getPendingStreamsCount() == 1) {	1✔
183	syncContext.executeLater(reportTransportInUse);	1✔
184	}
185	for (ClientStreamTracer streamTracer : tracers) {	1✔
186	streamTracer.createPendingStream();	1✔
187	}
188	return pendingStream;	1✔
189	}
190
191	@Override
192	public final void ping(final PingCallback callback, Executor executor) {
193	throw new UnsupportedOperationException("This method is not expected to be called");	×
194	}
195
196	@Override
197	public ListenableFuture<SocketStats> getStats() {
198	SettableFuture<SocketStats> ret = SettableFuture.create();	×
199	ret.set(null);	×
200	return ret;	×
201	}
202
203	/**
204	* Prevents creating any new streams. Buffered streams are not failed and may still proceed
205	* when {@link #reprocess} is called. The delayed transport will be terminated when there is no
206	* more buffered streams.
207	*/
208	@Override
209	public final void shutdown(final Status status) {
210	synchronized (lock) {	1✔
211	if (pickerState.shutdownStatus != null) {	1✔
212	return;	1✔
213	}
214	pickerState = pickerState.withShutdownStatus(status);	1✔
215	syncContext.executeLater(new Runnable() {	1✔
216	@Override
217	public void run() {
218	listener.transportShutdown(status);	1✔
219	}	1✔
220	});
221	if (!hasPendingStreams() && reportTransportTerminated != null) {	1✔
222	syncContext.executeLater(reportTransportTerminated);	1✔
223	reportTransportTerminated = null;	1✔
224	}
225	}	1✔
226	syncContext.drain();	1✔
227	}	1✔
228
229	/**
230	* Shuts down this transport and cancels all streams that it owns, hence immediately terminates
231	* this transport.
232	*/
233	@Override
234	public final void shutdownNow(Status status) {
235	shutdown(status);	1✔
236	Collection<PendingStream> savedPendingStreams;
237	Runnable savedReportTransportTerminated;
238	synchronized (lock) {	1✔
239	savedPendingStreams = pendingStreams;	1✔
240	savedReportTransportTerminated = reportTransportTerminated;	1✔
241	reportTransportTerminated = null;	1✔
242	if (!pendingStreams.isEmpty()) {	1✔
243	pendingStreams = Collections.emptyList();	1✔
244	}
245	}	1✔
246	if (savedReportTransportTerminated != null) {	1✔
247	for (PendingStream stream : savedPendingStreams) {	1✔
248	Runnable runnable = stream.setStream(	1✔
249	new FailingClientStream(status, RpcProgress.REFUSED, stream.tracers));	1✔
250	if (runnable != null) {	1✔
251	// Drain in-line instead of using an executor as failing stream just throws everything
252	// away. This is essentially the same behavior as DelayedStream.cancel() but can be done
253	// before stream.start().
254	runnable.run();	1✔
255	}
256	}	1✔
257	syncContext.execute(savedReportTransportTerminated);	1✔
258	}
259	// If savedReportTransportTerminated == null, transportTerminated() has already been called in
260	// shutdown().
261	}	1✔
262
263	public final boolean hasPendingStreams() {
264	synchronized (lock) {	1✔
265	return !pendingStreams.isEmpty();	1✔
266	}
267	}
268
269	@VisibleForTesting
270	final int getPendingStreamsCount() {
271	synchronized (lock) {	1✔
272	return pendingStreams.size();	1✔
273	}
274	}
275
276	/**
277	* Use the picker to try picking a transport for every pending stream, proceed the stream if the
278	* pick is successful, otherwise keep it pending.
279	*
280	* <p>This method may be called concurrently with {@code newStream()}, and it's safe. All pending
281	* streams will be served by the latest picker (if a same picker is given more than once, they are
282	* considered different pickers) as soon as possible.
283	*
284	* <p>This method <strong>must not</strong> be called concurrently with itself.
285	*/
286	final void reprocess(@Nullable SubchannelPicker picker) {
287	ArrayList<PendingStream> toProcess;
288	synchronized (lock) {	1✔
289	pickerState = pickerState.withPicker(picker);	1✔
290	if (picker == null \|\| !hasPendingStreams()) {	1✔
291	return;	1✔
292	}
293	toProcess = new ArrayList<>(pendingStreams);	1✔
294	}	1✔
295	ArrayList<PendingStream> toRemove = new ArrayList<>();	1✔
296
297	for (final PendingStream stream : toProcess) {	1✔
298	PickResult pickResult = picker.pickSubchannel(stream.args);	1✔
299	CallOptions callOptions = stream.args.getCallOptions();	1✔
300	if (callOptions.isWaitForReady() && pickResult.hasResult()) {	1✔
301	stream.lastPickStatus = pickResult.getStatus();	1✔
302	}
303	final ClientTransport transport = GrpcUtil.getTransportFromPickResult(pickResult,	1✔
304	callOptions.isWaitForReady());	1✔
305	if (transport != null) {	1✔
306	Executor executor = defaultAppExecutor;	1✔
307	// createRealStream may be expensive. It will start real streams on the transport. If
308	// there are pending requests, they will be serialized too, which may be expensive. Since
309	// we are now on transport thread, we need to offload the work to an executor.
310	if (callOptions.getExecutor() != null) {	1✔
311	executor = callOptions.getExecutor();	1✔
312	}
313	Runnable runnable = stream.createRealStream(transport, pickResult.getAuthorityOverride());	1✔
314	if (runnable != null) {	1✔
315	executor.execute(runnable);	1✔
316	}
317	toRemove.add(stream);	1✔
318	} // else: stay pending
319	}	1✔
320
321	synchronized (lock) {	1✔
322	// Between this synchronized and the previous one:
323	// - Streams may have been cancelled, which may turn pendingStreams into emptiness.
324	// - shutdown() may be called, which may turn pendingStreams into null.
325	if (!hasPendingStreams()) {	1✔
326	return;	×
327	}
328	pendingStreams.removeAll(toRemove);	1✔
329	// Because delayed transport is long-lived, we take this opportunity to down-size the
330	// hashmap.
331	if (pendingStreams.isEmpty()) {	1✔
332	pendingStreams = new LinkedHashSet<>();	1✔
333	}
334	if (!hasPendingStreams()) {	1✔
335	// There may be a brief gap between delayed transport clearing in-use state, and first real
336	// transport starting streams and setting in-use state. During the gap the whole channel's
337	// in-use state may be false. However, it shouldn't cause spurious switching to idleness
338	// (which would shutdown the transports and LoadBalancer) because the gap should be shorter
339	// than IDLE_MODE_DEFAULT_TIMEOUT_MILLIS (1 second).
340	syncContext.executeLater(reportTransportNotInUse);	1✔
341	if (pickerState.shutdownStatus != null && reportTransportTerminated != null) {	1✔
342	syncContext.executeLater(reportTransportTerminated);	1✔
343	reportTransportTerminated = null;	1✔
344	}
345	}
346	}	1✔
347	syncContext.drain();	1✔
348	}	1✔
349
350	@Override
351	public InternalLogId getLogId() {
352	return logId;	×
353	}
354
355	private class PendingStream extends DelayedStream {
356	private final PickSubchannelArgs args;
357	private final Context context = Context.current();	1✔
358	private final ClientStreamTracer[] tracers;
359	private volatile Status lastPickStatus;
360
361	private PendingStream(PickSubchannelArgs args, ClientStreamTracer[] tracers) {	1✔
362	this.args = args;	1✔
363	this.tracers = tracers;	1✔
364	}	1✔
365
366	/** Runnable may be null. */
367	private Runnable createRealStream(ClientTransport transport, String authorityOverride) {
368	ClientStream realStream;
369	Context origContext = context.attach();	1✔
370	try {
371	realStream = transport.newStream(	1✔
372	args.getMethodDescriptor(), args.getHeaders(), args.getCallOptions(),	1✔
373	tracers);
374	} finally {
375	context.detach(origContext);	1✔
376	}
377	if (authorityOverride != null) {	1✔
378	// User code provided authority takes precedence over the LB provided one; this will be
379	// overwritten by an enqueud call from ClientCallImpl if the application sets an authority
380	// override. We must call the real stream directly because stream.start() has likely already
381	// been called on the delayed stream.
382	realStream.setAuthority(authorityOverride);	1✔
383	}
384	return setStream(realStream);	1✔
385	}
386
387	@Override
388	public void cancel(Status reason) {
389	super.cancel(reason);	1✔
390	synchronized (lock) {	1✔
391	if (reportTransportTerminated != null) {	1✔
392	boolean justRemovedAnElement = pendingStreams.remove(this);	1✔
393	if (!hasPendingStreams() && justRemovedAnElement) {	1✔
394	syncContext.executeLater(reportTransportNotInUse);	1✔
395	if (pickerState.shutdownStatus != null) {	1✔
396	syncContext.executeLater(reportTransportTerminated);	1✔
397	reportTransportTerminated = null;	1✔
398	}
399	}
400	}
401	}	1✔
402	syncContext.drain();	1✔
403	}	1✔
404
405	@Override
406	protected void onEarlyCancellation(Status reason) {
407	for (ClientStreamTracer tracer : tracers) {	1✔
408	tracer.streamClosed(reason);	1✔
409	}
410	}	1✔
411
412	@Override
413	public void appendTimeoutInsight(InsightBuilder insight) {
414	if (args.getCallOptions().isWaitForReady()) {	1✔
415	insight.append("wait_for_ready");	1✔
416	Status status = lastPickStatus;	1✔
417	if (status != null && !status.isOk()) {	1✔
418	insight.appendKeyValue("Last Pick Failure", status);	1✔
419	}
420	}
421	super.appendTimeoutInsight(insight);	1✔
422	}	1✔
423	}
424
425	static final class PickerState {
426	/**
427	* The last picker that {@link #reprocess} has used. May be set to null when the channel has
428	* moved to idle.
429	*/
430	@Nullable
431	final SubchannelPicker lastPicker;
432	/**
433	* When {@code shutdownStatus != null && !hasPendingStreams()}, then the transport is considered
434	* terminated.
435	*/
436	@Nullable
437	final Status shutdownStatus;
438
439	private PickerState(SubchannelPicker lastPicker, Status shutdownStatus) {	1✔
440	this.lastPicker = lastPicker;	1✔
441	this.shutdownStatus = shutdownStatus;	1✔
442	}	1✔
443
444	public PickerState withPicker(SubchannelPicker newPicker) {
445	return new PickerState(newPicker, this.shutdownStatus);	1✔
446	}
447
448	public PickerState withShutdownStatus(Status newShutdownStatus) {
449	return new PickerState(this.lastPicker, newShutdownStatus);	1✔
450	}
451	}
452	}

grpc / grpc-java / #19691

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