feat(aws-crt-client): add numEventLoopThreads

.changes/next-release/feature-AWSCRTHTTPClient-3532d9a.json

@@ -0,0 +1,6 @@
+{
+    "type": "feature",
+    "category": "AWS CRT HTTP Client",
+    "contributor": "",
+    "description": "Add `numEventLoopThreads(Integer)` to `AwsCrtAsyncHttpClient.Builder` and `AwsCrtHttpClient.Builder` to configure the number of CRT event-loop (IO) threads. When set, the client owns a private `EventLoopGroup` of that size (must be greater than 1); when unset, it shares the process-wide default group."
+}

http-clients/aws-crt-client/pom.xml

@@ -213,6 +213,17 @@
                     </archive>
                 </configuration>
             </plugin>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-surefire-plugin</artifactId>
+                <configuration>
+                    <!-- Enable CRT native-resource tracking for the whole fork. The flag is latched once at CrtResource
+                         class-load, so it must be present at JVM launch (before any test loads a CRT type). -->
+                    <systemPropertyVariables>
+                        <aws.crt.debugnative>true</aws.crt.debugnative>
+                    </systemPropertyVariables>
+                </configuration>
+            </plugin>
         </plugins>
     </build>
 </project>

http-clients/aws-crt-client/src/main/java/software/amazon/awssdk/http/crt/AwsCrtAsyncHttpClient.java

@@ -129,6 +129,26 @@ public interface Builder extends SdkAsyncHttpClient.Builder<AwsCrtAsyncHttpClien
          */
         AwsCrtAsyncHttpClient.Builder readBufferSizeInBytes(Long readBufferSize);
 
+        /**
+         * Configure the number of event-loop (IO) threads in this client's event-loop group.
+         *
+         * <p>By default (when this is not set), the client shares a single, process-wide event-loop group sized to
+         * {@code Runtime.getRuntime().availableProcessors()}, shared with every other CRT client in the JVM. When this value is
+         * set, the client instead creates and owns a private event-loop group of the given size; that group is shut down when
+         * this client is closed and is not shared with any other client.
+         *
+         * <p>This is an advanced tuning and isolation control, and each client configured with an explicit size consumes that
+         * many additional IO threads. Oversizing wastes threads and adds context-switching and memory overhead without improving
+         * throughput; undersizing can leave the client's IO as a bottleneck and underutilize available cores. The best value
+         * depends on your workload and hardware, so benchmark your own application before changing it from the default. An
+         * excessively high value relative to the number of available processors is logged as a warning.
+         *
+         * @param numEventLoopThreads the number of event-loop threads; must be greater than 1, or {@code null} to use the shared
+         *                           default.
+         * @return The builder for method chaining.
+         */
+        AwsCrtAsyncHttpClient.Builder numEventLoopThreads(Integer numEventLoopThreads);
+
         /**
          * Sets the http proxy configuration to use for this client.
          * @param proxyConfiguration The http proxy configuration to use

http-clients/aws-crt-client/src/main/java/software/amazon/awssdk/http/crt/AwsCrtHttpClient.java

@@ -56,12 +56,16 @@
 public final class AwsCrtHttpClient extends AwsCrtHttpClientBase implements SdkHttpClient {
 
     private AwsCrtHttpClient(DefaultBuilder builder, AttributeMap config) {
-        super(builder, config);
-        if (this.protocol == Protocol.HTTP2) {
+        super(builder, validateProtocol(config));
+    }
+
+    private static AttributeMap validateProtocol(AttributeMap config) {
+        if (config.get(SdkHttpConfigurationOption.PROTOCOL) == Protocol.HTTP2) {
             throw new UnsupportedOperationException(
                 "HTTP/2 is not supported for sync HTTP clients. Either use HTTP/1.1 (the default) or use an async "
                 + "HTTP client (e.g., AwsCrtAsyncHttpClient).");
         }
+        return config;
     }
 
     public static AwsCrtHttpClient.Builder builder() {
@@ -181,6 +185,26 @@ public interface Builder extends SdkHttpClient.Builder<AwsCrtHttpClient.Builder>
          */
         AwsCrtHttpClient.Builder readBufferSizeInBytes(Long readBufferSize);
 
+        /**
+         * Configure the number of event-loop (IO) threads in this client's event-loop group.
+         *
+         * <p>By default (when this is not set), the client shares a single, process-wide event-loop group sized to
+         * {@code Runtime.getRuntime().availableProcessors()}, shared with every other CRT client in the JVM. When this value is
+         * set, the client instead creates and owns a private event-loop group of the given size; that group is shut down when
+         * this client is closed and is not shared with any other client.
+         *
+         * <p>This is an advanced tuning and isolation control, and each client configured with an explicit size consumes that
+         * many additional IO threads. Oversizing wastes threads and adds context-switching and memory overhead without improving
+         * throughput; undersizing can leave the client's IO as a bottleneck and underutilize available cores. The best value
+         * depends on your workload and hardware, so benchmark your own application before changing it from the default. An
+         * excessively high value relative to the number of available processors is logged as a warning.
+         *
+         * @param numEventLoopThreads the number of event-loop threads; must be greater than 1, or {@code null} to use the shared
+         *                           default.
+         * @return The builder for method chaining.
+         */
+        AwsCrtHttpClient.Builder numEventLoopThreads(Integer numEventLoopThreads);
+
         /**
          * Sets the http proxy configuration to use for this client.
          * @param proxyConfiguration The http proxy configuration to use

http-clients/aws-crt-client/src/main/java/software/amazon/awssdk/http/crt/AwsCrtHttpClientBase.java

@@ -40,6 +40,7 @@
 import software.amazon.awssdk.crt.http.HttpStreamManagerOptions;
 import software.amazon.awssdk.crt.http.HttpVersion;
 import software.amazon.awssdk.crt.io.ClientBootstrap;
+import software.amazon.awssdk.crt.io.EventLoopGroup;
 import software.amazon.awssdk.crt.io.SocketOptions;
 import software.amazon.awssdk.crt.io.TlsConnectionOptions;
 import software.amazon.awssdk.crt.io.TlsContext;
@@ -72,6 +73,10 @@ abstract class AwsCrtHttpClientBase implements SdkAutoCloseable {
     private static final String AWS_COMMON_RUNTIME = "AwsCommonRuntime";
     private static final long DEFAULT_STREAM_WINDOW_SIZE = 16L * 1024L * 1024L; // 16 MB
 
+    // Heuristic threshold (not an API contract) for warning about likely-accidental oversizing of the per-client
+    // event-loop group. A value at or above this multiple of the available processors is almost certainly unintended.
+    private static final int NUM_EVENT_LOOP_THREADS_WARN_MULTIPLIER = 4;
+
     protected final long readBufferSize;
     protected final Protocol protocol;
     private final Map<URI, HttpStreamManager> connectionPools = new ConcurrentHashMap<>();
@@ -89,35 +94,77 @@ abstract class AwsCrtHttpClientBase implements SdkAutoCloseable {
     private boolean isClosed = false;
 
     AwsCrtHttpClientBase(AwsCrtClientBuilderBase builder, AttributeMap config) {
-        ClientBootstrap clientBootstrap = new ClientBootstrap(null, null);
-        SocketOptions clientSocketOptions = buildSocketOptions(builder.getTcpKeepAliveConfiguration(),
-                                                               config.get(SdkHttpConfigurationOption.CONNECTION_TIMEOUT));
-        TlsContextOptions clientTlsContextOptions =
-            TlsContextOptions.createDefaultClient()
-                             .withCipherPreference(resolveCipherPreference(builder.getPostQuantumTlsEnabled()))
-                             .withMinimumTlsVersion(resolveMinTlsVersion(builder.getMinTlsVersion()))
-                             .withVerifyPeer(!config.get(SdkHttpConfigurationOption.TRUST_ALL_CERTIFICATES));
-        this.protocol = config.get(PROTOCOL);
-        if (protocol == Protocol.HTTP2) {
-            clientTlsContextOptions = clientTlsContextOptions.withAlpnList("h2");
+        // These native resources are created before being registered as owned, and each creation can throw
+        // (e.g. CrtRuntimeException). Because close() is never invoked on a client that failed to construct, any
+        // already-created resource must be released here if a later step throws - otherwise the private EventLoopGroup
+        // (and its OS threads) and the other native handles would leak.
+        EventLoopGroup eventLoopGroup = null;
+        ClientBootstrap clientBootstrap = null;
+        SocketOptions clientSocketOptions = null;
+        TlsContextOptions clientTlsContextOptions = null;
+        TlsContext clientTlsContext = null;
+        try {
+            Integer numEventLoopThreads = builder.getNumEventLoopThreads();
+            if (numEventLoopThreads != null) {
+                warnIfNumEventLoopThreadsIsExcessive(numEventLoopThreads);
+                eventLoopGroup = new EventLoopGroup(numEventLoopThreads);
+            }
+            clientBootstrap = new ClientBootstrap(eventLoopGroup, null);
+            clientSocketOptions = buildSocketOptions(builder.getTcpKeepAliveConfiguration(),
+                                                     config.get(SdkHttpConfigurationOption.CONNECTION_TIMEOUT));
+            clientTlsContextOptions =
+                TlsContextOptions.createDefaultClient()
+                                 .withCipherPreference(resolveCipherPreference(builder.getPostQuantumTlsEnabled()))
+                                 .withMinimumTlsVersion(resolveMinTlsVersion(builder.getMinTlsVersion()))
+                                 .withVerifyPeer(!config.get(SdkHttpConfigurationOption.TRUST_ALL_CERTIFICATES));
+            this.protocol = config.get(PROTOCOL);
+            if (protocol == Protocol.HTTP2) {
+                clientTlsContextOptions = clientTlsContextOptions.withAlpnList("h2");
+            }
+
+            this.tlsContextOptions = registerOwnedResource(clientTlsContextOptions);
+            clientTlsContext = new TlsContext(clientTlsContextOptions);
+
+            // The bootstrap holds a native reference to its event-loop group, so the group is registered before (and thus
+            // closed after) the bootstrap to keep CRT teardown ordering correct. A null group leaves the shared static
+            // default group untouched.
+            registerOwnedResource(eventLoopGroup);
+            this.bootstrap = registerOwnedResource(clientBootstrap);
+            this.socketOptions = registerOwnedResource(clientSocketOptions);
+            this.tlsContext = registerOwnedResource(clientTlsContext);
+            this.tlsNegotiationTimeout = config.get(SdkHttpConfigurationOption.TLS_NEGOTIATION_TIMEOUT);
+            this.readBufferSize = builder.getReadBufferSizeInBytes() == null ?
+                                  DEFAULT_STREAM_WINDOW_SIZE : builder.getReadBufferSizeInBytes();
+            this.maxStreamsPerEndpoint = config.get(SdkHttpConfigurationOption.MAX_CONNECTIONS);
+            this.monitoringOptions =
+                resolveHttpMonitoringOptions(builder.getConnectionHealthConfiguration())
+                    .orElse(null);
+            this.maxConnectionIdleInMilliseconds = config.get(SdkHttpConfigurationOption.CONNECTION_MAX_IDLE_TIMEOUT).toMillis();
+            this.connectionAcquisitionTimeout = config.get(SdkHttpConfigurationOption.CONNECTION_ACQUIRE_TIMEOUT).toMillis();
+            this.proxyOptions = resolveProxy(builder.getProxyConfiguration(), tlsContext).orElse(null);
+        } catch (RuntimeException e) {
+            // Release in reverse dependency order: the TlsContext wraps its options, and the bootstrap holds a
+            // reference to the event-loop group, so those wrappers are closed before what they depend on.
+            IoUtils.closeQuietly(clientTlsContext, log.logger());
+            IoUtils.closeQuietly(clientTlsContextOptions, log.logger());
+            IoUtils.closeQuietly(clientSocketOptions, log.logger());
+            IoUtils.closeQuietly(clientBootstrap, log.logger());
+            IoUtils.closeQuietly(eventLoopGroup, log.logger());
+            throw e;
         }
+    }
 
-        this.tlsContextOptions = registerOwnedResource(clientTlsContextOptions);
-        TlsContext clientTlsContext = new TlsContext(clientTlsContextOptions);
-
-        this.bootstrap = registerOwnedResource(clientBootstrap);
-        this.socketOptions = registerOwnedResource(clientSocketOptions);
-        this.tlsContext = registerOwnedResource(clientTlsContext);
-        this.tlsNegotiationTimeout = config.get(SdkHttpConfigurationOption.TLS_NEGOTIATION_TIMEOUT);
-        this.readBufferSize = builder.getReadBufferSizeInBytes() == null ?
-                              DEFAULT_STREAM_WINDOW_SIZE : builder.getReadBufferSizeInBytes();
-        this.maxStreamsPerEndpoint = config.get(SdkHttpConfigurationOption.MAX_CONNECTIONS);
-        this.monitoringOptions =
-            resolveHttpMonitoringOptions(builder.getConnectionHealthConfiguration())
-                .orElse(null);
-        this.maxConnectionIdleInMilliseconds = config.get(SdkHttpConfigurationOption.CONNECTION_MAX_IDLE_TIMEOUT).toMillis();
-        this.connectionAcquisitionTimeout = config.get(SdkHttpConfigurationOption.CONNECTION_ACQUIRE_TIMEOUT).toMillis();
-        this.proxyOptions = resolveProxy(builder.getProxyConfiguration(), tlsContext).orElse(null);
+    private static void warnIfNumEventLoopThreadsIsExcessive(int numEventLoopThreads) {
+        int availableProcessors = Math.max(1, Runtime.getRuntime().availableProcessors());
+        if (numEventLoopThreads >= NUM_EVENT_LOOP_THREADS_WARN_MULTIPLIER * availableProcessors) {
+            log.warn(() -> String.format(
+                "numEventLoopThreads is set to %d, which is unusually high relative to the %d available processor(s). "
+                + "Each client configured with numEventLoopThreads gets its own private event-loop group, so a high count "
+                + "multiplied across multiple clients can lead to thread explosion, excessive context-switching, and increased "
+                + "memory use without improving throughput. Consider benchmarking your workload to confirm this value is "
+                + "necessary.",
+                numEventLoopThreads, availableProcessors));
+        }
     }
 
     /**

http-clients/aws-crt-client/src/main/java/software/amazon/awssdk/http/crt/internal/AwsCrtClientBuilderBase.java

@@ -35,6 +35,7 @@ public class AwsCrtClientBuilderBase<BuilderT> {
     private TcpKeepAliveConfiguration tcpKeepAliveConfiguration;
     private Boolean postQuantumTlsEnabled;
     private TlsVersion minTlsVersion;
+    private Integer numEventLoopThreads;
 
     protected AwsCrtClientBuilderBase() {
     }
@@ -63,6 +64,17 @@ public Long getReadBufferSizeInBytes() {
         return this.readBufferSize;
     }
 
+    public BuilderT numEventLoopThreads(Integer numEventLoopThreads) {
+        Validate.isTrue(numEventLoopThreads == null || numEventLoopThreads > 1,
+                        "numEventLoopThreads must be greater than 1");
+        this.numEventLoopThreads = numEventLoopThreads;
+        return thisBuilder();
+    }
+
+    public Integer getNumEventLoopThreads() {
+        return this.numEventLoopThreads;
+    }
+
 
     public BuilderT proxyConfiguration(ProxyConfiguration proxyConfiguration) {
         this.proxyConfiguration = proxyConfiguration;