Ignore exception on cancel/fail race in CancellableContinuation

There seems to be no other way to satisfactory fix the problem of
a race between cancellation on CancellationContinuation and
concurrent failure of the corresponding background process in a
way the distinguishes "failure from cancellation attempt" from
a "true failure" without placing undue burden on anyone who is
implementing `await()` extension function for various future types.

- Added testTimeoutCancellationFailRace that works as a perfect litmus
  test; being allowed to run for a while it reliably detects various
  problems in alternative approaches.
- Optimized CancellableContinuationImpl; merged its code with
  AbstractContinuation class that is not needed as a separate
  class and removed.
- Deprecated and hidden CancellableContinuation.initCancellability();
  it is now always invoked after the end of the block that was
  passed to suspendCancellableCoroutine.
- Deprecated `holdCancellability` parameter to an internal
  suspendAtomicCancellableCoroutine function.

Fixes #892
Fixes #830

Author: elizarov

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

@@ -34,7 +34,7 @@ public final class kotlinx/coroutines/BuildersKt {
 public abstract interface class kotlinx/coroutines/CancellableContinuation : kotlin/coroutines/Continuation {
 	public abstract fun cancel (Ljava/lang/Throwable;)Z
 	public abstract fun completeResume (Ljava/lang/Object;)V
-	public abstract fun initCancellability ()V
+	public abstract synthetic fun initCancellability ()V
 	public abstract fun invokeOnCancellation (Lkotlin/jvm/functions/Function1;)V
 	public abstract fun isActive ()Z
 	public abstract fun isCancelled ()Z
@@ -50,17 +50,28 @@ public final class kotlinx/coroutines/CancellableContinuation$DefaultImpls {
 	public static synthetic fun tryResume$default (Lkotlinx/coroutines/CancellableContinuation;Ljava/lang/Object;Ljava/lang/Object;ILjava/lang/Object;)Ljava/lang/Object;
 }
 
-public class kotlinx/coroutines/CancellableContinuationImpl : java/lang/Runnable, kotlin/coroutines/jvm/internal/CoroutineStackFrame, kotlinx/coroutines/CancellableContinuation {
+public class kotlinx/coroutines/CancellableContinuationImpl : kotlin/coroutines/jvm/internal/CoroutineStackFrame, kotlinx/coroutines/CancellableContinuation {
 	public fun <init> (Lkotlin/coroutines/Continuation;I)V
+	public fun cancel (Ljava/lang/Throwable;)Z
 	public fun completeResume (Ljava/lang/Object;)V
 	public fun getCallerFrame ()Lkotlin/coroutines/jvm/internal/CoroutineStackFrame;
 	public fun getContext ()Lkotlin/coroutines/CoroutineContext;
+	public fun getContinuationCancellationCause (Lkotlinx/coroutines/Job;)Ljava/lang/Throwable;
+	public final fun getDelegate ()Lkotlin/coroutines/Continuation;
+	public final fun getResult ()Ljava/lang/Object;
 	public fun getStackTraceElement ()Ljava/lang/StackTraceElement;
 	public fun getSuccessfulResult (Ljava/lang/Object;)Ljava/lang/Object;
-	public fun initCancellability ()V
+	public synthetic fun initCancellability ()V
+	public fun invokeOnCancellation (Lkotlin/jvm/functions/Function1;)V
+	public fun isActive ()Z
+	public fun isCancelled ()Z
+	public fun isCompleted ()Z
 	protected fun nameString ()Ljava/lang/String;
 	public fun resumeUndispatched (Lkotlinx/coroutines/CoroutineDispatcher;Ljava/lang/Object;)V
 	public fun resumeUndispatchedWithException (Lkotlinx/coroutines/CoroutineDispatcher;Ljava/lang/Throwable;)V
+	public fun resumeWith (Ljava/lang/Object;)V
+	public fun takeState ()Ljava/lang/Object;
+	public fun toString ()Ljava/lang/String;
 	public fun tryResume (Ljava/lang/Object;Ljava/lang/Object;)Ljava/lang/Object;
 	public fun tryResumeWithException (Ljava/lang/Throwable;)Ljava/lang/Object;
 }

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

@@ -7,7 +7,6 @@ package kotlinx.coroutines
 import kotlinx.coroutines.internal.*
 import kotlin.coroutines.*
 import kotlin.coroutines.intrinsics.*
-import kotlin.jvm.*
 
 // --------------- cancellable continuations ---------------
 
@@ -98,9 +97,17 @@ public interface CancellableContinuation<in T> : Continuation<T> {
     public fun completeResume(token: Any)
 
     /**
-     * Makes this continuation cancellable. Use it with `holdCancellability` optional parameter to
-     * [suspendCancellableCoroutine] function. It throws [IllegalStateException] if invoked more than once.
+     * Legacy function that turned on cancellation behavior in [suspendCancellableCoroutine] before kotlinx.coroutines 1.1.0.
+     * This function does nothing and is left only for binary compatibility with old compiled code.
+     *
+     * @suppress **Deprecated**: This function is no longer used.
+     *   It is left for binary compatibility with code compiled before kotlinx.coroutines 1.1.0.
      */
+    @Deprecated(
+        level = DeprecationLevel.HIDDEN,
+        message = "This function is no longer used. " +
+            "It is left for binary compatibility with code compiled before kotlinx.coroutines 1.1.0. "
+    )
     @InternalCoroutinesApi
     public fun initCancellability()
 
@@ -155,7 +162,9 @@ public suspend inline fun <T> suspendCancellableCoroutine(
 ): T =
     suspendCoroutineUninterceptedOrReturn { uCont ->
         val cancellable = CancellableContinuationImpl(uCont.intercepted(), resumeMode = MODE_CANCELLABLE)
-        cancellable.initCancellability()
+        // NOTE: Before version 1.1.0 the following invocation was inlined here, so invocation of this
+        // method indicates that the code was compiled by kotlinx.coroutines < 1.1.0
+        // cancellable.initCancellability()
         block(cancellable)
         cancellable.getResult()
     }
@@ -172,16 +181,28 @@ public suspend inline fun <T> suspendCancellableCoroutine(
  */
 @InternalCoroutinesApi
 public suspend inline fun <T> suspendAtomicCancellableCoroutine(
-    holdCancellability: Boolean = false,
     crossinline block: (CancellableContinuation<T>) -> Unit
 ): T =
     suspendCoroutineUninterceptedOrReturn { uCont ->
         val cancellable = CancellableContinuationImpl(uCont.intercepted(), resumeMode = MODE_ATOMIC_DEFAULT)
-        if (!holdCancellability) cancellable.initCancellability()
         block(cancellable)
         cancellable.getResult()
     }
 
+/**
+ * @suppress **Deprecated**
+ */
+@Deprecated(
+    message = "holdCancellability parameter is deprecated and is no longer used",
+    replaceWith = ReplaceWith("suspendAtomicCancellableCoroutine(block)")
+)
+@InternalCoroutinesApi
+public suspend inline fun <T> suspendAtomicCancellableCoroutine(
+    holdCancellability: Boolean = false,
+    crossinline block: (CancellableContinuation<T>) -> Unit
+): T =
+    suspendAtomicCancellableCoroutine(block)
+
 /**
  * Removes a given node on cancellation.
  */
@@ -213,79 +234,3 @@ private class DisposeOnCancel(private val handle: DisposableHandle) : CancelHand
     override fun invoke(cause: Throwable?) = handle.dispose()
     override fun toString(): String = "DisposeOnCancel[$handle]"
 }
-
-@PublishedApi
-internal open class CancellableContinuationImpl<in T>(
-    delegate: Continuation<T>,
-    resumeMode: Int
-) : AbstractContinuation<T>(delegate, resumeMode), CancellableContinuation<T>, Runnable, CoroutineStackFrame {
-
-    public override val context: CoroutineContext = delegate.context
-
-    override val callerFrame: CoroutineStackFrame?
-        get() = delegate as? CoroutineStackFrame
-
-    override fun getStackTraceElement(): StackTraceElement? = null
-
-    override fun initCancellability() {
-        initParentJobInternal(delegate.context[Job])
-    }
-
-    override fun tryResume(value: T, idempotent: Any?): Any? {
-        loopOnState { state ->
-            when (state) {
-                is NotCompleted -> {
-                    val update: Any? = if (idempotent == null) value else
-                        CompletedIdempotentResult(idempotent, value, state)
-                    if (tryUpdateStateToFinal(state, update)) return state
-                }
-                is CompletedIdempotentResult -> {
-                    if (state.idempotentResume === idempotent) {
-                        check(state.result === value) { "Non-idempotent resume" }
-                        return state.token
-                    } else
-                        return null
-                }
-                else -> return null // cannot resume -- not active anymore
-            }
-        }
-    }
-
-    override fun tryResumeWithException(exception: Throwable): Any? {
-        loopOnState { state ->
-            when (state) {
-                is NotCompleted -> {
-                    if (tryUpdateStateToFinal(state, CompletedExceptionally(exception))) return state
-                }
-                else -> return null // cannot resume -- not active anymore
-            }
-        }
-    }
-
-    override fun completeResume(token: Any) = completeStateUpdate(token as NotCompleted, state, resumeMode)
-
-    override fun CoroutineDispatcher.resumeUndispatched(value: T) {
-        val dc = delegate as? DispatchedContinuation
-        resumeImpl(value, if (dc?.dispatcher === this) MODE_UNDISPATCHED else resumeMode)
-    }
-
-    override fun CoroutineDispatcher.resumeUndispatchedWithException(exception: Throwable) {
-        val dc = delegate as? DispatchedContinuation
-        resumeImpl(CompletedExceptionally(exception), if (dc?.dispatcher === this) MODE_UNDISPATCHED else resumeMode)
-    }
-
-    @Suppress("UNCHECKED_CAST")
-    override fun <T> getSuccessfulResult(state: Any?): T =
-        if (state is CompletedIdempotentResult) state.result as T else state as T
-
-    protected override fun nameString(): String =
-        "CancellableContinuation(${delegate.toDebugString()})"
-}
-
-private class CompletedIdempotentResult(
-    @JvmField val idempotentResume: Any?,
-    @JvmField val result: Any?,
-    @JvmField val token: NotCompleted
-) {
-    override fun toString(): String = "CompletedIdempotentResult[$result]"
-}