azalea/kotlin-fork
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[Wasm] Support coroutines

Browse Source 
- Reuse JS IR Suspend function lowering
  - Fix types
  - Disable reinterpretCast-based optimization for inline
    classes
- Add basic support to Wasm stdlib
- Explicitly transform suspend functions into regular functions
  with continuations
- Clean suspend function handling from JS IR codegen
This commit is contained in:
Svyatoslav Kuzmich
2021-10-11 15:57:38 +03:00
parent e3f826d1b9
commit 3bce0cc055
27 changed files with 817 additions and 238 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libraries/stdlib/wasm/src/kotlin/coroutines/CoroutineImpl.kt
+103
View File
@@ -0,0 +1,103 @@
/*
* Copyright 2010-2021 JetBrains s.r.o. and Kotlin Programming Language contributors.
* 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.COROUTINE_SUSPENDED
@SinceKotlin("1.3")
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
private val _context: CoroutineContext? = resultContinuation?.context
public override val context: CoroutineContext get() = _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: Result<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) {
// 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 === COROUTINE_SUSPENDED) return
currentResult = outcome
currentException = null
} catch (exception: Throwable) { // Catch all exceptions
currentResult = null
currentException = exception
}
releaseIntercepted() // this state machine instance is terminating
val completion = resultContinuation!!
if (completion is CoroutineImpl) {
// unrolling recursion via loop
current = completion
} else {
// top-level completion reached -- invoke and return
if (currentException != null) {
completion.resumeWithException(currentException!!)
} else {
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?
public open fun create(completion: Continuation<*>): Continuation<Unit> {
throw UnsupportedOperationException("create(Continuation) has not been overridden")
}
public open fun create(value: Any?, completion: Continuation<*>): Continuation<Unit> {
throw UnsupportedOperationException("create(Any?;Continuation) has not been overridden")
}
}
internal object CompletedContinuation : Continuation<Any?> {
override val context: CoroutineContext
get() = error("This continuation is already complete")
override fun resumeWith(result: Result<Any?>) {
error("This continuation is already complete")
}
override fun toString(): String = "This continuation is already complete"
}
libraries/stdlib/wasm/src/kotlin/coroutines/Coroutines.kt
+43 -11
View File
@@ -1,21 +1,53 @@
/*
* Copyright 2010-2020 JetBrains s.r.o. and Kotlin Programming Language contributors.
* Copyright 2010-2021 JetBrains s.r.o. and Kotlin Programming Language contributors.
* 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.*
import kotlin.coroutines.intrinsics.COROUTINE_SUSPENDED
@PublishedApi
@SinceKotlin("1.3")
internal actual class SafeContinuation<in T> : Continuation<T> {
actual internal constructor(delegate: Continuation<T>, initialResult: Any?) { TODO("Wasm stdlib: Coroutines") }
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: Result<T>) {
val cur = this.result
when {
cur === UNDECIDED -> {
this.result = result.value
}
cur === COROUTINE_SUSPENDED -> {
this.result = RESUMED
delegate.resumeWith(result)
}
else -> throw IllegalStateException("Already resumed")
}
}
@PublishedApi
actual internal constructor(delegate: Continuation<T>) { TODO("Wasm stdlib: Coroutines") }
@PublishedApi
actual internal fun getOrThrow(): Any? = TODO("Wasm stdlib: Coroutines")
actual override val context: CoroutineContext = TODO("Wasm stdlib: Coroutines")
actual override fun resumeWith(result: Result<T>): Unit { TODO("Wasm stdlib: Coroutines") }
}
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 Result.Failure -> throw result.exception
else -> result // either COROUTINE_SUSPENDED or data
}
}
}
libraries/stdlib/wasm/src/kotlin/coroutines/CoroutinesIntrinsics.kt
+81 -17
View File
@@ -1,14 +1,12 @@
/*
* Copyright 2010-2020 JetBrains s.r.o. and Kotlin Programming Language contributors.
* Copyright 2010-2021 JetBrains s.r.o. and Kotlin Programming Language contributors.
* 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.intrinsics
import kotlin.coroutines.Continuation
import kotlin.coroutines.ContinuationInterceptor
import kotlin.coroutines.CoroutineContext
import kotlin.internal.InlineOnly
import kotlin.coroutines.*
import kotlin.wasm.internal.*
/**
* Starts an unintercepted coroutine without a receiver and with result type [T] and executes it until its first suspension.
@@ -22,10 +20,11 @@ import kotlin.internal.InlineOnly
* This function is designed to be used from inside of [suspendCoroutineUninterceptedOrReturn] to resume the execution of the suspended
* coroutine using a reference to the suspending function.
*/
@SinceKotlin("1.3")
@Suppress("UNCHECKED_CAST")
@kotlin.internal.InlineOnly
public actual inline fun <T> (suspend () -> T).startCoroutineUninterceptedOrReturn(
completion: Continuation<T>
): Any? = TODO("Wasm stdlib: Coroutines intrinsics")
): Any? = startCoroutineUninterceptedOrReturnIntrinsic0(this, completion)
/**
* Starts an unintercepted coroutine with receiver type [R] and result type [T] and executes it until its first suspension.
@@ -39,29 +38,81 @@ public actual inline fun <T> (suspend () -> T).startCoroutineUninterceptedOrRetu
* This function is designed to be used from inside of [suspendCoroutineUninterceptedOrReturn] to resume the execution of the suspended
* coroutine using a reference to the suspending function.
*/
@SinceKotlin("1.3")
@Suppress("UNCHECKED_CAST")
@kotlin.internal.InlineOnly
public actual inline fun <R, T> (suspend R.() -> T).startCoroutineUninterceptedOrReturn(
receiver: R,
completion: Continuation<T>
): Any? = TODO("Wasm stdlib: Coroutines intrinsics")
): Any? = startCoroutineUninterceptedOrReturnIntrinsic1(this, receiver, completion)
@InlineOnly
@Suppress("UNCHECKED_CAST")
@kotlin.internal.InlineOnly
internal actual inline fun <R, P, T> (suspend R.(P) -> T).startCoroutineUninterceptedOrReturn(
receiver: R,
param: P,
completion: Continuation<T>
): Any? = TODO("Wasm stdlib: Coroutines intrinsics")
): Any? = startCoroutineUninterceptedOrReturnIntrinsic2(this, receiver, param, completion)
@SinceKotlin("1.3")
/**
* 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 immediately in the invoker's call stack without going through the
* [ContinuationInterceptor] that might be present in the completion's [CoroutineContext].
* It is the invoker's responsibility to ensure that a proper invocation context is established.
* Note that [completion] of this function may get invoked in an arbitrary context.
*
* [Continuation.intercepted] can be used to acquire the intercepted continuation.
* Invocation of `resume(Unit)` on intercepted continuation guarantees that execution of
* both the coroutine and [completion] happens in the invocation context established by
* [ContinuationInterceptor].
*
* 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.
*/
@Suppress("UNCHECKED_CAST")
public actual fun <T> (suspend () -> T).createCoroutineUnintercepted(
completion: Continuation<T>
): Continuation<Unit> = TODO("Wasm stdlib: Coroutines intrinsics")
): Continuation<Unit> {
return createCoroutineFromSuspendFunction(completion) {
this.startCoroutineUninterceptedOrReturn(completion)
}
}
@SinceKotlin("1.3")
/**
* 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 immediately in the invoker's call stack without going through the
* [ContinuationInterceptor] that might be present in the completion's [CoroutineContext].
* It is the invoker's responsibility to ensure that a proper invocation context is established.
* Note that [completion] of this function may get invoked in an arbitrary context.
*
* [Continuation.intercepted] can be used to acquire the intercepted continuation.
* Invocation of `resume(Unit)` on intercepted continuation guarantees that execution of
* both the coroutine and [completion] happens in the invocation context established by
* [ContinuationInterceptor].
*
* 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.
*/
@Suppress("UNCHECKED_CAST")
public actual fun <R, T> (suspend R.() -> T).createCoroutineUnintercepted(
receiver: R,
completion: Continuation<T>
): Continuation<Unit> = TODO("Wasm stdlib: Coroutines intrinsics")
): Continuation<Unit> {
return createCoroutineFromSuspendFunction(completion) {
this.startCoroutineUninterceptedOrReturn(receiver, completion)
}
}
/**
* Intercepts this continuation with [ContinuationInterceptor].
@@ -72,5 +123,18 @@ public actual fun <R, T> (suspend R.() -> T).createCoroutineUnintercepted(
*
* If this function is invoked on other [Continuation] instances it returns `this` continuation unchanged.
*/
@SinceKotlin("1.3")
public actual fun <T> Continuation<T>.intercepted(): Continuation<T> = TODO("Wasm stdlib: Coroutines intrinsics")
public actual fun <T> Continuation<T>.intercepted(): Continuation<T> =
(this as? CoroutineImpl)?.intercepted() ?: this
@Suppress("UNCHECKED_CAST")
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()
}
}
}