Exception propagation logic from cancelled coroutines is adjusted

* When cancelled coroutine crashes due to some other exception, this
other exception becomes the cancellation reason of the coroutine, while
the original cancellation reason is suppressed.
* UnexpectedCoroutineException is no longer used to report those cases
as is removed.
* This fixes a race between crash of CPU-consuming coroutine and
cancellation which resulted in an unhandled exception and lead to
crashes on Android.

Fixes #152

Author: elizarov

core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Job.kt

@@ -376,13 +376,6 @@ public class JobCancellationException(
         (message!!.hashCode() * 31 + job.hashCode()) * 31 + (cause?.hashCode() ?: 0)
 }
 
-/**
- * Represents an exception in the coroutine that was not caught by it and was not expected to be thrown.
- * This happens when coroutine is cancelled, but it completes with the different exception than its cancellation
- * cause was.
- */
-public class UnexpectedCoroutineException(message: String, cause: Throwable) : IllegalStateException(message, cause)
-
 /**
  * Unregisters a specified [registration] when this job is complete.
  *
@@ -621,38 +614,31 @@ public open class JobSupport(active: Boolean) : Job, SelectClause0, SelectClause
         val update = coerceProposedUpdate(expect, proposedUpdate)
         if (!tryUpdateState(expect, update)) return false
         completeUpdateState(expect, update, mode)
-        // if an exceptional completion was suppressed (because cancellation was in progress), then report it separately
-        if (proposedUpdate is CompletedExceptionally && proposedUpdate.cause != null && !incorporatedCause(update, proposedUpdate.cause)) {
-            handleException(UnexpectedCoroutineException("Unexpected exception while cancellation is in progress; job=$this", proposedUpdate.cause))
-        }
         return true
     }
 
-    /**
-     * Checks if the cause that was proposed for state update is consistent with the resulting updated state
-     * and not exception information was lost. The key observation here is that [getCancellationException] wraps
-     * exceptions that are not [CancellationException] into an instance of [JobCancellationException] and we allow
-     * that [JobCancellationException] to be unwrapped again when it reaches the coroutine that was cancelled.
-     *
-     * NOTE: equality comparison of exceptions is performed here by design, see equals of JobCancellationException
-     */
-    private fun incorporatedCause(update: Any?, proposedCause: Throwable) =
-        update is CompletedExceptionally && update.exception.let { ex ->
-            ex == proposedCause || proposedCause is JobCancellationException && ex == proposedCause.cause
-        }
-
-    // when Job is in Cancelling state, it can only be promoted to Cancelled state with the same cause
-    // however, null cause can be replaced with more specific JobCancellationException (that contains better stack trace)
+    // when Job is in Cancelling state, it can only be promoted to Cancelled state,
+    // so if the proposed Update is not an appropriate Cancelled (preserving the cancellation cause),
+    // then the corresponding Cancelled state is constructed.
     private fun coerceProposedUpdate(expect: Incomplete, proposedUpdate: Any?): Any? =
-        if (expect is Finishing && expect.cancelled != null && !correspondinglyCancelled(expect.cancelled, proposedUpdate))
-            expect.cancelled else proposedUpdate
+        if (expect is Finishing && expect.cancelled != null && !isCorrespondinglyCancelled(expect.cancelled, proposedUpdate))
+            createCancelled(expect.cancelled, proposedUpdate) else proposedUpdate
 
-    private fun correspondinglyCancelled(cancelled: Cancelled, proposedUpdate: Any?): Boolean {
+    private fun isCorrespondinglyCancelled(cancelled: Cancelled, proposedUpdate: Any?): Boolean {
         if (proposedUpdate !is Cancelled) return false
-        return proposedUpdate.cause === cancelled.cause ||
+        // NOTE: equality comparison of causes is performed here by design, see equals of JobCancellationException
+        return proposedUpdate.cause == cancelled.cause ||
             proposedUpdate.cause is JobCancellationException && cancelled.cause == null
     }
 
+    private fun createCancelled(cancelled: Cancelled, proposedUpdate: Any?): Cancelled {
+        if (proposedUpdate !is CompletedExceptionally) return cancelled // not exception -- just use original cancelled
+        val exception = proposedUpdate.exception
+        if (cancelled.exception == exception) return cancelled // that is the cancelled we need already!
+        cancelled.cause?.let { exception.addSuppressed(it) }
+        return Cancelled(this, exception)
+    }
+
     /**
      * Tries to initiate update of the current [state] of this job.
      * @suppress **This is unstable API and it is subject to change.**

core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/CoroutinesTest.kt

@@ -162,13 +162,14 @@ class CoroutinesTest : TestBase() {
 
     @Test
     fun testCancelParentOnChildException() = runTest(
-        expected = { it is IOException },
+        expected = { it is JobCancellationException && it.cause is IOException },
         unhandled = listOf({ it -> it is IOException })
     ) {
         expect(1)
         launch(coroutineContext) {
             finish(3)
-            throw IOException() // does not propagate exception to launch, but cancels parent (!)
+            throwIOException() // does not propagate exception to launch, but cancels parent (!)
+            expectUnreached()
         }
         expect(2)
         yield()
@@ -177,7 +178,7 @@ class CoroutinesTest : TestBase() {
 
     @Test
     fun testCancelParentOnNestedException() = runTest(
-        expected = { it is IOException },
+        expected = { it is JobCancellationException && it.cause is IOException },
         unhandled = listOf(
             { it -> it is IOException },
             { it -> it is IOException }
@@ -188,7 +189,8 @@ class CoroutinesTest : TestBase() {
             expect(3)
             launch(coroutineContext) {
                 finish(6)
-                throw IOException() // unhandled exception kills all parents
+                throwIOException() // unhandled exception kills all parents
+                expectUnreached()
             }
             expect(4)
             yield()
@@ -246,13 +248,14 @@ class CoroutinesTest : TestBase() {
 
     @Test
     fun testCancelAndJoinChildCrash() = runTest(
-        expected = { it is IOException && it.message == "OK" },
+        expected = { it is JobCancellationException && it.cause is IOException },
         unhandled = listOf({it -> it is IOException })
     ) {
         expect(1)
         val job = launch(coroutineContext, CoroutineStart.UNDISPATCHED) {
             expect(2)
-            throw IOException("OK")
+            throwIOException()
+            expectUnreached()
         }
         // now we have a failed job with IOException
         finish(3)
@@ -310,5 +313,38 @@ class CoroutinesTest : TestBase() {
         expectUnreached()
     }
 
+    @Test
+    fun testNotCancellableCodeWithExceptionCancelled() = runTest {
+        expect(1)
+        // CoroutineStart.ATOMIC makes sure it will not get cancelled for it starts executing
+        val job = launch(start = CoroutineStart.ATOMIC) {
+            Thread.sleep(100) // cannot be cancelled
+            throwIOException() // will throw
+            expectUnreached()
+        }
+        expect(2)
+        job.cancel()
+        finish(3)
+    }
+
+    @Test
+    fun testNotCancellableChildWithExceptionCancelled() = runTest(
+        expected = { it is IOException }
+    ) {
+        expect(1)
+        // CoroutineStart.ATOMIC makes sure it will not get cancelled for it starts executing
+        val d = async(coroutineContext, start = CoroutineStart.ATOMIC) {
+            finish(4)
+            throwIOException() // will throw
+            expectUnreached()
+        }
+        expect(2)
+        // now cancel with some other exception
+        d.cancel(IllegalArgumentException())
+        // now await to see how it got crashed -- IAE should have been suppressed by IOException
+        expect(3)
+        d.await()
+    }
+
     private fun throwIOException() { throw IOException() }
 }

core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/WithTimeoutOrNullTest.kt

@@ -92,7 +92,7 @@ class WithTimeoutOrNullTest : TestBase() {
     }
 
     @Test
-    fun testSuppressException() = runTest {
+    fun testSuppressExceptionWithResult() = runTest {
         expect(1)
         val result = withTimeoutOrNull(100) {
             expect(2)
@@ -108,23 +108,22 @@ class WithTimeoutOrNullTest : TestBase() {
     }
 
     @Test
-    fun testReplaceException() = runTest(
-        unhandled = listOf({ it -> it is UnexpectedCoroutineException && it.cause is IOException })
+    fun testSuppressExceptionWithAnotherException() = runTest(
+        expected = { it is IOException }
     ) {
         expect(1)
         val result = withTimeoutOrNull(100) {
             expect(2)
             try {
                 delay(1000)
             } catch (e: CancellationException) {
-                expect(3)
+                finish(3)
                 throw IOException(e)
             }
             expectUnreached()
             "OK"
         }
-        assertThat(result, IsNull())
-        finish(4)
+        expectUnreached()
     }
 
     /**

core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/WithTimeoutTest.kt

@@ -95,7 +95,7 @@ class WithTimeoutTest : TestBase() {
     }
 
     @Test
-    fun testSuppressException() = runTest(
+    fun testSuppressExceptionWithResult() = runTest(
         expected = { it is CancellationException }
     ) {
         expect(1)
@@ -112,9 +112,8 @@ class WithTimeoutTest : TestBase() {
     }
 
     @Test
-    fun testReplaceException() = runTest(
-        expected = { it is CancellationException },
-        unhandled = listOf({ it -> it is UnexpectedCoroutineException && it.cause is IOException })
+    fun testSuppressExceptionWithAnotherException() = runTest(
+        expected = { it is IOException }
     ) {
         expect(1)
         withTimeout(100) {

core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/channels/ProduceTest.kt

@@ -16,31 +16,51 @@
 
 package kotlinx.coroutines.experimental.channels
 
+import kotlinx.coroutines.experimental.Job
+import kotlinx.coroutines.experimental.JobCancellationException
 import kotlinx.coroutines.experimental.TestBase
-import kotlinx.coroutines.experimental.runBlocking
 import org.junit.Test
 
 class ProduceTest : TestBase() {
     @Test
-    fun testBasic() = runBlocking {
+    fun testBasic() = runTest {
         val c = produce(coroutineContext) {
+            expect(2)
             send(1)
+            expect(3)
             send(2)
+            expect(6)
         }
+        expect(1)
         check(c.receive() == 1)
+        expect(4)
         check(c.receive() == 2)
+        expect(5)
         check(c.receiveOrNull() == null)
+        finish(7)
     }
 
     @Test
-    fun testCancel() = runBlocking {
+    fun testCancel() = runTest {
         val c = produce(coroutineContext) {
+            expect(2)
             send(1)
-            send(2)
+            expect(3)
+            try {
+                send(2) // will get cancelled
+            } catch (e: Throwable) {
+                expect(6)
+                check(e is JobCancellationException && e.job == coroutineContext[Job])
+                throw e
+            }
             expectUnreached()
         }
+        expect(1)
         check(c.receive() == 1)
+        expect(4)
         c.cancel()
+        expect(5)
         check(c.receiveOrNull() == null)
+        finish(7)
     }
 }