Skip to content

Commit b78042a

Browse files
kannanjgithubkannanjgithub
authored
Merge branch 'grpc:master' into ext-proc-server
2 parents de52d88 + d49c0b1 commit b78042a

8 files changed

Lines changed: 46 additions & 139 deletions

File tree

‎core/src/main/java/io/grpc/internal/DelayedClientCall.java‎

Lines changed: 10 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -49,6 +49,9 @@
4949
*/
5050
public class DelayedClientCall<ReqT, RespT> extends ClientCall<ReqT, RespT> {
5151
private static final Logger logger = Logger.getLogger(DelayedClientCall.class.getName());
52+
53+
/** A string describing what this call is waiting on. */
54+
private final String bufferContext;
5255
/**
5356
* A timer to monitor the initial deadline. The timer must be cancelled on transition to the real
5457
* call.
@@ -76,7 +79,11 @@ public class DelayedClientCall<ReqT, RespT> extends ClientCall<ReqT, RespT> {
7679
private DelayedListener<RespT> delayedListener;
7780

7881
protected DelayedClientCall(
79-
Executor callExecutor, ScheduledExecutorService scheduler, @Nullable Deadline deadline) {
82+
String bufferContext,
83+
Executor callExecutor,
84+
ScheduledExecutorService scheduler,
85+
@Nullable Deadline deadline) {
86+
this.bufferContext = checkNotNull(bufferContext, "bufferContext");
8087
this.callExecutor = checkNotNull(callExecutor, "callExecutor");
8188
checkNotNull(scheduler, "scheduler");
8289
context = Context.current();
@@ -143,7 +150,8 @@ public void run() {
143150
}
144151
buf.append(seconds);
145152
buf.append(String.format(Locale.US, ".%09d", nanos));
146-
buf.append("s");
153+
buf.append("s waiting for ");
154+
buf.append(bufferContext);
147155
cancel(
148156
Status.DEADLINE_EXCEEDED.withDescription(buf.toString()),
149157
// We should not cancel the call if the realCall is set because there could be a

‎core/src/main/java/io/grpc/internal/ManagedChannelImpl.java‎

Lines changed: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -986,9 +986,12 @@ private final class PendingCall<ReqT, RespT> extends DelayedClientCall<ReqT, Res
986986
final CallOptions callOptions;
987987
private final long callCreationTime;
988988

989-
PendingCall(
990-
Context context, MethodDescriptor<ReqT, RespT> method, CallOptions callOptions) {
991-
super(getCallExecutor(callOptions), scheduledExecutor, callOptions.getDeadline());
989+
PendingCall(Context context, MethodDescriptor<ReqT, RespT> method, CallOptions callOptions) {
990+
super(
991+
"name_resolver",
992+
getCallExecutor(callOptions),
993+
scheduledExecutor,
994+
callOptions.getDeadline());
992995
this.context = context;
993996
this.method = method;
994997
this.callOptions = callOptions;

‎core/src/test/java/io/grpc/internal/DelayedClientCallTest.java‎

Lines changed: 22 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -66,8 +66,8 @@ public class DelayedClientCallTest {
6666

6767
@Test
6868
public void allMethodsForwarded() throws Exception {
69-
DelayedClientCall<String, Integer> delayedClientCall =
70-
new DelayedClientCall<>(callExecutor, fakeClock.getScheduledExecutorService(), null);
69+
DelayedClientCall<String, Integer> delayedClientCall = new DelayedClientCall<>(
70+
"test", callExecutor, fakeClock.getScheduledExecutorService(), null);
7171
callMeMaybe(delayedClientCall.setCall(mockRealCall));
7272
ForwardingTestUtil.testMethodsForwarded(
7373
ClientCall.class,
@@ -94,18 +94,22 @@ public Object get(Method method, int argPos, Class<?> clazz) {
9494
@Test
9595
public void deadlineExceededWhileCallIsStartedButStillPending() {
9696
DelayedClientCall<String, Integer> delayedClientCall = new DelayedClientCall<>(
97-
callExecutor, fakeClock.getScheduledExecutorService(), Deadline.after(10, SECONDS));
97+
"tESt", callExecutor, fakeClock.getScheduledExecutorService(),
98+
Deadline.after(10, SECONDS, fakeClock.getDeadlineTicker()));
9899

99100
delayedClientCall.start(listener, new Metadata());
100101
fakeClock.forwardTime(10, SECONDS);
101102
verify(listener).onClose(statusCaptor.capture(), any(Metadata.class));
102103
assertThat(statusCaptor.getValue().getCode()).isEqualTo(Status.Code.DEADLINE_EXCEEDED);
104+
assertThat(statusCaptor.getValue().getDescription())
105+
.isEqualTo("Deadline CallOptions was exceeded after 10.000000000s waiting for tESt");
103106
}
104107

105108
@Test
106109
public void listenerEventsPropagated() {
107110
DelayedClientCall<String, Integer> delayedClientCall = new DelayedClientCall<>(
108-
callExecutor, fakeClock.getScheduledExecutorService(), Deadline.after(10, SECONDS));
111+
"test", callExecutor, fakeClock.getScheduledExecutorService(),
112+
Deadline.after(10, SECONDS, fakeClock.getDeadlineTicker()));
109113
delayedClientCall.start(listener, new Metadata());
110114
callMeMaybe(delayedClientCall.setCall(mockRealCall));
111115
@SuppressWarnings("unchecked")
@@ -130,7 +134,7 @@ public void listenerEventsPropagated() {
130134
@Test
131135
public void setCallThenStart() {
132136
DelayedClientCall<String, Integer> delayedClientCall = new DelayedClientCall<>(
133-
callExecutor, fakeClock.getScheduledExecutorService(), null);
137+
"test", callExecutor, fakeClock.getScheduledExecutorService(), null);
134138
callMeMaybe(delayedClientCall.setCall(mockRealCall));
135139
delayedClientCall.start(listener, new Metadata());
136140
delayedClientCall.request(1);
@@ -146,7 +150,7 @@ public void setCallThenStart() {
146150
@Test
147151
public void startThenSetCall() {
148152
DelayedClientCall<String, Integer> delayedClientCall = new DelayedClientCall<>(
149-
callExecutor, fakeClock.getScheduledExecutorService(), null);
153+
"test", callExecutor, fakeClock.getScheduledExecutorService(), null);
150154
delayedClientCall.start(listener, new Metadata());
151155
delayedClientCall.request(1);
152156
Runnable r = delayedClientCall.setCall(mockRealCall);
@@ -167,7 +171,7 @@ public void startThenSetCall() {
167171
@SuppressWarnings("unchecked")
168172
public void cancelThenSetCall() {
169173
DelayedClientCall<String, Integer> delayedClientCall = new DelayedClientCall<>(
170-
callExecutor, fakeClock.getScheduledExecutorService(), null);
174+
"test", callExecutor, fakeClock.getScheduledExecutorService(), null);
171175
delayedClientCall.start(listener, new Metadata());
172176
delayedClientCall.request(1);
173177
delayedClientCall.cancel("cancel", new StatusException(Status.CANCELLED));
@@ -183,7 +187,7 @@ public void cancelThenSetCall() {
183187
@SuppressWarnings("unchecked")
184188
public void setCallThenCancel() {
185189
DelayedClientCall<String, Integer> delayedClientCall = new DelayedClientCall<>(
186-
callExecutor, fakeClock.getScheduledExecutorService(), null);
190+
"test", callExecutor, fakeClock.getScheduledExecutorService(), null);
187191
delayedClientCall.start(listener, new Metadata());
188192
delayedClientCall.request(1);
189193
Runnable r = delayedClientCall.setCall(mockRealCall);
@@ -206,7 +210,8 @@ public void delayedCallsRunUnderContext() throws Exception {
206210
Object goldenValue = new Object();
207211
DelayedClientCall<String, Integer> delayedClientCall =
208212
Context.current().withValue(contextKey, goldenValue).call(() ->
209-
new DelayedClientCall<>(callExecutor, fakeClock.getScheduledExecutorService(), null));
213+
new DelayedClientCall<>(
214+
"test", callExecutor, fakeClock.getScheduledExecutorService(), null));
210215
AtomicReference<Context> readyContext = new AtomicReference<>();
211216
delayedClientCall.start(new ClientCall.Listener<Integer>() {
212217
@Override public void onReady() {
@@ -232,7 +237,7 @@ public void delayedCallsRunUnderContext() throws Exception {
232237
@Test
233238
public void listenerThrowsInPendingCallback_cancelsRealCall() {
234239
DelayedClientCall<String, Integer> delayedClientCall = new DelayedClientCall<>(
235-
callExecutor, fakeClock.getScheduledExecutorService(), null);
240+
"test", callExecutor, fakeClock.getScheduledExecutorService(), null);
236241
final RuntimeException boom = new RuntimeException("boom");
237242
ClientCall.Listener<Integer> throwingListener = new ClientCall.Listener<Integer>() {
238243
@Override
@@ -261,7 +266,7 @@ public void start(Listener<Integer> listener, Metadata metadata) {
261266
@Test
262267
public void listenerThrowsInPendingOnHeaders_cancelsRealCall() {
263268
DelayedClientCall<String, Integer> delayedClientCall = new DelayedClientCall<>(
264-
callExecutor, fakeClock.getScheduledExecutorService(), null);
269+
"test", callExecutor, fakeClock.getScheduledExecutorService(), null);
265270
final RuntimeException boom = new RuntimeException("boom");
266271
ClientCall.Listener<Integer> throwingListener = new ClientCall.Listener<Integer>() {
267272
@Override
@@ -286,7 +291,7 @@ public void start(Listener<Integer> listener, Metadata metadata) {
286291
@Test
287292
public void listenerThrowsInPendingOnReady_cancelsRealCall() {
288293
DelayedClientCall<String, Integer> delayedClientCall = new DelayedClientCall<>(
289-
callExecutor, fakeClock.getScheduledExecutorService(), null);
294+
"test", callExecutor, fakeClock.getScheduledExecutorService(), null);
290295
final RuntimeException boom = new RuntimeException("boom");
291296
ClientCall.Listener<Integer> throwingListener = new ClientCall.Listener<Integer>() {
292297
@Override
@@ -311,7 +316,7 @@ public void start(Listener<Integer> listener, Metadata metadata) {
311316
@Test
312317
public void onCloseExceptionCaughtAndLogged() {
313318
DelayedClientCall<String, Integer> delayedClientCall = new DelayedClientCall<>(
314-
callExecutor, fakeClock.getScheduledExecutorService(), null);
319+
"test", callExecutor, fakeClock.getScheduledExecutorService(), null);
315320
final RuntimeException boom = new RuntimeException("boom");
316321
final AtomicReference<Status> observed = new AtomicReference<>();
317322
ClientCall.Listener<Integer> throwingListener = new ClientCall.Listener<Integer>() {
@@ -339,7 +344,7 @@ public void start(Listener<Integer> listener, Metadata metadata) {
339344
@Test
340345
public void listenerThrowsInPassThroughOnMessage_cancelsRealCall() {
341346
DelayedClientCall<String, Integer> delayedClientCall = new DelayedClientCall<>(
342-
callExecutor, fakeClock.getScheduledExecutorService(), null);
347+
"test", callExecutor, fakeClock.getScheduledExecutorService(), null);
343348
final RuntimeException boom = new RuntimeException("boom");
344349
ClientCall.Listener<Integer> throwingListener = new ClientCall.Listener<Integer>() {
345350
@Override
@@ -362,7 +367,7 @@ public void onMessage(Integer msg) {
362367
@Test
363368
public void listenerThrowsInPassThroughOnHeaders_cancelsRealCall() {
364369
DelayedClientCall<String, Integer> delayedClientCall = new DelayedClientCall<>(
365-
callExecutor, fakeClock.getScheduledExecutorService(), null);
370+
"test", callExecutor, fakeClock.getScheduledExecutorService(), null);
366371
final RuntimeException boom = new RuntimeException("boom");
367372
ClientCall.Listener<Integer> throwingListener = new ClientCall.Listener<Integer>() {
368373
@Override
@@ -385,7 +390,7 @@ public void onHeaders(Metadata headers) {
385390
@Test
386391
public void listenerThrowsInPassThroughOnReady_cancelsRealCall() {
387392
DelayedClientCall<String, Integer> delayedClientCall = new DelayedClientCall<>(
388-
callExecutor, fakeClock.getScheduledExecutorService(), null);
393+
"test", callExecutor, fakeClock.getScheduledExecutorService(), null);
389394
final RuntimeException boom = new RuntimeException("boom");
390395
ClientCall.Listener<Integer> throwingListener = new ClientCall.Listener<Integer>() {
391396
@Override
@@ -408,7 +413,7 @@ public void onReady() {
408413
@Test
409414
public void listenerThrowsInPassThrough_subsequentCallbacksSwallowedAndOnCloseOverridden() {
410415
DelayedClientCall<String, Integer> delayedClientCall = new DelayedClientCall<>(
411-
callExecutor, fakeClock.getScheduledExecutorService(), null);
416+
"test", callExecutor, fakeClock.getScheduledExecutorService(), null);
412417
final RuntimeException boom = new RuntimeException("boom");
413418
final AtomicReference<Integer> lastMessage = new AtomicReference<>();
414419
final AtomicReference<Status> closeStatus = new AtomicReference<>();

‎xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -262,7 +262,7 @@ public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
262262
private static class DataPlaneDelayedCall<ReqT, RespT> extends DelayedClientCall<ReqT, RespT> {
263263
DataPlaneDelayedCall(
264264
Executor executor, ScheduledExecutorService scheduler, @Nullable Deadline deadline) {
265-
super(executor, scheduler, deadline);
265+
super("ext_proc", executor, scheduler, deadline);
266266
}
267267
}
268268

‎xds/src/main/java/io/grpc/xds/FaultFilter.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -414,7 +414,7 @@ private final class DelayInjectedCall<ReqT, RespT> extends DelayedClientCall<Req
414414
long delayNanos, Executor callExecutor, ScheduledExecutorService scheduler,
415415
@Nullable Deadline deadline,
416416
final Supplier<? extends ClientCall<ReqT, RespT>> callSupplier) {
417-
super(callExecutor, scheduler, deadline);
417+
super("httpfault_filter", callExecutor, scheduler, deadline);
418418
activeFaultCounter.incrementAndGet();
419419
ScheduledFuture<?> task = scheduler.schedule(
420420
new Runnable() {

‎xds/src/main/java/io/grpc/xds/GrpcXdsTransportFactory.java‎

Lines changed: 5 additions & 57 deletions
Original file line numberDiff line numberDiff line change
@@ -31,74 +31,39 @@
3131
import io.grpc.Status;
3232
import io.grpc.xds.client.Bootstrapper;
3333
import io.grpc.xds.client.XdsTransportFactory;
34-
import java.util.Map;
35-
import java.util.concurrent.ConcurrentHashMap;
3634
import java.util.concurrent.TimeUnit;
3735

38-
/**
39-
* A factory for creating gRPC-based transports for xDS communication.
40-
*
41-
* <p>WARNING: This class reuses channels when possible, based on the provided {@link
42-
* Bootstrapper.ServerInfo} with important considerations. The {@link Bootstrapper.ServerInfo}
43-
* includes {@link ChannelCredentials}, which is compared by reference equality. This means every
44-
* {@link Bootstrapper.BootstrapInfo} would have non-equal copies of {@link
45-
* Bootstrapper.ServerInfo}, even if they all represent the same xDS server configuration. For gRPC
46-
* name resolution with the {@code xds} and {@code google-c2p} scheme, this transport sharing works
47-
* as expected as it internally reuses a single {@link Bootstrapper.BootstrapInfo} instance.
48-
* Otherwise, new transports would be created for each {@link Bootstrapper.ServerInfo} despite them
49-
* possibly representing the same xDS server configuration and defeating the purpose of transport
50-
* sharing.
51-
*/
5236
final class GrpcXdsTransportFactory implements XdsTransportFactory {
5337

5438
private final CallCredentials callCredentials;
55-
// The map of xDS server info to its corresponding gRPC xDS transport.
56-
// This enables reusing and sharing the same underlying gRPC channel.
57-
//
58-
// NOTE: ConcurrentHashMap is used as a per-entry lock and all reads and writes must be a mutation
59-
// via the ConcurrentHashMap APIs to acquire the per-entry lock in order to ensure thread safety
60-
// for reference counting of each GrpcXdsTransport instance.
61-
private static final Map<Bootstrapper.ServerInfo, GrpcXdsTransport> xdsServerInfoToTransportMap =
62-
new ConcurrentHashMap<>();
6339

6440
GrpcXdsTransportFactory(CallCredentials callCredentials) {
6541
this.callCredentials = callCredentials;
6642
}
6743

6844
@Override
6945
public XdsTransport create(Bootstrapper.ServerInfo serverInfo) {
70-
return xdsServerInfoToTransportMap.compute(
71-
serverInfo,
72-
(info, transport) -> {
73-
if (transport == null) {
74-
transport = new GrpcXdsTransport(serverInfo, callCredentials);
75-
}
76-
++transport.refCount;
77-
return transport;
78-
});
46+
return new GrpcXdsTransport(serverInfo, callCredentials);
7947
}
8048

8149
@VisibleForTesting
8250
public XdsTransport createForTest(ManagedChannel channel) {
83-
return new GrpcXdsTransport(channel, callCredentials, null);
51+
return new GrpcXdsTransport(channel, callCredentials);
8452
}
8553

8654
@VisibleForTesting
8755
static class GrpcXdsTransport implements XdsTransport {
8856

8957
private final ManagedChannel channel;
9058
private final CallCredentials callCredentials;
91-
private final Bootstrapper.ServerInfo serverInfo;
92-
// Must only be accessed via the ConcurrentHashMap APIs which act as the locking methods.
93-
private int refCount = 0;
9459

9560
public GrpcXdsTransport(Bootstrapper.ServerInfo serverInfo) {
9661
this(serverInfo, null);
9762
}
9863

9964
@VisibleForTesting
10065
public GrpcXdsTransport(ManagedChannel channel) {
101-
this(channel, null, null);
66+
this(channel, null);
10267
}
10368

10469
public GrpcXdsTransport(Bootstrapper.ServerInfo serverInfo, CallCredentials callCredentials) {
@@ -108,17 +73,12 @@ public GrpcXdsTransport(Bootstrapper.ServerInfo serverInfo, CallCredentials call
10873
.keepAliveTime(5, TimeUnit.MINUTES)
10974
.build();
11075
this.callCredentials = callCredentials;
111-
this.serverInfo = serverInfo;
11276
}
11377

11478
@VisibleForTesting
115-
public GrpcXdsTransport(
116-
ManagedChannel channel,
117-
CallCredentials callCredentials,
118-
Bootstrapper.ServerInfo serverInfo) {
79+
public GrpcXdsTransport(ManagedChannel channel, CallCredentials callCredentials) {
11980
this.channel = checkNotNull(channel, "channel");
12081
this.callCredentials = callCredentials;
121-
this.serverInfo = serverInfo;
12282
}
12383

12484
@Override
@@ -138,19 +98,7 @@ public <ReqT, RespT> StreamingCall<ReqT, RespT> createStreamingCall(
13898

13999
@Override
140100
public void shutdown() {
141-
if (serverInfo == null) {
142-
channel.shutdown();
143-
return;
144-
}
145-
xdsServerInfoToTransportMap.computeIfPresent(
146-
serverInfo,
147-
(info, transport) -> {
148-
if (--transport.refCount == 0) { // Prefix decrement and return the updated value.
149-
transport.channel.shutdown();
150-
return null; // Remove mapping.
151-
}
152-
return transport;
153-
});
101+
channel.shutdown();
154102
}
155103

156104
private class XdsStreamingCall<ReqT, RespT> implements

0 commit comments

Comments
 (0)