~elastic Dispatchers.IO

kotlinx-coroutines-core/concurrent/src/Dispatchers.kt

@@ -8,9 +8,26 @@ package kotlinx.coroutines
  * The [CoroutineDispatcher] that is designed for offloading blocking IO tasks to a shared pool of threads.
  * Additional threads in this pool are created on demand.
  *
- * By default, the pool is backed by `64` standalone threads, for the additional details
- * such as elasticity, dynamic sizing and interaction with [CoroutineDispatcher.limitedParallelism],
- * please refer to the documentation of `Dispatchers.IO` on specific platform.
+ * ### Elasticity for limited parallelism
+ *
+ * `Dispatchers.IO` has a unique property of elasticity: its views
+ * obtained with [CoroutineDispatcher.limitedParallelism] are
+ * not restricted by the `Dispatchers.IO` parallelism. Conceptually, there is
+ * a dispatcher backed by an unlimited pool of threads, and both `Dispatchers.IO`
+ * and views of `Dispatchers.IO` are actually views of that dispatcher. In practice
+ * this means that, despite not abiding by `Dispatchers.IO`'s parallelism
+ * restrictions, its views share threads and resources with it.
+ *
+ * In the following example
+ * ```
+ * // 100 threads for MySQL connection
+ * val myMysqlDbDispatcher = Dispatchers.IO.limitedParallelism(100)
+ * // 60 threads for MongoDB connection
+ * val myMongoDbDispatcher = Dispatchers.IO.limitedParallelism(60)
+ * ```
+ * the system may have up to `64 + 100 + 60` threads dedicated to blocking tasks during peak loads,
+ * but during its steady state there is only a small number of threads shared
+ * among `Dispatchers.IO`, `myMysqlDbDispatcher` and `myMongoDbDispatcher`
  */
 @Suppress("EXTENSION_SHADOWED_BY_MEMBER")
 public expect val Dispatchers.IO: CoroutineDispatcher

kotlinx-coroutines-core/native/src/Dispatchers.kt

@@ -4,6 +4,9 @@
 
 package kotlinx.coroutines
 
+import kotlinx.coroutines.internal.*
+import kotlin.coroutines.*
+
 
 public actual object Dispatchers {
  public actual val Default: CoroutineDispatcher = createDefaultDispatcher()
@@ -20,17 +23,33 @@ public actual object Dispatchers {
  injectedMainDispatcher = dispatcher
  }
 
- internal val IO: CoroutineDispatcher = newFixedThreadPoolContext(64, "Dispatchers.IO")
+ internal val IO: CoroutineDispatcher = DefaultIoScheduler
 }
 
-/**
- * The [CoroutineDispatcher] that is designed for offloading blocking IO tasks to a shared pool of threads.
- * Additional threads in this pool are created on demand.
- *
- * On Native platforms it is backed by a standalone [newFixedThreadPoolContext] with `64` worker threads in it.
- * **NB**: this dispatcher **does not** share the same elasticity behaviour for [CoroutineDispatcher.limitedParallelism]
- * as `Dispatchers.IO` on JVM.
- */
+internal object DefaultIoScheduler : CoroutineDispatcher() {
+ // 2048 is an arbitrary KMP-friendly constant
+ private val unlimitedPool = newFixedThreadPoolContext(2048, "Dispatchers.IO")
+ private val io = unlimitedPool.limitedParallelism(64) // Default JVM size
+
+ @ExperimentalCoroutinesApi
+ override fun limitedParallelism(parallelism: Int): CoroutineDispatcher {
+ // See documentation to Dispatchers.IO for the rationale
+ return unlimitedPool.limitedParallelism(parallelism)
+ }
+
+ override fun dispatch(context: CoroutineContext, block: Runnable) {
+ io.dispatch(context, block)
+ }
+
+ @InternalCoroutinesApi
+ override fun dispatchYield(context: CoroutineContext, block: Runnable) {
+ io.dispatchYield(context, block)
+ }
+
+ override fun toString(): String = "Dispatchers.IO"
+}
+
+
 @Suppress("EXTENSION_SHADOWED_BY_MEMBER")
 public actual val Dispatchers.IO: CoroutineDispatcher get() = IO