JS: implement release coroutines in stdlib

2018-07-08 22:43:03 +03:00
parent 2604da6f0e
commit 99ac43eb84
5 changed files with 313 additions and 6 deletions
@@ -0,0 +1,52 @@
+/*
+ * Copyright 2010-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license
+ * that can be found in the license/LICENSE.txt file.
+ */
+
+package kotlin.coroutines
+
+import kotlin.coroutines.intrinsics.CoroutineSingletons.*
+
+@PublishedApi
+@SinceKotlin("1.3")
+internal actual class SafeContinuation<in T>
+internal actual constructor(
+    private val delegate: Continuation<T>,
+    initialResult: Any?
+) : Continuation<T> {
+    @PublishedApi
+    internal actual constructor(delegate: Continuation<T>) : this(delegate, UNDECIDED)
+
+    public actual override val context: CoroutineContext
+        get() = delegate.context
+
+    private var result: Any? = initialResult
+
+    public actual override fun resumeWith(result: SuccessOrFailure<T>) {
+        val cur = this.result
+        when {
+            cur === UNDECIDED -> {
+                this.result = result
+            }
+            cur === COROUTINE_SUSPENDED -> {
+                this.result = RESUMED
+                delegate.resumeWith(result)
+            }
+            else -> throw IllegalStateException("Already resumed")
+        }
+    }
+
+    @PublishedApi
+    internal actual fun getOrThrow(): Any? {
+        if (result === UNDECIDED) {
+            result = COROUTINE_SUSPENDED
+            return COROUTINE_SUSPENDED
+        }
+        val result = this.result
+        return when {
+            result === RESUMED -> COROUTINE_SUSPENDED // already called continuation, indicate COROUTINE_SUSPENDED upstream
+            result is SuccessOrFailure.Failure -> throw result.exception
+            else -> result // either COROUTINE_SUSPENDED or data
+        }
+    }
+}
@@ -0,0 +1,130 @@
+/*
+ * Copyright 2010-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license
+ * that can be found in the license/LICENSE.txt file.
+ */
+
+@file:Suppress("INVISIBLE_REFERENCE", "INVISIBLE_MEMBER", "UNCHECKED_CAST")
+
+package kotlin.coroutines.intrinsics
+
+import kotlin.coroutines.*
+import kotlin.internal.InlineOnly
+
+/**
+ * Starts unintercepted coroutine without receiver and with result type [T] and executes it until its first suspension.
+ * Returns the result of the coroutine or throws its exception if it does not suspend or [COROUTINE_SUSPENDED] if it suspends.
+ * In the later case, the [completion] continuation is invoked when coroutine completes with result or exception.
+ *
+ * The coroutine is started directly in the invoker's thread without going through the [ContinuationInterceptor] that might
+ * be present in the completion's [CoroutineContext]. It is invoker's responsibility to ensure that the proper invocation
+ * context is established.
+ *
+ * This function is designed to be used from inside of [suspendCoroutineUninterceptedOrReturn] to resume the execution of suspended
+ * coroutine using a reference to the suspending function.
+ */
+@SinceKotlin("1.3")
+@InlineOnly
+public actual inline fun <T> (suspend () -> T).startCoroutineUninterceptedOrReturn(
+    completion: Continuation<T>
+): Any? = this.asDynamic()(completion, false)
+
+/**
+ * Starts unintercepted coroutine with receiver type [R] and result type [T] and executes it until its first suspension.
+ * Returns the result of the coroutine or throws its exception if it does not suspend or [COROUTINE_SUSPENDED] if it suspends.
+ * In the later case, the [completion] continuation is invoked when coroutine completes with result or exception.
+ *
+ * The coroutine is started directly in the invoker's thread without going through the [ContinuationInterceptor] that might
+ * be present in the completion's [CoroutineContext]. It is invoker's responsibility to ensure that the proper invocation
+ * context is established.
+ *
+ * This function is designed to be used from inside of [suspendCoroutineUninterceptedOrReturn] to resume the execution of suspended
+ * coroutine using a reference to the suspending function.
+ */
+@SinceKotlin("1.3")
+@InlineOnly
+public actual inline fun <R, T> (suspend R.() -> T).startCoroutineUninterceptedOrReturn(
+    receiver: R,
+    completion: Continuation<T>
+): Any? = this.asDynamic()(receiver, completion, false)
+
+
+/**
+ * Creates unintercepted coroutine without receiver and with result type [T].
+ * This function creates a new, fresh instance of suspendable computation every time it is invoked.
+ *
+ * To start executing the created coroutine, invoke `resume(Unit)` on the returned [Continuation] instance.
+ * The [completion] continuation is invoked when coroutine completes with result or exception.
+ *
+ * This function returns unintercepted continuation.
+ * Invocation of `resume(Unit)` starts coroutine directly in the invoker's thread without going through the
+ * [ContinuationInterceptor] that might be present in the completion's [CoroutineContext].
+ * It is invoker's responsibility to ensure that the proper invocation context is established.
+ * [Continuation.intercepted] can be used to acquire the intercepted continuation.
+ *
+ * Repeated invocation of any resume function on the resulting continuation corrupts the
+ * state machine of the coroutine and may result in arbitrary behaviour or exception.
+ */
+@SinceKotlin("1.3")
+public actual fun <T> (suspend () -> T).createCoroutineUnintercepted(
+    completion: Continuation<T>
+): Continuation<Unit> =
+    // Kotlin/JS suspend lambdas have an extra parameter `suspended`
+    if (this.asDynamic().length == 2) {
+        // When `suspended` is true the continuation is created, but not executed
+        this.asDynamic()(completion, true)
+    } else {
+        createCoroutineFromSuspendFunction(completion) {
+            this.asDynamic()(completion)
+        }
+    }
+
+/**
+ * Creates unintercepted coroutine with receiver type [R] and result type [T].
+ * This function creates a new, fresh instance of suspendable computation every time it is invoked.
+ *
+ * To start executing the created coroutine, invoke `resume(Unit)` on the returned [Continuation] instance.
+ * The [completion] continuation is invoked when coroutine completes with result or exception.
+ *
+ * This function returns unintercepted continuation.
+ * Invocation of `resume(Unit)` starts coroutine directly in the invoker's thread without going through the
+ * [ContinuationInterceptor] that might be present in the completion's [CoroutineContext].
+ * It is invoker's responsibility to ensure that the proper invocation context is established.
+ * [Continuation.intercepted] can be used to acquire the intercepted continuation.
+ *
+ * Repeated invocation of any resume function on the resulting continuation corrupts the
+ * state machine of the coroutine and may result in arbitrary behaviour or exception.
+ */
+@SinceKotlin("1.3")
+public actual fun <R, T> (suspend R.() -> T).createCoroutineUnintercepted(
+    receiver: R,
+    completion: Continuation<T>
+): Continuation<Unit> =
+    // Kotlin/JS suspend lambdas have an extra parameter `suspended`
+    if (this.asDynamic().length == 3) {
+        // When `suspended` is true the continuation is created, but not executed
+        this.asDynamic()(receiver, completion, true)
+    } else {
+        createCoroutineFromSuspendFunction(completion) {
+            this.asDynamic()(receiver, completion)
+        }
+    }
+
+/**
+ * Intercepts this continuation with [ContinuationInterceptor].
+ */
+@SinceKotlin("1.3")
+public actual fun <T> Continuation<T>.intercepted(): Continuation<T> =
+    (this as? CoroutineImpl)?.intercepted() ?: this
+
+
+private inline fun <T> createCoroutineFromSuspendFunction(
+    completion: Continuation<T>,
+    crossinline block: () -> Any?
+): Continuation<Unit> {
+    return object : CoroutineImpl(completion as Continuation<Any?>) {
+        override fun doResume(): Any? {
+            exception?.let { throw it }
+            return block()
+        }
+    }
+}
@@ -0,0 +1,92 @@
+/*
+ * Copyright 2010-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license
+ * that can be found in the license/LICENSE.txt file.
+ */
+
+package kotlin.coroutines
+
+import kotlin.coroutines.intrinsics.CoroutineSingletons
+
+@SinceKotlin("1.3")
+@JsName("CoroutineImpl")
+internal abstract class CoroutineImpl(private val resultContinuation: Continuation<Any?>) : Continuation<Any?> {
+    protected var state = 0
+    protected var exceptionState = 0
+    protected var result: Any? = null
+    protected var exception: Throwable? = null
+    protected var finallyPath: Array<Int>? = null
+
+    public override val context: CoroutineContext = resultContinuation.context
+
+    private var intercepted_: Continuation<Any?>? = null
+
+    public fun intercepted(): Continuation<Any?> =
+        intercepted_
+                ?: (context[ContinuationInterceptor]?.interceptContinuation(this) ?: this)
+                    .also { intercepted_ = it }
+
+    override fun resumeWith(result: SuccessOrFailure<Any?>) {
+        var current = this
+        var currentResult: Any? = result.getOrNull()
+        var currentException: Throwable? = result.exceptionOrNull()
+
+        // This loop unrolls recursion in current.resumeWith(param) to make saner and shorter stack traces on resume
+        while (true) {
+            with(current) {
+                val completion = resultContinuation
+
+                // Set result and exception fields in the current continuation
+                if (currentException == null) {
+                    this.result = currentResult
+                } else {
+                    state = exceptionState
+                    exception = currentException
+                }
+
+                try {
+                    val outcome = doResume()
+                    if (outcome === CoroutineSingletons.COROUTINE_SUSPENDED) return
+                    currentResult = outcome
+                    currentException = null
+                } catch (exception: dynamic) { // Catch all exceptions
+                    currentResult = null
+                    currentException = exception.unsafeCast<Throwable>()
+                }
+
+                releaseIntercepted() // this state machine instance is terminating
+
+                if (completion is CoroutineImpl) {
+                    // unrolling recursion via loop
+                    current = completion
+                } else {
+                    // top-level completion reached -- invoke and return
+                    currentException?.let {
+                        completion.resumeWithException(it)
+                    } ?: completion.resume(currentResult)
+                    return
+                }
+            }
+        }
+    }
+
+    private fun releaseIntercepted() {
+        val intercepted = intercepted_
+        if (intercepted != null && intercepted !== this) {
+            context[ContinuationInterceptor]!!.releaseInterceptedContinuation(intercepted)
+        }
+        this.intercepted_ = CompletedContinuation // just in case
+    }
+
+    protected abstract fun doResume(): Any?
+}
+
+internal object CompletedContinuation : Continuation<Any?> {
+    override val context: CoroutineContext
+        get() = error("This continuation is already complete")
+
+    override fun resumeWith(result: SuccessOrFailure<Any?>) {
+        error("This continuation is already complete")
+    }
+
+    override fun toString(): String = "This continuation is already complete"
+}
@@ -10,7 +10,8 @@
    "NON_PUBLIC_PRIMARY_CONSTRUCTOR_OF_INLINE_CLASS",
    "UNSUPPORTED_FEATURE",
    "INVISIBLE_REFERENCE",
-    "INVISIBLE_MEMBER"
+    "INVISIBLE_MEMBER",
+    "CANNOT_OVERRIDE_INVISIBLE_MEMBER"
 )

 package kotlin