WIP

Author: dkhalanskyjb

Diff for: docs/topics/coroutine-context-and-dispatchers.md

@@ -115,7 +115,7 @@ Produces the output:
 ```text
 Unconfined      : I'm working in thread main
 main runBlocking: I'm working in thread main
-Unconfined      : After delay in thread kotlinx.coroutines.DefaultDelay
+Unconfined      : After delay in thread DefaultDispatcher oroutine#2
 main runBlocking: After delay in thread main
 ```
 

Diff for: kotlinx-coroutines-core/common/src/EventLoop.common.kt

@@ -6,6 +6,48 @@ import kotlin.concurrent.Volatile
 import kotlin.coroutines.*
 import kotlin.jvm.*
 
+internal interface UnconfinedEventLoop {
+    /**
+     * Returns `true` if calling [yield] in a coroutine in this event loop can avoid yielding and continue executing
+     * due to there being no other tasks in the queue.
+     *
+     * This can only be called from the thread that owns this event loop.
+     */
+    val thisLoopsTaskCanAvoidYielding: Boolean
+
+    /**
+     * Returns `true` if someone (typically a call to [runUnconfinedEventLoop]) is currently processing the tasks,
+     * so calling [dispatchUnconfined] is guaranteed to be processed eventually.
+     *
+     * This can only be called from the thread that owns this event loop.
+     */
+    val isUnconfinedLoopActive: Boolean
+
+    /**
+     * Executes [initialBlock] and then processes unconfined tasks until there are no more, blocking the current thread.
+     *
+     * This can only be called when no other [runUnconfinedEventLoop] is currently active on this event loop.
+     *
+     * This can only be called from the thread that owns this event loop.
+     */
+    fun runUnconfinedEventLoop(initialBlock: () -> Unit)
+
+    /**
+     * Sends the [task] to this event loop for execution.
+     *
+     * This method should only be called while [isUnconfinedLoopActive] is `true`.
+     * Otherwise, the task may be left unprocessed.
+     *
+     * This can only be called from the thread that owns this event loop.
+     */
+    fun dispatchUnconfined(task: DispatchedTask<*>)
+
+    /**
+     * Tries to interpret this event loop for unconfined tasks as a proper event loop and returns it if successful.
+     */
+    fun tryUseAsEventLoop(): EventLoop?
+}
+
 /**
  * Extended by [CoroutineDispatcher] implementations that have event loop inside and can
  * be asked to process next event from their event queue.
@@ -16,7 +58,7 @@ import kotlin.jvm.*
  *
  * @suppress **This an internal API and should not be used from general code.**
  */
-internal abstract class EventLoop : CoroutineDispatcher() {
+internal abstract class EventLoop : CoroutineDispatcher(), UnconfinedEventLoop {
     /**
      * Counts the number of nested `runBlocking` and [Dispatchers.Unconfined] that use this event loop.
      */
@@ -51,8 +93,6 @@ internal abstract class EventLoop : CoroutineDispatcher() {
         return 0
     }
 
-    protected open val isEmpty: Boolean get() = isUnconfinedQueueEmpty
-
     protected open val nextTime: Long
         get() {
             val queue = unconfinedQueue ?: return Long.MAX_VALUE
@@ -65,6 +105,7 @@ internal abstract class EventLoop : CoroutineDispatcher() {
         task.run()
         return true
     }
+
     /**
      * Returns `true` if the invoking `runBlocking(context) { ... }` that was passed this event loop in its context
      * parameter should call [processNextEvent] for this event loop (otherwise, it will process thread-local one).
@@ -77,28 +118,26 @@ internal abstract class EventLoop : CoroutineDispatcher() {
      * Dispatches task whose dispatcher returned `false` from [CoroutineDispatcher.isDispatchNeeded]
      * into the current event loop.
      */
-    fun dispatchUnconfined(task: DispatchedTask<*>) {
-        val queue = unconfinedQueue ?:
-            ArrayDeque<DispatchedTask<*>>().also { unconfinedQueue = it }
+    override fun dispatchUnconfined(task: DispatchedTask<*>) {
+        val queue = unconfinedQueue ?: ArrayDeque<DispatchedTask<*>>().also { unconfinedQueue = it }
         queue.addLast(task)
     }
 
     val isActive: Boolean
         get() = useCount > 0
 
-    val isUnconfinedLoopActive: Boolean
+    override val isUnconfinedLoopActive: Boolean
         get() = useCount >= delta(unconfined = true)
 
-    // May only be used from the event loop's thread
-    val isUnconfinedQueueEmpty: Boolean
-        get() = unconfinedQueue?.isEmpty() ?: true
+    override val thisLoopsTaskCanAvoidYielding: Boolean
+        get() = unconfinedQueue?.isEmpty() != false
 
     private fun delta(unconfined: Boolean) =
         if (unconfined) (1L shl 32) else 1L
 
     fun incrementUseCount(unconfined: Boolean = false) {
         useCount += delta(unconfined)
-        if (!unconfined) shared = true 
+        if (!unconfined) shared = true
     }
 
     fun decrementUseCount(unconfined: Boolean = false) {
@@ -117,22 +156,37 @@ internal abstract class EventLoop : CoroutineDispatcher() {
     }
 
     open fun shutdown() {}
+
+    override fun runUnconfinedEventLoop(initialBlock: () -> Unit) {
+        incrementUseCount(unconfined = true)
+        try {
+            initialBlock()
+            while (true) {
+                // break when all unconfined continuations where executed
+                if (!processUnconfinedEvent()) break
+            }
+        } finally {
+            decrementUseCount(unconfined = true)
+        }
+    }
+
+    override fun tryUseAsEventLoop(): EventLoop? = this
 }
 
 internal object ThreadLocalEventLoop {
-    private val ref = commonThreadLocal<EventLoop?>(Symbol("ThreadLocalEventLoop"))
+    private val ref = commonThreadLocal<UnconfinedEventLoop?>(Symbol("ThreadLocalEventLoop"))
 
-    internal val eventLoop: EventLoop
+    internal val unconfinedEventLoop: UnconfinedEventLoop
         get() = ref.get() ?: createEventLoop().also { ref.set(it) }
 
-    internal fun currentOrNull(): EventLoop? =
+    internal fun currentOrNull(): UnconfinedEventLoop? =
         ref.get()
 
     internal fun resetEventLoop() {
         ref.set(null)
     }
 
-    internal fun setEventLoop(eventLoop: EventLoop) {
+    internal fun setEventLoop(eventLoop: UnconfinedEventLoop) {
         ref.set(eventLoop)
     }
 }
@@ -183,8 +237,10 @@ internal abstract class EventLoopImplBase: EventLoopImplPlatform(), Delay {
         get() = _isCompleted.value
         set(value) { _isCompleted.value = value }
 
-    override val isEmpty: Boolean get() {
-        if (!isUnconfinedQueueEmpty) return false
+    /**
+     * Checks that at the moment this method is called, there are no tasks in the delayed tasks queue.
+     */
+    protected val delayedQueueIsEmpty: Boolean get() {
         val delayed = _delayed.value
         if (delayed != null && !delayed.isEmpty) return false
         return when (val queue = _queue.value) {
@@ -383,12 +439,6 @@ internal abstract class EventLoopImplBase: EventLoopImplPlatform(), Delay {
         return delayedTask.scheduleTask(now, delayedQueue, this)
     }
 
-    // It performs "hard" shutdown for test cleanup purposes
-    protected fun resetAll() {
-        _queue.value = null
-        _delayed.value = null
-    }
-
     // This is a "soft" (normal) shutdown
     private fun rescheduleAllDelayed() {
         val now = nanoTime()

Diff for: kotlinx-coroutines-core/common/src/internal/DispatchedContinuation.kt

@@ -292,12 +292,12 @@ internal fun DispatchedContinuation<Unit>.yieldUndispatched(): Boolean =
  */
 private inline fun DispatchedContinuation<*>.executeUnconfined(
     contState: Any?, mode: Int, doYield: Boolean = false,
-    block: () -> Unit
+    noinline block: () -> Unit
 ): Boolean {
     assert { mode != MODE_UNINITIALIZED } // invalid execution mode
-    val eventLoop = ThreadLocalEventLoop.eventLoop
+    val eventLoop = ThreadLocalEventLoop.unconfinedEventLoop
     // If we are yielding and unconfined queue is empty, we can bail out as part of fast path
-    if (doYield && eventLoop.isUnconfinedQueueEmpty) return false
+    if (doYield && eventLoop.thisLoopsTaskCanAvoidYielding) return false
     return if (eventLoop.isUnconfinedLoopActive) {
         // When unconfined loop is active -- dispatch continuation for execution to avoid stack overflow
         _state = contState

Diff for: kotlinx-coroutines-core/common/src/internal/DispatchedTask.kt

@@ -165,7 +165,7 @@ internal fun <T> DispatchedTask<T>.resume(delegate: Continuation<T>, undispatche
 }
 
 private fun DispatchedTask<*>.resumeUnconfined() {
-    val eventLoop = ThreadLocalEventLoop.eventLoop
+    val eventLoop = ThreadLocalEventLoop.unconfinedEventLoop
     if (eventLoop.isUnconfinedLoopActive) {
         // When unconfined loop is active -- dispatch continuation for execution to avoid stack overflow
         eventLoop.dispatchUnconfined(this)
@@ -177,25 +177,18 @@ private fun DispatchedTask<*>.resumeUnconfined() {
     }
 }
 
-internal inline fun DispatchedTask<*>.runUnconfinedEventLoop(
-    eventLoop: EventLoop,
+internal fun DispatchedTask<*>.runUnconfinedEventLoop(
+    eventLoop: UnconfinedEventLoop,
     block: () -> Unit
 ) {
-    eventLoop.incrementUseCount(unconfined = true)
     try {
-        block()
-        while (true) {
-            // break when all unconfined continuations where executed
-            if (!eventLoop.processUnconfinedEvent()) break
-        }
+        eventLoop.runUnconfinedEventLoop(block)
     } catch (e: Throwable) {
         /*
          * This exception doesn't happen normally, only if we have a bug in implementation.
          * Report it as a fatal exception.
          */
         handleFatalException(e)
-    } finally {
-        eventLoop.decrementUseCount(unconfined = true)
     }
 }
 

Diff for: kotlinx-coroutines-core/common/test/flow/VirtualTime.kt

@@ -19,7 +19,7 @@ internal class VirtualTimeDispatcher(enclosingScope: CoroutineScope) : Coroutine
          */
         enclosingScope.launch {
             while (true) {
-                val delayNanos = ThreadLocalEventLoop.currentOrNull()?.processNextEvent()
+                val delayNanos = ThreadLocalEventLoop.currentOrNull()?.tryUseAsEventLoop()?.processNextEvent()
                     ?: error("Event loop is missing, virtual time source works only as part of event loop")
                 if (delayNanos <= 0) continue
                 if (delayNanos > 0 && delayNanos != Long.MAX_VALUE) {

Diff for: kotlinx-coroutines-core/concurrent/src/Builders.concurrent.kt

@@ -25,3 +25,6 @@ import kotlin.coroutines.*
  * they are resubmitted to [Dispatchers.IO].
  */
 public expect fun <T> runBlocking(context: CoroutineContext = EmptyCoroutineContext, block: suspend CoroutineScope.() -> T): T
+
+internal fun UnconfinedEventLoop.useAsEventLoopForRunBlockingOrFail(): EventLoop =
+    tryUseAsEventLoop() ?: throw IllegalStateException("runBlocking can not be run in direct dispatchers")

Diff for: kotlinx-coroutines-core/jsAndWasmJsShared/src/EventLoop.kt

@@ -2,11 +2,11 @@ package kotlinx.coroutines
 
 import kotlin.coroutines.*
 
-internal actual fun createEventLoop(): EventLoop = UnconfinedEventLoop()
+internal actual fun createEventLoop(): EventLoop = UnconfinedEventLoopImpl()
 
 internal actual fun nanoTime(): Long = unsupported()
 
-internal class UnconfinedEventLoop : EventLoop() {
+private class UnconfinedEventLoopImpl : EventLoop() {
     override fun dispatch(context: CoroutineContext, block: Runnable): Unit = unsupported()
 }
 

Diff for: kotlinx-coroutines-core/jvm/src/Builders.kt

@@ -58,13 +58,13 @@ public actual fun <T> runBlocking(context: CoroutineContext, block: suspend Coro
     val newContext: CoroutineContext
     if (contextInterceptor == null) {
         // create or use private event loop if no dispatcher is specified
-        eventLoop = ThreadLocalEventLoop.eventLoop
+        eventLoop = ThreadLocalEventLoop.unconfinedEventLoop.useAsEventLoopForRunBlockingOrFail()
         newContext = GlobalScope.newCoroutineContext(context + eventLoop)
     } else {
         // See if context's interceptor is an event loop that we shall use (to support TestContext)
         // or take an existing thread-local event loop if present to avoid blocking it (but don't create one)
         eventLoop = (contextInterceptor as? EventLoop)?.takeIf { it.shouldBeProcessedFromContext() }
-            ?: ThreadLocalEventLoop.currentOrNull()
+            ?: ThreadLocalEventLoop.currentOrNull()?.useAsEventLoopForRunBlockingOrFail()
         newContext = GlobalScope.newCoroutineContext(context)
     }
     val coroutine = BlockingCoroutine<T>(newContext, currentThread, eventLoop)

Diff for: kotlinx-coroutines-core/jvm/src/DefaultDelay.kt

@@ -1,3 +1,4 @@
+@file:JvmName("DefaultExecutorKt")
 package kotlinx.coroutines
 
 import kotlinx.atomicfu.*
@@ -70,20 +71,21 @@ private object DefaultDelayImpl : EventLoopImplBase(), Runnable {
         val oldName = currentThread.name
         currentThread.name = THREAD_NAME
         try {
-            ThreadLocalEventLoop.setEventLoop(DefaultDelayImpl)
+            ThreadLocalEventLoop.setEventLoop(DelegatingUnconfinedEventLoop)
             registerTimeLoopThread()
             try {
                 while (true) {
                     Thread.interrupted() // just reset interruption flag
                     val parkNanos = processNextEvent()
                     if (parkNanos == Long.MAX_VALUE) break // no more events
-                    parkNanos(DefaultDelayImpl, parkNanos)
+                    parkNanos(this@DefaultDelayImpl, parkNanos)
                 }
             } finally {
                 _thread.value = null
                 unregisterTimeLoopThread()
+                ThreadLocalEventLoop.resetEventLoop()
                 // recheck if queues are empty after _thread reference was set to null (!!!)
-                if (isEmpty) {
+                if (delayedQueueIsEmpty) {
                     notifyAboutThreadExiting()
                 } else {
                     /* recreate the thread, as there is still work to do,
@@ -111,7 +113,7 @@ private object DefaultDelayImpl : EventLoopImplBase(), Runnable {
         if (_thread.value != null) {
             val end = System.currentTimeMillis() + timeout.inWholeMilliseconds
             while (true) {
-                check(isEmpty) { "There are tasks in the DefaultExecutor" }
+                check(delayedQueueIsEmpty) { "There are tasks in the DefaultExecutor" }
                 synchronized(this) {
                     unpark(_thread.value ?: return)
                     val toWait = end - System.currentTimeMillis()
@@ -129,6 +131,31 @@ private object DefaultDelayImpl : EventLoopImplBase(), Runnable {
     override fun toString(): String = "DefaultDelay"
 }
 
+private object DelegatingUnconfinedEventLoop: UnconfinedEventLoop {
+    override val thisLoopsTaskCanAvoidYielding: Boolean
+        get() = defaultDelayRunningUnconfinedLoop()
+
+    override val isUnconfinedLoopActive: Boolean get() = false
+
+    override fun runUnconfinedEventLoop(initialBlock: () -> Unit) {
+        ioView.dispatch(ioView, Runnable {
+            ThreadLocalEventLoop.unconfinedEventLoop.runUnconfinedEventLoop(initialBlock)
+        })
+    }
+
+    override fun dispatchUnconfined(task: DispatchedTask<*>) =
+        defaultDelayRunningUnconfinedLoop()
+
+    override fun tryUseAsEventLoop(): EventLoop? = null
+}
+
+private fun defaultDelayRunningUnconfinedLoop(): Nothing {
+    throw UnsupportedOperationException(
+        "This method can only be called from the thread where an unconfined event loop is running, " +
+        "but no tasks can run on this thread."
+    )
+}
+
 /** A view separate from [Dispatchers.IO].
  * [Int.MAX_VALUE] instead of `1` to avoid needlessly using the [LimitedDispatcher] machinery. */
 private val ioView = Dispatchers.IO.limitedParallelism(Int.MAX_VALUE)

Diff for: kotlinx-coroutines-core/jvm/src/Dispatchers.kt

@@ -19,7 +19,8 @@ public actual object Dispatchers {
     public actual val Main: MainCoroutineDispatcher get() = MainDispatcherLoader.dispatcher
 
     @JvmStatic
-    public actual val Unconfined: CoroutineDispatcher = kotlinx.coroutines.Unconfined
+    public actual val Unconfined: CoroutineDispatcher get() =
+        kotlinx.coroutines.Unconfined
 
     /**
      * The [CoroutineDispatcher] that is designed for offloading blocking IO tasks to a shared pool of threads.

Diff for: kotlinx-coroutines-core/jvm/src/EventLoop.kt

@@ -46,7 +46,7 @@ internal actual fun createEventLoop(): EventLoop = BlockingEventLoop(Thread.curr
 public fun processNextEventInCurrentThread(): Long =
     // This API is used in Ktor for serverless integration where a single thread awaits a blocking call
     // (and, to avoid actual blocking, does something via this call), see #850
-    ThreadLocalEventLoop.currentOrNull()?.processNextEvent() ?: Long.MAX_VALUE
+    ThreadLocalEventLoop.currentOrNull()?.tryUseAsEventLoop()?.processNextEvent() ?: Long.MAX_VALUE
 
 internal actual inline fun platformAutoreleasePool(crossinline block: () -> Unit) = block()
 

Diff for: kotlinx-coroutines-core/jvm/test/channels/TickerChannelCommonTest.kt

@@ -9,7 +9,7 @@ import org.junit.runners.*
 import kotlin.test.*
 
 @RunWith(Parameterized::class)
-class TickerChannelCommonTest(private val channelFactory: Channel) : TestBase() {
+class TickerChannelCommonTest(private val channelFactory: Channel) : TestBase(disableOutCheck = true) {
     companion object {
         @Parameterized.Parameters(name = "{0}")
         @JvmStatic

Diff for: kotlinx-coroutines-core/jvm/test/guide/test/DispatcherGuideTest.kt

@@ -20,7 +20,7 @@ class DispatcherGuideTest {
         test("ExampleContext02") { kotlinx.coroutines.guide.exampleContext02.main() }.verifyLinesStart(
             "Unconfined      : I'm working in thread main",
             "main runBlocking: I'm working in thread main",
-            "Unconfined      : After delay in thread kotlinx.coroutines.DefaultDelay",
+            "Unconfined      : After delay in thread DefaultDispatcher oroutine#2",
             "main runBlocking: After delay in thread main"
         )
     }