Skip to content

core: add perAttemptRecvTimeout to retry policy #8301

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
dapengzhang0 merged 2 commits into from
Jul 12, 2021
Merged

core: add perAttemptRecvTimeout to retry policy #8301

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
3d54e9f
core: add perAttemptRecvTimeout to retry policy
dapengzhang0 Jun 30, 2021
a4927d6
add tests
dapengzhang0 Jul 2, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 16 additions & 1 deletion core/src/main/java/io/grpc/internal/ManagedChannelServiceConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,11 +27,13 @@
import io.grpc.InternalConfigSelector;
import io.grpc.LoadBalancer.PickSubchannelArgs;
import io.grpc.MethodDescriptor;
import io.grpc.Status.Code;
import io.grpc.internal.RetriableStream.Throttle;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import javax.annotation.Nullable;

/**
Expand Down Expand Up @@ -354,9 +356,22 @@ private static RetryPolicy retryPolicy(Map<String, ?> retryPolicy, int maxAttemp
"backoffMultiplier must be greater than 0: %s",
backoffMultiplier);

Long perAttemptRecvTimeout =
ServiceConfigUtil.getPerAttemptRecvTimeoutNanosFromRetryPolicy(retryPolicy);
checkArgument(
perAttemptRecvTimeout == null || perAttemptRecvTimeout >= 0,
"perAttemptRecvTimeout cannot be negative: %s",
perAttemptRecvTimeout);

Set<Code> retryableCodes =
ServiceConfigUtil.getRetryableStatusCodesFromRetryPolicy(retryPolicy);
checkArgument(
perAttemptRecvTimeout != null || !retryableCodes.isEmpty(),

@ejona86 ejona86 Jul 2, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

No new tests? Seems we need tests for these cases.

@dapengzhang0 dapengzhang0 Jul 2, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why do we need tests for assertions. They will throw if the app is in wrong state and people will notice.

@dapengzhang0 dapengzhang0 Jul 2, 2021 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Say if I had a bug here and the code is like

checkArgument(
     perAttemptRecvTimeout != null && !retryableCodes.isEmpty()

with && not ||. Then existing code going through this code path would throw.

@ejona86 ejona86 Jul 2, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Then existing code going through this code path would throw.

The unit tests don't look like they would fail. If we can pass one or the other, it seems there should be tests for passing one or the other. I wouldn't be such a stickler but this is for service config and a bug in the logic can poison the well.

@dapengzhang0 dapengzhang0 Jul 2, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done.

"retryableStatusCodes cannot be empty without perAttemptRecvTimeout");

return new RetryPolicy(
maxAttempts, initialBackoffNanos, maxBackoffNanos, backoffMultiplier,
ServiceConfigUtil.getRetryableStatusCodesFromRetryPolicy(retryPolicy));
perAttemptRecvTimeout, retryableCodes);
}

private static HedgingPolicy hedgingPolicy(
Expand Down
8 changes: 8 additions & 0 deletions core/src/main/java/io/grpc/internal/RetryPolicy.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
import io.grpc.Status.Code;
import java.util.Set;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import javax.annotation.concurrent.Immutable;

/**
Expand All @@ -33,6 +34,8 @@ final class RetryPolicy {
final long initialBackoffNanos;
final long maxBackoffNanos;
final double backoffMultiplier;
@Nullable
final Long perAttemptRecvTimeoutNanos;
final Set<Code> retryableStatusCodes;

/**
Expand All @@ -44,11 +47,13 @@ final class RetryPolicy {
long initialBackoffNanos,
long maxBackoffNanos,
double backoffMultiplier,
@Nullable Long perAttemptRecvTimeoutNanos,
@Nonnull Set<Code> retryableStatusCodes) {
this.maxAttempts = maxAttempts;
this.initialBackoffNanos = initialBackoffNanos;
this.maxBackoffNanos = maxBackoffNanos;
this.backoffMultiplier = backoffMultiplier;
this.perAttemptRecvTimeoutNanos = perAttemptRecvTimeoutNanos;
this.retryableStatusCodes = ImmutableSet.copyOf(retryableStatusCodes);
}

Expand All @@ -59,6 +64,7 @@ public int hashCode() {
initialBackoffNanos,
maxBackoffNanos,
backoffMultiplier,
perAttemptRecvTimeoutNanos,
retryableStatusCodes);
}

Expand All @@ -72,6 +78,7 @@ public boolean equals(Object other) {
&& this.initialBackoffNanos == that.initialBackoffNanos
&& this.maxBackoffNanos == that.maxBackoffNanos
&& Double.compare(this.backoffMultiplier, that.backoffMultiplier) == 0
&& Objects.equal(this.perAttemptRecvTimeoutNanos, that.perAttemptRecvTimeoutNanos)
&& Objects.equal(this.retryableStatusCodes, that.retryableStatusCodes);
}

Expand All @@ -82,6 +89,7 @@ public String toString() {
.add("initialBackoffNanos", initialBackoffNanos)
.add("maxBackoffNanos", maxBackoffNanos)
.add("backoffMultiplier", backoffMultiplier)
.add("perAttemptRecvTimeoutNanos", perAttemptRecvTimeoutNanos)
.add("retryableStatusCodes", retryableStatusCodes)
.toString();
}
Expand Down
6 changes: 5 additions & 1 deletion core/src/main/java/io/grpc/internal/ServiceConfigUtil.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -140,6 +140,11 @@ static Double getBackoffMultiplierFromRetryPolicy(Map<String, ?> retryPolicy) {
return JsonUtil.getNumber(retryPolicy, "backoffMultiplier");
}

@Nullable
static Long getPerAttemptRecvTimeoutNanosFromRetryPolicy(Map<String, ?> retryPolicy) {
return JsonUtil.getStringAsDuration(retryPolicy, "perAttemptRecvTimeout");
}

private static Set<Status.Code> getListOfStatusCodesAsSet(Map<String, ?> obj, String key) {
List<?> statuses = JsonUtil.getList(obj, key);
if (statuses == null) {
Expand Down Expand Up @@ -178,7 +183,6 @@ static Set<Status.Code> getRetryableStatusCodesFromRetryPolicy(Map<String, ?> re
String retryableStatusCodesKey = "retryableStatusCodes";
Set<Status.Code> codes = getListOfStatusCodesAsSet(retryPolicy, retryableStatusCodesKey);
verify(codes != null, "%s is required in retry policy", retryableStatusCodesKey);
Comment thread
ejona86 marked this conversation as resolved.
verify(!codes.isEmpty(), "%s must not be empty", retryableStatusCodesKey);
verify(!codes.contains(Status.Code.OK), "%s must not contain OK", retryableStatusCodesKey);
return codes;
}
Expand Down
96 changes: 89 additions & 7 deletions core/src/test/java/io/grpc/internal/ManagedChannelServiceConfigTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
import static io.grpc.MethodDescriptor.MethodType.UNARY;
import static io.grpc.Status.Code.UNAVAILABLE;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static org.junit.Assert.fail;

import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
Expand Down Expand Up @@ -155,13 +156,14 @@ public void managedChannelServiceConfig_parseMethodConfig() {
"name", ImmutableList.of(name1, name2),
"timeout", "1.234s",
"retryPolicy",
ImmutableMap.of(
"maxAttempts", 3.0D,
"initialBackoff", "1s",
"maxBackoff", "10s",
"backoffMultiplier", 1.5D,
"retryableStatusCodes", ImmutableList.of("UNAVAILABLE")
));
ImmutableMap.builder()
.put("maxAttempts", 3.0D)
.put("initialBackoff", "1s")
.put("maxBackoff", "10s")
.put("backoffMultiplier", 1.5D)
.put("perAttemptRecvTimeout", "2.5s")
.put("retryableStatusCodes", ImmutableList.of("UNAVAILABLE"))
.build());
Map<String, ?> defaultMethodConfig = ImmutableMap.of(
"name", ImmutableList.of(ImmutableMap.of()),
"timeout", "4.321s");
Expand All @@ -187,6 +189,8 @@ public void managedChannelServiceConfig_parseMethodConfig() {
methodInfo = serviceConfig.getMethodConfig(methodForName("service1", "method1"));
assertThat(methodInfo.timeoutNanos).isEqualTo(MILLISECONDS.toNanos(1234));
assertThat(methodInfo.retryPolicy.maxAttempts).isEqualTo(2);
assertThat(methodInfo.retryPolicy.perAttemptRecvTimeoutNanos)
.isEqualTo(MILLISECONDS.toNanos(2500));
assertThat(methodInfo.retryPolicy.retryableStatusCodes).containsExactly(UNAVAILABLE);
methodInfo = serviceConfig.getMethodConfig(methodForName("service1", "methodX"));
assertThat(methodInfo.timeoutNanos).isEqualTo(MILLISECONDS.toNanos(4321));
Expand All @@ -212,6 +216,84 @@ public void getDefaultConfigSelectorFromConfig() {
.isEqualTo(serviceConfig.getMethodConfig(method));
}

@Test
public void retryConfig_emptyRetriableStatusCodesAllowedWithPerAttemptRecvTimeoutGiven() {
Map<String, ?> retryPolicy = ImmutableMap.<String, Object>builder()
.put("maxAttempts", 3.0D)
.put("initialBackoff", "1s")
.put("maxBackoff", "10s")
.put("backoffMultiplier", 1.5D)
.put("perAttemptRecvTimeout", "2.5s")
.put("retryableStatusCodes", ImmutableList.of())
.build();
Map<String, ?> methodConfig = ImmutableMap.of(
"name", ImmutableList.of(ImmutableMap.of()), "retryPolicy", retryPolicy);
Map<String, ?> rawServiceConfig =
ImmutableMap.of("methodConfig", ImmutableList.of(methodConfig));
assertThat(ManagedChannelServiceConfig.fromServiceConfig(rawServiceConfig, true, 5, 5, null))
.isNotNull();
}

@Test
public void retryConfig_PerAttemptRecvTimeoutUnsetAllowedIfRetryableStatusCodesNonempty() {
Map<String, ?> retryPolicy = ImmutableMap.<String, Object>builder()
.put("maxAttempts", 3.0D)
.put("initialBackoff", "1s")
.put("maxBackoff", "10s")
.put("backoffMultiplier", 1.5D)
.put("retryableStatusCodes", ImmutableList.of("UNAVAILABLE"))
.build();
Map<String, ?> methodConfig = ImmutableMap.of(
"name", ImmutableList.of(ImmutableMap.of()), "retryPolicy", retryPolicy);
Map<String, ?> rawServiceConfig =
ImmutableMap.of("methodConfig", ImmutableList.of(methodConfig));
assertThat(ManagedChannelServiceConfig.fromServiceConfig(rawServiceConfig, true, 5, 5, null))
.isNotNull();
}

@Test
public void retryConfig_emptyRetriableStatusCodesNotAllowedWithPerAttemptRecvTimeoutUnset() {
Map<String, ?> retryPolicy = ImmutableMap.<String, Object>builder()
.put("maxAttempts", 3.0D)
.put("initialBackoff", "1s")
.put("maxBackoff", "10s")
.put("backoffMultiplier", 1.5D)
.put("retryableStatusCodes", ImmutableList.of())
.build();
Map<String, ?> methodConfig = ImmutableMap.of(
"name", ImmutableList.of(ImmutableMap.of()), "retryPolicy", retryPolicy);
Map<String, ?> rawServiceConfig =
ImmutableMap.of("methodConfig", ImmutableList.of(methodConfig));
try {
ManagedChannelServiceConfig.fromServiceConfig(rawServiceConfig, true, 5, 5, null);
fail("The expected IllegalArgumentException is not thrown");
} catch (IllegalArgumentException e) {
assertThat(e).hasMessageThat().contains(
"retryableStatusCodes cannot be empty without perAttemptRecvTimeout");
}
}

// For now we allow perAttemptRecvTimeout being zero although it does not make sense.
// TODO(zdapeng): disallow zero perAttemptRecvTimeout if hedging is not enabled once we support

@ejona86 ejona86 Jul 7, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Once retries are no longer experimental, we can't be adding a restriction. Right? And it sounded like your plan was to try to get retries stable here before supporting hedging.

@dapengzhang0 dapengzhang0 Jul 12, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Once retries are no longer experimental, we can't be adding a restriction. Right?

That's right.

And it sounded like your plan was to try to get retries stable here before supporting hedging.

Why? I'm not.

I'm currently allowing zero perAttemptRecvTimeout, because in the future, we have usecase for hedging.
Once hedge_on_per_try_timeout is supported, we will add restriction for zero perAttemptRecvTimeout without hedge_on_per_try_timeout=true. The planned change is still before getting retries stable.

@ejona86 ejona86 Jul 12, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Oh, I see.

// hedge_on_per_try_timeout.
@Test
public void retryConfig_AllowPerAttemptRecvTimeoutZero() {
Map<String, ?> retryPolicy = ImmutableMap.<String, Object>builder()
.put("maxAttempts", 3.0D)
.put("initialBackoff", "1s")
.put("maxBackoff", "10s")
.put("backoffMultiplier", 1.5D)
.put("perAttemptRecvTimeout", "0s")
.put("retryableStatusCodes", ImmutableList.of())
.build();
Map<String, ?> methodConfig = ImmutableMap.of(
"name", ImmutableList.of(ImmutableMap.of()), "retryPolicy", retryPolicy);
Map<String, ?> rawServiceConfig =
ImmutableMap.of("methodConfig", ImmutableList.of(methodConfig));
assertThat(ManagedChannelServiceConfig.fromServiceConfig(rawServiceConfig, true, 5, 5, null))
.isNotNull();
}

private static MethodDescriptor<?, ?> methodForName(String service, String method) {
return MethodDescriptor.<Void, Void>newBuilder()
.setFullMethodName(service + "/" + method)
Expand Down
1 change: 1 addition & 0 deletions core/src/test/java/io/grpc/internal/RetriableStreamTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -120,6 +120,7 @@ public double nextDouble() {
TimeUnit.SECONDS.toNanos(INITIAL_BACKOFF_IN_SECONDS),
TimeUnit.SECONDS.toNanos(MAX_BACKOFF_IN_SECONDS),
BACKOFF_MULTIPLIER,
null,
ImmutableSet.of(RETRIABLE_STATUS_CODE_1, RETRIABLE_STATUS_CODE_2));
private static final HedgingPolicy HEDGING_POLICY =
new HedgingPolicy(
Expand Down
3 changes: 3 additions & 0 deletions core/src/test/java/io/grpc/internal/RetryPolicyTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -78,6 +78,7 @@ public void getRetryPolicies() throws Exception {
TimeUnit.MILLISECONDS.toNanos(2100),
TimeUnit.MILLISECONDS.toNanos(2200),
parseDouble("3"),
null,
ImmutableSet.of(Code.UNAVAILABLE, Code.RESOURCE_EXHAUSTED)));

method = builder.setFullMethodName("SimpleService1/Foo1").build();
Expand All @@ -87,6 +88,7 @@ public void getRetryPolicies() throws Exception {
TimeUnit.MILLISECONDS.toNanos(100),
TimeUnit.MILLISECONDS.toNanos(1000),
parseDouble("2"),
null,
ImmutableSet.of(Code.UNAVAILABLE)));

method = builder.setFullMethodName("SimpleService2/not_exist").build();
Expand All @@ -99,6 +101,7 @@ public void getRetryPolicies() throws Exception {
TimeUnit.MILLISECONDS.toNanos(100),
TimeUnit.MILLISECONDS.toNanos(1000),
parseDouble("2"),
null,
ImmutableSet.of(Code.UNAVAILABLE)));
} finally {
if (reader != null) {
Expand Down