Exception unwrapping logic rework, do not unwrap cancellation exception if parent can handle exceptions

It allows to properly handle only relevant exceptions by child without reporting the same exception multiple times, fixes suppressions hierarchy and allows to implement supervisorScope properly

Author: qwwdfsad

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

@@ -234,18 +234,18 @@ internal open class JobSupport constructor(active: Boolean) : Job, ChildJob, Sel
          * 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 it is cancelled, adding its exception to parent's list of exceptions
-         * 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.
+         * If parent is cancelling, it cancels its children with JCE(rootCause).
+         * When child is building final exception, it can skip JCE(anything) if it knows
+         * that parent handles exceptions, because parent should already have this exception.
+         * If parent does not, then we should unwrap exception, otherwise in the following code
+         * ```
+         * val parent = Job()
+         * launch(parent) {
+         *   try { delay() } finally { throw E2() }
+         * }
+         * parent.cancel(E1)
+         * ```
+         * E1 will be lost.
          */
         var rootCause = exceptions[0]
         if (rootCause is CancellationException) {
@@ -275,13 +275,17 @@ internal open class JobSupport constructor(active: Boolean) : Job, ChildJob, Sel
         return suppressed
     }
 
-    private tailrec fun unwrap(exception: Throwable): Throwable? =
-        if (exception is CancellationException) {
+    private tailrec fun unwrap(exception: Throwable): Throwable? {
+        if (exception is CancellationException && parentHandlesExceptions) {
+            return null
+        }
+        return if (exception is CancellationException) {
             val cause = exception.cause
             if (cause !== null) unwrap(cause) else null
         } else {
             exception
         }
+    }
 
     // fast-path method to finalize normally completed coroutines without children
     private fun tryFinalizeSimpleState(state: Incomplete, update: Any?, mode: Int): Boolean {
@@ -923,6 +927,10 @@ internal open class JobSupport constructor(active: Boolean) : Job, ChildJob, Sel
      */
     protected open val handlesException: Boolean get() = true
 
+    // returns true when we know that parent handles exceptions
+    private val parentHandlesExceptions: Boolean get() =
+        (parentHandle as? ChildHandleNode)?.job?.handlesException ?: false
+
     /**
      * 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.

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

@@ -107,22 +107,73 @@ class SupervisorTest : TestBase() {
     }
 
     @Test
-    fun testSupervisorFlaw() = runTest {
+    fun testSupervisorThrows() = runTest {
         try {
             supervisorScope {
                 expect(1)
                 launch {
-                    // Test exception from supervisor is handled by launch though it clearly will be rethtown by supervisorScope{}
                     expect(2)
                     delay(Long.MAX_VALUE)
                 }
 
+                launch {
+                    expect(3)
+                    delay(Long.MAX_VALUE)
+                }
+
                 yield()
-                expect(3)
+                expect(4)
                 throw TestException1()
             }
         } catch (e: TestException1) {
-            finish(4)
+            finish(5)
+        }
+    }
+
+    @Test
+    @Ignore // JS BE bug
+    fun testSupervisorThrowsWithFailingChild() = runTest(unhandled = listOf({e -> e is TestException2})) {
+        try {
+            supervisorScope {
+                expect(1)
+                launch {
+                    expect(2)
+                    delay(Long.MAX_VALUE)
+                }
+
+                launch {
+                    expect(3)
+                    try {
+                        delay(Long.MAX_VALUE)
+                    } finally {
+                        throw TestException2()
+                    }
+                }
+
+                yield()
+                expect(4)
+                throw TestException1()
+            }
+        } catch (e: TestException1) {
+            finish(5)
+        }
+    }
+
+    @Test
+    fun testAsyncCancellation() = runTest {
+        val parent = SupervisorJob()
+        val deferred = async(parent) {
+            expect(2)
+            delay(Long.MAX_VALUE)
+        }
+        expect(1)
+        yield()
+        parent.cancel(TestException1())
+        try {
+            deferred.await()
+            expectUnreached()
+        } catch (e: TestException1) {
+            finish(3)
         }
     }
 
@@ -146,4 +197,4 @@ class SupervisorTest : TestBase() {
 
     private class TestException1 : Exception()
     private class TestException2 : Exception()
-}
+}

core/kotlinx-coroutines-core/test/WithTimeoutOrNullJvmTest.kt

@@ -4,7 +4,6 @@
 
 package kotlinx.coroutines.experimental
 
-import java.io.*
 import kotlin.test.*
 
 class WithTimeoutOrNullJvmTest : TestBase() {

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

@@ -34,6 +34,45 @@ class JobExceptionHandlingTest : TestBase() {
         checkException<IllegalStateException>(exception)
     }
 
+    @Test
+    fun testAsyncCancellationWithCause() = runTest {
+        val deferred = async {
+            expect(2)
+            delay(Long.MAX_VALUE)
+        }
+
+        expect(1)
+        yield()
+        deferred.cancel(IOException())
+        try {
+            deferred.await()
+            expectUnreached()
+        } catch (e: IOException) {
+            assertTrue(e.suppressed().isEmpty())
+            finish(3)
+        }
+    }
+
+    @Test
+    fun testAsyncCancellationWithCauseAndParent() = runTest {
+        val parent = Job()
+        val deferred = async(parent) {
+            expect(2)
+            delay(Long.MAX_VALUE)
+        }
+
+        expect(1)
+        yield()
+        parent.cancel(IOException())
+        try {
+            deferred.await()
+            expectUnreached()
+        } catch (e: IOException) {
+            assertTrue(e.suppressed().isEmpty())
+            finish(3)
+        }
+    }
+
     @Test
     fun testExceptionDuringCancellation() {
         /*
@@ -191,7 +230,6 @@ class JobExceptionHandlingTest : TestBase() {
         checkException<ArithmeticException>(suppressed[0])
     }
 
-
     @Test
     fun testMultipleChildrenThrowAtomically() {
         /*

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

@@ -112,9 +112,9 @@ class JobNestedExceptionsTest : TestBase() {
         val exception = exceptions[0]
         assertTrue(exception is ArithmeticException, "Exception is $exception")
         val suppressed = exception.suppressed()
-        // the order of suppressed exceptions here is a bit hacky, may change in the future
-        checkException<NullPointerException>(suppressed[0])
-        checkException<IOException>(suppressed[1])
+        val ioe = suppressed[0]
+        assertTrue(ioe is IOException)
+        checkException<NullPointerException>(ioe.suppressed()[0])
         checkCycles(exception)
     }
 }

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

@@ -70,6 +70,102 @@ class ProduceExceptionsTest : TestBase() {
         }
     }
 
+    @Test
+    fun testCancelProduceChannel() = runTest {
+        var channel: ReceiveChannel<Int>? = null
+        channel = produce {
+            expect(2)
+            channel!!.cancel()
+            try {
+                send(1)
+            } catch (e: ClosedSendChannelException) {
+                expect(3)
+                throw e
+            }
+        }
+
+        expect(1)
+        yield()
+        try {
+            channel.receive()
+        } catch (e: ClosedReceiveChannelException) {
+            assertTrue(e.suppressed().isEmpty())
+            finish(4)
+        }
+    }
+
+    @Test
+    fun testCancelProduceChannelWithException() = runTest {
+        var channel: ReceiveChannel<Int>? = null
+        channel = produce {
+            expect(2)
+            channel!!.cancel(TestException())
+            try {
+                send(1)
+                // Not a ClosedForSendException
+            } catch (e: TestException) {
+                expect(3)
+                throw e
+            }
+        }
+
+        expect(1)
+        yield()
+        try {
+            channel.receive()
+        } catch (e: TestException) {
+            assertTrue(e.suppressed().isEmpty())
+            finish(4)
+        }
+    }
+
+    @Test
+    fun testCancelChannelWithJob() = runTest {
+        val job = Job()
+        val channel = produce(job) {
+            expect(2)
+            job.cancel()
+            try {
+                send(1)
+            } catch (e: CancellationException) {
+                expect(3)
+                throw e
+            }
+        }
+
+        expect(1)
+        yield()
+        try {
+            channel.receive()
+        } catch (e: CancellationException) {
+            assertTrue(e.suppressed().isEmpty())
+            finish(4)
+        }
+    }
+
+    @Test
+    fun testCancelChannelWithJobWithException() = runTest {
+        val job = Job()
+        val channel = produce(job) {
+            expect(2)
+            job.cancel(TestException2())
+            try {
+                send(1)
+            } catch (e: CancellationException) { // Not a TestException2
+                expect(3)
+                throw e
+            }
+        }
+
+        expect(1)
+        yield()
+        try {
+            channel.receive()
+        } catch (e: TestException2) {
+            finish(4)
+        }
+    }
+
     class TestException : Exception()
     class TestException2 : Exception()
 }