Structured concurrency implementation:

  * Introducing async, launch, produce, actor and broadcast extensions on CoroutineScope
  * Deprecate top-level coroutine builders
  * Introducing currentScope and coroutineScope for manipulation with CoroutineScope interface
  * Introducing CoroutineScope factories
  * Introducing extension CoroutineScope.isActive

Fixes #410

Author: qwwdfsad

binary-compatibility-validator/test/PublicApiTest.kt

@@ -12,6 +12,7 @@ import java.util.*
 import java.util.jar.*
 import kotlin.collections.ArrayList
 
+@Ignore
 @RunWith(Parameterized::class)
 class PublicApiTest(
     private val rootDir: String,

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

@@ -36,7 +36,9 @@ public abstract class AbstractCoroutine<in T>(
     @Suppress("LeakingThis")
     public final override val context: CoroutineContext = parentContext + this
     @Deprecated("Replaced with context", replaceWith = ReplaceWith("context"))
-    public final override val coroutineContext: CoroutineContext get() = context
+    public override val coroutineContext: CoroutineContext get() = context
+
+    override val isActive: Boolean get() = super<JobSupport>.isActive
 
     /**
      * Initializes parent job from the `parentContext` of this coroutine that was passed to it during construction.

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

@@ -18,17 +18,13 @@ import kotlin.coroutines.experimental.intrinsics.*
  * Launches new coroutine without blocking current thread and returns a reference to the coroutine as a [Job].
  * The coroutine is cancelled when the resulting job is [cancelled][Job.cancel].
  *
- * The [context] for the new coroutine can be explicitly specified.
- * See [CoroutineDispatcher] for the standard context implementations that are provided by `kotlinx.coroutines`.
- * The [coroutineContext](https://kotlinlang.org/api/latest/jvm/stdlib/kotlin.coroutines.experimental/coroutine-context.html)
- * of the parent coroutine may be used,
- * in which case the [Job] of the resulting coroutine is a child of the job of the parent coroutine.
- * The parent job may be also explicitly specified using [parent] parameter.
- *
+ * Coroutine context is inherited from a [CoroutineScope], additional context elements can be specified with [context] argument.
  * If the context does not have any dispatcher nor any other [ContinuationInterceptor], then [DefaultDispatcher] is used.
+ * The parent job is inherited from a [CoroutineScope] as well, but it can also be overridden
+ * with corresponding [coroutineContext] element.
  *
  * By default, the coroutine is immediately scheduled for execution.
- * Other options can be specified via `start` parameter. See [CoroutineStart] for details.
+ * Other start options can be specified via `start` parameter. See [CoroutineStart] for details.
  * An optional [start] parameter can be set to [CoroutineStart.LAZY] to start coroutine _lazily_. In this case,
  * the coroutine [Job] is created in _new_ state. It can be explicitly started with [start][Job.start] function
  * and will be started implicitly on the first invocation of [join][Job.join].
@@ -39,35 +35,68 @@ import kotlin.coroutines.experimental.intrinsics.*
  *
  * See [newCoroutineContext] for a description of debugging facilities that are available for newly created coroutine.
  *
- * @param context context of the coroutine. The default value is [DefaultDispatcher].
+ * @param context additional to [CoroutineScope.coroutineContext] context of the coroutine
  * @param start coroutine start option. The default value is [CoroutineStart.DEFAULT].
- * @param parent explicitly specifies the parent job, overrides job from the [context] (if any).
  * @param onCompletion optional completion handler for the coroutine (see [Job.invokeOnCompletion]).
- * @param block the coroutine code.
- */
-public fun launch(
-    context: CoroutineContext = DefaultDispatcher,
+ * @param block the coroutine code which will be invoked in the context of the provided scope
+ **/
+public fun CoroutineScope.launch(
+    context: CoroutineContext = EmptyCoroutineContext,
     start: CoroutineStart = CoroutineStart.DEFAULT,
-    parent: Job? = null,
     onCompletion: CompletionHandler? = null,
     block: suspend CoroutineScope.() -> Unit
 ): Job {
-    val newContext = newCoroutineContext(context, parent)
+    val newContext = newCoroutineContext(context)
     val coroutine = if (start.isLazy)
         LazyStandaloneCoroutine(newContext, block) else
         StandaloneCoroutine(newContext, active = true)
     if (onCompletion != null) coroutine.invokeOnCompletion(handler = onCompletion)
     coroutine.start(start, coroutine, block)
     return coroutine
 }
+
+/**
+ * Launches new coroutine without blocking current thread and returns a reference to the coroutine as a [Job].
+ * @suppress **Deprecated** Use [CoroutineScope.launch] instead.
+ */
+@Deprecated(
+    message = "Standalone coroutine builders are deprecated, use extensions on CoroutineScope instead",
+    replaceWith = ReplaceWith("GlobalScope.launch(context, start, onCompletion, block)", imports = ["kotlinx.coroutines.experimental.*"])
+)
+public fun launch(
+    context: CoroutineContext = DefaultDispatcher,
+    start: CoroutineStart = CoroutineStart.DEFAULT,
+    onCompletion: CompletionHandler? = null,
+    block: suspend CoroutineScope.() -> Unit
+): Job =
+    GlobalScope.launch(context, start, onCompletion, block)
+
+/**
+ * Launches new coroutine without blocking current thread and returns a reference to the coroutine as a [Job].
+ * @suppress **Deprecated** Use [CoroutineScope.launch] instead.
+ */
+@Deprecated(
+    message = "Standalone coroutine builders are deprecated, use extensions on CoroutineScope instead. This API will be hidden in the next release",
+    replaceWith = ReplaceWith("GlobalScope.launch(context + parent, start, onCompletion, block)", imports = ["kotlinx.coroutines.experimental.*"])
+)
+public fun launch(
+    context: CoroutineContext = DefaultDispatcher,
+    start: CoroutineStart = CoroutineStart.DEFAULT,
+    parent: Job? = null, // nullable for binary compatibility
+    onCompletion: CompletionHandler? = null,
+    block: suspend CoroutineScope.() -> Unit
+): Job =
+    GlobalScope.launch(context + (parent ?: EmptyCoroutineContext), start, onCompletion, block)
+
 /** @suppress **Deprecated**: Binary compatibility */
 @Deprecated(message = "Binary compatibility", level = DeprecationLevel.HIDDEN)
 public fun launch(
     context: CoroutineContext = DefaultDispatcher,
     start: CoroutineStart = CoroutineStart.DEFAULT,
     parent: Job? = null,
     block: suspend CoroutineScope.() -> Unit
-): Job = launch(context, start, parent, block = block)
+): Job =
+    GlobalScope.launch(context + (parent ?: EmptyCoroutineContext), start, block = block)
 
 /** @suppress **Deprecated**: Binary compatibility */
 @Deprecated(message = "Binary compatibility", level = DeprecationLevel.HIDDEN)
@@ -84,7 +113,7 @@ public fun launch(
 @Deprecated(message = "Use `start = CoroutineStart.XXX` parameter",
     replaceWith = ReplaceWith("launch(context, if (start) CoroutineStart.DEFAULT else CoroutineStart.LAZY, block)"))
 public fun launch(context: CoroutineContext, start: Boolean, block: suspend CoroutineScope.() -> Unit): Job =
-    launch(context, if (start) CoroutineStart.DEFAULT else CoroutineStart.LAZY, block = block)
+    GlobalScope.launch(context, if (start) CoroutineStart.DEFAULT else CoroutineStart.LAZY, block = block)
 
 /**
  * Calls the specified suspending block with a given coroutine context, suspends until it completes, and returns
@@ -103,6 +132,19 @@ public fun launch(context: CoroutineContext, start: Boolean, block: suspend Coro
  * A value of [CoroutineStart.LAZY] is not supported and produces [IllegalArgumentException].
  */
 public suspend fun <T> withContext(
+    context: CoroutineContext,
+    start: CoroutineStart = CoroutineStart.DEFAULT,
+    block: suspend CoroutineScope.() -> T
+): T =
+    // todo: optimize fast-path to work without allocation (when there is a already a coroutine implementing scope)
+    withContextImpl(context, start) {
+        currentScope {
+            block()
+        }
+    }
+
+// todo: optimize it to reduce allocations
+private suspend fun <T> withContextImpl(
     context: CoroutineContext,
     start: CoroutineStart = CoroutineStart.DEFAULT,
     block: suspend () -> T
@@ -137,6 +179,16 @@ public suspend fun <T> withContext(
     completion.getResult()
 }
 
+/** @suppress **Deprecated**: Binary compatibility */
+@Deprecated(level = DeprecationLevel.HIDDEN, message = "Binary compatibility")
+@JvmName("withContext")
+public suspend fun <T> withContext0(
+    context: CoroutineContext,
+    start: CoroutineStart = CoroutineStart.DEFAULT,
+    block: suspend () -> T
+): T =
+    withContextImpl(context, start, block)
+
 /** @suppress **Deprecated**: Renamed to [withContext]. */
 @Deprecated(message = "Renamed to `withContext`", level=DeprecationLevel.WARNING,
     replaceWith = ReplaceWith("withContext(context, start, block)"))
@@ -145,12 +197,12 @@ public suspend fun <T> run(
     start: CoroutineStart = CoroutineStart.DEFAULT,
     block: suspend () -> T
 ): T =
-    withContext(context, start, block)
+    withContextImpl(context, start, block)
 
 /** @suppress **Deprecated** */
 @Deprecated(message = "It is here for binary compatibility only", level=DeprecationLevel.HIDDEN)
 public suspend fun <T> run(context: CoroutineContext, block: suspend () -> T): T =
-    withContext(context, start = CoroutineStart.ATOMIC, block = block)
+    withContextImpl(context, start = CoroutineStart.ATOMIC, block = block)
 
 // --------------- implementation ---------------
 

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

@@ -4,12 +4,51 @@
 
 package kotlinx.coroutines.experimental
 
+import kotlinx.coroutines.experimental.internal.*
 import kotlin.coroutines.experimental.*
-import kotlin.internal.*
 
 /**
- * Receiver interface for generic coroutine builders, so that the code inside coroutine has a convenient
- * and fast access to its own cancellation status via [isActive].
+ * Defines a scope for new coroutines. Every coroutine builder
+ * is an extension on [CoroutineScope] and inherits its [coroutineContext][CoroutineScope.coroutineContext]
+ * to automatically propagate both context elements and cancellation.
+ *
+ * [CoroutineScope] should be implemented on entities with well-defined lifecycle that are responsible
+ * for launching children coroutines. Example of such entity on Android is Activity.
+ * Usage of this interface may look like this:
+ *
+ * ```
+ * class MyActivity : AppCompatActivity(), CoroutineScope {
+ *
+ *     override val coroutineContext: CoroutineContext
+ *         get() = job + UI
+ *
+ *     override fun onCreate(savedInstanceState: Bundle?) {
+ *         super.onCreate(savedInstanceState)
+ *         job = Job()
+ *     }
+ *
+ *     override fun onDestroy() {
+ *         super.onDestroy()
+ *         job.cancel() // Cancel job on activity destroy. After destroy all children jobs will be cancelled automatically
+ *     }
+ *
+ *     /*
+ *      * Note how coroutine builders are scoped: if activity is destroyed or any of the launched coroutines
+ *      * in this method throws an exception, then all nested coroutines will be cancelled.
+ *      */
+ *     fun loadDataFromUI() = launch { // <- extension on current activity, launched in CommonPool
+ *        val ioData = async(IO) { // <- extension on launch scope, launched in IO dispatcher
+ *          // long computation
+ *        }
+ *
+ *        withContext(UI) {
+ *            val data = ioData.await()
+ *            draw(data)
+ *        }
+ *     }
+ * }
+ *
+ * ```
  */
 public interface CoroutineScope {
     /**
@@ -26,18 +65,140 @@ public interface CoroutineScope {
      * [CoroutineScope] is available.
      * See [coroutineContext][kotlin.coroutines.experimental.coroutineContext],
      * [isActive][kotlinx.coroutines.experimental.isActive] and [Job.isActive].
+     *
+     * @suppress **Deprecated**: Deprecated in favor of top-level extension property
      */
+    @Deprecated(level = DeprecationLevel.HIDDEN, message = "Deprecated in favor of top-level extension property")
     public val isActive: Boolean
+        get() = coroutineContext[Job]?.isActive ?: true
 
     /**
-     * Returns the context of this coroutine.
-     *
-     * @suppress: **Deprecated**: Replaced with top-level [kotlin.coroutines.experimental.coroutineContext].
+     * Returns the context of this scope.
      */
-    @Deprecated("Replace with top-level coroutineContext",
-        replaceWith = ReplaceWith("coroutineContext",
-            imports = ["kotlin.coroutines.experimental.coroutineContext"]))
-    @LowPriorityInOverloadResolution
-    @Suppress("INVISIBLE_MEMBER", "INVISIBLE_REFERENCE")
     public val coroutineContext: CoroutineContext
-}
+}
+
+/**
+ * Adds the specified coroutine context to this scope, overriding existing elements in the current
+ * scope's context with the corresponding keys.
+ *
+ * This is a shorthand for `CoroutineScope(thisScope + context)`.
+ */
+public operator fun CoroutineScope.plus(context: CoroutineContext): CoroutineScope =
+    CoroutineScope(context + context)
+
+/**
+ * Returns `true` when current [Job] is still active (has not completed and was not cancelled yet).
+ *
+ * Check this property in long-running computation loops to support cancellation:
+ * ```
+ * while (_isActive) {
+ *     // do some computation
+ * }
+ * ```
+ *
+ * This property is a shortcut for `coroutineContext.isActive` in the scope when
+ * [CoroutineScope] is available.
+ * See [coroutineContext][kotlin.coroutines.experimental.coroutineContext],
+ * [isActive][kotlinx.coroutines.experimental.isActive] and [Job.isActive].
+ */
+@Suppress("EXTENSION_SHADOWED_BY_MEMBER")
+public val CoroutineScope.isActive: Boolean
+    get() = coroutineContext[Job]?.isActive ?: true
+
+/**
+ * A global [CoroutineScope] not bound to any job.
+ *
+ * Global scope is used to launch top-level coroutines which are operating on the whole application lifetime
+ * and are not cancelled prematurely.
+ * Another use of the global scope is [Unconfined] operators, which don't have any job associated with them.
+ *
+ * Application code usually should use application-defined [CoroutineScope], using [async] or [launch]
+ * on the instance of [GlobalScope] is highly discouraged.
+ *
+ * Usage of this interface may look like this:
+ *
+ * ```
+ * fun ReceiveChannel<Int>.sqrt(): ReceiveChannel<Double> = GlobalScope.produce(Unconfined) {
+ *     for (number in this) {
+ *         send(Math.sqrt(number))
+ *     }
+ * }
+ *
+ * ```
+ */
+object GlobalScope : CoroutineScope {
+    /**
+     * @suppress **Deprecated**: Deprecated in favor of top-level extension property
+     */
+    @Deprecated(level = DeprecationLevel.HIDDEN, message = "Deprecated in favor of top-level extension property")
+    override val isActive: Boolean
+        get() = true
+
+    /**
+     * Returns [EmptyCoroutineContext].
+     */
+    override val coroutineContext: CoroutineContext
+        get() = EmptyCoroutineContext
+}
+
+/**
+ * Creates new [CoroutineScope] and calls the specified suspend block with this scope.
+ * The provided scope inherits its [coroutineContext][CoroutineScope.coroutineContext] from the outer scope, but overrides
+ * context's [Job].
+ *
+ * This methods returns as soon as given block and all launched from within the scope children coroutines are completed.
+ * Example of the scope usages looks like this:
+ *
+ * ```
+ * suspend fun loadDataForUI() = coroutineScope {
+ *
+ *   val data = async { // <- extension on current scope
+ *      ... load some UI data ...
+ *   }
+ *
+ *   withContext(UI) {
+ *     doSomeWork()
+ *     val result = data.await()
+ *     display(result)
+ *   }
+ * }
+ * ```
+ *
+ * Semantics of the scope in this example:
+ * 1) `loadDataForUI` returns as soon as data is loaded and UI is updated.
+ * 2) If `doSomeWork` throws an exception, then `async` task is cancelled and `loadDataForUI` rethrows that exception.
+ * 3) If outer scope of `loadDataForUI` is cancelled, both started `async` and `withContext` are cancelled.
+ *
+ * Method may throw [JobCancellationException] if the current job was cancelled externally
+ * or may throw the corresponding unhandled [Throwable] if there is any unhandled exception in this scope
+ * (for example, from a crashed coroutine that was started with [launch] in this scope).
+ */
+public suspend fun <R> coroutineScope(block: suspend CoroutineScope.() -> R): R {
+    // todo: optimize implementation to a single allocated object
+    val owner = ScopeOwnerCoroutine<R>(coroutineContext)
+    owner.start(CoroutineStart.UNDISPATCHED, owner, block)
+    owner.join()
+    if (owner.isCancelled) {
+        throw owner.getCancellationException().let { it.cause ?: it }
+    }
+    val state = owner.state
+    if (state is CompletedExceptionally) {
+        throw state.cause
+    }
+    @Suppress("UNCHECKED_CAST")
+    return state as R
+}
+
+/**
+ * Provides [CoroutineScope] that is already present in the current [coroutineContext] to the given [block].
+ * Note, this method doesn't wait for all launched children to complete (as opposed to [coroutineContext]).
+ */
+public suspend inline fun <R> currentScope(block: CoroutineScope.() -> R): R =
+    CoroutineScope(coroutineContext).block()
+
+/**
+ * Creates [CoroutineScope] that wraps the given [coroutineContext].
+ */
+@Suppress("FunctionName")
+public fun CoroutineScope(context: CoroutineContext): CoroutineScope = ContextScope(context)