KTOR-6292 Make client use Dispatchers.IO by default (#3748)

* KTOR-6292 Make client use Dispatchers.IO by default

Author: e5l

ktor-client/ktor-client-android/api/ktor-client-android.api

@@ -8,7 +8,6 @@ public final class io/ktor/client/engine/android/AndroidClientEngine : io/ktor/c
 	public fun execute (Lio/ktor/client/request/HttpRequestData;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public synthetic fun getConfig ()Lio/ktor/client/engine/HttpClientEngineConfig;
 	public fun getConfig ()Lio/ktor/client/engine/android/AndroidEngineConfig;
-	public fun getDispatcher ()Lkotlinx/coroutines/CoroutineDispatcher;
 	public fun getSupportedCapabilities ()Ljava/util/Set;
 }
 

ktor-client/ktor-client-android/jvm/src/io/ktor/client/engine/android/AndroidClientEngine.kt

@@ -9,7 +9,6 @@ import io.ktor.client.engine.*
 import io.ktor.client.plugins.*
 import io.ktor.client.plugins.sse.*
 import io.ktor.client.request.*
-import io.ktor.client.utils.*
 import io.ktor.http.*
 import io.ktor.http.content.*
 import io.ktor.util.*
@@ -31,13 +30,6 @@ private val METHODS_WITHOUT_BODY = listOf(HttpMethod.Get, HttpMethod.Head)
 @OptIn(InternalAPI::class)
 public class AndroidClientEngine(override val config: AndroidEngineConfig) : HttpClientEngineBase("ktor-android") {
 
-    override val dispatcher: CoroutineDispatcher by lazy {
-        Dispatchers.clientDispatcher(
-            config.threadsCount,
-            "ktor-android-dispatcher"
-        )
-    }
-
     override val supportedCapabilities: Set<HttpClientEngineCapability<*>> = setOf(HttpTimeout, SSECapability)
 
     override suspend fun execute(data: HttpRequestData): HttpResponseData {

ktor-client/ktor-client-apache/jvm/src/io/ktor/client/engine/apache/ApacheEngine.kt

@@ -8,7 +8,6 @@ import io.ktor.client.engine.*
 import io.ktor.client.plugins.*
 import io.ktor.client.plugins.sse.*
 import io.ktor.client.request.*
-import io.ktor.client.utils.*
 import io.ktor.utils.io.*
 import kotlinx.coroutines.*
 import org.apache.http.*
@@ -22,13 +21,6 @@ private const val IO_THREAD_COUNT_DEFAULT = 4
 @OptIn(InternalAPI::class)
 internal class ApacheEngine(override val config: ApacheEngineConfig) : HttpClientEngineBase("ktor-apache") {
 
-    override val dispatcher by lazy {
-        Dispatchers.clientDispatcher(
-            config.threadsCount,
-            "ktor-apache-dispatcher"
-        )
-    }
-
     override val supportedCapabilities = setOf(HttpTimeout, SSECapability)
 
     private val engine: CloseableHttpAsyncClient = prepareClient().apply { start() }

ktor-client/ktor-client-apache5/jvm/src/io/ktor/client/engine/apache5/Apache5Engine.kt

@@ -8,7 +8,6 @@ import io.ktor.client.engine.*
 import io.ktor.client.plugins.*
 import io.ktor.client.plugins.sse.*
 import io.ktor.client.request.*
-import io.ktor.client.utils.*
 import io.ktor.utils.io.*
 import kotlinx.coroutines.*
 import org.apache.hc.client5.http.config.*
@@ -30,13 +29,6 @@ private const val IO_THREAD_COUNT_DEFAULT = 4
 @OptIn(InternalAPI::class)
 internal class Apache5Engine(override val config: Apache5EngineConfig) : HttpClientEngineBase("ktor-apache") {
 
-    override val dispatcher by lazy {
-        Dispatchers.clientDispatcher(
-            config.threadsCount,
-            "ktor-apache-dispatcher"
-        )
-    }
-
     override val supportedCapabilities = setOf(HttpTimeout, SSECapability)
 
     @Volatile

ktor-client/ktor-client-cio/jvmAndNix/src/io/ktor/client/engine/cio/CIOEngine.kt

@@ -9,7 +9,6 @@ import io.ktor.client.plugins.*
 import io.ktor.client.plugins.sse.*
 import io.ktor.client.plugins.websocket.*
 import io.ktor.client.request.*
-import io.ktor.client.utils.*
 import io.ktor.http.*
 import io.ktor.network.selector.*
 import io.ktor.util.*
@@ -25,9 +24,6 @@ internal class CIOEngine(
     override val config: CIOEngineConfig
 ) : HttpClientEngineBase("ktor-cio") {
 
-    override val dispatcher: CoroutineDispatcher =
-        Dispatchers.clientDispatcher(config.threadsCount, "ktor-cio-dispatcher")
-
     override val supportedCapabilities =
         setOf(HttpTimeout, WebSocketCapability, WebSocketExtensionsCapability, SSECapability)
 

ktor-client/ktor-client-core/api/ktor-client-core.api

@@ -150,6 +150,7 @@ public abstract class io/ktor/client/engine/HttpClientEngineBase : io/ktor/clien
 	public fun <init> (Ljava/lang/String;)V
 	public fun close ()V
 	public fun getCoroutineContext ()Lkotlin/coroutines/CoroutineContext;
+	public fun getDispatcher ()Lkotlinx/coroutines/CoroutineDispatcher;
 	public fun getSupportedCapabilities ()Ljava/util/Set;
 	public fun install (Lio/ktor/client/HttpClient;)V
 }
@@ -188,7 +189,6 @@ public abstract class io/ktor/client/engine/HttpClientJvmEngine : io/ktor/client
 	public fun close ()V
 	protected final fun createCallContext (Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public fun getCoroutineContext ()Lkotlin/coroutines/CoroutineContext;
-	public fun getDispatcher ()Lkotlinx/coroutines/CoroutineDispatcher;
 	public fun getSupportedCapabilities ()Ljava/util/Set;
 	public fun install (Lio/ktor/client/HttpClient;)V
 }

ktor-client/ktor-client-core/common/src/io/ktor/client/engine/HttpClientEngineBase.kt

@@ -18,6 +18,8 @@ import kotlin.coroutines.*
 public abstract class HttpClientEngineBase(private val engineName: String) : HttpClientEngine {
     private val closed = atomic(false)
 
+    override val dispatcher: CoroutineDispatcher = ioDispatcher()
+
     override val coroutineContext: CoroutineContext by lazy {
         SilentSupervisor() + dispatcher + CoroutineName("$engineName-context")
     }
@@ -28,9 +30,6 @@ public abstract class HttpClientEngineBase(private val engineName: String) : Htt
         val requestJob = coroutineContext[Job] as? CompletableJob ?: return
 
         requestJob.complete()
-        requestJob.invokeOnCompletion {
-            dispatcher.close()
-        }
     }
 }
 
@@ -54,3 +53,5 @@ private fun CoroutineDispatcher.close() {
         // Some closeable dispatchers like Dispatchers.IO can't be closed.
     }
 }
+
+internal expect fun ioDispatcher(): CoroutineDispatcher

ktor-client/ktor-client-core/common/src/io/ktor/client/engine/HttpClientEngineConfig.kt

@@ -14,6 +14,10 @@ public open class HttpClientEngineConfig {
     /**
      * Specifies network threads count advice.
      */
+    @Deprecated(
+        "The [threadsCount] property is deprecated. The [Dispatchers.IO] is used by default.",
+        level = DeprecationLevel.ERROR
+    )
     public var threadsCount: Int = 4
 
     /**

ktor-client/ktor-client-core/js/src/io/ktor/client/engine/HttpClientEngineBase.js.kt

@@ -0,0 +1,9 @@
+/*
+ * Copyright 2014-2023 JetBrains s.r.o and contributors. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package io.ktor.client.engine
+
+import kotlinx.coroutines.*
+
+internal actual fun ioDispatcher(): CoroutineDispatcher = Dispatchers.Default

ktor-client/ktor-client-core/js/src/io/ktor/client/engine/js/JsClientEngine.kt

@@ -25,8 +25,6 @@ internal class JsClientEngine(
     override val config: HttpClientEngineConfig
 ) : HttpClientEngineBase("ktor-js") {
 
-    override val dispatcher = Dispatchers.Default
-
     override val supportedCapabilities = setOf(HttpTimeout, WebSocketCapability, SSECapability)
 
     init {

ktor-client/ktor-client-core/jvm/src/io/ktor/client/engine/HttpClientEngineBase.jvm.kt

@@ -0,0 +1,9 @@
+/*
+ * Copyright 2014-2023 JetBrains s.r.o and contributors. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package io.ktor.client.engine
+
+import kotlinx.coroutines.*
+
+internal actual fun ioDispatcher(): CoroutineDispatcher = Dispatchers.IO

ktor-client/ktor-client-core/jvm/src/io/ktor/client/engine/HttpClientJvmEngine.kt

@@ -6,7 +6,6 @@ package io.ktor.client.engine
 
 import io.ktor.util.*
 import kotlinx.coroutines.*
-import java.util.concurrent.*
 import kotlin.coroutines.*
 
 /**
@@ -20,21 +19,9 @@ import kotlin.coroutines.*
 )
 public abstract class HttpClientJvmEngine(engineName: String) : HttpClientEngine {
     private val clientContext = SilentSupervisor()
-    private val _dispatcher by lazy {
-        Executors.newFixedThreadPool(config.threadsCount) {
-            Thread(it).apply {
-                isDaemon = true
-            }
-        }.asCoroutineDispatcher()
-    }
-
-    @OptIn(InternalCoroutinesApi::class)
-    override val dispatcher: CoroutineDispatcher
-        get() = _dispatcher
 
-    @OptIn(InternalCoroutinesApi::class)
     override val coroutineContext: CoroutineContext by lazy {
-        _dispatcher + clientContext + CoroutineName("$engineName-context")
+        dispatcher + clientContext + CoroutineName("$engineName-context")
     }
 
     /**
@@ -60,12 +47,8 @@ public abstract class HttpClientJvmEngine(engineName: String) : HttpClientEngine
     }
 
     override fun close() {
-        val job = clientContext[Job] as CompletableJob
-
+        val job = clientContext.job as CompletableJob
         job.complete()
-        job.invokeOnCompletion {
-            _dispatcher.close()
-        }
     }
 }
 

ktor-client/ktor-client-core/posix/src/io/ktor/client/engine/HttpClientEngineBase.posix.kt

@@ -0,0 +1,9 @@
+/*
+ * Copyright 2014-2023 JetBrains s.r.o and contributors. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package io.ktor.client.engine
+
+import kotlinx.coroutines.*
+
+internal actual fun ioDispatcher(): CoroutineDispatcher = Dispatchers.IO

ktor-client/ktor-client-java/api/ktor-client-java.api

@@ -15,7 +15,6 @@ public final class io/ktor/client/engine/java/JavaHttpEngine : io/ktor/client/en
 	public fun execute (Lio/ktor/client/request/HttpRequestData;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public synthetic fun getConfig ()Lio/ktor/client/engine/HttpClientEngineConfig;
 	public fun getConfig ()Lio/ktor/client/engine/java/JavaHttpConfig;
-	public fun getDispatcher ()Lkotlinx/coroutines/CoroutineDispatcher;
 	public fun getSupportedCapabilities ()Ljava/util/Set;
 }
 

ktor-client/ktor-client-java/jvm/src/io/ktor/client/engine/java/JavaHttpEngine.kt

@@ -10,37 +10,16 @@ import io.ktor.client.plugins.sse.*
 import io.ktor.client.plugins.websocket.*
 import io.ktor.client.request.*
 import io.ktor.utils.io.*
-import kotlinx.atomicfu.*
 import kotlinx.coroutines.*
 import java.net.*
 import java.net.http.*
 import java.time.*
 import java.time.temporal.*
-import java.util.concurrent.*
 
 public class JavaHttpEngine(override val config: JavaHttpConfig) : HttpClientEngineBase("ktor-java") {
 
-    private val executorThreadCounter = atomic(0L)
-
     private val protocolVersion = config.protocolVersion
 
-    /**
-     * Exposed for tests only.
-     */
-    internal val executor by lazy {
-        Executors.newFixedThreadPool(config.threadsCount) {
-            val number = executorThreadCounter.getAndIncrement()
-            Thread(it, "ktor-client-java-$number").apply {
-                isDaemon = true
-                setUncaughtExceptionHandler { _, _ -> }
-            }
-        }
-    }
-
-    public override val dispatcher: CoroutineDispatcher by lazy {
-        executor.asCoroutineDispatcher()
-    }
-
     public override val supportedCapabilities: Set<HttpClientEngineCapability<*>> =
         setOf(HttpTimeout, WebSocketCapability, SSECapability)
 
@@ -75,7 +54,7 @@ public class JavaHttpEngine(override val config: JavaHttpConfig) : HttpClientEng
         return httpClient ?: synchronized(this) {
             httpClient ?: HttpClient.newBuilder().apply {
                 version(protocolVersion)
-                executor(executor)
+                executor(dispatcher.asExecutor())
 
                 apply(config.config)
 
@@ -106,6 +85,7 @@ public class JavaHttpEngine(override val config: JavaHttpConfig) : HttpClientEng
 
                 proxy(ProxySelector.of(address))
             }
+
             Proxy.Type.DIRECT -> proxy(HttpClient.Builder.NO_PROXY)
             else -> error("Java HTTP engine does not currently support $type proxies.")
         }

ktor-client/ktor-client-java/jvm/test/io/ktor/client/engine/java/JavaEngineTests.kt

@@ -13,19 +13,9 @@ import io.ktor.http.*
 import kotlinx.coroutines.*
 import java.net.*
 import java.time.*
-import java.util.concurrent.*
 import kotlin.test.*
 
 class JavaEngineTests {
-    @Test
-    fun testClose() {
-        val engine = JavaHttpEngine(JavaHttpConfig())
-        engine.close()
-
-        assertTrue("Java HTTP dispatcher is not working.") {
-            engine.executor.isShutdown
-        }
-    }
 
     @Test
     fun testProxy() = runBlocking {
@@ -41,49 +31,6 @@ class JavaEngineTests {
         assertEquals("proxy", body)
     }
 
-    @Test
-    fun testThreadLeak() = runBlocking {
-        System.setProperty("jdk.internal.httpclient.selectorTimeout", "50")
-
-        val initialNumberOfThreads = Thread.getAllStackTraces().size
-        val repeats = 25
-        val executors = ArrayList<ExecutorService>()
-
-        try {
-            repeat(repeats) {
-                HttpClient(Java).use { client ->
-                    val response = client.get("http://www.google.com").body<String>()
-                    assertNotNull(response)
-                    executors += (client.engine as JavaHttpEngine).executor
-                }
-            }
-
-            // When engine is disposed HttpClient's SelectorManager thread remains active
-            // until it realizes that no more reference on HttpClient.
-            // Minimum polling interval SelectorManager thread is 1000ms.
-            var retry = 0
-            do {
-                System.gc()
-                Thread.sleep(1000)
-                System.gc()
-                System.gc()
-            } while (Thread.getAllStackTraces().size >= initialNumberOfThreads && retry++ < 10)
-        } finally {
-            System.clearProperty("jdk.internal.httpclient.selectorTimeout")
-        }
-
-        val totalNumberOfThreads = Thread.getAllStackTraces().size
-        val threadsCreated = totalNumberOfThreads - initialNumberOfThreads
-
-        executors.forEach { pool ->
-            assertTrue(pool.isTerminated)
-        }
-
-        assertTrue("Number of threads should be less $repeats, but was $threadsCreated") {
-            threadsCreated < repeats
-        }
-    }
-
     @Test
     fun testRequestAfterRecreate() {
         runBlocking {

ktor-client/ktor-client-jetty-jakarta/jvm/src/io/ktor/client/engine/jetty/jakarta/JettyHttp2Engine.kt

@@ -7,25 +7,16 @@ package io.ktor.client.engine.jetty.jakarta
 import io.ktor.client.engine.*
 import io.ktor.client.plugins.*
 import io.ktor.client.request.*
-import io.ktor.client.utils.*
 import io.ktor.util.*
 import io.ktor.utils.io.*
 import kotlinx.coroutines.*
 import org.eclipse.jetty.http2.client.*
-import org.eclipse.jetty.util.thread.*
 
 @OptIn(InternalAPI::class)
 internal class JettyHttp2Engine(
     override val config: JettyEngineConfig
 ) : HttpClientEngineBase("ktor-jetty") {
 
-    override val dispatcher: CoroutineDispatcher by lazy {
-        Dispatchers.clientDispatcher(
-            config.threadsCount,
-            "ktor-jetty-dispatcher"
-        )
-    }
-
     override val supportedCapabilities = setOf(HttpTimeout)
 
     /**
@@ -59,9 +50,7 @@ internal class JettyHttp2Engine(
             addBean(config.sslContextFactory)
             check(config.proxy == null) { "Proxy unsupported in Jetty engine." }
 
-            executor = QueuedThreadPool().apply {
-                name = "ktor-jetty-client-qtp"
-            }
+            executor = Dispatchers.IO.asExecutor()
 
             setupTimeoutAttributes(timeoutExtension)