Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fix: http call waiting for idle connection monitor to finish #1178

Merged
merged 5 commits into from
Nov 14, 2024
Merged

fix: http call waiting for idle connection monitor to finish #1178

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
ac87eab
fix: http call waiting for idle connection monitor to finish
0marperez Nov 11, 2024
330353e
lint
0marperez Nov 11, 2024
38e3cf2
PR feedback
0marperez Nov 13, 2024
6ef2a48
SoTimeout -> soTimeout in logs
0marperez Nov 13, 2024
3b23c49
Cancel and join monitor job
0marperez Nov 14, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 14 additions & 15 deletions ...gine-okhttp/jvm/src/aws/smithy/kotlin/runtime/http/engine/okhttp/ConnectionIdleMonitor.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,14 +5,7 @@
package aws.smithy.kotlin.runtime.http.engine.okhttp

import aws.smithy.kotlin.runtime.telemetry.logging.logger
import kotlinx.coroutines.CoroutineName
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.Job
import kotlinx.coroutines.cancelAndJoin
import kotlinx.coroutines.isActive
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.*
import okhttp3.Call
import okhttp3.Connection
import okhttp3.ConnectionListener
Expand All @@ -24,14 +17,20 @@ import okio.source
import java.net.SocketException
import java.net.SocketTimeoutException
import java.util.concurrent.ConcurrentHashMap
import kotlin.coroutines.CoroutineContext
import kotlin.coroutines.coroutineContext
import kotlin.time.Duration
import kotlin.time.measureTime

@OptIn(ExperimentalOkHttpApi::class)
internal class ConnectionIdleMonitor(val pollInterval: Duration) : ConnectionListener() {
private val monitorScope = CoroutineScope(Dispatchers.IO + SupervisorJob())
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Question: HTTP engines have their own coroutine scope. Could we reuse that as a parent scope instead of creating a new one from scratch?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think so, let me test this.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm seeing some issues trying to use the HTTP engine scope. We try to wait for the HTTP engine job to complete before calling shutdown. And that make us wait on the connection idle monitor to just fail it seems like, since it's running in a while loop.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hmm, that's unfortunate. We could probably structure engine closure a bit better to allow for child scopes/jobs but that's a bigger and riskier change. We can stick with this for now.

private val monitors = ConcurrentHashMap<Connection, Job>()

fun close(): Unit = runBlocking {
monitors.values.forEach { it.cancelAndJoin() }
}

private fun Call.callContext() =
request()
.tag(SdkRequestTag::class.java)
Expand All @@ -58,21 +57,20 @@ internal class ConnectionIdleMonitor(val pollInterval: Duration) : ConnectionLis

override fun connectionReleased(connection: Connection, call: Call) {
val connId = System.identityHashCode(connection)
val context = call.callContext()
val scope = CoroutineScope(context)
val monitor = scope.launch(CoroutineName("okhttp-conn-monitor-for-$connId")) {
doMonitor(connection)
val callContext = call.callContext()
val monitor = monitorScope.launch(CoroutineName("okhttp-conn-monitor-for-$connId")) {
doMonitor(connection, callContext)
}
context.logger<ConnectionIdleMonitor>().trace { "Launched coroutine $monitor to monitor $connection" }
callContext.logger<ConnectionIdleMonitor>().trace { "Launched coroutine $monitor to monitor $connection" }

// Non-locking map access is okay here because this code will only execute synchronously as part of a
// `connectionReleased` event and will be complete before any future `connectionAcquired` event could fire for
// the same connection.
monitors[connection] = monitor
}

private suspend fun doMonitor(conn: Connection) {
val logger = coroutineContext.logger<ConnectionIdleMonitor>()
private suspend fun doMonitor(conn: Connection, callContext: CoroutineContext) {
val logger = callContext.logger<ConnectionIdleMonitor>()

val socket = conn.socket()
val source = try {
Expand Down Expand Up @@ -111,6 +109,7 @@ internal class ConnectionIdleMonitor(val pollInterval: Duration) : ConnectionLis
logger.trace { "Attempting to reset soTimeout..." }
try {
conn.socket().soTimeout = oldTimeout
logger.trace { "soTimeout reset." }
} catch (e: Throwable) {
logger.warn(e) { "Failed to reset socket timeout on $conn. Connection may be unstable now." }
}
Expand Down
55 changes: 37 additions & 18 deletions ...client-engine-okhttp/jvm/src/aws/smithy/kotlin/runtime/http/engine/okhttp/OkHttpEngine.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,9 +15,9 @@ import aws.smithy.kotlin.runtime.net.TlsVersion
import aws.smithy.kotlin.runtime.operation.ExecutionContext
import aws.smithy.kotlin.runtime.time.Instant
import aws.smithy.kotlin.runtime.time.fromEpochMilliseconds
import kotlinx.coroutines.ExperimentalCoroutinesApi
import kotlinx.coroutines.job
import okhttp3.*
import okhttp3.ConnectionPool
import okhttp3.coroutines.executeAsync
import java.util.concurrent.TimeUnit
import kotlin.time.toJavaDuration
Expand All @@ -44,15 +44,15 @@ public class OkHttpEngine(
}

private val metrics = HttpClientMetrics(TELEMETRY_SCOPE, config.telemetryProvider)
private val client = config.buildClient(metrics)
private val connectionIdleMonitor = config.connectionIdlePollingInterval?.let { ConnectionIdleMonitor(it) }
private val client = config.buildClientWithConnectionListener(metrics, connectionIdleMonitor)

override suspend fun roundTrip(context: ExecutionContext, request: HttpRequest): HttpCall {
val callContext = callContext()

val engineRequest = request.toOkHttpRequest(context, callContext, metrics)
val engineCall = client.newCall(engineRequest)

@OptIn(ExperimentalCoroutinesApi::class)
val engineResponse = mapOkHttpExceptions { engineCall.executeAsync() }

val response = engineResponse.toSdkResponse()
Expand All @@ -71,17 +71,17 @@ public class OkHttpEngine(
}

override fun shutdown() {
connectionIdleMonitor?.close()
client.connectionPool.evictAll()
client.dispatcher.executorService.shutdown()
metrics.close()
}
}

/**
* Convert SDK version of HTTP configuration to OkHttp specific configuration and return the configured client
*/
@InternalApi
public fun OkHttpEngineConfig.buildClient(metrics: HttpClientMetrics): OkHttpClient {
private fun OkHttpEngineConfig.buildClientFromConfig(
metrics: HttpClientMetrics,
poolOverride: ConnectionPool? = null,
): OkHttpClient {
val config = this

return OkHttpClient.Builder().apply {
Expand All @@ -99,20 +99,11 @@ public fun OkHttpEngineConfig.buildClient(metrics: HttpClientMetrics): OkHttpCli
readTimeout(config.socketReadTimeout.toJavaDuration())
writeTimeout(config.socketWriteTimeout.toJavaDuration())

@OptIn(ExperimentalOkHttpApi::class)
val connectionListener = if (config.connectionIdlePollingInterval == null) {
ConnectionListener.NONE
} else {
ConnectionIdleMonitor(connectionIdlePollingInterval)
}

// use our own pool configured with the timeout settings taken from config
@OptIn(ExperimentalOkHttpApi::class)
val pool = ConnectionPool(
val pool = poolOverride ?: ConnectionPool(
maxIdleConnections = 5, // The default from the no-arg ConnectionPool() constructor
keepAliveDuration = config.connectionIdleTimeout.inWholeMilliseconds,
TimeUnit.MILLISECONDS,
connectionListener = connectionListener,
)
connectionPool(pool)

Expand Down Expand Up @@ -147,6 +138,34 @@ public fun OkHttpEngineConfig.buildClient(metrics: HttpClientMetrics): OkHttpCli
}.build()
}

/**
* Convert SDK version of HTTP configuration to OkHttp specific configuration and return the configured client
*/
// Used by OkHttp4Engine - OkHttp4 does NOT have `connectionListener`
// TODO - Refactor in next minor version - Move this to OkHttp4Engine and make it private
@InternalApi
public fun OkHttpEngineConfig.buildClient(
metrics: HttpClientMetrics,
): OkHttpClient = this.buildClientFromConfig(metrics)

/**
* Convert SDK version of HTTP configuration to OkHttp specific configuration and return the configured client
*/
// Used by OkHttpEngine - OkHttp5 does have `connectionListener`
@OptIn(ExperimentalOkHttpApi::class)
private fun OkHttpEngineConfig.buildClientWithConnectionListener(
metrics: HttpClientMetrics,
connectionListener: ConnectionIdleMonitor?,
): OkHttpClient = this.buildClientFromConfig(
metrics,
ConnectionPool(
maxIdleConnections = 5, // The default from the no-arg ConnectionPool() constructor
keepAliveDuration = this.connectionIdleTimeout.inWholeMilliseconds,
timeUnit = TimeUnit.MILLISECONDS,
connectionListener = connectionListener ?: ConnectionListener.NONE,
),
)

private fun minTlsConnectionSpec(tlsContext: TlsContext): ConnectionSpec {
val minVersion = tlsContext.minVersion ?: TlsVersion.TLS_1_2
val okHttpTlsVersions = SdkTlsVersion
Expand Down
6 changes: 6 additions & 0 deletions ...ent-engine-okhttp4/jvm/src/aws/smithy/kotlin/runtime/http/engine/okhttp4/OkHttp4Engine.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -70,6 +70,12 @@ public class OkHttp4Engine(
}
}
}

override fun shutdown() {
client.connectionPool.evictAll()
client.dispatcher.executorService.shutdown()
metrics.close()
}
Comment on lines +74 to +78
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nice catch

}

// Copied from okhttp3 5.x:
Expand Down
Loading