diff --git a/js/js.libraries/src/core/coroutines.kt b/js/js.libraries/src/core/coroutines.kt index 637f22d7d92..0eda0d91e2a 100644 --- a/js/js.libraries/src/core/coroutines.kt +++ b/js/js.libraries/src/core/coroutines.kt @@ -34,19 +34,6 @@ public fun (suspend R.() -> T).createCoroutine( COROUTINE_SUSPENDED ) -/** - * 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. - */ -@SinceKotlin("1.1") -public fun (suspend R.() -> T).startCoroutine( - receiver: R, - completion: Continuation -) { - createCoroutine(receiver, completion).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. @@ -62,33 +49,6 @@ public fun (suspend () -> T).createCoroutine( COROUTINE_SUSPENDED ) -/** - * 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. - */ -@SinceKotlin("1.1") -public fun (suspend () -> T).startCoroutine( - completion: Continuation -) { - createCoroutine(completion).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 suspend fun suspendCoroutine(block: (Continuation) -> Unit): T = suspendCoroutineOrReturn { c -> - val safe = SafeContinuation(c) - block(safe) - safe.getResult() -} - // ------- internal stuff ------- @JsName("CoroutineImpl") @@ -135,11 +95,16 @@ private val UNDECIDED: Any? = Any() private val RESUMED: Any? = Any() private class Fail(val exception: Throwable) +@PublishedApi internal class SafeContinuation -@PublishedApi internal constructor( +internal constructor( private val delegate: Continuation, - initialResult: Any? = UNDECIDED + initialResult: Any? ) : Continuation { + + @PublishedApi + internal constructor(delegate: Continuation) : this(delegate, UNDECIDED) + public override val context: CoroutineContext get() = delegate.context @@ -175,6 +140,7 @@ internal class SafeContinuation } } + @PublishedApi internal fun getResult(): Any? { if (result === UNDECIDED) { result = COROUTINE_SUSPENDED diff --git a/libraries/stdlib/common/src/kotlin/CoroutinesH.kt b/libraries/stdlib/common/src/kotlin/CoroutinesH.kt index 79b7b65608d..de81587163c 100644 --- a/libraries/stdlib/common/src/kotlin/CoroutinesH.kt +++ b/libraries/stdlib/common/src/kotlin/CoroutinesH.kt @@ -14,18 +14,6 @@ public header fun (suspend R.() -> T).createCoroutine( completion: Continuation ): Continuation -/** - * 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") -public header fun (suspend R.() -> T).startCoroutine( - receiver: R, - completion: Continuation -) - /** * Creates coroutine without receiver and with result type [T]. * This function creates a new, fresh instance of suspendable computation every time it is invoked. @@ -38,25 +26,17 @@ public header fun (suspend () -> T).createCoroutine( completion: Continuation ): Continuation -/** - * 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 header fun (suspend () -> T).startCoroutine( - completion: Continuation -) +@PublishedApi +internal header class SafeContinuation : Continuation { + internal constructor(delegate: Continuation, initialResult: Any?) -/** - * 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 header inline suspend fun suspendCoroutine(crossinline block: (Continuation) -> Unit): T + @PublishedApi + internal constructor(delegate: Continuation) + + @PublishedApi + internal fun getResult(): Any? + + override val context: CoroutineContext + override fun resume(value: T): Unit + override fun resumeWithException(exception: Throwable): Unit +} diff --git a/libraries/stdlib/src/kotlin/coroutines/experimental/CoroutinesLibrary.kt b/libraries/stdlib/src/kotlin/coroutines/experimental/CoroutinesLibrary.kt index 1ecd680f07d..31b2b1c70e3 100644 --- a/libraries/stdlib/src/kotlin/coroutines/experimental/CoroutinesLibrary.kt +++ b/libraries/stdlib/src/kotlin/coroutines/experimental/CoroutinesLibrary.kt @@ -15,37 +15,10 @@ */ @file:kotlin.jvm.JvmName("CoroutinesKt") -@file:kotlin.jvm.JvmVersion package kotlin.coroutines.experimental -import java.lang.IllegalStateException -import java.util.concurrent.atomic.AtomicReferenceFieldUpdater import kotlin.coroutines.experimental.intrinsics.COROUTINE_SUSPENDED import kotlin.coroutines.experimental.intrinsics.suspendCoroutineOrReturn -import kotlin.coroutines.experimental.jvm.internal.CoroutineImpl -import kotlin.coroutines.experimental.jvm.internal.interceptContinuationIfNeeded - -/** - * 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 or exception. - */ -@SinceKotlin("1.1") -@Suppress("UNCHECKED_CAST") -public fun (suspend R.() -> T).createCoroutine( - receiver: R, - completion: Continuation -): Continuation = - SafeContinuation( - if (this !is CoroutineImpl) - buildContinuationByInvokeCall(completion) { - (this as Function2, Any?>).invoke(receiver, completion) - } - else - ((this as CoroutineImpl).create(receiver, completion) as CoroutineImpl).facade, - COROUTINE_SUSPENDED - ) /** * Starts coroutine with receiver type [R] and result type [T]. @@ -61,27 +34,6 @@ public fun (suspend R.() -> T).startCoroutine( createCoroutine(receiver, completion).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 or exception. - */ -@SinceKotlin("1.1") -@Suppress("UNCHECKED_CAST") -public fun (suspend () -> T).createCoroutine( - completion: Continuation -): Continuation = - SafeContinuation( - if (this !is CoroutineImpl) - buildContinuationByInvokeCall(completion) { - (this as Function1, Any?>).invoke(completion) - } - else - ((this as CoroutineImpl).create(completion) as CoroutineImpl).facade, - COROUTINE_SUSPENDED - ) - /** * 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. @@ -113,28 +65,7 @@ public inline suspend fun suspendCoroutine(crossinline block: (Continuation< // INTERNAL DECLARATIONS -private inline fun buildContinuationByInvokeCall( - completion: Continuation, - crossinline block: () -> Any? -): Continuation { - val continuation = - object : Continuation { - override val context: CoroutineContext - get() = completion.context - - override fun resume(value: Unit) { - processInvokeCallOnCoroutine(completion, block) - } - - override fun resumeWithException(exception: Throwable) { - completion.resumeWithException(exception) - } - } - - return completion.context.interceptContinuationIfNeeded(continuation) -} - -private inline fun processInvokeCallOnCoroutine(completion: Continuation<*>, block: () -> Any?) { +internal inline fun processInvokeCallOnCoroutine(completion: Continuation<*>, block: () -> Any?) { try { val result = block() if (result !== COROUTINE_SUSPENDED) { @@ -145,69 +76,3 @@ private inline fun processInvokeCallOnCoroutine(completion: Continuation<*>, blo completion.resumeWithException(t) } } - -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, - initialResult: Any? = UNDECIDED -) : Continuation { - public override val context: CoroutineContext - get() = delegate.context - - @Volatile - private var result: Any? = initialResult - - companion object { - @Suppress("UNCHECKED_CAST") - @JvmStatic - private val RESULT = AtomicReferenceFieldUpdater.newUpdater, Any?>( - SafeContinuation::class.java, Any::class.java as Class, "result") - } - - override fun resume(value: T) { - while (true) { // lock-free loop - val result = this.result // atomic read - when { - result === UNDECIDED -> if (RESULT.compareAndSet(this, UNDECIDED, value)) return - result === COROUTINE_SUSPENDED -> if (RESULT.compareAndSet(this, COROUTINE_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 (RESULT.compareAndSet(this, UNDECIDED, Fail(exception))) return - result === COROUTINE_SUSPENDED -> if (RESULT.compareAndSet(this, COROUTINE_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 (RESULT.compareAndSet(this, UNDECIDED, COROUTINE_SUSPENDED)) return COROUTINE_SUSPENDED - result = this.result // reread volatile var - } - when { - result === RESUMED -> return COROUTINE_SUSPENDED // already called continuation, indicate COROUTINE_SUSPENDED upstream - result is Fail -> throw result.exception - else -> return result // either COROUTINE_SUSPENDED or data - } - } -} diff --git a/libraries/stdlib/src/kotlin/coroutines/experimental/CoroutinesLibraryJvm.kt b/libraries/stdlib/src/kotlin/coroutines/experimental/CoroutinesLibraryJvm.kt new file mode 100644 index 00000000000..42e4b23a081 --- /dev/null +++ b/libraries/stdlib/src/kotlin/coroutines/experimental/CoroutinesLibraryJvm.kt @@ -0,0 +1,90 @@ +/* + * Copyright 2010-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. + */ + +@file:kotlin.jvm.JvmName("CoroutinesJvmKt") +@file:kotlin.jvm.JvmVersion +package kotlin.coroutines.experimental + +import kotlin.coroutines.experimental.intrinsics.COROUTINE_SUSPENDED +import kotlin.coroutines.experimental.jvm.internal.CoroutineImpl +import kotlin.coroutines.experimental.jvm.internal.interceptContinuationIfNeeded + +/** + * 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 or exception. + */ +@SinceKotlin("1.1") +@Suppress("UNCHECKED_CAST") +public fun (suspend R.() -> T).createCoroutine( + receiver: R, + completion: Continuation +): Continuation = + SafeContinuation( + if (this !is CoroutineImpl) + buildContinuationByInvokeCall(completion) { + (this as Function2, Any?>).invoke(receiver, completion) + } + else + ((this as CoroutineImpl).create(receiver, completion) as CoroutineImpl).facade, + COROUTINE_SUSPENDED + ) + +/** + * 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 or exception. + */ +@SinceKotlin("1.1") +@Suppress("UNCHECKED_CAST") +public fun (suspend () -> T).createCoroutine( + completion: Continuation +): Continuation = + SafeContinuation( + if (this !is CoroutineImpl) + buildContinuationByInvokeCall(completion) { + (this as Function1, Any?>).invoke(completion) + } + else + ((this as CoroutineImpl).create(completion) as CoroutineImpl).facade, + COROUTINE_SUSPENDED + ) + +// INTERNAL DECLARATIONS + +private inline fun buildContinuationByInvokeCall( + completion: Continuation, + crossinline block: () -> Any? +): Continuation { + val continuation = + object : Continuation { + override val context: CoroutineContext + get() = completion.context + + override fun resume(value: Unit) { + processInvokeCallOnCoroutine(completion, block) + } + + override fun resumeWithException(exception: Throwable) { + completion.resumeWithException(exception) + } + } + + return completion.context.interceptContinuationIfNeeded(continuation) +} + diff --git a/libraries/stdlib/src/kotlin/coroutines/experimental/SafeContinuationJvm.kt b/libraries/stdlib/src/kotlin/coroutines/experimental/SafeContinuationJvm.kt new file mode 100644 index 00000000000..ac673ce8e00 --- /dev/null +++ b/libraries/stdlib/src/kotlin/coroutines/experimental/SafeContinuationJvm.kt @@ -0,0 +1,91 @@ +/* + * Copyright 2010-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. + */ +@file:kotlin.jvm.JvmVersion +package kotlin.coroutines.experimental + +import java.util.concurrent.atomic.AtomicReferenceFieldUpdater +import kotlin.coroutines.experimental.intrinsics.COROUTINE_SUSPENDED + +@PublishedApi +internal class SafeContinuation +internal constructor( + private val delegate: Continuation, + initialResult: Any? +) : Continuation { + + @PublishedApi + internal constructor(delegate: Continuation) : this(delegate, UNDECIDED) + + public override val context: CoroutineContext + get() = delegate.context + + @Volatile + private var result: Any? = initialResult + + companion object { + private val UNDECIDED: Any? = Any() + private val RESUMED: Any? = Any() + + @Suppress("UNCHECKED_CAST") + @JvmStatic + private val RESULT = AtomicReferenceFieldUpdater.newUpdater, Any?>( + SafeContinuation::class.java, Any::class.java as Class, "result") + } + + private class Fail(val exception: Throwable) + + override fun resume(value: T) { + while (true) { // lock-free loop + val result = this.result // atomic read + when { + result === UNDECIDED -> if (RESULT.compareAndSet(this, UNDECIDED, value)) return + result === COROUTINE_SUSPENDED -> if (RESULT.compareAndSet(this, COROUTINE_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 (RESULT.compareAndSet(this, UNDECIDED, Fail(exception))) return + result === COROUTINE_SUSPENDED -> if (RESULT.compareAndSet(this, COROUTINE_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 (RESULT.compareAndSet(this, UNDECIDED, COROUTINE_SUSPENDED)) return COROUTINE_SUSPENDED + result = this.result // reread volatile var + } + when { + result === RESUMED -> return COROUTINE_SUSPENDED // already called continuation, indicate COROUTINE_SUSPENDED upstream + result is Fail -> throw result.exception + else -> return result // either COROUTINE_SUSPENDED or data + } + } +}