Streamlined Job APIs that support cancelling state:

Job.isCancelledOrCompleted and Job.invokeOnCancellation are removed,
the later replaced with invokeOnCompletion(handler, onCancelling=true)

Author: elizarov

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

@@ -41,11 +41,23 @@ import kotlin.coroutines.experimental.suspendCoroutine
  * Invocation of [cancel] transitions this continuation from _active_ to _cancelled_ state, while
  * invocation of [resume] or [resumeWithException] transitions it from _active_ to _resumed_ state.
  *
- * A [cancelled][isCancelled] continuation implies that it is [completed][isCompleted], so
- * [invokeOnCancellation] and [invokeOnCompletion] have the same effect.
+ * A [cancelled][isCancelled] continuation implies that it is [completed][isCompleted].
  *
  * Invocation of [resume] or [resumeWithException] in _resumed_ state produces [IllegalStateException]
  * but is ignored in _cancelled_ state.
+ *
+ * ```
+ *    +-----------+   resume    +---------+
+ *    |  Active   | ----------> | Resumed |
+ *    +-----------+             +---------+
+ *          |
+ *          | cancel
+ *          V
+ *    +-----------+
+ *    | Cancelled |
+ *    +-----------+
+ *
+ * ```
  */
 public interface CancellableContinuation<in T> : Continuation<T>, Job {
     /**

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

@@ -115,7 +115,7 @@ internal class DispatchTask<in T>(
         val job = if (cancellable) context[Job] else null
         withCoroutineContext(context) {
             when {
-                job != null && job.isCancelledOrCompleted -> continuation.resumeWithException(job.getCompletionException())
+                job != null && !job.isActive -> continuation.resumeWithException(job.getCompletionException())
                 exception -> continuation.resumeWithException(value as Throwable)
                 else -> continuation.resume(value as T)
             }

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

@@ -24,7 +24,7 @@ import kotlin.coroutines.experimental.CoroutineContext
  */
 public interface CoroutineScope {
     /**
-     * Returns `true` when this coroutine is still active (has not completed yet).
+     * Returns `true` when this coroutine is still active (has not completed and was not cancelled yet).
      *
      * Check this property in long-running computation loops to support cancellation:
      * ```

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

@@ -115,11 +115,6 @@ public interface Job : CoroutineContext.Element {
      */
     public val isCancelled: Boolean
 
-    /**
-     * Returns `true` when this job is either [isCancelled] or [isCompleted].
-     */
-    public val isCancelledOrCompleted: Boolean
-
     /**
      * Returns the exception that signals the completion of this job -- it returns the original
      * [cancel] cause or an instance of [CancellationException] if this job had completed
@@ -171,35 +166,40 @@ public interface Job : CoroutineContext.Element {
     // ------------ low-level state-notification ------------
 
     /**
-     * Registers handler that is **synchronously** invoked on cancellation or completion of this job.
-     * When job is already in _cancelling_ state or is complete for any reason, then the handler
-     * is immediately invoked with a job's cancellation cause or `null`. Otherwise, handler will be
-     * invoked once when this job is [cancelled][cancel] or becomes complete.
-     *
-     * Unlike [invokeOnCompletion], here the handler is immediately invoked on invocation of [cancel]
-     * even if the corresponding coroutine has not finished its execution yet.
+     * Registers handler that is **synchronously** invoked once on completion of this job.
+     * When job is already complete, then the handler is immediately invoked
+     * with a job's cancellation cause or `null`. Otherwise, handler will be invoked once when this
+     * job is complete.
      *
      * 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.
      * There is no need to dispose the handler after completion of this job. The references to
      * all the handlers are released when this job completes.
      *
+     * Note, that the handler is not invoked on invocation of [cancel] when
+     * job becomes _cancelling_, but only when the corresponding coroutine had finished execution
+     * of its code and became _cancelled_. There is an overloaded version of this function
+     * with `onCancelling` parameter to receive notification on _cancelling_ state.
+     *
      * **Note**: This function is a part of internal machinery that supports parent-child hierarchies
      * and allows for implementation of suspending functions that wait on the Job's state.
      * This function should not be used in general application code.
-     * Implementations of `CompletionHandler` must be fast and _lock-free_
+     * Implementations of `CompletionHandler` must be fast and _lock-free_.
      */
-    public fun invokeOnCancellation(handler: CompletionHandler): DisposableHandle
+    public fun invokeOnCompletion(handler: CompletionHandler): DisposableHandle
 
     /**
-     * Registers handler that is **synchronously** invoked on completion of this job.
+     * Registers handler that is **synchronously** invoked once on cancellation or completion of this job.
      * When job is already complete, then the handler is immediately invoked
      * with a job's cancellation cause or `null`. Otherwise, handler will be invoked once when this
-     * job is complete.
+     * job is cancelled or complete.
      *
-     * Unlike [invokeOnCancellation], here the handler is not invoked on invocation of [cancel] when
-     * job becomes _cancelling_, but only when the corresponding coroutine had finished execution
-     * of its code and became _cancelled_.
+     * 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_ when [onCancelling] parameters 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.
      *
      * 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.
@@ -209,9 +209,9 @@ public interface Job : CoroutineContext.Element {
      * **Note**: This function is a part of internal machinery that supports parent-child hierarchies
      * and allows for implementation of suspending functions that wait on the Job's state.
      * This function should not be used in general application code.
-     * Implementations of `CompletionHandler` must be fast and _lock-free_
+     * Implementations of `CompletionHandler` must be fast and _lock-free_.
      */
-    public fun invokeOnCompletion(handler: CompletionHandler): DisposableHandle
+    public fun invokeOnCompletion(handler: CompletionHandler, onCancelling: Boolean): DisposableHandle
 
     // ------------ unstable internal API ------------
 
@@ -272,12 +272,12 @@ public interface DisposableHandle : Job.Registration {
 }
 
 /**
- * Handler for [Job.invokeOnCompletion] and [Job.invokeOnCancellation].
+ * Handler for [Job.invokeOnCompletion].
  *
  * **Note**: This type is a part of internal machinery that supports parent-child hierarchies
  * and allows for implementation of suspending functions that wait on the Job's state.
  * This type should not be used in general application code.
- * Implementations of `CompletionHandler` must be fast and _lock-free_
+ * Implementations of `CompletionHandler` must be fast and _lock-free_.
  */
 public typealias CompletionHandler = (Throwable?) -> Unit
 
@@ -441,13 +441,19 @@ public open class JobSupport(active: Boolean) : AbstractCoroutineContextElement(
             parentHandle = NonDisposableHandle
             return
         }
+        parent.start() // make sure the parent is started
         // directly pass HandlerNode to parent scope to optimize one closure object (see makeNode)
-        val newRegistration = parent.invokeOnCancellation(ParentOnCancellation(parent, this))
+        val newRegistration = parent.invokeOnCompletion(ParentOnCancellation(parent), onCancelling = true)
         parentHandle = newRegistration
         // now check our state _after_ registering (see updateState order of actions)
         if (isCompleted) newRegistration.dispose()
     }
 
+    private inner class ParentOnCancellation(parent: Job) : JobCancellationNode<Job>(parent) {
+        override fun invokeOnce(reason: Throwable?) { onParentCancellation(reason) }
+        override fun toString(): String = "ParentOnCancellation[${this@JobSupport}]"
+    }
+
     /**
      * Invoked at most once on parent completion.
      * @suppress **This is unstable API and it is subject to change.**
@@ -490,11 +496,6 @@ public open class JobSupport(active: Boolean) : AbstractCoroutineContextElement(
         return state is Cancelled || state is Cancelling
     }
 
-    public final override val isCancelledOrCompleted: Boolean get() {
-        val state = this.state
-        return state !is Incomplete || state is Cancelling
-    }
-
     // ------------ state update ------------
 
     /**
@@ -637,20 +638,20 @@ public open class JobSupport(active: Boolean) : AbstractCoroutineContextElement(
         }
     }
 
-    public final override fun invokeOnCancellation(handler: CompletionHandler): DisposableHandle =
-        installHandler(handler, onCancellation = hasCancellingState)
-
     public final override fun invokeOnCompletion(handler: CompletionHandler): DisposableHandle =
-        installHandler(handler, onCancellation = false)
+        installHandler(handler, onCancelling = false)
+
+    public final override fun invokeOnCompletion(handler: CompletionHandler, onCancelling: Boolean): DisposableHandle =
+        installHandler(handler, onCancelling = onCancelling && hasCancellingState)
 
-    private fun installHandler(handler: CompletionHandler, onCancellation: Boolean): DisposableHandle {
+    private fun installHandler(handler: CompletionHandler, onCancelling: Boolean): DisposableHandle {
         var nodeCache: JobNode<*>? = null
         lockFreeLoopOnState { state ->
             when (state) {
                 is Empty -> { // EMPTY_X state -- no completion handlers
                     if (state.isActive) {
                         // try move to SINGLE state
-                        val node = nodeCache ?: makeNode(handler, onCancellation).also { nodeCache = it }
+                        val node = nodeCache ?: makeNode(handler, onCancelling).also { nodeCache = it }
                         if (STATE.compareAndSet(this, state, node)) return node
                     } else
                         promoteEmptyToNodeList(state) // that way we can add listener for non-active coroutine
@@ -659,15 +660,15 @@ public open class JobSupport(active: Boolean) : AbstractCoroutineContextElement(
                     promoteSingleToNodeList(state)
                 }
                 is NodeList -> { // LIST -- a list of completion handlers (either new or active)
-                    val node = nodeCache ?: makeNode(handler, onCancellation).also { nodeCache = it }
+                    val node = nodeCache ?: makeNode(handler, onCancelling).also { nodeCache = it }
                     if (addLastAtomic(state, state, node)) return node
                 }
                 is Cancelling -> { // CANCELLING -- has a list of completion handlers
-                    if (onCancellation) { // installing cancellation handler on job that is being cancelled
+                    if (onCancelling) { // installing cancellation handler on job that is being cancelled
                         handler((state as? CompletedExceptionally)?.exception)
                         return NonDisposableHandle
                     }
-                    val node = nodeCache ?: makeNode(handler, onCancellation).also { nodeCache = it }
+                    val node = nodeCache ?: makeNode(handler, onCancelling).also { nodeCache = it }
                     if (addLastAtomic(state, state.list, node)) return node
                 }
                 else -> { // is inactive
@@ -678,8 +679,8 @@ public open class JobSupport(active: Boolean) : AbstractCoroutineContextElement(
         }
     }
 
-    private fun makeNode(handler: CompletionHandler, onCancellation: Boolean): JobNode<*> =
-        if (onCancellation)
+    private fun makeNode(handler: CompletionHandler, onCancelling: Boolean): JobNode<*> =
+        if (onCancelling)
             (handler as? JobCancellationNode<*>)?.also { require(it.job === this) }
                 ?: InvokeOnCancellation(this, handler)
         else
@@ -816,14 +817,15 @@ public open class JobSupport(active: Boolean) : AbstractCoroutineContextElement(
 
     /**
      * Override to process any exceptions that were encountered while invoking completion handlers
-     * installed via [invokeOnCancellation] or [invokeOnCompletion].
+     * installed via [invokeOnCompletion].
      */
     protected open fun handleException(exception: Throwable) {
         throw exception
     }
 
     /**
-     * It is invoked once when job is cancelled or is completed, similarly to [invokeOnCancellation].
+     * It is invoked once when job is cancelled or is completed, similarly to [invokeOnCompletion] with
+     * `onCancelling` set to `true`.
      */
     protected open fun onCancellation() {}
 
@@ -912,14 +914,6 @@ public open class JobSupport(active: Boolean) : AbstractCoroutineContextElement(
     ) : CompletedExceptionally(cause)
 
 
-    private class ParentOnCancellation(
-        parentJob: Job,
-        private val subordinateJob: JobSupport
-    ) : JobCancellationNode<Job>(parentJob) {
-        override fun invokeOnce(reason: Throwable?) { subordinateJob.onParentCancellation(reason) }
-        override fun toString(): String = "ParentOnCancellation[$subordinateJob]"
-    }
-
     /*
      * =================================================================================================
      * This is ready-to-use implementation for Deferred interface.

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

@@ -41,9 +41,6 @@ object NonCancellable : AbstractCoroutineContextElement(Job), Job {
     /** Always returns `false`. */
     override val isCancelled: Boolean get() = false
 
-    /** Always returns `false`. */
-    override val isCancelledOrCompleted: Boolean get() = false
-
     /** Always returns `false`. */
     override fun start(): Boolean = false
 
@@ -64,10 +61,10 @@ object NonCancellable : AbstractCoroutineContextElement(Job), Job {
     override fun getCompletionException(): CancellationException = throw IllegalStateException("This job is always active")
 
     /** Always returns [NonDisposableHandle]. */
-    override fun invokeOnCancellation(handler: CompletionHandler): DisposableHandle = NonDisposableHandle
+    override fun invokeOnCompletion(handler: CompletionHandler): DisposableHandle = NonDisposableHandle
 
     /** Always returns [NonDisposableHandle]. */
-    override fun invokeOnCompletion(handler: CompletionHandler): DisposableHandle = NonDisposableHandle
+    override fun invokeOnCompletion(handler: CompletionHandler, onCancelling: Boolean): DisposableHandle = NonDisposableHandle
 
     /** Always returns `false`. */
     override fun cancel(cause: Throwable?): Boolean = false

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

@@ -30,7 +30,7 @@ import kotlin.coroutines.experimental.intrinsics.suspendCoroutineOrReturn
 suspend fun yield(): Unit = suspendCoroutineOrReturn sc@ { cont ->
     val context = cont.context
     val job = context[Job]
-    if (job != null && job.isCancelledOrCompleted) throw job.getCompletionException()
+    if (job != null && !job.isActive) throw job.getCompletionException()
     if (cont !is DispatchedContinuation<Unit>) return@sc Unit
     if (!cont.dispatcher.isDispatchNeeded(context)) return@sc Unit
     cont.dispatchYield(Unit)

kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/selects/Select.kt

@@ -284,15 +284,20 @@ internal class SelectBuilderImpl<in R>(
 
     private fun initCancellability() {
         val parent = context[Job] ?: return
-        val newRegistration = parent.invokeOnCancellation { cause ->
-            if (trySelect(null))
-                resumeSelectCancellableWithException(cause ?: CancellationException("Select was cancelled"))
-        }
+        val newRegistration = parent.invokeOnCompletion(SelectOnCancellation(parent), onCancelling = true)
         parentHandle = newRegistration
         // now check our state _after_ registering
         if (isSelected) newRegistration.dispose()
     }
 
+    private inner class SelectOnCancellation(job: Job) : JobCancellationNode<Job>(job) {
+        override fun invokeOnce(reason: Throwable?) {
+            if (trySelect(null))
+                resumeSelectCancellableWithException(reason ?: CancellationException("Select was cancelled"))
+        }
+        override fun toString(): String = "SelectOnCancellation[${this@SelectBuilderImpl}]"
+    }
+
     private val state: Any? get() {
         while (true) { // lock-free helping loop
             val state = _state

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

@@ -51,7 +51,7 @@ class JobDisposeTest: TestBase() {
         threads += testThread("creator") {
             while (!done) {
                 val job = TestJob()
-                val handle = job.invokeOnCancellation { /* nothing */ }
+                val handle = job.invokeOnCompletion({ /* nothing */ }, onCancelling = true)
                 this.job = job // post job to cancelling thread
                 this.handle = handle // post handle to concurrent disposer thread
                 handle.dispose() // dispose of handle from this thread (concurrently with other disposer)

reactive/kotlinx-coroutines-reactive/src/main/kotlin/kotlinx/coroutines/experimental/reactive/Publish.kt

@@ -113,7 +113,7 @@ private class PublisherCoroutine<T>(
     // assert: mutex.isLocked()
     private fun doLockedNext(elem: T) {
         // check if already closed for send
-        if (isCancelledOrCompleted) {
+        if (!isActive) {
             doLockedSignalCompleted()
             throw sendException()
         }
@@ -141,14 +141,14 @@ private class PublisherCoroutine<T>(
             }
         }
         /*
-           There is no sense to check for `isCompleted` before doing `unlock`, because completion might
-           happen after this check and before `unlock` (see `afterCompleted` that does not do anything
+           There is no sense to check for `isActive` before doing `unlock`, because cancellation/completion might
+           happen after this check and before `unlock` (see `onCancellation` that does not do anything
            if it fails to acquire the lock that we are still holding).
-           We have to recheck `isCompleted` after `unlock` anyway.
+           We have to recheck `isActive` after `unlock` anyway.
          */
         mutex.unlock()
-        // recheck isCancelledOrCompleted
-        if (isCancelledOrCompleted && mutex.tryLock())
+        // recheck isActive
+        if (!isActive && mutex.tryLock())
             doLockedSignalCompleted()
     }
 
@@ -188,8 +188,8 @@ private class PublisherCoroutine<T>(
                 // unlock the mutex when we don't have back-pressure anymore
                 if (cur == 0L) {
                     mutex.unlock()
-                    // recheck isCancelledOrCompleted
-                    if (isCancelledOrCompleted && mutex.tryLock())
+                    // recheck isActive
+                    if (!isActive && mutex.tryLock())
                         doLockedSignalCompleted()
                 }
                 return
@@ -200,7 +200,7 @@ private class PublisherCoroutine<T>(
     override fun onCancellation() {
         while (true) { // lock-free loop for nRequested
             val cur = nRequested
-            if (cur == SIGNALLED) return // some other thread holding lock already signalled completion
+            if (cur == SIGNALLED) return // some other thread holding lock already signalled cancellation/completion
             check(cur >= 0) // no other thread could have marked it as CLOSED, because onCancellation is invoked once
             if (!N_REQUESTED.compareAndSet(this, cur, CLOSED)) continue // retry on failed CAS
             // Ok -- marked as CLOSED, now can unlock the mutex if it was locked due to backpressure