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
* Job.isFailed is introduced as a consistent way to query the "failing"
  state of the job that was previously only implicitly available via
  invokeOnCompletion handler (cause != null means a failed job) and
  the documentation for both Job & Deferred is updated to reflect that.
* 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/AbstractCoroutine : kotlin/coroutines/Continuation, kotlinx/coroutines/CoroutineScope, kotlinx/coroutines/Job {
+	protected final field parentContext Lkotlin/coroutines/experimental/CoroutineContext;
 	public fun <init> (Lkotlin/coroutines/CoroutineContext;Z)V
 	public synthetic fun <init> (Lkotlin/coroutines/CoroutineContext;ZILkotlin/jvm/internal/DefaultConstructorMarker;)V
 	public final fun getContext ()Lkotlin/coroutines/CoroutineContext;
@@ -376,7 +377,9 @@ public abstract interface class kotlinx/coroutines/Job : kotlin/coroutines/Corou
 	public static final field Key Lkotlinx/coroutines/Job$Key;
 	public abstract fun attachChild (Lkotlinx/coroutines/Job;)Lkotlinx/coroutines/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;
@@ -388,6 +391,7 @@ public abstract interface class kotlinx/coroutines/Job : kotlin/coroutines/Corou
 	public abstract fun isActive ()Z
 	public abstract fun isCancelled ()Z
 	public abstract fun isCompleted ()Z
+	public abstract fun isFailed ()Z
 	public abstract fun join (Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public abstract fun plus (Lkotlinx/coroutines/Job;)Lkotlinx/coroutines/Job;
 	public abstract fun start ()Z
@@ -447,7 +451,9 @@ public final class kotlinx/coroutines/NonCancellable : kotlin/coroutines/Abstrac
 	public static final field INSTANCE Lkotlinx/coroutines/NonCancellable;
 	public fun attachChild (Lkotlinx/coroutines/Job;)Lkotlinx/coroutines/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;
@@ -459,6 +465,7 @@ public final class kotlinx/coroutines/NonCancellable : kotlin/coroutines/Abstrac
 	public fun isActive ()Z
 	public fun isCancelled ()Z
 	public fun isCompleted ()Z
+	public fun isFailed ()Z
 	public fun join (Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public fun plus (Lkotlinx/coroutines/Job;)Lkotlinx/coroutines/Job;
 	public fun start ()Z

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
 
 import kotlinx.coroutines.CoroutineStart.*
+import kotlinx.coroutines.internal.*
 import kotlinx.coroutines.intrinsics.*
 import kotlin.coroutines.*
 
@@ -30,7 +31,11 @@ import kotlin.coroutines.*
  */
 @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
@@ -105,6 +116,13 @@ public abstract class AbstractCoroutine<in T>(
         makeCompletingOnce(result.toState(), 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

@@ -210,20 +210,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

@@ -17,10 +17,8 @@ public fun <T> Result<T>.toState(): Any? =
  * Class for an internal state of a job that had completed exceptionally, including cancellation.
  *
  * **Note: This class cannot be used outside of internal coroutines framework**.
- * **Note: cannot be internal until we get rid of MutableDelegateContinuation in IO**
+ * **Note: cannot be internal and renamed 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(
@@ -34,14 +32,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

@@ -180,7 +180,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/Deferred.kt

@@ -12,69 +12,35 @@ import kotlin.coroutines.*
 import kotlin.experimental.*
 
 /**
- * Deferred value is a non-blocking cancellable future.
+ * Deferred value is a non-blocking cancellable future — it is a [Job] that has a result.
  *
  * It is created with [async][CoroutineScope.async] coroutine builder or via constructor of [CompletableDeferred] class.
  * It is in [active][isActive] state while the value is being computed.
  *
- * Deferred value has the following states:
- *
- * | **State**                               | [isActive] | [isCompleted] | [isCompletedExceptionally] | [isCancelled] |
- * | --------------------------------------- | ---------- | ------------- | -------------------------- | ------------- |
- * | _New_ (optional initial state)          | `false`    | `false`       | `false`                    | `false`       |
- * | _Active_ (default initial state)        | `true`     | `false`       | `false`                    | `false`       |
- * | _Completing_ (optional transient state) | `true`     | `false`       | `false`                    | `false`       |
- * | _Cancelling_ (optional transient state) | `false`    | `false`       | `false`                    | `true`        |
- * | _Cancelled_ (final state)               | `false`    | `true`        | `true`                     | `true`        |
- * | _Resolved_  (final state)               | `false`    | `true`        | `false`                    | `false`       |
- * | _Failed_    (final state)               | `false`    | `true`        | `true`                     | `false`       |
+ * Deferred value has the same state machine as the [Job] with additional convenience methods to retrieve
+ * successful or failed result of the computation that was carried out. The result of the deferred is
+ * available when it is [completed][isCompleted] and can be retrieved by [await] method, which throws
+ * exception if the deferred had failed.
+ * A _failed_ deferred is considered to be [completed exceptionally][isCompletedExceptionally].
+ * The corresponding exception can be retrieved via [getCompletionExceptionOrNull] from a completed instance of deferred.
  *
  * Usually, a deferred value is created in _active_ state (it is created and started).
  * However, [async][CoroutineScope.async] coroutine builder has an optional `start` parameter that creates a deferred value in _new_ state
  * when this parameter is set to [CoroutineStart.LAZY].
  * Such a deferred can be be made _active_ by invoking [start], [join], or [await].
  *
- * A deferred can be _cancelled_ at any time with [cancel] function that forces it to transition to
- * _cancelling_ state immediately. Deferred that is not backed by a coroutine (see [CompletableDeferred]) and does not have
- * [children] becomes _cancelled_ on [cancel] immediately.
- * Otherwise, deferred becomes _cancelled_  when it finishes executing its code and
- * when all its children [complete][isCompleted].
- *
- * ```
- *                                                     wait children
- *    +-----+       start      +--------+   complete  +-------------+ finish +-----------+
- *    | New | ---------------> | Active | ----------> | Completing  | ---+-> | Resolved  |
- *    +-----+                  +--------+             +-------------+    |   |(completed)|
- *       |                         |                        |            |   +-----------+
- *       | cancel                  | cancel                 | cancel     |
- *       V                         V                        |            |   +-----------+
- *  +-----------+   finish   +------------+                 |            +-> |  Failed   |
- *  | Cancelled | <--------- | Cancelling | <---------------+                |(completed)|
- *  |(completed)|            +------------+                                  +-----------+
- *  +-----------+
- * ```
- *
  * A deferred value is a [Job]. A job in the
  * [coroutineContext](https://kotlinlang.org/api/latest/jvm/stdlib/kotlin.coroutines/coroutine-context.html)
  * of [async][CoroutineScope.async] builder represents the coroutine itself.
- * A deferred value is active while the coroutine is working and cancellation aborts the coroutine when
- * the coroutine is suspended on a _cancellable_ suspension point by throwing [CancellationException]
- * or the cancellation cause inside the coroutine.
- *
- * A deferred value can have a _parent_ job. A deferred value with a parent is cancelled when its parent is
- * cancelled or completes. Parent waits for all its [children] to complete in _completing_ or
- * _cancelling_ state. _Completing_ state is purely internal. For an outside observer a _completing_
- * deferred is still active, while internally it is waiting for its children.
  *
  * All functions on this interface and on all interfaces derived from it are **thread-safe** and can
  * be safely invoked from concurrent coroutines without external synchronization.
  */
 public interface Deferred<out T> : Job {
     /**
-     * Returns `true` if computation of this deferred value has _completed exceptionally_ -- it had
-     * either _failed_ with exception during computation or was [cancelled][cancel].
-     *
-     * It implies that [isActive] is `false` and [isCompleted] is `true`.
+     * Returns `true` if computation of this deferred value has _completed exceptionally_.
+     * It is `true` when both [isCompleted] and [isFailed] are true.
+     * It implies that [isActive] is `false`.
      */
     public val isCompletedExceptionally: Boolean