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

@@ -5,7 +5,6 @@
 
 package kotlinx.coroutines
 
-import java.util.concurrent.locks.*
 import kotlin.contracts.*
 import kotlin.coroutines.*
 
@@ -39,6 +38,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.**

kotlinx-coroutines-core/jvm/test/DefaultExecutorStressTest.kt

@@ -8,7 +8,7 @@ class DefaultExecutorStressTest : TestBase() {
     @Test
     fun testDelay() = runTest {
         val iterations = 100_000 * stressTestMultiplier
-        withContext(DefaultExecutor) {
+        withContext(DefaultDelay as CoroutineDispatcher) {
             expect(1)
             var expected = 1
             repeat(iterations) {

kotlinx-coroutines-core/jvm/test/EventLoopsTest.kt

@@ -49,25 +49,6 @@ class EventLoopsTest : TestBase() {
         finish(5)
     }
 
-    @Test
-    fun testEventLoopInDefaultExecutor() = runTest {
-        expect(1)
-        withContext(Dispatchers.Unconfined) {
-            delay(1)
-            assertTrue(Thread.currentThread().name.startsWith(DefaultExecutor.THREAD_NAME))
-            expect(2)
-            // now runBlocking inside default executor thread --> should use outer event loop
-            DefaultExecutor.enqueue(Runnable {
-                expect(4) // will execute when runBlocking runs loop
-            })
-            expect(3)
-            runBlocking {
-                expect(5)
-            }
-        }
-        finish(6)
-    }
-
     /**
      * Simple test for [processNextEventInCurrentThread] API use-case.
      */
@@ -159,4 +140,4 @@ class EventLoopsTest : TestBase() {
             waitingThread.value = null
         }
     }
-}
+}

kotlinx-coroutines-core/jvm/test/RunBlockingJvmTest.kt

@@ -13,4 +13,3 @@ class RunBlockingJvmTest : TestBase() {
         rb.hashCode() // unused
     }
 }
-

kotlinx-coroutines-core/jvm/test/knit/TestUtil.kt

@@ -35,9 +35,8 @@ fun <R> test(name: String, block: () -> R): List<String> = outputException(name)
             } finally {
                 // the shutdown
                 log.println("--- shutting down")
-                DefaultScheduler.shutdown(SHUTDOWN_TIMEOUT)
                 shutdownDispatcherPools(SHUTDOWN_TIMEOUT)
-                DefaultExecutor.shutdownForTests(SHUTDOWN_TIMEOUT) // the last man standing -- cleanup all pending tasks
+                DefaultScheduler.shutdown(SHUTDOWN_TIMEOUT) // the last man standing -- cleanup all pending tasks
             }
             checkTestThreads(threadsBefore) // check thread if the main completed successfully
         }
@@ -55,7 +54,7 @@ private fun shutdownDispatcherPools(timeout: Long) {
             (thread.dispatcher.executor as ExecutorService).apply {
                 shutdown()
                 awaitTermination(timeout, TimeUnit.MILLISECONDS)
-                shutdownNow().forEach { DefaultExecutor.enqueue(it) }
+                shutdownNow().forEach { rescheduleTaskFromClosedDispatcher(it) }
             }
     }
 }

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

@@ -35,8 +35,8 @@ import kotlin.native.concurrent.*
  * the specified dispatcher while the current thread is blocked. If the specified dispatcher is an event loop of another `runBlocking`,
  * then this invocation uses the outer event loop.
  *
- * 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/native/src/CoroutineContext.kt

@@ -3,32 +3,11 @@ package kotlinx.coroutines
 import kotlinx.coroutines.internal.*
 import kotlin.coroutines.*
 
-internal actual object DefaultExecutor : CoroutineDispatcher(), Delay {
-
-    private val delegate = WorkerDispatcher(name = "DefaultExecutor")
-
-    override fun dispatch(context: CoroutineContext, block: Runnable) {
-        delegate.dispatch(context, block)
-    }
-
-    override fun scheduleResumeAfterDelay(timeMillis: Long, continuation: CancellableContinuation<Unit>) {
-        delegate.scheduleResumeAfterDelay(timeMillis, continuation)
-    }
-
-    override fun invokeOnTimeout(timeMillis: Long, block: Runnable, context: CoroutineContext): DisposableHandle {
-        return delegate.invokeOnTimeout(timeMillis, block, context)
-    }
-
-    actual fun enqueue(task: Runnable): Unit {
-        delegate.dispatch(EmptyCoroutineContext, task)
-    }
-}
+@PublishedApi
+internal actual val DefaultDelay: Delay = WorkerDispatcher(name = "DefaultDelay")
 
 internal expect fun createDefaultDispatcher(): CoroutineDispatcher
 
-@PublishedApi
-internal actual val DefaultDelay: Delay = DefaultExecutor
-
 public actual fun CoroutineScope.newCoroutineContext(context: CoroutineContext): CoroutineContext {
     val combined = coroutineContext + context
     return if (combined !== Dispatchers.Default && combined[ContinuationInterceptor] == null)

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

@@ -1,6 +1,5 @@
 package kotlinx.coroutines
 
-import kotlinx.coroutines.internal.*
 import kotlin.coroutines.*