From 72401efb9f4704a5e109b657476c4a5a9674143e Mon Sep 17 00:00:00 2001 From: Denis Zharkov Date: Fri, 9 Dec 2016 13:04:22 +0300 Subject: [PATCH] Update coroutine related runtime parts --- core/builtins/native/kotlin/Coroutines.kt | 2 +- .../src/kotlin/coroutines/Coroutines.kt | 7 +- .../src/kotlin/coroutines/Builders.kt | 65 ++++--------------- .../kotlin/coroutines/ContinuationCapture.kt | 24 +++---- .../src/kotlin/jvm/internal/CoroutineImpl.kt | 38 +---------- .../org/jetbrains/kotlin/utils/addToStdlib.kt | 5 +- 6 files changed, 39 insertions(+), 102 deletions(-) diff --git a/core/builtins/native/kotlin/Coroutines.kt b/core/builtins/native/kotlin/Coroutines.kt index 51d3feab3cc..c4b464fa075 100644 --- a/core/builtins/native/kotlin/Coroutines.kt +++ b/core/builtins/native/kotlin/Coroutines.kt @@ -37,4 +37,4 @@ package kotlin.coroutines * Use [runWithCurrentContinuation] as a safer way to obtain current continuation instance. */ @SinceKotlin("1.1") -public inline suspend fun maySuspendWithCurrentContinuation(body: (Continuation) -> Any?): T +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 9a77bf9d15a..8862e00248f 100644 --- a/core/builtins/src/kotlin/coroutines/Coroutines.kt +++ b/core/builtins/src/kotlin/coroutines/Coroutines.kt @@ -49,10 +49,15 @@ public annotation class AllowSuspendExtensions @SinceKotlin("1.1") public val SUSPENDED: Any? = Any() +@SinceKotlin("1.1") +@Target(AnnotationTarget.TYPE) +@Retention(AnnotationRetention.BINARY) +public annotation class Suspend + /** * 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 +public annotation class RestrictSuspension diff --git a/core/runtime.jvm/src/kotlin/coroutines/Builders.kt b/core/runtime.jvm/src/kotlin/coroutines/Builders.kt index 5cc6211313c..0e247efef61 100644 --- a/core/runtime.jvm/src/kotlin/coroutines/Builders.kt +++ b/core/runtime.jvm/src/kotlin/coroutines/Builders.kt @@ -16,9 +16,6 @@ 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 @@ -47,24 +44,11 @@ public interface ResumeInterceptor { */ @SinceKotlin("1.1") @Suppress("UNCHECKED_CAST") -public fun (/*suspend*/ R.() -> T).createCoroutine( +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) - } - } -} + 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]. @@ -73,17 +57,12 @@ public fun (/*suspend*/ R.() -> T).createCoroutine( */ @SinceKotlin("1.1") @Suppress("UNCHECKED_CAST") -public fun (/*suspend*/ R.() -> T).startCoroutine( +public fun (@Suspend() (R.() -> T)).startCoroutine( receiver: R, - resultHandler: Continuation + resultHandler: Continuation, + resumeInterceptor: ResumeInterceptor? = null ) { - 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) - } + createCoroutine(receiver, resultHandler, resumeInterceptor).resume(Unit) } /** @@ -95,24 +74,10 @@ public fun (/*suspend*/ R.() -> T).startCoroutine( */ @SinceKotlin("1.1") @Suppress("UNCHECKED_CAST") -public fun (/*suspend*/ () -> T).createCoroutine( +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) - } - } -} +): Continuation = (this as (Continuation) -> Continuation).invoke(withInterceptor(resultHandler, resumeInterceptor)) /** * Starts coroutine without receiver and with result type [T]. @@ -122,17 +87,11 @@ public fun (/*suspend*/ () -> T).createCoroutine( */ @SinceKotlin("1.1") @Suppress("UNCHECKED_CAST") -public fun (/*suspend*/ () -> T).startCoroutine( +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) - } + createCoroutine(resultHandler, resumeInterceptor).resume(Unit) } // ------- internal stuff ------- diff --git a/core/runtime.jvm/src/kotlin/coroutines/ContinuationCapture.kt b/core/runtime.jvm/src/kotlin/coroutines/ContinuationCapture.kt index 089ae40d4f0..32f67367a9d 100644 --- a/core/runtime.jvm/src/kotlin/coroutines/ContinuationCapture.kt +++ b/core/runtime.jvm/src/kotlin/coroutines/ContinuationCapture.kt @@ -16,8 +16,8 @@ package kotlin.coroutines +import java.lang.IllegalStateException import java.util.concurrent.atomic.AtomicReferenceFieldUpdater -import kotlin.IllegalStateException /** * This function allows to obtain the current continuation instance inside suspend functions and suspend @@ -32,12 +32,12 @@ import kotlin.IllegalStateException * 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 - } +//@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 @@ -50,8 +50,8 @@ public inline suspend fun suspendWithCurrentContinuation(crossinline body: ( * Repeated invocation of any resume function produces [IllegalStateException]. */ @SinceKotlin("1.1") -public inline suspend fun runWithCurrentContinuation(body: (Continuation) -> Unit): T = - maySuspendWithCurrentContinuation { c: Continuation -> +public inline suspend fun runWithCurrentContinuation(crossinline body: (Continuation) -> Unit): T = + suspendWithCurrentContinuation { c: Continuation -> val safe = SafeContinuation(c) body(safe) safe.getResult() @@ -65,7 +65,8 @@ private class Fail(val exception: Throwable) 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 { +@PublishedApi +internal class SafeContinuation @PublishedApi internal constructor(private val delegate: Continuation) : Continuation { @Volatile private var result: Any? = UNDECIDED @@ -100,7 +101,8 @@ internal class SafeContinuation @PublishedApi constructor(private val delegat } } - fun getResult(): Any? { + @PublishedApi + internal fun getResult(): Any? { val result = this.result // atomic read if (result == UNDECIDED && cas(UNDECIDED, SUSPENDED)) return SUSPENDED when (result) { diff --git a/core/runtime.jvm/src/kotlin/jvm/internal/CoroutineImpl.kt b/core/runtime.jvm/src/kotlin/jvm/internal/CoroutineImpl.kt index 9e10ee745cd..6a0fbbb8279 100644 --- a/core/runtime.jvm/src/kotlin/jvm/internal/CoroutineImpl.kt +++ b/core/runtime.jvm/src/kotlin/jvm/internal/CoroutineImpl.kt @@ -16,8 +16,6 @@ package kotlin.jvm.internal -import kotlin.coroutines.* - private const val INTERCEPT_BIT_SET = 1 shl 31 private const val INTERCEPT_BIT_CLEAR = INTERCEPT_BIT_SET.inv() @@ -28,23 +26,11 @@ abstract class CoroutineImpl : RestrictedCoroutineImpl, InterceptableContinuatio 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?) { if (_resumeInterceptor != null) { if (label and INTERCEPT_BIT_SET == 0) { @@ -71,37 +57,19 @@ abstract class CoroutineImpl : RestrictedCoroutineImpl, InterceptableContinuatio @SinceKotlin("1.1") abstract class RestrictedCoroutineImpl : Lambda, Continuation { @JvmField - protected val resultContinuation: Continuation? + protected var 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 - } - // this constructor is used to create a continuation instance for coroutine constructor(arity: Int, resultContinuation: Continuation?) : super(arity) { this.resultContinuation = resultContinuation - label = 0 + label = if (resultContinuation != null) 0 else -1 } - // 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) @@ -195,4 +163,4 @@ class XXXCoroutine : RestrictedCoroutineImpl, Function2 Pair.swap(): Pair = Pair(second, first) fun T.check(predicate: (T) -> Boolean): T? = if (predicate(this)) this else null +inline fun Any?.safeAs(): T? = this as? T +inline fun Any?.cast(): T = this as T + fun constant(calculator: () -> T): T { val cached = constantMap[calculator] @Suppress("UNCHECKED_CAST") @@ -124,4 +127,4 @@ inline fun Iterable.sumByLong(selector: (T) -> Long): Long { inline fun , O> C.ifNotEmpty(body: C.() -> O?): O? = if (isNotEmpty()) this.body() else null -inline fun Array.ifNotEmpty(body: Array.() -> O?): O? = if (isNotEmpty()) this.body() else null \ No newline at end of file +inline fun Array.ifNotEmpty(body: Array.() -> O?): O? = if (isNotEmpty()) this.body() else null