Removed suspendCoroutineOrReturn use in preparation for Kotlin 1.3

Replaced with suspendCoroutineUninterceptedOrReturn

Author: elizarov

common/kotlinx-coroutines-core-common/src/Builders.common.kt

@@ -106,27 +106,27 @@ public suspend fun <T> withContext(
     context: CoroutineContext,
     start: CoroutineStart = CoroutineStart.DEFAULT,
     block: suspend () -> T
-): T = suspendCoroutineOrReturn sc@ { cont ->
-    val oldContext = cont.context
+): T = suspendCoroutineUninterceptedOrReturn sc@ { uCont ->
+    val oldContext = uCont.context
     // fast path #1 if there is no change in the actual context:
     if (context === oldContext || context is CoroutineContext.Element && oldContext[context.key] === context)
-        return@sc block.startCoroutineUninterceptedOrReturn(cont)
+        return@sc block.startCoroutineUninterceptedOrReturn(uCont)
     // compute new context
     val newContext = oldContext + context
     // fast path #2 if the result is actually the same
     if (newContext === oldContext)
-        return@sc block.startCoroutineUninterceptedOrReturn(cont)
+        return@sc block.startCoroutineUninterceptedOrReturn(uCont)
     // fast path #3 if the new dispatcher is the same as the old one.
     // `equals` is used by design (see equals implementation is wrapper context like ExecutorCoroutineDispatcher)
     if (newContext[ContinuationInterceptor] == oldContext[ContinuationInterceptor]) {
-        val newContinuation = RunContinuationDirect(newContext, cont)
+        val newContinuation = RunContinuationDirect(newContext, uCont)
         return@sc block.startCoroutineUninterceptedOrReturn(newContinuation)
     }
     // slowest path otherwise -- use new interceptor, sync to its result via a full-blown instance of RunCompletion
     require(!start.isLazy) { "$start start is not supported" }
     val completion = RunCompletion(
         context = newContext,
-        delegate = cont,
+        delegate = uCont.intercepted(), // delegate to continuation intercepted with old dispatcher on completion
         resumeMode = if (start == CoroutineStart.ATOMIC) MODE_ATOMIC_DEFAULT else MODE_CANCELLABLE
     )
     completion.initParentJobInternal(newContext[Job]) // attach to job

common/kotlinx-coroutines-core-common/src/CancellableContinuation.kt

@@ -172,8 +172,8 @@ public suspend inline fun <T> suspendCancellableCoroutine(
     holdCancellability: Boolean = false,
     crossinline block: (CancellableContinuation<T>) -> Unit
 ): T =
-    suspendCoroutineOrReturn { cont ->
-        val cancellable = CancellableContinuationImpl(cont, resumeMode = MODE_CANCELLABLE)
+    suspendCoroutineUninterceptedOrReturn { uCont ->
+        val cancellable = CancellableContinuationImpl(uCont.intercepted(), resumeMode = MODE_CANCELLABLE)
         if (!holdCancellability) cancellable.initCancellability()
         block(cancellable)
         cancellable.getResult()
@@ -191,8 +191,8 @@ public suspend inline fun <T> suspendAtomicCancellableCoroutine(
     holdCancellability: Boolean = false,
     crossinline block: (CancellableContinuation<T>) -> Unit
 ): T =
-    suspendCoroutineOrReturn { cont ->
-        val cancellable = CancellableContinuationImpl(cont, resumeMode = MODE_ATOMIC_DEFAULT)
+    suspendCoroutineUninterceptedOrReturn { uCont ->
+        val cancellable = CancellableContinuationImpl(uCont.intercepted(), resumeMode = MODE_ATOMIC_DEFAULT)
         if (!holdCancellability) cancellable.initCancellability()
         block(cancellable)
         cancellable.getResult()

common/kotlinx-coroutines-core-common/src/JobSupport.kt

@@ -402,10 +402,8 @@ internal open class JobSupport constructor(active: Boolean) : Job, SelectClause0
 
     public final override suspend fun join() {
         if (!joinInternal()) { // fast-path no wait
-            return suspendCoroutineOrReturn { cont ->
-                cont.context.checkCompletion()
-                Unit // do not suspend
-            }
+            coroutineContext.checkCompletion()
+            return // do not suspend
         }
         return joinSuspend() // slow-path wait
     }

common/kotlinx-coroutines-core-common/src/ResumeMode.kt

@@ -4,7 +4,8 @@
 
 package kotlinx.coroutines.experimental
 
-import kotlin.coroutines.experimental.Continuation
+import kotlin.coroutines.experimental.*
+import kotlin.coroutines.experimental.intrinsics.*
 
 @PublishedApi internal const val MODE_ATOMIC_DEFAULT = 0 // schedule non-cancellable dispatch for suspendCoroutine
 @PublishedApi internal const val MODE_CANCELLABLE = 1    // schedule cancellable dispatch for suspendCancellableCoroutine
@@ -36,3 +37,25 @@ internal fun <T> Continuation<T>.resumeWithExceptionMode(exception: Throwable, m
         else -> error("Invalid mode $mode")
     }
 }
+
+internal fun <T> Continuation<T>.resumeUninterceptedMode(value: T, mode: Int) {
+    when (mode) {
+        MODE_ATOMIC_DEFAULT -> intercepted().resume(value)
+        MODE_CANCELLABLE -> intercepted().resumeCancellable(value)
+        MODE_DIRECT -> resume(value)
+        MODE_UNDISPATCHED -> resume(value)
+        MODE_IGNORE -> {}
+        else -> error("Invalid mode $mode")
+    }
+}
+
+internal fun <T> Continuation<T>.resumeUninterceptedWithExceptionMode(exception: Throwable, mode: Int) {
+    when (mode) {
+        MODE_ATOMIC_DEFAULT -> intercepted().resumeWithException(exception)
+        MODE_CANCELLABLE -> intercepted().resumeCancellableWithException(exception)
+        MODE_DIRECT -> resumeWithException(exception)
+        MODE_UNDISPATCHED -> resumeWithException(exception)
+        MODE_IGNORE -> {}
+        else -> error("Invalid mode $mode")
+    }
+}

common/kotlinx-coroutines-core-common/src/Scheduled.kt

@@ -51,8 +51,8 @@ public suspend fun <T> withTimeout(time: Int, block: suspend CoroutineScope.() -
  */
 public suspend fun <T> withTimeout(time: Long, unit: TimeUnit = TimeUnit.MILLISECONDS, block: suspend CoroutineScope.() -> T): T {
     if (time <= 0L) throw CancellationException("Timed out immediately")
-    return suspendCoroutineOrReturn { cont: Continuation<T> ->
-        setupTimeout(TimeoutCoroutine(time, unit, cont), block)
+    return suspendCoroutineUninterceptedOrReturn { uCont ->
+        setupTimeout(TimeoutCoroutine(time, unit, uCont), block)
     }
 }
 
@@ -61,7 +61,7 @@ private fun <U, T: U> setupTimeout(
     block: suspend CoroutineScope.() -> T
 ): Any? {
     // schedule cancellation of this coroutine on time
-    val cont = coroutine.cont
+    val cont = coroutine.uCont
     val context = cont.context
     coroutine.disposeOnCompletion(context.delay.invokeOnTimeout(coroutine.time, coroutine.unit, coroutine))
     // restart block using new coroutine with new job,
@@ -79,8 +79,8 @@ public suspend fun <T> withTimeout(time: Long, unit: TimeUnit = TimeUnit.MILLISE
 private open class TimeoutCoroutine<U, in T: U>(
     @JvmField val time: Long,
     @JvmField val unit: TimeUnit,
-    @JvmField val cont: Continuation<U>
-) : AbstractCoroutine<T>(cont.context, active = true), Runnable, Continuation<T> {
+    @JvmField val uCont: Continuation<U> // unintercepted continuation
+) : AbstractCoroutine<T>(uCont.context, active = true), Runnable, Continuation<T> {
     override val defaultResumeMode: Int get() = MODE_DIRECT
 
     @Suppress("LeakingThis")
@@ -91,9 +91,9 @@ private open class TimeoutCoroutine<U, in T: U>(
     @Suppress("UNCHECKED_CAST")
     internal override fun onCompletionInternal(state: Any?, mode: Int) {
         if (state is CompletedExceptionally)
-            cont.resumeWithExceptionMode(state.cause, mode)
+            uCont.resumeUninterceptedWithExceptionMode(state.cause, mode)
         else
-            cont.resumeMode(state as T, mode)
+            uCont.resumeUninterceptedMode(state as T, mode)
     }
 
     override fun nameString(): String =
@@ -140,8 +140,8 @@ public suspend fun <T> withTimeoutOrNull(time: Int, block: suspend CoroutineScop
  */
 public suspend fun <T> withTimeoutOrNull(time: Long, unit: TimeUnit = TimeUnit.MILLISECONDS, block: suspend CoroutineScope.() -> T): T? {
     if (time <= 0L) return null
-    return suspendCoroutineOrReturn { cont: Continuation<T?> ->
-        setupTimeout(TimeoutOrNullCoroutine(time, unit, cont), block)
+    return suspendCoroutineUninterceptedOrReturn { uCont ->
+        setupTimeout(TimeoutOrNullCoroutine(time, unit, uCont), block)
     }
 }
 
@@ -155,17 +155,17 @@ public suspend fun <T> withTimeoutOrNull(time: Long, unit: TimeUnit = TimeUnit.M
 private class TimeoutOrNullCoroutine<T>(
     time: Long,
     unit: TimeUnit,
-    cont: Continuation<T?>
-) : TimeoutCoroutine<T?, T>(time, unit, cont) {
+    uCont: Continuation<T?> // unintercepted continuation
+) : TimeoutCoroutine<T?, T>(time, unit, uCont) {
     @Suppress("UNCHECKED_CAST")
     internal override fun onCompletionInternal(state: Any?, mode: Int) {
         if (state is CompletedExceptionally) {
             val exception = state.cause
             if (exception is TimeoutCancellationException && exception.coroutine === this)
-                cont.resumeMode(null, mode) else
-                cont.resumeWithExceptionMode(exception, mode)
+                uCont.resumeUninterceptedMode(null, mode) else
+                uCont.resumeUninterceptedWithExceptionMode(exception, mode)
         } else
-            cont.resumeMode(state as T, mode)
+            uCont.resumeUninterceptedMode(state as T, mode)
     }
 }
 

common/kotlinx-coroutines-core-common/src/Yield.kt

@@ -15,10 +15,10 @@ import kotlin.coroutines.experimental.intrinsics.*
  * If the [Job] of the current coroutine is cancelled or completed when this suspending function is invoked or while
  * this function is waiting for dispatching, it resumes with [CancellationException].
  */
-public suspend fun yield(): Unit = suspendCoroutineOrReturn sc@ { cont ->
-    val context = cont.context
+public suspend fun yield(): Unit = suspendCoroutineUninterceptedOrReturn sc@ { uCont ->
+    val context = uCont.context
     context.checkCompletion()
-    if (cont !is DispatchedContinuation<Unit>) return@sc Unit
+    val cont = uCont.intercepted() as? DispatchedContinuation<Unit> ?: return@sc Unit
     if (!cont.dispatcher.isDispatchNeeded(context)) return@sc Unit
     cont.dispatchYield(Unit)
     COROUTINE_SUSPENDED

common/kotlinx-coroutines-core-common/src/selects/Select.kt

@@ -169,9 +169,8 @@ public interface SelectInstance<in R> {
  * Use [yield] or [CoroutineScope.isActive] to periodically check for cancellation in tight loops if needed.
  */
 public suspend inline fun <R> select(crossinline builder: SelectBuilder<R>.() -> Unit): R =
-    // TODO suspendCoroutineUnintercepted
-    suspendCoroutineOrReturn { cont ->
-        val scope = SelectBuilderImpl(cont)
+    suspendCoroutineUninterceptedOrReturn { uCont ->
+        val scope = SelectBuilderImpl(uCont)
         try {
             builder(scope)
         } catch (e: Throwable) {
@@ -187,7 +186,7 @@ private val RESUMED: Any = Symbol("RESUMED")
 
 @PublishedApi
 internal class SelectBuilderImpl<in R>(
-    private val delegate: Continuation<R>
+    private val uCont: Continuation<R> // unintercepted delegate continuation
 ) : LockFreeLinkedListHead(), SelectBuilder<R>,
     SelectInstance<R>, Continuation<R> {
     // selection state is "this" (list of nodes) initially and is replaced by idempotent marker (or null) when selected
@@ -216,7 +215,7 @@ internal class SelectBuilderImpl<in R>(
               +-------------------+
      */
 
-    override val context: CoroutineContext get() = delegate.context
+    override val context: CoroutineContext get() = uCont.context
 
     override val completion: Continuation<R> get() = this
 
@@ -239,21 +238,21 @@ internal class SelectBuilderImpl<in R>(
     // Resumes in MODE_DIRECT
     override fun resume(value: R) {
         doResume({ value }) {
-            delegate.resumeDirect(value)
+            uCont.resume(value)
         }
     }
 
     // Resumes in MODE_DIRECT
     override fun resumeWithException(exception: Throwable) {
         doResume({ Fail(exception) }) {
-            delegate.resumeDirectWithException(exception)
+            uCont.resumeWithException(exception)
         }
     }
 
     // Resumes in MODE_CANCELLABLE
     override fun resumeSelectCancellableWithException(exception: Throwable) {
         doResume({ Fail(exception) }) {
-            delegate.resumeCancellableWithException(exception)
+            uCont.intercepted().resumeCancellableWithException(exception)
         }
     }
 

common/kotlinx-coroutines-core-common/src/selects/SelectUnbiased.kt

@@ -19,8 +19,8 @@ import kotlin.coroutines.experimental.intrinsics.*
  * See [select] function description for all the other details.
  */
 public suspend inline fun <R> selectUnbiased(crossinline builder: SelectBuilder<R>.() -> Unit): R =
-    suspendCoroutineOrReturn { cont ->
-        val scope = UnbiasedSelectBuilderImpl(cont)
+    suspendCoroutineUninterceptedOrReturn { uCont ->
+        val scope = UnbiasedSelectBuilderImpl(uCont)
         try {
             builder(scope)
         } catch (e: Throwable) {
@@ -31,9 +31,9 @@ public suspend inline fun <R> selectUnbiased(crossinline builder: SelectBuilder<
 
 
 @PublishedApi
-internal class UnbiasedSelectBuilderImpl<in R>(cont: Continuation<R>) :
+internal class UnbiasedSelectBuilderImpl<in R>(uCont: Continuation<R>) :
     SelectBuilder<R> {
-    val instance = SelectBuilderImpl(cont)
+    val instance = SelectBuilderImpl(uCont)
     val clauses = arrayListOf<() -> Unit>()
 
     @PublishedApi

core/kotlinx-coroutines-core/test/channels/TickerChannelCommonTest.kt

@@ -12,7 +12,7 @@ import org.junit.runners.*
 import kotlin.test.*
 
 @RunWith(Parameterized::class)
-class TimerChannelCommonTest(private val channelFactory: Channel) : TestBase() {
+class TickerChannelCommonTest(private val channelFactory: Channel) : TestBase() {
     companion object {
         @Parameterized.Parameters(name = "{0}")
         @JvmStatic

core/kotlinx-coroutines-io/src/ByteBufferChannel.kt

@@ -2216,10 +2216,10 @@ internal class ByteBufferChannel(
     private suspend fun readSuspendImpl(size: Int): Boolean {
         if (!readSuspendPredicate(size)) return true
 
-        return suspendCoroutineOrReturn { raw ->
+        return suspendCoroutineUninterceptedOrReturn { uCont ->
             val c = readSuspendContinuationCache
             suspensionForSize(size, c)
-            c.swap(raw)
+            c.swap(uCont.intercepted())
         }
     }
 
@@ -2269,13 +2269,15 @@ internal class ByteBufferChannel(
 
         writeSuspensionSize = size
         if (attachedJob != null) {
-            return suspendCoroutineOrReturn(writeSuspension)
+            return suspendCoroutineUninterceptedOrReturn { uCont ->
+                writeSuspension(uCont.intercepted())
+            }
         }
 
-        return suspendCoroutineOrReturn { raw ->
+        return suspendCoroutineUninterceptedOrReturn { uCont ->
             val c = writeSuspendContinuationCache
             writeSuspension(c)
-            c.swap(raw)
+            c.swap(uCont.intercepted())
         }
     }