Fix DefaultExecutor not being able to exit

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

@@ -52,7 +52,7 @@ internal abstract class EventLoop : CoroutineDispatcher() {
      */
     public open fun processNextEvent(): Long {
         if (!processUnconfinedEvent()) return Long.MAX_VALUE
-        return nextTime
+        return 0
     }
 
     protected open val isEmpty: Boolean get() = isUnconfinedQueueEmpty
@@ -251,7 +251,7 @@ internal abstract class EventLoopImplBase: EventLoopImplPlatform(), Delay {
 
     override fun processNextEvent(): Long {
         // unconfined events take priority
-        if (processUnconfinedEvent()) return nextTime
+        if (processUnconfinedEvent()) return 0
         // queue all delayed tasks that are due to be executed
         val delayed = _delayed.value
         if (delayed != null && !delayed.isEmpty) {
@@ -269,7 +269,11 @@ internal abstract class EventLoopImplBase: EventLoopImplPlatform(), Delay {
             }
         }
         // then process one event from queue
-        dequeue()?.run()
+        val task = dequeue()
+        if (task != null) {
+            task.run()
+            return 0
+        }
         return nextTime
     }
 

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

@@ -68,15 +68,13 @@ internal actual object DefaultExecutor : EventLoopImplBase(), Runnable {
                 var parkNanos = processNextEvent()
                 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)
+                } else
+                    shutdownNanos = Long.MAX_VALUE
                 if (parkNanos > 0) {
                     // check if shutdown was requested and bail out in this case
                     if (isShutdownRequested) return
@@ -142,4 +140,7 @@ internal actual object DefaultExecutor : EventLoopImplBase(), Runnable {
         resetAll() // clear queues
         (this as Object).notifyAll()
     }
+
+    internal val isThreadPresent
+        get() = _thread != null
 }

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

@@ -4,7 +4,8 @@
 
 package kotlinx.coroutines
 
-import org.junit.*
+import org.junit.Test
+import kotlin.test.*
 
 class DefaultExecutorStressTest : TestBase() {
     @Test
@@ -35,4 +36,30 @@ class DefaultExecutorStressTest : TestBase() {
         }
         finish(2 + iterations * 4)
     }
+
+    @Test
+    fun testWorkerShutdown() = withVirtualTimeSource {
+        val iterations = 1_000 * stressTestMultiplier
+        // wait for the worker to shut down
+        suspend fun awaitWorkerShutdown() {
+            val executorTimeoutMs = 1000L
+            delay(executorTimeoutMs)
+            while (DefaultExecutor.isThreadPresent) { delay(10) } // hangs if the thread refuses to stop
+            assertFalse(DefaultExecutor.isThreadPresent) // just to make sure
+        }
+        runTest {
+            awaitWorkerShutdown() // so that the worker shuts down after the initial launch
+            repeat (iterations) {
+                val job = launch(Dispatchers.Unconfined) {
+                    // this line runs in the main thread
+                    delay(1)
+                    // this line runs in the DefaultExecutor worker
+                }
+                delay(100) // yield the execution, allow the worker to spawn
+                assertTrue(DefaultExecutor.isThreadPresent) // the worker spawned
+                job.join()
+                awaitWorkerShutdown()
+            }
+        }
+    }
 }

kotlinx-coroutines-core/jvm/test/channels/TickerChannelTest.kt

@@ -47,17 +47,17 @@ class TickerChannelTest : TestBase() {
     @Test
     fun testDelayChannelBackpressure2() = withVirtualTimeSource {
         runTest {
-            val delayChannel = ticker(delayMillis = 1000, initialDelayMillis = 0)
+            val delayChannel = ticker(delayMillis = 200, initialDelayMillis = 0)
             delayChannel.checkNotEmpty()
             delayChannel.checkEmpty()
 
-            delay(2500)
+            delay(500)
             delayChannel.checkNotEmpty()
-            delay(510)
+            delay(110)
             delayChannel.checkNotEmpty()
-            delay(510)
+            delay(110)
             delayChannel.checkEmpty()
-            delay(510)
+            delay(110)
             delayChannel.checkNotEmpty()
             delayChannel.cancel()
         }