feat(rt)!:introduce maxConcurrency HTTP setting and rename OkHttp maxConnectionsPerHost (#899)

Author: aajtodd

.changes/c5f04082-06d1-4425-b7b9-b1ac2b9b8edd.json

@@ -0,0 +1,8 @@
+{
+    "id": "c5f04082-06d1-4425-b7b9-b1ac2b9b8edd",
+    "type": "feature",
+    "description": "BREAKING: introduce `maxConcurrency` HTTP engine setting and rename OkHttp specific `maxConnectionsPerHost` to `maxConcurrencyPerHost`.",
+    "issues": [
+        "awslabs/smithy-kotlin#898"
+    ]
+}

runtime/protocol/http-client-engines/http-client-engine-crt/common/src/aws/smithy/kotlin/runtime/http/engine/crt/CrtHttpEngine.kt

@@ -26,11 +26,11 @@ import aws.smithy.kotlin.runtime.io.internal.SdkDispatchers
 import aws.smithy.kotlin.runtime.operation.ExecutionContext
 import aws.smithy.kotlin.runtime.telemetry.logging.logger
 import aws.smithy.kotlin.runtime.time.Instant
-import kotlinx.coroutines.job
+import kotlinx.coroutines.*
 import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.Semaphore
 import kotlinx.coroutines.sync.withLock
-import kotlinx.coroutines.withContext
-import kotlinx.coroutines.withTimeoutOrNull
+import kotlinx.coroutines.sync.withPermit
 import aws.sdk.kotlin.crt.io.TlsContext as CrtTlsContext
 import aws.sdk.kotlin.crt.io.TlsVersion as CrtTlsVersion
 import aws.smithy.kotlin.runtime.config.TlsVersion as SdkTlsVersion
@@ -83,15 +83,17 @@ public class CrtHttpEngine(public override val config: CrtHttpEngineConfig) : Ht
             connectTimeoutMs = config.connectTimeout.inWholeMilliseconds.toInt(),
         )
         initialWindowSize = config.initialWindowSizeBytes
+        // FIXME - given managers are _per host_ the maxConnections parameter is not actually respected here
         maxConnections = config.maxConnections.toInt()
         maxConnectionIdleMs = config.connectionIdleTimeout.inWholeMilliseconds
     }
 
     // connection managers are per host
     private val connManagers = mutableMapOf<String, HttpClientConnectionManager>()
     private val mutex = Mutex()
+    private val requestLimiter = Semaphore(config.maxConcurrency.toInt())
 
-    override suspend fun roundTrip(context: ExecutionContext, request: HttpRequest): HttpCall {
+    override suspend fun roundTrip(context: ExecutionContext, request: HttpRequest): HttpCall = requestLimiter.withPermit {
         val callContext = callContext()
         val logger = callContext.logger<CrtHttpEngine>()
         val proxyConfig = config.proxySelector.select(request.url)

runtime/protocol/http-client-engines/http-client-engine-okhttp/api/http-client-engine-okhttp.api

@@ -15,15 +15,15 @@ public final class aws/smithy/kotlin/runtime/http/engine/okhttp/OkHttpEngine$Com
 public final class aws/smithy/kotlin/runtime/http/engine/okhttp/OkHttpEngineConfig : aws/smithy/kotlin/runtime/http/engine/HttpClientEngineConfigImpl {
 	public static final field Companion Laws/smithy/kotlin/runtime/http/engine/okhttp/OkHttpEngineConfig$Companion;
 	public synthetic fun <init> (Laws/smithy/kotlin/runtime/http/engine/okhttp/OkHttpEngineConfig$Builder;Lkotlin/jvm/internal/DefaultConstructorMarker;)V
-	public final fun getMaxConnectionsPerHost-pVg5ArA ()I
+	public final fun getMaxConcurrencyPerHost-pVg5ArA ()I
 	public fun toBuilderApplicator ()Lkotlin/jvm/functions/Function1;
 }
 
 public final class aws/smithy/kotlin/runtime/http/engine/okhttp/OkHttpEngineConfig$Builder : aws/smithy/kotlin/runtime/http/engine/HttpClientEngineConfigImpl$BuilderImpl {
 	public fun <init> ()V
-	public final fun getMaxConnectionsPerHost-0hXNFcg ()Lkotlin/UInt;
+	public final fun getMaxConcurrencyPerHost-0hXNFcg ()Lkotlin/UInt;
 	public fun getTelemetryProvider ()Laws/smithy/kotlin/runtime/telemetry/TelemetryProvider;
-	public final fun setMaxConnectionsPerHost-ExVfyTY (Lkotlin/UInt;)V
+	public final fun setMaxConcurrencyPerHost-ExVfyTY (Lkotlin/UInt;)V
 	public fun setTelemetryProvider (Laws/smithy/kotlin/runtime/telemetry/TelemetryProvider;)V
 }
 

runtime/protocol/http-client-engines/http-client-engine-okhttp/jvm/src/aws/smithy/kotlin/runtime/http/engine/okhttp/OkHttpEngine.kt

@@ -95,6 +95,7 @@ private fun OkHttpEngineConfig.buildClient(metrics: HttpClientMetrics): OkHttpCl
 
         // FIXME - register a [ConnectionListener](https://github.com/square/okhttp/blob/master/okhttp/src/jvmMain/kotlin/okhttp3/ConnectionListener.kt#L27)
         // when a new okhttp release is cut that contains this abstraction and wireup connection uptime metrics
+        // FIXME - with that ConnectionListener implement our own max connections gate?
 
         // use our own pool configured with the timeout settings taken from config
         val pool = ConnectionPool(
@@ -104,11 +105,9 @@ private fun OkHttpEngineConfig.buildClient(metrics: HttpClientMetrics): OkHttpCl
         )
         connectionPool(pool)
 
-        // Configure a dispatcher that uses maxConnections as a proxy for maxRequests. Note that this isn't precisely
-        // the same since some protocols (e.g., HTTP2) may use a single connection for multiple requests.
         val dispatcher = Dispatcher().apply {
-            maxRequests = config.maxConnections.toInt()
-            maxRequestsPerHost = config.maxConnectionsPerHost.toInt()
+            maxRequests = config.maxConcurrency.toInt()
+            maxRequestsPerHost = config.maxConcurrencyPerHost.toInt()
         }
         dispatcher(dispatcher)
 

runtime/protocol/http-client-engines/http-client-engine-okhttp/jvm/src/aws/smithy/kotlin/runtime/http/engine/okhttp/OkHttpEngineConfig.kt

@@ -29,15 +29,15 @@ public class OkHttpEngineConfig private constructor(builder: Builder) : HttpClie
     }
 
     /**
-     * The maximum number of connections to open to a single host.
+     * The maximum number of requests to execute concurrently for a single host.
      */
-    public val maxConnectionsPerHost: UInt = builder.maxConnectionsPerHost ?: builder.maxConnections
+    public val maxConcurrencyPerHost: UInt = builder.maxConcurrencyPerHost ?: builder.maxConcurrency
 
     override fun toBuilderApplicator(): HttpClientEngineConfig.Builder.() -> Unit = {
         super.toBuilderApplicator()()
 
         if (this is Builder) {
-            maxConnectionsPerHost = this@OkHttpEngineConfig.maxConnectionsPerHost
+            maxConcurrencyPerHost = this@OkHttpEngineConfig.maxConcurrencyPerHost
         }
     }
 
@@ -46,9 +46,9 @@ public class OkHttpEngineConfig private constructor(builder: Builder) : HttpClie
      */
     public class Builder : BuilderImpl() {
         /**
-         * The maximum number of connections to open to a single host. Defaults to [maxConnections].
+         * The maximum number of requests to execute concurrently for a single host. Defaults to [maxConcurrency].
          */
-        public var maxConnectionsPerHost: UInt? = null
+        public var maxConcurrencyPerHost: UInt? = null
 
         override var telemetryProvider: TelemetryProvider = TelemetryProvider.Global
     }

runtime/protocol/http-client/api/http-client.api

@@ -73,6 +73,7 @@ public abstract interface class aws/smithy/kotlin/runtime/http/engine/HttpClient
 	public abstract fun getConnectionAcquireTimeout-UwyO8pc ()J
 	public abstract fun getConnectionIdleTimeout-UwyO8pc ()J
 	public abstract fun getHostResolver ()Laws/smithy/kotlin/runtime/net/HostResolver;
+	public abstract fun getMaxConcurrency-pVg5ArA ()I
 	public abstract fun getMaxConnections-pVg5ArA ()I
 	public abstract fun getProxySelector ()Laws/smithy/kotlin/runtime/http/engine/ProxySelector;
 	public abstract fun getSocketReadTimeout-UwyO8pc ()J
@@ -87,6 +88,7 @@ public abstract interface class aws/smithy/kotlin/runtime/http/engine/HttpClient
 	public abstract fun getConnectionAcquireTimeout-UwyO8pc ()J
 	public abstract fun getConnectionIdleTimeout-UwyO8pc ()J
 	public abstract fun getHostResolver ()Laws/smithy/kotlin/runtime/net/HostResolver;
+	public abstract fun getMaxConcurrency-pVg5ArA ()I
 	public abstract fun getMaxConnections-pVg5ArA ()I
 	public abstract fun getProxySelector ()Laws/smithy/kotlin/runtime/http/engine/ProxySelector;
 	public abstract fun getSocketReadTimeout-UwyO8pc ()J
@@ -97,6 +99,7 @@ public abstract interface class aws/smithy/kotlin/runtime/http/engine/HttpClient
 	public abstract fun setConnectionAcquireTimeout-LRDsOJo (J)V
 	public abstract fun setConnectionIdleTimeout-LRDsOJo (J)V
 	public abstract fun setHostResolver (Laws/smithy/kotlin/runtime/net/HostResolver;)V
+	public abstract fun setMaxConcurrency-WZ4Q5Ns (I)V
 	public abstract fun setMaxConnections-WZ4Q5Ns (I)V
 	public abstract fun setProxySelector (Laws/smithy/kotlin/runtime/http/engine/ProxySelector;)V
 	public abstract fun setSocketReadTimeout-LRDsOJo (J)V

runtime/protocol/http-client/common/src/aws/smithy/kotlin/runtime/http/engine/HttpClientEngineConfig.kt

@@ -65,6 +65,12 @@ public interface HttpClientEngineConfig {
      */
     public val connectionIdleTimeout: Duration
 
+    /**
+     * The maximum number of requests that will be executed concurrently by an engine. Beyond this requests
+     * will be queued waiting to be executed by the engine.
+     */
+    public val maxConcurrency: UInt
+
     /**
      * The proxy selection policy
      */
@@ -132,6 +138,12 @@ public interface HttpClientEngineConfig {
          */
         public var connectionIdleTimeout: Duration
 
+        /**
+         * The maximum number of requests that will be executed concurrently by an engine. Beyond this requests
+         * will be queued waiting to be executed by the engine.
+         */
+        public var maxConcurrency: UInt
+
         /**
          * Set the proxy selection policy to be used.
          *
@@ -191,6 +203,7 @@ public open class HttpClientEngineConfigImpl(builder: HttpClientEngineConfig.Bui
     override val connectTimeout: Duration = builder.connectTimeout
     override val connectionAcquireTimeout: Duration = builder.connectionAcquireTimeout
     override val connectionIdleTimeout: Duration = builder.connectionIdleTimeout
+    override val maxConcurrency: UInt = builder.maxConcurrency
     override val proxySelector: ProxySelector = builder.proxySelector
     override val hostResolver: HostResolver = builder.hostResolver
     override val tlsContext: TlsContext = builder.tlsContext
@@ -203,6 +216,7 @@ public open class HttpClientEngineConfigImpl(builder: HttpClientEngineConfig.Bui
         connectTimeout = this@HttpClientEngineConfigImpl.connectTimeout
         connectionAcquireTimeout = this@HttpClientEngineConfigImpl.connectionAcquireTimeout
         connectionIdleTimeout = this@HttpClientEngineConfigImpl.connectionIdleTimeout
+        maxConcurrency = this@HttpClientEngineConfigImpl.maxConcurrency
         proxySelector = this@HttpClientEngineConfigImpl.proxySelector
         hostResolver = this@HttpClientEngineConfigImpl.hostResolver
         tlsContext = this@HttpClientEngineConfigImpl.tlsContext
@@ -213,10 +227,11 @@ public open class HttpClientEngineConfigImpl(builder: HttpClientEngineConfig.Bui
     public open class BuilderImpl : HttpClientEngineConfig.Builder {
         override var socketReadTimeout: Duration = 30.seconds
         override var socketWriteTimeout: Duration = 30.seconds
-        override var maxConnections: UInt = 16u
+        override var maxConnections: UInt = 64u
         override var connectTimeout: Duration = 2.seconds
         override var connectionAcquireTimeout: Duration = 10.seconds
         override var connectionIdleTimeout: Duration = 60.seconds
+        override var maxConcurrency: UInt = 128u
         override var proxySelector: ProxySelector = EnvironmentProxySelector()
         override var hostResolver: HostResolver = HostResolver.Default
         override var tlsContext: TlsContext = TlsContext.Default