Change order of handlers in JobSupport and CancellableContinuation

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

@@ -188,7 +188,7 @@ internal abstract class AbstractContinuation<in T>(
         return updateStateToFinal(state, CancelledContinuation(this, cause), mode = MODE_ATOMIC_DEFAULT)
     }
 
-    private fun onCompletionInternal(mode: Int) {
+    private fun dispatchResume(mode: Int) {
         if (tryResume()) return // completed before getResult invocation -- bail out
         // otherwise, getResult has already commenced, i.e. completed later or in other thread
         dispatch(mode)
@@ -326,16 +326,18 @@ internal abstract class AbstractContinuation<in T>(
 
     protected fun completeStateUpdate(expect: NotCompleted, update: Any?, mode: Int) {
         val exceptionally = update as? CompletedExceptionally
-        onCompletionInternal(mode)
 
-        // Invoke cancellation handlers only if necessary
         if (update is CancelledContinuation && expect is CancelHandler) {
             try {
                 expect.invoke(exceptionally?.cause)
             } catch (ex: Throwable) {
                 handleException(CompletionHandlerException("Exception in completion handler $expect for $this", ex))
             }
         }
+
+        // Notify all handlers before dispatching, otherwise behaviour will be timing-dependent
+        // and confusing with Unconfined
+        dispatchResume(mode)
     }
 
     private fun handleException(exception: Throwable) {

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

@@ -207,8 +207,17 @@ internal open class JobSupport constructor(active: Boolean) : Job, SelectClause0
     private fun completeUpdateState(expect: Incomplete, update: Any?, mode: Int) {
         val exceptionally = update as? CompletedExceptionally
         // Do overridable processing before completion handlers
-        if (!expect.isCancelling) onCancellationInternal(exceptionally) // only notify when was not cancelling before
-        onCompletionInternal(update, mode)
+
+        /*
+         * 1) Invoke onCancellationInternal: exception handling, parent/resource cancellation etc.
+         * 2) Invoke completion handlers: .join(), callbacks etc. It's important to invoke them only AFTER exception handling, see #208
+         * 3) Invoke onCompletionInternal: onNext(), timeout deregistration etc. I should be last so all callbacks observe consistent state
+         *    of the job which doesn't depend on callback scheduling
+         *
+         * Only notify on cancellation once (expect.isCancelling)
+         */
+        if (!expect.isCancelling) onCancellationInternal(exceptionally)
+
         // Invoke completion handlers
         val cause = exceptionally?.cause
         if (expect is JobNode<*>) { // SINGLE/SINGLE+ state -- one completion handler (common case)
@@ -220,6 +229,8 @@ internal open class JobSupport constructor(active: Boolean) : Job, SelectClause0
         } else {
             expect.list?.notifyCompletion(cause)
         }
+
+        onCompletionInternal(update, mode)
     }
 
     private inline fun <reified T: JobNode<*>> notifyHandlers(list: NodeList, cause: Throwable?) {

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

@@ -24,20 +24,22 @@ class AbstractCoroutineTest : TestBase() {
 
             override fun onCompleted(value: String) {
                 assertEquals("OK", value)
-                expect(6)
+                expect(8)
             }
 
             override fun onCompletedExceptionally(exception: Throwable) {
                 expectUnreached()
             }
         }
+
         coroutine.invokeOnCompletion(onCancelling = true) {
             assertTrue(it == null)
-            expect(7)
+            expect(6)
         }
+
         coroutine.invokeOnCompletion {
             assertTrue(it == null)
-            expect(8)
+            expect(7)
         }
         expect(2)
         coroutine.start()
@@ -66,7 +68,7 @@ class AbstractCoroutineTest : TestBase() {
 
             override fun onCompletedExceptionally(exception: Throwable) {
                 assertTrue(exception is TestException1)
-                expect(8)
+                expect(9)
             }
         }
         coroutine.invokeOnCompletion(onCancelling = true) {
@@ -75,7 +77,7 @@ class AbstractCoroutineTest : TestBase() {
         }
         coroutine.invokeOnCompletion {
             assertTrue(it is TestException1)
-            expect(9)
+            expect(8)
         }
         expect(2)
         coroutine.start()

common/kotlinx-coroutines-core-common/test/channels/ProduceTest.kt

@@ -9,6 +9,7 @@ import kotlin.coroutines.experimental.*
 import kotlin.test.*
 
 class ProduceTest : TestBase() {
+
     @Test
     fun testBasic() = runTest {
         val c = produce(coroutineContext) {
@@ -81,5 +82,34 @@ class ProduceTest : TestBase() {
         }
     }
 
+    @Test
+    fun testCancelOnCompletionUnconfined() = runTest {
+        cancelOnCompletion(Unconfined)
+    }
+
+    @Test
+    fun testCancelOnCompletion() = runTest {
+        cancelOnCompletion(coroutineContext)
+    }
+
+    private suspend fun cancelOnCompletion(coroutineContext: CoroutineContext) {
+        val source = Channel<Int>()
+        expect(1)
+        val produced = produce<Int>(coroutineContext, onCompletion = source.consumes()) {
+            expect(2)
+            source.receive()
+        }
+
+        yield()
+        expect(3)
+        produced.cancel()
+        try {
+            source.receive()
+            // TODO shouldn't it be ClosedReceiveChannelException ?
+        } catch (e: JobCancellationException) {
+            finish(4)
+        }
+    }
+
     private class TestException : Exception()
 }