Failed parent Job cancels all children jobs, then waits them

Also improved debug output a little bit

Author: elizarov

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

@@ -242,5 +242,4 @@ private class BlockingCoroutine<T>(
         (state as? CompletedExceptionally)?.let { throw it.exception }
         return state as T
     }
-
 }

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

@@ -113,19 +113,23 @@ internal class DispatchTask<in T>(
 ) : Runnable {
     @Suppress("UNCHECKED_CAST")
     override fun run() {
-        val context = continuation.context
-        val job = if (cancellable) context[Job] else null
-        withCoroutineContext(context) {
-            when {
-                job != null && !job.isActive -> continuation.resumeWithException(job.getCancellationException())
-                exception -> continuation.resumeWithException(value as Throwable)
-                else -> continuation.resume(value as T)
+        try {
+            val context = continuation.context
+            val job = if (cancellable) context[Job] else null
+            withCoroutineContext(context) {
+                when {
+                    job != null && !job.isActive -> continuation.resumeWithException(job.getCancellationException())
+                    exception -> continuation.resumeWithException(value as Throwable)
+                    else -> continuation.resume(value as T)
+                }
             }
+        } catch (e: Throwable) {
+            throw RuntimeException("Unexpected exception running $this", e)
         }
     }
 
     override fun toString(): String =
-        "DispatchTask[$value, cancellable=$cancellable, ${continuation.toDebugString()}]"
+        "DispatchTask[${continuation.toDebugString()}, cancellable=$cancellable, value=${value.toSafeString()}]"
 }
 
 internal class DispatchedContinuation<in T>(
@@ -186,16 +190,8 @@ internal class DispatchedContinuation<in T>(
         dispatcher.dispatch(context, DispatchTask(continuation, value,false, true))
     }
 
-    override fun toString(): String = "DispatchedContinuation[$dispatcher, ${continuation.toDebugString()}]"
-}
-
-// **KLUDGE**: there is no reason to include continuation into debug string until the following ticket is resolved:
-// KT-18986 Debug-friendly toString implementation for CoroutineImpl
-// (the current string representation of continuation is useless and uses buggy reflection internals)
-// So, this function is a replacement that extract a usable information from continuation -> its class name, at least
-internal fun Continuation<*>.toDebugString(): String = when (this) {
-    is DispatchedContinuation -> toString()
-    else -> this::class.java.name
+    override fun toString(): String =
+        "DispatchedContinuation[$dispatcher, ${continuation.toDebugString()}]"
 }
 
 internal fun <T> Continuation<T>.resumeCancellable(value: T) = when (this) {

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

@@ -0,0 +1,38 @@
+/*
+ * Copyright 2016-2017 JetBrains s.r.o.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kotlinx.coroutines.experimental
+
+import kotlin.coroutines.experimental.Continuation
+
+// internal debugging tools
+
+internal val Any.hexAddress: String
+    get() = Integer.toHexString(System.identityHashCode(this))
+
+internal fun Any?.toSafeString(): String =
+    try { toString() }
+    catch (e: Throwable) { "toString() failed with $e" }
+
+// **KLUDGE**: there is no reason to include continuation into debug string until the following ticket is resolved:
+// KT-18986 Debug-friendly toString implementation for CoroutineImpl
+// (the current string representation of continuation is useless and uses buggy reflection internals)
+// So, this function is a replacement that extract a usable information from continuation -> its class name, at least
+internal fun Continuation<*>.toDebugString(): String = when (this) {
+    is DispatchedContinuation -> toString()
+    else -> "${this::class.java.name}@$hexAddress"
+}
+

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

@@ -52,7 +52,7 @@ import kotlin.coroutines.experimental.intrinsics.suspendCoroutineOrReturn
  * | _Completing_ (optional transient state) | `true`     | `false`       | `false`       |
  * | _Cancelling_ (optional transient state) | `false`    | `false`       | `true`        |
  * | _Cancelled_ (final state)               | `false`    | `true`        | `true`        |
- * | _Completed normally_ (final state)      | `false`    | `true`        | `false`       |
+ * | _Completed_ (final state)               | `false`    | `true`        | `false`       |
  *
  * Usually, a job is created in _active_ state (it is created and started). However, coroutine builders
  * that provide an optional `start` parameter create a coroutine in _new_ state when this parameter is set to
@@ -68,8 +68,8 @@ import kotlin.coroutines.experimental.intrinsics.suspendCoroutineOrReturn
  *                                                      wait children
  *    +-----+       start      +--------+   complete   +-------------+  finish  +-----------+
  *    | New | ---------------> | Active | -----------> | Completing  | -------> | Completed |
- *    +-----+                  +--------+              +-------------+          | normally  |
- *       |                         |                         |                  +-----------+
+ *    +-----+                  +--------+              +-------------+          +-----------+
+ *       |                         |                         |
  *       | cancel                  | cancel                  | cancel
  *       V                         V                         |
  *  +-----------+   finish   +------------+                  |
@@ -82,7 +82,7 @@ import kotlin.coroutines.experimental.intrinsics.suspendCoroutineOrReturn
  * A job is active while the coroutine is working and job's cancellation aborts the coroutine when
  * the coroutine is suspended on a _cancellable_ suspension point by throwing [CancellationException].
  *
- * A job can have a _parent_ job. A job with a parent is cancelled when its parent is cancelled or completes.
+ * A job can have a _parent_ job. A job with a parent is cancelled when its parent is cancelled or completes exceptionally.
  * Parent job waits for all its [children][attachChild] to complete in _completing_ or _cancelling_ state.
  * _Completing_ state is purely internal to the job. For an outside observer a _completing_ job is still active,
  * while internally it is waiting for its children.
@@ -185,9 +185,10 @@ public interface Job : CoroutineContext.Element {
      * returns a handle that should be used to detach it.
      *
      * A parent-child relation has the following effect:
-     * * Cancellation of parent with [cancel] immediately cancels all its children with the same cause.
+     * * Cancellation of parent with [cancel] or its exceptional completion (failure)
+     *   immediately cancels all its children.
      * * Parent cannot complete until all its children are complete. Parent waits for all its children to
-     *   complete first in _completing_ or _cancelling_ state.
+     *   complete in _completing_ or _cancelling_ state.
      *
      * A child must store the resulting [DisposableHandle] and [dispose][DisposableHandle.dispose] the attachment
      * to its parent on its own completion.
@@ -988,30 +989,38 @@ public open class JobSupport(active: Boolean) : Job, SelectClause0, SelectClause
                 throw IllegalStateException("Job $this is already complete, but is being completed with $proposedUpdate", proposedUpdate.exceptionOrNull)
             if (state is Finishing && state.completing)
                 throw IllegalStateException("Job $this is already completing, but is being completed with $proposedUpdate", proposedUpdate.exceptionOrNull)
-            val waitChild: Child = firstChild(state) ?: // or else complete immediately
+            val child: Child = firstChild(state) ?: // or else complete immediately w/o children
                 if (updateState(state, proposedUpdate, mode)) return true else return@loopOnState
-            // switch to completing state
+            // must promote to list to correct operate on child lists
             if (state is JobNode<*>) {
-                // must promote to list to make completing & retry
                 promoteSingleToNodeList(state)
-            } else {
-                val completing = Finishing(state.list!!, (state as? Finishing)?.cancelled, true)
-                if (_state.compareAndSet(state, completing)) {
-                    waitForChild(waitChild, proposedUpdate)
-                    return false
-                }
+                return@loopOnState // retry
+            }
+            // cancel all children in list on exceptional completion
+            if (proposedUpdate is CompletedExceptionally)
+                child.cancelChildrenInternal(proposedUpdate.exception)
+            // switch to completing state
+            val completing = Finishing(state.list!!, (state as? Finishing)?.cancelled, true)
+            if (_state.compareAndSet(state, completing)) {
+                waitForChild(child, proposedUpdate)
+                return false
             }
         }
     }
 
+    private tailrec fun Child.cancelChildrenInternal(cause: Throwable) {
+        childJob.cancel(JobCancellationException("Child job was cancelled because of parent failure", cause, childJob))
+        nextChild()?.cancelChildrenInternal(cause)
+    }
+
     private val Any?.exceptionOrNull: Throwable?
         get() = (this as? CompletedExceptionally)?.exception
 
     private fun firstChild(state: Incomplete) =
         state as? Child ?: state.list?.nextChild()
 
-    private fun waitForChild(waitChild: Child, proposedUpdate: Any?) {
-        waitChild.child.invokeOnCompletion(handler = ChildCompletion(this, waitChild, proposedUpdate))
+    private fun waitForChild(child: Child, proposedUpdate: Any?) {
+        child.childJob.invokeOnCompletion(handler = ChildCompletion(this, child, proposedUpdate))
     }
 
     internal fun continueCompleting(lastChild: Child, proposedUpdate: Any?) {
@@ -1044,13 +1053,13 @@ public open class JobSupport(active: Boolean) : Job, SelectClause0, SelectClause
     public override fun cancelChildren(cause: Throwable?) {
         val state = this.state
         when (state) {
-            is Child -> state.child.cancel(cause)
+            is Child -> state.childJob.cancel(cause)
             is Incomplete -> state.list?.cancelChildrenList(cause)
         }
     }
 
     private fun NodeList.cancelChildrenList(cause: Throwable?) {
-        forEach<Child> { it.child.cancel(cause) }
+        forEach<Child> { it.childJob.cancel(cause) }
     }
 
     /**
@@ -1079,7 +1088,7 @@ public open class JobSupport(active: Boolean) : Job, SelectClause0, SelectClause
 
     // for nicer debugging
     override final fun toString(): String =
-        "${nameString()}{${stateString()}}@${Integer.toHexString(System.identityHashCode(this))}"
+        "${nameString()}{${stateString()}}@$hexAddress"
 
     /**
      * @suppress **This is unstable API and it is subject to change.**
@@ -1393,22 +1402,22 @@ private class InvokeOnCancellation(
 
 internal class Child(
     parent: JobSupport,
-    val child: Job
+    @JvmField val childJob: Job
 ) : JobCancellationNode<JobSupport>(parent) {
     override fun invoke(reason: Throwable?) {
         // Always materialize the actual instance of parent's completion exception and cancel child with it
-        child.cancel(job.getCancellationException())
+        childJob.cancel(job.getCancellationException())
     }
-    override fun toString(): String = "Child[$child]"
+    override fun toString(): String = "Child[$childJob]"
 }
 
 private class ChildCompletion(
     private val parent: JobSupport,
-    private val waitChild: Child,
+    private val child: Child,
     private val proposedUpdate: Any?
-) : JobNode<Job>(waitChild.child) {
+) : JobNode<Job>(child.childJob) {
     override fun invoke(reason: Throwable?) {
-        parent.continueCompleting(waitChild, proposedUpdate)
+        parent.continueCompleting(child, proposedUpdate)
     }
 }
 

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

@@ -266,4 +266,49 @@ class CoroutinesTest : TestBase() {
         }
         expectUnreached()
     }
+
+    @Test
+    fun testParentCrashCancelsChildren() = runTest(
+        unhandled = listOf({ it -> it is IOException })
+    ) {
+        expect(1)
+        val parent = launch(coroutineContext + Job()) {
+            expect(4)
+            throw IOException("Crashed")
+        }
+        launch(coroutineContext + parent, CoroutineStart.UNDISPATCHED) {
+            expect(2)
+            try {
+                yield() // to test
+            } finally {
+                expect(5)
+                run(NonCancellable) { yield() } // to test
+                expect(7)
+            }
+            expectUnreached() // will get cancelled, because parent crashes
+        }
+        expect(3)
+        yield() // to parent
+        expect(6)
+        parent.join() // make sure crashed parent still waits for its child
+        finish(8)
+    }
+
+    @Test
+    fun testYieldInFinally() = runTest(
+        expected = { it is IOException }
+    ) {
+        expect(1)
+        try {
+            expect(2)
+            throwIOException()
+        } finally {
+            expect(3)
+            yield()
+            finish(4)
+        }
+        expectUnreached()
+    }
+
+    private fun throwIOException() { throw IOException() }
 }