Force implementors of AbstractCoroutine deliberately choose whether parent-child relationship should be established

    * Test all coroutine builders
    * Fix FlowSubscription

Author: qwwdfsad

integration/kotlinx-coroutines-guava/src/ListenableFuture.kt

@@ -299,7 +299,7 @@ private class ToContinuation<T>(
  */
 private class ListenableFutureCoroutine<T>(
     context: CoroutineContext
-) : AbstractCoroutine<T>(context) {
+) : AbstractCoroutine<T>(context, initParentJob = true, active = true) {
 
     // JobListenableFuture propagates external cancellation to `this` coroutine. See JobListenableFuture.
     @JvmField

integration/kotlinx-coroutines-guava/test/ListenableFutureTest.kt

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ * Copyright 2016-2021 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
  */
 
 package kotlinx.coroutines.guava
@@ -747,4 +747,12 @@ class ListenableFutureTest : TestBase() {
         latch.countDown()
         return future
     }
+
+    @Test
+    fun testCancelledParent() = runTest({ it is CancellationException }) {
+        cancel()
+        future { expectUnreached() }
+        future(start = CoroutineStart.ATOMIC) { }
+        future(start = CoroutineStart.UNDISPATCHED) { }
+    }
 }

integration/kotlinx-coroutines-jdk8/src/future/Future.kt

@@ -48,7 +48,7 @@ public fun <T> CoroutineScope.future(
 private class CompletableFutureCoroutine<T>(
     context: CoroutineContext,
     private val future: CompletableFuture<T>
-) : AbstractCoroutine<T>(context), BiConsumer<T?, Throwable?> {
+) : AbstractCoroutine<T>(context, initParentJob = true, active = true), BiConsumer<T?, Throwable?> {
     override fun accept(value: T?, exception: Throwable?) {
         cancel()
     }

integration/kotlinx-coroutines-jdk8/test/future/FutureTest.kt

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ * Copyright 2016-2021 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
  */
 
 package kotlinx.coroutines.future
@@ -567,4 +567,12 @@ class FutureTest : TestBase() {
         assertFailsWith<CancellationException> { stage.await() }
         finish(4)
     }
+
+    @Test
+    fun testCancelledParent() = runTest({ it is java.util.concurrent.CancellationException }) {
+        cancel()
+        future { expectUnreached() }
+        future(start = CoroutineStart.ATOMIC) { }
+        future(start = CoroutineStart.UNDISPATCHED) { }
+    }
 }

kotlinx-coroutines-core/api/kotlinx-coroutines-core.api

@@ -1,6 +1,5 @@
 public abstract class kotlinx/coroutines/AbstractCoroutine : kotlinx/coroutines/JobSupport, kotlin/coroutines/Continuation, kotlinx/coroutines/CoroutineScope, kotlinx/coroutines/Job {
-	public fun <init> (Lkotlin/coroutines/CoroutineContext;Z)V
-	public synthetic fun <init> (Lkotlin/coroutines/CoroutineContext;ZILkotlin/jvm/internal/DefaultConstructorMarker;)V
+	public fun <init> (Lkotlin/coroutines/CoroutineContext;ZZ)V
 	protected fun afterResume (Ljava/lang/Object;)V
 	protected fun cancellationExceptionMessage ()Ljava/lang/String;
 	public final fun getContext ()Lkotlin/coroutines/CoroutineContext;
@@ -418,6 +417,7 @@ public class kotlinx/coroutines/JobSupport : kotlinx/coroutines/ChildJob, kotlin
 	public final fun getKey ()Lkotlin/coroutines/CoroutineContext$Key;
 	public final fun getOnJoin ()Lkotlinx/coroutines/selects/SelectClause0;
 	protected fun handleJobException (Ljava/lang/Throwable;)Z
+	protected final fun initParentJob (Lkotlinx/coroutines/Job;)V
 	public final fun invokeOnCompletion (Lkotlin/jvm/functions/Function1;)Lkotlinx/coroutines/DisposableHandle;
 	public final fun invokeOnCompletion (ZZLkotlin/jvm/functions/Function1;)Lkotlinx/coroutines/DisposableHandle;
 	public fun isActive ()Z

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

@@ -25,33 +25,38 @@ import kotlin.coroutines.*
  * * [onCancelled] in invoked when the coroutine completes with an exception (cancelled).
  *
  * @param parentContext the context of the parent coroutine.
+ * @param initParentJob specifies whether the parent-child relationship should be instantiated directly
+ *               in `AbstractCoroutine` constructor. If set to `false`, it's the responsibility of the child class
+ *               to invoke [initParentJob] manually.
  * @param active when `true` (by default), the coroutine is created in the _active_ state, otherwise it is created in the _new_ state.
  *               See [Job] for details.
  *
  * @suppress **This an internal API and should not be used from general code.**
  */
 @InternalCoroutinesApi
 public abstract class AbstractCoroutine<in T>(
-    /**
-     * The context of the parent coroutine.
-     */
     parentContext: CoroutineContext,
-    active: Boolean = true
+    initParentJob: Boolean,
+    active: Boolean
 ) : JobSupport(active), Job, Continuation<T>, CoroutineScope {
+
+    init {
+        /*
+         * Setup parent-child relationship between the parent in the context and the current coroutine.
+         * It may cause this coroutine to become _cancelling_ if the parent is already cancelled.
+         * It is dangerous to install parent-child relationship here if the coroutine class
+         * operates its state from within onCancelled or onCancelling
+         * (with exceptions for rx integrations that can't have any parent)
+         */
+        if (initParentJob) initParentJob(parentContext[Job])
+    }
+
     /**
      * The context of this coroutine that includes this coroutine as a [Job].
      */
     @Suppress("LeakingThis")
     public final override val context: CoroutineContext = parentContext + this
 
-    init {
-        /*
-         * Setup parent-child relationship between the parent in the context
-         * and the current coroutine.
-         * It may cause this coroutine to become cancelled if the parent is already cancelled
-         */
-        initParentJobInternal(parentContext[Job])
-    }
     /**
      * The context of this scope which is the same as the [context] of this coroutine.
      */

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

@@ -96,7 +96,7 @@ public fun <T> CoroutineScope.async(
 private open class DeferredCoroutine<T>(
     parentContext: CoroutineContext,
     active: Boolean
-) : AbstractCoroutine<T>(parentContext, active), Deferred<T>, SelectClause1<T> {
+) : AbstractCoroutine<T>(parentContext, true, active = active), Deferred<T>, SelectClause1<T> {
     override fun getCompleted(): T = getCompletedInternal() as T
     override suspend fun await(): T = awaitInternal() as T
     override val onAwait: SelectClause1<T> get() = this
@@ -187,7 +187,7 @@ public suspend inline operator fun <T> CoroutineDispatcher.invoke(
 private open class StandaloneCoroutine(
     parentContext: CoroutineContext,
     active: Boolean
-) : AbstractCoroutine<Unit>(parentContext, active) {
+) : AbstractCoroutine<Unit>(parentContext, initParentJob = true, active = active) {
     override fun handleJobException(exception: Throwable): Boolean {
         handleCoroutineException(context, exception)
         return true

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

@@ -80,7 +80,7 @@ public fun <T> CompletableDeferred(value: T): CompletableDeferred<T> = Completab
 private class CompletableDeferredImpl<T>(
     parent: Job?
 ) : JobSupport(true), CompletableDeferred<T>, SelectClause1<T> {
-    init { initParentJobInternal(parent) }
+    init { initParentJob(parent) }
     override val onCancelComplete get() = true
     override fun getCompleted(): T = getCompletedInternal() as T
     override suspend fun await(): T = awaitInternal() as T

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

@@ -139,7 +139,7 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
      * Initializes parent job.
      * It shall be invoked at most once after construction after all other initialization.
      */
-    internal fun initParentJobInternal(parent: Job?) {
+    protected fun initParentJob(parent: Job?) {
         assert { parentHandle == null }
         if (parent == null) {
             parentHandle = NonDisposableHandle
@@ -1311,7 +1311,7 @@ private class Empty(override val isActive: Boolean) : Incomplete {
 }
 
 internal open class JobImpl(parent: Job?) : JobSupport(true), CompletableJob {
-    init { initParentJobInternal(parent) }
+    init { initParentJob(parent) }
     override val onCancelComplete get() = true
     /*
      * Check whether parent is able to handle exceptions as well.

kotlinx-coroutines-core/common/src/channels/Broadcast.kt

@@ -127,7 +127,13 @@ private open class BroadcastCoroutine<E>(
     parentContext: CoroutineContext,
     protected val _channel: BroadcastChannel<E>,
     active: Boolean
-) : AbstractCoroutine<Unit>(parentContext, active), ProducerScope<E>, BroadcastChannel<E> by _channel {
+) : AbstractCoroutine<Unit>(parentContext, initParentJob = false, active = active),
+    ProducerScope<E>, BroadcastChannel<E> by _channel {
+
+    init {
+        initParentJob(parentContext[Job])
+    }
+
     override val isActive: Boolean get() = super.isActive
 
     override val channel: SendChannel<E>

kotlinx-coroutines-core/common/src/channels/ChannelCoroutine.kt

@@ -11,8 +11,10 @@ import kotlin.coroutines.*
 internal open class ChannelCoroutine<E>(
     parentContext: CoroutineContext,
     protected val _channel: Channel<E>,
+    initParentJob: Boolean,
     active: Boolean
-) : AbstractCoroutine<Unit>(parentContext, active), Channel<E> by _channel {
+) : AbstractCoroutine<Unit>(parentContext, initParentJob, active), Channel<E> by _channel {
+
     val channel: Channel<E> get() = this
 
     override fun cancel() {

kotlinx-coroutines-core/common/src/channels/Produce.kt

@@ -139,7 +139,7 @@ internal fun <E> CoroutineScope.produce(
 
 internal open class ProducerCoroutine<E>(
     parentContext: CoroutineContext, channel: Channel<E>
-) : ChannelCoroutine<E>(parentContext, channel, active = true), ProducerScope<E> {
+) : ChannelCoroutine<E>(parentContext, channel, true, active = true), ProducerScope<E> {
     override val isActive: Boolean
         get() = super.isActive
 

kotlinx-coroutines-core/common/src/internal/Scopes.kt

@@ -15,7 +15,7 @@ import kotlin.jvm.*
 internal open class ScopeCoroutine<in T>(
     context: CoroutineContext,
     @JvmField val uCont: Continuation<T> // unintercepted continuation
-) : AbstractCoroutine<T>(context, true), CoroutineStackFrame {
+) : AbstractCoroutine<T>(context, true, true), CoroutineStackFrame {
 
     final override val callerFrame: CoroutineStackFrame? get() = uCont as? CoroutineStackFrame
     final override fun getStackTraceElement(): StackTraceElement? = null

kotlinx-coroutines-core/common/test/AbstractCoroutineTest.kt

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ * Copyright 2016-2021 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
  */
 
 package kotlinx.coroutines
@@ -13,7 +13,7 @@ class AbstractCoroutineTest : TestBase() {
     fun testNotifications() = runTest {
         expect(1)
         val coroutineContext = coroutineContext // workaround for KT-22984
-        val coroutine = object : AbstractCoroutine<String>(coroutineContext, false) {
+        val coroutine = object : AbstractCoroutine<String>(coroutineContext, true, false) {
             override fun onStart() {
                 expect(3)
             }
@@ -53,7 +53,7 @@ class AbstractCoroutineTest : TestBase() {
     fun testNotificationsWithException() = runTest {
         expect(1)
         val coroutineContext = coroutineContext // workaround for KT-22984
-        val coroutine = object : AbstractCoroutine<String>(coroutineContext + NonCancellable, false) {
+        val coroutine = object : AbstractCoroutine<String>(coroutineContext + NonCancellable, true, false) {
             override fun onStart() {
                 expect(3)
             }
@@ -91,4 +91,4 @@ class AbstractCoroutineTest : TestBase() {
         coroutine.resumeWithException(TestException2())
         finish(10)
     }
-}
+}

kotlinx-coroutines-core/common/test/CancelledParentAttachTest.kt

@@ -0,0 +1,85 @@
+/*
+ * Copyright 2016-2021 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package kotlinx.coroutines
+
+import kotlinx.coroutines.channels.*
+import kotlinx.coroutines.flow.internal.*
+import kotlin.test.*
+
+class CancelledParentAttachTest : TestBase() {
+
+    @Test
+    fun testAsync() = CoroutineStart.values().forEach(::testAsyncCancelledParent)
+
+    private fun testAsyncCancelledParent(start: CoroutineStart) =
+        runTest({ it is CancellationException }) {
+            cancel()
+            expect(1)
+            val d = async<Int>(start = start) { 42 }
+            expect(2)
+            d.invokeOnCompletion {
+                finish(3)
+                reset()
+            }
+        }
+
+    @Test
+    fun testLaunch() = CoroutineStart.values().forEach(::testLaunchCancelledParent)
+
+    private fun testLaunchCancelledParent(start: CoroutineStart) =
+        runTest({ it is CancellationException }) {
+            cancel()
+            expect(1)
+            val d = launch(start = start) { }
+            expect(2)
+            d.invokeOnCompletion {
+                finish(3)
+                reset()
+            }
+        }
+
+    @Test
+    fun testProduce() =
+        runTest({ it is CancellationException }) {
+            cancel()
+            expect(1)
+            val d = produce<Int> { }
+            expect(2)
+            (d as Job).invokeOnCompletion {
+                finish(3)
+                reset()
+            }
+        }
+
+    @Test
+    fun testBroadcast() = CoroutineStart.values().forEach(::testBroadcastCancelledParent)
+
+    private fun testBroadcastCancelledParent(start: CoroutineStart) =
+        runTest({ it is CancellationException }) {
+            cancel()
+            expect(1)
+            val bc = broadcast<Int>(start = start) {}
+            expect(2)
+            (bc as Job).invokeOnCompletion {
+                finish(3)
+                reset()
+            }
+        }
+
+    @Test
+    fun testScopes() {
+        testScope { coroutineScope {  } }
+        testScope { supervisorScope {  } }
+        testScope { flowScope {  } }
+        testScope { withTimeout(Long.MAX_VALUE) {  } }
+        testScope { withContext(Job()) {  } }
+        testScope { withContext(CoroutineName("")) {  } }
+    }
+
+    private inline fun testScope(crossinline block: suspend () -> Unit) = runTest({ it is CancellationException }) {
+        cancel()
+        block()
+    }
+}

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

@@ -63,7 +63,8 @@ private class BlockingCoroutine<T>(
     parentContext: CoroutineContext,
     private val blockedThread: Thread,
     private val eventLoop: EventLoop?
-) : AbstractCoroutine<T>(parentContext, true) {
+) : AbstractCoroutine<T>(parentContext, true, true) {
+
     override val isScopedCoroutine: Boolean get() = true
 
     override fun afterCompletion(state: Any?) {

kotlinx-coroutines-core/jvm/src/channels/Actor.kt

@@ -127,7 +127,11 @@ private open class ActorCoroutine<E>(
     parentContext: CoroutineContext,
     channel: Channel<E>,
     active: Boolean
-) : ChannelCoroutine<E>(parentContext, channel, active), ActorScope<E> {
+) : ChannelCoroutine<E>(parentContext, channel, initParentJob = false, active = active), ActorScope<E> {
+
+    init {
+        initParentJob(parentContext[Job])
+    }
 
     override fun onCancelling(cause: Throwable?) {
         _channel.cancel(cause?.let {

kotlinx-coroutines-core/jvm/test/JoinStrTest.kt renamed to kotlinx-coroutines-core/jvm/test/JoinStressTest.kt

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ * Copyright 2016-2021 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
  */
 
 package kotlinx.coroutines

kotlinx-coroutines-core/jvm/test/RunBlockingTest.kt

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ * Copyright 2016-2021 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
  */
 
 package kotlinx.coroutines
@@ -171,4 +171,15 @@ class RunBlockingTest : TestBase() {
         }
         rb.hashCode() // unused
     }
+
+    @Test
+    fun testCancelledParent() {
+        val job = Job()
+        job.cancel()
+        assertFailsWith<CancellationException> {
+            runBlocking(job) {
+                expectUnreached()
+            }
+        }
+    }
 }