New coroutine builder convention API

This commit is contained in:
Roman Elizarov
2016-12-08 21:59:45 +03:00
committed by Stanislav Erokhin
parent b4af532cf0
commit 852a5ee064
6 changed files with 492 additions and 28 deletions
+16 -10
View File
@@ -17,18 +17,24 @@
package kotlin.coroutines
/**
* This function allows to obtain a continuation instance inside the suspend functions.
* As a suspend function may be only tail-called inside another suspend function, this call will be used as a return value of the latter one.
* 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 [Suspend] object, that means that suspend-function cannot return a value immediately,
* i.e. it's literally suspends continuation, and the continuation will be resumed at some moment in the future.
* 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 return value must have a type assignable to 'T'.
* As the latter part cannot be correctly typechecked, it remains on the conscience of the suspend function's author.
*
* Note that it's not recommended to call a [Continuation] method in the same stackframe where suspension function is run,
* they should be called asynchronously later (probably from the different thread).
* 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 <T> suspendWithCurrentContinuation(body: (Continuation<T>) -> Any?): T
public inline suspend fun <T> maySuspendWithCurrentContinuation(body: (Continuation<T>) -> Any?): T
@@ -43,9 +43,16 @@ public interface Continuation<in P> {
public annotation class AllowSuspendExtensions
/**
* This object can be used as a return value of [kotlin.coroutines.suspendWithCurrentContinuation] lambda-argument to state that
* the continuation will be resumed at some moment in the future, that means that suspend-function cannot return a value immediately,
* i.e. it's literally suspends continuation, so no stack-unwinding will be performed by the continuation.
* 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.
*/
@SinceKotlin("1.1")
public object Suspend
public val SUSPENDED: Any? = Any()
/**
* 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
* receiver only and are restricted from calling arbitrary suspension functions.
*/
@SinceKotlin("1.1")
public annotation class RestrictSuspension
@@ -0,0 +1,149 @@
/*
* 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 kotlin.jvm.internal.CoroutineImpl
import kotlin.jvm.internal.RestrictedCoroutineImpl
/**
* 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 <P> interceptResume(data: P, continuation: Continuation<P>): 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 <R, T> (/*suspend*/ R.() -> T).createCoroutine(
receiver: R,
resultHandler: Continuation<T>
): Continuation<Unit> {
// check if the RestrictedCoroutineImpl was passed and do efficient creation
if (this is RestrictedCoroutineImpl)
return doCreateInternal(receiver, resultHandler)
// otherwise, it is just a callable reference to some suspend function
return object : Continuation<Unit> {
override fun resume(data: Unit) {
startCoroutine(receiver, resultHandler)
}
override fun resumeWithException(exception: Throwable) {
resultHandler.resumeWithException(exception)
}
}
}
/**
* 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 <R, T> (/*suspend*/ R.() -> T).startCoroutine(
receiver: R,
resultHandler: Continuation<T>
) {
try {
val result = (this as Function2<R, Continuation<T>, Any?>).invoke(receiver, resultHandler)
if (result == SUSPENDED) return
resultHandler.resume(result as T)
} catch (e: Throwable) {
resultHandler.resumeWithException(e)
}
}
/**
* 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 <T> (/*suspend*/ () -> T).createCoroutine(
resultHandler: Continuation<T>,
resumeInterceptor: ResumeInterceptor? = null
): Continuation<Unit> {
// check if the CoroutineImpl was passed and do efficient creation
if (this is CoroutineImpl)
return doCreateInternal(null, withInterceptor(resultHandler, resumeInterceptor))
// otherwise, it is just a callable reference to some suspend function
return object : Continuation<Unit> {
override fun resume(data: Unit) {
startCoroutine(resultHandler, resumeInterceptor)
}
override fun resumeWithException(exception: Throwable) {
resultHandler.resumeWithException(exception)
}
}
}
/**
* 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 <T> (/*suspend*/ () -> T).startCoroutine(
resultHandler: Continuation<T>,
resumeInterceptor: ResumeInterceptor? = null
) {
try {
val result = (this as Function1<Continuation<T>, Any?>).invoke(withInterceptor(resultHandler, resumeInterceptor))
if (result == SUSPENDED) return
resultHandler.resume(result as T)
} catch (e: Throwable) {
resultHandler.resumeWithException(e)
}
}
// ------- internal stuff -------
internal interface InterceptableContinuation<P> : Continuation<P> {
val resumeInterceptor: ResumeInterceptor?
}
private fun <T> withInterceptor(resultHandler: Continuation<T>, resumeInterceptor: ResumeInterceptor?): Continuation<T> {
return if (resumeInterceptor == null) resultHandler else
object : InterceptableContinuation<T>, Continuation<T> by resultHandler {
override val resumeInterceptor: ResumeInterceptor? = resumeInterceptor
}
}
@@ -0,0 +1,113 @@
/*
* 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.util.concurrent.atomic.AtomicReferenceFieldUpdater
import kotlin.IllegalStateException
/**
* 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 <T> suspendWithCurrentContinuation(crossinline body: (Continuation<T>) -> Unit): T =
maySuspendWithCurrentContinuation<T> { c: Continuation<T> ->
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 <T> runWithCurrentContinuation(body: (Continuation<T>) -> Unit): T =
maySuspendWithCurrentContinuation<T> { c: Continuation<T> ->
val safe = SafeContinuation(c)
body(safe)
safe.getResult()
}
private val UNDECIDED: Any? = Any()
private val RESUMED: Any? = Any()
private class Fail(val exception: Throwable)
@Suppress("UNCHECKED_CAST")
private val RESULT_UPDATER = AtomicReferenceFieldUpdater.newUpdater<SafeContinuation<*>, Any?>(
SafeContinuation::class.java, Any::class.java as Class<Any?>, "result")
internal class SafeContinuation<T> @PublishedApi constructor(private val delegate: Continuation<T>) : Continuation<T> {
@Volatile
private var result: Any? = UNDECIDED
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")
}
}
}
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
}
}
}
@@ -0,0 +1,33 @@
/*
* 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<R> {
* await(...) // calls one suspend function
* await(...) // calls another suspend function
* }
*/
@SinceKotlin("1.1")
suspend fun <T> suspendable(/*suspend*/ lambda: () -> T): T = suspendWithCurrentContinuation<T> { c ->
@Suppress("UNCHECKED_CAST")
(lambda as Function1<Continuation<T>, Any?>).invoke(c)
}
@@ -16,27 +16,183 @@
package kotlin.jvm.internal
abstract class CoroutineImpl(arity: Int) : Lambda(arity), Continuation<Any?> {
@JvmField
protected var _controller: Any? = null
import kotlin.coroutines.*
// It's not protected because can be used from noinline lambdas inside coroutine (when calling non-suspend functions)
// Also there might be needed a way to access a controller by Continuation instance when it's inherited from CoroutineImpl
val controller: Any? get() = _controller
private const val INTERCEPT_BIT_SET = 1 shl 31
private const val INTERCEPT_BIT_CLEAR = INTERCEPT_BIT_SET.inv()
// Any label state less then zero indicates that coroutine is not run and can't be resumed in any way.
// Specific values do not matter by now, but currently -2 used for uninitialized coroutine (no controller is assigned),
// and -1 will mean that coroutine execution is over (does not work yet).
@JvmField
protected var label: Int = -2
@SinceKotlin("1.1")
abstract class CoroutineImpl : RestrictedCoroutineImpl, InterceptableContinuation<Any?> {
private val _resumeInterceptor: ResumeInterceptor?
override val resumeInterceptor: ResumeInterceptor?
get() = _resumeInterceptor
// this constructor is used to create initial "factory" lambda object
constructor(arity: Int) : super(arity) {
_resumeInterceptor = null
}
// this constructor is used to create a continuation instance for coroutine
constructor(arity: Int, resultContinuation: Continuation<Any?>?) : super(arity, resultContinuation) {
_resumeInterceptor = (resultContinuation as? InterceptableContinuation<*>)?.resumeInterceptor
}
// coroutine factory implementation for unrestricted coroutines, it will implement Function1.invoke
// in the actual coroutine implementation
fun invoke(resultContinuation: Continuation<*>): Any? {
// create and run it until first suspension
return (doCreate(null, resultContinuation) as CoroutineImpl).doResume(Unit, null)
}
override fun resume(data: Any?) {
doResume(data, null)
if (_resumeInterceptor != null) {
if (label and INTERCEPT_BIT_SET == 0) {
label = label or INTERCEPT_BIT_SET
if (_resumeInterceptor.interceptResume(data, this)) return
}
label = label and INTERCEPT_BIT_CLEAR
}
super.resume(data)
}
override fun resumeWithException(exception: Throwable) {
doResume(null, exception)
if (_resumeInterceptor != null) {
if (label and INTERCEPT_BIT_SET == 0) {
label = label or INTERCEPT_BIT_SET
if (_resumeInterceptor.interceptResumeWithException(exception, this)) return
}
label = label and INTERCEPT_BIT_CLEAR
}
super.resumeWithException(exception)
}
}
@SinceKotlin("1.1")
abstract class RestrictedCoroutineImpl : Lambda, Continuation<Any?> {
@JvmField
protected val resultContinuation: Continuation<Any?>?
// 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
@JvmField
protected var label: Int
// this constructor is used to create initial "factory" lambda object
constructor(arity: Int) : super(arity) {
resultContinuation = null
label = -1 // don't use this object as coroutine
}
protected abstract fun doResume(data: Any?, exception: Throwable?)
// this constructor is used to create a continuation instance for coroutine
constructor(arity: Int, resultContinuation: Continuation<Any?>?) : super(arity) {
this.resultContinuation = resultContinuation
label = 0
}
// coroutine factory implementation for restricted coroutines, it will implement Function2.invoke
// in the actual restricted coroutine implementation
fun invoke(receiver: Any, resultContinuation: Continuation<*>): Any? {
// create and run it until first suspension
return (doCreate(receiver, resultContinuation) as RestrictedCoroutineImpl).doResume(Unit, null)
}
protected abstract fun doCreate(receiver: Any?, resultContinuation: Continuation<*>): Continuation<Unit>
internal fun doCreateInternal(receiver: Any?, resultContinuation: Continuation<*>) =
doCreate(receiver, resultContinuation)
override fun resume(data: Any?) {
try {
val result = doResume(data, null)
if (result != SUSPENDED)
resultContinuation!!.resume(result)
} catch (e: Throwable) {
resultContinuation!!.resumeWithException(e)
}
}
override fun resumeWithException(exception: Throwable) {
try {
val result = doResume(null, exception)
if (result != SUSPENDED)
resultContinuation!!.resume(result)
} catch (e: Throwable) {
resultContinuation!!.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
}
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<Receiver, Continuation<*>, 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<Any?>) : 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<Unit> {
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
}
}
}
---------------------------------------------------------------------------------------------
*/