diff --git a/core/builtins/native/kotlin/Coroutines.kt b/core/builtins/native/kotlin/Coroutines.kt deleted file mode 100644 index c4b464fa075..00000000000 --- a/core/builtins/native/kotlin/Coroutines.kt +++ /dev/null @@ -1,40 +0,0 @@ -/* - * Copyright 2010-2016 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 kotlin.coroutines - -/** - * This function allows to obtain the current continuation instance inside suspend functions and either suspend - * currently running coroutine or return result immediately without suspension. - * This function can be used in a tail-call position as the return value of another suspend function. - * - * If the [body] returns the special [SUSPENDED_COMPUTATION] value, it means that suspend function did suspend the execution and will - * not return any result immediately. In this case, the [Continuation] provided to the [body] shall be invoked at some moment in the future when - * the result becomes available to resume the computation. - * - * Otherwise, the return value of the [body] must have a type assignable to [T] and represents the result of this suspend function. - * It means that the execution was not suspended and the [Continuation] provided to the [body] shall not be invoked. - * As the result type of the [body] is declared as `Any?` and cannot be correctly type-checked, - * its proper return type remains on the conscience of the suspend function's author. - * - * Note that it is not recommended to call either [Continuation.resume] nor [Continuation.resumeWithException] functions synchronously in - * the same stackframe where suspension function is run. They should be called asynchronously either later in the same thread or - * from a different thread of execution. - * Repeated invocation of any resume function on continuation produces unspecified behavior. - * Use [runWithCurrentContinuation] as a safer way to obtain current continuation instance. - */ -@SinceKotlin("1.1") -public inline suspend fun suspendWithCurrentContinuation(body: (Continuation) -> Any?): T diff --git a/core/builtins/src/kotlin/coroutines/Coroutines.kt b/core/builtins/src/kotlin/coroutines/Coroutines.kt index 8d9785b2412..470aee6c998 100644 --- a/core/builtins/src/kotlin/coroutines/Coroutines.kt +++ b/core/builtins/src/kotlin/coroutines/Coroutines.kt @@ -17,35 +17,83 @@ package kotlin.coroutines /** - * Interface representing a continuation after a suspension point that returns value of type `P` + * Interface representing a continuation after a suspension point that returns value of type `T`. */ @SinceKotlin("1.1") -public interface Continuation { +public interface Continuation { /** - * Resumes the execution of the corresponding coroutine passing `data` as the return value of the last suspension point + * Resumes the execution of the corresponding coroutine passing [value] as the return value of the last suspension point. */ - public fun resume(data: P) + public fun resume(value: T) /** - * Resumes the execution of the corresponding coroutine so that the `exception` is re-thrown right after the - * last suspension point + * Resumes the execution of the corresponding coroutine so that the [exception] is re-thrown right after the + * last suspension point. */ public fun resumeWithException(exception: Throwable) } /** - * This value can be used as a return value of [kotlin.coroutines.maySuspendWithCurrentContinuation] `body` argument to state that - * the execution was suspended and will not return any result immediately. + * An interface to customise dispatch of continuations. */ @SinceKotlin("1.1") -public val SUSPENDED: Any? = Any() +public interface ContinuationDispatcher { + /** + * Dispatches [Continuation.resume] invocation. + * This function must either return `false` or return `true` and invoke `continuation.resume(value)` asynchronously. + */ + public fun dispatchResume(value: T, continuation: Continuation): Boolean = false + + /** + * Dispatches [Continuation.resumeWithException] invocation. + * This function must either return `false` or return `true` and invoke `continuation.resumeWithException(exception)` asynchronously. + */ + public fun dispatchResumeWithException(exception: Throwable, continuation: Continuation<*>): Boolean = false +} /** * Classes and interfaces marked with this annotation are restricted when used as receivers for extension - * `suspend` functions. These `suspend` extensions can only invoke other member or extension `suspend` functions on this parcitular + * `suspend` functions. These `suspend` extensions can only invoke other member or extension `suspend` functions on this particular * receiver only and are restricted from calling arbitrary suspension functions. */ @SinceKotlin("1.1") @Target(AnnotationTarget.CLASS) @Retention(AnnotationRetention.BINARY) -public annotation class RestrictSuspension +public annotation class RestrictsSuspendExtensions + +/** + * Contains intrinsic functions for coroutines. + */ +@SinceKotlin("1.1") +public object CoroutineIntrinsics { + /** + * Obtains the current continuation instance inside suspend functions and either suspend + * currently running coroutine or return result immediately without suspension. + * + * If the [block] returns the special [SUSPENDED] value, it means that suspend function did suspend the execution and will + * not return any result immediately. In this case, the [Continuation] provided to the [block] shall be invoked at some moment in the + * future when the result becomes available to resume the computation. + * + * Otherwise, the return value of the [block] must have a type assignable to [T] and represents the result of this suspend function. + * It means that the execution was not suspended and the [Continuation] provided to the [block] shall not be invoked. + * As the result type of the [block] is declared as `Any?` and cannot be correctly type-checked, + * its proper return type remains on the conscience of the suspend function's author. + * + * Note that it is not recommended to call either [Continuation.resume] nor [Continuation.resumeWithException] functions synchronously + * in the same stackframe where suspension function is run. Use [suspendCoroutine] as a safer way to obtain current + * continuation instance. + */ + public inline suspend fun suspendCoroutineOrReturn( + block: (Continuation) -> Any? + ): T = suspendWithCurrentContinuation(block) + + /** + * This value is used as a return value of [suspendCoroutineOrReturn] `block` argument to state that + * the execution was suspended and will not return any result immediately. + */ + @SinceKotlin("1.1") + public val SUSPENDED: Any = Any() +} + +@PublishedApi +internal inline suspend fun suspendWithCurrentContinuation(body: (Continuation) -> Any?): T = null!! diff --git a/core/runtime.jvm/src/kotlin/coroutines/Builders.kt b/core/runtime.jvm/src/kotlin/coroutines/Builders.kt deleted file mode 100644 index 814c3731d6b..00000000000 --- a/core/runtime.jvm/src/kotlin/coroutines/Builders.kt +++ /dev/null @@ -1,108 +0,0 @@ -/* - * Copyright 2010-2016 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 kotlin.coroutines - -/** - * A strategy to intercept resumptions inside coroutine. - * Interceptor may shift resumption into another execution frame by scheduling asynchronous execution - * in this or another thread. - */ -@SinceKotlin("1.1") -public interface ResumeInterceptor { - /** - * Intercepts [Continuation.resume] invocation. - * This function must either return `false` or return `true` and invoke `continuation.resume(data)` asynchronously. - */ - public fun

interceptResume(data: P, continuation: Continuation

): Boolean = false - - /** - * Intercepts [Continuation.resumeWithException] invocation. - * This function must either return `false` or return `true` and invoke `continuation.resumeWithException(exception)` asynchronously. - */ - public fun interceptResumeWithException(exception: Throwable, continuation: Continuation<*>): Boolean = false -} - -/** - * Creates 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 result of the coroutine's execution is provided via invocation of [resultHandler]. - */ -@SinceKotlin("1.1") -@Suppress("UNCHECKED_CAST") -public fun (suspend R.() -> T).createCoroutine( - receiver: R, - resultHandler: Continuation, - resumeInterceptor: ResumeInterceptor? = null -): Continuation = (this as (R, Continuation) -> Continuation).invoke(receiver, withInterceptor(resultHandler, resumeInterceptor)) - -/** - * Starts coroutine with receiver type [R] and result type [T]. - * This function creates and start a new, fresh instance of suspendable computation every time it is invoked. - * The result of the coroutine's execution is provided via invocation of [resultHandler]. - */ -@SinceKotlin("1.1") -@Suppress("UNCHECKED_CAST") -public fun (suspend R.() -> T).startCoroutine( - receiver: R, - resultHandler: Continuation, - resumeInterceptor: ResumeInterceptor? = null -) { - createCoroutine(receiver, resultHandler, resumeInterceptor).resume(Unit) -} - -/** - * Creates 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 result of the coroutine's execution is provided via invocation of [resultHandler]. - * An optional [resumeInterceptor] may be specified to intercept resumes at suspension points inside the coroutine. - */ -@SinceKotlin("1.1") -@Suppress("UNCHECKED_CAST") -public fun (suspend () -> T).createCoroutine( - resultHandler: Continuation, - resumeInterceptor: ResumeInterceptor? = null -): Continuation = (this as (Continuation) -> Continuation).invoke(withInterceptor(resultHandler, resumeInterceptor)) - -/** - * Starts coroutine without receiver and with result type [T]. - * This function creates and start a new, fresh instance of suspendable computation every time it is invoked. - * The result of the coroutine's execution is provided via invocation of [resultHandler]. - * An optional [resumeInterceptor] may be specified to intercept resumes at suspension points inside the coroutine. - */ -@SinceKotlin("1.1") -@Suppress("UNCHECKED_CAST") -public fun (suspend () -> T).startCoroutine( - resultHandler: Continuation, - resumeInterceptor: ResumeInterceptor? = null -) { - createCoroutine(resultHandler, resumeInterceptor).resume(Unit) -} - -// ------- internal stuff ------- - -internal interface InterceptableContinuation

: Continuation

{ - val resumeInterceptor: ResumeInterceptor? -} - -private fun withInterceptor(resultHandler: Continuation, resumeInterceptor: ResumeInterceptor?): Continuation { - return if (resumeInterceptor == null) resultHandler else - object : InterceptableContinuation, Continuation by resultHandler { - override val resumeInterceptor: ResumeInterceptor? = resumeInterceptor - } -} diff --git a/core/runtime.jvm/src/kotlin/coroutines/ContinuationCapture.kt b/core/runtime.jvm/src/kotlin/coroutines/ContinuationCapture.kt deleted file mode 100644 index 0824a1e8084..00000000000 --- a/core/runtime.jvm/src/kotlin/coroutines/ContinuationCapture.kt +++ /dev/null @@ -1,119 +0,0 @@ -/* - * Copyright 2010-2016 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 kotlin.coroutines - -import java.lang.IllegalStateException -import java.util.concurrent.atomic.AtomicReferenceFieldUpdater - -/** - * This function allows to obtain the current continuation instance inside suspend functions and suspend - * currently running coroutine. - * This function can be used in a tail-call position as the return value of another suspend function. - * - * Note that it is not recommended to call either [Continuation.resume] nor [Continuation.resumeWithException] functions synchronously in - * the same stackframe where suspension function is run. They should be called asynchronously either later in the same thread or - * from a different thread of execution. - * In practise, this restriction means that only _asynchronous_ promises and callbacks can be used to resume the continuation - * in this function. - * Repeated invocation of any resume function on continuation produces unspecified behavior. - * Use [runWithCurrentContinuation] as a safer way to obtain current continuation instance. - */ -//@SinceKotlin("1.1") -//public inline suspend fun suspendWithCurrentContinuation(crossinline body: (Continuation) -> Unit): T = -// maySuspendWithCurrentContinuation { c: Continuation -> -// body(c) -// SUSPENDED -// } - -/** - * This function allows to safely obtain the current continuation instance inside suspend functions and suspend - * currently running coroutine. - * This function can be used in a tail-call position as the return value of another suspend function. - * - * In this function both [Continuation.resume] and [Continuation.resumeWithException] can be used either synchronously in - * the same stackframe where suspension function is run or asynchronously later in the same thread or - * from a different thread of execution. - * Repeated invocation of any resume function produces [IllegalStateException]. - */ -@SinceKotlin("1.1") -public inline suspend fun runWithCurrentContinuation(crossinline body: (Continuation) -> Unit): T = - suspendWithCurrentContinuation { c: Continuation -> - val safe = SafeContinuation(c) - body(safe) - safe.getResult() - } - -private val UNDECIDED: Any? = Any() -private val RESUMED: Any? = Any() -private class Fail(val exception: Throwable) - - -@PublishedApi -internal class SafeContinuation @PublishedApi internal constructor(private val delegate: Continuation) : Continuation { - @Volatile - private var result: Any? = UNDECIDED - - companion object { - @Suppress("UNCHECKED_CAST") - @JvmStatic - private val RESULT_UPDATER = AtomicReferenceFieldUpdater.newUpdater, Any?>( - SafeContinuation::class.java, Any::class.java as Class, "result") - } - - private fun cas(expect: Any?, update: Any?): Boolean = - RESULT_UPDATER.compareAndSet(this, expect, update) - - override fun resume(data: T) { - while (true) { // lock-free loop - val result = this.result // atomic read - when (result) { - UNDECIDED -> if (cas(UNDECIDED, data)) return - SUSPENDED -> if (cas(SUSPENDED, RESUMED)) { - delegate.resume(data) - return - } - else -> throw IllegalStateException("Already resumed") - } - } - } - - override fun resumeWithException(exception: Throwable) { - while (true) { // lock-free loop - val result = this.result // atomic read - when (result) { - UNDECIDED -> if (cas(UNDECIDED, Fail(exception))) return - SUSPENDED -> if (cas(SUSPENDED, RESUMED)) { - delegate.resumeWithException(exception) - return - } - else -> throw IllegalStateException("Already resumed") - } - } - } - - @PublishedApi - internal fun getResult(): Any? { - val result = this.result // atomic read - if (result == UNDECIDED && cas(UNDECIDED, SUSPENDED)) return SUSPENDED - when (result) { - RESUMED -> return SUSPENDED // already called continuation, indicate SUSPENDED upstream - is Fail -> throw result.exception - else -> return result // either SUSPENDED or data - } - } -} - diff --git a/core/runtime.jvm/src/kotlin/coroutines/Suspendable.kt b/core/runtime.jvm/src/kotlin/coroutines/Suspendable.kt deleted file mode 100644 index 7966f498966..00000000000 --- a/core/runtime.jvm/src/kotlin/coroutines/Suspendable.kt +++ /dev/null @@ -1,33 +0,0 @@ -/* - * Copyright 2010-2016 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 kotlin.coroutines - -/** - * Converts a callable references to `suspend` lambda into a tail-callable suspend function. - * It can be used to define arbitrary suspension function without tail-call restrictions, for example - * - * ``` - * suspend fun awaitTwice(...): R = suspendable { - * await(...) // calls one suspend function - * await(...) // calls another suspend function - * } - */ -@SinceKotlin("1.1") -suspend fun suspendable(/*suspend*/ lambda: () -> T): T = suspendWithCurrentContinuation { c -> - @Suppress("UNCHECKED_CAST") - (lambda as Function1, Any?>).invoke(c) -} diff --git a/core/runtime.jvm/src/kotlin/jvm/internal/CoroutineImpl.kt b/core/runtime.jvm/src/kotlin/jvm/internal/CoroutineImpl.kt index 2885486885f..8837bee0f18 100644 --- a/core/runtime.jvm/src/kotlin/jvm/internal/CoroutineImpl.kt +++ b/core/runtime.jvm/src/kotlin/jvm/internal/CoroutineImpl.kt @@ -21,34 +21,33 @@ import kotlin.coroutines.* private const val INTERCEPT_BIT_SET = 1 shl 31 private const val INTERCEPT_BIT_CLEAR = INTERCEPT_BIT_SET.inv() -@SinceKotlin("1.1") -abstract class CoroutineImpl : RestrictedCoroutineImpl, InterceptableContinuation { - private val _resumeInterceptor: ResumeInterceptor? +abstract class CoroutineImpl : RestrictedCoroutineImpl, DispatchedContinuation { + private val _dispatcher: ContinuationDispatcher? - override val resumeInterceptor: ResumeInterceptor? - get() = _resumeInterceptor + override val dispatcher: ContinuationDispatcher? + get() = _dispatcher // this constructor is used to create a continuation instance for coroutine - constructor(arity: Int, resultContinuation: Continuation?) : super(arity, resultContinuation) { - _resumeInterceptor = (resultContinuation as? InterceptableContinuation<*>)?.resumeInterceptor + constructor(arity: Int, completion: Continuation?) : super(arity, completion) { + _dispatcher = (completion as? DispatchedContinuation<*>)?.dispatcher } - override fun resume(data: Any?) { - if (_resumeInterceptor != null) { + override fun resume(value: Any?) { + if (_dispatcher != null) { if (label and INTERCEPT_BIT_SET == 0) { label = label or INTERCEPT_BIT_SET - if (_resumeInterceptor.interceptResume(data, this)) return + if (_dispatcher.dispatchResume(value, this)) return } label = label and INTERCEPT_BIT_CLEAR } - super.resume(data) + super.resume(value) } override fun resumeWithException(exception: Throwable) { - if (_resumeInterceptor != null) { + if (_dispatcher != null) { if (label and INTERCEPT_BIT_SET == 0) { label = label or INTERCEPT_BIT_SET - if (_resumeInterceptor.interceptResumeWithException(exception, this)) return + if (_dispatcher.dispatchResumeWithException(exception, this)) return } label = label and INTERCEPT_BIT_CLEAR } @@ -56,10 +55,9 @@ abstract class CoroutineImpl : RestrictedCoroutineImpl, InterceptableContinuatio } } -@SinceKotlin("1.1") abstract class RestrictedCoroutineImpl : Lambda, Continuation { @JvmField - protected var resultContinuation: Continuation? + protected var completion: Continuation? // label == -1 when coroutine cannot be started (it is just a factory object) or has already finished execution // label == 0 in initial part of the coroutine @@ -67,102 +65,34 @@ abstract class RestrictedCoroutineImpl : Lambda, Continuation { protected var label: Int // this constructor is used to create a continuation instance for coroutine - constructor(arity: Int, resultContinuation: Continuation?) : super(arity) { - this.resultContinuation = resultContinuation - label = if (resultContinuation != null) 0 else -1 + constructor(arity: Int, completion: Continuation?) : super(arity) { + this.completion = completion + label = if (completion != null) 0 else -1 } - override fun resume(data: Any?) { + override fun resume(value: Any?) { try { - val result = doResume(data, null) - if (result != SUSPENDED) - resultContinuation!!.resume(result) + val result = doResume(value, null) + if (result != CoroutineIntrinsics.SUSPENDED) + completion!!.resume(result) } catch (e: Throwable) { - resultContinuation!!.resumeWithException(e) + completion!!.resumeWithException(e) } } override fun resumeWithException(exception: Throwable) { try { val result = doResume(null, exception) - if (result != SUSPENDED) - resultContinuation!!.resume(result) + if (result != CoroutineIntrinsics.SUSPENDED) + completion!!.resume(result) } catch (e: Throwable) { - resultContinuation!!.resumeWithException(e) + completion!!.resumeWithException(e) } } protected abstract fun doResume(data: Any?, exception: Throwable?): Any? } -/* -=========================================================================================================================== -Showcase of coroutine compilation strategy. - -Given this following "sample" suspend function code: - ---------------------------------------------------------------------------------------------- -@RestrictSuspension -class Receiver { - suspend fun yield(): SomeResult +internal interface DispatchedContinuation : Continuation { + val dispatcher: ContinuationDispatcher? } - -suspend fun Receiver.sample(): String { - doSomethingBefore() - val yieldResult = yield() // suspension point - doSomethingAfter(yieldResult) - return "Done" -} ---------------------------------------------------------------------------------------------- - -The compiler emits the class with the following logic: - ---------------------------------------------------------------------------------------------- - -class XXXCoroutine : RestrictedCoroutineImpl, Function2, Any?> { - // ^^^^^^^^^^^^^^^^^^^^^^^ - // Replace with CoroutineImpl if the coroutine is non-restricted - - val receiver: Receiver? - // ^^^^^^^^^^^^^^^^^^^ - // receiver is just a part of the captured scope and is only declared here is coroutine has it - - // this constructor is used to create initial "factory" lambda object - constructor() : super(2, null) { - this.receiver = null - } - - // this constructor is used to create a continuation instance for coroutine - constructor(receiver: Receiver, resultContinuation: Continuation) : super(2, resultContinuation) { - // ^^^^^^^^^^^^^^^^^^^^ - // no receiver parameter when compiling coroutine that does not have one - this.receiver = receiver - } - - override fun doCreate(receiver: Any, resultContinuation: Continuation<*>): Continuation { - return XXXCoroutine(receiver as Receiver, resultContinuation) // create the actual instance - // ^^^^^^^^^^^^^^^^^^^^ - // ignore receiver parameter when compiling coroutine that does not have one, - // check that it is not null as check its proper type for coroutine that has a receiver - } - - override fun doResume(data: Any?, exception: Throwable?): Any? { - var suspensionResult = data - switch (label) { - default: - throw IllegalStateException() - case 0: - doSomethingBefore() - label = 1 // set next label before calling suspend function - suspensionResult = receiver.yield(this) - if (suspensionResult == SUSPENDED) return SUSPENDED - // falls through with yeild result if it is available - case 1: - doSomethingAfter(supensionResult) - label = -1 // execution is over - return "Done" // we have result of execution - } - } -} ---------------------------------------------------------------------------------------------- -*/ diff --git a/js/js.libraries/src/core/coroutines.kt b/js/js.libraries/src/core/coroutines.kt index bfc481a75f6..b9da7e94768 100644 --- a/js/js.libraries/src/core/coroutines.kt +++ b/js/js.libraries/src/core/coroutines.kt @@ -16,91 +16,72 @@ package kotlin.coroutines -/** - * A strategy to intercept resumptions inside coroutine. - * Interceptor may shift resumption into another execution frame by scheduling asynchronous execution - * in this or another thread. - */ -@SinceKotlin("1.1") -public interface ResumeInterceptor { - /** - * Intercepts [Continuation.resume] invocation. - * This function must either return `false` or return `true` and invoke `continuation.resume(data)` asynchronously. - */ - public fun

interceptResume(data: P, continuation: Continuation

): Boolean = false - - /** - * Intercepts [Continuation.resumeWithException] invocation. - * This function must either return `false` or return `true` and invoke `continuation.resumeWithException(exception)` asynchronously. - */ - public fun interceptResumeWithException(exception: Throwable, continuation: Continuation<*>): Boolean = false -} /** * Creates 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 result of the coroutine's execution is provided via invocation of [resultHandler]. + * The [completion] continuation is invoked when coroutine completes with result of exception. + * An optional [dispatcher] may be specified to customise dispatch of continuations between suspension points inside the coroutine. */ @SinceKotlin("1.1") public fun (suspend R.() -> T).createCoroutine( receiver: R, - resultHandler: Continuation, - resumeInterceptor: ResumeInterceptor? = null -): Continuation = this.asDynamic().call(receiver, withInterceptor(resultHandler, resumeInterceptor)) + completion: Continuation, + dispatcher: ContinuationDispatcher? = null +): Continuation = this.asDynamic().call(receiver, withDispatcher(completion, dispatcher)) /** * Starts coroutine with receiver type [R] and result type [T]. * This function creates and start a new, fresh instance of suspendable computation every time it is invoked. - * The result of the coroutine's execution is provided via invocation of [resultHandler]. + * The [completion] continuation is invoked when coroutine completes with result of exception. + * An optional [dispatcher] may be specified to customise dispatch of continuations between suspension points inside the coroutine. */ @SinceKotlin("1.1") -@Suppress("UNCHECKED_CAST") public fun (suspend R.() -> T).startCoroutine( receiver: R, - resultHandler: Continuation, - resumeInterceptor: ResumeInterceptor? = null + completion: Continuation, + dispatcher: ContinuationDispatcher? = null ) { - createCoroutine(receiver, resultHandler, resumeInterceptor).resume(Unit) + createCoroutine(receiver, completion, dispatcher).resume(Unit) } /** * Creates 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 result of the coroutine's execution is provided via invocation of [resultHandler]. - * An optional [resumeInterceptor] may be specified to intercept resumes at suspension points inside the coroutine. + * The [completion] continuation is invoked when coroutine completes with result of exception. + * An optional [dispatcher] may be specified to customise dispatch of continuations between suspension points inside the coroutine. */ @SinceKotlin("1.1") public fun (suspend () -> T).createCoroutine( - resultHandler: Continuation, - resumeInterceptor: ResumeInterceptor? = null -): Continuation = this.asDynamic()(withInterceptor(resultHandler, resumeInterceptor)) + completion: Continuation, + dispatcher: ContinuationDispatcher? = null +): Continuation = this.asDynamic()(withDispatcher(completion, dispatcher)) /** * Starts coroutine without receiver and with result type [T]. * This function creates and start a new, fresh instance of suspendable computation every time it is invoked. - * The result of the coroutine's execution is provided via invocation of [resultHandler]. - * An optional [resumeInterceptor] may be specified to intercept resumes at suspension points inside the coroutine. + * The [completion] continuation is invoked when coroutine completes with result of exception. + * An optional [dispatcher] may be specified to customise dispatch of continuations between suspension points inside the coroutine. */ @SinceKotlin("1.1") -@Suppress("UNCHECKED_CAST") -public fun (suspend () -> T).startCoroutine( - resultHandler: Continuation, - resumeInterceptor: ResumeInterceptor? = null +public fun (suspend () -> T).startCoroutine( + completion: Continuation, + dispatcher: ContinuationDispatcher? = null ) { - createCoroutine(resultHandler, resumeInterceptor).resume(Unit) + createCoroutine(completion, dispatcher).resume(Unit) } // ------- internal stuff ------- -private fun withInterceptor(resultHandler: Continuation, resumeInterceptor: ResumeInterceptor?): Continuation { - val finalResumeInterceptor = resumeInterceptor ?: object : ResumeInterceptor { - override fun

interceptResume(data: P, continuation: Continuation

) = false +private fun withDispatcher(completion: Continuation, dispatcher: ContinuationDispatcher?): Continuation { + val finalContinuationDispatcher = dispatcher ?: object : ContinuationDispatcher { + override fun dispatchResume(value: T, continuation: Continuation) = false - override fun interceptResumeWithException(exception: Throwable, continuation: Continuation<*>) = false + override fun dispatchResumeWithException(exception: Throwable, continuation: Continuation<*>) = false } - return object : Continuation by resultHandler, ResumeInterceptor by finalResumeInterceptor {} + return object : Continuation by completion, ContinuationDispatcher by finalContinuationDispatcher {} } @JsName("CoroutineImpl") @@ -110,12 +91,12 @@ internal abstract class CoroutineImpl(private val resultContinuation: Continuati protected var result: Any? = null protected var exception: Throwable? = null protected var finallyPath: Array? = null - private val resumeInterceptor = resultContinuation as? ResumeInterceptor + private val continuationDispatcher = resultContinuation as? ContinuationDispatcher override fun resume(data: Any?) { - if (resumeInterceptor != null && (state and INTERCEPTING) == 0) { + if (continuationDispatcher != null && (state and INTERCEPTING) == 0) { state = state or INTERCEPTING - if (resumeInterceptor.interceptResume(data, this)) { + if (continuationDispatcher.dispatchResume(data, this)) { state = state and INTERCEPTING.inv() return } @@ -125,7 +106,7 @@ internal abstract class CoroutineImpl(private val resultContinuation: Continuati this.result = data try { val result = doResume() - if (result != SUSPENDED) { + if (result != CoroutineIntrinsics.SUSPENDED) { resultContinuation.resume(result) } } @@ -135,9 +116,9 @@ internal abstract class CoroutineImpl(private val resultContinuation: Continuati } override fun resumeWithException(exception: Throwable) { - if (resumeInterceptor != null && (state and INTERCEPTING) == 0) { + if (continuationDispatcher != null && (state and INTERCEPTING) == 0) { state = state or INTERCEPTING - if (resumeInterceptor.interceptResumeWithException(exception, this)) { + if (continuationDispatcher.dispatchResumeWithException(exception, this)) { state = state and INTERCEPTING.inv() return } @@ -147,7 +128,7 @@ internal abstract class CoroutineImpl(private val resultContinuation: Continuati this.exception = exception try { val result = doResume() - if (result != SUSPENDED) { + if (result != CoroutineIntrinsics.SUSPENDED) { resultContinuation.resume(result) } } diff --git a/libraries/stdlib/src/kotlin/coroutines/CoroutinesLibrary.kt b/libraries/stdlib/src/kotlin/coroutines/CoroutinesLibrary.kt new file mode 100644 index 00000000000..ee92f0b1ad0 --- /dev/null +++ b/libraries/stdlib/src/kotlin/coroutines/CoroutinesLibrary.kt @@ -0,0 +1,155 @@ +@file:kotlin.jvm.JvmName("CoroutinesLibraryKt") +@file:kotlin.jvm.JvmVersion +package kotlin.coroutines + +import java.util.concurrent.atomic.AtomicReferenceFieldUpdater +import java.lang.IllegalStateException +import kotlin.coroutines.CoroutineIntrinsics.SUSPENDED + +/** + * Creates 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 of exception. + * An optional [dispatcher] may be specified to customise dispatch of continuations between suspension points inside the coroutine. + */ +@SinceKotlin("1.1") +@Suppress("UNCHECKED_CAST") +public fun (suspend R.() -> T).createCoroutine( + receiver: R, + completion: Continuation, + dispatcher: ContinuationDispatcher? = null +): Continuation = (this as (R, Continuation) -> Continuation).invoke(receiver, withDispatcher(completion, dispatcher)) + +/** + * Starts coroutine with receiver type [R] and result type [T]. + * This function creates and start a new, fresh instance of suspendable computation every time it is invoked. + * The [completion] continuation is invoked when coroutine completes with result of exception. + * An optional [dispatcher] may be specified to customise dispatch of continuations between suspension points inside the coroutine. + */ +@SinceKotlin("1.1") +@Suppress("UNCHECKED_CAST") +public fun (suspend R.() -> T).startCoroutine( + receiver: R, + completion: Continuation, + dispatcher: ContinuationDispatcher? = null +) { + createCoroutine(receiver, completion, dispatcher).resume(Unit) +} + +/** + * Creates 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 of exception. + * An optional [dispatcher] may be specified to customise dispatch of continuations between suspension points inside the coroutine. + */ +@SinceKotlin("1.1") +@Suppress("UNCHECKED_CAST") +public fun (suspend () -> T).createCoroutine( + completion: Continuation, + dispatcher: ContinuationDispatcher? = null +): Continuation = (this as (Continuation) -> Continuation).invoke(withDispatcher(completion, dispatcher)) + +/** + * Starts coroutine without receiver and with result type [T]. + * This function creates and start a new, fresh instance of suspendable computation every time it is invoked. + * The [completion] continuation is invoked when coroutine completes with result of exception. + * An optional [dispatcher] may be specified to customise dispatch of continuations between suspension points inside the coroutine. + */ +@SinceKotlin("1.1") +@Suppress("UNCHECKED_CAST") +public fun (suspend () -> T).startCoroutine( + completion: Continuation, + dispatcher: ContinuationDispatcher? = null +) { + createCoroutine(completion, dispatcher).resume(Unit) +} + +/** + * Obtains the current continuation instance inside suspend functions and suspends + * currently running coroutine. + * + * In this function both [Continuation.resume] and [Continuation.resumeWithException] can be used either synchronously in + * the same stack-frame where suspension function is run or asynchronously later in the same thread or + * from a different thread of execution. Repeated invocation of any resume function produces [IllegalStateException]. + */ +@SinceKotlin("1.1") +public inline suspend fun suspendCoroutine(crossinline block: (Continuation) -> Unit): T = + CoroutineIntrinsics.suspendCoroutineOrReturn { c: Continuation -> + val safe = SafeContinuation(c) + block(safe) + safe.getResult() + } + +// INTERNAL DECLARATIONS + +@Suppress("INVISIBLE_REFERENCE", "INVISIBLE_MEMBER", "CANNOT_OVERRIDE_INVISIBLE_MEMBER") +private fun withDispatcher(completion: Continuation, dispatcher: ContinuationDispatcher?): Continuation { + return if (dispatcher == null) completion else + object : kotlin.jvm.internal.DispatchedContinuation, Continuation by completion { + override val dispatcher: ContinuationDispatcher? = dispatcher + } +} + +private val UNDECIDED: Any? = Any() +private val RESUMED: Any? = Any() +private class Fail(val exception: Throwable) + +@PublishedApi +internal class SafeContinuation @PublishedApi internal constructor(private val delegate: Continuation) : Continuation { + @Volatile + private var result: Any? = UNDECIDED + + companion object { + @Suppress("UNCHECKED_CAST") + @JvmStatic + private val RESULT_UPDATER = AtomicReferenceFieldUpdater.newUpdater, Any?>( + SafeContinuation::class.java, Any::class.java as Class, "result") + } + + private fun cas(expect: Any?, update: Any?): Boolean = + RESULT_UPDATER.compareAndSet(this, expect, update) + + override fun resume(value: T) { + while (true) { // lock-free loop + val result = this.result // atomic read + when (result) { + UNDECIDED -> if (cas(UNDECIDED, value)) return + SUSPENDED -> if (cas(SUSPENDED, RESUMED)) { + delegate.resume(value) + return + } + else -> throw IllegalStateException("Already resumed") + } + } + } + + override fun resumeWithException(exception: Throwable) { + while (true) { // lock-free loop + val result = this.result // atomic read + when (result) { + UNDECIDED -> if (cas(UNDECIDED, Fail(exception))) return + SUSPENDED -> if (cas(SUSPENDED, RESUMED)) { + delegate.resumeWithException(exception) + return + } + else -> throw IllegalStateException("Already resumed") + } + } + } + + @PublishedApi + internal fun getResult(): Any? { + var result = this.result // atomic read + if (result == UNDECIDED) { + if (cas(UNDECIDED, SUSPENDED)) return SUSPENDED + result = this.result // reread volatile var + } + when (result) { + RESUMED -> return SUSPENDED // already called continuation, indicate SUSPENDED upstream + is Fail -> throw result.exception + else -> return result // either SUSPENDED or data + } + } +}