alts: Enable user to configure max number of concurrent ALTS handshakes. (#10016)

alts/src/main/java/io/grpc/alts/internal/AltsProtocolNegotiator.java

@@ -54,10 +54,12 @@ import javax.annotation.Nullable;
 // TODO(carl-mastrangelo): rename this AltsProtocolNegotiators.
 public final class AltsProtocolNegotiator {
   private static final Logger logger = Logger.getLogger(AltsProtocolNegotiator.class.getName());
-  // Avoid performing too many handshakes in parallel, as it may cause queuing in the handshake
+
-  // server and cause unbounded blocking on the event loop (b/168808426). This is a workaround until
+  static final String ALTS_MAX_CONCURRENT_HANDSHAKES_ENV_VARIABLE =
-  // there is an async TSI handshaking API to avoid the blocking.
+      "GRPC_ALTS_MAX_CONCURRENT_HANDSHAKES";
-  private static final AsyncSemaphore handshakeSemaphore = new AsyncSemaphore(32);
+  @VisibleForTesting static final int DEFAULT_ALTS_MAX_CONCURRENT_HANDSHAKES = 32;
+  private static final AsyncSemaphore handshakeSemaphore =
+      new AsyncSemaphore(getAltsMaxConcurrentHandshakes());
 
   @Grpc.TransportAttr
   public static final Attributes.Key<TsiPeer> TSI_PEER_KEY =
@@ -424,5 +426,30 @@ public final class AltsProtocolNegotiator {
     }
   }
 
+  @VisibleForTesting
+  static int getAltsMaxConcurrentHandshakes(String altsMaxConcurrentHandshakes) {
+    if (altsMaxConcurrentHandshakes == null) {
+      return DEFAULT_ALTS_MAX_CONCURRENT_HANDSHAKES;
+    }
+    try {
+      int effectiveMaxConcurrentHandshakes = Integer.parseInt(altsMaxConcurrentHandshakes);
+      if (effectiveMaxConcurrentHandshakes < 0) {
+        logger.warning(
+            "GRPC_ALTS_MAX_CONCURRENT_HANDSHAKES environment variable set to invalid value.");
+        return DEFAULT_ALTS_MAX_CONCURRENT_HANDSHAKES;
+      }
+      return effectiveMaxConcurrentHandshakes;
+    } catch (NumberFormatException e) {
+      logger.warning(
+          "GRPC_ALTS_MAX_CONCURRENT_HANDSHAKES environment variable set to invalid value.");
+      return DEFAULT_ALTS_MAX_CONCURRENT_HANDSHAKES;
+    }
+  }
+
+  private static int getAltsMaxConcurrentHandshakes() {
+    return getAltsMaxConcurrentHandshakes(
+        System.getenv(ALTS_MAX_CONCURRENT_HANDSHAKES_ENV_VARIABLE));
+  }
+
   private AltsProtocolNegotiator() {}
 }

alts/src/test/java/io/grpc/alts/internal/AltsProtocolNegotiatorTest.java

@@ -354,6 +354,29 @@ public class AltsProtocolNegotiatorTest {
         .isEqualTo(SecurityLevel.PRIVACY_AND_INTEGRITY);
   }
 
+  @Test
+  public void getAltsMaxConcurrentHandshakes_success() throws Exception {
+    assertThat(AltsProtocolNegotiator.getAltsMaxConcurrentHandshakes("10")).isEqualTo(10);
+  }
+
+  @Test
+  public void getAltsMaxConcurrentHandshakes_envVariableNotSet() throws Exception {
+    assertThat(AltsProtocolNegotiator.getAltsMaxConcurrentHandshakes(null))
+        .isEqualTo(AltsProtocolNegotiator.DEFAULT_ALTS_MAX_CONCURRENT_HANDSHAKES);
+  }
+
+  @Test
+  public void getAltsMaxConcurrentHandshakes_envVariableNotANumber() throws Exception {
+    assertThat(AltsProtocolNegotiator.getAltsMaxConcurrentHandshakes("not-a-number"))
+        .isEqualTo(AltsProtocolNegotiator.DEFAULT_ALTS_MAX_CONCURRENT_HANDSHAKES);
+  }
+
+  @Test
+  public void getAltsMaxConcurrentHandshakes_envVariableNegative() throws Exception {
+    assertThat(AltsProtocolNegotiator.getAltsMaxConcurrentHandshakes("-10"))
+        .isEqualTo(AltsProtocolNegotiator.DEFAULT_ALTS_MAX_CONCURRENT_HANDSHAKES);
+  }
+
   private void doHandshake() throws Exception {
     // Capture the client frame and add to the server.
     assertEquals(1, channel.outboundMessages().size());