From 852a5ee06465fa3a1268cee5c38646ef581ac98b Mon Sep 17 00:00:00 2001 From: Roman Elizarov Date: Thu, 8 Dec 2016 21:59:45 +0300 Subject: [PATCH] New coroutine builder convention API --- core/builtins/native/kotlin/Coroutines.kt | 26 ++- .../src/kotlin/coroutines/Coroutines.kt | 15 +- .../src/kotlin/coroutines/Builders.kt | 149 ++++++++++++++ .../kotlin/coroutines/ContinuationCapture.kt | 113 +++++++++++ .../src/kotlin/coroutines/Suspendable.kt | 33 ++++ .../src/kotlin/jvm/internal/CoroutineImpl.kt | 184 ++++++++++++++++-- 6 files changed, 492 insertions(+), 28 deletions(-) create mode 100644 core/runtime.jvm/src/kotlin/coroutines/Builders.kt create mode 100644 core/runtime.jvm/src/kotlin/coroutines/ContinuationCapture.kt create mode 100644 core/runtime.jvm/src/kotlin/coroutines/Suspendable.kt diff --git a/core/builtins/native/kotlin/Coroutines.kt b/core/builtins/native/kotlin/Coroutines.kt index abf1dbebd5b..51d3feab3cc 100644 --- a/core/builtins/native/kotlin/Coroutines.kt +++ b/core/builtins/native/kotlin/Coroutines.kt @@ -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 suspendWithCurrentContinuation(body: (Continuation) -> Any?): T +public inline suspend fun maySuspendWithCurrentContinuation(body: (Continuation) -> Any?): T diff --git a/core/builtins/src/kotlin/coroutines/Coroutines.kt b/core/builtins/src/kotlin/coroutines/Coroutines.kt index 3f25cf0350f..9a77bf9d15a 100644 --- a/core/builtins/src/kotlin/coroutines/Coroutines.kt +++ b/core/builtins/src/kotlin/coroutines/Coroutines.kt @@ -43,9 +43,16 @@ public interface Continuation { 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 \ No newline at end of file diff --git a/core/runtime.jvm/src/kotlin/coroutines/Builders.kt b/core/runtime.jvm/src/kotlin/coroutines/Builders.kt new file mode 100644 index 00000000000..5cc6211313c --- /dev/null +++ b/core/runtime.jvm/src/kotlin/coroutines/Builders.kt @@ -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

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 +): Continuation { + // 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 { + 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 (/*suspend*/ R.() -> T).startCoroutine( + receiver: R, + resultHandler: Continuation +) { + try { + val result = (this as Function2, 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 (/*suspend*/ () -> T).createCoroutine( + resultHandler: Continuation, + resumeInterceptor: ResumeInterceptor? = null +): Continuation { + // 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 { + 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 (/*suspend*/ () -> T).startCoroutine( + resultHandler: Continuation, + resumeInterceptor: ResumeInterceptor? = null +) { + try { + val result = (this as Function1, 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

: 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 new file mode 100644 index 00000000000..089ae40d4f0 --- /dev/null +++ b/core/runtime.jvm/src/kotlin/coroutines/ContinuationCapture.kt @@ -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 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(body: (Continuation) -> Unit): T = + maySuspendWithCurrentContinuation { 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) + +@Suppress("UNCHECKED_CAST") +private val RESULT_UPDATER = AtomicReferenceFieldUpdater.newUpdater, Any?>( + SafeContinuation::class.java, Any::class.java as Class, "result") + +internal class SafeContinuation @PublishedApi constructor(private val delegate: Continuation) : Continuation { + @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 + } + } +} + diff --git a/core/runtime.jvm/src/kotlin/coroutines/Suspendable.kt b/core/runtime.jvm/src/kotlin/coroutines/Suspendable.kt new file mode 100644 index 00000000000..7966f498966 --- /dev/null +++ b/core/runtime.jvm/src/kotlin/coroutines/Suspendable.kt @@ -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 { + * 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 aa48c58249c..9e10ee745cd 100644 --- a/core/runtime.jvm/src/kotlin/jvm/internal/CoroutineImpl.kt +++ b/core/runtime.jvm/src/kotlin/jvm/internal/CoroutineImpl.kt @@ -16,27 +16,183 @@ package kotlin.jvm.internal -abstract class CoroutineImpl(arity: Int) : Lambda(arity), Continuation { - @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 { + 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?) : 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 { + @JvmField + protected val resultContinuation: 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 + @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?) : 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 + + 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, 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 + } + } +} +--------------------------------------------------------------------------------------------- +*/ \ No newline at end of file