Skip to content

Commit cf96549

Browse files
kannanjgithubkannanjgithub
committed
core: respect perAttemptRecvTimeout in RetriableStream
This change respects the `perAttemptRecvTimeoutNanos` configured in the retry policy of a service config, which was previously ignored by the client channel implementation. - Added an `overallDeadline` field to `RetriableStream` to store the overall call deadline. - Calculated the attempt-specific deadline when spawning a new retry attempt (`Substream`) based on `perAttemptRecvTimeoutNanos`. - Bounded each substream's deadline by the minimum of the overall call deadline and the attempt-specific timeout. - Added unit tests in `RetriableStreamTest` and `ManagedChannelImplTest` to verify that the attempt timeout is correctly calculated and applied.
1 parent 53ebe2d commit cf96549

3 files changed

Lines changed: 148 additions & 2 deletions

File tree

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

Lines changed: 32 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -123,6 +123,8 @@ public void uncaughtException(Thread t, Throwable e) {
123123
private long nextBackoffIntervalNanos;
124124
private Status cancellationStatus;
125125
private boolean isClosed;
126+
@GuardedBy("lock")
127+
private Deadline overallDeadline;
126128

127129
RetriableStream(
128130
MethodDescriptor<ReqT, ?> method, Metadata headers,
@@ -257,7 +259,11 @@ private Substream createSubstream(int previousAttemptCount, boolean isTransparen
257259
return null;
258260
}
259261
} while (!inFlightSubStreams.compareAndSet(inFlight, inFlight + 1));
260-
Substream sub = new Substream(previousAttemptCount);
262+
Deadline attemptDeadline = null;
263+
if (retryPolicy != null && retryPolicy.perAttemptRecvTimeoutNanos != null) {
264+
attemptDeadline = Deadline.after(retryPolicy.perAttemptRecvTimeoutNanos, TimeUnit.NANOSECONDS);
265+
}
266+
Substream sub = new Substream(previousAttemptCount, attemptDeadline);
261267
// one tracer per substream
262268
final ClientStreamTracer bufferSizeTracer = new BufferSizeTracer(sub);
263269
ClientStreamTracer.Factory tracerFactory = new ClientStreamTracer.Factory() {
@@ -272,6 +278,17 @@ public ClientStreamTracer newClientStreamTracer(
272278
// NOTICE: This set _must_ be done before stream.start() and it actually is.
273279
sub.stream = newSubstream(newHeaders, tracerFactory, previousAttemptCount, isTransparentRetry,
274280
isHedgedStream);
281+
282+
Deadline combinedDeadline = attemptDeadline;
283+
synchronized (lock) {
284+
if (overallDeadline != null) {
285+
combinedDeadline = (attemptDeadline != null) ? overallDeadline.minimum(attemptDeadline) : overallDeadline;
286+
}
287+
}
288+
if (combinedDeadline != null) {
289+
sub.stream.setDeadline(combinedDeadline);
290+
}
291+
275292
return sub;
276293
}
277294

@@ -751,10 +768,17 @@ public void runWith(Substream substream) {
751768

752769
@Override
753770
public final void setDeadline(final Deadline deadline) {
771+
synchronized (lock) {
772+
overallDeadline = deadline;
773+
}
754774
class DeadlineEntry implements BufferEntry {
755775
@Override
756776
public void runWith(Substream substream) {
757-
substream.stream.setDeadline(deadline);
777+
Deadline combinedDeadline = deadline;
778+
if (substream.attemptDeadline != null) {
779+
combinedDeadline = deadline.minimum(substream.attemptDeadline);
780+
}
781+
substream.stream.setDeadline(combinedDeadline);
758782
}
759783
}
760784

@@ -1382,9 +1406,15 @@ private static final class Substream {
13821406
boolean bufferLimitExceeded;
13831407

13841408
final int previousAttemptCount;
1409+
@Nullable final Deadline attemptDeadline;
13851410

13861411
Substream(int previousAttemptCount) {
1412+
this(previousAttemptCount, null);
1413+
}
1414+
1415+
Substream(int previousAttemptCount, @Nullable Deadline attemptDeadline) {
13871416
this.previousAttemptCount = previousAttemptCount;
1417+
this.attemptDeadline = attemptDeadline;
13881418
}
13891419
}
13901420

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

Lines changed: 66 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -80,6 +80,7 @@
8080
import io.grpc.ConnectivityState;
8181
import io.grpc.ConnectivityStateInfo;
8282
import io.grpc.Context;
83+
import io.grpc.Deadline;
8384
import io.grpc.EquivalentAddressGroup;
8485
import io.grpc.InsecureChannelCredentials;
8586
import io.grpc.IntegerMarshaller;
@@ -3508,6 +3509,71 @@ public double nextDouble() {
35083509
channel.isTerminated());
35093510
}
35103511

3512+
@Test
3513+
public void retryPerAttemptTimeout() {
3514+
Map<String, Object> retryPolicy = new HashMap<>();
3515+
retryPolicy.put("maxAttempts", 3D);
3516+
retryPolicy.put("initialBackoff", "10s");
3517+
retryPolicy.put("maxBackoff", "30s");
3518+
retryPolicy.put("backoffMultiplier", 2D);
3519+
retryPolicy.put("retryableStatusCodes", Arrays.<Object>asList("UNAVAILABLE", "DEADLINE_EXCEEDED"));
3520+
retryPolicy.put("perAttemptRecvTimeout", "5s");
3521+
Map<String, Object> methodConfig = new HashMap<>();
3522+
Map<String, Object> name = new HashMap<>();
3523+
name.put("service", "service");
3524+
methodConfig.put("name", Arrays.<Object>asList(name));
3525+
methodConfig.put("retryPolicy", retryPolicy);
3526+
Map<String, Object> rawServiceConfig = new HashMap<>();
3527+
rawServiceConfig.put("methodConfig", Arrays.<Object>asList(methodConfig));
3528+
3529+
FakeNameResolverFactory nameResolverFactory =
3530+
new FakeNameResolverFactory.Builder(expectedUri)
3531+
.setServers(Collections.singletonList(new EquivalentAddressGroup(socketAddress)))
3532+
.build();
3533+
ManagedChannelServiceConfig managedChannelServiceConfig =
3534+
createManagedChannelServiceConfig(rawServiceConfig, null);
3535+
nameResolverFactory.nextConfigOrError.set(
3536+
ConfigOrError.fromConfig(managedChannelServiceConfig));
3537+
3538+
channelBuilder.nameResolverFactory(nameResolverFactory);
3539+
channelBuilder.executor(MoreExecutors.directExecutor());
3540+
channelBuilder.enableRetry();
3541+
3542+
requestConnection = false;
3543+
createChannel();
3544+
3545+
ClientCall<String, Integer> call = channel.newCall(method, CallOptions.DEFAULT);
3546+
call.start(mockCallListener, new Metadata());
3547+
ArgumentCaptor<Helper> helperCaptor = ArgumentCaptor.forClass(Helper.class);
3548+
verify(mockLoadBalancerProvider).newLoadBalancer(helperCaptor.capture());
3549+
helper = helperCaptor.getValue();
3550+
verify(mockLoadBalancer).acceptResolvedAddresses(resolvedAddressCaptor.capture());
3551+
3552+
Subchannel subchannel =
3553+
createSubchannelSafely(helper, addressGroup, Attributes.EMPTY, subchannelStateListener);
3554+
when(mockPicker.pickSubchannel(any(PickSubchannelArgs.class)))
3555+
.thenReturn(PickResult.withSubchannel(subchannel));
3556+
requestConnectionSafely(helper, subchannel);
3557+
MockClientTransportInfo transportInfo = transports.poll();
3558+
ConnectionClientTransport mockTransport = transportInfo.transport;
3559+
ClientStream mockStream = mock(ClientStream.class);
3560+
3561+
when(mockTransport.newStream(
3562+
same(method), any(Metadata.class), any(CallOptions.class),
3563+
ArgumentMatchers.<ClientStreamTracer[]>any()))
3564+
.thenReturn(mockStream);
3565+
transportInfo.listener.transportReady();
3566+
updateBalancingStateSafely(helper, READY, mockPicker);
3567+
3568+
executor.runDueTasks();
3569+
3570+
ArgumentCaptor<Deadline> deadlineCaptor = ArgumentCaptor.forClass(Deadline.class);
3571+
verify(mockStream).setDeadline(deadlineCaptor.capture());
3572+
long remaining = deadlineCaptor.getValue().timeRemaining(TimeUnit.MILLISECONDS);
3573+
assertThat(remaining).isAtLeast(4000L);
3574+
assertThat(remaining).isAtMost(6000L);
3575+
}
3576+
35113577
@Test
35123578
public void hedgingScheduledThenChannelShutdown_hedgeShouldStillHappen_newCallShouldFail() {
35133579
Map<String, Object> hedgingPolicy = new HashMap<>();

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

Lines changed: 50 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -47,6 +47,7 @@
4747
import com.google.common.util.concurrent.MoreExecutors;
4848
import io.grpc.ClientStreamTracer;
4949
import io.grpc.Codec;
50+
import io.grpc.Deadline;
5051
import io.grpc.Compressor;
5152
import io.grpc.DecompressorRegistry;
5253
import io.grpc.Metadata;
@@ -2963,6 +2964,55 @@ public void hedging_throttledByHedgingStreams() {
29632964
assertEquals(0, fakeClock.numPendingTasks());
29642965
}
29652966

2967+
@Test
2968+
public void perAttemptTimeout_deadlineSetOnSubstream() {
2969+
long perAttemptTimeoutNanos = TimeUnit.SECONDS.toNanos(5);
2970+
RetryPolicy retryPolicy = new RetryPolicy(
2971+
3,
2972+
TimeUnit.SECONDS.toNanos(1),
2973+
TimeUnit.SECONDS.toNanos(2),
2974+
2D,
2975+
perAttemptTimeoutNanos,
2976+
ImmutableSet.of(Code.UNAVAILABLE, Code.DEADLINE_EXCEEDED));
2977+
2978+
// Case 1: Overall deadline is longer than per-attempt timeout (15s > 5s)
2979+
RetriableStream<String> stream = new RecordedRetriableStream(
2980+
method, new Metadata(), channelBufferUsed, PER_RPC_BUFFER_LIMIT, CHANNEL_BUFFER_LIMIT,
2981+
MoreExecutors.directExecutor(), fakeClock.getScheduledExecutorService(),
2982+
retryPolicy, null, null);
2983+
2984+
ClientStream mockStream1 = mock(ClientStream.class);
2985+
doReturn(mockStream1).when(retriableStreamRecorder).newSubstream(0);
2986+
2987+
Deadline overallDeadline = Deadline.after(15, TimeUnit.SECONDS);
2988+
stream.setDeadline(overallDeadline);
2989+
stream.start(masterListener);
2990+
2991+
ArgumentCaptor<Deadline> deadlineCaptor = ArgumentCaptor.forClass(Deadline.class);
2992+
verify(mockStream1, times(2)).setDeadline(deadlineCaptor.capture());
2993+
assertThat(deadlineCaptor.getValue()).isNotEqualTo(overallDeadline);
2994+
long remaining = deadlineCaptor.getValue().timeRemaining(TimeUnit.MILLISECONDS);
2995+
assertThat(remaining).isAtLeast(4000L);
2996+
assertThat(remaining).isAtMost(6000L);
2997+
2998+
// Case 2: Overall deadline is shorter than per-attempt timeout (2s < 5s)
2999+
RetriableStream<String> stream2 = new RecordedRetriableStream(
3000+
method, new Metadata(), channelBufferUsed, PER_RPC_BUFFER_LIMIT, CHANNEL_BUFFER_LIMIT,
3001+
MoreExecutors.directExecutor(), fakeClock.getScheduledExecutorService(),
3002+
retryPolicy, null, null);
3003+
3004+
ClientStream mockStream2 = mock(ClientStream.class);
3005+
doReturn(mockStream2).when(retriableStreamRecorder).newSubstream(0);
3006+
3007+
Deadline overallDeadline2 = Deadline.after(2, TimeUnit.SECONDS);
3008+
stream2.setDeadline(overallDeadline2);
3009+
stream2.start(masterListener);
3010+
3011+
ArgumentCaptor<Deadline> deadlineCaptor2 = ArgumentCaptor.forClass(Deadline.class);
3012+
verify(mockStream2, times(2)).setDeadline(deadlineCaptor2.capture());
3013+
assertThat(deadlineCaptor2.getValue()).isEqualTo(overallDeadline2);
3014+
}
3015+
29663016
/**
29673017
* Used to stub a retriable stream as well as to record methods of the retriable stream being
29683018
* called.

0 commit comments

Comments
 (0)