MPP: Delay, EventLoop & runBlocking moved to common code

elizarov · elizarov · commit a12ee150abf1 · 2017-12-21T17:36:55.000+03:00
diff --git a/common/kotlinx-coroutines-core-common/src/main/kotlin/kotlinx/coroutines/experimental/CommonBuilders.kt b/common/kotlinx-coroutines-core-common/src/main/kotlin/kotlinx/coroutines/experimental/CommonBuilders.kt
@@ -1,6 +1,7 @@
 package kotlinx.coroutines.experimental
 
 import kotlin.coroutines.experimental.CoroutineContext
+import kotlin.coroutines.experimental.EmptyCoroutineContext
 
 @Suppress("EXPECTED_DECLARATION_WITH_DEFAULT_PARAMETER")
 public expect fun launch(
@@ -9,3 +10,8 @@ public expect fun launch(
     parent: Job? = null,
     block: suspend CoroutineScope.() -> Unit
 ): Job
+
+@Suppress("EXPECTED_DECLARATION_WITH_DEFAULT_PARAMETER")
+public expect fun <T> runBlocking(
+    context: CoroutineContext = EmptyCoroutineContext,
+    block: suspend CoroutineScope.() -> T): T
diff --git a/common/kotlinx-coroutines-core-common/src/main/kotlin/kotlinx/coroutines/experimental/CommonDelay.kt b/common/kotlinx-coroutines-core-common/src/main/kotlin/kotlinx/coroutines/experimental/CommonDelay.kt
@@ -0,0 +1,5 @@
+package kotlinx.coroutines.experimental
+
+public expect interface Delay
+
+public expect suspend fun delay(time: Long)
diff --git a/common/kotlinx-coroutines-core-common/src/main/kotlin/kotlinx/coroutines/experimental/CommonEventLoop.kt b/common/kotlinx-coroutines-core-common/src/main/kotlin/kotlinx/coroutines/experimental/CommonEventLoop.kt
@@ -0,0 +1,3 @@
+package kotlinx.coroutines.experimental
+
+public expect interface EventLoop
diff --git a/common/kotlinx-coroutines-core-common/src/main/kotlin/kotlinx/coroutines/experimental/CommonJob.kt b/common/kotlinx-coroutines-core-common/src/main/kotlin/kotlinx/coroutines/experimental/CommonJob.kt
@@ -43,7 +43,7 @@ public expect class JobCancellationException(
 
 public expect fun Job.disposeOnCompletion(handle: DisposableHandle): DisposableHandle
 public expect suspend fun Job.cancelAndJoin()
-@Suppress("EXPECTED_DECLARATION_WITH_DEFAULT_PARAMETER")
+@Suppress("EXPECTED_DECLARATION_WITH_DEFAULT_PARAMETER", "EXTENSION_SHADOWED_BY_MEMBER") // See KT-21598
 public expect fun Job.cancelChildren(cause: Throwable? = null)
 public expect suspend fun Job.joinChildren()
 
diff --git a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Builders.kt b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Builders.kt
@@ -164,7 +164,7 @@ public suspend fun <T> run(context: CoroutineContext, block: suspend () -> T): T
  * @param block the coroutine code.
  */
 @Throws(InterruptedException::class)
-public fun <T> runBlocking(context: CoroutineContext = EmptyCoroutineContext, block: suspend CoroutineScope.() -> T): T {
+public actual fun <T> runBlocking(context: CoroutineContext = EmptyCoroutineContext, block: suspend CoroutineScope.() -> T): T {
     val currentThread = Thread.currentThread()
     val eventLoop = if (context[ContinuationInterceptor] == null) BlockingEventLoop(currentThread) else null
     val newContext = newCoroutineContext(context + (eventLoop ?: EmptyCoroutineContext))
diff --git a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Delay.kt b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Delay.kt
@@ -30,7 +30,7 @@ import kotlin.coroutines.experimental.CoroutineContext
  * Implementation of this interface affects operation of
  * [delay][kotlinx.coroutines.experimental.delay] and [withTimeout] functions.
  */
-public interface Delay {
+public actual interface Delay {
     /**
      * Delays coroutine for a given time without blocking a thread and resumes it after a specified time.
      * This suspending function is cancellable.
@@ -81,8 +81,32 @@ public interface Delay {
  *
  * This function delegates to [Delay.scheduleResumeAfterDelay] if the context [CoroutineDispatcher]
  * implements [Delay] interface, otherwise it resumes using a built-in single-threaded scheduled executor service.
+ *
+ * @param time time in milliseconds.
+ */
+public actual suspend fun delay(time: Long) {
+    kotlin.require(time >= 0) { "Delay time $time cannot be negative" }
+    if (time <= 0) return // don't delay
+    return suspendCancellableCoroutine sc@ { cont: CancellableContinuation<Unit> ->
+        cont.context.delay.scheduleResumeAfterDelay(time, TimeUnit.MILLISECONDS, cont)
+    }
+}
+
+/**
+ * Delays coroutine for a given time without blocking a thread and resumes it after a specified time.
+ * This suspending function is cancellable.
+ * If the [Job] of the current coroutine is cancelled or completed while this suspending function is waiting, this function
+ * immediately resumes with [CancellationException].
+ *
+ * Note, that delay can be used in [select] invocation with [onTimeout][SelectBuilder.onTimeout] clause.
+ *
+ * This function delegates to [Delay.scheduleResumeAfterDelay] if the context [CoroutineDispatcher]
+ * implements [Delay] interface, otherwise it resumes using a built-in single-threaded scheduled executor service.
+ *
+ * @param time time in the specified [unit].
+ * @param unit time unit.
  */
-suspend fun delay(time: Long, unit: TimeUnit = TimeUnit.MILLISECONDS) {
+public 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> ->
diff --git a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/EventLoop.kt b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/EventLoop.kt
@@ -31,7 +31,7 @@ import kotlin.coroutines.experimental.CoroutineContext
  * It may optionally implement [Delay] interface and support time-scheduled tasks. It is used by [runBlocking] to
  * continue processing events when invoked from the event dispatch thread.
  */
-public interface EventLoop {
+public actual interface EventLoop {
     /**
      * Processes next event in this event loop.
      *
diff --git a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Job.kt b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Job.kt
@@ -267,6 +267,9 @@ public actual interface Job : CoroutineContext.Element {
     @Deprecated(message = "For binary compatibility", level = DeprecationLevel.HIDDEN)
     public fun invokeOnCompletion(handler: CompletionHandler, onCancelling: Boolean): DisposableHandle
 
+    /**
+     * @suppress **Deprecated**: Use with named `onCancelling` and `handler` parameters.
+     */
     @Deprecated(message = "Use with named `onCancelling` and `handler` parameters", level = DeprecationLevel.WARNING,
         replaceWith = ReplaceWith("this.invokeOnCompletion(onCancelling = onCancelling_, handler = handler)"))
     public fun invokeOnCompletion(onCancelling_: Boolean = false, handler: CompletionHandler): DisposableHandle
diff --git a/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/NonCancellable.kt b/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/NonCancellable.kt
@@ -67,7 +67,7 @@ public actual object NonCancellable : AbstractCoroutineContextElement(Job), Job
 
     /** Always returns [NonDisposableHandle]. */
     @Suppress("OverridingDeprecatedMember")
-    override fun invokeOnCompletion(onCancelling: Boolean, handler: CompletionHandler): DisposableHandle =
+    override fun invokeOnCompletion(onCancelling_: Boolean, handler: CompletionHandler): DisposableHandle =
         NonDisposableHandle
 
     /** Always returns [NonDisposableHandle]. */
diff --git a/js/kotlinx-coroutines-core-js/src/main/kotlin/kotlinx/coroutines/experimental/Builders.kt b/js/kotlinx-coroutines-core-js/src/main/kotlin/kotlinx/coroutines/experimental/Builders.kt
@@ -2,6 +2,8 @@ package kotlinx.coroutines.experimental
 
 import kotlin.coroutines.experimental.ContinuationInterceptor
 import kotlin.coroutines.experimental.CoroutineContext
+import kotlin.coroutines.experimental.EmptyCoroutineContext
+import kotlin.coroutines.experimental.startCoroutine
 
 /**
  * Launches new coroutine without blocking current thread and returns a reference to the coroutine as a [Job].
@@ -47,6 +49,27 @@ public actual fun launch(
     return coroutine
 }
 
+/**
+ * Runs new coroutine with the private event loop until its completion.
+ * This function should not be used from coroutine. It is designed to bridge regular code
+ * to libraries that are written in suspending style, to be used in `main` functions and in tests.
+ *
+ * The default [CoroutineDispatcher] for this builder in an implementation of [EventLoop] that processes continuations
+ * in this blocked thread until the completion of this coroutine.
+ * See [CoroutineDispatcher] for the other implementations that are provided by `kotlinx.coroutines`.
+ *
+ * @param context context of the coroutine. The default value is an implementation of [EventLoop].
+ * @param block the coroutine code.
+ */
+public actual fun <T> runBlocking(context: CoroutineContext = EmptyCoroutineContext, block: suspend CoroutineScope.() -> T): T {
+    val eventLoop = if (context[ContinuationInterceptor] == null) BlockingEventLoop() else null
+    val newContext = newCoroutineContext(context + (eventLoop ?: EmptyCoroutineContext))
+    val coroutine = BlockingCoroutine<T>(newContext, privateEventLoop = eventLoop != null)
+    coroutine.initParentJob(newContext[Job])
+    block.startCoroutine(coroutine, coroutine)
+    return coroutine.joinBlocking()
+}
+
 // --------------- implementation ---------------
 
 private open class StandaloneCoroutine(
@@ -68,3 +91,36 @@ private class LazyStandaloneCoroutine(
     }
 }
 
+private class BlockingCoroutine<T>(
+    parentContext: CoroutineContext,
+    private val privateEventLoop: Boolean
+) : AbstractCoroutine<T>(parentContext, true) {
+    private val eventLoop: EventLoop? = parentContext[ContinuationInterceptor] as? EventLoop
+
+    init {
+        if (privateEventLoop) require(eventLoop is BlockingEventLoop)
+    }
+
+    fun joinBlocking(): T {
+        while (true) {
+            val delay = eventLoop?.processNextEvent() ?: Double.MAX_VALUE
+            if (isCompleted) break
+            if (delay > 0) throw IllegalStateException("JS thread cannot be blocked, " +
+                "runBlocking { ... } cannot be waiting for its completion with timeout")
+        }
+        // process queued events (that could have been added after last processNextEvent and before cancel
+        if (privateEventLoop) (eventLoop as BlockingEventLoop).apply {
+            // We exit the "while" loop above when this coroutine's state "isCompleted",
+            // Here we should signal that BlockingEventLoop should not accept any more tasks
+            isCompleted = true
+            shutdown()
+        }
+        // now return result
+        val state = this.state
+        (state as? CompletedExceptionally)?.let { throw it.exception }
+        @Suppress("UNCHECKED_CAST")
+        return state as T
+    }
+}
+
+
diff --git a/js/kotlinx-coroutines-core-js/src/main/kotlin/kotlinx/coroutines/experimental/CoroutineContext.kt b/js/kotlinx-coroutines-core-js/src/main/kotlin/kotlinx/coroutines/experimental/CoroutineContext.kt
@@ -26,12 +26,38 @@ public actual object Unconfined : CoroutineDispatcher() {
  * [launch], [async], etc if no dispatcher nor any other [ContinuationInterceptor] is specified in their context.
  */
 @Suppress("PropertyName")
-public actual val DefaultDispatcher: CoroutineDispatcher = DefaultDispatcherImpl
+public actual val DefaultDispatcher: CoroutineDispatcher = DefaultExecutor
+
+internal object DefaultExecutor : CoroutineDispatcher(), Delay {
+    fun enqueue(block: Runnable) {
+        window.setTimeout({ block.run() }, 0)
+    }
+
+    fun schedule(time: Double, block: Runnable): Int =
+        window.setTimeout({ block.run() }, time.timeToInt())
+
+    fun removeScheduled(handle: Int) {
+        window.clearTimeout(handle)
+    }
 
-private object DefaultDispatcherImpl : CoroutineDispatcher() {
     override fun dispatch(context: CoroutineContext, block: Runnable) {
         window.setTimeout({ block.run() }, 0)
     }
+
+    override fun scheduleResumeAfterDelay(time: Double, continuation: CancellableContinuation<Unit>) {
+        window.setTimeout({ with(continuation) { resumeUndispatched(Unit) } }, time.timeToInt())
+    }
+
+    override fun invokeOnTimeout(time: Double, block: Runnable): DisposableHandle {
+        val handle = window.setTimeout({ block.run() }, time.timeToInt())
+        return object : DisposableHandle {
+            override fun dispose() {
+                window.clearTimeout(handle)
+            }
+        }
+    }
+
+    private fun Double.timeToInt(): Int = coerceIn(0.0..Int.MAX_VALUE.toDouble()).toInt()
 }
 
 /**
diff --git a/js/kotlinx-coroutines-core-js/src/main/kotlin/kotlinx/coroutines/experimental/Delay.kt b/js/kotlinx-coroutines-core-js/src/main/kotlin/kotlinx/coroutines/experimental/Delay.kt
@@ -0,0 +1,76 @@
+/*
+ * Copyright 2016-2017 JetBrains s.r.o.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kotlinx.coroutines.experimental
+
+import kotlin.coroutines.experimental.ContinuationInterceptor
+import kotlin.coroutines.experimental.CoroutineContext
+
+/**
+ * This dispatcher _feature_ is implemented by [CoroutineDispatcher] implementations that natively support
+ * scheduled execution of tasks.
+ *
+ * Implementation of this interface affects operation of
+ * [delay][kotlinx.coroutines.experimental.delay] and [withTimeout] functions.
+ */
+public actual interface Delay {
+    /**
+     * Schedules resume of a specified [continuation] after a specified delay [time].
+     *
+     * Continuation **must be scheduled** to resume even if it is already cancelled, because a cancellation is just
+     * an exception that the coroutine that used `delay` might wanted to catch and process. It might
+     * need to close some resources in its `finally` blocks, for example.
+     *
+     * This implementation is supposed to use dispatcher's native ability for scheduled execution in its thread(s).
+     * In order to avoid an extra delay of execution, the following code shall be used to resume this
+     * [continuation] when the code is already executing in the appropriate dispatcher:
+     *
+     * ```kotlin
+     * with(continuation) { resumeUndispatched(Unit) }
+     * ```
+     */
+    fun scheduleResumeAfterDelay(time: Double, continuation: CancellableContinuation<Unit>)
+
+    /**
+     * Schedules invocation of a specified [block] after a specified delay [time].
+     * The resulting [DisposableHandle] can be used to [dispose][DisposableHandle.dispose] of this invocation
+     * request if it is not needed anymore.
+     */
+    fun invokeOnTimeout(time: Double, block: Runnable): DisposableHandle
+}
+
+/**
+ * Delays coroutine for a given time without blocking and resumes it after a specified time.
+ * This suspending function is cancellable.
+ * If the [Job] of the current coroutine is cancelled or completed while this suspending function is waiting, this function
+ * immediately resumes with [CancellationException].
+ *
+ * This function delegates to [Delay.scheduleResumeAfterDelay] if the context [CoroutineDispatcher]
+ * implements [Delay] interface, otherwise it resumes using a built-in scheduler.
+ *
+ * @param time time in milliseconds.
+ */
+public actual suspend fun delay(time: Long) {
+    val dt = time.toDouble()
+    kotlin.require(dt >= 0) { "Delay time $time cannot be negative" }
+    if (dt <= 0) return // don't delay
+    return suspendCancellableCoroutine sc@ { cont: CancellableContinuation<Unit> ->
+        cont.context.delay.scheduleResumeAfterDelay(dt, cont)
+    }
+}
+
+/** Returns [Delay] implementation of the given context */
+internal val CoroutineContext.delay: Delay get() = get(ContinuationInterceptor) as? Delay ?: DefaultExecutor
diff --git a/js/kotlinx-coroutines-core-js/src/main/kotlin/kotlinx/coroutines/experimental/EventLoop.kt b/js/kotlinx-coroutines-core-js/src/main/kotlin/kotlinx/coroutines/experimental/EventLoop.kt
diff --git a/js/kotlinx-coroutines-core-js/src/main/kotlin/kotlinx/coroutines/experimental/internal/Heap.kt b/js/kotlinx-coroutines-core-js/src/main/kotlin/kotlinx/coroutines/experimental/internal/Heap.kt
diff --git a/js/kotlinx-coroutines-core-js/src/main/kotlin/kotlinx/coroutines/experimental/internal/LinkedList.kt b/js/kotlinx-coroutines-core-js/src/main/kotlin/kotlinx/coroutines/experimental/internal/LinkedList.kt

Original file line number	Diff line number	Diff line change
`@@ -0,0 +1,3 @@`
	`1`	`+package kotlinx.coroutines.experimental`
	`2`	`+`
	`3`	`+public expect interface EventLoop`
Original file line number	Diff line number	Diff line change
`@@ -31,7 +31,7 @@ import kotlin.coroutines.experimental.CoroutineContext`
`31`	`31`	`* It may optionally implement [Delay] interface and support time-scheduled tasks. It is used by [runBlocking] to`
`32`	`32`	`* continue processing events when invoked from the event dispatch thread.`
`33`	`33`	`*/`
`34`		`-public interface EventLoop {`
	`34`	`+public actual interface EventLoop {`
`35`	`35`	`/**`
`36`	`36`	`* Processes next event in this event loop.`
`37`	`37`	`*`