Default scheduler for delay is rewritten with support for virtual time

Guide tests use virtual test time for faster & more predictable operation

Author: elizarov

coroutines-guide.md

@@ -354,13 +354,13 @@ example shows:
 ```kotlin
 fun main(args: Array<String>) = runBlocking<Unit> {
     val job = launch(CommonPool) {
-        var nextPrintTime = 0L
+        var nextPrintTime = System.currentTimeMillis()
         var i = 0
         while (i < 10) { // computation loop
             val currentTime = System.currentTimeMillis()
             if (currentTime >= nextPrintTime) {
                 println("I'm sleeping ${i++} ...")
-                nextPrintTime = currentTime + 500L
+                nextPrintTime += 500L
             }
         }
     }
@@ -613,7 +613,7 @@ The answer is 42
 Completed in 2017 ms
 ```
 
-<!--- TEST FLEXIBLE_TIME -->
+<!--- TEST ARBITRARY_TIME -->
 
 ### Concurrent using async
 
@@ -646,7 +646,7 @@ The answer is 42
 Completed in 1017 ms
 ```
 
-<!--- TEST FLEXIBLE_TIME -->
+<!--- TEST ARBITRARY_TIME -->
 
 This is twice as fast, because we have concurrent execution of two coroutines. 
 Note, that concurrency with coroutines is always explicit.
@@ -678,7 +678,7 @@ The answer is 42
 Completed in 2017 ms
 ```
 
-<!--- TEST FLEXIBLE_TIME -->
+<!--- TEST ARBITRARY_TIME -->
 
 So, we are back to sequential execution, because we _first_ start and await for `one`, _and then_ start and await
 for `two`. It is not the intended use-case for laziness. It is designed as a replacement for
@@ -728,7 +728,7 @@ fun main(args: Array<String>) {
 
 > You can get full code [here](kotlinx-coroutines-core/src/test/kotlin/guide/example-compose-04.kt)
 
-<!--- TEST FLEXIBLE_TIME
+<!--- TEST ARBITRARY_TIME
 The answer is 42
 Completed in 1085 ms
 -->
@@ -1371,13 +1371,13 @@ fun launchProcessor(id: Int, channel: ReceiveChannel<Int>) = launch(CommonPool)
 }
 ```
 
-Now let us launch five processors and let them work for a second. See what happens:
+Now let us launch five processors and let them work for almost a second. See what happens:
 
 ```kotlin
 fun main(args: Array<String>) = runBlocking<Unit> {
     val producer = produceNumbers()
     repeat(5) { launchProcessor(it, producer) }
-    delay(1000)
+    delay(950)
     producer.cancel() // cancel producer coroutine and thus kill them all
 }
 ```

kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Builders.kt

@@ -206,7 +206,7 @@ private class BlockingCoroutine<T>(
     private val blockedThread: Thread,
     private val privateEventLoop: Boolean
 ) : AbstractCoroutine<T>(parentContext, true) {
-    val eventLoop: EventLoop? = parentContext[ContinuationInterceptor] as? EventLoop
+    private val eventLoop: EventLoop? = parentContext[ContinuationInterceptor] as? EventLoop
 
     init {
         if (privateEventLoop) require(eventLoop is EventLoopImpl)
@@ -219,15 +219,17 @@ private class BlockingCoroutine<T>(
 
     @Suppress("UNCHECKED_CAST")
     fun joinBlocking(): T {
+        timeSource.registerTimeLoopThread()
         while (true) {
             if (Thread.interrupted()) throw InterruptedException().also { cancel(it) }
             val parkNanos = eventLoop?.processNextEvent() ?: Long.MAX_VALUE
-            // note: process next even may look unpark flag, so check !isActive before parking
+            // note: process next even may loose unpark flag, so check if completed before parking
             if (isCompleted) break
-            LockSupport.parkNanos(this, parkNanos)
+            timeSource.parkNanos(this, parkNanos)
         }
         // process queued events (that could have been added after last processNextEvent and before cancel
         if (privateEventLoop) (eventLoop as EventLoopImpl).shutdown()
+        timeSource.unregisterTimeLoopThread()
         // now return result
         val state = this.state
         (state as? CompletedExceptionally)?.let { throw it.exception }

kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/CancellableContinuation.kt

@@ -94,9 +94,6 @@ public interface CancellableContinuation<in T> : Continuation<T>, Job {
      * [dispatch][CoroutineDispatcher.dispatch] function of the [CoroutineDispatcher] in the [context].
      * This function is designed to be used only by the [CoroutineDispatcher] implementations themselves.
      * **It should not be used in general code**.
-     *
-     * The receiver [CoroutineDispatcher] of this function be equal to the context dispatcher or
-     * [IllegalArgumentException] if thrown.
      */
     public fun CoroutineDispatcher.resumeUndispatched(value: T)
 
@@ -105,9 +102,6 @@ public interface CancellableContinuation<in T> : Continuation<T>, Job {
      * [dispatch][CoroutineDispatcher.dispatch] function of the [CoroutineDispatcher] in the [context].
      * This function is designed to be used only by the [CoroutineDispatcher] implementations themselves.
      * **It should not be used in general code**.
-     *
-     * The receiver [CoroutineDispatcher] of this function be equal to the context dispatcher or
-     * [IllegalArgumentException] if thrown.
      */
     public fun CoroutineDispatcher.resumeUndispatchedWithException(exception: Throwable)
 }
@@ -244,14 +238,12 @@ internal class CancellableContinuationImpl<in T>(
 
     override fun CoroutineDispatcher.resumeUndispatched(value: T) {
         val dc = delegate as? DispatchedContinuation ?: throw IllegalArgumentException("Must be used with DispatchedContinuation")
-        check(dc.dispatcher === this) { "Must be invoked from the context CoroutineDispatcher"}
-        resumeImpl(value, MODE_UNDISPATCHED)
+        resumeImpl(value, if (dc.dispatcher === this) MODE_UNDISPATCHED else resumeMode)
     }
 
     override fun CoroutineDispatcher.resumeUndispatchedWithException(exception: Throwable) {
         val dc = delegate as? DispatchedContinuation ?: throw IllegalArgumentException("Must be used with DispatchedContinuation")
-        check(dc.dispatcher === this) { "Must be invoked from the context CoroutineDispatcher"}
-        resumeWithExceptionImpl(exception, MODE_UNDISPATCHED)
+        resumeWithExceptionImpl(exception, if (dc.dispatcher === this) MODE_UNDISPATCHED else resumeMode)
     }
 
     override fun toString(): String = super.toString() + "[${delegate.toDebugString()}]"

kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/CommonPool.kt

@@ -60,8 +60,11 @@ object CommonPool : CoroutineDispatcher() {
         _pool ?: createPool().also { _pool = it }
 
     override fun dispatch(context: CoroutineContext, block: Runnable) =
-        try { (_pool ?: getOrCreatePoolSync()).execute(block) }
-        catch (e: RejectedExecutionException) { defaultExecutor.execute(block) }
+        try { (_pool ?: getOrCreatePoolSync()).execute(timeSource.trackTask(block)) }
+        catch (e: RejectedExecutionException) {
+            timeSource.unTrackTask()
+            DefaultExecutor.execute(block)
+        }
 
     // used for tests
     @Synchronized
@@ -78,7 +81,7 @@ object CommonPool : CoroutineDispatcher() {
             shutdown()
             if (timeout > 0)
                 awaitTermination(timeout, TimeUnit.MILLISECONDS)
-            shutdownNow().forEach { defaultExecutor.execute(it) }
+            shutdownNow().forEach { DefaultExecutor.execute(it) }
         }
         _pool = Executor { throw RejectedExecutionException("CommonPool was shutdown") }
     }

kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/DefaultExecutor.kt

@@ -16,46 +16,125 @@
 
 package kotlinx.coroutines.experimental
 
-import java.util.concurrent.ScheduledExecutorService
-import java.util.concurrent.ScheduledThreadPoolExecutor
 import java.util.concurrent.TimeUnit
 
-private const val DEFAULT_KEEP_ALIVE = 1000L
+@Suppress("PLATFORM_CLASS_MAPPED_TO_KOTLIN")
+internal object DefaultExecutor : EventLoopBase(), Runnable {
 
-private val KEEP_ALIVE =
-    try { java.lang.Long.getLong("kotlinx.coroutines.DefaultExecutor.keepAlive", DEFAULT_KEEP_ALIVE) }
-    catch (e: SecurityException) { DEFAULT_KEEP_ALIVE }
+    override val canComplete: Boolean get() = false
+    override val isCompleted: Boolean get() = false
 
-@Volatile
-private var _executor: ScheduledExecutorService? = null
+    private const val DEFAULT_KEEP_ALIVE = 1000L // in milliseconds
 
-internal val defaultExecutor: ScheduledExecutorService
-    get() = _executor ?: getOrCreateExecutorSync()
+    private val KEEP_ALIVE_NANOS = TimeUnit.MILLISECONDS.toNanos(
+        try {
+            java.lang.Long.getLong("kotlinx.coroutines.DefaultExecutor.keepAlive", DEFAULT_KEEP_ALIVE)
+        } catch (e: SecurityException) {
+            DEFAULT_KEEP_ALIVE
+        })
+
+    @Volatile
+    private var _thread: Thread? = null
+
+    private const val FRESH = 0
+    private const val ACTIVE = 1
+    private const val SHUTDOWN_REQ = 2
+    private const val SHUTDOWN_ACK = 3
+
+    @Volatile
+    private var debugStatus: Int = FRESH
 
-@Synchronized
-private fun getOrCreateExecutorSync(): ScheduledExecutorService =
-    _executor ?: ScheduledThreadPoolExecutor(1) { r ->
-        Thread(r, "kotlinx.coroutines.DefaultExecutor").apply { isDaemon = true }
-    }.apply {
-        setKeepAliveTime(KEEP_ALIVE, TimeUnit.MILLISECONDS)
-        allowCoreThreadTimeOut(true)
-        executeExistingDelayedTasksAfterShutdownPolicy = false
-        // "setRemoveOnCancelPolicy" is available only since JDK7, so try it via reflection
+    override fun run() {
+        var shutdownNanos = Long.MAX_VALUE
+        timeSource.registerTimeLoopThread()
+        notifyStartup()
         try {
-            val m = this::class.java.getMethod("setRemoveOnCancelPolicy", Boolean::class.javaPrimitiveType)
-            m.invoke(this, true)
-        } catch (ex: Throwable) { /* ignore */ }
-        _executor = this
+            runLoop@ while (true) {
+                Thread.interrupted() // just reset interruption flag
+                var parkNanos = processNextEvent()
+                if (parkNanos == Long.MAX_VALUE) {
+                    // nothing to do, initialize shutdown timeout
+                    if (shutdownNanos == Long.MAX_VALUE) {
+                        val now = timeSource.nanoTime()
+                        if (shutdownNanos == Long.MAX_VALUE) shutdownNanos = now + KEEP_ALIVE_NANOS
+                        val tillShutdown = shutdownNanos - now
+                        if (tillShutdown <= 0) break@runLoop // shut thread down
+                        parkNanos = parkNanos.coerceAtMost(tillShutdown)
+                    } else
+                        parkNanos = parkNanos.coerceAtMost(KEEP_ALIVE_NANOS) // limit wait time anyway
+                }
+                if (parkNanos > 0) {
+                    // check if shutdown was requested and bail out in this case
+                    if (debugStatus == SHUTDOWN_REQ) {
+                        acknowledgeShutdown()
+                        break@runLoop
+                    } else {
+                        timeSource.parkNanos(this, parkNanos)
+                    }
+                }
+            }
+        } finally {
+            _thread = null // this thread is dead
+            timeSource.unregisterTimeLoopThread()
+            // recheck if queues are empty after _thread reference was set to null (!!!)
+            if (!isEmpty) thread() // recreate thread if it is needed
+        }
     }
 
-// used for tests
-@Synchronized
-internal fun shutdownDefaultExecutor(timeout: Long) {
-    _executor?.apply {
-        shutdown()
-        awaitTermination(timeout, TimeUnit.MILLISECONDS)
-        shutdownNow() // ignore all remaining
-        _executor = null
+    // ensure that thread is there
+    private fun thread(): Thread = _thread ?: createThreadSync()
+
+    @Synchronized
+    private fun createThreadSync() = _thread ?:
+        Thread(this, "kotlinx.coroutines.DefaultExecutor").apply {
+            _thread = this
+            isDaemon = true
+            start()
+        }
+
+    override fun unpark() {
+        timeSource.unpark(thread()) // as a side effect creates thread if it is not there
     }
-}
 
+    override fun isCorrectThread(): Boolean = true
+
+    // used for tests
+    @Synchronized
+    internal fun ensureStarted() {
+        assert(_thread == null) // ensure we are at a clean state
+        debugStatus = FRESH
+        createThreadSync() // create fresh thread
+        while (debugStatus == FRESH) (this as Object).wait()
+    }
+
+    @Synchronized
+    private fun notifyStartup() {
+        debugStatus = ACTIVE
+        (this as Object).notifyAll()
+    }
+
+    // used for tests
+    @Synchronized
+    internal fun shutdown(timeout: Long) {
+        if (_thread != null) {
+            val deadline = System.currentTimeMillis() + timeout
+            if (debugStatus == ACTIVE) debugStatus = SHUTDOWN_REQ
+            unpark()
+            // loop while there is anything to do immediately or deadline passes
+            while (debugStatus != SHUTDOWN_ACK && _thread != null) {
+                val remaining = deadline - System.currentTimeMillis()
+                if (remaining <= 0) break
+                (this as Object).wait(timeout)
+            }
+        }
+        // restore fresh status
+        debugStatus = FRESH
+    }
+
+    @Synchronized
+    private fun acknowledgeShutdown() {
+        debugStatus = SHUTDOWN_ACK
+        clearAll() // clear queues
+        (this as Object).notifyAll()
+    }
+}

kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Delay.kt

@@ -21,6 +21,7 @@ import kotlinx.coroutines.experimental.selects.select
 import java.util.concurrent.Future
 import java.util.concurrent.TimeUnit
 import kotlin.coroutines.experimental.ContinuationInterceptor
+import kotlin.coroutines.experimental.CoroutineContext
 
 /**
  * This dispatcher _feature_ is implemented by [CoroutineDispatcher] implementations that natively support
@@ -67,7 +68,7 @@ public interface Delay {
      * This implementation uses a built-in single-threaded scheduled executor service.
      */
     fun invokeOnTimeout(time: Long, unit: TimeUnit, block: Runnable): DisposableHandle =
-        DisposableFutureHandle(defaultExecutor.schedule(block, time, unit))
+        DefaultExecutor.invokeOnTimeout(time, unit, block)
 }
 
 /**
@@ -85,10 +86,7 @@ suspend fun delay(time: Long, unit: TimeUnit = TimeUnit.MILLISECONDS) {
     require(time >= 0) { "Delay time $time cannot be negative" }
     if (time <= 0) return // don't delay
     return suspendCancellableCoroutine sc@ { cont: CancellableContinuation<Unit> ->
-        val delay = cont.context[ContinuationInterceptor] as? Delay
-        if (delay != null)
-            delay.scheduleResumeAfterDelay(time, unit, cont) else
-            cont.cancelFutureOnCompletion(defaultExecutor.schedule(ResumeRunnable(cont), time, unit))
+        cont.context.delay.scheduleResumeAfterDelay(time, unit, cont)
     }
 }
 
@@ -101,3 +99,6 @@ public class DisposableFutureHandle(private val future: Future<*>) : DisposableH
     }
     override fun toString(): String = "DisposableFutureHandle[$future]"
 }
+
+/** Returns [Delay] implementation of the given context */
+internal val CoroutineContext.delay: Delay get() = get(ContinuationInterceptor) as? Delay ?: DefaultExecutor