Do not use DefaultExecutor for cleanup work

Before this change, DefaultExecutor was occasionally used for
executing the work of dispatchers that no longer function.
This is no longer the case: instead, Dispatchers.IO is used for
that on our multithreaded targets.

Author: dkhalanskyjb

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

@@ -71,3 +71,27 @@ public expect object Dispatchers {
      */
     public val Unconfined: CoroutineDispatcher
 }
+
+/**
+ * If a task can no longer run because its dispatcher is closed, it is rescheduled to another dispatcher.
+ *
+ * This is required to avoid a situation where some finalizers do not run:
+ * ```
+ * val dispatcher = newSingleThreadContext("test")
+ * launch(dispatcher) {
+ *   val resource = Resource()
+ *   try {
+ *     // do something `suspending` with resource
+ *   } finally {
+ *     resource.close()
+ *   }
+ * }
+ * dispatcher.close()
+ * ```
+ *
+ * `close` needs to run somewhere, but it can't run on the closed dispatcher.
+ *
+ * On the JVM and Native, we reschedule to the thread pool backing `Dispatchers.IO`,
+ * because an arbitrary task may well have blocking behavior.
+ */
+internal expect fun rescheduleTaskFromClosedDispatcher(task: Runnable)

kotlinx-coroutines-core/common/src/EventLoop.common.kt

@@ -169,9 +169,6 @@ private typealias Queue<T> = LockFreeTaskQueueCore<T>
 internal expect abstract class EventLoopImplPlatform() : EventLoop {
     // Called to unpark this event loop's thread
     protected fun unpark()
-
-    // Called to reschedule to DefaultExecutor when this event loop is complete
-    protected fun reschedule(now: Long, delayedTask: EventLoopImplBase.DelayedTask)
 }
 
 internal abstract class EventLoopImplBase: EventLoopImplPlatform(), Delay {
@@ -275,7 +272,7 @@ internal abstract class EventLoopImplBase: EventLoopImplPlatform(), Delay {
             // todo: we should unpark only when this delayed task became first in the queue
             unpark()
         } else {
-            DefaultExecutor.enqueue(task)
+            rescheduleTaskFromClosedDispatcher(task)
         }
     }
 
@@ -408,6 +405,14 @@ internal abstract class EventLoopImplBase: EventLoopImplPlatform(), Delay {
         }
     }
 
+    // Called to reschedule when this event loop is complete
+    protected open fun reschedule(now: Long, delayedTask: DelayedTask) {
+        val delayTimeMillis = delayNanosToMillis(delayedTask.nanoTime - now)
+        DefaultDelay.invokeOnTimeout(delayTimeMillis, Runnable {
+            rescheduleTaskFromClosedDispatcher(delayedTask)
+        }, EmptyCoroutineContext)
+    }
+
     internal abstract class DelayedTask(
         /**
          * This field can be only modified in [scheduleTask] before putting this DelayedTask
@@ -530,10 +535,6 @@ internal expect fun createEventLoop(): EventLoop
 
 internal expect fun nanoTime(): Long
 
-internal expect object DefaultExecutor {
-    fun enqueue(task: Runnable)
-}
-
 /**
  * Used by Darwin targets to wrap a [Runnable.run] call in an Objective-C Autorelease Pool. It is a no-op on JVM, JS and
  * non-Darwin native targets.

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

@@ -20,5 +20,8 @@ import kotlin.coroutines.*
  *
  * Here, instead of releasing the thread on which `loadConfiguration` runs if `fetchConfigurationData` suspends, it will
  * block, potentially leading to thread starvation issues.
+ *
+ * If new tasks are submitted to the dispatcher created by [runBlocking] after this function returns,
+ * they are resubmitted to [Dispatchers.IO].
  */
 public expect fun <T> runBlocking(context: CoroutineContext = EmptyCoroutineContext, block: suspend CoroutineScope.() -> T): T

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

@@ -39,4 +39,26 @@ package kotlinx.coroutines
 @Suppress("EXTENSION_SHADOWED_BY_MEMBER")
 public expect val Dispatchers.IO: CoroutineDispatcher
 
-
+internal actual fun rescheduleTaskFromClosedDispatcher(task: Runnable) {
+    /**
+     * We do not create a separate view of [Dispatchers.IO] for the cleanup needs.
+     *
+     * If [Dispatchers.IO] is not flooded with other tasks + the cleanup view does not have more threads than
+     * [Dispatchers.IO], there is no difference between sending cleanup tasks to [Dispatchers.IO] and creating
+     * a separate view of [Dispatchers.IO] for cleanup.
+     *
+     * If [Dispatchers.IO] is flooded with other tasks, we are at a crossroads:
+     * - On the one hand, we do not want to create too many threads.
+     *   Some clients are carefully monitoring the number of threads and are relying on it not being larger than
+     *   the system property + the sum of explicit `limitedParallelism` arguments in the system.
+     * - On the other hand, we don't want to delay productive tasks in favor of cleanup tasks.
+     *
+     * The first consideration wins on two accounts:
+     * - As of writing this, [Dispatchers.IO] has been in use as the cleanup dispatcher for dispatchers obtained
+     *   from JVM executors for years, and this has not caused any issues that we know of.
+     * - On the other hand, some people likely rely on the number of threads not exceeding the number they control.
+     *   If we were to create a separate view of [Dispatchers.IO] for cleanup, this number would be exceeded, which
+     *   is a regression.
+     */
+    Dispatchers.IO.dispatch(Dispatchers.IO, task)
+}

kotlinx-coroutines-core/jsAndWasmJsShared/src/EventLoop.kt

@@ -12,14 +12,13 @@ internal class UnconfinedEventLoop : EventLoop() {
 
 internal actual abstract class EventLoopImplPlatform : EventLoop() {
     protected actual fun unpark(): Unit = unsupported()
-    protected actual fun reschedule(now: Long, delayedTask: EventLoopImplBase.DelayedTask): Unit = unsupported()
-}
-
-internal actual object DefaultExecutor {
-    public actual fun enqueue(task: Runnable): Unit = unsupported()
 }
 
 private fun unsupported(): Nothing =
     throw UnsupportedOperationException("runBlocking event loop is not supported")
 
 internal actual inline fun platformAutoreleasePool(crossinline block: () -> Unit) = block()
+
+internal actual fun rescheduleTaskFromClosedDispatcher(task: Runnable) {
+    Dispatchers.Default.dispatch(Dispatchers.Default, task)
+}

kotlinx-coroutines-core/jvm/src/Builders.kt

@@ -4,7 +4,6 @@
 
 package kotlinx.coroutines
 
-import java.util.concurrent.locks.*
 import kotlin.contracts.*
 import kotlin.coroutines.*
 
@@ -38,6 +37,9 @@ import kotlin.coroutines.*
  * If this blocked thread is interrupted (see [Thread.interrupt]), then the coroutine job is cancelled and
  * this `runBlocking` invocation throws [InterruptedException].
  *
+ * If new tasks are submitted to the dispatcher created by [runBlocking] after this function returns,
+ * they are resubmitted to [Dispatchers.IO].
+ *
  * See [newCoroutineContext][CoroutineScope.newCoroutineContext] for a description of debugging facilities that are available
  * for a newly created coroutine.
  *

kotlinx-coroutines-core/jvm/src/DefaultExecutor.kt

@@ -11,18 +11,27 @@ internal actual val DefaultDelay: Delay = initializeDefaultDelay()
 
 private fun initializeDefaultDelay(): Delay {
     // Opt-out flag
-    if (!defaultMainDelayOptIn) return DefaultExecutor
+    if (!defaultMainDelayOptIn) return DefaultDelayImpl
     val main = Dispatchers.Main
     /*
      * When we already are working with UI and Main threads, it makes
      * no sense to create a separate thread with timer that cannot be controller
      * by the UI runtime.
      */
-    return if (main.isMissing() || main !is Delay) DefaultExecutor else main
+    return if (main.isMissing() || main !is Delay) DefaultDelayImpl else main
 }
 
-@Suppress("PLATFORM_CLASS_MAPPED_TO_KOTLIN")
-internal actual object DefaultExecutor : EventLoopImplBase(), Runnable {
+internal object DefaultExecutor {
+    fun shutdown() = DefaultDelayImpl.shutdown()
+
+    fun ensureStarted() = DefaultDelayImpl.ensureStarted()
+
+    fun shutdownForTests(timeout: Long) = DefaultDelayImpl.shutdownForTests(timeout)
+
+    val isThreadPresent: Boolean get() = DefaultDelayImpl.isThreadPresent
+}
+
+private object DefaultDelayImpl : EventLoopImplBase(), Runnable {
     const val THREAD_NAME = "kotlinx.coroutines.DefaultExecutor"
 
     init {
@@ -61,7 +70,7 @@ internal actual object DefaultExecutor : EventLoopImplBase(), Runnable {
         return debugStatus == SHUTDOWN_REQ || debugStatus == SHUTDOWN_ACK
     }
 
-    actual override fun enqueue(task: Runnable) {
+    override fun enqueue(task: Runnable) {
         if (isShutDown) shutdownError()
         super.enqueue(task)
     }
@@ -137,7 +146,7 @@ internal actual object DefaultExecutor : EventLoopImplBase(), Runnable {
              * the singleton itself instead of using parent' thread one
              * in order not to accidentally capture temporary application classloader.
              */
-            contextClassLoader = this@DefaultExecutor.javaClass.classLoader
+            contextClassLoader = this@DefaultDelayImpl.javaClass.classLoader
             isDaemon = true
             start()
         }

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

@@ -2,7 +2,6 @@ package kotlinx.coroutines
 
 import kotlinx.coroutines.internal.*
 import kotlinx.coroutines.scheduling.*
-import kotlin.coroutines.*
 
 /**
  * Name of the property that defines the maximal number of threads that are used by [Dispatchers.IO] coroutines dispatcher.

kotlinx-coroutines-core/jvm/src/EventLoop.kt

@@ -3,6 +3,7 @@ package kotlinx.coroutines
 import kotlinx.coroutines.Runnable
 import kotlinx.coroutines.scheduling.*
 import kotlinx.coroutines.scheduling.CoroutineScheduler
+import kotlin.coroutines.EmptyCoroutineContext
 
 internal actual abstract class EventLoopImplPlatform: EventLoop() {
 
@@ -14,9 +15,6 @@ internal actual abstract class EventLoopImplPlatform: EventLoop() {
             unpark(thread)
     }
 
-    protected actual open fun reschedule(now: Long, delayedTask: EventLoopImplBase.DelayedTask) {
-        DefaultExecutor.schedule(now, delayedTask)
-    }
 }
 
 internal class BlockingEventLoop(
@@ -122,4 +120,3 @@ internal fun Thread.isIoDispatcherThread(): Boolean {
     if (this !is CoroutineScheduler.Worker) return false
     return isIo()
 }
-

kotlinx-coroutines-core/jvm/src/Executors.kt

@@ -131,7 +131,7 @@ internal class ExecutorCoroutineDispatcherImpl(override val executor: Executor)
         } catch (e: RejectedExecutionException) {
             unTrackTask()
             cancelJobOnRejection(context, e)
-            Dispatchers.IO.dispatch(context, block)
+            rescheduleTaskFromClosedDispatcher(block)
         }
     }
 
@@ -146,15 +146,15 @@ internal class ExecutorCoroutineDispatcherImpl(override val executor: Executor)
             continuation.invokeOnCancellation(CancelFutureOnCancel(future))
             return
         }
-        // Otherwise fallback to default executor
-        DefaultExecutor.scheduleResumeAfterDelay(timeMillis, continuation)
+        // Otherwise fallback to default delay
+        DefaultDelay.scheduleResumeAfterDelay(timeMillis, continuation)
     }
 
     override fun invokeOnTimeout(timeMillis: Long, block: Runnable, context: CoroutineContext): DisposableHandle {
         val future = (executor as? ScheduledExecutorService)?.scheduleBlock(block, context, timeMillis)
         return when {
             future != null -> DisposableFutureHandle(future)
-            else -> DefaultExecutor.invokeOnTimeout(timeMillis, block, context)
+            else -> DefaultDelay.invokeOnTimeout(timeMillis, block, context)
         }
     }
 
@@ -189,6 +189,14 @@ private class ResumeUndispatchedRunnable(
     }
 }
 
+private class ResumeDispatchedRunnable(
+    private val continuation: CancellableContinuation<Unit>
+) : Runnable {
+    override fun run() {
+        continuation.resume(Unit)
+    }
+}
+
 /**
  * An implementation of [DisposableHandle] that cancels the specified future on dispose.
  * @suppress **This is unstable API and it is subject to change.**