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.
  * Now exception is handled only once

First part of fix for #689

Author: qwwdfsad

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

@@ -382,7 +382,7 @@ public class kotlinx/coroutines/JobSupport : kotlinx/coroutines/ChildJob, kotlin
 	public fun get (Lkotlin/coroutines/CoroutineContext$Key;)Lkotlin/coroutines/CoroutineContext$Element;
 	public final fun getCancellationException ()Ljava/util/concurrent/CancellationException;
 	protected fun getCancelsParent ()Z
-	public fun getChildJobCancellationCause ()Ljava/lang/Throwable;
+	public fun getChildJobCancellationCause ()Ljava/util/concurrent/CancellationException;
 	public final fun getChildren ()Lkotlin/sequences/Sequence;
 	protected final fun getCompletionCause ()Ljava/lang/Throwable;
 	protected final fun getCompletionCauseHandled ()Z
@@ -447,7 +447,7 @@ public abstract interface annotation class kotlinx/coroutines/ObsoleteCoroutines
 }
 
 public abstract interface class kotlinx/coroutines/ParentJob : kotlinx/coroutines/Job {
-	public abstract fun getChildJobCancellationCause ()Ljava/lang/Throwable;
+	public abstract fun getChildJobCancellationCause ()Ljava/util/concurrent/CancellationException;
 }
 
 public final class kotlinx/coroutines/ParentJob$DefaultImpls {

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?.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 }