Get rid of AbstractCoroutine.defaultResumeMode

It was only needed to customize resume mode from within of
AbstractCoroutine.resumeWith method. This is now achieved by
a separate open fun afterResume. Also, afterCompletionInternal is
now renamed to afterCompletion and is only used from the job was
completed from unknown context and thus should be resumed in
a default way. This "default way" is now "cancellable resume"
(to minimize the amount of code duplication) which does not introduce
any functional difference, since that only happens on cancellation.

Remove MODE_DIRECT resume mode

Author: elizarov

binary-compatibility-validator/reference-public-api/kotlinx-coroutines-core.txt

@@ -2,6 +2,7 @@ public abstract class kotlinx/coroutines/AbstractCoroutine : kotlinx/coroutines/
 	protected final field parentContext Lkotlin/coroutines/CoroutineContext;
 	public fun <init> (Lkotlin/coroutines/CoroutineContext;Z)V
 	public synthetic fun <init> (Lkotlin/coroutines/CoroutineContext;ZILkotlin/jvm/internal/DefaultConstructorMarker;)V
+	protected fun afterResume (Ljava/lang/Object;)V
 	public final fun getContext ()Lkotlin/coroutines/CoroutineContext;
 	public fun getCoroutineContext ()Lkotlin/coroutines/CoroutineContext;
 	public fun isActive ()Z
@@ -380,7 +381,7 @@ public final class kotlinx/coroutines/JobKt {
 
 public class kotlinx/coroutines/JobSupport : kotlinx/coroutines/ChildJob, kotlinx/coroutines/Job, kotlinx/coroutines/ParentJob, kotlinx/coroutines/selects/SelectClause0 {
 	public fun <init> (Z)V
-	protected fun afterCompletionInternal (Ljava/lang/Object;I)V
+	protected fun afterCompletion (Ljava/lang/Object;)V
 	public final fun attachChild (Lkotlinx/coroutines/ChildJob;)Lkotlinx/coroutines/ChildHandle;
 	public synthetic fun cancel ()V
 	public synthetic fun cancel (Ljava/lang/Throwable;)Z

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

@@ -102,17 +102,17 @@ public abstract class AbstractCoroutine<in T>(
             onCompleted(state as T)
     }
 
-    internal open val defaultResumeMode: Int get() = MODE_ATOMIC_DEFAULT
-
     /**
      * Completes execution of this with coroutine with the specified result.
      */
     public final override fun resumeWith(result: Result<T>) {
         val state = makeCompletingOnce(result.toState())
         if (state === COMPLETING_WAITING_CHILDREN) return
-        afterCompletionInternal(state, defaultResumeMode)
+        afterResume(state)
     }
 
+    protected open fun afterResume(state: Any?) = afterCompletion(state)
+
     internal final override fun handleOnCompletionException(exception: Throwable) {
         handleCoroutineException(context, exception)
     }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
  */
 
 @file:JvmMultifileClass
@@ -142,20 +142,20 @@ public suspend fun <T> withContext(
     newContext.checkCompletion()
     // FAST PATH #1 -- new context is the same as the old one
     if (newContext === oldContext) {
-        val coroutine = ScopeCoroutine(newContext, uCont) // MODE_DIRECT
+        val coroutine = ScopeCoroutine(newContext, uCont)
         return@sc coroutine.startUndispatchedOrReturn(coroutine, block)
     }
     // FAST PATH #2 -- the new dispatcher is the same as the old one (something else changed)
     // `equals` is used by design (see equals implementation is wrapper context like ExecutorCoroutineDispatcher)
     if (newContext[ContinuationInterceptor] == oldContext[ContinuationInterceptor]) {
-        val coroutine = UndispatchedCoroutine(newContext, uCont) // MODE_UNDISPATCHED
+        val coroutine = UndispatchedCoroutine(newContext, uCont)
         // There are changes in the context, so this thread needs to be updated
         withCoroutineContext(newContext, null) {
             return@sc coroutine.startUndispatchedOrReturn(coroutine, block)
         }
     }
     // SLOW PATH -- use new dispatcher
-    val coroutine = DispatchedCoroutine(newContext, uCont) // MODE_CANCELLABLE
+    val coroutine = DispatchedCoroutine(newContext, uCont)
     coroutine.initParentJob()
     block.startCoroutineCancellable(coroutine, coroutine)
     coroutine.getResult()
@@ -200,7 +200,13 @@ private class UndispatchedCoroutine<in T>(
     context: CoroutineContext,
     uCont: Continuation<T>
 ) : ScopeCoroutine<T>(context, uCont) {
-    override val defaultResumeMode: Int get() = MODE_UNDISPATCHED
+    override fun afterResume(state: Any?) {
+        // resume undispatched -- update context by stay on the same dispatcher
+        val result = recoverResult(state, uCont)
+        withCoroutineContext(uCont.context, null) {
+            uCont.resumeWith(result)
+        }
+    }
 }
 
 private const val UNDECIDED = 0
@@ -212,8 +218,6 @@ private class DispatchedCoroutine<in T>(
     context: CoroutineContext,
     uCont: Continuation<T>
 ) : ScopeCoroutine<T>(context, uCont) {
-    override val defaultResumeMode: Int get() = MODE_CANCELLABLE
-
     // this is copy-and-paste of a decision state machine inside AbstractionContinuation
     // todo: we may some-how abstract it via inline class
     private val _decision = atomic(UNDECIDED)
@@ -238,10 +242,16 @@ private class DispatchedCoroutine<in T>(
         }
     }
 
-    override fun afterCompletionInternal(state: Any?, mode: Int) {
+    override fun afterCompletion(state: Any?) {
+        // Call afterResume from afterCompletion and not vice-versa, because stack-size is more
+        // important for afterResume implementation
+        afterResume(state)
+    }
+
+    override fun afterResume(state: Any?) {
         if (tryResume()) return // completed before getResult invocation -- bail out
-        // otherwise, getResult has already commenced, i.e. completed later or in other thread
-        super.afterCompletionInternal(state, mode)
+        // Resume in a cancellable way because we have to switch back to the original dispatcher
+        uCont.intercepted().resumeCancellableWith(recoverResult(state, uCont))
     }
 
     fun getResult(): Any? {

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

@@ -1,16 +1,24 @@
 /*
- * Copyright 2016-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
  */
 
 package kotlinx.coroutines
 
 import kotlinx.atomicfu.*
+import kotlinx.coroutines.internal.*
 import kotlin.coroutines.*
 import kotlin.jvm.*
 
 internal fun <T> Result<T>.toState(): Any? =
     if (isSuccess) getOrThrow() else CompletedExceptionally(exceptionOrNull()!!) // todo: need to do it better
 
+@Suppress("RESULT_CLASS_IN_RETURN_TYPE", "UNCHECKED_CAST")
+internal fun <T> recoverResult(state: Any?, uCont: Continuation<T>): Result<T> =
+    if (state is CompletedExceptionally)
+        Result.failure(recoverStackTrace(state.cause, uCont))
+    else
+        Result.success(state as T)
+
 /**
  * Class for an internal state of a job that was cancelled (completed exceptionally).
  *

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

@@ -194,16 +194,16 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
     // Finalizes Finishing -> Completed (terminal state) transition.
     // ## IMPORTANT INVARIANT: Only one thread can be concurrently invoking this method.
     // Returns final state that was created and updated to
-    private fun tryFinalizeFinishingState(state: Finishing, proposedUpdate: Any?): Any? {
+    private fun finalizeFinishingState(state: Finishing, proposedUpdate: Any?): Any? {
         /*
          * Note: proposed state can be Incomplete, e.g.
          * async {
          *     something.invokeOnCompletion {} // <- returns handle which implements Incomplete under the hood
          * }
          */
-        require(this.state === state) // consistency check -- it cannot change
-        require(!state.isSealed) // consistency check -- cannot be sealed yet
-        require(state.isCompleting) // consistency check -- must be marked as completing
+        assert { this.state === state } // consistency check -- it cannot change
+        assert { !state.isSealed } // consistency check -- cannot be sealed yet
+        assert { state.isCompleting } // consistency check -- must be marked as completing
         val proposedException = (proposedUpdate as? CompletedExceptionally)?.cause
         // Create the final exception and seal the state so that no more exceptions can be added
         var wasCancelling = false // KLUDGE: we cannot have contract for our own expect fun synchronized
@@ -233,7 +233,8 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
         if (!wasCancelling) onCancelling(finalException)
         onCompletionInternal(finalState)
         // Then CAS to completed state -> it must succeed
-        require(_state.compareAndSet(state, finalState.boxIncomplete())) { "Unexpected state: ${_state.value}, expected: $state, update: $finalState" }
+        val casSuccess = _state.compareAndSet(state, finalState.boxIncomplete())
+        assert { casSuccess }
         // And process all post-completion actions
         completeStateFinalization(state, finalState)
         return finalState
@@ -269,7 +270,7 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
     }
 
     // fast-path method to finalize normally completed coroutines without children
-    // returns true if complete, and afterCompletionInternal(update, mode) shall be called
+    // returns true if complete, and afterCompletion(update) shall be called
     private fun tryFinalizeSimpleState(state: Incomplete, update: Any?): Boolean {
         assert { state is Empty || state is JobNode<*> } // only simple state without lists where children can concurrently add
         assert { update !is CompletedExceptionally } // only for normal completion
@@ -495,10 +496,10 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
 
     private fun makeNode(handler: CompletionHandler, onCancelling: Boolean): JobNode<*> {
         return if (onCancelling)
-            (handler as? JobCancellingNode<*>)?.also { require(it.job === this) }
+            (handler as? JobCancellingNode<*>)?.also { assert { it.job === this } }
                 ?: InvokeOnCancelling(this, handler)
         else
-            (handler as? JobNode<*>)?.also { require(it.job === this && it !is JobCancellingNode) }
+            (handler as? JobNode<*>)?.also { assert { it.job === this && it !is JobCancellingNode } }
                 ?: InvokeOnCompletion(this, handler)
     }
 
@@ -653,7 +654,7 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
             finalState === COMPLETING_WAITING_CHILDREN -> true
             finalState === TOO_LATE_TO_CANCEL -> false
             else -> {
-                afterCompletionInternal(finalState, MODE_ATOMIC_DEFAULT)
+                afterCompletion(finalState)
                 true
             }
         }
@@ -663,7 +664,7 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
     // It contains a loop and never returns COMPLETING_RETRY, can return
     // COMPLETING_ALREADY -- if already completed/completing
     // COMPLETING_WAITING_CHILDREN -- if started waiting for children
-    // final state -- when completed, for call to afterCompletionInternal
+    // final state -- when completed, for call to afterCompletion
     private fun cancelMakeCompleting(cause: Any?): Any? {
         loopOnState { state ->
             if (state !is Incomplete || state is Finishing && state.isCompleting) {
@@ -703,7 +704,7 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
     // COMPLETING_ALREADY -- if already completing or successfully made cancelling, added exception
     // COMPLETING_WAITING_CHILDREN -- if started waiting for children, added exception
     // TOO_LATE_TO_CANCEL -- too late to cancel, did not add exception
-    // final state -- when completed, for call to afterCompletionInternal
+    // final state -- when completed, for call to afterCompletion
     private fun makeCancelling(cause: Any?): Any? {
         var causeExceptionCache: Throwable? = null // lazily init result of createCauseException(cause)
         loopOnState { state ->
@@ -786,7 +787,7 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
                 finalState === COMPLETING_WAITING_CHILDREN -> return true
                 finalState === COMPLETING_RETRY -> return@loopOnState
                 else -> {
-                    afterCompletionInternal(finalState, MODE_ATOMIC_DEFAULT)
+                    afterCompletion(finalState)
                     return true
                 }
             }
@@ -798,7 +799,7 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
      * It throws [IllegalStateException] on repeated invocation (when this job is already completing).
      * Returns:
      * * [COMPLETING_WAITING_CHILDREN] if started waiting for children.
-     * * Final state otherwise (caller should do [afterCompletionInternal])
+     * * Final state otherwise (caller should do [afterCompletion])
      */
     internal fun makeCompletingOnce(proposedUpdate: Any?): Any? {
         loopOnState { state ->
@@ -819,7 +820,7 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
     // COMPLETING_ALREADY -- when already complete or completing
     // COMPLETING_RETRY -- when need to retry due to interference
     // COMPLETING_WAITING_CHILDREN -- when made completing and is waiting for children
-    // final state -- when completed, for call to afterCompletionInternal
+    // final state -- when completed, for call to afterCompletion
     private fun tryMakeCompleting(state: Any?, proposedUpdate: Any?): Any? {
         if (state !is Incomplete)
             return COMPLETING_ALREADY
@@ -844,7 +845,7 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
     // COMPLETING_ALREADY -- when already complete or completing
     // COMPLETING_RETRY -- when need to retry due to interference
     // COMPLETING_WAITING_CHILDREN -- when made completing and is waiting for children
-    // final state -- when completed, for call to afterCompletionInternal
+    // final state -- when completed, for call to afterCompletion
     private fun tryMakeCompletingSlowPath(state: Incomplete, proposedUpdate: Any?): Any? {
         // get state's list or else promote to list to correctly operate on child lists
         val list = getOrPromoteCancellingList(state) ?: return COMPLETING_RETRY
@@ -866,7 +867,7 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
                 if (!_state.compareAndSet(state, finishing)) return COMPLETING_RETRY
             }
             // ## IMPORTANT INVARIANT: Only one thread (that had set isCompleting) can go past this point
-            require(!finishing.isSealed) // cannot be sealed
+            assert { !finishing.isSealed } // cannot be sealed
             // add new proposed exception to the finishing state
             val wasCancelling = finishing.isCancelling
             (proposedUpdate as? CompletedExceptionally)?.let { finishing.addExceptionLocked(it.cause) }
@@ -880,7 +881,7 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
         if (child != null && tryWaitForChild(finishing, child, proposedUpdate))
             return COMPLETING_WAITING_CHILDREN
         // otherwise -- we have not children left (all were already cancelled?)
-        return tryFinalizeFinishingState(finishing, proposedUpdate)
+        return finalizeFinishingState(finishing, proposedUpdate)
     }
 
     private val Any?.exceptionOrNull: Throwable?
@@ -903,14 +904,14 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
 
     // ## IMPORTANT INVARIANT: Only one thread can be concurrently invoking this method.
     private fun continueCompleting(state: Finishing, lastChild: ChildHandleNode, proposedUpdate: Any?) {
-        require(this.state === state) // consistency check -- it cannot change while we are waiting for children
+        assert { this.state === state } // consistency check -- it cannot change while we are waiting for children
         // figure out if we need to wait for next child
         val waitChild = lastChild.nextChild()
         // try wait for next child
         if (waitChild != null && tryWaitForChild(state, waitChild, proposedUpdate)) return // waiting for next child
         // no more children to wait -- try update state
-        val finalState = tryFinalizeFinishingState(state, proposedUpdate)
-        afterCompletionInternal(finalState, MODE_ATOMIC_DEFAULT)
+        val finalState = finalizeFinishingState(state, proposedUpdate)
+        afterCompletion(finalState)
     }
 
     private fun LockFreeLinkedListNode.nextChild(): ChildHandleNode? {
@@ -1014,14 +1015,13 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
     protected open fun onCompletionInternal(state: Any?) {}
 
     /**
-     * Override for the very last action on job's completion to resume the rest of the code in scoped coroutines.
-     *
-     * @param state the final state.
-     * @param mode completion mode.
+     * Override for the very last action on job's completion to resume the rest of the code in
+     * scoped coroutines. It is called when this job is externally completed in an unknown
+     * context and thus should resume with a default mode.
      *
      * @suppress **This is unstable API and it is subject to change.**
      */
-    protected open fun afterCompletionInternal(state: Any?, mode: Int) {}
+    protected open fun afterCompletion(state: Any?) {}
 
     // for nicer debugging
     public override fun toString(): String =

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

@@ -8,6 +8,13 @@ import kotlinx.coroutines.internal.*
 import kotlin.coroutines.*
 import kotlin.jvm.*
 
+@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
+@PublishedApi internal const val MODE_UNDISPATCHED = 2   // when the thread is right, but need to mark it with current coroutine
+
+internal val Int.isCancellableMode get() = this == MODE_CANCELLABLE
+internal val Int.isDispatchedMode get() = this == MODE_ATOMIC_DEFAULT || this == MODE_CANCELLABLE
+
 internal abstract class DispatchedTask<in T>(
     @JvmField public var resumeMode: Int
 ) : SchedulerTask() {
@@ -89,7 +96,7 @@ internal abstract class DispatchedTask<in T>(
     }
 }
 
-internal fun <T> DispatchedTask<T>.dispatch(mode: Int = MODE_CANCELLABLE) {
+internal fun <T> DispatchedTask<T>.dispatch(mode: Int) {
     val delegate = this.delegate
     if (mode.isDispatchedMode && delegate is DispatchedContinuation<*> && mode.isCancellableMode == resumeMode.isCancellableMode) {
         // dispatch directly using this instance's Runnable implementation
@@ -125,7 +132,6 @@ internal fun <T> DispatchedTask<T>.resume(delegate: Continuation<T>, useMode: In
     when (useMode) {
         MODE_ATOMIC_DEFAULT -> delegate.resumeWith(result)
         MODE_CANCELLABLE -> delegate.resumeCancellableWith(result)
-        MODE_DIRECT -> ((delegate as? DispatchedContinuation)?.continuation ?: delegate).resumeWith(result)
         MODE_UNDISPATCHED -> (delegate as DispatchedContinuation).resumeUndispatchedWith(result)
         else -> error("Invalid mode $useMode")
     }