EventLoop is integrated as runBlocking default and is used for tests, coroutine builders provide CoroutineScope with context

Author: elizarov

kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Builders.kt

@@ -1,10 +1,7 @@
 package kotlinx.coroutines.experimental
 
 import java.util.concurrent.locks.LockSupport
-import kotlin.coroutines.Continuation
-import kotlin.coroutines.CoroutineContext
-import kotlin.coroutines.startCoroutine
-import kotlin.coroutines.suspendCoroutine
+import kotlin.coroutines.*
 
 // --------------- basic coroutine builders ---------------
 
@@ -21,29 +18,30 @@ import kotlin.coroutines.suspendCoroutine
  *
  * See [newCoroutineContext] for a description of debugging facilities that are available for newly created coroutine.
  */
-fun launch(context: CoroutineContext, block: suspend () -> Unit): Job =
-    StandaloneCoroutine(newCoroutineContext(context)).also { block.startCoroutine(it) }
+fun launch(context: CoroutineContext, block: suspend CoroutineScope.() -> Unit): Job =
+    StandaloneCoroutine(newCoroutineContext(context)).also { block.startCoroutine(it, it) }
 
 /**
  * Calls the specified suspending block with a given coroutine context, suspends until it completes, and returns
  * the result. It immediately applies dispatcher from the new context, shifting execution of the block into the
  * different thread inside the block, and back when it completes.
  * The specified [context] is merged onto the current coroutine context.
  */
-public suspend fun <T> run(context: CoroutineContext, block: suspend () -> T): T =
+public suspend fun <T> run(context: CoroutineContext, block: suspend CoroutineScope.() -> T): T =
     suspendCoroutine { cont ->
-        block.startCoroutine(object : Continuation<T> by cont {
-            override val context: CoroutineContext = cont.context + context
-        })
+        // new don't invoke `newCoroutineContext`, but consider this being the same coroutine in the new context
+        InnerCoroutine(cont.context + context, cont).also { block.startCoroutine(it, it) }
     }
 
 /**
  * Runs new coroutine and *blocks* current thread *interruptibly* until its completion.
- * This function should not be used from coroutine. It is designed to bridge regular code blocking code
- * to libraries that are written in suspending style.
- * The [context] for the new coroutine must be explicitly specified and must include [CoroutineDispatcher] element.
- * See [CoroutineDispatcher] for the standard [context] implementations that are provided by `kotlinx.coroutines`.
- * The specified context is added to the context of the parent running coroutine (if any) inside which this function
+ * This function should not be used from coroutine. It is designed to bridge regular blocking code
+ * to libraries that are written in suspending style, to be used in `main` functions and in tests.
+ *
+ * The default [CoroutineDispatcher] for this builder in an implementation of [EventLoop] that processes continuations
+ * in this blocked thread until the completion of this coroutine.
+ * See [CoroutineDispatcher] for the other implementations that are provided by `kotlinx.coroutines`.
+ * The specified [context] is added to the context of the parent running coroutine (if any) inside which this function
  * is invoked. The [Job] of the resulting coroutine is a child of the job of the parent coroutine (if any).
  *
  * If this blocked thread is interrupted (see [Thread.interrupt]), then the coroutine job is cancelled and
@@ -52,26 +50,45 @@ public suspend fun <T> run(context: CoroutineContext, block: suspend () -> T): T
  * See [newCoroutineContext] for a description of debugging facilities that are available for newly created coroutine.
  */
 @Throws(InterruptedException::class)
-public fun <T> runBlocking(context: CoroutineContext, block: suspend () -> T): T =
-    BlockingCoroutine<T>(newCoroutineContext(context)).also { block.startCoroutine(it) }.joinBlocking()
+public fun <T> runBlocking(context: CoroutineContext = EmptyCoroutineContext, block: suspend CoroutineScope.() -> T): T {
+    val currentThread = Thread.currentThread()
+    val privateEventLoop = if (context[ContinuationInterceptor] as? CoroutineDispatcher == null)
+        EventLoopImpl(currentThread) else null
+    val newContext = newCoroutineContext(context + (privateEventLoop ?: EmptyCoroutineContext))
+    val coroutine = BlockingCoroutine<T>(newContext, currentThread, privateEventLoop != null)
+    privateEventLoop?.initParentJob(coroutine)
+    block.startCoroutine(coroutine, coroutine)
+    return coroutine.joinBlocking()
+}
 
 // --------------- implementation ---------------
 
 private class StandaloneCoroutine(
-    val parentContext: CoroutineContext
-) : AbstractCoroutine<Unit>(parentContext) {
-    init { initParentJob(parentContext[Job]) }
+    val newContext: CoroutineContext
+) : AbstractCoroutine<Unit>(newContext) {
+    init { initParentJob(newContext[Job]) }
 
     override fun afterCompletion(state: Any?) {
-        // note the use of the parent context below!
-        if (state is CompletedExceptionally) handleCoroutineException(parentContext, state.cancelReason)
+        // note the use of the parent's job context below!
+        if (state is CompletedExceptionally) handleCoroutineException(newContext, state.cancelReason)
     }
 }
 
-private class BlockingCoroutine<T>(parentContext: CoroutineContext) : AbstractCoroutine<T>(parentContext) {
-    val blockedThread: Thread = Thread.currentThread()
+private class InnerCoroutine<T>(
+    override val context: CoroutineContext,
+    continuation: Continuation<T>
+) : Continuation<T> by continuation, CoroutineScope {
+    override val isActive: Boolean = context[Job]?.isActive ?: true
+}
+
+private class BlockingCoroutine<T>(
+    newContext: CoroutineContext,
+    val blockedThread: Thread,
+    val hasPrivateEventLoop: Boolean
+) : AbstractCoroutine<T>(newContext) {
+    val eventLoop: EventLoop? = newContext[ContinuationInterceptor] as? EventLoop
 
-    init { initParentJob(parentContext[Job])  }
+    init { initParentJob(newContext[Job]) }
 
     override fun afterCompletion(state: Any?) {
         LockSupport.unpark(blockedThread)
@@ -81,8 +98,14 @@ private class BlockingCoroutine<T>(parentContext: CoroutineContext) : AbstractCo
     fun joinBlocking(): T {
         while (isActive) {
             if (Thread.interrupted()) throw InterruptedException().also { cancel(it) }
-            LockSupport.park(this)
+            if (eventLoop == null || !eventLoop.processNextEvent())
+                LockSupport.park(this)
+        }
+        // process remaining events (that could have been added after last processNextEvent and before cancel
+        if (hasPrivateEventLoop) {
+            while (eventLoop!!.processNextEvent()) { /* just spin */ }
         }
+        // now return result
         val state = getState()
         (state as? CompletedExceptionally)?.let { throw it.exception }
         return state as T

kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/CommonPool.kt

@@ -11,7 +11,7 @@ import java.util.concurrent.atomic.AtomicInteger
  * coroutine resumption is dispatched as a separate task even when it already executes inside the pool.
  * When available, it wraps [ForkJoinPool.commonPool] and provides a similar shared pool where not.
  */
-object CommonPool : CoroutineDispatcher() {
+object CommonPool : CoroutineDispatcher(), Yield {
     private val pool: Executor = findPool()
 
     private inline fun <T> Try(block: () -> T) = try { block() } catch (e: Throwable) { null }
@@ -37,4 +37,8 @@ object CommonPool : CoroutineDispatcher() {
 
     override fun isDispatchNeeded(): Boolean = true
     override fun dispatch(block: Runnable) = pool.execute(block)
+
+    override fun scheduleResume(continuation: CancellableContinuation<Unit>) {
+        pool.execute { continuation.resume(Unit) }
+    }
 }

kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/AbstractCoroutine.kt renamed to kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/CoroutineScope.kt

@@ -3,14 +3,30 @@ package kotlinx.coroutines.experimental
 import kotlin.coroutines.Continuation
 import kotlin.coroutines.CoroutineContext
 
+/**
+ * Receiver interface for generic coroutine builders, so that the code inside coroutine has a convenient access
+ * to its [context] and cancellation status via [isActive].
+ */
+public interface CoroutineScope {
+    /**
+     * Returns `true` when this coroutine is still active (was not cancelled).
+     */
+    public val isActive: Boolean
+
+    /**
+     * Returns the context of this coroutine.
+     */
+    public val context: CoroutineContext
+}
+
 /**
  * Abstract class to simplify writing of coroutine completion objects that
  * implements [Continuation] and [Job] interfaces.
  * It stores the result of continuation in the state of the job.
  */
 @Suppress("LeakingThis")
-public abstract class AbstractCoroutine<in T>(parentContext: CoroutineContext) : JobSupport(), Continuation<T> {
-    override val context: CoroutineContext = parentContext + this // merges this job into this context
+public abstract class AbstractCoroutine<in T>(newContext: CoroutineContext) : JobSupport(), Continuation<T>, CoroutineScope {
+    override val context: CoroutineContext = newContext + this // merges this job into this context
 
     final override fun resume(value: T) {
         while (true) { // lock-free loop on state

kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/CurrentCoroutineContext.kt

@@ -3,7 +3,6 @@ package kotlinx.coroutines.experimental
 import java.util.concurrent.atomic.AtomicLong
 import kotlin.coroutines.AbstractCoroutineContextElement
 import kotlin.coroutines.ContinuationInterceptor
-import kotlin.coroutines.ContinuationInterceptor.Key
 import kotlin.coroutines.CoroutineContext
 import kotlin.coroutines.EmptyCoroutineContext
 
@@ -35,7 +34,6 @@ public object Here : CoroutineDispatcher() {
 public val currentCoroutineContext: CoroutineContext
     get() = CURRENT_CONTEXT.get() ?: throw IllegalStateException("Not inside a coroutine")
 
-
 /**
  * Returns the context of the coroutine that this function is invoked in or a specified [default]
  * if not invoked inside a coroutine. A [default] must be a singleton [CoroutineDispatcher] element.

kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Deferred.kt

@@ -33,13 +33,13 @@ public interface Deferred<out T> : Job {
  * The specified context is added to the context of the parent running coroutine (if any) inside which this function
  * is invoked. The [Job] of the resulting coroutine is a child of the job of the parent coroutine (if any).
  */
-public fun <T> defer(context: CoroutineContext, block: suspend () -> T) : Deferred<T> =
-    DeferredCoroutine<T>(newCoroutineContext(context)).also { block.startCoroutine(it) }
+public fun <T> defer(context: CoroutineContext, block: suspend CoroutineScope.() -> T) : Deferred<T> =
+    DeferredCoroutine<T>(newCoroutineContext(context)).also { block.startCoroutine(it, it) }
 
 private class DeferredCoroutine<T>(
-        parentContext: CoroutineContext
-) : AbstractCoroutine<T>(parentContext), Deferred<T> {
-    init { initParentJob(parentContext[Job]) }
+    newContext: CoroutineContext
+) : AbstractCoroutine<T>(newContext), Deferred<T> {
+    init { initParentJob(newContext[Job]) }
 
     @Suppress("UNCHECKED_CAST")
     suspend override fun await(): T {

kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Delay.kt

@@ -4,7 +4,8 @@ import java.util.concurrent.TimeUnit
 import kotlin.coroutines.ContinuationInterceptor
 
 /**
- * Implemented by [CoroutineDispatcher] implementations that natively support non-blocking [delay] function.
+ * This dispatcher _feature_ is implemented by [CoroutineDispatcher] implementations that natively support
+ * non-blocking [delay] function.
  */
 public interface Delay {
     /**
@@ -16,13 +17,13 @@ public interface Delay {
     suspend fun delay(time: Long, unit: TimeUnit = TimeUnit.MILLISECONDS) {
         require(time >= 0) { "Delay time $time cannot be negative" }
         if (time <= 0) return // don't delay
-        return suspendCancellableCoroutine { resumeAfterDelay(time, unit, it) }
+        return suspendCancellableCoroutine { scheduleResumeAfterDelay(time, unit, it) }
     }
 
     /**
-     * Resumes a specified continuation after a specified delay.
+     * Schedules resume of a specified [continuation] after a specified delay [time].
      */
-    fun resumeAfterDelay(time: Long, unit: TimeUnit, continuation: CancellableContinuation<Unit>)
+    fun scheduleResumeAfterDelay(time: Long, unit: TimeUnit, continuation: CancellableContinuation<Unit>)
 }
 
 /**
@@ -39,7 +40,7 @@ suspend fun delay(time: Long, unit: TimeUnit = TimeUnit.MILLISECONDS) {
     if (time <= 0) return // don't delay
     return suspendCancellableCoroutine sc@ { cont: CancellableContinuation<Unit> ->
         (cont.context[ContinuationInterceptor] as? Delay)?.apply {
-            resumeAfterDelay(time, unit, cont)
+            scheduleResumeAfterDelay(time, unit, cont)
             return@sc
         }
         val timeout = scheduledExecutor.schedule({ cont.resume(Unit) }, time, unit)

kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/EventLoop.kt

@@ -4,65 +4,91 @@ import kotlinx.coroutines.experimental.internal.LockFreeLinkedListHead
 import kotlinx.coroutines.experimental.internal.LockFreeLinkedListNode
 import java.util.concurrent.locks.LockSupport
 import kotlin.coroutines.Continuation
-import kotlin.coroutines.startCoroutine
 
+/**
+ * Implemented by [CoroutineDispatcher] implementations that have event loop inside and can
+ * be asked to process next event from their event queue. It is used by [runBlocking] to
+ * continue processing events when invoked from the event dispatch thread.
+ */
 public interface EventLoop {
-    public val thisEventLoop: CoroutineDispatcher
-    public suspend fun yield()
-}
+    /**
+     * Processes next event in this event loop and returns `true` or returns `false` if there are
+     * no events to process or when invoked from the wrong thread.
+     */
+    public fun processNextEvent(): Boolean
 
-@Throws(InterruptedException::class)
-public fun <T> runEventLoop(block: suspend EventLoop.() -> T): T =
-    EventLoopImpl<T>().also { block.startCoroutine(it, it.coroutine) }.coroutine.joinBlocking()
+    public companion object Factory {
+        /**
+         * Creates a new event loop that is bound the specified [thread] (current thread by default) and
+         * stops accepting new events when [parentJob] completes. Every continuation that is scheduled
+         * onto this event loop unparks the specified thread via [LockSupport.unpark].
+         *
+         * The main event-processing loop using the resulting `eventLoop` object should look like this:
+         * ```
+         * while (needsToBeRunning) {
+         *     if (Thread.interrupted()) break // or handle somehow
+         *     if (!eventLoop.processNextEvent()) LockSupport.park() // event loop will unpark
+         * }
+         * ```
+         */
+        public operator fun invoke(thread: Thread = Thread.currentThread(), parentJob: Job? = null): CoroutineDispatcher =
+            EventLoopImpl(thread).apply {
+                if (parentJob != null) initParentJob(parentJob)
+            }
+    }
+}
 
-private class EventLoopImpl<T> : CoroutineDispatcher(), EventLoop {
-    val thread: Thread = Thread.currentThread()
+internal class EventLoopImpl(
+    val thread: Thread
+) : CoroutineDispatcher(), EventLoop, Yield {
     val queue = LockFreeLinkedListHead()
-    val coroutine = Coroutine()
-
-    public override val thisEventLoop: CoroutineDispatcher = this
+    var parentJob: Job? = null
 
-    public override suspend fun yield(): Unit = suspendCancellableCoroutine { cont ->
-        val node = Resume(cont)
-        schedule(node)
-        cont.removeOnCompletion(node)
+    fun initParentJob(coroutine: Job) {
+        require(this.parentJob == null)
+        this.parentJob = coroutine
     }
 
     override fun isDispatchNeeded(): Boolean = Thread.currentThread() != thread
 
     override fun dispatch(block: Runnable) {
         schedule(Dispatch(block))
-        queue.addLast(Dispatch(block))
     }
 
-    fun schedule(node: LockFreeLinkedListNode) {
-        check(queue.addLastIf(node) { coroutine.isActive }) {
-            "EventLoop is already complete... cannot schedule any tasks"
-        }
-        LockSupport.unpark(thread)
+    override fun scheduleResume(continuation: CancellableContinuation<Unit>) {
+        val node = Resume(continuation)
+        if (schedule(node))
+            continuation.removeOnCompletion(node)
     }
 
-    inner class Coroutine : AbstractCoroutine<T>(this@EventLoopImpl) {
-        override fun afterCompletion(state: Any?) {
+    fun schedule(node: Node): Boolean {
+        val added = if (parentJob == null) {
+            queue.addLast(node)
+            true
+        } else
+            queue.addLastIf(node) { parentJob!!.isActive }
+        if (added) {
             LockSupport.unpark(thread)
+        } else {
+            node.run()
         }
+        return added
+    }
 
-        @Suppress("UNCHECKED_CAST")
-        fun joinBlocking(): T {
-            while (isActive) {
-                if (Thread.interrupted()) throw InterruptedException().also { cancel(it) }
-                (queue.removeFirstOrNull() as? Runnable)?.run() ?: LockSupport.park(this)
-            }
-            check(queue.isEmpty) { "There are still tasks in event loop queue... Stray coroutines?"}
-            val state = getState()
-            (state as? CompletedExceptionally)?.let { throw it.exception }
-            return state as T
+    override fun processNextEvent(): Boolean {
+        if (Thread.currentThread() != thread) return false
+        (queue.removeFirstOrNull() as? Runnable)?.apply {
+            run()
+            return true
         }
+        return false
     }
 
-    class Dispatch(block: Runnable) : LockFreeLinkedListNode(), Runnable by block
+    abstract class Node : LockFreeLinkedListNode(), Runnable
+
+    class Dispatch(block: Runnable) : Node(), Runnable by block
 
-    class Resume(val cont: Continuation<Unit>) : LockFreeLinkedListNode(), Runnable {
+    class Resume(val cont: Continuation<Unit>) : Node() {
         override fun run() = cont.resume(Unit)
     }
 }

kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Job.kt

@@ -23,6 +23,9 @@ import kotlin.coroutines.CoroutineContext
  * All functions on this interface are thread-safe.
  */
 public interface Job : CoroutineContext.Element {
+    /**
+     * Key for [Job] instance in the coroutine context.
+     */
     public companion object Key : CoroutineContext.Key<Job> {
         /**
          * Creates new job object. It is optionally a child of a [parent] job.