diff --git a/grpc-gcp-java/src/main/java/com/google/cloud/grpc/GcpManagedChannel.java b/grpc-gcp-java/src/main/java/com/google/cloud/grpc/GcpManagedChannel.java index 8c8f2a302972..7c1ab00e5325 100644 --- a/grpc-gcp-java/src/main/java/com/google/cloud/grpc/GcpManagedChannel.java +++ b/grpc-gcp-java/src/main/java/com/google/cloud/grpc/GcpManagedChannel.java @@ -56,7 +56,6 @@ import java.util.Collections; import java.util.Comparator; import java.util.HashMap; -import java.util.HashSet; import java.util.LinkedList; import java.util.List; import java.util.LongSummaryStatistics; @@ -72,6 +71,7 @@ import java.util.concurrent.ScheduledThreadPoolExecutor; import java.util.concurrent.ThreadLocalRandom; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; import java.util.function.IntUnaryOperator; @@ -162,8 +162,9 @@ private static int stateFromChannelId(int channelId) { private int minRpcPerChannel = 0; private int maxRpcPerChannel = 0; private Duration scaleDownInterval = Duration.ZERO; + private Duration scaleUpCooldown = Duration.ofSeconds(10); private int scaleDownConsecutiveLowLoadChecks = 3; - private int consecutiveLowLoadChecks; + private int maxScaleUpPercent = 30; private int maxScaleDownChannels = 2; private boolean isDynamicScalingEnabled = false; private int maxConcurrentStreamsLowWatermark = DEFAULT_MAX_STREAM; @@ -186,7 +187,15 @@ private static int stateFromChannelId(int channelId) { private final Map channelIdToChannelRef = new ConcurrentHashMap<>(); // A set of channels that we removed from the pool and wait for their RPCs to be completed before // we can shut them down. - final Set removedChannelRefs = new HashSet<>(); + final Set removedChannelRefs = ConcurrentHashMap.newKeySet(); + + // 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 volatile long lastScaleUpNanos = Long.MIN_VALUE; + private int consecutiveLowLoadChecks; + private volatile boolean shuttingDown; private final ExecutorService stateNotificationExecutor = Executors.newCachedThreadPool( @@ -279,6 +288,11 @@ private static int stateFromChannelId(int channelId) { private IntUnaryOperator candidateIndexPicker = bound -> ThreadLocalRandom.current().nextInt(bound); + @VisibleForTesting + void setNanoClock(Supplier nanoClock) { + this.nanoClock = nanoClock; + } + @VisibleForTesting void setCandidateIndexPickerForTest(IntUnaryOperator candidateIndexPicker) { this.candidateIndexPicker = candidateIndexPicker; @@ -294,11 +308,6 @@ int readyChannelCountForTest() { return readyChannels.get(); } - @VisibleForTesting - void setNanoClock(Supplier nanoClock) { - this.nanoClock = nanoClock; - } - private static ScheduledThreadPoolExecutor createSharedBackgroundService() { ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor( @@ -489,7 +498,9 @@ private void initOptions() { minRpcPerChannel = poolOptions.getMinRpcPerChannel(); maxRpcPerChannel = poolOptions.getMaxRpcPerChannel(); scaleDownInterval = poolOptions.getScaleDownInterval(); + scaleUpCooldown = poolOptions.getScaleUpCooldown(); scaleDownConsecutiveLowLoadChecks = poolOptions.getScaleDownConsecutiveLowLoadChecks(); + maxScaleUpPercent = poolOptions.getMaxScaleUpPercent(); maxScaleDownChannels = poolOptions.getMaxScaleDownChannels(); isDynamicScalingEnabled = minRpcPerChannel > 0 && maxRpcPerChannel > 0 && !scaleDownInterval.isZero(); @@ -517,7 +528,13 @@ private synchronized void initScaleDownChecker(Duration scaleDownInterval) { scaleDownTask = SHARED_BACKGROUND_SERVICE.scheduleAtFixedRate( - this::checkScaleDown, + () -> { + try { + checkScaleDown(); + } catch (Throwable failure) { + logger.log(Level.WARNING, log("Scale-down check failed"), failure); + } + }, scaleDownInterval.toMillis(), scaleDownInterval.toMillis(), MILLISECONDS); @@ -1650,11 +1667,19 @@ public int getStreamsLowWatermark() { } public int getMinActiveStreams() { - return channelRefs.stream().mapToInt(ChannelRef::getActiveStreamsCount).min().orElse(0); + return channelRefs.stream() + .filter(ChannelRef::isActive) + .mapToInt(ChannelRef::getActiveStreamsCount) + .min() + .orElse(0); } public int getMaxActiveStreams() { - return channelRefs.stream().mapToInt(ChannelRef::getActiveStreamsCount).max().orElse(0); + return channelRefs.stream() + .filter(ChannelRef::isActive) + .mapToInt(ChannelRef::getActiveStreamsCount) + .max() + .orElse(0); } /** @@ -1688,15 +1713,24 @@ protected ChannelRef getChannelRefForBind() { * @return {@link ChannelRef} */ protected synchronized ChannelRef getChannelRefRoundRobin() { + ChannelRef first = createFirstChannel(); + if (first != null) { + return first; + } if (!isDynamicScalingEnabled && channelRefs.size() < maxSize) { return createNewChannel(); } - maybeDynamicUpscale(); - bindingIndex++; - if (bindingIndex >= channelRefs.size()) { - bindingIndex = 0; + Object[] snapshot = channelRefs.toArray(); + if (snapshot.length > 0) { + for (int attempts = 0; attempts < snapshot.length; attempts++) { + bindingIndex = (bindingIndex + 1) % snapshot.length; + ChannelRef candidate = (ChannelRef) snapshot[bindingIndex]; + if (candidate.isActive()) { + return candidate; + } + } } - return channelRefs.get(bindingIndex); + return pickFromCandidates(channelRefs); } /** @@ -1710,7 +1744,6 @@ protected synchronized ChannelRef getChannelRefRoundRobin() { * Otherwise pick the one with the smallest number of streams. */ protected ChannelRef getChannelRef(@Nullable String key) { - maybeDynamicUpscale(); if (key == null || key.isEmpty()) { return pickLeastBusyChannel(/* forFallback= */ false); } @@ -1777,7 +1810,6 @@ protected ChannelRef getChannelRef(@Nullable String key) { * Pick a {@link ChannelRef} using a caller-owned reference instead of grpc-gcp's affinity map. */ protected ChannelRef getChannelRefByAffinityRef(ChannelAffinityRef affinityRef) { - maybeDynamicUpscale(); // Retry if another thread updates the caller-owned affinity ref while we are picking a channel. while (true) { int state = affinityRef.state.get(); @@ -1853,16 +1885,20 @@ ChannelRef createNewChannel() { } private Optional pickChannelForReuse() { - // Pick the most recently connected, if any. - Optional chRef = - removedChannelRefs.stream().max(Comparator.comparing(ChannelRef::getConnectedSinceNanos)); - - // Make sure it is ready, because connectedSinceNanos may be 0. - if (chRef.isPresent() && chRef.get().getState() != ConnectivityState.READY) { - return Optional.empty(); - } + // Pick the most recently connected ready channel, if any. + return removedChannelRefs.stream() + .filter(channelRef -> channelRef.getState() == ConnectivityState.READY) + .max(Comparator.comparing(ChannelRef::getConnectedSinceNanos)); + } - return chRef; + @GuardedBy("this") + private ChannelRef addBuiltChannel(ManagedChannel channel) { + ChannelRef channelRef = new ChannelRef(channel); + channelRefs.add(channelRef); + channelIdToChannelRef.put(channelRef.getId(), channelRef); + channelRef.activateAndAccountReadiness(); + maxChannels.accumulateAndGet(getNumberOfChannels(), Math::max); + return channelRef; } // Returns first newly created channel or null if there are already some channels in the pool. @@ -1894,25 +1930,133 @@ private ChannelRef tryCreateNewChannel() { return null; } - private void maybeDynamicUpscale() { - if (!isDynamicScalingEnabled || channelRefs.size() >= maxSize) { + private void maybeSignalScaleUp(ChannelRef selectedChannel) { + int activeChannels = channelRefs.size(); + if (!selectedChannel.isActive() + || !isDynamicScalingEnabled + || shuttingDown + || activeChannels == 0 + || activeChannels >= maxSize) { return; } - - if ((totalActiveStreams.get() / channelRefs.size()) >= maxRpcPerChannel) { - dynamicUpscale(); + if (selectedChannel.getActiveStreamsCount() <= maxRpcPerChannel + && ((double) totalActiveStreams.get() / activeChannels) <= maxRpcPerChannel) { + return; } + signalScaleUp(); } - private synchronized void dynamicUpscale() { - if (!isDynamicScalingEnabled || channelRefs.size() >= maxSize) { + private void signalScaleUp() { + scaleUpSignalPending.set(true); + if (!scaleUpWorkerRunning.compareAndSet(false, true)) { return; } + try { + SHARED_BACKGROUND_SERVICE.execute(this::runScaleUpWorker); + } catch (RejectedExecutionException e) { + scaleUpWorkerRunning.set(false); + logger.fine(log("Scale-up task rejected: %s", e.getMessage())); + } + } - if ((totalActiveStreams.get() / channelRefs.size()) >= maxRpcPerChannel) { - createNewChannel(); - scaleUpCount.incrementAndGet(); + private void runScaleUpWorker() { + try { + do { + scaleUpSignalPending.set(false); + try { + dynamicUpscale(); + } catch (Throwable failure) { + logger.log(Level.WARNING, log("Scale-up failed"), failure); + } + } while (scaleUpSignalPending.get() && !shuttingDown); + } finally { + scaleUpWorkerRunning.set(false); + // Close the race where a signal arrives between the final test and clearing running. + if (scaleUpSignalPending.get() && !shuttingDown) { + signalScaleUp(); + } + } + } + + private void dynamicUpscale() { + final int channelsToBuild; + int reused = 0; + synchronized (this) { + if (!isDynamicScalingEnabled || shuttingDown || channelRefs.size() >= maxSize) { + return; + } + long now = nanoClock.get(); + if (lastScaleUpNanos != Long.MIN_VALUE + && now - lastScaleUpNanos < scaleUpCooldown.toNanos()) { + return; + } + int active = channelRefs.size(); + if (active == 0) { + return; + } + int targetRpcPerChannel = Math.max(1, (minRpcPerChannel + maxRpcPerChannel) / 2); + long load = totalActiveStreams.get(); + int desired = + load == 0 + ? active + : (int) Math.min(Integer.MAX_VALUE, 1 + ((load - 1) / targetRpcPerChannel)); + int add = desired - active; + // Small pools may add two channels per event before percentage growth dominates. + int percentCap = Math.max(2, (int) (1 + (((long) active * maxScaleUpPercent - 1) / 100))); + add = Math.min(add, percentCap); + add = Math.min(add, maxSize - active); + if (add <= 0) { + return; + } + while (reused < add) { + Optional reusable = pickChannelForReuse(); + if (!reusable.isPresent()) { + break; + } + ChannelRef channelRef = reusable.get(); + removedChannelRefs.remove(channelRef); + channelRefs.add(channelRef); + channelIdToChannelRef.put(channelRef.getId(), channelRef); + channelRef.activateAndAccountReadiness(); + maxChannels.accumulateAndGet(getNumberOfChannels(), Math::max); + reused++; + } + channelsToBuild = add - reused; + // Claim cooldown before delegate construction begins. + lastScaleUpNanos = now; } + + scaleUpCount.addAndGet(reused); + List builtChannels = new ArrayList<>(channelsToBuild); + try { + for (int i = 0; i < channelsToBuild; i++) { + builtChannels.add(delegateChannelBuilder.build()); + } + } catch (Throwable failure) { + for (ManagedChannel channel : builtChannels) { + try { + channel.shutdownNow(); + } catch (Throwable shutdownFailure) { + failure.addSuppressed(shutdownFailure); + } + } + throw failure; + } + + int added = 0; + List surplus = new ArrayList<>(); + synchronized (this) { + for (ManagedChannel channel : builtChannels) { + if (shuttingDown || channelRefs.size() >= maxSize) { + surplus.add(channel); + } else { + addBuiltChannel(channel); + added++; + } + } + } + surplus.forEach(ManagedChannel::shutdownNow); + scaleUpCount.addAndGet(added); } // This is pre-dynamic scaling functionality where we only scale up when the minimum number of @@ -1952,24 +2096,22 @@ private ChannelRef pickLeastBusyChannel(boolean forFallback) { */ private ChannelRef pickLeastBusyNoFallback() { ChannelRef channelCandidate = pickFromCandidates(channelRefs); - int minStreams; - - if (channelPickStrategy == GcpManagedChannelOptions.ChannelPickStrategy.POWER_OF_TWO) { + if (!isDynamicScalingEnabled && channelRefs.size() < maxSize) { // 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 { - minStreams = channelCandidate.getActiveStreamsCount(); - } - - if (shouldScaleUp(minStreams)) { - ChannelRef newChannel = tryCreateNewChannel(); - if (newChannel != null) { - scaleUpCount.incrementAndGet(); - return newChannel; + int streams = + channelPickStrategy == GcpManagedChannelOptions.ChannelPickStrategy.POWER_OF_TWO + ? getMaxActiveStreams() + : channelCandidate.getActiveStreamsCount(); + if (streams >= maxConcurrentStreamsLowWatermark) { + ChannelRef newChannel = tryCreateNewChannel(); + if (newChannel != null) { + scaleUpCount.incrementAndGet(); + return newChannel; + } } } return channelCandidate; @@ -1982,13 +2124,16 @@ private ChannelRef pickLeastBusyNoFallback() { private ChannelRef pickLeastBusyWithFallback(boolean forFallback) { // Full scan to collect eligible ("ready") channels not in fallbackMap and under max streams. List readyCandidates = new ArrayList<>(); - ChannelRef overallCandidate = channelRefs.get(0); - int overallMinStreams = overallCandidate.getActiveStreamsCount(); + ChannelRef overallCandidate = null; + int overallMinStreams = Integer.MAX_VALUE; int readyMaxStreams = 0; for (ChannelRef channelRef : channelRefs) { + if (!channelRef.isActive()) { + continue; + } int cnt = channelRef.getActiveStreamsCount(); - if (cnt < overallMinStreams) { + if (overallCandidate == null || cnt < overallMinStreams) { overallMinStreams = cnt; overallCandidate = channelRef; } @@ -2000,6 +2145,10 @@ private ChannelRef pickLeastBusyWithFallback(boolean forFallback) { } } + if (overallCandidate == null) { + return pickFromCandidates(channelRefs); + } + // For scale-up, use maxStreams among ready channels (consistent with non-fallback path). int scaleUpStreams = readyCandidates.isEmpty() ? Integer.MAX_VALUE : readyMaxStreams; if (shouldScaleUp(scaleUpStreams)) { @@ -2044,8 +2193,7 @@ private ChannelRef pickLeastBusyWithFallback(boolean forFallback) { * Picks a channel from the given candidate list using the configured strategy. * *

For {@code POWER_OF_TWO}: samples twice with replacement and picks the less busy candidate. - * The first sample wins ties. Draining or inactive candidates are retried before falling back to - * a full scan. + * The first sample wins ties. Inactive candidates are retried before falling back to a full scan. * *

For {@code LINEAR_SCAN}: deterministic scan picking the first least-busy active channel. */ @@ -2146,6 +2294,8 @@ private String keyFromOptsCtx(CallOptions callOptions) { } private synchronized void cancelBackgroundTasks() { + shuttingDown = true; + scaleUpSignalPending.set(false); if (cleanupTask != null) { cleanupTask.cancel(false); cleanupTask = null; @@ -2163,17 +2313,19 @@ private synchronized void cancelBackgroundTasks() { @Override public ManagedChannel shutdownNow() { logger.finer(log("Shutdown now started.")); - for (ChannelRef channelRef : channelRefs) { + cancelBackgroundTasks(); + List activeSnapshot = new ArrayList<>(channelRefs); + List removedSnapshot = new ArrayList<>(removedChannelRefs); + for (ChannelRef channelRef : activeSnapshot) { if (!channelRef.getChannel().isTerminated()) { channelRef.getChannel().shutdownNow(); } } - for (ChannelRef channelRef : removedChannelRefs) { + for (ChannelRef channelRef : removedSnapshot) { if (!channelRef.getChannel().isTerminated()) { channelRef.getChannel().shutdownNow(); } } - cancelBackgroundTasks(); if (!stateNotificationExecutor.isTerminated()) { stateNotificationExecutor.shutdownNow(); } @@ -2183,13 +2335,15 @@ public ManagedChannel shutdownNow() { @Override public ManagedChannel shutdown() { logger.finer(log("Shutdown started.")); - for (ChannelRef channelRef : channelRefs) { + cancelBackgroundTasks(); + List activeSnapshot = new ArrayList<>(channelRefs); + List removedSnapshot = new ArrayList<>(removedChannelRefs); + for (ChannelRef channelRef : activeSnapshot) { channelRef.getChannel().shutdown(); } - for (ChannelRef channelRef : removedChannelRefs) { + for (ChannelRef channelRef : removedSnapshot) { channelRef.getChannel().shutdown(); } - cancelBackgroundTasks(); stateNotificationExecutor.shutdown(); return this; } @@ -2540,6 +2694,7 @@ protected void activeStreamsCountIncr() { maxActiveStreams.accumulateAndGet(actStreams, Math::max); int totalActStreams = totalActiveStreams.incrementAndGet(); maxTotalActiveStreams.accumulateAndGet(totalActStreams, Math::max); + maybeSignalScaleUp(this); } protected void activeStreamsCountDecr(long startNanos, Status status, boolean fromClientSide) { @@ -2572,6 +2727,12 @@ protected int getActiveStreamsCount() { return activeStreamsCount.get(); } + @VisibleForTesting + void setActiveStreamsForTest(int streams) { + int previous = activeStreamsCount.getAndSet(streams); + totalActiveStreams.addAndGet(streams - previous); + } + protected long getAndResetOkCalls() { return okCalls.getAndSet(0); } diff --git a/grpc-gcp-java/src/main/java/com/google/cloud/grpc/GcpManagedChannelOptions.java b/grpc-gcp-java/src/main/java/com/google/cloud/grpc/GcpManagedChannelOptions.java index 795a264ec185..615ed094532f 100644 --- a/grpc-gcp-java/src/main/java/com/google/cloud/grpc/GcpManagedChannelOptions.java +++ b/grpc-gcp-java/src/main/java/com/google/cloud/grpc/GcpManagedChannelOptions.java @@ -50,7 +50,7 @@ public enum ChannelPickStrategy { /** * Samples two channels at random with replacement and returns the one with fewer active - * streams. The first sample wins ties. Inactive or draining samples are retried. + * streams. The first sample wins ties. Inactive samples are retried. * *

This is the default strategy. It avoids the thundering herd problem without preferring * channel warmth. The trade-off is that it may not always find the global minimum, but in @@ -211,7 +211,13 @@ public static class GcpChannelPoolOptions { private final int maxRpcPerChannel; // How often to check for a possibility to scale down. private final Duration scaleDownInterval; + // Minimum interval between successful scale-up actions. + private final Duration scaleUpCooldown; + // Consecutive low-load checks required before scaling down. private final int scaleDownConsecutiveLowLoadChecks; + // Maximum percentage growth in one scale-up action. + private final int maxScaleUpPercent; + // Maximum channels removed in one scale-down check. private final int maxScaleDownChannels; // Use round-robin channel selection for affinity binding calls. @@ -230,7 +236,9 @@ public GcpChannelPoolOptions(Builder builder) { minRpcPerChannel = builder.minRpcPerChannel; maxRpcPerChannel = builder.maxRpcPerChannel; scaleDownInterval = builder.scaleDownInterval; + scaleUpCooldown = builder.scaleUpCooldown; scaleDownConsecutiveLowLoadChecks = builder.scaleDownConsecutiveLowLoadChecks; + maxScaleUpPercent = builder.maxScaleUpPercent; maxScaleDownChannels = builder.maxScaleDownChannels; concurrentStreamsLowWatermark = builder.concurrentStreamsLowWatermark; useRoundRobinOnBind = builder.useRoundRobinOnBind; @@ -263,10 +271,18 @@ public Duration getScaleDownInterval() { return scaleDownInterval; } + public Duration getScaleUpCooldown() { + return scaleUpCooldown; + } + public int getScaleDownConsecutiveLowLoadChecks() { return scaleDownConsecutiveLowLoadChecks; } + public int getMaxScaleUpPercent() { + return maxScaleUpPercent; + } + public int getMaxScaleDownChannels() { return maxScaleDownChannels; } @@ -304,8 +320,27 @@ public static GcpChannelPoolOptions.Builder newBuilder(GcpChannelPoolOptions opt @Override public String toString() { return String.format( - "{maxSize: %d, minSize: %d, concurrentStreamsLowWatermark: %d, useRoundRobinOnBind: %s}", - getMaxSize(), getMinSize(), getConcurrentStreamsLowWatermark(), isUseRoundRobinOnBind()); + "{maxSize: %d, minSize: %d, initSize: %d, minRpcPerChannel: %d, " + + "maxRpcPerChannel: %d, scaleDownInterval: %s, scaleUpCooldown: %s, " + + "scaleDownConsecutiveLowLoadChecks: %d, maxScaleUpPercent: %d, " + + "maxScaleDownChannels: %d, " + + "concurrentStreamsLowWatermark: %d, useRoundRobinOnBind: %s, " + + "affinityKeyLifetime: %s, cleanupInterval: %s, channelPickStrategy: %s}", + getMaxSize(), + getMinSize(), + getInitSize(), + getMinRpcPerChannel(), + getMaxRpcPerChannel(), + getScaleDownInterval(), + getScaleUpCooldown(), + getScaleDownConsecutiveLowLoadChecks(), + getMaxScaleUpPercent(), + getMaxScaleDownChannels(), + getConcurrentStreamsLowWatermark(), + isUseRoundRobinOnBind(), + getAffinityKeyLifetime(), + getCleanupInterval(), + getChannelPickStrategy()); } public static class Builder { @@ -315,7 +350,9 @@ public static class Builder { private int minRpcPerChannel = 0; private int maxRpcPerChannel = 0; private Duration scaleDownInterval = Duration.ZERO; + private Duration scaleUpCooldown = Duration.ofSeconds(10); private int scaleDownConsecutiveLowLoadChecks = 3; + private int maxScaleUpPercent = 30; private int maxScaleDownChannels = 2; private int concurrentStreamsLowWatermark = GcpManagedChannel.DEFAULT_MAX_STREAM; private boolean useRoundRobinOnBind = false; @@ -336,7 +373,9 @@ public Builder(GcpChannelPoolOptions options) { this.minRpcPerChannel = options.getMinRpcPerChannel(); this.maxRpcPerChannel = options.getMaxRpcPerChannel(); this.scaleDownInterval = options.getScaleDownInterval(); + this.scaleUpCooldown = options.getScaleUpCooldown(); this.scaleDownConsecutiveLowLoadChecks = options.getScaleDownConsecutiveLowLoadChecks(); + this.maxScaleUpPercent = options.getMaxScaleUpPercent(); this.maxScaleDownChannels = options.getMaxScaleDownChannels(); this.concurrentStreamsLowWatermark = options.getConcurrentStreamsLowWatermark(); this.useRoundRobinOnBind = options.isUseRoundRobinOnBind(); @@ -390,23 +429,14 @@ public Builder setInitSize(int initSize) { /** * Enables dynamic scaling functionality. * - *

When the average number of concurrent calls per channel reaches maxRpcPerChannel - * the pool will create and add a new channel unless already at max size. - * - *

Every scaleDownInterval a check for downscaling is performed. Based on the - * maximum total concurrent calls observed since the last check, the desired number of - * channels is calculated as: + *

After a call is counted, load above maxRpcPerChannel on its selected + * channel or across the pool average signals a background scale-up worker. * - *

(max_total_concurrent_calls / minRpcPerChannel) rounded up. - * - *

If the calculated desired number of channels is lower than the current number of - * channels, the pool will be downscaled to the desired number or min size (whichever is - * greater). - * - *

When downscaling, channels with the oldest connections are selected. Then the selected - * channels are removed from the pool but are not instructed to shutdown until all calls are - * completed. In a case when the pool is scaling up and there is a ready channel awaiting - * calls completion, the channel will be re-used instead of creating a new channel. + *

Every scaleDownInterval, after consecutive low-load checks, the + * longest-connected channels (by connectedSinceNanos) are removed from selection, bounded by + * maxScaleDownChannels. A removed channel is shut down on a later check once its in-flight + * calls reach zero. A READY removed channel can be reused by a later scale-up before it + * closes. * * @param minRpcPerChannel minimum desired average concurrent calls per channel. * @param maxRpcPerChannel maximum desired average concurrent calls per channel. @@ -418,6 +448,9 @@ public Builder setDynamicScaling( minRpcPerChannel > 0, "Minimum RPCs per channel must be positive."); Preconditions.checkArgument( maxRpcPerChannel > 0, "Maximum RPCs per channel must be positive."); + Preconditions.checkArgument( + minRpcPerChannel <= maxRpcPerChannel, + "Minimum RPCs per channel must not exceed maximum RPCs per channel."); Preconditions.checkArgument( !scaleDownInterval.isNegative() && !scaleDownInterval.isZero(), "Scale down interval must be positive."); @@ -439,12 +472,35 @@ public Builder disableDynamicScaling() { return this; } + /** + * Sets the minimum interval between successful scale-up operations. Zero uses the 10-second + * default. + */ + public Builder setScaleUpCooldown(Duration scaleUpCooldown) { + Preconditions.checkNotNull(scaleUpCooldown, "Scale up cooldown must not be null."); + Preconditions.checkArgument( + !scaleUpCooldown.isNegative(), "Scale up cooldown must not be negative."); + this.scaleUpCooldown = scaleUpCooldown.isZero() ? Duration.ofSeconds(10) : scaleUpCooldown; + return this; + } + public Builder setScaleDownConsecutiveLowLoadChecks(int checks) { Preconditions.checkArgument(checks > 0, "Scale down checks must be positive."); this.scaleDownConsecutiveLowLoadChecks = checks; return this; } + /** + * Sets the maximum percentage of active channels added by one scale-up operation. The + * percentage cap has a two-channel floor before desired-size and maximum-size clamps. + */ + public Builder setMaxScaleUpPercent(int percent) { + Preconditions.checkArgument( + percent > 0 && percent <= 100, "Scale up percent must be in (0, 100]."); + this.maxScaleUpPercent = percent; + return this; + } + public Builder setMaxScaleDownChannels(int channels) { Preconditions.checkArgument(channels > 0, "Scale down channel limit must be positive."); this.maxScaleDownChannels = channels; @@ -510,8 +566,7 @@ public Builder setCleanupInterval(Duration cleanupInterval) { * *

Defaults to {@link ChannelPickStrategy#POWER_OF_TWO} which avoids the thundering herd * problem by sampling two channels with replacement and picking the less busy one. The first - * sample wins ties, with no channel-warmth preference. Inactive or draining samples are - * retried. + * sample wins ties, with no channel-warmth preference. Inactive samples are retried. * *

Use {@link ChannelPickStrategy#LINEAR_SCAN} to restore the legacy behavior of scanning * all channels and always picking the one with the fewest active streams. diff --git a/grpc-gcp-java/src/test/java/com/google/cloud/grpc/GcpManagedChannelHotChannelReproducerTest.java b/grpc-gcp-java/src/test/java/com/google/cloud/grpc/GcpManagedChannelHotChannelReproducerTest.java index b5538d963348..f25794edaea8 100644 --- a/grpc-gcp-java/src/test/java/com/google/cloud/grpc/GcpManagedChannelHotChannelReproducerTest.java +++ b/grpc-gcp-java/src/test/java/com/google/cloud/grpc/GcpManagedChannelHotChannelReproducerTest.java @@ -356,6 +356,7 @@ private RunResult runScenario(long seed, Variant variant, LoadShape loadShape) t poolOptions.disableDynamicScaling(); } else { poolOptions.setDynamicScaling(MIN_RPC_PER_CHANNEL, MAX_RPC_PER_CHANNEL, scaleDownInterval); + compressBranchSpecificScaleUpTimer(poolOptions); } pool = @@ -426,6 +427,10 @@ private RunResult runScenario(long seed, Variant variant, LoadShape loadShape) t } } + private static void compressBranchSpecificScaleUpTimer(GcpChannelPoolOptions.Builder builder) { + builder.setScaleUpCooldown(Duration.ofNanos(1)); + } + private static void runTransactionsAcrossScaleDown( GcpManagedChannel pool, ExecutorService executor, diff --git a/grpc-gcp-java/src/test/java/com/google/cloud/grpc/GcpManagedChannelOptionsTest.java b/grpc-gcp-java/src/test/java/com/google/cloud/grpc/GcpManagedChannelOptionsTest.java index 7dc407ff1381..64e6479c5824 100644 --- a/grpc-gcp-java/src/test/java/com/google/cloud/grpc/GcpManagedChannelOptionsTest.java +++ b/grpc-gcp-java/src/test/java/com/google/cloud/grpc/GcpManagedChannelOptionsTest.java @@ -20,6 +20,7 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertThrows; import static org.junit.Assert.assertTrue; import com.google.cloud.grpc.GcpManagedChannelOptions.GcpChannelPoolOptions; @@ -208,6 +209,58 @@ public void testAffinityKeysCleanupZeroByDefault() { assertThat(channelPoolOptions.getCleanupInterval()).isEqualTo(Duration.ZERO); } + @Test + public void testDynamicScalingKnobsHaveDefaultsAndSurviveCopy() { + GcpChannelPoolOptions defaults = GcpChannelPoolOptions.newBuilder().build(); + + assertThat(defaults.getScaleUpCooldown()).isEqualTo(Duration.ofSeconds(10)); + assertThat(defaults.getMaxScaleUpPercent()).isEqualTo(30); + + GcpChannelPoolOptions configured = + GcpChannelPoolOptions.newBuilder(defaults) + .setScaleUpCooldown(Duration.ofSeconds(1)) + .setMaxScaleUpPercent(40) + .build(); + GcpChannelPoolOptions copied = GcpChannelPoolOptions.newBuilder(configured).build(); + + assertThat(copied.getScaleUpCooldown()).isEqualTo(Duration.ofSeconds(1)); + assertThat(copied.getMaxScaleUpPercent()).isEqualTo(40); + } + + @Test + public void dynamicScalingDefaultKnobsRejectNegativeAndDefaultZeroCooldown() { + assertThrows( + IllegalArgumentException.class, + () -> GcpChannelPoolOptions.newBuilder().setScaleUpCooldown(Duration.ofNanos(-1))); + assertThat( + GcpChannelPoolOptions.newBuilder() + .setScaleUpCooldown(Duration.ZERO) + .build() + .getScaleUpCooldown()) + .isEqualTo(Duration.ofSeconds(10)); + } + + @Test + public void channelPoolOptionsToStringIncludesEveryKnob() { + String options = GcpChannelPoolOptions.newBuilder().build().toString(); + + assertThat(options).contains("maxSize:"); + assertThat(options).contains("minSize:"); + assertThat(options).contains("initSize:"); + assertThat(options).contains("minRpcPerChannel:"); + assertThat(options).contains("maxRpcPerChannel:"); + assertThat(options).contains("scaleDownInterval:"); + assertThat(options).contains("scaleUpCooldown:"); + assertThat(options).contains("scaleDownConsecutiveLowLoadChecks:"); + assertThat(options).contains("maxScaleUpPercent:"); + assertThat(options).contains("maxScaleDownChannels:"); + assertThat(options).contains("concurrentStreamsLowWatermark:"); + assertThat(options).contains("useRoundRobinOnBind:"); + assertThat(options).contains("affinityKeyLifetime:"); + assertThat(options).contains("cleanupInterval:"); + assertThat(options).contains("channelPickStrategy:"); + } + @Test public void testCleanupDefault() { GcpManagedChannelOptions opts = diff --git a/grpc-gcp-java/src/test/java/com/google/cloud/grpc/GcpManagedChannelScaleUpWorkerTest.java b/grpc-gcp-java/src/test/java/com/google/cloud/grpc/GcpManagedChannelScaleUpWorkerTest.java new file mode 100644 index 000000000000..394d184e3652 --- /dev/null +++ b/grpc-gcp-java/src/test/java/com/google/cloud/grpc/GcpManagedChannelScaleUpWorkerTest.java @@ -0,0 +1,402 @@ +/* + * Copyright 2026 Google LLC + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.google.cloud.grpc; + +import static com.google.common.truth.Truth.assertThat; +import static org.awaitility.Awaitility.await; + +import com.google.cloud.grpc.GcpManagedChannel.ChannelRef; +import com.google.cloud.grpc.GcpManagedChannelOptions.GcpChannelPoolOptions; +import com.google.common.util.concurrent.MoreExecutors; +import java.time.Duration; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicReference; +import org.junit.After; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +@RunWith(JUnit4.class) +public final class GcpManagedChannelScaleUpWorkerTest { + private final ExecutorService executor = MoreExecutors.newDirectExecutorService(); + private GcpManagedChannel pool; + + @After + public void tearDown() { + if (pool != null) { + pool.shutdownNow(); + } + executor.shutdownNow(); + } + + @Test + public void scaleUpAddsAtMostThirtyPercent() { + pool = newPool(10, 30); + ChannelRef hot = pool.channelRefs.get(0); + hot.setActiveStreamsForTest(99); + + hot.activeStreamsCountIncr(); + + await().atMost(Duration.ofSeconds(5)).until(() -> pool.getNumberOfChannels() == 13); + } + + @Test + public void scaleUpUsesTwoChannelFloorForSmallPool() { + pool = newPool(1, 5); + ChannelRef hot = pool.channelRefs.get(0); + hot.setActiveStreamsForTest(9); + + hot.activeStreamsCountIncr(); + + await().atMost(Duration.ofSeconds(5)).until(() -> pool.getNumberOfChannels() == 3); + } + + @Test + public void hottestChannelCanSignalBeforePoolAverageCrossesLimit() { + pool = newPool(4, 8); + pool.channelRefs.get(0).setActiveStreamsForTest(5); + pool.channelRefs.get(1).setActiveStreamsForTest(3); + pool.channelRefs.get(2).setActiveStreamsForTest(2); + pool.channelRefs.get(3).setActiveStreamsForTest(2); + + pool.channelRefs.get(0).activeStreamsCountIncr(); + + await().atMost(Duration.ofSeconds(5)).until(() -> pool.getNumberOfChannels() == 5); + } + + @Test + public void poolAverageCanSignalWhenSelectedChannelIsBelowMaximum() { + pool = newPool(2, 4); + ChannelRef selected = pool.channelRefs.get(0); + pool.channelRefs.get(1).setActiveStreamsForTest(12); + + selected.activeStreamsCountIncr(); + + await().atMost(Duration.ofSeconds(5)).until(() -> pool.getNumberOfChannels() == 4); + } + + @Test + public void scaleUpBuildsOffCallerThread() { + AtomicInteger builds = new AtomicInteger(); + AtomicReference scaleUpThread = new AtomicReference<>(); + GcpManagedChannelTest.FakeManagedChannelBuilder delegate = + new GcpManagedChannelTest.FakeManagedChannelBuilder( + () -> { + if (builds.incrementAndGet() > 2) { + scaleUpThread.set(Thread.currentThread().getName()); + } + return new GcpManagedChannelTest.FakeManagedChannel(executor); + }); + pool = newPool(2, 4, delegate); + String callerThread = Thread.currentThread().getName(); + + for (int i = 0; i < 7; i++) { + pool.channelRefs.get(0).activeStreamsCountIncr(); + } + + await().atMost(Duration.ofSeconds(5)).until(() -> scaleUpThread.get() != null); + assertThat(scaleUpThread.get()).isNotEqualTo(callerThread); + } + + @Test + public void burstSignalsCoalesceWhileWorkerIsBusy() throws Exception { + AtomicInteger builds = new AtomicInteger(); + CountDownLatch buildStarted = new CountDownLatch(1); + CountDownLatch releaseBuild = new CountDownLatch(1); + GcpManagedChannelTest.FakeManagedChannelBuilder delegate = + new GcpManagedChannelTest.FakeManagedChannelBuilder( + () -> { + if (builds.incrementAndGet() == 3) { + buildStarted.countDown(); + try { + releaseBuild.await(5, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + return new GcpManagedChannelTest.FakeManagedChannel(executor); + }); + pool = newPool(2, 10, delegate); + AtomicLong clock = new AtomicLong(1); + pool.setNanoClock(clock::get); + ChannelRef hot = pool.channelRefs.get(0); + for (int i = 0; i < 7; i++) { + hot.activeStreamsCountIncr(); + } + assertThat(buildStarted.await(5, TimeUnit.SECONDS)).isTrue(); + + for (int i = 7; i < 30; i++) { + hot.activeStreamsCountIncr(); + } + clock.incrementAndGet(); + releaseBuild.countDown(); + + await().atMost(Duration.ofSeconds(5)).until(() -> pool.getNumberOfChannels() == 5); + await() + .during(Duration.ofMillis(100)) + .atMost(Duration.ofSeconds(1)) + .until(() -> pool.getNumberOfChannels() == 5); + } + + @Test + public void buildExceptionClosesPartialBatchAndFutureSignalStillScales() { + AtomicInteger builds = new AtomicInteger(); + AtomicLong clock = new AtomicLong(1); + AtomicReference partialBuild = + new AtomicReference<>(); + GcpManagedChannelTest.FakeManagedChannelBuilder delegate = + new GcpManagedChannelTest.FakeManagedChannelBuilder( + () -> { + int build = builds.incrementAndGet(); + if (build == 3) { + GcpManagedChannelTest.FakeManagedChannel channel = + new GcpManagedChannelTest.FakeManagedChannel(executor); + partialBuild.set(channel); + return channel; + } + if (build == 4) { + throw new AssertionError("build failed"); + } + return new GcpManagedChannelTest.FakeManagedChannel(executor); + }); + pool = newPool(2, 6, delegate); + pool.setNanoClock(clock::get); + ChannelRef hot = pool.channelRefs.get(0); + hot.setActiveStreamsForTest(9); + + hot.activeStreamsCountIncr(); + + await().atMost(Duration.ofSeconds(5)).until(() -> builds.get() == 4); + await().atMost(Duration.ofSeconds(5)).until(() -> partialBuild.get().isShutdown()); + clock.incrementAndGet(); + hot.activeStreamsCountIncr(); + await().atMost(Duration.ofSeconds(5)).until(() -> pool.getNumberOfChannels() == 4); + } + + @Test + public void scaleUpCooldownIsHonored() { + AtomicLong clock = new AtomicLong(100); + pool = newPool(2, 6, Duration.ofNanos(10)); + pool.setNanoClock(clock::get); + ChannelRef hot = pool.channelRefs.get(0); + hot.setActiveStreamsForTest(20); + + hot.activeStreamsCountIncr(); + await().atMost(Duration.ofSeconds(5)).until(() -> pool.getNumberOfChannels() == 4); + + hot.activeStreamsCountIncr(); + await() + .during(Duration.ofMillis(100)) + .atMost(Duration.ofSeconds(1)) + .until(() -> pool.getNumberOfChannels() == 4); + + clock.addAndGet(11); + hot.activeStreamsCountIncr(); + await().atMost(Duration.ofSeconds(5)).until(() -> pool.getNumberOfChannels() == 6); + } + + @Test + public void scaleUpClampsToMaxSize() { + pool = newPool(4, 5); + ChannelRef hot = pool.channelRefs.get(0); + hot.setActiveStreamsForTest(30); + + hot.activeStreamsCountIncr(); + + await().atMost(Duration.ofSeconds(5)).until(() -> pool.getNumberOfChannels() == 5); + } + + @Test + public void shutdownDuringScaleUpClosesUnpublishedChannel() throws Exception { + AtomicInteger builds = new AtomicInteger(); + CountDownLatch buildStarted = new CountDownLatch(1); + CountDownLatch releaseBuild = new CountDownLatch(1); + AtomicReference unpublished = new AtomicReference<>(); + GcpManagedChannelTest.FakeManagedChannelBuilder delegate = + new GcpManagedChannelTest.FakeManagedChannelBuilder( + () -> { + if (builds.incrementAndGet() == 3) { + GcpManagedChannelTest.FakeManagedChannel channel = + new GcpManagedChannelTest.FakeManagedChannel(executor); + unpublished.set(channel); + buildStarted.countDown(); + try { + releaseBuild.await(5, TimeUnit.SECONDS); + } catch (InterruptedException failure) { + Thread.currentThread().interrupt(); + } + return channel; + } + return new GcpManagedChannelTest.FakeManagedChannel(executor); + }); + pool = newPool(2, 3, delegate); + ChannelRef hot = pool.channelRefs.get(0); + hot.setActiveStreamsForTest(9); + hot.activeStreamsCountIncr(); + assertThat(buildStarted.await(5, TimeUnit.SECONDS)).isTrue(); + + pool.shutdownNow(); + releaseBuild.countDown(); + + await().atMost(Duration.ofSeconds(5)).until(() -> unpublished.get().isShutdown()); + assertThat(pool.getNumberOfChannels()).isEqualTo(2); + } + + @Test + public void shutdownDuringScaleUpClosesChannelPublishingUnderPoolMonitor() throws Exception { + AtomicInteger builds = new AtomicInteger(); + CountDownLatch publishing = new CountDownLatch(1); + CountDownLatch releasePublish = new CountDownLatch(1); + AtomicReference publishingChannel = + new AtomicReference<>(); + GcpManagedChannelTest.FakeManagedChannelBuilder delegate = + new GcpManagedChannelTest.FakeManagedChannelBuilder( + () -> { + if (builds.incrementAndGet() == 3) { + GcpManagedChannelTest.FakeManagedChannel channel = + new GcpManagedChannelTest.FakeManagedChannel(executor) { + @Override + public io.grpc.ConnectivityState getState(boolean requestConnection) { + publishing.countDown(); + try { + releasePublish.await(5, TimeUnit.SECONDS); + } catch (InterruptedException failure) { + Thread.currentThread().interrupt(); + } + return super.getState(requestConnection); + } + }; + publishingChannel.set(channel); + return channel; + } + return new GcpManagedChannelTest.FakeManagedChannel(executor); + }); + pool = newPool(2, 3, delegate); + ChannelRef hot = pool.channelRefs.get(0); + hot.setActiveStreamsForTest(9); + hot.activeStreamsCountIncr(); + assertThat(publishing.await(5, TimeUnit.SECONDS)).isTrue(); + + AtomicReference shutdownThread = new AtomicReference<>(); + ExecutorService shutdownExecutor = + Executors.newSingleThreadExecutor( + command -> { + Thread thread = new Thread(command); + shutdownThread.set(thread); + return thread; + }); + try { + Future shutdown = shutdownExecutor.submit(pool::shutdown); + await() + .atMost(Duration.ofSeconds(5)) + .until( + () -> + shutdownThread.get() != null + && shutdownThread.get().getState() == Thread.State.BLOCKED); + releasePublish.countDown(); + shutdown.get(5, TimeUnit.SECONDS); + await().atMost(Duration.ofSeconds(5)).until(() -> publishingChannel.get().isShutdown()); + assertThat(pool.channelRefs).hasSize(3); + assertThat( + pool.channelRefs.stream() + .allMatch(channelRef -> channelRef.getChannel().isShutdown())) + .isTrue(); + } finally { + releasePublish.countDown(); + shutdownExecutor.shutdownNow(); + } + } + + @Test + public void removedChannelReadersTolerateConcurrentMutation() throws Exception { + pool = newPool(2, 3); + ChannelRef removed = pool.channelRefs.get(0); + ExecutorService mutator = Executors.newSingleThreadExecutor(); + AtomicReference failure = new AtomicReference<>(); + CountDownLatch started = new CountDownLatch(1); + Future mutation = + mutator.submit( + () -> { + started.countDown(); + try { + for (int i = 0; i < 200; i++) { + pool.removedChannelRefs.add(removed); + pool.removedChannelRefs.remove(removed); + } + } catch (Throwable throwable) { + failure.set(throwable); + } + }); + try { + assertThat(started.await(1, TimeUnit.SECONDS)).isTrue(); + for (int i = 0; i < 200; i++) { + pool.isShutdown(); + pool.isTerminated(); + pool.awaitTermination(0, TimeUnit.NANOSECONDS); + } + await().atMost(Duration.ofSeconds(1)).until(mutation::isDone); + assertThat(failure.get()).isNull(); + } finally { + mutator.shutdownNow(); + } + } + + private GcpManagedChannel newPool(int initial, int maximum) { + return newPool(initial, maximum, Duration.ofNanos(1)); + } + + private GcpManagedChannel newPool(int initial, int maximum, Duration scaleUpCooldown) { + return newPool( + initial, + maximum, + new GcpManagedChannelTest.FakeManagedChannelBuilder( + () -> new GcpManagedChannelTest.FakeManagedChannel(executor)), + scaleUpCooldown); + } + + private GcpManagedChannel newPool( + int initial, int maximum, GcpManagedChannelTest.FakeManagedChannelBuilder delegate) { + return newPool(initial, maximum, delegate, Duration.ofNanos(1)); + } + + private GcpManagedChannel newPool( + int initial, + int maximum, + GcpManagedChannelTest.FakeManagedChannelBuilder delegate, + Duration scaleUpCooldown) { + GcpChannelPoolOptions options = + GcpChannelPoolOptions.newBuilder() + .setInitSize(initial) + .setMinSize(initial) + .setMaxSize(maximum) + .setDynamicScaling(2, 5, Duration.ofMinutes(1)) + .setScaleUpCooldown(scaleUpCooldown) + .setMaxScaleUpPercent(30) + .build(); + return (GcpManagedChannel) + GcpManagedChannelBuilder.forDelegateBuilder(delegate) + .withOptions( + GcpManagedChannelOptions.newBuilder().withChannelPoolOptions(options).build()) + .build(); + } +} diff --git a/grpc-gcp-java/src/test/java/com/google/cloud/grpc/GcpManagedChannelSkewFixTest.java b/grpc-gcp-java/src/test/java/com/google/cloud/grpc/GcpManagedChannelSkewFixTest.java index 9fe5b60d2613..8991101b7005 100644 --- a/grpc-gcp-java/src/test/java/com/google/cloud/grpc/GcpManagedChannelSkewFixTest.java +++ b/grpc-gcp-java/src/test/java/com/google/cloud/grpc/GcpManagedChannelSkewFixTest.java @@ -22,6 +22,8 @@ import com.google.cloud.grpc.GcpManagedChannel.ChannelRef; import com.google.cloud.grpc.GcpManagedChannelOptions.ChannelPickStrategy; import com.google.cloud.grpc.GcpManagedChannelOptions.GcpChannelPoolOptions; +import com.google.cloud.grpc.GcpManagedChannelOptions.GcpResiliencyOptions; +import com.google.cloud.grpc.proto.ApiConfig; import com.google.common.util.concurrent.MoreExecutors; import java.time.Duration; import java.util.concurrent.ExecutorService; @@ -109,6 +111,87 @@ public void pickLeastBusyNoFallback_linearScan_skipsInactiveChannels() { assertThat(pool.getChannelRef(null)).isSameInstanceAs(active); } + @Test + public void roundRobin_skipsInactiveChannels() { + pool = newPool(2, 2); + ChannelRef active = pool.channelRefs.get(0); + pool.channelRefs.get(1).deactivateForTest(); + + assertThat(pool.getChannelRefRoundRobin()).isSameInstanceAs(active); + } + + @Test + public void roundRobin_dynamicEmptyPool_createsFirstChannel() { + GcpChannelPoolOptions options = + GcpChannelPoolOptions.newBuilder() + .setInitSize(0) + .setMinSize(0) + .setMaxSize(2) + .setDynamicScaling(10, 20, Duration.ofMinutes(1)) + .build(); + pool = + (GcpManagedChannel) + GcpManagedChannelBuilder.forDelegateBuilder( + new GcpManagedChannelTest.FakeManagedChannelBuilder( + () -> new GcpManagedChannelTest.FakeManagedChannel(executor))) + .withOptions( + GcpManagedChannelOptions.newBuilder().withChannelPoolOptions(options).build()) + .build(); + + assertThat(pool.channelRefs).isEmpty(); + assertThat(pool.getChannelRefRoundRobin()).isNotNull(); + assertThat(pool.channelRefs).hasSize(1); + } + + @Test + public void fallbackPicker_skipsInactiveChannels() { + pool = newPoolWithFallback(); + ChannelRef inactive = pool.channelRefs.get(0); + ChannelRef active = pool.channelRefs.get(1); + inactive.deactivateForTest(); + pool.fallbackMapForTest().put(active.getId(), new java.util.concurrent.ConcurrentHashMap<>()); + + assertThat(pool.getChannelRef(null)).isSameInstanceAs(active); + } + + @Test + public void activeStreamExtrema_skipInactiveChannels() { + pool = newPool(2, 2); + ChannelRef inactive = pool.channelRefs.get(0); + ChannelRef active = pool.channelRefs.get(1); + inactive.setActiveStreamsForTest(99); + active.setActiveStreamsForTest(3); + inactive.deactivateForTest(); + + assertThat(pool.getMinActiveStreams()).isEqualTo(3); + assertThat(pool.getMaxActiveStreams()).isEqualTo(3); + } + + @Test + public void dynamicPowerOfTwoPickDoesNotReadLegacyWatermarkLoad() { + GcpChannelPoolOptions options = + GcpChannelPoolOptions.newBuilder() + .setInitSize(2) + .setMinSize(2) + .setMaxSize(4) + .setDynamicScaling(10, 20, Duration.ofMinutes(1)) + .setChannelPickStrategy(ChannelPickStrategy.POWER_OF_TWO) + .build(); + pool = + new GcpManagedChannel( + new GcpManagedChannelTest.FakeManagedChannelBuilder( + () -> new GcpManagedChannelTest.FakeManagedChannel(executor)), + ApiConfig.getDefaultInstance(), + GcpManagedChannelOptions.newBuilder().withChannelPoolOptions(options).build()) { + @Override + public int getMaxActiveStreams() { + throw new AssertionError("legacy watermark load must be lazy"); + } + }; + + assertThat(pool.getChannelRef(null)).isNotNull(); + } + @Test public void affinityReferenceStaysStickyUntilDelegateShutdown() { pool = newPool(4, 2); @@ -151,6 +234,22 @@ private GcpManagedChannel newPool( .build(); } + private GcpManagedChannel newPoolWithFallback() { + GcpChannelPoolOptions options = + GcpChannelPoolOptions.newBuilder().setInitSize(2).setMinSize(2).setMaxSize(2).build(); + return (GcpManagedChannel) + GcpManagedChannelBuilder.forDelegateBuilder( + new GcpManagedChannelTest.FakeManagedChannelBuilder( + () -> new GcpManagedChannelTest.FakeManagedChannel(executor))) + .withOptions( + GcpManagedChannelOptions.newBuilder() + .withChannelPoolOptions(options) + .withResiliencyOptions( + GcpResiliencyOptions.newBuilder().setNotReadyFallback(true).build()) + .build()) + .build(); + } + private void reserveOneStreamPerChannel() { for (ChannelRef channelRef : pool.channelRefs) { channelRef.activeStreamsCountIncr(); diff --git a/grpc-gcp-java/src/test/java/com/google/cloud/grpc/GcpManagedChannelTest.java b/grpc-gcp-java/src/test/java/com/google/cloud/grpc/GcpManagedChannelTest.java index 6153f4259678..fa652b48b321 100644 --- a/grpc-gcp-java/src/test/java/com/google/cloud/grpc/GcpManagedChannelTest.java +++ b/grpc-gcp-java/src/test/java/com/google/cloud/grpc/GcpManagedChannelTest.java @@ -19,6 +19,7 @@ import static com.google.cloud.grpc.GcpManagedChannel.getKeysFromMessage; import static com.google.common.truth.Truth.assertThat; import static com.google.common.truth.Truth.assertWithMessage; +import static org.awaitility.Awaitility.await; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; @@ -426,6 +427,52 @@ public void readyAccountingRemainsExactWhenReadyChannelIsReused() { } } + @Test + public void readyRemovedChannelIsReusedWhenNewerRemovedChannelIsNotReady() { + resetGcpChannel(); + ExecutorService executorService = MoreExecutors.newDirectExecutorService(); + try { + gcpChannel = + new GcpManagedChannel( + new FakeManagedChannelBuilder(() -> new FakeManagedChannel(executorService)), + ApiConfig.getDefaultInstance(), + GcpManagedChannelOptions.newBuilder().build()); + ChannelRef olderReady = + gcpChannel.new ChannelRef(new FakeManagedChannel(executorService)) { + @Override + protected long getConnectedSinceNanos() { + return 1; + } + + @Override + protected ConnectivityState getState() { + return ConnectivityState.READY; + } + }; + ChannelRef newerNotReady = + gcpChannel.new ChannelRef(new FakeManagedChannel(executorService)) { + @Override + protected long getConnectedSinceNanos() { + return 2; + } + + @Override + protected ConnectivityState getState() { + return ConnectivityState.IDLE; + } + }; + olderReady.deactivateForTest(); + newerNotReady.deactivateForTest(); + gcpChannel.removedChannelRefs.add(olderReady); + gcpChannel.removedChannelRefs.add(newerNotReady); + + assertThat(gcpChannel.createNewChannel()).isSameInstanceAs(olderReady); + } finally { + gcpChannel.shutdownNow(); + executorService.shutdownNow(); + } + } + @Test public void testChannelAffinityRefRemovedChannelPicksAvailableChannel() throws Exception { resetGcpChannel(); @@ -649,11 +696,12 @@ public void testPickLeastBusyWithDynamicScaleUp() throws InterruptedException { // One more call triggers scale-up. pool.getChannelRef(null).activeStreamsCountIncr(); - assertThat(pool.getNumberOfChannels()).isEqualTo(minSize + 1); + await().atMost(Duration.ofSeconds(5)).until(() -> pool.getNumberOfChannels() == minSize + 2); - // Mark the new channel as READY. - ((FakeManagedChannel) pool.channelRefs.get(minSize).getChannel()) - .setState(ConnectivityState.READY); + // Mark the new channels as READY. + for (int i = minSize; i < pool.getNumberOfChannels(); i++) { + ((FakeManagedChannel) pool.channelRefs.get(i).getChannel()).setState(ConnectivityState.READY); + } // Now pick many times without incrementing. The new (less busy) channel should be favored, // but picks should still be distributed across channels. @@ -834,6 +882,9 @@ public void testGetChannelRefWithFallback() { GcpMetricsOptions.newBuilder().withMetricRegistry(fakeRegistry).build()) .build()) .build(); + AtomicInteger fallbackSample = new AtomicInteger(); + pool.setCandidateIndexPickerForTest( + bound -> Math.floorMod(fallbackSample.getAndIncrement(), bound)); final int currentIndex = GcpManagedChannel.channelPoolIndex.get(); final String poolIndex = String.format("pool-%d", currentIndex);