Tweaked order of notifications for better error handling & consistency

* JobSupport.onCompletionInternal(onCompleted/onCancelled) is now invoked
  before all the user-installed listeners with is consistent with how
  the onCancelling/invokeOnCancelling(onCancelling=true) works.
* Now all the state processing (updating the future, reporting unhandled
  exception, etc) in onCompletionInternal happens before observers that
  .join/.wait the coroutine are resumed and even before the state is set
  to final (to avoid exception handing races) with the exception of
  fast-path successful completion of coroutine.
* JobSupport.afterCompletionInternal is introduced. It is invoked
  after all use-installed listeners and that is where scoped coroutines
  resume the rest of the code.
* Remove empty AbstractCoroutine.onCancellation, move docs to JobSupport
* onCancellation renamed to onCancelling for consistency with
  invokeOnCompletion(onCancelling=true)

Author: elizarov

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

@@ -5,9 +5,9 @@ public abstract class kotlinx/coroutines/AbstractCoroutine : kotlinx/coroutines/
 	public final fun getContext ()Lkotlin/coroutines/CoroutineContext;
 	public fun getCoroutineContext ()Lkotlin/coroutines/CoroutineContext;
 	public fun isActive ()Z
-	protected fun onCancellation (Ljava/lang/Throwable;)V
 	protected fun onCancelled (Ljava/lang/Throwable;Z)V
 	protected fun onCompleted (Ljava/lang/Object;)V
+	protected final fun onCompletionInternal (Ljava/lang/Object;)V
 	protected fun onStart ()V
 	public final fun resumeWith (Ljava/lang/Object;)V
 	public final fun start (Lkotlinx/coroutines/CoroutineStart;Ljava/lang/Object;Lkotlin/jvm/functions/Function2;)V
@@ -368,6 +368,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
 	public final fun attachChild (Lkotlinx/coroutines/ChildJob;)Lkotlinx/coroutines/ChildHandle;
 	public synthetic fun cancel ()V
 	public synthetic fun cancel (Ljava/lang/Throwable;)Z
@@ -396,7 +397,8 @@ public class kotlinx/coroutines/JobSupport : kotlinx/coroutines/ChildJob, kotlin
 	public final fun isCompletedExceptionally ()Z
 	public final fun join (Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public fun minusKey (Lkotlin/coroutines/CoroutineContext$Key;)Lkotlin/coroutines/CoroutineContext;
-	protected fun onCancellation (Ljava/lang/Throwable;)V
+	protected fun onCancelling (Ljava/lang/Throwable;)V
+	protected fun onCompletionInternal (Ljava/lang/Object;)V
 	public final fun parentCancelled (Lkotlinx/coroutines/ParentJob;)V
 	public fun plus (Lkotlin/coroutines/CoroutineContext;)Lkotlin/coroutines/CoroutineContext;
 	public fun plus (Lkotlinx/coroutines/Job;)Lkotlinx/coroutines/Job;

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

@@ -20,9 +20,8 @@ import kotlin.jvm.*
  *
  * The following methods are available for override:
  *
- * * [onStart] is invoked when coroutine is create in not active state and is [started][Job.start].
- * * [onCancellation] is invoked as soon as coroutine is _failing_, or is cancelled,
- *   or when it completes for any reason.
+ * * [onStart] is invoked when coroutine was created in not active state and is being [started][Job.start].
+ * * [onCancelling] is invoked as soon as coroutine is being cancelled for any reason (or completes).
  * * [onCompleted] is invoked when coroutine completes with a value.
  * * [onCancelled] in invoked when coroutines completes with exception (cancelled).
  *
@@ -77,31 +76,26 @@ public abstract class AbstractCoroutine<in T>(
     }
 
     /**
-     * This function is invoked once when this coroutine is cancelled
-     * similarly to [invokeOnCompletion] with `onCancelling` set to `true`.
-     *
-     * The meaning of [cause] parameter:
-     * * Cause is `null` when job has completed normally.
-     * * Cause is an instance of [CancellationException] when job was cancelled _normally_.
-     *   **It should not be treated as an error**. In particular, it should not be reported to error logs.
-     * * Otherwise, the job had been cancelled or failed with exception.
-     */
-    protected override fun onCancellation(cause: Throwable?) {}
-
-    /**
-     * This function is invoked once when job was completed normally with the specified [value].
+     * This function is invoked once when job was completed normally with the specified [value],
+     * right before all the waiters for coroutine's completion are notified.
      */
     protected open fun onCompleted(value: T) {}
 
     /**
-     * This function is invoked once when job was cancelled with the specified [cause].
+     * This function is invoked once when job was cancelled with the specified [cause],
+     * right before all the waiters for coroutine's completion are notified.
+     *
+     * **Note:** the state of the coroutine might not be final yet in this function and should not be queried.
+     * You can use [completionCause] and [completionCauseHandled] to recover parameters that we passed
+     * to this `onCancelled` invocation only when [isCompleted] returns `true`.
+     *
      * @param cause The cancellation (failure) cause
      * @param handled `true` if the exception was handled by parent (always `true` when it is a [CancellationException])
      */
     protected open fun onCancelled(cause: Throwable, handled: Boolean) {}
 
     @Suppress("UNCHECKED_CAST")
-    internal override fun onCompletionInternal(state: Any?, mode: Int) {
+    protected final override fun onCompletionInternal(state: Any?) {
         if (state is CompletedExceptionally)
             onCancelled(state.cause, state.handled)
         else

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

@@ -241,10 +241,10 @@ private class DispatchedCoroutine<in T>(
         }
     }
 
-    override fun onCompletionInternal(state: Any?, mode: Int) {
+    override fun afterCompletionInternal(state: Any?, mode: Int) {
         if (tryResume()) return // completed before getResult invocation -- bail out
         // otherwise, getResult has already commenced, i.e. completed later or in other thread
-        super.onCompletionInternal(state, mode)
+        super.afterCompletionInternal(state, mode)
     }
 
     fun getResult(): Any? {

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

@@ -98,9 +98,9 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
          ~ active coroutine is working (or scheduled to execution)
          >> childCancelled / cancelImpl invoked
        ## CANCELLING: state is Finishing, state.rootCause != null
-        ------ cancelling listeners are not admitted anymore, invokeOnCompletion(onCancellation=true) returns NonDisposableHandle
+        ------ cancelling listeners are not admitted anymore, invokeOnCompletion(onCancelling=true) returns NonDisposableHandle
         ------ new children get immediately cancelled, but are still admitted to the list
-         + onCancellation
+         + onCancelling
          + notifyCancelling (invoke all cancelling listeners -- cancel all children, suspended functions resume with exception)
          + cancelParent (rootCause of cancellation is communicated to the parent, parent is cancelled, too)
          ~ waits for completion of coroutine body
@@ -203,7 +203,9 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
         require(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
         val finalException = synchronized(state) {
+            wasCancelling = state.isCancelling
             val exceptions = state.sealLocked(proposedException)
             val finalCause = getFinalRootCause(state, exceptions)
             if (finalCause != null) addSuppressedExceptions(finalCause, exceptions)
@@ -223,6 +225,10 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
             val handled = cancelParent(finalException) || handleJobException(finalException)
             if (handled) (finalState as CompletedExceptionally).makeHandled()
         }
+        // Process state updates for the final state before the state of the Job is actually set to the final state
+        // to avoid races where outside observer may see the job in the final state, yet exception is not handled yet.
+        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" }
         // And process all post-completion actions
@@ -257,6 +263,8 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
         check(state is Empty || state is JobNode<*>) // only simple state without lists where children can concurrently add
         check(update !is CompletedExceptionally) // only for normal completion
         if (!_state.compareAndSet(state, update.boxIncomplete())) return false
+        onCancelling(null) // simple state is not a failure
+        onCompletionInternal(update)
         completeStateFinalization(state, update, mode)
         return true
     }
@@ -275,14 +283,8 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
         }
         val cause = (update as? CompletedExceptionally)?.cause
         /*
-         * 2) Invoke onCancellation: for resource cancellation resource cancellation etc.
-         *    Only notify is was not notified yet.
-         *    Note: we do not use notifyCancelling here, since we are going to invoke all completion as our next step
-         */
-        if (!state.isCancelling) onCancellation(cause)
-        /*
-         * 3) Invoke completion handlers: .join(), callbacks etc.
-         *    It's important to invoke them only AFTER exception handling, see #208
+         * 2) Invoke completion handlers: .join(), callbacks etc.
+         *    It's important to invoke them only AFTER exception handling and everything else, see #208
          */
         if (state is JobNode<*>) { // SINGLE/SINGLE+ state -- one completion handler (common case)
             try {
@@ -294,16 +296,15 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
             state.list?.notifyCompletion(cause)
         }
         /*
-         * 4) Invoke onCompletionInternal: onNext(), timeout de-registration etc.
-         *    It should be last so all callbacks observe consistent state
-         *    of the job which doesn't depend on callback scheduling.
+         * 3) Resumes the rest of the code in scoped coroutines
+         *    (runBlocking, coroutineScope, withContext, withTimeout, etc)
          */
-        onCompletionInternal(update, mode)
+        afterCompletionInternal(update, mode)
     }
 
     private fun notifyCancelling(list: NodeList, cause: Throwable) {
         // first cancel our own children
-        onCancellation(cause)
+        onCancelling(cause)
         notifyHandlers<JobCancellingNode<*>>(list, cause)
         // then cancel parent
         cancelParent(cause) // tentative cancellation -- does not matter if there is no parent
@@ -671,7 +672,7 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
                             val causeException = causeExceptionCache ?: createCauseException(cause).also { causeExceptionCache = it }
                             state.addExceptionLocked(causeException)
                         }
-                        // take cause for notification is was not cancelling before
+                        // take cause for notification if was not in cancelling state before
                         state.rootCause.takeIf { !wasCancelling }
                     }
                     notifyRootCause?.let { notifyCancelling(state.list, it) }
@@ -890,12 +891,21 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
     }
 
     /**
-     * This function is invoked once when job is being cancelled, fails, or is completed.
-     * It's an optimization for [invokeOnCompletion] with `onCancellation` set to `true`.
+     * This function is invoked once as soon as this job is being cancelled for any reason or completes,
+     * similarly to [invokeOnCompletion] with `onCancelling` set to `true`.
+     *
+     * The meaning of [cause] parameter:
+     * * Cause is `null` when job has completed normally.
+     * * Cause is an instance of [CancellationException] when job was cancelled _normally_.
+     *   **It should not be treated as an error**. In particular, it should not be reported to error logs.
+     * * Otherwise, the job had been cancelled or failed with exception.
+     *
+     * The specified [cause] is not the final cancellation cause of this job.
+     * A job may produce other exceptions while it is failing and the final cause might be different.
      *
      * @suppress **This is unstable API and it is subject to change.*
      */
-    protected open fun onCancellation(cause: Throwable?) {}
+    protected open fun onCancelling(cause: Throwable?) {}
 
     /**
      * When this function returns `true` the parent is cancelled on cancellation of this job.
@@ -940,13 +950,24 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
     }
 
     /**
-     * Override for post-completion actions that need to do something with the state.
+     * Override for completion actions that need to update some external object depending on job's state,
+     * right before all the waiters for coroutine's completion are notified.
+     *
+     * @param state the final state.
+     *
+     * @suppress **This is unstable API and it is subject to change.**
+     */
+    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.
      *
      * @suppress **This is unstable API and it is subject to change.**
      */
-    internal open fun onCompletionInternal(state: Any?, mode: Int) {}
+    protected open fun afterCompletionInternal(state: Any?, mode: Int) {}
 
     // for nicer debugging
     public override fun toString(): String =

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

@@ -95,7 +95,7 @@ private open class TimeoutCoroutine<U, in T: U>(
     }
 
     @Suppress("UNCHECKED_CAST")
-    internal override fun onCompletionInternal(state: Any?, mode: Int) {
+    override fun afterCompletionInternal(state: Any?, mode: Int) {
         if (state is CompletedExceptionally)
             uCont.resumeUninterceptedWithExceptionMode(state.cause, mode)
         else

kotlinx-coroutines-core/common/src/internal/Scopes.kt

@@ -23,7 +23,7 @@ internal open class ScopeCoroutine<in T>(
         get() = false // it throws exception to parent instead of cancelling it
 
     @Suppress("UNCHECKED_CAST")
-    internal override fun onCompletionInternal(state: Any?, mode: Int) {
+    override fun afterCompletionInternal(state: Any?, mode: Int) {
         if (state is CompletedExceptionally) {
             val exception = if (mode == MODE_IGNORE) state.cause else recoverStackTrace(state.cause, uCont)
             uCont.resumeUninterceptedWithExceptionMode(exception, mode)

kotlinx-coroutines-core/common/test/AbstractCoroutineTest.kt

@@ -18,14 +18,14 @@ class AbstractCoroutineTest : TestBase() {
                 expect(3)
             }
 
-            override fun onCancellation(cause: Throwable?) {
+            override fun onCancelling(cause: Throwable?) {
                 assertEquals(null, cause)
                 expect(5)
             }
 
             override fun onCompleted(value: String) {
                 assertEquals("OK", value)
-                expect(8)
+                expect(6)
             }
 
             override fun onCancelled(cause: Throwable, handled: Boolean) {
@@ -35,12 +35,12 @@ class AbstractCoroutineTest : TestBase() {
 
         coroutine.invokeOnCompletion(onCancelling = true) {
             assertEquals(null, it)
-            expect(6)
+            expect(7)
         }
 
         coroutine.invokeOnCompletion {
             assertEquals(null, it)
-            expect(7)
+            expect(8)
         }
         expect(2)
         coroutine.start()
@@ -58,7 +58,7 @@ class AbstractCoroutineTest : TestBase() {
                 expect(3)
             }
 
-            override fun onCancellation(cause: Throwable?) {
+            override fun onCancelling(cause: Throwable?) {
                 assertTrue(cause is TestException1)
                 expect(5)
             }
@@ -69,7 +69,7 @@ class AbstractCoroutineTest : TestBase() {
 
             override fun onCancelled(cause: Throwable, handled: Boolean) {
                 assertTrue(cause is TestException1)
-                expect(9)
+                expect(8)
             }
         }
 
@@ -80,7 +80,7 @@ class AbstractCoroutineTest : TestBase() {
 
         coroutine.invokeOnCompletion {
             assertTrue(it is TestException1)
-            expect(8)
+            expect(9)
         }
 
         expect(2)

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

@@ -61,7 +61,7 @@ private class BlockingCoroutine<T>(
     override val cancelsParent: Boolean
         get() = false // it throws exception to parent instead of cancelling it
 
-    override fun onCompletionInternal(state: Any?, mode: Int) {
+    override fun afterCompletionInternal(state: Any?, mode: Int) {
         // wake up blocked thread
         if (Thread.currentThread() != blockedThread)
             LockSupport.unpark(blockedThread)

kotlinx-coroutines-core/jvm/src/channels/Actor.kt

@@ -129,7 +129,7 @@ private open class ActorCoroutine<E>(
 ) : ChannelCoroutine<E>(parentContext, channel, active), ActorScope<E> {
     override val cancelsParent: Boolean get() = true
 
-    override fun onCancellation(cause: Throwable?) {
+    override fun onCancelling(cause: Throwable?) {
         _channel.cancel(cause?.let {
             it as? CancellationException ?: CancellationException("$classSimpleName was cancelled", it)
         })

kotlinx-coroutines-core/jvm/test/exceptions/SuppressionTests.kt

@@ -21,7 +21,7 @@ class SuppressionTests : TestBase() {
                 expect(3)
             }
 
-            override fun onCancellation(cause: Throwable?) {
+            override fun onCancelling(cause: Throwable?) {
                 assertTrue(cause is ArithmeticException)
                 assertTrue(cause.suppressed.isEmpty())
                 expect(5)
@@ -34,7 +34,7 @@ class SuppressionTests : TestBase() {
             override fun onCancelled(cause: Throwable, handled: Boolean) {
                 assertTrue(cause is ArithmeticException)
                 checkException<IOException>(cause.suppressed[0])
-                expect(9)
+                expect(8)
             }
         }
 
@@ -47,7 +47,7 @@ class SuppressionTests : TestBase() {
         coroutine.invokeOnCompletion {
             assertTrue(it is ArithmeticException)
             checkException<IOException>(it.suppressed[0])
-            expect(8)
+            expect(9)
         }
 
         expect(2)