Skip to content

perf: use power-of-two random choices for channel selection to avoid thundering herd #235

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
rahul2393 merged 5 commits into from
Apr 10, 2026
Merged

perf: use power-of-two random choices for channel selection to avoid thundering herd #235

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
5ce9933
feat: use power-of-two random choices for channel selection to avoid …
rahul2393 Apr 10, 2026
8fcd27d
fix tests
rahul2393 Apr 10, 2026
c7f6076
address comments
rahul2393 Apr 10, 2026
76be1c6
include fallback case
rahul2393 Apr 10, 2026
eefff1e
refactoring
rahul2393 Apr 10, 2026
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
193 changes: 148 additions & 45 deletions grpc-gcp/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 @@ -68,7 +68,9 @@
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
Expand Down Expand Up @@ -122,6 +124,8 @@ public class GcpManagedChannel extends ManagedChannel {
private Duration scaleDownInterval = Duration.ZERO;
private boolean isDynamicScalingEnabled = false;
private int maxConcurrentStreamsLowWatermark = DEFAULT_MAX_STREAM;
private GcpManagedChannelOptions.ChannelPickStrategy channelPickStrategy =

@olavloite olavloite Apr 10, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

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

For a follow-up PR: This field (and probably most of the other fields here) can be made final.

@rahul2393 rahul2393 Apr 10, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

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

Agreed, will address in a follow-up. Most of these fields are set once in initOptions() during construction and never mutated after.

GcpManagedChannelOptions.ChannelPickStrategy.POWER_OF_TWO;
private Duration affinityKeyLifetime = Duration.ZERO;

@VisibleForTesting final Map<String, AffinityConfig> methodToAffinity = new HashMap<>();
Expand Down Expand Up @@ -179,8 +183,12 @@ public class GcpManagedChannel extends ManagedChannel {
private final String metricPoolIndex =
String.format("pool-%d", channelPoolIndex.incrementAndGet());
private final Map<String, Long> cumulativeMetricValues = new ConcurrentHashMap<>();
private final ScheduledExecutorService backgroundService =
Executors.newSingleThreadScheduledExecutor(GcpThreadFactory.newThreadFactory("gcp-mc-bg-%d"));
private static final ScheduledThreadPoolExecutor SHARED_BACKGROUND_SERVICE =
createSharedBackgroundService();

private ScheduledFuture<?> cleanupTask;
private ScheduledFuture<?> scaleDownTask;
private ScheduledFuture<?> logMetricsTask;

// Metrics counters.
private final AtomicInteger readyChannels = new AtomicInteger();
Expand Down Expand Up @@ -223,6 +231,17 @@ public class GcpManagedChannel extends ManagedChannel {
private AtomicLong scaleUpCount = new AtomicLong();
private AtomicLong scaleDownCount = new AtomicLong();

private static ScheduledThreadPoolExecutor createSharedBackgroundService() {
ScheduledThreadPoolExecutor executor =
new ScheduledThreadPoolExecutor(
Math.max(2, Math.min(4, Runtime.getRuntime().availableProcessors() / 2)),
GcpThreadFactory.newThreadFactory("gcp-mc-bg-%d"));
executor.setRemoveOnCancelPolicy(true);
executor.setExecuteExistingDelayedTasksAfterShutdownPolicy(false);
executor.setContinueExistingPeriodicTasksAfterShutdownPolicy(false);
return executor;
}

/**
* Constructor for GcpManagedChannel.
*
Expand Down Expand Up @@ -396,6 +415,7 @@ private void initOptions() {
scaleDownInterval = poolOptions.getScaleDownInterval();
isDynamicScalingEnabled =
minRpcPerChannel > 0 && maxRpcPerChannel > 0 && !scaleDownInterval.isZero();
channelPickStrategy = poolOptions.getChannelPickStrategy();
}
initMetrics();
}
Expand All @@ -404,27 +424,30 @@ private synchronized void initCleanupTask(Duration cleanupInterval) {
if (cleanupInterval.isZero()) {
return;
}
backgroundService.scheduleAtFixedRate(
this::cleanupAffinityKeys,
cleanupInterval.toMillis(),
cleanupInterval.toMillis(),
MILLISECONDS);
cleanupTask =
SHARED_BACKGROUND_SERVICE.scheduleAtFixedRate(
this::cleanupAffinityKeys,
cleanupInterval.toMillis(),
cleanupInterval.toMillis(),
MILLISECONDS);
}

private synchronized void initScaleDownChecker(Duration scaleDownInterval) {
if (!isDynamicScalingEnabled || scaleDownInterval.isZero()) {
return;
}

backgroundService.scheduleAtFixedRate(
this::checkScaleDown,
scaleDownInterval.toMillis(),
scaleDownInterval.toMillis(),
MILLISECONDS);
scaleDownTask =
SHARED_BACKGROUND_SERVICE.scheduleAtFixedRate(
this::checkScaleDown,
scaleDownInterval.toMillis(),
scaleDownInterval.toMillis(),
MILLISECONDS);
}

private synchronized void initLogMetrics() {
backgroundService.scheduleAtFixedRate(this::logMetrics, 60, 60, SECONDS);
logMetricsTask =
SHARED_BACKGROUND_SERVICE.scheduleAtFixedRate(this::logMetrics, 60, 60, SECONDS);
}

private void logMetricsOptions() {
Expand Down Expand Up @@ -1757,8 +1780,57 @@ private ChannelRef pickLeastBusyChannel(boolean forFallback) {
return first;
}

// Pick the least busy channel and the least busy ready and not overloaded channel (this could
// be the same channel or different or no channel).
if (!fallbackEnabled) {
return pickLeastBusyNoFallback();
}

return pickLeastBusyWithFallback(forFallback);
}

/**
* Non-fallback channel selection. Uses the configured {@link
* GcpManagedChannelOptions.ChannelPickStrategy}.
*/
private ChannelRef pickLeastBusyNoFallback() {
ChannelRef channelCandidate;
int minStreams;

if (channelPickStrategy == GcpManagedChannelOptions.ChannelPickStrategy.POWER_OF_TWO) {
channelCandidate = pickPowerOfTwo();
// With power-of-two, streams distribute approximately (not exactly) evenly.
// Use max streams for scale-up: if ANY channel hits the watermark, it's overloaded now
// and we should add capacity before other channels follow. This preserves the original
// per-channel watermark semantics (with LINEAR_SCAN, min == max so it didn't matter).
// Global min would delay scale-up; sampled min would be noisy.
minStreams = getMaxActiveStreams();
} else {
channelCandidate = channelRefs.get(0);
minStreams = channelCandidate.getActiveStreamsCount();
for (ChannelRef channelRef : channelRefs) {
int cnt = channelRef.getActiveStreamsCount();
if (cnt < minStreams) {
minStreams = cnt;
channelCandidate = channelRef;
}
}
}

if (shouldScaleUp(minStreams)) {
ChannelRef newChannel = tryCreateNewChannel();
if (newChannel != null) {
scaleUpCount.incrementAndGet();
return newChannel;
}
}
return channelCandidate;
}

/**
* Fallback-enabled channel selection. Always uses a full linear scan because the fallback logic
* needs to filter channels by readiness state and max stream limits.
*/
private ChannelRef pickLeastBusyWithFallback(boolean forFallback) {
// Full scan required: readyCandidate must be filtered by fallbackMap and DEFAULT_MAX_STREAM.
Comment thread
rahul2393 marked this conversation as resolved.
Outdated
ChannelRef channelCandidate = channelRefs.get(0);
int minStreams = channelCandidate.getActiveStreamsCount();
ChannelRef readyCandidate = null;
Expand All @@ -1778,17 +1850,6 @@ private ChannelRef pickLeastBusyChannel(boolean forFallback) {
}
}

if (!fallbackEnabled) {
if (shouldScaleUp(minStreams)) {
ChannelRef newChannel = tryCreateNewChannel();
if (newChannel != null) {
scaleUpCount.incrementAndGet();
return newChannel;
}
}
return channelCandidate;
}

if (shouldScaleUp(readyMinStreams)) {
ChannelRef newChannel = tryCreateNewChannel();
if (newChannel != null) {
Expand Down Expand Up @@ -1825,6 +1886,41 @@ private ChannelRef pickLeastBusyChannel(boolean forFallback) {
return channelCandidate;
}

/**
* Power-of-two random choices: pick two channels at random and return the less busy one. On tie,
* prefer the channel with more recent activity (warmer) to preserve connection warmth under low
* traffic.
*/
private ChannelRef pickPowerOfTwo() {
int size = channelRefs.size();
if (size == 1) {
return channelRefs.get(0);
}

ThreadLocalRandom random = ThreadLocalRandom.current();
int i = random.nextInt(size);
int j = random.nextInt(size - 1);
if (j >= i) {
j++;
}

ChannelRef a = channelRefs.get(i);
ChannelRef b = channelRefs.get(j);

int aStreams = a.getActiveStreamsCount();
int bStreams = b.getActiveStreamsCount();

if (aStreams < bStreams) {
return a;
}
if (bStreams < aStreams) {
return b;
}

// Tie: prefer the warmer channel (more recent activity) to preserve connection warmth.
return a.lastResponseNanos >= b.lastResponseNanos ? a : b;
}

@Override
public String authority() {
if (!channelRefs.isEmpty()) {
Expand Down Expand Up @@ -1882,6 +1978,21 @@ private String keyFromOptsCtx(CallOptions callOptions) {
return key;
}

private void cancelBackgroundTasks() {
Comment thread
rahul2393 marked this conversation as resolved.
Outdated
if (cleanupTask != null) {
cleanupTask.cancel(false);
cleanupTask = null;
}
if (scaleDownTask != null) {
scaleDownTask.cancel(false);
scaleDownTask = null;
}
if (logMetricsTask != null) {
logMetricsTask.cancel(false);
logMetricsTask = null;
}
}

@Override
public ManagedChannel shutdownNow() {
logger.finer(log("Shutdown now started."));
Expand All @@ -1895,9 +2006,7 @@ public ManagedChannel shutdownNow() {
channelRef.getChannel().shutdownNow();
}
}
if (backgroundService != null && !backgroundService.isTerminated()) {
backgroundService.shutdownNow();
}
cancelBackgroundTasks();
if (!stateNotificationExecutor.isTerminated()) {
stateNotificationExecutor.shutdownNow();
}
Expand All @@ -1913,9 +2022,7 @@ public ManagedChannel shutdown() {
for (ChannelRef channelRef : removedChannelRefs) {
channelRef.getChannel().shutdown();
}
if (backgroundService != null) {
backgroundService.shutdown();
}
cancelBackgroundTasks();
stateNotificationExecutor.shutdown();
return this;
}
Expand All @@ -1936,10 +2043,6 @@ public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedE
channelRef.getChannel().awaitTermination(awaitTimeNanos, NANOSECONDS);
}
long awaitTimeNanos = endTimeNanos - System.nanoTime();
if (backgroundService != null && awaitTimeNanos > 0) {
//noinspection ResultOfMethodCallIgnored
backgroundService.awaitTermination(awaitTimeNanos, NANOSECONDS);
}
awaitTimeNanos = endTimeNanos - System.nanoTime();
if (awaitTimeNanos > 0) {
// noinspection ResultOfMethodCallIgnored
Expand All @@ -1957,10 +2060,10 @@ public boolean isShutdown() {
return false;
}
}
if (backgroundService != null && !backgroundService.isShutdown()) {
return false;
}
return stateNotificationExecutor.isShutdown();
return cleanupTask == null
&& scaleDownTask == null
&& logMetricsTask == null
&& stateNotificationExecutor.isShutdown();
}

@Override
Expand All @@ -1972,10 +2075,10 @@ public boolean isTerminated() {
return false;
}
}
if (backgroundService != null && !backgroundService.isTerminated()) {
return false;
}
return stateNotificationExecutor.isTerminated();
return cleanupTask == null
&& scaleDownTask == null
&& logMetricsTask == null
&& stateNotificationExecutor.isTerminated();
}

/** Get the current connectivity state of the channel pool. */
Expand Down
Loading