max concurrent streams for h2 stream manager (#977)

Author: TingDaoK

crt/aws-c-http

@@ -123,6 +123,7 @@ private Http2StreamManager(Http2StreamManagerOptions options) {
                 maxConnections,
                 idealConcurrentStreamsPerConnection,
                 maxConcurrentStreamsPerConnection,
+                options.getMaxConcurrentStreams(),
                 options.hasPriorKnowledge(),
                 options.shouldCloseConnectionOnServerError(),
                 options.getConnectionPingPeriodMs(),
@@ -265,6 +266,7 @@ private static native long http2StreamManagerNew(Http2StreamManager thisObj,
             int maxConns,
             int ideal_concurrent_streams_per_connection,
             int max_concurrent_streams_per_connection,
+            int max_concurrent_streams,
             boolean priorKnowledge,
             boolean closeConnectionOnServerError,
             int connectionPingPeriodMs,

src/main/java/software/amazon/awssdk/crt/http/Http2StreamManager.java

@@ -9,9 +9,7 @@
  * instance
  */
 public class Http2StreamManagerOptions {
-    public static final int DEFAULT_MAX_WINDOW_SIZE = Integer.MAX_VALUE;
     public static final int DEFAULT_MAX = Integer.MAX_VALUE;
-    public static final int DEFAULT_MAX_CONNECTIONS = 2;
     public static final int DEFAULT_CONNECTION_PING_TIMEOUT_MS = 3000;
     private static final String HTTPS = "https";
 
@@ -20,6 +18,7 @@ public class Http2StreamManagerOptions {
     private int idealConcurrentStreamsPerConnection = 100;
     private boolean connectionManualWindowManagement = false;
     private int maxConcurrentStreamsPerConnection = DEFAULT_MAX;
+    private int maxConcurrentStreams = 0;
 
     private boolean priorKnowledge = false;
     private boolean closeConnectionOnServerError = false;
@@ -108,6 +107,28 @@ public int getMaxConcurrentStreamsPerConnection() {
         return maxConcurrentStreamsPerConnection;
     }
 
+    /**
+     * The max number of concurrent streams that can be active across all connections
+     * at the same time. 0 means no limit (default). When this limit is reached, the
+     * stream manager will wait for existing streams to complete before creating new
+     * ones, even if connections have available capacity.
+     *
+     * @param maxConcurrentStreams The max number of concurrent streams across all connections
+     * @return this
+     */
+    public Http2StreamManagerOptions withMaxConcurrentStreams(int maxConcurrentStreams) {
+        this.maxConcurrentStreams = maxConcurrentStreams;
+        return this;
+    }
+
+    /**
+     * @return The max number of concurrent streams across all connections.
+     *         0 means no limit (default).
+     */
+    public int getMaxConcurrentStreams() {
+        return maxConcurrentStreams;
+    }
+
     /**
      * @return The connection level manual flow control enabled or not.
      */

src/main/java/software/amazon/awssdk/crt/http/Http2StreamManagerOptions.java

@@ -113,6 +113,7 @@ JNIEXPORT jlong JNICALL Java_software_amazon_awssdk_crt_http_Http2StreamManager_
     jint jni_max_conns,
     jint jni_ideal_concurrent_streams_per_connection,
     jint jni_max_concurrent_streams_per_connection,
+    jint jni_max_concurrent_streams,
     jboolean jni_prior_knowledge,
     jboolean jni_close_connection_on_server_error,
     jint jni_connection_ping_period_ms,
@@ -207,6 +208,7 @@ JNIEXPORT jlong JNICALL Java_software_amazon_awssdk_crt_http_Http2StreamManager_
         .ideal_concurrent_streams_per_connection = (size_t)jni_ideal_concurrent_streams_per_connection,
         .max_concurrent_streams_per_connection = (size_t)jni_max_concurrent_streams_per_connection,
         .max_connections = (size_t)jni_max_conns,
+        .max_concurrent_streams = (size_t)jni_max_concurrent_streams,
     };
 
     struct aws_http_connection_monitoring_options monitoring_options;

src/native/http2_stream_manager.c

@@ -191,4 +191,104 @@ public void testStreamManagerMetrics() throws Exception {
         CrtResource.logNativeResources();
         CrtResource.waitForNoResources();
     }
+
+    @Test
+    public void testMaxConcurrentStreamsEnforcement() throws Exception {
+        skipIfAndroid();
+        skipIfNetworkUnavailable();
+
+        URI uri = new URI(endpoint);
+        int maxConcurrentStreams = 1;
+        int numRequestsToMake = 20;
+
+        try (EventLoopGroup eventLoopGroup = new EventLoopGroup(1);
+                HostResolver resolver = new HostResolver(eventLoopGroup);
+                ClientBootstrap bootstrap = new ClientBootstrap(eventLoopGroup, resolver);
+                SocketOptions sockOpts = new SocketOptions();
+                TlsContextOptions tlsOpts = TlsContextOptions.createDefaultClient().withAlpnList("h2");
+                TlsContext tlsContext = createHttpClientTlsContext(tlsOpts)) {
+
+            Http2StreamManagerOptions options = new Http2StreamManagerOptions();
+            options.withMaxConcurrentStreams(maxConcurrentStreams)
+                   .withMaxConcurrentStreamsPerConnection(100)
+                   .withIdealConcurrentStreamsPerConnection(100);
+
+            HttpClientConnectionManagerOptions connectionManagerOptions = new HttpClientConnectionManagerOptions();
+            connectionManagerOptions.withClientBootstrap(bootstrap)
+                    .withSocketOptions(sockOpts)
+                    .withTlsContext(tlsContext)
+                    .withUri(uri)
+                    .withMaxConnections(2);
+            options.withConnectionManagerOptions(connectionManagerOptions);
+
+            Http2StreamManager streamManager = Http2StreamManager.create(options);
+
+            try {
+                Http2Request request = createHttp2Request("GET", endpoint, path, EMPTY_BODY);
+
+                final AtomicInteger completedRequests = new AtomicInteger(0);
+                final AtomicInteger maxConcurrentActive = new AtomicInteger(0);
+                final AtomicInteger currentActive = new AtomicInteger(0);
+                final CountDownLatch allRequestsComplete = new CountDownLatch(numRequestsToMake);
+
+                // Acquire multiple streams rapidly
+                for (int i = 0; i < numRequestsToMake; i++) {
+                    streamManager.acquireStream(request, new HttpStreamBaseResponseHandler() {
+                        @Override
+                        public void onResponseHeaders(HttpStreamBase stream, int responseStatusCode, int blockType,
+                                HttpHeader[] nextHeaders) {
+                            // Track concurrent active streams
+                            int active = currentActive.incrementAndGet();
+                            int prevMax = maxConcurrentActive.get();
+                            while (active > prevMax) {
+                                if (maxConcurrentActive.compareAndSet(prevMax, active)) {
+                                    break;
+                                }
+                                prevMax = maxConcurrentActive.get();
+                            }
+                        }
+
+                        @Override
+                        public void onResponseComplete(HttpStreamBase stream, int errorCode) {
+                            currentActive.decrementAndGet();
+                            completedRequests.incrementAndGet();
+                            stream.close();
+                            allRequestsComplete.countDown();
+                        }
+                    }).whenComplete((stream, throwable) -> {
+                        if (throwable != null) {
+                            allRequestsComplete.countDown();
+                        }
+                    });
+                }
+
+                // Wait for all requests to complete (with timeout)
+                boolean completed = allRequestsComplete.await(60, TimeUnit.SECONDS);
+                Assert.assertTrue("Requests did not complete within timeout", completed);
+
+                // Verify all requests completed
+                Assert.assertEquals(numRequestsToMake, completedRequests.get());
+
+                // Verify that max concurrent streams limit was respected
+                // The actual concurrent count should not exceed our limit significantly
+                // (allowing small margin for timing)
+                int observedMax = maxConcurrentActive.get();
+                Log.log(Log.LogLevel.Info, Log.LogSubject.HttpConnectionManager,
+                        String.format("Max concurrent streams observed: %d (limit: %d)", observedMax, maxConcurrentStreams));
+
+                // The observed max should be close to our limit (allowing some flexibility for race conditions)
+                Assert.assertTrue(
+                    String.format("Expected max concurrent streams around %d, but observed %d",
+                                  maxConcurrentStreams, observedMax),
+                    observedMax <= maxConcurrentStreams);
+
+            } finally {
+                streamManager.close();
+                streamManager.getShutdownCompleteFuture().get(60, TimeUnit.SECONDS);
+            }
+        }
+
+        CrtResource.logNativeResources();
+        CrtResource.waitForNoResources();
+    }
 }