Improve coroutines library layout

- Split CoroutinesLibrary into common and JVM parts
- Get rid of startCoroutine duplications
- Make suspendCoroutine implementation to be platform independent
This commit is contained in:
Denis Zharkov
2017-01-31 14:47:05 +03:00
parent 258ee0db75
commit a65d86293b
5 changed files with 203 additions and 211 deletions
@@ -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 <R, T> (suspend R.() -> T).createCoroutine(
receiver: R,
completion: Continuation<T>
): Continuation<Unit> =
SafeContinuation(
if (this !is CoroutineImpl)
buildContinuationByInvokeCall(completion) {
(this as Function2<R, Continuation<T>, 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 <R, T> (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 <T> (suspend () -> T).createCoroutine(
completion: Continuation<T>
): Continuation<Unit> =
SafeContinuation(
if (this !is CoroutineImpl)
buildContinuationByInvokeCall(completion) {
(this as Function1<Continuation<T>, 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 <T> suspendCoroutine(crossinline block: (Continuation<
// INTERNAL DECLARATIONS
private inline fun <T> buildContinuationByInvokeCall(
completion: Continuation<T>,
crossinline block: () -> Any?
): Continuation<Unit> {
val continuation =
object : Continuation<Unit> {
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<in T>
@PublishedApi internal constructor(
private val delegate: Continuation<T>,
initialResult: Any? = UNDECIDED
) : Continuation<T> {
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<SafeContinuation<*>, Any?>(
SafeContinuation::class.java, Any::class.java as Class<Any?>, "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
}
}
}
@@ -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 <R, T> (suspend R.() -> T).createCoroutine(
receiver: R,
completion: Continuation<T>
): Continuation<Unit> =
SafeContinuation(
if (this !is CoroutineImpl)
buildContinuationByInvokeCall(completion) {
(this as Function2<R, Continuation<T>, 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 <T> (suspend () -> T).createCoroutine(
completion: Continuation<T>
): Continuation<Unit> =
SafeContinuation(
if (this !is CoroutineImpl)
buildContinuationByInvokeCall(completion) {
(this as Function1<Continuation<T>, Any?>).invoke(completion)
}
else
((this as CoroutineImpl).create(completion) as CoroutineImpl).facade,
COROUTINE_SUSPENDED
)
// INTERNAL DECLARATIONS
private inline fun <T> buildContinuationByInvokeCall(
completion: Continuation<T>,
crossinline block: () -> Any?
): Continuation<Unit> {
val continuation =
object : Continuation<Unit> {
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)
}
@@ -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<in T>
internal constructor(
private val delegate: Continuation<T>,
initialResult: Any?
) : Continuation<T> {
@PublishedApi
internal constructor(delegate: Continuation<T>) : 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<SafeContinuation<*>, Any?>(
SafeContinuation::class.java, Any::class.java as Class<Any?>, "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
}
}
}