feat(grpc-gcp): move scale-up to background worker

[rahul2393]

No description provided.

[gemini-code-assist]

Code Review

This pull request introduces asynchronous background scale-up logic for GcpManagedChannel to prevent blocking caller threads during channel creation, adding new configuration options like scaleUpCooldown and maxScaleUpPercent along with comprehensive tests. The review feedback highlights critical concurrency issues that need to be addressed: first, recursive retries in submitScaleUpWorker upon executor rejection could block gRPC transport threads or cause a StackOverflowError; second, race conditions in shutdownNow() and shutdown() could allow background channels to be published after shutdown has initiated, leading to silent resource leaks. These state updates should be properly synchronized.

[rahul2393]

/gemini review

[gemini-code-assist]

Code Review

This pull request introduces an asynchronous background scale-up worker for dynamic scaling in GcpManagedChannel, moving the scale-up logic off the caller thread. It adds new configuration options such as scaleUpCooldown and maxScaleUpPercent to control scaling behavior, ensures inactive channels are skipped during channel selection, and adds comprehensive unit tests. The review feedback identifies critical issues, including a potential ArithmeticException when the channel pool is empty, thread safety concerns in shutdown() and shutdownNow() due to unsynchronized snapshot creation, and exception-handling robustness during channel building. Additionally, it suggests using Google Truth assertions consistently across the new test file for better readability.

[gemini-code-assist]

If channelRefs is empty (for example, during initialization or shutdown), snapshot.length will be 0. This will cause an ArithmeticException: / by zero when calculating bindingIndex % snapshot.length. Guarding against an empty snapshot prevents potential runtime crashes.

    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 pickFromCandidates(channelRefs);

[gemini-code-assist]

If delegateChannelBuilder.build() throws an exception, any successfully built channels in builtChannels are shut down. However, if calling shutdownNow() on one of the channels throws an exception, subsequent channels won't be shut down, and the original failure will be lost. Using Throwable.addSuppressed ensures all channels are attempted to be shut down and the original exception is preserved.

    } catch (Throwable failure) {
      for (ManagedChannel channel : builtChannels) {
        try {
          channel.shutdownNow();
        } catch (Throwable t) {
          failure.addSuppressed(t);
        }
      }
      throw failure;
    }

[rahul2393]

/gemini review

[gemini-code-assist]

Code Review

This pull request introduces an asynchronous background scale-up worker to GcpManagedChannel to handle dynamic scaling off the caller thread, along with new configuration options (scaleUpCooldown, maxScaleUpPercent) and robust shutdown handling. It also updates channel selection strategies to skip inactive channels and adds comprehensive unit tests. The review feedback suggests optimizing the performance of monitoring and routing methods (getMinActiveStreams, getMaxActiveStreams, and pickLeastBusyWithFallback) to avoid potential lock contention on the critical path by utilizing atomic variables or thread-safe collections.

[olavloite]

This is correctly taking a snapshot of removedChannelRefs while holding a lock to ensure that there are no concurrent access to the list. However, these methods also read from the same list without holding a lock:

awaitTermination()
isShutdown()
isTerminated()
logChannelsStats()

(This test shows the problem, but don't add it to the code base, because I think it is too slow)

@Test
public void concurrentIsShutdownDoesNotThrowConcurrentModificationException() throws Exception {
  pool = newPool(2, 4);
  ChannelRef channelRef = pool.channelRefs.get(0);
  AtomicBoolean running = new AtomicBoolean(true);
  CountDownLatch latch = new CountDownLatch(1);
  AtomicReference<Throwable> error = new AtomicReference<>();

  Thread readerThread =
      new Thread(
          () -> {
            latch.countDown();
            while (running.get()) {
              try {
                pool.isShutdown();
                pool.isTerminated();
              } catch (Throwable t) {
                error.set(t);
                break;
              }
            }
          });
  readerThread.start();
  assertTrue(latch.await(5, TimeUnit.SECONDS));

  for (int i = 0; i < 10_000; i++) {
    synchronized (pool) {
      pool.removedChannelRefs.add(channelRef);
      pool.removedChannelRefs.remove(channelRef);
    }
    if (error.get() != null) {
      break;
    }
  }
  running.set(false);
  readerThread.join(5000);

  assertNull(
      "Expected no ConcurrentModificationException in isShutdown/isTerminated", error.get());
}

[olavloite]

Although this code is not changed in this PR, it is referenced in a couple of places. And I think that this could be implemented more efficiently by first filtering for state == Ready and then picking the one that was most recently connected. Now it first selects the most recently connected, and if that one does not have state Ready, then Optional.empty() is returned.

[olavloite]

I'm not sure this is completely accurate. I think that the oldest channels are selected for scale-down. And there is no grace period before they close once the number of active RPCs reaches zero.

[olavloite]

For a follow-up pull request: I think that this class needs to be split. It does more than 'just being a managed channel' and is almost 3,000 LoC, which makes it hard to read.

[olavloite]

I don't think this is used. (And if it is, then it is probably a code smell)

[olavloite]

This method allocates a number of different (small) objects on every invocation, so it should preferably only be called on the hot path when it is really necessary. It is currently always called in pickLeastBusyNoFallback() when p2c is used for channel selection, including when dynamic scaling is disabled. Meaning that it is potentially called on every RPC for no good reason. Could we fix that?