Ignore resume on cancelled continuation

There is a race between cancellation of outer job and invocation of
resume on a cancellable continuation and this race cannot be
worked around by the code that uses suspendCancellableCoroutine
function. Thus resume attempt on a cancelled continuation shall
be simply ignored.

Fixes #450

Author: elizarov

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

@@ -272,15 +272,17 @@ internal abstract class AbstractContinuation<in T>(
                         }
                     }
                 }
-
                 is NotCompleted -> {
                     if (updateStateToFinal(state, proposedUpdate, resumeMode)) return
                 }
                 is CancelledContinuation -> {
-                    if (proposedUpdate is NotCompleted || proposedUpdate is CompletedExceptionally) {
-                        error("Unexpected update, state: $state, update: $proposedUpdate")
-                    }
-                    // Coroutine is dispatched normally (e.g.via `delay()`) after cancellation
+                    /*
+                     * If continuation was cancelled, then all further updates (resumes or exceptions) must be
+                     * ignored, because cancellation is asynchronous and may race with resume/resumeWithException.
+                     * This race is normal.
+                     *
+                     * :todo: we should somehow remember the attempt to invoke resume and fail on the second attempt.
+                     */
                     return
                 }
                 else -> error("Already resumed, but proposed with update $proposedUpdate")

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

@@ -0,0 +1,120 @@
+/*
+ * Copyright 2016-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package kotlinx.coroutines.experimental
+
+import kotlin.coroutines.experimental.*
+import kotlin.test.*
+
+class CancellableContinuationTest : TestBase() {
+    @Test
+    fun testResumeWithExceptionAndResumeWithException() = runTest {
+        var continuation: Continuation<Unit>? = null
+        val job = launch(coroutineContext) {
+            try {
+                expect(2)
+                suspendCancellableCoroutine<Unit> { c ->
+                    continuation = c
+                }
+            } catch (e: TestException) {
+                expect(3)
+            }
+        }
+        expect(1)
+        yield()
+        continuation!!.resumeWithException(TestException())
+        yield()
+        assertFailsWith<IllegalStateException> { continuation!!.resumeWithException(TestException()) }
+        job.join()
+        finish(4)
+    }
+
+    @Test
+    fun testResumeAndResumeWithException() = runTest {
+        var continuation: Continuation<Unit>? = null
+        val job = launch(coroutineContext) {
+            expect(2)
+            suspendCancellableCoroutine<Unit> { c ->
+                continuation = c
+            }
+            expect(3)
+        }
+        expect(1)
+        yield()
+        continuation!!.resume(Unit)
+        job.join()
+        assertFailsWith<IllegalStateException> { continuation!!.resumeWithException(TestException()) }
+        finish(4)
+    }
+
+    @Test
+    fun testResumeAndResume() = runTest {
+        var continuation: Continuation<Unit>? = null
+        val job = launch(coroutineContext) {
+            expect(2)
+            suspendCancellableCoroutine<Unit> { c ->
+                continuation = c
+            }
+            expect(3)
+        }
+        expect(1)
+        yield()
+        continuation!!.resume(Unit)
+        job.join()
+        assertFailsWith<IllegalStateException> { continuation!!.resume(Unit) }
+        finish(4)
+    }
+
+    /**
+     * Cancelling outer job may, in practise, race with attempt to resume continuation and resumes
+     * should be ignored. Here suspended coroutine is cancelled but then resumed with exception.
+     */
+    @Test
+    fun testCancelAndResumeWithException() = runTest {
+        var continuation: Continuation<Unit>? = null
+        val job = launch(coroutineContext) {
+            try {
+                expect(2)
+                suspendCancellableCoroutine<Unit> { c ->
+                    continuation = c
+                }
+            } catch (e: JobCancellationException) {
+                expect(3)
+            }
+        }
+        expect(1)
+        yield()
+        job.cancel() // Cancel job
+        yield()
+        continuation!!.resumeWithException(TestException()) // Should not fail
+        finish(4)
+    }
+
+    /**
+     * Cancelling outer job may, in practise, race with attempt to resume continuation and resumes
+     * should be ignored. Here suspended coroutine is cancelled but then resumed with exception.
+     */
+    @Test
+    fun testCancelAndResume() = runTest {
+        var continuation: Continuation<Unit>? = null
+        val job = launch(coroutineContext) {
+            try {
+                expect(2)
+                suspendCancellableCoroutine<Unit> { c ->
+                    continuation = c
+                }
+            } catch (e: JobCancellationException) {
+                expect(3)
+            }
+        }
+        expect(1)
+        yield()
+        job.cancel() // Cancel job
+        yield()
+        continuation!!.resume(Unit) // Should not fail
+        finish(4)
+    }
+
+    private class TestException : Exception()
+}

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

@@ -215,53 +215,6 @@ class CancellableContinuationExceptionHandlingTest : TestBase() {
         }
     }
 
-    @Test
-    fun testMultipleCancellations() = runTest {
-        var continuation: Continuation<Unit>? = null
-
-        val job = launch(coroutineContext) {
-            try {
-                expect(2)
-                suspendCancellableCoroutine<Unit> { c ->
-                    continuation = c
-                }
-            } catch (e: IOException) {
-                expect(3)
-            }
-        }
-
-        expect(1)
-        yield()
-        continuation!!.resumeWithException(IOException())
-        yield()
-        assertFailsWith<IllegalStateException> { continuation!!.resumeWithException(ClosedChannelException()) }
-        try {
-            job.join()
-        } finally {
-            finish(4)
-        }
-    }
-
-    @Test
-    fun testResumeAndCancel() = runTest {
-        var continuation: Continuation<Unit>? = null
-
-        val job = launch(coroutineContext) {
-            expect(2)
-            suspendCancellableCoroutine<Unit> { c ->
-                continuation = c
-            }
-            expect(3)
-        }
-
-        expect(1)
-        yield()
-        continuation!!.resume(Unit)
-        job.join()
-        assertFailsWith<IllegalStateException> { continuation!!.resumeWithException(ClosedChannelException()) }
-        finish(4)
-    }
-
     private fun wrapperDispatcher(context: CoroutineContext): CoroutineContext {
         val dispatcher = context[ContinuationInterceptor] as CoroutineDispatcher
         return object : CoroutineDispatcher() {