EventLoop work in progress

Author: elizarov

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

@@ -3,14 +3,18 @@ package kotlinx.coroutines.experimental
 import kotlin.coroutines.Continuation
 import kotlin.coroutines.CoroutineContext
 
-// internal helper class for various primitives that combines Job and Continuation implementations
+/**
+ * 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")
-internal open class JobContinuation<in T>(
+public abstract class AbstractCoroutine<in T>(
     parentContext: CoroutineContext
 ) : JobSupport(parentContext[Job]), Continuation<T> {
     override val context: CoroutineContext = parentContext + this // mixes this job into this context
 
-    override fun resume(value: T) {
+    final override fun resume(value: T) {
         while (true) { // lock-free loop on state
             val state = getState() // atomic read
             when (state) {
@@ -21,7 +25,7 @@ internal open class JobContinuation<in T>(
         }
     }
 
-    override fun resumeWithException(exception: Throwable) {
+    final override fun resumeWithException(exception: Throwable) {
         while (true) { // lock-free loop on state
             val state = getState() // atomic read
             when (state) {
@@ -36,7 +40,7 @@ internal open class JobContinuation<in T>(
         }
     }
 
-    override fun afterCompletion(state: Any?, closeException: Throwable?) {
-        if (closeException != null) handleCoroutineException(context, closeException)
+    final override fun handleCompletionException(closeException: Throwable) {
+        handleCoroutineException(context, closeException)
     }
 }

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

@@ -59,19 +59,17 @@ public fun <T> runBlocking(context: CoroutineContext, block: suspend () -> T): T
 
 private class StandaloneCoroutine(
     val parentContext: CoroutineContext
-) : JobContinuation<Unit>(parentContext) {
-    override fun afterCompletion(state: Any?, closeException: Throwable?) {
-        super.afterCompletion(state, closeException) // handle closeException
+) : AbstractCoroutine<Unit>(parentContext) {
+    override fun afterCompletion(state: Any?) {
         // note the use of the parent context below!
         if (state is CompletedExceptionally) handleCoroutineException(parentContext, state.exception)
     }
 }
 
-private class BlockingCoroutine<T>(parentContext: CoroutineContext) : JobContinuation<T>(parentContext) {
+private class BlockingCoroutine<T>(parentContext: CoroutineContext) : AbstractCoroutine<T>(parentContext) {
     val blockedThread: Thread = Thread.currentThread()
 
-    override fun afterCompletion(state: Any?, closeException: Throwable?) {
-        super.afterCompletion(state, closeException) // handle closeException
+    override fun afterCompletion(state: Any?) {
         LockSupport.unpark(blockedThread)
     }
 

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

@@ -31,7 +31,7 @@ public inline suspend fun <T> suspendCancellableCoroutine(crossinline block: (Ca
 @PublishedApi
 internal class SafeCancellableContinuation<in T>(
     private val delegate: Continuation<T>
-) : JobContinuation<T>(delegate.context), CancellableContinuation<T> {
+) : AbstractCoroutine<T>(delegate.context), CancellableContinuation<T> {
     // only updated from the thread that invoked suspendCancellableCoroutine
     private var suspendedThread: Thread? = Thread.currentThread()
 
@@ -46,8 +46,7 @@ internal class SafeCancellableContinuation<in T>(
     }
 
     @Suppress("UNCHECKED_CAST")
-    override fun afterCompletion(state: Any?, closeException: Throwable?) {
-        super.afterCompletion(state, closeException) // handle closeException
+    override fun afterCompletion(state: Any?) {
         if (suspendedThread === Thread.currentThread()) {
             // cancelled during suspendCancellableCoroutine in its thread
             suspendedThread = null

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

@@ -38,7 +38,7 @@ public fun <T> defer(context: CoroutineContext, block: suspend () -> T) : Deferr
 
 private class DeferredCoroutine<T>(
         parentContext: CoroutineContext
-) : JobContinuation<T>(parentContext), Deferred<T> {
+) : AbstractCoroutine<T>(parentContext), Deferred<T> {
     @Suppress("UNCHECKED_CAST")
     suspend override fun await(): T {
         // quick check if already complete (avoid extra object creation)

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

@@ -0,0 +1,69 @@
+package kotlinx.coroutines.experimental
+
+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
+
+public interface EventLoop {
+    public val thisEventLoop: CoroutineDispatcher
+    public suspend fun yield()
+}
+
+@Throws(InterruptedException::class)
+public fun <T> runEventLoop(block: suspend EventLoop.() -> T): T =
+    EventLoopImpl<T>().also { block.startCoroutine(it, it.coroutine) }.coroutine.joinBlocking()
+
+private class EventLoopImpl<T> : CoroutineDispatcher(), EventLoop {
+    val thread: Thread = Thread.currentThread()
+    val queue = LockFreeLinkedListHead()
+    val coroutine = Coroutine()
+
+    public override val thisEventLoop: CoroutineDispatcher = this
+
+    public override suspend fun yield(): Unit = suspendCancellableCoroutine { cont ->
+        val node = Resume(cont)
+        schedule(node)
+        cont.removeOnCompletion(node)
+    }
+
+    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)
+    }
+
+    inner class Coroutine : AbstractCoroutine<T>(this@EventLoopImpl) {
+        override fun afterCompletion(state: Any?) {
+            LockSupport.unpark(thread)
+        }
+
+        @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
+        }
+    }
+
+    class Dispatch(block: Runnable) : LockFreeLinkedListNode(), Runnable by block
+
+    class Resume(val cont: Continuation<Unit>) : LockFreeLinkedListNode(), Runnable {
+        override fun run() = cont.resume(Unit)
+    }
+}
+

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

@@ -91,6 +91,9 @@ public fun Job.unregisterOnCompletion(registration: Job.Registration): Job.Regis
 public fun Job.cancelFutureOnCompletion(future: Future<*>): Job.Registration =
     onCompletion(CancelFutureOnCompletion(this, future))
 
+internal fun Job.removeOnCompletion(node: LockFreeLinkedListNode): Job.Registration =
+    onCompletion(RemoveOnCompletion(this, node))
+
 /**
  * Suspends coroutine until this job is complete. This invocation resumes normally (without exception)
  * when the job is complete for any reason.
@@ -140,21 +143,18 @@ public open class JobSupport(
         // #1. Unregister from parent job
         registration?.unregister()
         // #2 Invoke completion handlers
-        var closeException: Throwable? = null
-        val reason = when (update) {
-            is Cancelled -> update.cancelReason
-            is CompletedExceptionally -> update.exception
-            else -> null
-        }
+        val reason = (update as? CompletedExceptionally)?.cancelReason
+        var completionException: Throwable? = null
         expect.forEach<JobNode> { node ->
             try {
                 node.invoke(reason)
             } catch (ex: Throwable) {
-                if (closeException == null) closeException = ex else closeException!!.addSuppressed(ex)
+                completionException?.apply { addSuppressed(ex) } ?: run { completionException = ex }
             }
         }
         // #3 Do other (overridable) processing
-        afterCompletion(update, closeException)
+        completionException?.let { handleCompletionException(it) }
+        afterCompletion(update)
         return true
     }
 
@@ -181,10 +181,18 @@ public open class JobSupport(
         }
     }
 
-    protected open fun afterCompletion(state: Any?, closeException: Throwable?) {
-        if (closeException != null) throw closeException
+    /**
+     * Override to process any exceptions that were encountered while invoking [onCompletion] handlers.
+     */
+    protected open fun handleCompletionException(closeException: Throwable) {
+        throw closeException
     }
 
+    /**
+     * Override for post-completion actions that need to do something with the state.
+     */
+    protected open fun afterCompletion(state: Any?) {}
+
     private fun makeNode(handler: CompletionHandler): JobNode =
             (handler as? JobNode)?.also { require(it.job === this) }
                     ?: InvokeOnCompletion(this, handler)
@@ -272,3 +280,11 @@ private class CancelFutureOnCompletion(
     override fun invoke(reason: Throwable?) { future.cancel(true) }
     override fun toString() = "CancelFutureOnCompletion[$future]"
 }
+
+private class RemoveOnCompletion(
+    job: Job,
+    val node: LockFreeLinkedListNode
+) : JobNode(job)  {
+    override fun invoke(reason: Throwable?) { node.remove() }
+    override fun toString() = "RemoveOnCompletion[$node]"
+}

kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/internal/LockFreeLinkedList.kt

@@ -139,21 +139,29 @@ internal open class LockFreeLinkedListNode {
     }
 
     /**
-     * Removes this node from the list.
+     * Removes this node from the list. Returns `true` when removed successfully.
      */
-    public open fun remove() {
+    public open fun remove(): Boolean {
         while (true) { // lock-free loop on next
             val next = this.next
-            if (next is Removed) return // was already removed -- don't try to help (original thread will take care)
+            if (next is Removed) return false // was already removed -- don't try to help (original thread will take care)
             if (NEXT.compareAndSet(this, next, Removed(next as Node))) {
                 // was removed successfully (linearized remove) -- fixup the list
                 helpDelete()
                 next.helpInsert(prev.unwrap())
-                return
+                return true
             }
         }
     }
 
+    internal fun removeFirstOrNull(): Node? {
+        while (true) { // try to linearize
+            val first = next()
+            if (first == this) return null
+            if (first.remove()) return first
+        }
+    }
+
     private fun markPrev(): Node {
         while (true) { // lock-free loop on prev
             val prev = this.prev
@@ -242,6 +250,8 @@ internal open class LockFreeLinkedListNode {
 }
 
 internal open class LockFreeLinkedListHead : LockFreeLinkedListNode() {
+    public val isEmpty: Boolean get() = next() == this
+
     /**
      * Iterates over all elements in this list of a specified type.
      */

kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/CoroutinesTest.kt

@@ -0,0 +1,52 @@
+package kotlinx.coroutines.experimental
+
+import org.junit.After
+import org.junit.Test
+import java.util.concurrent.atomic.AtomicBoolean
+import java.util.concurrent.atomic.AtomicInteger
+
+class CoroutinesTest {
+    var actionIndex = AtomicInteger()
+    var finished = AtomicBoolean()
+
+    fun expect(index: Int) {
+        val wasIndex = actionIndex.incrementAndGet()
+        check(index == wasIndex) { "Expecting action index $index but it is actually $wasIndex"}
+    }
+
+    fun finish(index: Int) {
+        expect(index)
+        finished.set(true)
+    }
+
+    @After
+    fun onCompletion() {
+        check(finished.get()) { "Expecting that 'finish(...)' was invoked, but it was not" }
+    }
+
+    @Test
+    fun testSimple() = runEventLoop {
+        expect(1)
+        finish(2)
+    }
+
+    @Test
+    fun testYield() = runEventLoop {
+        expect(1)
+        yield() // effectively does nothing, as we don't have other coroutines
+        finish(2)
+    }
+
+    @Test
+    fun testLaunchHereAndYield() = runEventLoop {
+        expect(1)
+        val job = launch(Here) {
+            expect(2)
+            yield()
+            expect(4)
+        }
+        expect(3)
+        job.join()
+        finish(5)
+    }
+}

kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/internal/LockFreeLinkedListTest.kt

@@ -18,13 +18,14 @@ class LockFreeLinkedListTest {
         assertContents(list, 3, 2, 1)
         val n4 = IntNode(4).apply { list.addFirst(this) }
         assertContents(list, 4, 3, 2, 1)
-        n1.remove()
+        assertTrue(n1.remove())
         assertContents(list, 4, 3, 2)
-        n3.remove()
+        assertTrue(n3.remove())
+        assertFalse(n3.remove())
         assertContents(list, 4, 2)
-        n4.remove()
+        assertTrue(n4.remove())
         assertContents(list, 2)
-        n2.remove()
+        assertTrue(n2.remove())
         assertContents(list)
     }
 
@@ -40,13 +41,14 @@ class LockFreeLinkedListTest {
         assertContents(list, 1, 2, 3)
         val n4 = IntNode(4).apply { list.addLast(this) }
         assertContents(list, 1, 2, 3, 4)
-        n1.remove()
+        assertTrue(n1.remove())
         assertContents(list, 2, 3, 4)
-        n3.remove()
+        assertTrue(n3.remove())
         assertContents(list, 2, 4)
-        n4.remove()
+        assertTrue(n4.remove())
         assertContents(list, 2)
-        n2.remove()
+        assertTrue(n2.remove())
+        assertFalse(n2.remove())
         assertContents(list)
     }
 
@@ -72,5 +74,6 @@ class LockFreeLinkedListTest {
         list.forEach<IntNode> { actual[index++] = it.i }
         assertEquals(n, index)
         for (i in 0 until n) assertEquals("item i", expected[i], actual[i])
+        assertEquals(expected.isEmpty(), list.isEmpty)
     }
 }