Update coroutine related runtime parts
This commit is contained in:
committed by
Stanislav Erokhin
parent
852a5ee064
commit
72401efb9f
@@ -37,4 +37,4 @@ package kotlin.coroutines
|
|||||||
* Use [runWithCurrentContinuation] as a safer way to obtain current continuation instance.
|
* Use [runWithCurrentContinuation] as a safer way to obtain current continuation instance.
|
||||||
*/
|
*/
|
||||||
@SinceKotlin("1.1")
|
@SinceKotlin("1.1")
|
||||||
public inline suspend fun <T> maySuspendWithCurrentContinuation(body: (Continuation<T>) -> Any?): T
|
public inline suspend fun <T> suspendWithCurrentContinuation(body: (Continuation<T>) -> Any?): T
|
||||||
|
|||||||
@@ -49,10 +49,15 @@ public annotation class AllowSuspendExtensions
|
|||||||
@SinceKotlin("1.1")
|
@SinceKotlin("1.1")
|
||||||
public val SUSPENDED: Any? = Any()
|
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
|
* 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
|
* `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.
|
* receiver only and are restricted from calling arbitrary suspension functions.
|
||||||
*/
|
*/
|
||||||
@SinceKotlin("1.1")
|
@SinceKotlin("1.1")
|
||||||
public annotation class RestrictSuspension
|
public annotation class RestrictSuspension
|
||||||
|
|||||||
@@ -16,9 +16,6 @@
|
|||||||
|
|
||||||
package kotlin.coroutines
|
package kotlin.coroutines
|
||||||
|
|
||||||
import kotlin.jvm.internal.CoroutineImpl
|
|
||||||
import kotlin.jvm.internal.RestrictedCoroutineImpl
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* A strategy to intercept resumptions inside coroutine.
|
* A strategy to intercept resumptions inside coroutine.
|
||||||
* Interceptor may shift resumption into another execution frame by scheduling asynchronous execution
|
* Interceptor may shift resumption into another execution frame by scheduling asynchronous execution
|
||||||
@@ -47,24 +44,11 @@ public interface ResumeInterceptor {
|
|||||||
*/
|
*/
|
||||||
@SinceKotlin("1.1")
|
@SinceKotlin("1.1")
|
||||||
@Suppress("UNCHECKED_CAST")
|
@Suppress("UNCHECKED_CAST")
|
||||||
public fun <R, T> (/*suspend*/ R.() -> T).createCoroutine(
|
public fun <R, T> (@Suspend() (R.() -> T)).createCoroutine(
|
||||||
receiver: R,
|
receiver: R,
|
||||||
resultHandler: Continuation<T>
|
resultHandler: Continuation<T>,
|
||||||
): Continuation<Unit> {
|
resumeInterceptor: ResumeInterceptor? = null
|
||||||
// check if the RestrictedCoroutineImpl was passed and do efficient creation
|
): Continuation<Unit> = (this as (R, Continuation<T>) -> Continuation<Unit>).invoke(receiver, withInterceptor(resultHandler, resumeInterceptor))
|
||||||
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].
|
* Starts coroutine with receiver type [R] and result type [T].
|
||||||
@@ -73,17 +57,12 @@ public fun <R, T> (/*suspend*/ R.() -> T).createCoroutine(
|
|||||||
*/
|
*/
|
||||||
@SinceKotlin("1.1")
|
@SinceKotlin("1.1")
|
||||||
@Suppress("UNCHECKED_CAST")
|
@Suppress("UNCHECKED_CAST")
|
||||||
public fun <R, T> (/*suspend*/ R.() -> T).startCoroutine(
|
public fun <R, T> (@Suspend() (R.() -> T)).startCoroutine(
|
||||||
receiver: R,
|
receiver: R,
|
||||||
resultHandler: Continuation<T>
|
resultHandler: Continuation<T>,
|
||||||
|
resumeInterceptor: ResumeInterceptor? = null
|
||||||
) {
|
) {
|
||||||
try {
|
createCoroutine(receiver, resultHandler, resumeInterceptor).resume(Unit)
|
||||||
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)
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
@@ -95,24 +74,10 @@ public fun <R, T> (/*suspend*/ R.() -> T).startCoroutine(
|
|||||||
*/
|
*/
|
||||||
@SinceKotlin("1.1")
|
@SinceKotlin("1.1")
|
||||||
@Suppress("UNCHECKED_CAST")
|
@Suppress("UNCHECKED_CAST")
|
||||||
public fun <T> (/*suspend*/ () -> T).createCoroutine(
|
public fun <T> (@Suspend() (() -> T)).createCoroutine(
|
||||||
resultHandler: Continuation<T>,
|
resultHandler: Continuation<T>,
|
||||||
resumeInterceptor: ResumeInterceptor? = null
|
resumeInterceptor: ResumeInterceptor? = null
|
||||||
): Continuation<Unit> {
|
): Continuation<Unit> = (this as (Continuation<T>) -> Continuation<Unit>).invoke(withInterceptor(resultHandler, resumeInterceptor))
|
||||||
// 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].
|
* Starts coroutine without receiver and with result type [T].
|
||||||
@@ -122,17 +87,11 @@ public fun <T> (/*suspend*/ () -> T).createCoroutine(
|
|||||||
*/
|
*/
|
||||||
@SinceKotlin("1.1")
|
@SinceKotlin("1.1")
|
||||||
@Suppress("UNCHECKED_CAST")
|
@Suppress("UNCHECKED_CAST")
|
||||||
public fun <T> (/*suspend*/ () -> T).startCoroutine(
|
public fun <T> (@Suspend() (() -> T)).startCoroutine(
|
||||||
resultHandler: Continuation<T>,
|
resultHandler: Continuation<T>,
|
||||||
resumeInterceptor: ResumeInterceptor? = null
|
resumeInterceptor: ResumeInterceptor? = null
|
||||||
) {
|
) {
|
||||||
try {
|
createCoroutine(resultHandler, resumeInterceptor).resume(Unit)
|
||||||
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 stuff -------
|
||||||
|
|||||||
@@ -16,8 +16,8 @@
|
|||||||
|
|
||||||
package kotlin.coroutines
|
package kotlin.coroutines
|
||||||
|
|
||||||
|
import java.lang.IllegalStateException
|
||||||
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater
|
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater
|
||||||
import kotlin.IllegalStateException
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* This function allows to obtain the current continuation instance inside suspend functions and suspend
|
* 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.
|
* Repeated invocation of any resume function on continuation produces unspecified behavior.
|
||||||
* Use [runWithCurrentContinuation] as a safer way to obtain current continuation instance.
|
* Use [runWithCurrentContinuation] as a safer way to obtain current continuation instance.
|
||||||
*/
|
*/
|
||||||
@SinceKotlin("1.1")
|
//@SinceKotlin("1.1")
|
||||||
public inline suspend fun <T> suspendWithCurrentContinuation(crossinline body: (Continuation<T>) -> Unit): T =
|
//public inline suspend fun <T> suspendWithCurrentContinuation(crossinline body: (Continuation<T>) -> Unit): T =
|
||||||
maySuspendWithCurrentContinuation<T> { c: Continuation<T> ->
|
// maySuspendWithCurrentContinuation<T> { c: Continuation<T> ->
|
||||||
body(c)
|
// body(c)
|
||||||
SUSPENDED
|
// SUSPENDED
|
||||||
}
|
// }
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* This function allows to safely obtain the current continuation instance inside suspend functions and suspend
|
* This function allows to safely obtain the current continuation instance inside suspend functions and suspend
|
||||||
@@ -50,8 +50,8 @@ public inline suspend fun <T> suspendWithCurrentContinuation(crossinline body: (
|
|||||||
* Repeated invocation of any resume function produces [IllegalStateException].
|
* Repeated invocation of any resume function produces [IllegalStateException].
|
||||||
*/
|
*/
|
||||||
@SinceKotlin("1.1")
|
@SinceKotlin("1.1")
|
||||||
public inline suspend fun <T> runWithCurrentContinuation(body: (Continuation<T>) -> Unit): T =
|
public inline suspend fun <T> runWithCurrentContinuation(crossinline body: (Continuation<T>) -> Unit): T =
|
||||||
maySuspendWithCurrentContinuation<T> { c: Continuation<T> ->
|
suspendWithCurrentContinuation { c: Continuation<T> ->
|
||||||
val safe = SafeContinuation(c)
|
val safe = SafeContinuation(c)
|
||||||
body(safe)
|
body(safe)
|
||||||
safe.getResult()
|
safe.getResult()
|
||||||
@@ -65,7 +65,8 @@ private class Fail(val exception: Throwable)
|
|||||||
private val RESULT_UPDATER = AtomicReferenceFieldUpdater.newUpdater<SafeContinuation<*>, Any?>(
|
private val RESULT_UPDATER = AtomicReferenceFieldUpdater.newUpdater<SafeContinuation<*>, Any?>(
|
||||||
SafeContinuation::class.java, Any::class.java as Class<Any?>, "result")
|
SafeContinuation::class.java, Any::class.java as Class<Any?>, "result")
|
||||||
|
|
||||||
internal class SafeContinuation<T> @PublishedApi constructor(private val delegate: Continuation<T>) : Continuation<T> {
|
@PublishedApi
|
||||||
|
internal class SafeContinuation<T> @PublishedApi internal constructor(private val delegate: Continuation<T>) : Continuation<T> {
|
||||||
@Volatile
|
@Volatile
|
||||||
private var result: Any? = UNDECIDED
|
private var result: Any? = UNDECIDED
|
||||||
|
|
||||||
@@ -100,7 +101,8 @@ internal class SafeContinuation<T> @PublishedApi constructor(private val delegat
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
fun getResult(): Any? {
|
@PublishedApi
|
||||||
|
internal fun getResult(): Any? {
|
||||||
val result = this.result // atomic read
|
val result = this.result // atomic read
|
||||||
if (result == UNDECIDED && cas(UNDECIDED, SUSPENDED)) return SUSPENDED
|
if (result == UNDECIDED && cas(UNDECIDED, SUSPENDED)) return SUSPENDED
|
||||||
when (result) {
|
when (result) {
|
||||||
|
|||||||
@@ -16,8 +16,6 @@
|
|||||||
|
|
||||||
package kotlin.jvm.internal
|
package kotlin.jvm.internal
|
||||||
|
|
||||||
import kotlin.coroutines.*
|
|
||||||
|
|
||||||
private const val INTERCEPT_BIT_SET = 1 shl 31
|
private const val INTERCEPT_BIT_SET = 1 shl 31
|
||||||
private const val INTERCEPT_BIT_CLEAR = INTERCEPT_BIT_SET.inv()
|
private const val INTERCEPT_BIT_CLEAR = INTERCEPT_BIT_SET.inv()
|
||||||
|
|
||||||
@@ -28,23 +26,11 @@ abstract class CoroutineImpl : RestrictedCoroutineImpl, InterceptableContinuatio
|
|||||||
override val resumeInterceptor: ResumeInterceptor?
|
override val resumeInterceptor: ResumeInterceptor?
|
||||||
get() = _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
|
// this constructor is used to create a continuation instance for coroutine
|
||||||
constructor(arity: Int, resultContinuation: Continuation<Any?>?) : super(arity, resultContinuation) {
|
constructor(arity: Int, resultContinuation: Continuation<Any?>?) : super(arity, resultContinuation) {
|
||||||
_resumeInterceptor = (resultContinuation as? InterceptableContinuation<*>)?.resumeInterceptor
|
_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?) {
|
override fun resume(data: Any?) {
|
||||||
if (_resumeInterceptor != null) {
|
if (_resumeInterceptor != null) {
|
||||||
if (label and INTERCEPT_BIT_SET == 0) {
|
if (label and INTERCEPT_BIT_SET == 0) {
|
||||||
@@ -71,37 +57,19 @@ abstract class CoroutineImpl : RestrictedCoroutineImpl, InterceptableContinuatio
|
|||||||
@SinceKotlin("1.1")
|
@SinceKotlin("1.1")
|
||||||
abstract class RestrictedCoroutineImpl : Lambda, Continuation<Any?> {
|
abstract class RestrictedCoroutineImpl : Lambda, Continuation<Any?> {
|
||||||
@JvmField
|
@JvmField
|
||||||
protected val resultContinuation: Continuation<Any?>?
|
protected var resultContinuation: Continuation<Any?>?
|
||||||
|
|
||||||
// label == -1 when coroutine cannot be started (it is just a factory object) or has already finished execution
|
// 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
|
// label == 0 in initial part of the coroutine
|
||||||
@JvmField
|
@JvmField
|
||||||
protected var label: Int
|
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
|
// this constructor is used to create a continuation instance for coroutine
|
||||||
constructor(arity: Int, resultContinuation: Continuation<Any?>?) : super(arity) {
|
constructor(arity: Int, resultContinuation: Continuation<Any?>?) : super(arity) {
|
||||||
this.resultContinuation = resultContinuation
|
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<Unit>
|
|
||||||
|
|
||||||
internal fun doCreateInternal(receiver: Any?, resultContinuation: Continuation<*>) =
|
|
||||||
doCreate(receiver, resultContinuation)
|
|
||||||
|
|
||||||
override fun resume(data: Any?) {
|
override fun resume(data: Any?) {
|
||||||
try {
|
try {
|
||||||
val result = doResume(data, null)
|
val result = doResume(data, null)
|
||||||
@@ -195,4 +163,4 @@ class XXXCoroutine : RestrictedCoroutineImpl, Function2<Receiver, Continuation<*
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
---------------------------------------------------------------------------------------------
|
---------------------------------------------------------------------------------------------
|
||||||
*/
|
*/
|
||||||
|
|||||||
@@ -78,6 +78,9 @@ fun <T1, T2> Pair<T1, T2>.swap(): Pair<T2, T1> = Pair(second, first)
|
|||||||
|
|
||||||
fun <T: Any> T.check(predicate: (T) -> Boolean): T? = if (predicate(this)) this else null
|
fun <T: Any> T.check(predicate: (T) -> Boolean): T? = if (predicate(this)) this else null
|
||||||
|
|
||||||
|
inline fun <reified T : Any> Any?.safeAs(): T? = this as? T
|
||||||
|
inline fun <reified T : Any> Any?.cast(): T = this as T
|
||||||
|
|
||||||
fun <T : Any> constant(calculator: () -> T): T {
|
fun <T : Any> constant(calculator: () -> T): T {
|
||||||
val cached = constantMap[calculator]
|
val cached = constantMap[calculator]
|
||||||
@Suppress("UNCHECKED_CAST")
|
@Suppress("UNCHECKED_CAST")
|
||||||
@@ -124,4 +127,4 @@ inline fun <T> Iterable<T>.sumByLong(selector: (T) -> Long): Long {
|
|||||||
|
|
||||||
inline fun <T, C : Collection<T>, O> C.ifNotEmpty(body: C.() -> O?): O? = if (isNotEmpty()) this.body() else null
|
inline fun <T, C : Collection<T>, O> C.ifNotEmpty(body: C.() -> O?): O? = if (isNotEmpty()) this.body() else null
|
||||||
|
|
||||||
inline fun <T, O> Array<out T>.ifNotEmpty(body: Array<out T>.() -> O?): O? = if (isNotEmpty()) this.body() else null
|
inline fun <T, O> Array<out T>.ifNotEmpty(body: Array<out T>.() -> O?): O? = if (isNotEmpty()) this.body() else null
|
||||||
|
|||||||
Reference in New Issue
Block a user