Job join/await waits for coroutine code completion,

"Cancelling" state is introduced for Job/Deferred objects.

Author: elizarov

integration/kotlinx-coroutines-guava/src/main/kotlin/kotlinx/coroutines/experimental/guava/ListenableFuture.kt

@@ -95,7 +95,7 @@ private class DeferredListenableFuture<T>(
  * Awaits for completion of the future without blocking a thread.
  *
  * This suspending function is cancellable.
- * If the [Job] of the current coroutine is completed while this suspending function is waiting, this function
+ * If the [Job] of the current coroutine is cancelled or completed while this suspending function is waiting, this function
  * stops waiting for the future and immediately resumes with [CancellationException].
  *
  * Note, that `ListenableFuture` does not support removal of installed listeners, so on cancellation of this wait

integration/kotlinx-coroutines-jdk8/src/main/kotlin/kotlinx/coroutines/experimental/future/Future.kt

@@ -102,7 +102,7 @@ public suspend fun <T> CompletionStage<T>.await(): T = suspendCoroutine { cont:
  * Awaits for completion of the future without blocking a thread.
  *
  * This suspending function is cancellable.
- * If the [Job] of the current coroutine is completed while this suspending function is waiting, this function
+ * If the [Job] of the current coroutine is cancelled or completed while this suspending function is waiting, this function
  * stops waiting for the future and immediately resumes with [CancellationException].
  *
  * Note, that `CompletableFuture` does not support prompt removal of installed listeners, so on cancellation of this wait

integration/kotlinx-coroutines-nio/src/main/kotlin/kotlinx/coroutines/experimental/nio/Nio.kt

@@ -28,7 +28,7 @@ import java.util.concurrent.TimeUnit
 /**
  * Performs [AsynchronousFileChannel.lock] without blocking a thread and resumes when asynchronous operation completes.
  * This suspending function is cancellable.
- * If the [Job] of the current coroutine is completed while this suspending function is suspended, this function
+ * If the [Job] of the current coroutine is cancelled or completed while this suspending function is waiting, this function
  * *closes the underlying channel* and immediately resumes with [CancellationException].
  */
 suspend fun AsynchronousFileChannel.aLock() = suspendCancellableCoroutine<FileLock> { cont ->
@@ -39,7 +39,7 @@ suspend fun AsynchronousFileChannel.aLock() = suspendCancellableCoroutine<FileLo
 /**
  * Performs [AsynchronousFileChannel.lock] without blocking a thread and resumes when asynchronous operation completes.
  * This suspending function is cancellable.
- * If the [Job] of the current coroutine is completed while this suspending function is suspended, this function
+ * If the [Job] of the current coroutine is cancelled or completed while this suspending function is waiting, this function
  * *closes the underlying channel* and immediately resumes with [CancellationException].
  */
 suspend fun AsynchronousFileChannel.aLock(
@@ -54,7 +54,7 @@ suspend fun AsynchronousFileChannel.aLock(
 /**
  * Performs [AsynchronousFileChannel.read] without blocking a thread and resumes when asynchronous operation completes.
  * This suspending function is cancellable.
- * If the [Job] of the current coroutine is completed while this suspending function is suspended, this function
+ * If the [Job] of the current coroutine is cancelled or completed while this suspending function is waiting, this function
  * *closes the underlying channel* and immediately resumes with [CancellationException].
  */
 suspend fun AsynchronousFileChannel.aRead(
@@ -68,7 +68,7 @@ suspend fun AsynchronousFileChannel.aRead(
 /**
  * Performs [AsynchronousFileChannel.write] without blocking a thread and resumes when asynchronous operation completes.
  * This suspending function is cancellable.
- * If the [Job] of the current coroutine is completed while this suspending function is suspended, this function
+ * If the [Job] of the current coroutine is cancelled or completed while this suspending function is waiting, this function
  * *closes the underlying channel* and immediately resumes with [CancellationException].
  */
 suspend fun AsynchronousFileChannel.aWrite(
@@ -82,7 +82,7 @@ suspend fun AsynchronousFileChannel.aWrite(
 /**
  * Performs [AsynchronousServerSocketChannel.accept] without blocking a thread and resumes when asynchronous operation completes.
  * This suspending function is cancellable.
- * If the [Job] of the current coroutine is completed while this suspending function is suspended, this function
+ * If the [Job] of the current coroutine is cancelled or completed while this suspending function is waiting, this function
  * *closes the underlying channel* and immediately resumes with [CancellationException].
  */
 suspend fun AsynchronousServerSocketChannel.aAccept() = suspendCancellableCoroutine<AsynchronousSocketChannel> { cont ->
@@ -93,7 +93,7 @@ suspend fun AsynchronousServerSocketChannel.aAccept() = suspendCancellableCorout
 /**
  * Performs [AsynchronousSocketChannel.connect] without blocking a thread and resumes when asynchronous operation completes.
  * This suspending function is cancellable.
- * If the [Job] of the current coroutine is completed while this suspending function is suspended, this function
+ * If the [Job] of the current coroutine is cancelled or completed while this suspending function is waiting, this function
  * *closes the underlying channel* and immediately resumes with [CancellationException].
  */
 suspend fun AsynchronousSocketChannel.aConnect(
@@ -106,7 +106,7 @@ suspend fun AsynchronousSocketChannel.aConnect(
 /**
  * Performs [AsynchronousSocketChannel.read] without blocking a thread and resumes when asynchronous operation completes.
  * This suspending function is cancellable.
- * If the [Job] of the current coroutine is completed while this suspending function is suspended, this function
+ * If the [Job] of the current coroutine is cancelled or completed while this suspending function is waiting, this function
  * *closes the underlying channel* and immediately resumes with [CancellationException].
  */
 suspend fun AsynchronousSocketChannel.aRead(
@@ -121,7 +121,7 @@ suspend fun AsynchronousSocketChannel.aRead(
 /**
  * Performs [AsynchronousSocketChannel.write] without blocking a thread and resumes when asynchronous operation completes.
  * This suspending function is cancellable.
- * If the [Job] of the current coroutine is completed while this suspending function is suspended, this function
+ * If the [Job] of the current coroutine is cancelled or completed while this suspending function is waiting, this function
  * *closes the underlying channel* and immediately resumes with [CancellationException].
  */
 suspend fun AsynchronousSocketChannel.aWrite(

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

@@ -55,7 +55,7 @@ internal abstract class AbstractContinuation<in T>(
     }
 
     protected fun trySuspend(): Boolean {
-        while (true) { // lock-free loop
+        while (true) { // lock-free loop on decision
             val decision = this.decision // volatile read
             when (decision) {
                 UNDECIDED -> if (DECISION.compareAndSet(this, UNDECIDED, SUSPENDED)) return true
@@ -66,7 +66,7 @@ internal abstract class AbstractContinuation<in T>(
     }
 
     protected fun tryResume(): Boolean {
-        while (true) { // lock-free loop
+        while (true) { // lock-free loop on decision
             val decision = this.decision // volatile read
             when (decision) {
                 UNDECIDED -> if (DECISION.compareAndSet(this, UNDECIDED, RESUMED)) return true
@@ -79,8 +79,7 @@ internal abstract class AbstractContinuation<in T>(
     override fun resume(value: T) = resumeImpl(value, resumeMode)
 
     protected fun resumeImpl(value: T, resumeMode: Int) {
-        while (true) { // lock-free loop on state
-            val state = this.state // atomic read
+        lockFreeLoopOnState { state ->
             when (state) {
                 is Incomplete -> if (updateState(state, value, resumeMode)) return
                 is Cancelled -> return // ignore resumes on cancelled continuation
@@ -92,8 +91,7 @@ internal abstract class AbstractContinuation<in T>(
     override fun resumeWithException(exception: Throwable) = resumeWithExceptionImpl(exception, resumeMode)
 
     protected fun resumeWithExceptionImpl(exception: Throwable, resumeMode: Int) {
-        while (true) { // lock-free loop on state
-            val state = this.state // atomic read
+        lockFreeLoopOnState { state ->
             when (state) {
                 is Incomplete -> {
                     if (updateState(state, CompletedExceptionally(exception), resumeMode)) return
@@ -108,7 +106,7 @@ internal abstract class AbstractContinuation<in T>(
         }
     }
 
-    override fun handleCompletionException(closeException: Throwable) {
-        handleCoroutineException(context, closeException)
+    override fun handleException(exception: Throwable) {
+        handleCoroutineException(context, exception)
     }
 }

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

@@ -0,0 +1,68 @@
+/*
+ * 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
+import kotlin.coroutines.experimental.CoroutineContext
+
+/**
+ * Abstract class to simplify writing of coroutine completion objects that
+ * implement completion [Continuation], [Job], and [CoroutineScope] interfaces.
+ * It stores the result of continuation in the state of the job.
+ *
+ * @param active when `true` coroutine is created in _active_ state, when `false` in _new_ state. See [Job] for details.
+ * @suppress **This is unstable API and it is subject to change.**
+ */
+public abstract class AbstractCoroutine<in T>(
+    private val parentContext: CoroutineContext,
+    active: Boolean
+) : JobSupport(active), Continuation<T>, CoroutineScope {
+    @Suppress("LeakingThis")
+    public final override val context: CoroutineContext = parentContext + this
+
+    final override val hasCancellingState: Boolean get() = true
+
+    final override fun resume(value: T) {
+        lockFreeLoopOnState { state ->
+            when (state) {
+                is Incomplete -> if (updateState(state, value, MODE_ATOMIC_DEFAULT)) return
+                is Cancelled -> return // ignore resumes on cancelled continuation
+                else -> error("Already resumed, but got value $value")
+            }
+        }
+    }
+
+    final override fun resumeWithException(exception: Throwable) {
+        lockFreeLoopOnState { state ->
+            when (state) {
+                is Incomplete -> {
+                    if (updateState(state, CompletedExceptionally(exception), MODE_ATOMIC_DEFAULT)) return
+                }
+                is Cancelled -> {
+                    // ignore resumes on cancelled continuation, but handle exception if a different one is here
+                    if (exception != state.exception) handleCoroutineException(context, exception)
+                    return
+                }
+                else -> throw IllegalStateException("Already resumed, but got exception $exception", exception)
+            }
+        }
+    }
+
+    final override fun handleException(exception: Throwable) {
+        handleCoroutineException(parentContext, exception)
+    }
+}

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

@@ -27,10 +27,10 @@ import kotlin.coroutines.experimental.suspendCoroutine
 
 /**
  * Cancellable continuation. Its job is _completed_ when it is resumed or cancelled.
- * When [cancel] function is explicitly invoked, this continuation resumes with [CancellationException] or
+ * When [cancel] function is explicitly invoked, this continuation immediately resumes with [CancellationException] or
  * with the specified cancel cause.
  *
- * Cancellable continuation has three states:
+ * Cancellable continuation has three states (as subset of [Job] states):
  *
  * | **State**                           | [isActive] | [isCompleted] | [isCancelled] |
  * | ----------------------------------- | ---------- | ------------- | ------------- |
@@ -41,6 +41,9 @@ 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.
+ *
  * Invocation of [resume] or [resumeWithException] in _resumed_ state produces [IllegalStateException]
  * but is ignored in _cancelled_ state.
  */
@@ -50,7 +53,7 @@ public interface CancellableContinuation<in T> : Continuation<T>, Job {
      *
      * It implies that [isActive] is `false` and [isCompleted] is `true`.
      */
-    public val isCancelled: Boolean
+    public override val isCancelled: Boolean
 
     /**
      * Tries to resume this continuation with a given value and returns non-null object token if it was successful,
@@ -111,7 +114,7 @@ public interface CancellableContinuation<in T> : Continuation<T>, Job {
 
 /**
  * Suspends coroutine similar to [suspendCoroutine], but provide an implementation of [CancellableContinuation] to
- * the [block]. This function throws [CancellationException] if the coroutine is cancelled while suspended.
+ * the [block]. This function throws [CancellationException] if the coroutine is cancelled or completed while suspended.
  *
  * If [holdCancellability] optional parameter is `true`, then the coroutine is suspended, but it is not
  * cancellable until [CancellableContinuation.initCancellability] is invoked.

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

@@ -38,8 +38,7 @@ public class CompletableDeferred<T> : JobSupport(true), Deferred<T> {
      * Repeated invocations of this function have no effect and always produce `false`.
      */
     public fun complete(value: T): Boolean {
-        while (true) { // lock-free loop on state
-            val state = this.state // atomic read
+        lockFreeLoopOnState { state ->
             when (state) {
                 is Incomplete -> {
                     // actually, we don't care about the mode here at all, so just use a default
@@ -58,8 +57,7 @@ public class CompletableDeferred<T> : JobSupport(true), Deferred<T> {
      * Repeated invocations of this function have no effect and always produce `false`.
      */
     public fun completeExceptionally(exception: Throwable): Boolean {
-        while (true) { // lock-free loop on state
-            val state = this.state // atomic read
+        lockFreeLoopOnState { state ->
             when (state) {
                 is Incomplete -> {
                     // actually, we don't care about the mode here at all, so just use a default

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

@@ -113,7 +113,7 @@ internal class DispatchTask<in T>(
     override fun run() {
         val job = if (cancellable) dispatched.context[Job] else null
         when {
-            job != null && job.isCompleted ->
+            job != null && job.isCancelledOrCompleted ->
                 dispatched.resumeUndispatchedWithException(job.getCompletionException())
             exception -> dispatched.resumeUndispatchedWithException(value as Throwable)
             else -> dispatched.resumeUndispatched(value as T)
@@ -181,7 +181,7 @@ internal class DispatchedContinuation<in T>(
         val context = continuation.context
         dispatcher.dispatch(context, Runnable {
             withCoroutineContext(context) {
-                if (job != null && job.isCompleted)
+                if (job != null && job.isCancelledOrCompleted)
                     continuation.resumeWithException(job.getCompletionException())
                 else
                     continuation.resume(value)

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

@@ -16,7 +16,6 @@
 
 package kotlinx.coroutines.experimental
 
-import kotlin.coroutines.experimental.Continuation
 import kotlin.coroutines.experimental.CoroutineContext
 
 /**
@@ -42,53 +41,4 @@ public interface CoroutineScope {
      * Returns the context of this coroutine.
      */
     public val context: CoroutineContext
-}
-
-/**
- * Abstract class to simplify writing of coroutine completion objects that
- * implement completion [Continuation], [Job], and [CoroutineScope] interfaces.
- * It stores the result of continuation in the state of the job.
- *
- * @param active when `true` coroutine is created in _active_ state, when `false` in _new_ state. See [Job] for details.
- * @suppress **This is unstable API and it is subject to change.**
- */
-public abstract class AbstractCoroutine<in T>(
-    private val parentContext: CoroutineContext,
-    active: Boolean
-) : JobSupport(active), Continuation<T>, CoroutineScope {
-    @Suppress("LeakingThis")
-    public final override val context: CoroutineContext = parentContext + this
-
-    final override fun resume(value: T) {
-        while (true) { // lock-free loop on state
-            val state = this.state // atomic read
-            when (state) {
-                is Incomplete -> if (updateState(state, value, MODE_ATOMIC_DEFAULT)) return
-                is Cancelled -> return // ignore resumes on cancelled continuation
-                else -> error("Already resumed, but got value $value")
-            }
-        }
-    }
-
-    final override fun resumeWithException(exception: Throwable) {
-        while (true) { // lock-free loop on state
-            val state = this.state // atomic read
-            when (state) {
-                is Incomplete -> {
-                    if (updateState(state, CompletedExceptionally(exception), MODE_ATOMIC_DEFAULT)) return
-                }
-                is Cancelled -> {
-                    // ignore resumes on cancelled continuation, but handle exception if a different one is here
-                    if (exception != state.exception) handleCoroutineException(context, exception)
-                    return
-                }
-                else -> throw IllegalStateException("Already resumed, but got exception $exception", exception)
-            }
-        }
-    }
-
-    final override fun handleCompletionException(closeException: Throwable) {
-        handleCoroutineException(parentContext, closeException)
-    }
-}
-
+}