MPP: Ported DispatchedTask perf improvements to JS and more code is made common

Author: elizarov

common/kotlinx-coroutines-core-common/src/main/kotlin/kotlinx/coroutines/experimental/CommonCoroutineContext.kt

@@ -16,9 +16,12 @@
 
 package kotlinx.coroutines.experimental
 
-import kotlin.coroutines.experimental.CoroutineContext
+import kotlin.coroutines.experimental.*
 
 public expect object Unconfined : CoroutineDispatcher {
     override fun isDispatchNeeded(context: CoroutineContext): Boolean
     override fun dispatch(context: CoroutineContext, block: Runnable)
 }
+
+internal expect inline fun <T> withCoroutineContext(context: CoroutineContext, block: () -> T): T
+internal expect fun Continuation<*>.toDebugString(): String

common/kotlinx-coroutines-core-common/src/main/kotlin/kotlinx/coroutines/experimental/CommonExceptions.kt

@@ -0,0 +1,33 @@
+/*
+ * 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
+
+public expect class CompletionHandlerException(message: String, cause: Throwable) : RuntimeException
+
+public expect open class CancellationException(message: String) : IllegalStateException
+
+public expect class JobCancellationException(
+    message: String,
+    cause: Throwable?,
+    job: Job
+) : CancellationException {
+    val job: Job
+}
+
+public expect class TimeoutCancellationException public constructor(message: String)
+
+internal expect class DispatchException(message: String, cause: Throwable) : RuntimeException

common/kotlinx-coroutines-core-common/src/main/kotlin/kotlinx/coroutines/experimental/CommonJob.kt

@@ -45,18 +45,6 @@ public expect interface DisposableHandle {
     public fun dispose()
 }
 
-public expect class CompletionHandlerException(message: String, cause: Throwable) : RuntimeException
-
-public open expect class CancellationException(message: String) : IllegalStateException
-
-public expect class JobCancellationException(
-    message: String,
-    cause: Throwable?,
-    job: Job
-) : CancellationException {
-    val job: Job
-}
-
 @Suppress("EXPECTED_DECLARATION_WITH_DEFAULT_PARAMETER")
 public expect fun CoroutineContext.cancel(cause: Throwable? = null): Boolean
 @Suppress("EXPECTED_DECLARATION_WITH_DEFAULT_PARAMETER")

common/kotlinx-coroutines-core-common/src/main/kotlin/kotlinx/coroutines/experimental/CommonScheduled.kt

@@ -19,5 +19,3 @@ package kotlinx.coroutines.experimental
 public expect suspend fun <T> withTimeout(time: Int, block: suspend CoroutineScope.() -> T): T
 
 public expect suspend fun <T> withTimeoutOrNull(time: Int, block: suspend CoroutineScope.() -> T): T?
-
-public expect class TimeoutCancellationException public constructor(message: String)

core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/CompletedExceptionally.kt renamed to common/kotlinx-coroutines-core-common/src/main/kotlin/kotlinx/coroutines/experimental/CompletedExceptionally.kt

@@ -17,15 +17,17 @@
 package kotlinx.coroutines.experimental
 
 /**
- * Class for a [state] of a job that had completed exceptionally, including cancellation.
+ * Class for an internal state of a job that had completed exceptionally, including cancellation.
+ *
+ * **Note: This class cannot be used outside of internal coroutines framework**.
  *
  * @param cause the exceptional completion cause. If `cause` is null, then an exception is
  *        if created via [createException] on first get from [exception] property.
  * @param allowNullCause if `null` cause is allowed.
  * @suppress **This is unstable API and it is subject to change.**
  */
 public open class CompletedExceptionally protected constructor(
-    public @JvmField val cause: Throwable?,
+    public val cause: Throwable?,
     allowNullCause: Boolean
 ) {
     /**
@@ -50,5 +52,23 @@ public open class CompletedExceptionally protected constructor(
 
     protected open fun createException(): Throwable = error("Completion exception was not specified")
 
-    override fun toString(): String = "${this::class.java.simpleName}[$exception]"
-}
+    override fun toString(): String = "${this::class.simpleName}[$exception]"
+}
+
+/**
+ * A specific subclass of [CompletedExceptionally] for cancelled jobs.
+ *
+ * **Note: This class cannot be used outside of internal coroutines framework**.
+ * 
+ * @param job the job that was cancelled.
+ * @param cause the exceptional completion cause. If `cause` is null, then a [JobCancellationException]
+ *        if created on first get from [exception] property.
+ * @suppress **This is unstable API and it is subject to change.**
+ */
+public class Cancelled(
+    private val job: Job,
+    cause: Throwable?
+) : CompletedExceptionally(cause, true) {
+    override fun createException(): Throwable = JobCancellationException("Job was cancelled normally", null, job)
+}
+

common/kotlinx-coroutines-core-common/src/main/kotlin/kotlinx/coroutines/experimental/Dispatched.kt

@@ -0,0 +1,196 @@
+/*
+ * 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 kotlinx.coroutines.experimental.internal.*
+import kotlin.coroutines.experimental.*
+
+@Suppress("PrivatePropertyName")
+private val UNDEFINED = Symbol("UNDEFINED")
+
+internal class DispatchedContinuation<in T>(
+    val dispatcher: CoroutineDispatcher,
+    val continuation: Continuation<T>
+) : Continuation<T> by continuation, DispatchedTask<T> {
+    private var _state: Any? = UNDEFINED
+    public override var resumeMode: Int = 0
+
+    override fun takeState(): Any? {
+        val state = _state
+        check(state !== UNDEFINED) // fail-fast if repeatedly invoked
+        _state = UNDEFINED
+        return state
+    }
+
+    override val delegate: Continuation<T>
+        get() = this
+
+    override fun resume(value: T) {
+        val context = continuation.context
+        if (dispatcher.isDispatchNeeded(context)) {
+            _state = value
+            resumeMode = MODE_ATOMIC_DEFAULT
+            dispatcher.dispatch(context, this)
+        } else
+            resumeUndispatched(value)
+    }
+
+    override fun resumeWithException(exception: Throwable) {
+        val context = continuation.context
+        if (dispatcher.isDispatchNeeded(context)) {
+            _state = CompletedExceptionally(exception)
+            resumeMode = MODE_ATOMIC_DEFAULT
+            dispatcher.dispatch(context, this)
+        } else
+            resumeUndispatchedWithException(exception)
+    }
+
+    @Suppress("NOTHING_TO_INLINE") // we need it inline to save us an entry on the stack
+    inline fun resumeCancellable(value: T) {
+        val context = continuation.context
+        if (dispatcher.isDispatchNeeded(context)) {
+            _state = value
+            resumeMode = MODE_CANCELLABLE
+            dispatcher.dispatch(context, this)
+        } else
+            resumeUndispatched(value)
+    }
+
+    @Suppress("NOTHING_TO_INLINE") // we need it inline to save us an entry on the stack
+    inline fun resumeCancellableWithException(exception: Throwable) {
+        val context = continuation.context
+        if (dispatcher.isDispatchNeeded(context)) {
+            _state = CompletedExceptionally(exception)
+            resumeMode = MODE_CANCELLABLE
+            dispatcher.dispatch(context, this)
+        } else
+            resumeUndispatchedWithException(exception)
+    }
+
+    @Suppress("NOTHING_TO_INLINE") // we need it inline to save us an entry on the stack
+    inline fun resumeUndispatched(value: T) {
+        withCoroutineContext(context) {
+            continuation.resume(value)
+        }
+    }
+
+    @Suppress("NOTHING_TO_INLINE") // we need it inline to save us an entry on the stack
+    inline fun resumeUndispatchedWithException(exception: Throwable) {
+        withCoroutineContext(context) {
+            continuation.resumeWithException(exception)
+        }
+    }
+
+    // used by "yield" implementation
+    internal fun dispatchYield(value: T) {
+        val context = continuation.context
+        _state = value
+        resumeMode = MODE_CANCELLABLE
+        dispatcher.dispatch(context, this)
+    }
+
+    override fun toString(): String =
+        "DispatchedContinuation[$dispatcher, ${continuation.toDebugString()}]"
+}
+
+internal fun <T> Continuation<T>.resumeCancellable(value: T) = when (this) {
+    is DispatchedContinuation -> resumeCancellable(value)
+    else -> resume(value)
+}
+
+internal fun <T> Continuation<T>.resumeCancellableWithException(exception: Throwable) = when (this) {
+    is DispatchedContinuation -> resumeCancellableWithException(exception)
+    else -> resumeWithException(exception)
+}
+
+internal fun <T> Continuation<T>.resumeDirect(value: T) = when (this) {
+    is DispatchedContinuation -> continuation.resume(value)
+    else -> resume(value)
+}
+
+internal fun <T> Continuation<T>.resumeDirectWithException(exception: Throwable) = when (this) {
+    is DispatchedContinuation -> continuation.resumeWithException(exception)
+    else -> resumeWithException(exception)
+}
+
+/**
+ * @suppress **This is unstable API and it is subject to change.**
+ */
+public interface DispatchedTask<in T> : Runnable {
+    public val delegate: Continuation<T>
+    public val resumeMode: Int get() = MODE_CANCELLABLE
+
+    public fun takeState(): Any?
+
+    @Suppress("UNCHECKED_CAST")
+    public fun <T> getSuccessfulResult(state: Any?): T =
+        state as T
+
+    public fun getExceptionalResult(state: Any?): Throwable? =
+        (state as? CompletedExceptionally)?.exception
+
+    public override fun run() {
+        try {
+            val delegate = delegate as DispatchedContinuation<T>
+            val continuation = delegate.continuation
+            val context = continuation.context
+            val job = if (resumeMode.isCancellableMode) context[Job] else null
+            val state = takeState() // NOTE: Must take state in any case, even if cancelled
+            withCoroutineContext(context) {
+                if (job != null && !job.isActive)
+                    continuation.resumeWithException(job.getCancellationException())
+                else {
+                    val exception = getExceptionalResult(state)
+                    if (exception != null)
+                        continuation.resumeWithException(exception)
+                    else
+                        continuation.resume(getSuccessfulResult(state))
+                }
+            }
+        } catch (e: Throwable) {
+            throw DispatchException("Unexpected exception running $this", e)
+        }
+    }
+}
+
+/**
+ * @suppress **This is unstable API and it is subject to change.**
+ */
+public fun <T> DispatchedTask<T>.dispatch(mode: Int = MODE_CANCELLABLE) {
+    var useMode = mode
+    val delegate = this.delegate
+    if (mode.isDispatchedMode && delegate is DispatchedContinuation<*> && mode.isCancellableMode == resumeMode.isCancellableMode) {
+        // dispatch directly using this instance's Runnable implementation
+        val dispatcher = delegate.dispatcher
+        val context = delegate.context
+        if (dispatcher.isDispatchNeeded(context)
+        ) {
+            dispatcher.dispatch(context, this)
+            return // and that's it -- dispatched via fast-path
+        } else {
+            useMode = MODE_UNDISPATCHED
+        }
+    }
+    // slow-path - use delegate
+    val state = takeState()
+    val exception = getExceptionalResult(state)
+    if (exception != null) {
+        delegate.resumeWithExceptionMode(exception, useMode)
+    } else {
+        delegate.resumeMode(getSuccessfulResult(state), useMode)
+    }
+}

core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/ResumeMode.kt renamed to common/kotlinx-coroutines-core-common/src/main/kotlin/kotlinx/coroutines/experimental/ResumeMode.kt

@@ -16,7 +16,11 @@
 
 package kotlinx.coroutines.experimental.internal
 
-/** @suppress **This is unstable API and it is subject to change.** */
+/**
+ * A symbol class that is used to define unique constants that are self-explanatory in debugger.
+ *
+ * @suppress **This is unstable API and it is subject to change.**
+ */
 public class Symbol(val symbol: String) {
     override fun toString(): String = symbol
 }

core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/internal/Symbol.kt renamed to common/kotlinx-coroutines-core-common/src/main/kotlin/kotlinx/coroutines/experimental/internal/Symbol.kt

@@ -104,7 +104,7 @@ public fun newCoroutineContext(context: CoroutineContext, parent: Job? = null):
 /**
  * Executes a block using a given coroutine context.
  */
-internal inline fun <T> withCoroutineContext(context: CoroutineContext, block: () -> T): T {
+internal actual inline fun <T> withCoroutineContext(context: CoroutineContext, block: () -> T): T {
     val oldName = context.updateThreadContext()
     try {
         return block()