Fix exception aggregation to ensure atomic handling of exceptions

* Removed legacy onFinishing handler support from JobSupport.
  - It is no longer needed, because handleJobException covers #208
* Fixed bugs that were masked by cancelling parent from onFinishing.
* Consistent "Finishing" state was introduced in internal machinery:
  - Job enters finishing state when it is failing or it is completing
    and has children
  - Finishing state cleanly aggregates all failures, tracks cancellation
    and completion status
* Source-incompatible change: Job.invokeOnCompletion boolean parameter is
  change to onFailing. Such handlers are now invoked as soon as the job
  starts failing and the root cause exception of the failure is consistently
  passed to all the handlers.
* The following internal methods were introduced to facilitate this:
  - Job.failParent(exception) is used by child to signal failure to parent
  - Job.cancelChild(parentJob) is used by parent to cancel child.
* Child never aggregates exception received from it parent, but uses
  it as it root cause if there is no other exception.
* JobSupport.handleJobException() logic for launch/actor is split into:
  - failParent - can be invoked multiple times on race;
  - handleJobException which is invoked exactly once.
* Exception materiazization is much lazier now, which should
  significantly improve performance when cancelling large hierarchies.
* Other minor perf improvements in JobSupport code.

Fixes #585

Author: elizarov

binary-compatibility-validator/reference-public-api/kotlinx-coroutines-core.txt

@@ -1,4 +1,5 @@
 public abstract class kotlinx/coroutines/experimental/AbstractCoroutine : kotlin/coroutines/experimental/Continuation, kotlinx/coroutines/experimental/CoroutineScope, kotlinx/coroutines/experimental/Job {
+	protected final field parentContext Lkotlin/coroutines/experimental/CoroutineContext;
 	public fun <init> (Lkotlin/coroutines/experimental/CoroutineContext;Z)V
 	public synthetic fun <init> (Lkotlin/coroutines/experimental/CoroutineContext;ZILkotlin/jvm/internal/DefaultConstructorMarker;)V
 	public final fun getContext ()Lkotlin/coroutines/experimental/CoroutineContext;
@@ -369,7 +370,9 @@ public abstract interface class kotlinx/coroutines/experimental/Job : kotlin/cor
 	public static final field Key Lkotlinx/coroutines/experimental/Job$Key;
 	public abstract fun attachChild (Lkotlinx/coroutines/experimental/Job;)Lkotlinx/coroutines/experimental/DisposableHandle;
 	public abstract fun cancel (Ljava/lang/Throwable;)Z
+	public abstract fun cancelChild (Lkotlinx/coroutines/experimental/Job;)V
 	public abstract synthetic fun cancelChildren (Ljava/lang/Throwable;)V
+	public abstract fun childFailed (Ljava/lang/Throwable;)Z
 	public abstract fun getCancellationException ()Ljava/util/concurrent/CancellationException;
 	public abstract fun getChildren ()Lkotlin/sequences/Sequence;
 	public abstract fun getCompletionException ()Ljava/lang/Throwable;
@@ -440,7 +443,9 @@ public final class kotlinx/coroutines/experimental/NonCancellable : kotlin/corou
 	public static final field INSTANCE Lkotlinx/coroutines/experimental/NonCancellable;
 	public fun attachChild (Lkotlinx/coroutines/experimental/Job;)Lkotlinx/coroutines/experimental/DisposableHandle;
 	public fun cancel (Ljava/lang/Throwable;)Z
+	public fun cancelChild (Lkotlinx/coroutines/experimental/Job;)V
 	public synthetic fun cancelChildren (Ljava/lang/Throwable;)V
+	public fun childFailed (Ljava/lang/Throwable;)Z
 	public fun getCancellationException ()Ljava/util/concurrent/CancellationException;
 	public fun getChildren ()Lkotlin/sequences/Sequence;
 	public fun getCompletionException ()Ljava/lang/Throwable;

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

@@ -89,7 +89,7 @@ internal abstract class AbstractContinuation<in T>(
             return
         }
         parent.start() // make sure the parent is started
-        val handle = parent.invokeOnCompletion(onCancelling = true,
+        val handle = parent.invokeOnCompletion(onFailing = true,
             handler = ChildContinuation(parent, this).asHandler)
 
         parentHandle = handle

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

@@ -5,6 +5,7 @@
 package kotlinx.coroutines.experimental
 
 import kotlinx.coroutines.experimental.CoroutineStart.*
+import kotlinx.coroutines.experimental.internal.*
 import kotlinx.coroutines.experimental.intrinsics.*
 import kotlin.coroutines.experimental.*
 
@@ -30,7 +31,11 @@ import kotlin.coroutines.experimental.*
  */
 @Suppress("EXPOSED_SUPER_CLASS")
 public abstract class AbstractCoroutine<in T>(
-    private val parentContext: CoroutineContext,
+    /**
+     * Context of the parent coroutine.
+     */
+    @JvmField
+    protected val parentContext: CoroutineContext,
     active: Boolean = true
 ) : JobSupport(active), Job, Continuation<T>, CoroutineScope {
     @Suppress("LeakingThis")
@@ -63,19 +68,25 @@ public abstract class AbstractCoroutine<in T>(
     }
 
     /**
-     * This function is invoked once when this coroutine is cancelled or is completed,
-     * similarly to [invokeOnCompletion] with `onCancelling` set to `true`.
+     * @suppress **Deprecated**: Override [onFailing].
+     */
+    @Deprecated("Override onFailing")
+    protected open fun onCancellation(cause: Throwable?) {}
+
+    /**
+     * This function is invoked once when this coroutine is failing or is completed,
+     * similarly to [invokeOnCompletion] with `onFailing` set to `true`.
      *
      * The meaning of [cause] parameter:
      * * Cause is `null` when job has completed normally.
      * * Cause is an instance of [CancellationException] when job was cancelled _normally_.
      *   **It should not be treated as an error**. In particular, it should not be reported to error logs.
      * * Otherwise, the job had _failed_.
+     *
+     * @suppress **Deprecated**: Override [onFailing].
      */
-    protected open fun onCancellation(cause: Throwable?) {}
-
-    internal override fun onCancellationInternal(exceptionally: CompletedExceptionally?) {
-        onCancellation(exceptionally?.cause)
+    override fun onFailing(cause: Throwable?) {
+        onCancellation(cause)
     }
 
     /**
@@ -89,7 +100,7 @@ public abstract class AbstractCoroutine<in T>(
     protected open fun onCompletedExceptionally(exception: Throwable) {}
 
     @Suppress("UNCHECKED_CAST")
-    internal override fun onCompletionInternal(state: Any?, mode: Int) {
+    internal override fun onCompletionInternal(state: Any?, mode: Int, suppressed: Boolean) {
         if (state is CompletedExceptionally)
             onCompletedExceptionally(state.cause)
         else
@@ -112,6 +123,13 @@ public abstract class AbstractCoroutine<in T>(
         makeCompletingOnce(CompletedExceptionally(exception), defaultResumeMode)
     }
 
+    // todo: make it for all kinds of coroutines, now only launch & actor override and handleExceptionViaJob
+    internal fun failParentImpl(exception: Throwable): Boolean {
+        if (exception is CancellationException) return true
+        val parentJob = parentContext[Job]
+        return parentJob !== null && parentJob.childFailed(exception)
+    }
+
     internal final override fun handleOnCompletionException(exception: Throwable) {
         handleCoroutineException(parentContext, exception, this)
     }

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

@@ -207,20 +207,11 @@ public suspend fun <T> run(context: CoroutineContext, block: suspend () -> T): T
 // --------------- implementation ---------------
 
 private open class StandaloneCoroutine(
-    private val parentContext: CoroutineContext,
+    parentContext: CoroutineContext,
     active: Boolean
 ) : AbstractCoroutine<Unit>(parentContext, active) {
-    override fun hasOnFinishingHandler(update: Any?) = update is CompletedExceptionally
-
-    override fun handleJobException(exception: Throwable) {
-        handleCoroutineException(parentContext, exception, this)
-    }
-
-    override fun onFinishingInternal(update: Any?) {
-        if (update is CompletedExceptionally && update.cause !is CancellationException) {
-            parentContext[Job]?.cancel(update.cause)
-        }
-    }
+    override fun failParent(exception: Throwable) = failParentImpl(exception)
+    override fun handleJobException(exception: Throwable) = handleExceptionViaHandler(parentContext, exception)
 }
 
 private class LazyStandaloneCoroutine(

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

@@ -61,7 +61,7 @@ private class CompletableDeferredImpl<T>(
     parent: Job?
 ) : JobSupport(true), CompletableDeferred<T>, SelectClause1<T> {
     init { initParentJobInternal(parent) }
-    override val onCancelMode: Int get() = ON_CANCEL_MAKE_COMPLETING
+    override val onFailComplete get() = true
     override fun getCompleted(): T = getCompletedInternal() as T
     override suspend fun await(): T = awaitInternal() as T
     override val onAwait: SelectClause1<T> get() = this

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

@@ -13,8 +13,6 @@ import kotlin.coroutines.experimental.*
  * **Note: This class cannot be used outside of internal coroutines framework**.
  * **Note: cannot be internal until we get rid of MutableDelegateContinuation in IO**
  *
- * @param cause the exceptional completion cause. It's either original exceptional cause
- *        or artificial JobCancellationException if no cause was provided
  * @suppress **This is unstable API and it is subject to change.**
  */
 open class CompletedExceptionally(
@@ -28,14 +26,9 @@ open class CompletedExceptionally(
  *
  * **Note: This class cannot be used outside of internal coroutines framework**.
  *
- * @param job the job that was cancelled.
- * @param cause the exceptional completion cause. If `cause` is null, then a [JobCancellationException] is created.
  * @suppress **This is unstable API and it is subject to change.**
  */
-internal class Cancelled(
-    job: Job,
-    cause: Throwable?
-) : CompletedExceptionally(cause ?: JobCancellationException("Job was cancelled normally", null, job))
+internal class Cancelled(cause: Throwable) : CompletedExceptionally(cause)
 
 /**
  * A specific subclass of [CompletedExceptionally] for cancelled [AbstractContinuation].

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

@@ -43,3 +43,5 @@ internal expect val CancelHandlerBase.asHandler: CompletionHandler
 // :KLUDGE: We have to invoke a handler in platform-specific way via `invokeIt` extension,
 // because we play type tricks on Kotlin/JS and handler is not necessarily a function there
 internal expect fun CompletionHandler.invokeIt(cause: Throwable?)
+
+internal inline fun <reified T> CompletionHandler.isHandlerOf(): Boolean = this is T

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

@@ -20,22 +20,27 @@ internal expect fun handleCoroutineExceptionImpl(context: CoroutineContext, exce
  * If invocation returned `true`, method terminates: now [Job] is responsible for handling an exception.
  * Otherwise, If there is [CoroutineExceptionHandler] in the context, it is used.
  * Otherwise all instances of [CoroutineExceptionHandler] found via [ServiceLoader] and [Thread.uncaughtExceptionHandler] are invoked
+ *
+ * todo: Deprecate/hide this function.
  */
 @JvmOverloads // binary compatibility
 public fun handleCoroutineException(context: CoroutineContext, exception: Throwable, caller: Job? = null) {
-    // if exception handling fails, make sure the original exception is not lost
+    if (!handleExceptionViaJob(context, exception, caller)) {
+        handleExceptionViaHandler(context, exception)
+    }
+}
+
+private fun handleExceptionViaJob(context: CoroutineContext, exception: Throwable, caller: Job?): Boolean {
+    // Ignore CancellationException (they are normal ways to terminate a coroutine)
+    if (exception is CancellationException) return true
+    // If job is successfully cancelled, we're done
+    val job = context[Job]
+    return job !== null && job !== caller && job.cancel(exception)
+}
+
+internal fun handleExceptionViaHandler(context: CoroutineContext, exception: Throwable) {
     try {
-        // Ignore CancellationException (they are normal ways to terminate a coroutine)
-        if (exception is CancellationException) {
-            return
-        }
-        // If parent is successfully cancelled, we're done, it is now its responsibility to handle the exception
-        val parent = context[Job]
-        // E.g. actor registers itself in the context, in that case we should invoke handler
-        if (parent !== null && parent !== caller && parent.cancel(exception)) {
-            return
-        }
-        // If not, invoke exception handler from the context
+        // Invoke exception handler from the context if present
         context[CoroutineExceptionHandler]?.let {
             it.handleException(context, exception)
             return

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

@@ -175,7 +175,7 @@ object GlobalScope : CoroutineScope {
  */
 public suspend fun <R> coroutineScope(block: suspend CoroutineScope.() -> R): R {
     // todo: optimize implementation to a single allocated object
-    val owner = ScopeOwnerCoroutine<R>(coroutineContext)
+    val owner = ScopeCoroutine<R>(coroutineContext)
     owner.start(CoroutineStart.UNDISPATCHED, owner, block)
     owner.join()
     if (owner.isCancelled) {

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

@@ -127,8 +127,9 @@ public interface Job : CoroutineContext.Element {
      * returned. The [JobCancellationException.cause] of the resulting [JobCancellationException] references
      * the original cancellation cause that was passed to [cancel] function.
      *
-     * This function throws [IllegalStateException] when invoked on a job that has not
-     * [completed][isCompleted] nor [cancelled][isCancelled] yet.
+     * This function throws [IllegalStateException] when invoked on a job that is still active.
+     *
+     * @suppress **This is unstable API and it is subject to change.**
      */
     public fun getCancellationException(): CancellationException
 
@@ -163,6 +164,25 @@ public interface Job : CoroutineContext.Element {
     public fun cancel(cause: Throwable? = null): Boolean
 
     // ------------ parent-child ------------
+    
+    /**
+     * Child is reporting failure to the parent by invoking this method.
+     * This method is invoked by the child twice. The first time child report its root cause as soon as possible,
+     * so that all its siblings and the parent can start finishing their work asap on failure. The second time
+     * child invokes this method when it had aggregated and determined its final termination cause.
+     *
+     * @suppress **This is unstable API and it is subject to change.**
+     */
+    public fun childFailed(cause: Throwable): Boolean
+
+    /**
+     * Cancels child job. This method is invoked by [parentJob] to cancel this child job.
+     * Child finds the cancellation cause using [getCancellationException] of the [parentJob].
+     * This method does nothing is the child is already being cancelled.
+     *
+     * @suppress **This is unstable API and it is subject to change.**
+     */
+    public fun cancelChild(parentJob: Job)
 
     /**
      * Returns a sequence of this job's children.
@@ -201,9 +221,9 @@ public interface Job : CoroutineContext.Element {
      * lookup a [Job] instance in the parent context and use this function to attach themselves as a child.
      * They also store a reference to the resulting [DisposableHandle] and dispose a handle when they complete.
      *
-     * @suppress This is an internal API. This method is too error prone for public API.
+     * @suppress **This is unstable API and it is subject to change.**
+     *           This is an internal API. This method is too error prone for public API.
      */
-    @Deprecated(message = "Start child coroutine with 'parent' parameter", level = DeprecationLevel.WARNING)
     public fun attachChild(child: Job): DisposableHandle
 
     /**
@@ -285,23 +305,20 @@ public interface Job : CoroutineContext.Element {
     public fun invokeOnCompletion(handler: CompletionHandler): DisposableHandle
 
     /**
-     * Registers handler that is **synchronously** invoked once on cancellation or completion of this job.
-     * When job is already cancelling or complete, then the handler is immediately invoked
+     * Registers handler that is **synchronously** invoked once on failure or completion of this job.
+     * When job is already failing or complete, then the handler is immediately invoked
      * with a job's cancellation cause or `null` unless [invokeImmediately] is set to false.
-     * Otherwise, handler will be invoked once when this job is cancelled or complete.
+     * Otherwise, handler will be invoked once when this job is failing or is complete.
      *
      * The meaning of `cause` that is passed to the handler:
      * * Cause is `null` when job has completed normally.
      * * Cause is an instance of [CancellationException] when job was cancelled _normally_.
      *   **It should not be treated as an error**. In particular, it should not be reported to error logs.
      * * Otherwise, the job had _failed_.
      *
-     * Invocation of this handler on a transition to a transient _cancelling_ state
-     * is controlled by [onCancelling] boolean parameter.
-     * The handler is invoked on invocation of [cancel] when
-     * job becomes _cancelling_ if [onCancelling] parameter is set to `true`. However,
-     * when this [Job] is not backed by a coroutine, like [CompletableDeferred] or [CancellableContinuation]
-     * (both of which do not posses a _cancelling_ state), then the value of [onCancelling] parameter is ignored.
+     * Invocation of this handler on a transition to a _failing_ state
+     * is controlled by [onFailing] boolean parameter.
+     * The handler is invoked when the job is failing when [onFailing] parameter is set to `true`.
      *
      * The resulting [DisposableHandle] can be used to [dispose][DisposableHandle.dispose] the
      * registration of this handler and release its memory if its invocation is no longer needed.
@@ -316,15 +333,15 @@ public interface Job : CoroutineContext.Element {
      * This function should not be used in general application code.
      * Implementations of `CompletionHandler` must be fast and _lock-free_.
      *
-     * @param onCancelling when `true`, then the [handler] is invoked as soon as this job transitions to _cancelling_ state;
+     * @param onFailing when `true`, then the [handler] is invoked as soon as this job transitions to _failing_ state;
      *        when `false` then the [handler] is invoked only when it transitions to _completed_ state.
-     * @param invokeImmediately when `true` and this job is already in the desired state (depending on [onCancelling]),
+     * @param invokeImmediately when `true` and this job is already in the desired state (depending on [onFailing]),
      *        then the [handler] is immediately and synchronously invoked and [NonDisposableHandle] is returned;
      *        when `false` then [NonDisposableHandle] is returned, but the [handler] is not invoked.
      * @param handler the handler.
      */
     public fun invokeOnCompletion(
-        onCancelling: Boolean = false,
+        onFailing: Boolean = false,
         invokeImmediately: Boolean = true,
         handler: CompletionHandler): DisposableHandle