Improved coroutine exception handling logic

* JobSupport.handleJobException (which is called before the coroutine goes
  to the final state) is now called only when parent did not handle
  exception and is used only in launch-like coroutines (launch and actor)
  to report uncaught exception.
* The final result of other types coroutines that, by default, have an
  object "to store exceptions" is consistently processed after all
  other notifications when the coroutine state is already complete and
  is reported as uncaught exception as a last-resort processing tactic
  when it cannot be otherwise handled.
* The above change makes the order of notifications for future { ... } and
  other async-like coroutine builders consistent with the order in which
  async { ... }.await() is notified.
* The "handled" state of exception (whether it was processed or not)
  is now stored in CompletedExceptionally marker class (pulled this
  field up from CancelledContinuation).
* AbstractCoroutine.onCompletedExceptionally is renamed to onCancelled
  and includes "handled" flag.
* protected val JobSupport.completionCauseHandled is introduced to access
  "handled" flag, which simplified reactive streams integration.
* "suppressed" flag is dropped from JobSupport.onCompletionInternal.
* Fixed exception processing in reactive integrations -- exception that
  is thrown from a cancelled (disposed) coroutine is not lost anymore.

Author: elizarov

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

@@ -6,8 +6,8 @@ public abstract class kotlinx/coroutines/AbstractCoroutine : kotlinx/coroutines/
 	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 fun onCompletedExceptionally (Ljava/lang/Throwable;)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
@@ -382,11 +382,12 @@ public class kotlinx/coroutines/JobSupport : kotlinx/coroutines/ChildJob, kotlin
 	public fun getChildJobCancellationCause ()Ljava/lang/Throwable;
 	public final fun getChildren ()Lkotlin/sequences/Sequence;
 	protected final fun getCompletionCause ()Ljava/lang/Throwable;
+	protected final fun getCompletionCauseHandled ()Z
 	public final fun getCompletionExceptionOrNull ()Ljava/lang/Throwable;
 	protected fun getHandlesException ()Z
 	public final fun getKey ()Lkotlin/coroutines/CoroutineContext$Key;
 	public final fun getOnJoin ()Lkotlinx/coroutines/selects/SelectClause0;
-	protected fun handleJobException (Ljava/lang/Throwable;Z)V
+	protected fun handleJobException (Ljava/lang/Throwable;)Z
 	public final fun invokeOnCompletion (Lkotlin/jvm/functions/Function1;)Lkotlinx/coroutines/DisposableHandle;
 	public final fun invokeOnCompletion (ZZLkotlin/jvm/functions/Function1;)Lkotlinx/coroutines/DisposableHandle;
 	public fun isActive ()Z

integration/kotlinx-coroutines-guava/src/ListenableFuture.kt

@@ -68,10 +68,10 @@ private class ListenableFutureCoroutine<T>(
         future.set(value)
     }
 
-    override fun handleJobException(exception: Throwable, handled: Boolean) {
-        if (!future.setException(exception) && !handled) {
+    override fun onCancelled(cause: Throwable, handled: Boolean) {
+        if (!future.setException(cause) && !handled) {
             // prevents loss of exception that was not handled by parent & could not be set to SettableFuture
-            handleCoroutineException(context, exception)
+            handleCoroutineException(context, cause)
         }
     }
 }

integration/kotlinx-coroutines-jdk8/src/future/Future.kt

@@ -57,10 +57,10 @@ private class CompletableFutureCoroutine<T>(
         future.complete(value)
     }
 
-    override fun handleJobException(exception: Throwable, handled: Boolean) {
-        if (!future.completeExceptionally(exception) && !handled) {
+    override fun onCancelled(cause: Throwable, handled: Boolean) {
+        if (!future.completeExceptionally(cause) && !handled) {
             // prevents loss of exception that was not handled by parent & could not be set to CompletableFuture
-            handleCoroutineException(context, exception)
+            handleCoroutineException(context, cause)
         }
     }
 }

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

@@ -24,7 +24,7 @@ import kotlin.jvm.*
  * * [onCancellation] is invoked as soon as coroutine is _failing_, or is cancelled,
  *   or when it completes for any reason.
  * * [onCompleted] is invoked when coroutine completes with a value.
- * * [onCompletedExceptionally] in invoked when coroutines completes with exception.
+ * * [onCancelled] in invoked when coroutines completes with exception (cancelled).
  *
  * @param parentContext context of the parent coroutine.
  * @param active when `true` (by default) coroutine is created in _active_ state, when `false` in _new_ state.
@@ -89,20 +89,21 @@ public abstract class AbstractCoroutine<in T>(
     protected override fun onCancellation(cause: Throwable?) {}
 
     /**
-     * This function is invoked once when job is completed normally with the specified [value].
+     * This function is invoked once when job was completed normally with the specified [value].
      */
     protected open fun onCompleted(value: T) {}
 
     /**
-     * This function is invoked once when job is completed exceptionally with the specified [exception].
+     * This function is invoked once when job was cancelled with the specified [cause].
+     * @param cause The cancellation (failure) cause
+     * @param handled `true` if the exception was handled by parent (always `true` when it is a [CancellationException])
      */
-    // todo: rename to onCancelled
-    protected open fun onCompletedExceptionally(exception: Throwable) {}
+    protected open fun onCancelled(cause: Throwable, handled: Boolean) {}
 
     @Suppress("UNCHECKED_CAST")
-    internal override fun onCompletionInternal(state: Any?, mode: Int, suppressed: Boolean) {
+    internal override fun onCompletionInternal(state: Any?, mode: Int) {
         if (state is CompletedExceptionally)
-            onCompletedExceptionally(state.cause)
+            onCancelled(state.cause, state.handled)
         else
             onCompleted(state as T)
     }

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

@@ -179,8 +179,9 @@ private open class StandaloneCoroutine(
     parentContext: CoroutineContext,
     active: Boolean
 ) : AbstractCoroutine<Unit>(parentContext, active) {
-    override fun handleJobException(exception: Throwable, handled: Boolean) {
-        if (!handled) handleCoroutineException(context, exception)
+    override fun handleJobException(exception: Throwable): Boolean {
+        handleCoroutineException(context, exception)
+        return true
     }
 }
 
@@ -240,10 +241,10 @@ private class DispatchedCoroutine<in T>(
         }
     }
 
-    override fun onCompletionInternal(state: Any?, mode: Int, suppressed: Boolean) {
+    override fun onCompletionInternal(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, suppressed)
+        super.onCompletionInternal(state, mode)
     }
 
     fun getResult(): Any? {

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

@@ -18,8 +18,12 @@ internal fun <T> Result<T>.toState(): Any? =
  *        or artificial [CancellationException] if no cause was provided
  */
 internal open class CompletedExceptionally(
-    @JvmField public val cause: Throwable
+    @JvmField public val cause: Throwable,
+    handled: Boolean = false
 ) {
+    private val _handled = atomic(handled)
+    val handled: Boolean get() = _handled.value
+    fun makeHandled(): Boolean = _handled.compareAndSet(false, true)
     override fun toString(): String = "$classSimpleName[$cause]"
 }
 
@@ -34,10 +38,7 @@ internal class CancelledContinuation(
     continuation: Continuation<*>,
     cause: Throwable?,
     handled: Boolean
-) : CompletedExceptionally(cause ?: CancellationException("Continuation $continuation was cancelled normally")) {
-    private val resumed = atomic(false)
-    private val handled = atomic(handled)
-
-    fun makeResumed(): Boolean = resumed.compareAndSet(false, true)
-    fun makeHandled(): Boolean = handled.compareAndSet(false, true)
+) : CompletedExceptionally(cause ?: CancellationException("Continuation $continuation was cancelled normally"), handled) {
+    private val _resumed = atomic(false)
+    fun makeResumed(): Boolean = _resumed.compareAndSet(false, true)
 }

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

@@ -118,7 +118,7 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
         ------ completion listeners are not admitted anymore, invokeOnCompletion returns NonDisposableHandle
          + parentHandle.dispose
          + notifyCompletion (invoke all completion listeners)
-         + onCompletionInternal / onCompleted / onCompletedExceptionally
+         + onCompletionInternal / onCompleted / onCancelled
 
        ---------------------------------------------------------------------------------
      */
@@ -193,22 +193,20 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
     // ## IMPORTANT INVARIANT: Only one thread can be concurrently invoking this method.
     private fun tryFinalizeFinishingState(state: Finishing, proposedUpdate: Any?, mode: Int): Boolean {
         /*
-         * Note: proposed state can be Incompleted, e.g.
+         * Note: proposed state can be Incomplete, e.g.
          * async {
-         *   smth.invokeOnCompletion {} // <- returns handle which implements Incomplete under the hood
+         *     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
         val proposedException = (proposedUpdate as? CompletedExceptionally)?.cause
         // Create the final exception and seal the state so that no more exceptions can be added
-        var suppressed = false
         val finalException = synchronized(state) {
             val exceptions = state.sealLocked(proposedException)
             val finalCause = getFinalRootCause(state, exceptions)
-            // Report suppressed exceptions if initial cause doesn't match final cause (due to JCE unwrapping)
-            if (finalCause != null) suppressed = suppressExceptions(finalCause, exceptions) || finalCause !== state.rootCause
+            if (finalCause != null) addSuppressedExceptions(finalCause, exceptions)
             finalCause
         }
         // Create the final state object
@@ -222,13 +220,13 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
         }
         // Now handle the final exception
         if (finalException != null) {
-            val handledByParent = cancelParent(finalException)
-            handleJobException(finalException, handledByParent)
+            val handled = cancelParent(finalException) || handleJobException(finalException)
+            if (handled) (finalState as CompletedExceptionally).makeHandled()
         }
         // 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
-        completeStateFinalization(state, finalState, mode, suppressed)
+        completeStateFinalization(state, finalState, mode)
         return true
     }
 
@@ -243,31 +241,28 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
         return exceptions.firstOrNull { it !is CancellationException } ?: exceptions[0]
     }
 
-    private fun suppressExceptions(rootCause: Throwable, exceptions: List<Throwable>): Boolean {
-        if (exceptions.size <= 1) return false // nothing more to do here
+    private fun addSuppressedExceptions(rootCause: Throwable, exceptions: List<Throwable>) {
+        if (exceptions.size <= 1) return // nothing more to do here
         val seenExceptions = identitySet<Throwable>(exceptions.size)
-        var suppressed = false
         for (exception in exceptions) {
             val unwrapped = unwrap(exception)
             if (unwrapped !== rootCause && unwrapped !is CancellationException && seenExceptions.add(unwrapped)) {
                 rootCause.addSuppressedThrowable(unwrapped)
-                suppressed = true
             }
         }
-        return suppressed
     }
 
     // fast-path method to finalize normally completed coroutines without children
     private fun tryFinalizeSimpleState(state: Incomplete, update: Any?, mode: Int): Boolean {
         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
-        completeStateFinalization(state, update, mode, false)
+        completeStateFinalization(state, update, mode)
         return true
     }
 
     // suppressed == true when any exceptions were suppressed while building the final completion cause
-    private fun completeStateFinalization(state: Incomplete, update: Any?, mode: Int, suppressed: Boolean) {
+    private fun completeStateFinalization(state: Incomplete, update: Any?, mode: Int) {
         /*
          * Now the job in THE FINAL state. We need to properly handle the resulting state.
          * Order of various invocations here is important.
@@ -303,7 +298,7 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
          *    It should be last so all callbacks observe consistent state
          *    of the job which doesn't depend on callback scheduling.
          */
-        onCompletionInternal(update, mode, suppressed)
+        onCompletionInternal(update, mode)
     }
 
     private fun notifyCancelling(list: NodeList, cause: Throwable) {
@@ -387,18 +382,21 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
      * [cancel] cause, [CancellationException] or **`null` if this job had completed normally**.
      * This function throws [IllegalStateException] when invoked for an job that has not [completed][isCompleted] nor
      * is being cancelled yet.
-     *
-     * @suppress **This is unstable API and it is subject to change.**
      */
-    protected fun getCompletionCause(): Throwable? = loopOnState { state ->
-        return when (state) {
+    protected val completionCause: Throwable?
+        get() = when (val state = state) {
             is Finishing -> state.rootCause
                 ?: error("Job is still new or active: $this")
             is Incomplete -> error("Job is still new or active: $this")
             is CompletedExceptionally -> state.cause
             else -> null
         }
-    }
+
+    /**
+     * Returns `true` when [completionCause] exception was handled by parent coroutine.
+     */
+    protected val completionCauseHandled: Boolean
+        get() = state.let { it is CompletedExceptionally && it.handled }
 
     @Suppress("OverridingDeprecatedMember")
     public final override fun invokeOnCompletion(handler: CompletionHandler): DisposableHandle =
@@ -859,8 +857,7 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
     }
 
     public final override val children: Sequence<Job> get() = sequence {
-        val state = this@JobSupport.state
-        when (state) {
+        when (val state = this@JobSupport.state) {
             is ChildHandleNode -> yield(state.childJob)
             is Incomplete -> state.list?.let { list ->
                 list.forEach<ChildHandleNode> { yield(it.childJob) }
@@ -885,6 +882,7 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
     /**
      * Override to process any exceptions that were encountered while invoking completion handlers
      * installed via [invokeOnCompletion].
+     *
      * @suppress **This is unstable API and it is subject to change.**
      */
     internal open fun handleOnCompletionException(exception: Throwable) {
@@ -910,7 +908,7 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
     protected open val cancelsParent: Boolean get() = true
 
     /**
-     * Returns `true` for jobs that handle their exceptions via [handleJobException] or integrate them
+     * Returns `true` for jobs that handle their exceptions or integrate them
      * into the job's result via [onCompletionInternal]. The only instance of the [Job] that does not
      * handle its exceptions is [JobImpl] and its subclass [SupervisorJobImpl].
      *
@@ -919,17 +917,18 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
     protected open val handlesException: Boolean get() = true
 
     /**
-     * Handles the final job [exception] after it was reported to the by the parent,
-     * where [handled] is `true` when parent had already handled exception and `false` otherwise.
+     * Handles the final job [exception] that was not handled by the parent coroutine.
+     * Returns `true` if it handles exception (so handling at later stages is not needed).
+     * It is designed to be overridden by launch-like coroutines
+     * (`StandaloneCoroutine` and `ActorCoroutine`) that don't have a result type
+     * that can represent exceptions.
      *
      * This method is invoked **exactly once** when the final exception of the job is determined
      * and before it becomes complete. At the moment of invocation the job and all its children are complete.
      *
-     * Note, [handled] is always `true` when [exception] is [CancellationException].
-     *
      * @suppress **This is unstable API and it is subject to change.*
      */
-    protected open fun handleJobException(exception: Throwable, handled: Boolean) {}
+    protected open fun handleJobException(exception: Throwable): Boolean = false
 
     private fun cancelParent(cause: Throwable): Boolean {
         // CancellationException is considered "normal" and parent is not cancelled when child produces it.
@@ -944,10 +943,10 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
      * Override for post-completion actions that need to do something with the state.
      * @param state the final state.
      * @param mode completion mode.
-     * @param suppressed true when any exceptions were suppressed while building the final completion cause.
+     *
      * @suppress **This is unstable API and it is subject to change.**
      */
-    internal open fun onCompletionInternal(state: Any?, mode: Int, suppressed: Boolean) {}
+    internal open fun onCompletionInternal(state: Any?, mode: Int) {}
 
     // for nicer debugging
     public override fun toString(): String =
@@ -1015,8 +1014,7 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
                 return
             }
             if (exception === rootCause) return // nothing to do
-            val eh = _exceptionsHolder // volatile read
-            when (eh) {
+            when (val eh = _exceptionsHolder) { // volatile read
                 null -> _exceptionsHolder = exception
                 is Throwable -> {
                     if (exception === eh) return // nothing to do

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, suppressed: Boolean) {
+    internal override fun onCompletionInternal(state: Any?, mode: Int) {
         if (state is CompletedExceptionally)
             uCont.resumeUninterceptedWithExceptionMode(state.cause, mode)
         else

kotlinx-coroutines-core/common/src/channels/Broadcast.kt

@@ -109,10 +109,13 @@ private open class BroadcastCoroutine<E>(
         return true // does not matter - result is used in DEPRECATED functions only
     }
 
-    override fun onCompletionInternal(state: Any?, mode: Int, suppressed: Boolean) {
-        val cause = (state as? CompletedExceptionally)?.cause
+    override fun onCompleted(value: Unit) {
+        _channel.close()
+    }
+
+    override fun onCancelled(cause: Throwable, handled: Boolean) {
         val processed = _channel.close(cause)
-        if (cause != null && !processed && suppressed) handleCoroutineException(context, cause)
+        if (!processed && !handled) handleCoroutineException(context, cause)
     }
 }
 

kotlinx-coroutines-core/common/src/channels/Produce.kt

@@ -94,9 +94,12 @@ private class ProducerCoroutine<E>(
     override val isActive: Boolean
         get() = super.isActive
 
-    override fun onCompletionInternal(state: Any?, mode: Int, suppressed: Boolean) {
-        val cause = (state as? CompletedExceptionally)?.cause
+    override fun onCompleted(value: Unit) {
+        _channel.close()
+    }
+
+    override fun onCancelled(cause: Throwable, handled: Boolean) {
         val processed = _channel.close(cause)
-        if (cause != null && !processed && suppressed) handleCoroutineException(context, cause)
+        if (!processed && !handled) handleCoroutineException(context, cause)
     }
 }