Introducing flowScope, builder necessary for creating cancellation-transparent flow operators

Author: qwwdfsad

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

@@ -377,11 +377,11 @@ public class kotlinx/coroutines/JobSupport : kotlinx/coroutines/ChildJob, kotlin
 	public fun cancel (Ljava/util/concurrent/CancellationException;)V
 	public final fun cancelCoroutine (Ljava/lang/Throwable;)Z
 	public fun cancelInternal (Ljava/lang/Throwable;)Z
+	public fun cancelOnChildCancellation (Ljava/util/concurrent/CancellationException;)Z
 	public fun childCancelled (Ljava/lang/Throwable;)Z
 	public fun fold (Ljava/lang/Object;Lkotlin/jvm/functions/Function2;)Ljava/lang/Object;
 	public fun get (Lkotlin/coroutines/CoroutineContext$Key;)Lkotlin/coroutines/CoroutineContext$Element;
 	public final fun getCancellationException ()Ljava/util/concurrent/CancellationException;
-	protected fun getCancelsParent ()Z
 	public fun getChildJobCancellationCause ()Ljava/util/concurrent/CancellationException;
 	public final fun getChildren ()Lkotlin/sequences/Sequence;
 	protected final fun getCompletionCause ()Ljava/lang/Throwable;
@@ -396,6 +396,7 @@ public class kotlinx/coroutines/JobSupport : kotlinx/coroutines/ChildJob, kotlin
 	public final fun isCancelled ()Z
 	public final fun isCompleted ()Z
 	public final fun isCompletedExceptionally ()Z
+	protected fun isScopedCoroutine ()Z
 	public final fun join (Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public fun minusKey (Lkotlin/coroutines/CoroutineContext$Key;)Lkotlin/coroutines/CoroutineContext;
 	protected fun onCancelling (Ljava/lang/Throwable;)V

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

@@ -319,6 +319,31 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
         cancelParent(cause) // tentative cancellation -- does not matter if there is no parent
     }
 
+    /**
+     * The method that is invoked when the job is cancelled to possible propagate cancellation to the parent.
+     * Returns `true` if the parent is responsible for handling the exception, `false` otherwise.
+     *
+     * Invariant: never returns `true` for instances of [CancellationException], otherwise such exception
+     * may leak to the [CoroutineExceptionHandler].
+     */
+    private fun cancelParent(cause: Throwable): Boolean {
+        /* CancellationException is considered "normal" and parent usually is not cancelled when child produces it.
+         * This allow parent to cancel its children (normally) without being cancelled itself, unless
+         * child crashes and produce some other exception during its completion.
+         */
+        val isCancellation = cause is CancellationException
+        val parent = parentHandle
+        // No parent -- ignore CE, report other exceptions.
+        if (parent === null || parent === NonDisposableHandle) {
+            return isCancellation
+        }
+
+        // Is scoped coroutine -- don't propagate, will be rethrown
+        if (isScopedCoroutine) return isCancellation
+        // Notify parent but don't forget to check cancellation
+        return parent.childCancelled(cause) || isCancellation
+    }
+
     private fun NodeList.notifyCompletion(cause: Throwable?) =
         notifyHandlers<JobNode<*>>(this, cause)
 
@@ -594,17 +619,28 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
         cancelImpl(parentJob)
     }
 
-    // Child was cancelled with cause
-    // It is overridden in supervisor implementations to ignore child cancellation
-    public open fun childCancelled(cause: Throwable): Boolean =
-        cancelImpl(cause) && handlesException
+    /**
+     * Returns `true` if job should cancel itself during on child [CancellationException].
+     */
+    public open fun cancelOnChildCancellation(cause: CancellationException) = false
+
+    /**
+     * Child was cancelled with a cause.
+     * At this point parent decides whether it cancels itself (e.g. on a critical failure) and
+     * whether it handles the exception of the child.
+     *
+     * It is overridden in supervisor implementations to completely ignore any child cancellation
+     */
+    public open fun childCancelled(cause: Throwable): Boolean {
+        if (cause is CancellationException && !cancelOnChildCancellation(cause)) return true
+        return cancelImpl(cause) && handlesException
+    }
 
     /**
      * Makes this [Job] cancelled with a specified [cause].
      * It is used in [AbstractCoroutine]-derived classes when there is an internal failure.
      */
-    public fun cancelCoroutine(cause: Throwable?) =
-        cancelImpl(cause)
+    public fun cancelCoroutine(cause: Throwable?) = cancelImpl(cause)
 
     // cause is Throwable or ParentJob when cancelChild was invoked
     // returns true is exception was handled, false otherwise
@@ -912,14 +948,15 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
     protected open fun onCancelling(cause: Throwable?) {}
 
     /**
-     * When this function returns `true` the parent is cancelled on cancellation of this job.
-     * Note that [CancellationException] is considered "normal" and parent is not cancelled when child produces it.
-     * This allows parent to cancel its children (normally) without being cancelled itself, unless
-     * child crashes and produce some other exception during its completion.
+     * Returns `true` for scoped coroutines.
+     * Scoped coroutine is a coroutine that is executed sequentially within the enclosing scope
+     * without any concurrency.
+     * Scoped coroutines always handle any exception happened within -- they just rethrow it
+     * to the enclosing scope.
      *
-     * @suppress **This is unstable API and it is subject to change.*
+     * Examples of scoped coroutines are `coroutineScope`, `withTimeout` and `runBlocking`.
      */
-    protected open val cancelsParent: Boolean get() = true
+    protected open val isScopedCoroutine: Boolean get() = false
 
     /**
      * Returns `true` for jobs that handle their exceptions or integrate them into the job's result via [onCompletionInternal].
@@ -939,20 +976,9 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
      *
      * This method is invoked **exactly once** when the final exception of the job is determined
      * and before it becomes complete. At the moment of invocation the job and all its children are complete.
-     *
-     * @suppress **This is unstable API and it is subject to change.*
      */
     protected open fun handleJobException(exception: Throwable): Boolean = false
 
-    private fun cancelParent(cause: Throwable): Boolean {
-        // CancellationException is considered "normal" and parent is not cancelled when child produces it.
-        // This allow parent to cancel its children (normally) without being cancelled itself, unless
-        // child crashes and produce some other exception during its completion.
-        if (cause is CancellationException) return true
-        if (!cancelsParent) return false
-        return parentHandle?.childCancelled(cause) == true
-    }
-
     /**
      * Override for completion actions that need to update some external object depending on job's state,
      * right before all the waiters for coroutine's completion are notified.

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

@@ -85,9 +85,7 @@ private open class TimeoutCoroutine<U, in T: U>(
     override val defaultResumeMode: Int get() = MODE_DIRECT
     override val callerFrame: CoroutineStackFrame? get() = (uCont as? CoroutineStackFrame)
     override fun getStackTraceElement(): StackTraceElement? = null
-
-    override val cancelsParent: Boolean
-        get() = false // it throws exception to parent instead of cancelling it
+    override val isScopedCoroutine: Boolean get() = true
 
     @Suppress("LeakingThis", "Deprecation")
     override fun run() {

kotlinx-coroutines-core/common/src/flow/internal/FlowScope.kt

@@ -0,0 +1,37 @@
+/*
+ * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package kotlinx.coroutines.flow.internal
+
+import kotlinx.coroutines.*
+import kotlinx.coroutines.internal.*
+import kotlinx.coroutines.intrinsics.*
+import kotlin.coroutines.*
+import kotlin.coroutines.intrinsics.*
+
+/**
+ * Creates a [CoroutineScope] and calls the specified suspend block with this scope.
+ * This builder is similar to [coroutineScope] with the only exception that it *ties* lifecycle of children
+ * and itself regarding the cancellation, thus being cancelled when one of the children becomes cancelled.
+ *
+ * For example:
+ * ```
+ * flowScope {
+ *     launch {
+ *         throw CancellationException()
+ *     }
+ * } // <- CE will be rethrown here
+ * ```
+ */
+internal suspend fun <R> flowScope(block: suspend CoroutineScope.() -> R): R =
+    suspendCoroutineUninterceptedOrReturn { uCont ->
+        val coroutine = FlowScope(uCont.context, uCont)
+        coroutine.startUndispatchedOrReturn(coroutine, block)
+    }
+
+private class FlowScope<T>(context: CoroutineContext, uCont: Continuation<T>) : ScopeCoroutine<T>(context, uCont) {
+    public override fun cancelOnChildCancellation(cause: CancellationException) = cause !is ChildCancelledException
+}
+
+internal class ChildCancelledException : CancellationException(null)

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

@@ -17,13 +17,12 @@ internal open class ScopeCoroutine<in T>(
 ) : AbstractCoroutine<T>(context, true), CoroutineStackFrame {
     final override val callerFrame: CoroutineStackFrame? get() = uCont as CoroutineStackFrame?
     final override fun getStackTraceElement(): StackTraceElement? = null
+    final override val isScopedCoroutine: Boolean get() = true
+
     override val defaultResumeMode: Int get() = MODE_DIRECT
 
     internal val parent: Job? get() = parentContext[Job]
 
-    override val cancelsParent: Boolean
-        get() = false // it throws exception to parent instead of cancelling it
-
     @Suppress("UNCHECKED_CAST")
     override fun afterCompletionInternal(state: Any?, mode: Int) {
         if (state is CompletedExceptionally) {

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

@@ -219,4 +219,22 @@ class SupervisorTest : TestBase() {
         yield() // to coroutineScope
         finish(7)
     }
+
+    @Test
+    fun testSupervisorJobCancellationException() = runTest {
+        val job = SupervisorJob()
+        val child = launch(job + CoroutineExceptionHandler { _, _ -> expectUnreached() }) {
+            expect(1)
+            hang {
+                expect(3)
+            }
+        }
+
+        yield()
+        expect(2)
+        child.cancelAndJoin()
+        job.complete()
+        job.join()
+        finish(4)
+    }
 }

kotlinx-coroutines-core/common/test/flow/internal/FlowScopeTest.kt

@@ -0,0 +1,77 @@
+/*
+ * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package kotlinx.coroutines.flow.internal
+
+import kotlinx.coroutines.*
+import kotlin.test.*
+
+class FlowScopeTest : TestBase() {
+
+    @Test
+    fun testCancellation() = runTest {
+        assertFailsWith<CancellationException> {
+            flowScope {
+                expect(1)
+                val child = launch {
+                    expect(3)
+                    hang { expect(5) }
+                }
+                expect(2)
+                yield()
+                expect(4)
+                child.cancel()
+            }
+        }
+        finish(6)
+    }
+
+    @Test
+    fun testCancellationWithChildCancelled() = runTest {
+        flowScope {
+            expect(1)
+            val child = launch {
+                expect(3)
+                hang { expect(5) }
+            }
+            expect(2)
+            yield()
+            expect(4)
+            child.cancel(ChildCancelledException())
+        }
+        finish(6)
+    }
+
+    @Test
+    fun testCancellationWithSuspensionPoint() = runTest {
+        assertFailsWith<CancellationException> {
+            flowScope {
+                expect(1)
+                val child = launch {
+                    expect(3)
+                    hang { expect(6) }
+                }
+                expect(2)
+                yield()
+                expect(4)
+                child.cancel()
+                hang { expect(5) }
+            }
+        }
+        finish(7)
+    }
+
+    @Test
+    fun testNestedScopes() = runTest {
+        assertFailsWith<CancellationException> {
+            flowScope {
+                flowScope {
+                    launch {
+                       throw CancellationException(null)
+                    }
+                }
+            }
+        }
+    }
+}

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

@@ -59,8 +59,7 @@ private class BlockingCoroutine<T>(
     private val blockedThread: Thread,
     private val eventLoop: EventLoop?
 ) : AbstractCoroutine<T>(parentContext, true) {
-    override val cancelsParent: Boolean
-        get() = false // it throws exception to parent instead of cancelling it
+    override val isScopedCoroutine: Boolean get() = true
 
     override fun afterCompletionInternal(state: Any?, mode: Int) {
         // wake up blocked thread

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

@@ -127,7 +127,6 @@ private open class ActorCoroutine<E>(
     channel: Channel<E>,
     active: Boolean
 ) : ChannelCoroutine<E>(parentContext, channel, active), ActorScope<E> {
-    override val cancelsParent: Boolean get() = true
 
     override fun onCancelling(cause: Throwable?) {
         _channel.cancel(cause?.let {

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

@@ -54,7 +54,7 @@ private class BlockingCoroutine<T>(
     parentContext: CoroutineContext,
     private val eventLoop: EventLoop?
 ) : AbstractCoroutine<T>(parentContext, true) {
-    override val cancelsParent: Boolean
+    override val isCoroutine: Boolean
         get() = false // it throws exception to parent instead of cancelling it
 
     @Suppress("UNCHECKED_CAST")