Improve coroutine exception handling logic

  * Wrap exceptions into CE if parent cancels its child. This is possible because now we can be sure that no one is invoking 'cancel(Throwable)', thus if parent cancels its children, an exception is definitely handled.
  * This changes allows exceptions in hierarchies to be handled only once even when coroutines in the hierarchy are capable of handling exceptions
  * Breaking change: CompletableJob.completeExceptionally(exception) uses its parameter as additional debug information for children ('cause' of CE), thus changing the result of its children. For example, if CJ was a parent of CompleteableDeferred and CJ was completed exceptionally with exception E1, now CD.await() will throw CE instead of E1

First part of fix for #689

Author: qwwdfsad

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

@@ -24,6 +24,7 @@ public interface CompletableJob : Job {
     /**
      * Completes this job exceptionally with a given [exception]. The result is `true` if this job was
      * completed as a result of this invocation and `false` otherwise (if it was already completed).
+     * [exception] parameter is used as an additional debug information that is not handled by any exception handlers.
      *
      * Subsequent invocations of this function have no effect and always produce `false`.
      *

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

@@ -422,10 +422,13 @@ public interface ParentJob : Job {
      * Child job is using this method to learn its cancellation cause when the parent cancels it with [ChildJob.parentCancelled].
      * This method is invoked only if the child was not already being cancelled.
      *
+     * Note that [CancellationException] is the method's return type: if child is cancelled by its parent,
+     * then the original exception is **already** handled by either the parent or the original source of failure.
+     *
      * @suppress **This is unstable API and it is subject to change.**
      */
     @InternalCoroutinesApi
-    public fun getChildJobCancellationCause(): Throwable
+    public fun getChildJobCancellationCause(): CancellationException
 }
 
 /**

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

@@ -635,29 +635,20 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
     private fun createJobCancellationException() =
         JobCancellationException("Job was cancelled", null, this)
 
-    override fun getChildJobCancellationCause(): Throwable {
+    override fun getChildJobCancellationCause(): CancellationException {
         // determine root cancellation cause of this job (why is it cancelling its children?)
         val state = this.state
         val rootCause = when (state) {
-            is Finishing -> state.rootCause
+            is Finishing -> state.rootCause as? CancellationException ?: JobCancellationException("Parent job was cancelled", state.rootCause, this)
+            is CompletedExceptionally -> state.cause as? CancellationException ?: JobCancellationException("Parent job was cancelled", state.cause, this)
             is Incomplete -> error("Cannot be cancelling child in this state: $state")
-            is CompletedExceptionally -> state.cause
             else -> null // create exception with the below code on normal completion
         }
-        /*
-         * If this parent job handles exceptions, then wrap cause into JobCancellationException, because we
-         * don't want the child to handle this exception on more time. Otherwise, pass our original rootCause
-         * to the child for cancellation.
-         */
-        return if (rootCause == null || handlesException && rootCause !is CancellationException) {
-            JobCancellationException("Parent job is ${stateString(state)}", rootCause, this)
-        } else {
-            rootCause
-        }
+        return rootCause ?: JobCancellationException("Parent job is ${stateString(state)}", rootCause, this)
     }
 
     // cause is Throwable or ParentJob when cancelChild was invoked
-    private fun createCauseException(cause: Any?): Throwable = when(cause) {
+    private fun createCauseException(cause: Any?): Throwable = when (cause) {
         is Throwable? -> cause ?: createJobCancellationException()
         else -> (cause as ParentJob).getChildJobCancellationCause()
     }

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

@@ -171,7 +171,8 @@ class SupervisorTest : TestBase() {
         try {
             deferred.await()
             expectUnreached()
-        } catch (e: TestException1) {
+        } catch (e: CancellationException) {
+            assertTrue(e.cause is TestException1)
             finish(3)
         }
     }

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

@@ -54,7 +54,7 @@ class CapturingHandler : AbstractCoroutineContextElement(CoroutineExceptionHandl
     }
 }
 
-internal fun runBlock(
+internal fun captureExceptionsRun(
     context: CoroutineContext = EmptyCoroutineContext,
     block: suspend CoroutineScope.() -> Unit
 ): Throwable {
@@ -63,7 +63,7 @@ internal fun runBlock(
     return handler.getException()
 }
 
-internal fun runBlockForMultipleExceptions(
+internal fun captureMultipleExceptionsRun(
     context: CoroutineContext = EmptyCoroutineContext,
     block: suspend CoroutineScope.() -> Unit
 ): List<Throwable> {

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

@@ -18,9 +18,9 @@ class JobExceptionHandlingTest : TestBase() {
         /*
          * Root parent: JobImpl()
          * Child: throws ISE
-         * Result: ISE in exception handler
+         * Result: ISE in exception handlerWithContextExceptionHandlingTest
          */
-        val exception = runBlock {
+        val exception = captureExceptionsRun {
             val job = Job()
             launch(job, start = ATOMIC) {
                 expect(2)
@@ -36,43 +36,39 @@ class JobExceptionHandlingTest : TestBase() {
     }
 
     @Test
-    fun testAsyncCancellationWithCause() = runTest {
-        val deferred = async(NonCancellable) {
+    fun testAsyncCancellationWithCauseAndParent() = runTest {
+        val parent = Job()
+        val deferred = async(parent) {
             expect(2)
             delay(Long.MAX_VALUE)
         }
 
         expect(1)
         yield()
-        deferred.cancel(TestCancellationException("TEST"))
+        parent.completeExceptionally(IOException())
         try {
             deferred.await()
             expectUnreached()
-        } catch (e: TestCancellationException) {
-            assertEquals("TEST", e.message)
+        } catch (e: CancellationException) {
             assertTrue(e.suppressed.isEmpty())
+            assertTrue(e.cause?.suppressed?.isEmpty() ?: false)
             finish(3)
         }
     }
 
     @Test
-    fun testAsyncCancellationWithCauseAndParent() = runTest {
+    fun testAsyncCancellationWithCauseAndParentDoesNotTriggerHandling() = runTest {
         val parent = Job()
-        val deferred = async(parent) {
+        val job = launch(parent) {
             expect(2)
             delay(Long.MAX_VALUE)
         }
 
         expect(1)
         yield()
         parent.completeExceptionally(IOException())
-        try {
-            deferred.await()
-            expectUnreached()
-        } catch (e: IOException) {
-            assertTrue(e.suppressed.isEmpty())
-            finish(3)
-        }
+        job.join()
+        finish(3)
     }
 
     @Test
@@ -85,7 +81,7 @@ class JobExceptionHandlingTest : TestBase() {
          *
          * Github issue #354
          */
-        val exception = runBlock {
+        val exception = captureExceptionsRun {
             val job = Job()
             val child = launch(job, start = ATOMIC) {
                 expect(2)
@@ -109,7 +105,7 @@ class JobExceptionHandlingTest : TestBase() {
          * Inner child: throws AE
          * Result: AE in exception handler
          */
-        val exception = runBlock {
+        val exception = captureExceptionsRun {
             val job = Job()
             launch(job) {
                 expect(2) // <- child is launched successfully
@@ -145,7 +141,7 @@ class JobExceptionHandlingTest : TestBase() {
         * Inner child: throws AE
         * Result: AE
         */
-        val exception = runBlock {
+        val exception = captureExceptionsRun {
             val job = Job()
             launch(job, start = ATOMIC) {
                 expect(2)
@@ -173,7 +169,7 @@ class JobExceptionHandlingTest : TestBase() {
          * Inner child: throws AE
          * Result: IOE with suppressed AE
          */
-        val exception = runBlock {
+        val exception = captureExceptionsRun {
             val job = Job()
             launch(job) {
                 expect(2) // <- child is launched successfully
@@ -196,11 +192,9 @@ class JobExceptionHandlingTest : TestBase() {
             finish(5)
         }
 
-        assertTrue(exception is IOException)
+        assertTrue(exception is ArithmeticException)
         assertNull(exception.cause)
-        val suppressed = exception.suppressed
-        assertEquals(1, suppressed.size)
-        checkException<ArithmeticException>(suppressed[0])
+        assertTrue(exception.suppressed.isEmpty())
     }
 
     @Test
@@ -211,7 +205,7 @@ class JobExceptionHandlingTest : TestBase() {
           * Child: launch 3 children, each of them throws an exception (AE, IOE, IAE) and calls delay()
           * Result: AE with suppressed IOE and IAE
           */
-        val exception = runBlock {
+        val exception = captureExceptionsRun {
             val job = Job()
             launch(job, start = ATOMIC) {
                 expect(2)
@@ -253,7 +247,7 @@ class JobExceptionHandlingTest : TestBase() {
          * Child: launch 2 children (each of them throws an exception (IOE, IAE)), throws AE
          * Result: AE with suppressed IOE and IAE
          */
-        val exception = runBlock {
+        val exception = captureExceptionsRun {
             val job = Job()
             launch(job, start = ATOMIC) {
                 expect(2)
@@ -282,6 +276,34 @@ class JobExceptionHandlingTest : TestBase() {
         assertTrue(suppressed[1] is IllegalArgumentException)
     }
 
+    @Test
+    fun testExceptionIsHandledOnce() = runTest(unhandled = listOf { e -> e is TestException }) {
+        val job = Job()
+        val j1 = launch(job) {
+            expect(1)
+            delay(Long.MAX_VALUE)
+        }
+
+        val j2 = launch(job) {
+            expect(2)
+            throw TestException()
+        }
+
+        joinAll(j1 ,j2)
+        finish(3)
+    }
+
+    @Test
+    fun testCancelledParent() = runTest {
+        expect(1)
+        val parent = Job()
+        parent.completeExceptionally(TestException())
+        launch(parent) {
+            expectUnreached()
+        }.join()
+        finish(2)
+    }
+
     @Test
     fun testBadException() = runTest(unhandled = listOf({e -> e is BadException})) {
         val job = launch(Job()) {

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

@@ -7,7 +7,6 @@ package kotlinx.coroutines.exceptions
 import kotlinx.coroutines.*
 import org.junit.*
 import org.junit.Test
-import java.io.*
 import java.util.concurrent.*
 import kotlin.test.*
 
@@ -28,7 +27,7 @@ class JobExceptionsStressTest : TestBase() {
          * Result: one of the exceptions with the rest two as suppressed
          */
         repeat(1000 * stressTestMultiplier) {
-            val exception = runBlock(executor) {
+            val exception = captureExceptionsRun(executor) {
                 val barrier = CyclicBarrier(4)
                 val job = launch(NonCancellable) {
                     launch(start = CoroutineStart.ATOMIC) {

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

@@ -13,7 +13,7 @@ class JobNestedExceptionsTest : TestBase() {
 
     @Test
     fun testExceptionUnwrapping() {
-        val exception = runBlock {
+        val exception = captureExceptionsRun {
             val job = Job()
             launch(job) {
                 expect(2)
@@ -37,7 +37,7 @@ class JobNestedExceptionsTest : TestBase() {
 
     @Test
     fun testExceptionUnwrappingWithSuspensions() {
-        val exception = runBlock {
+        val exception = captureExceptionsRun {
             val job = Job()
             launch(job) {
                 expect(2)
@@ -66,7 +66,7 @@ class JobNestedExceptionsTest : TestBase() {
 
     @Test
     fun testNestedAtomicThrow() {
-        val exception = runBlock {
+        val exception = captureExceptionsRun {
             expect(1)
             val job = launch(NonCancellable + CoroutineName("outer"), start = CoroutineStart.ATOMIC) {
                 expect(2)
@@ -86,7 +86,7 @@ class JobNestedExceptionsTest : TestBase() {
 
     @Test
     fun testChildThrowsDuringCompletion() {
-        val exceptions = runBlockForMultipleExceptions {
+        val exceptions = captureMultipleExceptionsRun {
             expect(1)
             val job = launch(NonCancellable + CoroutineName("outer"), start = CoroutineStart.ATOMIC) {
                 expect(2)

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

@@ -160,7 +160,9 @@ class ProduceExceptionsTest : TestBase() {
         yield()
         try {
             channel.receive()
-        } catch (e: TestException2) {
+        } catch (e: CancellationException) {
+            // RECOVER_STACK_TRACES
+            assertTrue(e.cause?.cause is TestException2)
             finish(4)
         }
     }

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

@@ -11,7 +11,6 @@ import org.junit.runners.*
 import kotlin.coroutines.*
 import kotlin.test.*
 
-@Suppress("DEPRECATION") // cancel(cause)
 @RunWith(Parameterized::class)
 class WithContextExceptionHandlingTest(private val mode: Mode) : TestBase() {
     enum class Mode { WITH_CONTEXT, ASYNC_AWAIT }