lazyDefer introduced
diff --git a/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Builders.kt b/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Builders.kt
index a8f4c9c..efd1651 100644
--- a/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Builders.kt
+++ b/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Builders.kt
@@ -21,7 +21,10 @@
* See [newCoroutineContext] for a description of debugging facilities that are available for newly created coroutine.
*/
fun launch(context: CoroutineContext, block: suspend CoroutineScope.() -> Unit): Job =
- StandaloneCoroutine(newCoroutineContext(context)).also { block.startCoroutine(it, it) }
+ StandaloneCoroutine(newCoroutineContext(context)).apply {
+ initParentJob(context[Job])
+ block.startCoroutine(this, this)
+ }
/**
* Calls the specified suspending block with a given coroutine context, suspends until it completes, and returns
@@ -54,10 +57,10 @@
@Throws(InterruptedException::class)
public fun <T> runBlocking(context: CoroutineContext = EmptyCoroutineContext, block: suspend CoroutineScope.() -> T): T {
val currentThread = Thread.currentThread()
- val privateEventLoop = if (context[ContinuationInterceptor] == null)
- EventLoopImpl(currentThread) else null
+ val privateEventLoop = if (context[ContinuationInterceptor] == null) EventLoopImpl(currentThread) else null
val newContext = newCoroutineContext(context + (privateEventLoop ?: EmptyCoroutineContext))
val coroutine = BlockingCoroutine<T>(newContext, currentThread, privateEventLoop != null)
+ coroutine.initParentJob(context[Job])
privateEventLoop?.initParentJob(coroutine)
block.startCoroutine(coroutine, coroutine)
return coroutine.joinBlocking()
@@ -66,13 +69,11 @@
// --------------- implementation ---------------
private class StandaloneCoroutine(
- val newContext: CoroutineContext
-) : AbstractCoroutine<Unit>(newContext) {
- init { initParentJob(newContext[Job]) }
-
+ val parentContext: CoroutineContext
+) : AbstractCoroutine<Unit>(parentContext) {
override fun afterCompletion(state: Any?) {
// note the use of the parent's job context below!
- if (state is CompletedExceptionally) handleCoroutineException(newContext, state.cancelReason)
+ if (state is CompletedExceptionally) handleCoroutineException(parentContext, state.cancelReason)
}
}
@@ -84,13 +85,11 @@
}
private class BlockingCoroutine<T>(
- newContext: CoroutineContext,
+ context: CoroutineContext,
val blockedThread: Thread,
val hasPrivateEventLoop: Boolean
-) : AbstractCoroutine<T>(newContext) {
- val eventLoop: EventLoop? = newContext[ContinuationInterceptor] as? EventLoop
-
- init { initParentJob(newContext[Job]) }
+) : AbstractCoroutine<T>(context) {
+ val eventLoop: EventLoop? = context[ContinuationInterceptor] as? EventLoop
override fun afterCompletion(state: Any?) {
if (Thread.currentThread() != blockedThread)
diff --git a/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/CoroutineScope.kt b/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/CoroutineScope.kt
index 3629f2d..3fc38c0 100644
--- a/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/CoroutineScope.kt
+++ b/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/CoroutineScope.kt
@@ -25,8 +25,8 @@
* It stores the result of continuation in the state of the job.
*/
@Suppress("LeakingThis")
-public abstract class AbstractCoroutine<in T>(newContext: CoroutineContext) : JobSupport(), Continuation<T>, CoroutineScope {
- override val context: CoroutineContext = newContext + this // merges this job into this context
+public abstract class AbstractCoroutine<in T>(context: CoroutineContext) : JobSupport(), Continuation<T>, CoroutineScope {
+ override val context: CoroutineContext = context + this // merges this job into this context
final override fun resume(value: T) {
while (true) { // lock-free loop on state
diff --git a/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Deferred.kt b/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Deferred.kt
index 70a5335..1348d3e 100644
--- a/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Deferred.kt
+++ b/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Deferred.kt
@@ -6,6 +6,7 @@
/**
* Deferred value is conceptually a non-blocking cancellable future.
* It is created with [defer] coroutine builder.
+ * It is in [active][isActive] state while the value is being computed.
*/
public interface Deferred<out T> : Job {
/**
@@ -35,20 +36,33 @@
* in which case the [Job] of the resulting coroutine is a child of the job of the parent coroutine.
*/
public fun <T> defer(context: CoroutineContext, block: suspend CoroutineScope.() -> T) : Deferred<T> =
- DeferredCoroutine<T>(newCoroutineContext(context)).also { block.startCoroutine(it, it) }
+ DeferredCoroutine<T>(newCoroutineContext(context)).apply {
+ initParentJob(context[Job])
+ block.startCoroutine(this, this)
+ }
-private class DeferredCoroutine<T>(
- newContext: CoroutineContext
-) : AbstractCoroutine<T>(newContext), Deferred<T> {
- init { initParentJob(newContext[Job]) }
+internal open class DeferredCoroutine<T>(
+ context: CoroutineContext
+) : AbstractCoroutine<T>(context), Deferred<T> {
+ protected open fun start(): Boolean = false // LazyDeferredCoroutine overrides
@Suppress("UNCHECKED_CAST")
suspend override fun await(): T {
// quick check if already complete (avoid extra object creation)
- val state = getState()
- if (state !is Active) {
- if (state is CompletedExceptionally) throw state.exception
- return state as T
+ getState().let { state ->
+ if (state !is Active) {
+ if (state is CompletedExceptionally) throw state.exception
+ return state as T
+ }
+ }
+ if (start()) { // LazyDeferredCoroutine overrides
+ // recheck state (may have started & already completed
+ getState().let { state ->
+ if (state !is Active) {
+ if (state is CompletedExceptionally) throw state.exception
+ return state as T
+ }
+ }
}
// Note: await is cancellable itself!
return awaitGetValue()
diff --git a/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Job.kt b/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Job.kt
index f3fd0a3..87f7305 100644
--- a/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Job.kt
+++ b/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Job.kt
@@ -139,7 +139,7 @@
}
// invoke at most once after construction after all other initialization
- protected fun initParentJob(parent: Job?) {
+ public fun initParentJob(parent: Job?) {
if (parent == null) return
check(registration == null)
// directly pass HandlerNode to parent scope to optimize one closure object (see makeNode)
@@ -155,9 +155,11 @@
expect as ActiveList // assert type
require(update !is Active) // only active -> inactive transition is allowed
if (!STATE.compareAndSet(this, expect, update)) return false
- // #1. Unregister from parent job
+ // #1. Update linked state before invoking completion handlers
+ onStateUpdate(update)
+ // #2. Unregister from parent job
registration?.unregister() // volatile read registration _after_ state was updated
- // #2 Invoke completion handlers
+ // #3. Invoke completion handlers
val reason = (update as? CompletedExceptionally)?.cancelReason
var completionException: Throwable? = null
expect.forEach<JobNode> { node ->
@@ -167,7 +169,7 @@
completionException?.apply { addSuppressed(ex) } ?: run { completionException = ex }
}
}
- // #3 Do other (overridable) processing
+ // #4. Do other (overridable) processing after completion handlers
completionException?.let { handleCompletionException(it) }
afterCompletion(update)
return true
@@ -197,6 +199,11 @@
}
/**
+ * Override to make linked state changes before completion handlers are invoked.
+ */
+ protected open fun onStateUpdate(update: Any?) {}
+
+ /**
* Override to process any exceptions that were encountered while invoking [onCompletion] handlers.
*/
protected open fun handleCompletionException(closeException: Throwable) {
diff --git a/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/LazyDeferred.kt b/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/LazyDeferred.kt
new file mode 100644
index 0000000..3121d4a
--- /dev/null
+++ b/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/LazyDeferred.kt
@@ -0,0 +1,112 @@
+package kotlinx.coroutines.experimental
+
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater
+import kotlin.coroutines.CoroutineContext
+import kotlin.coroutines.startCoroutine
+
+/**
+ * Lazy deferred value is conceptually a non-blocking cancellable future that is started on
+ * the first [await] or [start] invocation.
+ * It is created with [lazyDefer] coroutine builder.
+ *
+ * Unlike a simple [Deferred] value, a lazy deferred value has three states:
+ * * _Pending_ -- before the starts of the coroutine ([isActive] is `true`, but [isComputing] is `false`).
+ * * _Computing_ -- while computing the value ([isActive] is `true` and [isComputing] is `true`).
+ * * _Complete_ -- when done computing the value ([isActive] is `false` and [isComputing] is `false`).
+ *
+ * If this lazy deferred value is [cancelled][cancel], then it becomes immediately complete and
+ * cancels ongoing computation coroutine if it was started.
+ */
+public interface LazyDeferred<T> : Deferred<T> {
+ /**
+ * Returns `true` if the coroutine is computing its value.
+ */
+ public val isComputing: Boolean
+
+ /**
+ * Starts coroutine to compute this lazily deferred value. The result `true` if this invocation actually
+ * started coroutine or `false` if it was already started or cancelled.
+ */
+ public fun start(): Boolean
+}
+
+/**
+ * Lazily starts new coroutine on the first [await][Deferred.await] or [start][LazyDeferred.start] invocation
+ * on the resulting [LazyDeferred].
+ * The running coroutine is cancelled when the resulting value is [cancelled][Job.cancel].
+ *
+ * The [context] for the new coroutine must be explicitly specified.
+ * See [CoroutineDispatcher] for the standard [context] implementations that are provided by `kotlinx.coroutines`.
+ * The [context][CoroutineScope.context] of the parent coroutine from its [scope][CoroutineScope] may be used,
+ * in which case the [Job] of the resulting coroutine is a child of the job of the parent coroutine.
+ */
+public fun <T> lazyDefer(context: CoroutineContext, block: suspend CoroutineScope.() -> T) : LazyDeferred<T> =
+ LazyDeferredCoroutine<T>(newCoroutineContext(context), block).apply {
+ initParentJob(context[Job])
+ }
+
+private class LazyDeferredCoroutine<T>(
+ context: CoroutineContext,
+ val block: suspend CoroutineScope.() -> T
+) : DeferredCoroutine<T>(context), LazyDeferred<T> {
+
+ @Volatile
+ var lazyState: Int = STATE_PENDING
+
+ companion object {
+ private val STATE_PENDING = 0
+ private val STATE_COMPUTING = 1
+ private val STATE_COMPLETE = 2
+
+ private val LAZY_STATE: AtomicIntegerFieldUpdater<LazyDeferredCoroutine<*>> =
+ AtomicIntegerFieldUpdater.newUpdater(LazyDeferredCoroutine::class.java, "lazyState")
+ }
+
+ /*
+ === State linking & linearization of the overall state ===
+
+ There are two state variables in this object and they have to update atomically from the standpoint of
+ external observer:
+ 1. Job.state that is used by isActive function.
+ 2. lazyState that is used to make sure coroutine starts at most once.
+ External observer must see only three states, not four, i.e. it should not be able
+ to see `isActive == false`, but `isComputing == true`.
+
+ On completion/cancellation state variables are updated in this order:
+ a) state <- complete (isComplete starts returning true)
+ b) lazyState <- STATE_COMPLETE (see onStateUpdate)
+ This is why, `isComputing` checks state variables in reverse order:
+ a) lazyState is checked _first_
+ b) isActive is checked after it
+ This way cancellation/completion is atomic w.r.t to all state functions.
+
+ `start` function also has to check lazyState _before_ isActive.
+ */
+
+ override val isComputing: Boolean get() = lazyState == STATE_COMPUTING && isActive
+
+ override fun start(): Boolean {
+ while (true) { // lock-free loop on lazyState
+ when (lazyState) { // volatile read
+ STATE_PENDING -> {
+ if (isActive) { // then volatile read Job.state (inside isActive)
+ // can try to start
+ if (LAZY_STATE.compareAndSet(this, STATE_PENDING, STATE_COMPUTING)) {
+ block.startCoroutine(this, this)
+ return true
+ }
+ } else {
+ // cannot start -- already complete -- help update lazyState
+ lazyState = STATE_COMPLETE
+ return false
+ }
+ }
+ else -> return false
+ }
+ }
+ }
+
+ override fun onStateUpdate(update: Any?) {
+ lazyState = STATE_COMPLETE
+ }
+}