Fix DefaultExecutor not being able to exit

Solves #856.
Kotlin · Mar 23, 2020 · b1751db · b1751db
1 parent 3fdd3fe
commit b1751db
Show file tree

Hide file tree

Showing 2 changed files with 32 additions and 16 deletions.
diff --git a/kotlinx-coroutines-core/common/src/EventLoop.common.kt b/kotlinx-coroutines-core/common/src/EventLoop.common.kt
@@ -23,6 +23,8 @@ import kotlin.native.concurrent.*
 internal abstract class EventLoop : CoroutineDispatcher() {
     /**
      * Counts the number of nested `runBlocking` and [Dispatchers.Unconfined] that use this event loop.
+     * There are two 32-bit counters encoded in this 64-bit value, allowing to count [Dispatchers.Unconfined]
+     * separately from `runBlocking`; see [delta] and its uses.
      */
     private var useCount = 0L
 
@@ -51,7 +53,8 @@ internal abstract class EventLoop : CoroutineDispatcher() {
      *          (no check for performance reasons, may be added in the future).
      */
     public open fun processNextEvent(): Long {
-        if (!processUnconfinedEvent()) return Long.MAX_VALUE
+        val task = popNextTask() ?: return Long.MAX_VALUE
+        task.run()
         return nextTime
     }
 
@@ -64,11 +67,17 @@ internal abstract class EventLoop : CoroutineDispatcher() {
         }
 
     public fun processUnconfinedEvent(): Boolean {
-        val queue = unconfinedQueue ?: return false
-        val task = queue.removeFirstOrNull() ?: return false
+        val task = popUnconfinedTask() ?: return false
         task.run()
         return true
     }
+
+    public fun popUnconfinedTask(): DispatchedTask<*>? =
+        unconfinedQueue?.removeFirstOrNull()
+
+    protected open fun popNextTask(onlyUnconfined: Boolean = false): Runnable? =
+        popUnconfinedTask()
+
     /**
      * 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).
@@ -249,9 +258,15 @@ internal abstract class EventLoopImplBase: EventLoopImplPlatform(), Delay {
         }
     }
 
-    override fun processNextEvent(): Long {
+    override fun popNextTask(onlyUnconfined: Boolean): Runnable? {
+        val unconfined = popUnconfinedTask()
+        if (onlyUnconfined) {
+            return unconfined
+        }
         // unconfined events take priority
-        if (processUnconfinedEvent()) return nextTime
+        if (unconfined != null) {
+            return unconfined
+        }
         // queue all delayed tasks that are due to be executed
         val delayed = _delayed.value
         if (delayed != null && !delayed.isEmpty) {
@@ -269,8 +284,7 @@ internal abstract class EventLoopImplBase: EventLoopImplPlatform(), Delay {
             }
         }
         // then process one event from queue
-        dequeue()?.run()
-        return nextTime
+        return dequeue()
     }
 
     public final override fun dispatch(context: CoroutineContext, block: Runnable) = enqueue(block)

diff --git a/kotlinx-coroutines-core/jvm/src/DefaultExecutor.kt b/kotlinx-coroutines-core/jvm/src/DefaultExecutor.kt
@@ -65,17 +65,19 @@ internal actual object DefaultExecutor : EventLoopImplBase(), Runnable {
             if (!notifyStartup()) return
             while (true) {
                 Thread.interrupted() // just reset interruption flag
-                var parkNanos = processNextEvent()
+                val nextTask = popNextTask()
+                if (nextTask != null) {
+                    shutdownNanos = Long.MAX_VALUE
+                    nextTask.run()
+                }
+                var parkNanos = nextTime
                 if (parkNanos == Long.MAX_VALUE) {
                     // nothing to do, initialize shutdown timeout
-                    if (shutdownNanos == Long.MAX_VALUE) {
-                        val now = nanoTime()
-                        if (shutdownNanos == Long.MAX_VALUE) shutdownNanos = now + KEEP_ALIVE_NANOS
-                        val tillShutdown = shutdownNanos - now
-                        if (tillShutdown <= 0) return // shut thread down
-                        parkNanos = parkNanos.coerceAtMost(tillShutdown)
-                    } else
-                        parkNanos = parkNanos.coerceAtMost(KEEP_ALIVE_NANOS) // limit wait time anyway
+                    val now = nanoTime()
+                    if (shutdownNanos == Long.MAX_VALUE) shutdownNanos = now + KEEP_ALIVE_NANOS
+                    val tillShutdown = shutdownNanos - now
+                    if (tillShutdown <= 0) return // shut thread down
+                    parkNanos = parkNanos.coerceAtMost(tillShutdown)
                 }
                 if (parkNanos > 0) {
                     // check if shutdown was requested and bail out in this case