JobSupport cancellation rework:

Change cancel contract: if it returns true, then Job is responsible for handling given exception
Introduce handleJobException as the way to handle "final" exception on job completion
Do not report intermediate exceptions to exception handler
Propagate exceptions up to the Job which is responsible for exception handling
Change handleCoroutineException contract, so custom exception handlers can't break coroutines machinery
Unwrap JobCancellationException to provide exception transparency over whole call chain

Fixes #333, #452 and #451

Author: qwwdfsad

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

@@ -110,7 +110,7 @@ public abstract class AbstractCoroutine<in T>(
         makeCompletingOnce(CompletedExceptionally(exception), defaultResumeMode)
     }
 
-    internal final override fun handleException(exception: Throwable) {
+    internal final override fun handleOnCompletionException(exception: Throwable) {
         handleCoroutineException(parentContext, exception)
     }
 

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

@@ -157,9 +157,15 @@ private open class StandaloneCoroutine(
     active: Boolean
 ) : AbstractCoroutine<Unit>(parentContext, active) {
     override fun hasOnFinishingHandler(update: Any?) = update is CompletedExceptionally
+
+    override fun handleJobException(exception: Throwable) {
+        handleCoroutineException(parentContext, exception)
+    }
+
     override fun onFinishingInternal(update: Any?) {
-        // note the use of the parent's job context below!
-        if (update is CompletedExceptionally) handleCoroutineException(parentContext, update.cause)
+        if (update is CompletedExceptionally && update.cause !is CancellationException) {
+            parentContext[Job]?.cancel(update.cause)
+        }
     }
 }
 

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

@@ -12,26 +12,36 @@ internal expect fun handleCoroutineExceptionImpl(context: CoroutineContext, exce
  * Helper function for coroutine builder implementations to handle uncaught exception in coroutines.
  *
  * It tries to handle uncaught exception in the following way:
- * * If there is [CoroutineExceptionHandler] in the context, then it is used.
- * * Otherwise, if exception is [CancellationException] then it is ignored
- *   (because that is the supposed mechanism to cancel the running coroutine)
- * * Otherwise:
- *     * if there is a [Job] in the context, then [Job.cancel] is invoked;
- *     * all instances of [CoroutineExceptionHandler] found via [ServiceLoader] are invoked;
- *     * current thread's [Thread.uncaughtExceptionHandler] is invoked.
+ * If current exception is [CancellationException], it's ignored: [CancellationException] is a normal way to cancel
+ * coroutine.
+ *
+ * If there is a [Job] in the context, then [Job.cancel] is invoked.
+ * If invocation returned `true`, method terminates: now [Job] is responsible for handling an exception.
+ * Otherwise, If there is [CoroutineExceptionHandler] in the context, it is used.
+ * Otherwise all instances of [CoroutineExceptionHandler] found via [ServiceLoader] and [Thread.uncaughtExceptionHandler] are invoked
  */
 public fun handleCoroutineException(context: CoroutineContext, exception: Throwable) {
     // if exception handling fails, make sure the original exception is not lost
     try {
+
+        // Ignore CancellationException (they are normal ways to terminate a coroutine)
+        if (exception is CancellationException) {
+            return
+        }
+
+        // If parent is successfully cancelled, we're done, it is now its responsibility to handle the exception
+        val parent = context[Job]
+        if (parent != null && parent.cancel(exception)) {
+            return
+        }
+
+        // If not, invoke exception handler from the context
         context[CoroutineExceptionHandler]?.let {
             it.handleException(context, exception)
             return
         }
-        // ignore CancellationException (they are normal means to terminate a coroutine)
-        if (exception is CancellationException) return
-        // try cancel job in the context
-        context[Job]?.cancel(exception)
-        // platform-specific
+
+        // If handler is not present in the context, fallback to the global handler
         handleCoroutineExceptionImpl(context, exception)
     } catch (handlerException: Throwable) {
         // simply rethrow if handler threw the original exception

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

@@ -149,10 +149,12 @@ public interface Job : CoroutineContext.Element {
     public fun start(): Boolean
 
     /**
-     * Cancels this job with an optional cancellation [cause]. The result is `true` if this job was
-     * cancelled as a result of this invocation and `false` otherwise
-     * (if it was already _completed_ or if it is [NonCancellable]).
-     * Repeated invocations of this function have no effect and always produce `false`.
+     * Cancels this job with an optional cancellation [cause].
+     * The result is `true` if this job was either cancelled as a result of this invocation
+     * or it's being cancelled and given [cause] was successfully received by the job and will be properly handled, `false` otherwise.
+     *
+     * If this method returned `false`, then caller is responsible for handling [cause].
+     * If job is already completed, method returns `false`
      *
      * When cancellation has a clear reason in the code, an instance of [CancellationException] should be created
      * at the corresponding original cancellation site and passed into this method to aid in debugging by providing

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

@@ -155,12 +155,107 @@ internal open class JobSupport constructor(active: Boolean) : Job, SelectClause0
      * If this method succeeds, state of this job will never be changed again
      */
     private fun tryFinalizeState(expect: Incomplete, proposedUpdate: Any?, mode: Int): Boolean {
+        /*
+         * If job is in 'cancelling' state and we're finalizing job state, we start intricate dance:
+         * 1) Synchronize on state to avoid races with concurrent
+         *    mutations (e.g. when new child is added)
+         * 2) After synchronization check we're still in the expected state
+         * 3) Aggregate final exception under the same lock which protects exceptions
+         *    collection
+         * 4) Pass it upstream
+         */
+        if (expect is Finishing && expect.cancelled != null) {
+            val finalException = synchronized(expect) {
+                if (_state.value !== expect) {
+                    return false
+                }
+
+                if (proposedUpdate is CompletedExceptionally) {
+                    expect.addLocked(proposedUpdate.cause)
+                }
+
+                /*
+                 * Note that new exceptions cannot be added concurrently: state is guarded by lock
+                 * and storage is sealed in the end, so all new exceptions will be reported separately
+                 */
+                buildException(expect).also { expect.seal() }
+            }
+
+            val update = Cancelled(this, finalException ?: expect.cancelled.cause)
+            handleJobException(update.cause)
+            // This CAS never fails: we're in the state when no jobs can be attached, because state is already sealed
+            if (!tryFinalizeState(expect, update)) {
+                val error = AssertionError("Unexpected state: ${_state.value}, expected: $expect, update: $update")
+                handleOnCompletionException(error)
+                throw error
+            }
+
+            completeStateFinalization(expect, update, mode)
+            return true
+        }
+
         val update = coerceProposedUpdate(expect, proposedUpdate)
         if (!tryFinalizeState(expect, update)) return false
         completeStateFinalization(expect, update, mode)
         return true
     }
 
+    private fun buildException(state: Finishing): Throwable? {
+        val cancelled = state.cancelled!!
+        val suppressed = state.exceptions
+
+        /*
+         * This is a place where we step on our API limitation:
+         * We can't distinguish internal JobCancellationException from our parent
+         * from external cancellation, thus we ought to collect all exceptions.
+         *
+         * But it has negative consequences: same exception can be added as suppressed more than once.
+         * Consider concurrent parent-child relationship:
+         * 1) Child throws E1 and parent throws E2
+         * 2) Parent goes to "Cancelling(E1)" and cancels child with E1
+         * 3) Child goes to "Cancelling(E1)", but throws an exception E2
+         * 4) When child throws, it notifies parent that he is cancelling, adding its exception to parent list of exceptions
+         *    (again, parent don't know whether it's child exception or external exception)
+         * 5) Child builds final exception: E1 with suppressed E2, reports it to parent
+         * 6) Parent aggregates three exceptions: original E1, reported E2 and "final" E1.
+         *    It filters the third exception, but adds the second one to the first one, thus adding suppressed duplicate.
+         *
+         * Note that it's only happening when both parent and child throw exception simultaneously
+         */
+        var rootCause = cancelled.cause
+        if (rootCause is JobCancellationException) {
+            val cause = unwrap(rootCause)
+            rootCause = if (cause !== null) {
+                cause
+            } else {
+                suppressed.firstOrNull { unwrap(it) != null } ?: return rootCause
+            }
+        }
+
+        val seenExceptions = HashSet<Throwable>() // TODO it should be identity set
+        suppressed.forEach {
+            val unwrapped = unwrap(it)
+            if (unwrapped !== null && unwrapped !== rootCause) {
+                if (seenExceptions.add(unwrapped)) {
+                    rootCause.addSuppressedThrowable(unwrapped)
+                }
+            }
+        }
+
+        return rootCause
+    }
+
+    private tailrec fun unwrap(exception: Throwable): Throwable? {
+        if (exception is JobCancellationException) {
+            val cause = exception.cause
+            if (cause !== null) return unwrap(cause)
+            return null
+
+        } else {
+            return exception
+        }
+    }
+
     /**
      * Tries to update [_state] of this job to the final state and, if
      * succeeds, disposes parent handle (de-attaching child from parent)
@@ -200,7 +295,7 @@ internal open class JobSupport constructor(active: Boolean) : Job, SelectClause0
             try {
                 expect.invoke(cause)
             } catch (ex: Throwable) {
-                handleException(CompletionHandlerException("Exception in completion handler $expect for $this", ex))
+                handleOnCompletionException(CompletionHandlerException("Exception in completion handler $expect for $this", ex))
             }
         } else {
             expect.list?.notifyCompletion(cause)
@@ -225,8 +320,7 @@ internal open class JobSupport constructor(active: Boolean) : Job, SelectClause0
         if (proposedUpdate !is CompletedExceptionally) return cancelled // not exception -- just use original cancelled
         val exception = proposedUpdate.cause
         if (cancelled.cause == exception) return cancelled // that is the cancelled we need already!
-        // todo: We need to rework this logic to keep original cancellation cause in the state and suppress other exceptions
-        //       that could have occurred while coroutine is being cancelled.
+        // That could have occurred while coroutine is being cancelled.
         // Do not spam with JCE in suppressed exceptions
         if (cancelled.cause !is JobCancellationException) {
             exception.addSuppressedThrowable(cancelled.cause)
@@ -251,7 +345,7 @@ internal open class JobSupport constructor(active: Boolean) : Job, SelectClause0
                 }
             }
         }
-        exception?.let { handleException(it) }
+        exception?.let { handleOnCompletionException(it) }
     }
 
     public final override fun start(): Boolean {
@@ -509,7 +603,15 @@ internal open class JobSupport constructor(active: Boolean) : Job, SelectClause0
                     }
                 }
                 is Finishing -> { // Completing/Cancelling the job, may cancel
-                    if (state.cancelled != null) return false // already cancelling
+                    if (state.cancelled != null) {
+                        if (cause == null) {
+                            return true
+                        }
+
+                        // We either successfully added an exception or caller should handle it itself
+                        return cause.let { state.addException(it) }
+                    }
+
                     if (tryMakeCancelling(state, state.list, cause)) return true
                 }
                 else -> { // is inactive
@@ -666,20 +768,24 @@ internal open class JobSupport constructor(active: Boolean) : Job, SelectClause0
      * installed via [invokeOnCompletion].
      * @suppress **This is unstable API and it is subject to change.**
      */
-    internal open fun handleException(exception: Throwable) {
+    internal open fun handleOnCompletionException(exception: Throwable) {
         throw exception
     }
 
     /**
-     * This function is invoked once when job is cancelled or is completed, similarly to [invokeOnCompletion] with
-     * `onCancelling` set to `true`.
+     * This function is invoked once when job is cancelled or is completed.
+     * It's an optimization for [invokeOnCompletion] with `onCancelling` set to `true`.
+     *
      * @param exceptionally not null when the the job was cancelled or completed exceptionally,
      *               null when it has completed normally.
-     * @suppress **This is unstable API and it is subject to change.**
+     * @suppress **This is unstable API and it is subject to change.*
      */
-    internal open fun onCancellationInternal(exceptionally: CompletedExceptionally?) {}
+    internal open fun onCancellationInternal(exceptionally: CompletedExceptionally?) {
+        // TODO rename to "onCancelling"
+    }
 
     /**
+     * Whether job has [onFinishingInternal] handler for given [update]
      * @suppress **This is unstable API and it is subject to change.**
      */
     internal open fun hasOnFinishingHandler(update: Any?) = false
@@ -689,6 +795,12 @@ internal open class JobSupport constructor(active: Boolean) : Job, SelectClause0
      */
     internal open fun onFinishingInternal(update: Any?) {}
 
+    /**
+     * Method which is invoked once Job becomes `Cancelled`. It's guaranteed that at the moment
+     * of invocation the job and all its children are complete
+     */
+    internal open fun handleJobException(exception: Throwable) {}
+
     /**
      * Override for post-completion actions that need to do something with the state.
      * @param mode completion mode.
@@ -726,6 +838,36 @@ internal open class JobSupport constructor(active: Boolean) : Job, SelectClause0
         @JvmField val completing: Boolean /* true when completing */
     ) : Incomplete {
         override val isActive: Boolean get() = cancelled == null
+        val exceptions: List<Throwable> get() = _exceptionsHolder as List<Throwable>
+
+        // TODO optimize
+        private var _exceptionsHolder: Any? = if (cancelled == null) null else ArrayList<Throwable>(2)
+
+        fun addException(exception: Throwable): Boolean {
+            synchronized(this) {
+                return if (_exceptionsHolder == null) {
+                    false
+                } else {
+                    @Suppress("UNCHECKED_CAST")
+                    (_exceptionsHolder as MutableList<Throwable>).add(exception)
+                    true
+                }
+            }
+        }
+
+        fun addLocked(exception: Throwable) {
+            (_exceptionsHolder as MutableList<Throwable>).add(exception)
+        }
+
+        /**
+         * Seals current state. After [seal] call all consecutive calls to [addException]
+         * return `false` forcing callers to handle pending exception by themselves.
+         * This call should be guarded by `synchronized(finishingState)`
+         */
+        fun seal() {
+            _exceptionsHolder = null
+        }
+
     }
 
     private val Incomplete.isCancelling: Boolean
@@ -854,6 +996,12 @@ private class Empty(override val isActive: Boolean) : Incomplete {
 internal class JobImpl(parent: Job? = null) : JobSupport(true) {
     init { initParentJobInternal(parent) }
     override val onCancelMode: Int get() = ON_CANCEL_MAKE_COMPLETING
+
+    override fun cancel(cause: Throwable?): Boolean {
+        // JobImpl can't handle an exception, thus always returns false
+        super.cancel(cause)
+        return false
+    }
 }
 
 // -------- invokeOnCompletion nodes

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

@@ -8,6 +8,7 @@ import kotlin.coroutines.experimental.*
 import kotlin.test.*
 
 class AbstractCoroutineTest : TestBase() {
+
     @Test
     fun testNotifications() = runTest {
         expect(1)
@@ -67,18 +68,21 @@ class AbstractCoroutineTest : TestBase() {
             }
 
             override fun onCompletedExceptionally(exception: Throwable) {
-                assertTrue(exception is TestException1)
+                assertTrue(exception is TestException0)
                 expect(9)
             }
         }
+
         coroutine.invokeOnCompletion(onCancelling = true) {
             assertTrue(it is TestException0)
             expect(6)
         }
+
         coroutine.invokeOnCompletion {
-            assertTrue(it is TestException1)
+            assertTrue(it is TestException0)
             expect(8)
         }
+
         expect(2)
         coroutine.start()
         expect(4)

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

@@ -10,6 +10,7 @@ import kotlin.coroutines.experimental.*
 import kotlin.test.*
 
 class AsyncLazyTest : TestBase() {
+
     @Test
     fun testSimple() = runTest {
         expect(1)
@@ -180,7 +181,7 @@ class AsyncLazyTest : TestBase() {
         assertTrue(d.isActive && !d.isCompleted && !d.isCancelled)
         assertTrue(d.cancel())
         assertTrue(!d.isActive && !d.isCompletedExceptionally && d.isCancelled) // cancelling !
-        assertTrue(!d.cancel())
+        assertTrue(d.cancel())
         assertTrue(!d.isActive && !d.isCompletedExceptionally && d.isCancelled) // still cancelling
         finish(6)
         assertTrue(d.await() == 42) // await shall throw CancellationException