Skip to content

feat(grpc-gcp): penalize retryable channel errors #14219

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

Open
rahul2393 wants to merge 1 commit into
base: main
Choose a base branch
from
+772 −20
Open

feat(grpc-gcp): penalize retryable channel errors #14219

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
205 changes: 191 additions & 14 deletions grpc-gcp-java/src/main/java/com/google/cloud/grpc/GcpManagedChannel.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,6 +87,11 @@ public class GcpManagedChannel extends ManagedChannel {
private static final Logger logger = Logger.getLogger(GcpManagedChannel.class.getName());
static final AtomicInteger channelPoolIndex = new AtomicInteger();

@FunctionalInterface
interface NanoClock {
long get();
}

// Counter for tracking channel ids.
final AtomicInteger nextChannelId = new AtomicInteger();
static final int DEFAULT_MAX_CHANNEL = 10;
Expand Down Expand Up @@ -173,6 +178,9 @@ private static int stateFromChannelId(int channelId) {
private int maxScaleUpPercent = 30;
private int maxScaleDownChannels = 2;
private Duration drainIdleGrace = Duration.ofMinutes(1);
private int errorPenaltyStep = 5;
private Duration errorPenaltyDuration = Duration.ofSeconds(5);
private long errorPenaltyDurationNanos = Duration.ofSeconds(5).toNanos();
private boolean isDynamicScalingEnabled = false;
private int maxConcurrentStreamsLowWatermark = DEFAULT_MAX_STREAM;
private GcpManagedChannelOptions.ChannelPickStrategy channelPickStrategy =
Expand Down Expand Up @@ -202,6 +210,7 @@ private static int stateFromChannelId(int channelId) {
// One-slot scale-up signal. At most one worker mutates pool size at a time.
private final AtomicBoolean scaleUpSignalPending = new AtomicBoolean();
private final AtomicBoolean scaleUpWorkerRunning = new AtomicBoolean();
private final AtomicLong totalErrorPenaltyLoad = new AtomicLong();

private volatile long lastScaleUpNanos = Long.MIN_VALUE;
private int consecutiveLowLoadChecks;
Expand Down Expand Up @@ -295,14 +304,14 @@ private static int stateFromChannelId(int channelId) {
private AtomicLong scaleDownCount = new AtomicLong();

// Clock supplier for nanoTime, injectable for testing.
private Supplier<Long> nanoClock = System::nanoTime;
private NanoClock nanoClock = System::nanoTime;
private IntUnaryOperator candidateIndexPicker =
bound -> ThreadLocalRandom.current().nextInt(bound);
@Nullable private volatile Consumer<ChannelRef> pickerValidationHookForTest;
@Nullable private volatile Runnable inactiveMappingRemovalHookForTest;

@VisibleForTesting
void setNanoClock(Supplier<Long> nanoClock) {
void setNanoClock(NanoClock nanoClock) {
this.nanoClock = nanoClock;
}

Expand Down Expand Up @@ -467,6 +476,9 @@ void checkScaleDown() {
int removeCount = Math.min(maxScaleDownChannels, Math.max(0, channelCount - desiredSize));
removedChannels = removeChannels(removeCount);
}
for (ChannelRef channelRef : removedChannels) {
channelRef.clearErrorPenalty();
}
List<String> keysToUnbind =
affinityKeyToChannelRef.entrySet().stream()
.filter(entry -> removedChannels.contains(entry.getValue()))
Expand Down Expand Up @@ -632,6 +644,43 @@ private static int ceilDiv(long numerator, int denominator) {
return (int) Math.min(Integer.MAX_VALUE, 1 + ((numerator - 1) / denominator));
}

private static int ceilMultiplyDivide(int factor, long numerator, long denominator) {
if (factor <= 0 || numerator <= 0 || denominator <= 0) {
return 0;
}
if (numerator >= denominator) {
return factor;
}
if (numerator <= Long.MAX_VALUE / factor) {
long product = factor * numerator;
return (int) (product / denominator + (product % denominator == 0 ? 0 : 1));
}

// Overflow-safe binary long division for unusually large configured durations or penalties.
long quotient = 0;
long remainder = 0;
for (int bit = Integer.highestOneBit(factor); bit != 0; bit >>>= 1) {
quotient <<= 1;
long denominatorMinusRemainder = denominator - remainder;
if (remainder >= denominatorMinusRemainder) {
remainder -= denominatorMinusRemainder;
quotient++;
} else {
remainder += remainder;
}
if ((factor & bit) != 0) {
long denominatorMinusNumerator = denominator - numerator;
if (remainder >= denominatorMinusNumerator) {
remainder -= denominatorMinusNumerator;
quotient++;
} else {
remainder += numerator;
}
}
}
return (int) (quotient + (remainder == 0 ? 0 : 1));
}

private int targetRpcPerChannel() {
return Math.max(1, (minRpcPerChannel + maxRpcPerChannel) / 2);
}
Expand All @@ -646,11 +695,12 @@ private long activeLoad(List<ChannelRef> refs) {
return load;
}

private long pickerLoad(List<ChannelRef> refs) {
private long pickerLoad(List<ChannelRef> refs, long now) {
// Expiry accounting is swept before dynamicUpscale acquires the pool monitor.
long load = 0;
for (ChannelRef channelRef : refs) {
if (channelRef.isActive()) {
load += channelRef.getPickerLoad();
load += channelRef.getPickerLoad(now);
}
}
return load;
Expand Down Expand Up @@ -689,6 +739,10 @@ private void initOptions() {
maxScaleUpPercent = poolOptions.getMaxScaleUpPercent();
maxScaleDownChannels = poolOptions.getMaxScaleDownChannels();
drainIdleGrace = poolOptions.getDrainIdleGrace();
errorPenaltyStep = poolOptions.getErrorPenaltyStep();
errorPenaltyDuration = poolOptions.getErrorPenaltyDuration();
errorPenaltyDurationNanos =
errorPenaltyDuration.isNegative() ? 0 : errorPenaltyDuration.toNanos();
isDynamicScalingEnabled =
minRpcPerChannel > 0 && maxRpcPerChannel > 0 && !scaleDownInterval.isZero();
channelPickStrategy = poolOptions.getChannelPickStrategy();
Expand Down Expand Up @@ -2058,11 +2112,12 @@ private ChannelRef pickLeastBusyChannelDifferentFrom(@Nullable ChannelRef exclud
}
ChannelRef leastBusyChannelRef = null;
int leastBusyStreams = Integer.MAX_VALUE;
long now = nanoClock.get();
for (ChannelRef candidate : channelRefs) {
if (candidate == excludedChannelRef || !candidate.isActive()) {
continue;
}
int streams = candidate.getPickerLoad();
int streams = candidate.getPickerLoad(now);
if (leastBusyChannelRef == null || streams < leastBusyStreams) {
leastBusyChannelRef = candidate;
leastBusyStreams = streams;
Expand Down Expand Up @@ -2131,8 +2186,10 @@ private void maybeSignalScaleUp(ChannelRef selectedChannel) {
|| activeChannels >= maxSize) {
return;
}
if (selectedChannel.getActiveStreamsCount() <= maxRpcPerChannel
&& ((double) totalActiveStreams.get() / activeChannels) <= maxRpcPerChannel) {
int selectedLoad = selectedChannel.getPickerLoad();
long totalLoad = (long) totalActiveStreams.get() + totalErrorPenaltyLoad.get();
// A pool at the average cap has no spare capacity, including when penalty represents loss.
if (selectedLoad <= maxRpcPerChannel && totalLoad < (long) activeChannels * maxRpcPerChannel) {
return;
}
signalScaleUp();
Expand Down Expand Up @@ -2171,12 +2228,15 @@ private void runScaleUpWorker() {
}

private void dynamicUpscale() {
long now = nanoClock.get();
for (ChannelRef channelRef : channelRefs) {
channelRef.currentErrorPenalty(now);
}
final int channelsToBuild;
synchronized (this) {
if (!isDynamicScalingEnabled || shuttingDown || channelRefs.size() >= maxSize) {
return;
}
long now = nanoClock.get();
if (lastScaleUpNanos != Long.MIN_VALUE
&& now - lastScaleUpNanos < scaleUpCooldown.toNanos()) {
return;
Expand All @@ -2187,7 +2247,7 @@ private void dynamicUpscale() {
if (active == 0) {
return;
}
int desired = ceilDiv(pickerLoad(activeChannels), targetRpcPerChannel());
int desired = ceilDiv(pickerLoad(activeChannels, now), targetRpcPerChannel());
int add = desired - active;
// Small pools may add two channels per event before percentage growth dominates.
int percentCap = Math.max(2, ceilDiv((long) active * maxScaleUpPercent, 100));
Expand Down Expand Up @@ -2304,12 +2364,13 @@ private ChannelRef pickLeastBusyWithFallback(boolean forFallback) {
ChannelRef overallCandidate = null;
int overallMinStreams = Integer.MAX_VALUE;
int readyMaxStreams = 0;
long now = nanoClock.get();

for (ChannelRef channelRef : channelRefs) {
if (!channelRef.isActive()) {
continue;
}
int cnt = channelRef.getPickerLoad();
int cnt = channelRef.getPickerLoad(now);
if (overallCandidate == null || cnt < overallMinStreams) {
overallMinStreams = cnt;
overallCandidate = channelRef;
Expand All @@ -2323,7 +2384,7 @@ private ChannelRef pickLeastBusyWithFallback(boolean forFallback) {
}

if (overallCandidate == null) {
return leastLoadedActiveChannel(channelRefs);
return leastLoadedActiveChannel(channelRefs, now);
}

// For scale-up, use maxStreams among ready channels (consistent with non-fallback path).
Expand Down Expand Up @@ -2396,11 +2457,20 @@ ChannelRef pickFromCandidates(List<ChannelRef> candidates) {
}

private ChannelRef leastLoadedActiveChannel(List<ChannelRef> candidates) {
return leastLoadedActiveChannel(candidates, nanoClock.get());
}

private ChannelRef leastLoadedActiveChannel(List<ChannelRef> candidates, long now) {
ChannelRef best = null;
int bestLoad = Integer.MAX_VALUE;
for (ChannelRef candidate : candidates) {
if (candidate.isActive()
&& (best == null || candidate.getPickerLoad() < best.getPickerLoad())) {
if (!candidate.isActive()) {
continue;
}
int candidateLoad = candidate.getPickerLoad(now);
if (best == null || candidateLoad < bestLoad) {
best = candidate;
bestLoad = candidateLoad;
}
}
if (best != null) {
Expand Down Expand Up @@ -2822,6 +2892,10 @@ protected class ChannelRef {
private final long createdNanos = nanoClock.get();
private volatile long lastActivityNanos = createdNanos;
private long lastResponseNanos = createdNanos;

private volatile int errorPenaltyLoad;
private volatile long errorPenaltyExpiresAtNanos;

private final AtomicInteger deadlineExceededCount = new AtomicInteger();
private final AtomicLong okCalls = new AtomicLong();
private final AtomicLong errCalls = new AtomicLong();
Expand Down Expand Up @@ -2938,6 +3012,7 @@ protected void activeStreamsCountDecr(long startNanos, Status status, boolean fr
if (unresponsiveDetectionEnabled) {
detectUnresponsiveConnection(startNanos, status, fromClientSide);
}
applyErrorPenalty(status);
if (actStreams == 0 && !isActive()) {
scheduleDrain(this);
}
Expand All @@ -2957,7 +3032,109 @@ protected int getActiveStreamsCount() {
}

protected int getPickerLoad() {
return getActiveStreamsCount();
return (int)
Math.min(Integer.MAX_VALUE, (long) getActiveStreamsCount() + currentErrorPenalty());
}

@VisibleForTesting
int currentErrorPenalty() {
int penalty = errorPenaltyLoad;
if (penalty == 0) {
return 0;
}
return currentErrorPenalty(nanoClock.get());
}

private int currentErrorPenalty(long now) {
Comment thread
rahul2393 marked this conversation as resolved.
int penalty = errorPenaltyLoad;
if (penalty == 0) {
return 0;
}
if (now - errorPenaltyExpiresAtNanos < 0) {
return decayedErrorPenalty(penalty, now);
}
int expiredPenalty;
synchronized (this) {
penalty = errorPenaltyLoad;
if (penalty == 0) {
return 0;
}
if (now - errorPenaltyExpiresAtNanos < 0) {
return decayedErrorPenalty(penalty, now);
}
errorPenaltyLoad = 0;
errorPenaltyExpiresAtNanos = 0;
expiredPenalty = penalty;
}
totalErrorPenaltyLoad.addAndGet(-expiredPenalty);
return 0;
}

private int getPickerLoad(long now) {
int penalty = errorPenaltyLoad;
if (penalty != 0) {
penalty = decayedErrorPenalty(penalty, now);
}
return (int) Math.min(Integer.MAX_VALUE, (long) getActiveStreamsCount() + penalty);
}

private int decayedErrorPenalty(int penalty, long now) {
long remainingNanos = errorPenaltyExpiresAtNanos - now;
if (remainingNanos <= 0) {
return 0;
}
// Picker steering decays smoothly. Aggregate accounting deliberately retains the full
// contribution until expiry or clear, making scale-up load a conservative upper bound while
// preserving one atomic net delta per aggregate transition.
return ceilMultiplyDivide(
penalty, Math.min(remainingNanos, errorPenaltyDurationNanos), errorPenaltyDurationNanos);
}

private void applyErrorPenalty(Status status) {
if (!isDynamicScalingEnabled
|| !active
Comment thread
rahul2393 marked this conversation as resolved.
|| errorPenaltyStep == 0
|| errorPenaltyDuration.isZero()
|| errorPenaltyDuration.isNegative()
|| (status.getCode() != Code.UNAVAILABLE
&& status.getCode() != Code.RESOURCE_EXHAUSTED)) {

@olavloite olavloite Sep 1, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Should we indiscriminately apply errors for all RESOURCE_EXHAUSTED errors? Or only the ones that include certain metadata? (e.g. a RetryInfo)

@rahul2393 rahul2393 Sep 1, 2026

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.

In design review it was agreed that we will apply for all RESOURCE_EXHAUSTED errors so want to keep same here

return;
}
long now = nanoClock.get();
long addedPenalty;
synchronized (this) {
if (!active) {
return;
}
int previousContribution = errorPenaltyLoad;
int current =
previousContribution != 0 && now - errorPenaltyExpiresAtNanos < 0
? previousContribution
: 0;
int next = (int) Math.min(maxRpcPerChannel, (long) current + errorPenaltyStep);
Comment thread
rahul2393 marked this conversation as resolved.
errorPenaltyExpiresAtNanos = now + errorPenaltyDurationNanos;
errorPenaltyLoad = next;
addedPenalty = (long) next - previousContribution;
}
if (addedPenalty != 0) {
totalErrorPenaltyLoad.addAndGet(addedPenalty);
}
if (addedPenalty > 0) {
maybeSignalScaleUp(this);
}
}

private void clearErrorPenalty() {
int clearedPenalty;
synchronized (this) {
clearedPenalty = errorPenaltyLoad;
if (clearedPenalty == 0) {
return;
}
errorPenaltyLoad = 0;
errorPenaltyExpiresAtNanos = 0;
}
totalErrorPenaltyLoad.addAndGet(-clearedPenalty);
}

protected long getAndResetOkCalls() {
Expand Down
Loading
Loading