(WIP) EventLoop integration

Fixes #860

Author: elizarov

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

@@ -20,7 +20,12 @@ internal object UndispatchedEventLoop {
     )
 
     @JvmField
-    internal val threadLocalEventLoop = CommonThreadLocal { EventLoop() }
+    internal val threadLocalEventLoop = CommonThreadLocal<EventLoop?>()
+
+    private val eventLoop: EventLoop
+        get() = threadLocalEventLoop.get() ?: EventLoop().also {
+            threadLocalEventLoop.set(it)
+        }
 
     /**
      * Executes given [block] as part of current event loop, updating related to block [continuation]
@@ -30,7 +35,7 @@ internal object UndispatchedEventLoop {
      */
     inline fun execute(continuation: DispatchedContinuation<*>, contState: Any?, mode: Int,
                        doYield: Boolean = false, block: () -> Unit) : Boolean {
-        val eventLoop = threadLocalEventLoop.get()
+        val eventLoop = eventLoop
         if (eventLoop.isActive) {
             // If we are yielding and queue is empty, we can bail out as part of fast path
             if (doYield && eventLoop.queue.isEmpty) {
@@ -48,13 +53,12 @@ internal object UndispatchedEventLoop {
     }
 
     fun resumeUndispatched(task: DispatchedTask<*>): Boolean {
-        val eventLoop = threadLocalEventLoop.get()
+        val eventLoop = eventLoop
         if (eventLoop.isActive) {
             eventLoop.queue.addLast(task)
             return true
         }
-
-        runEventLoop(eventLoop, { task.resume(task.delegate, MODE_UNDISPATCHED) })
+        runEventLoop(eventLoop) { task.resume(task.delegate, MODE_UNDISPATCHED) }
         return false
     }
 

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

@@ -0,0 +1,38 @@
+/*
+ * Copyright 2016-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package kotlinx.coroutines
+
+import kotlinx.coroutines.internal.*
+import kotlin.coroutines.*
+import kotlin.jvm.*
+
+/**
+ * Implemented by [CoroutineDispatcher] implementations that have event loop inside and can
+ * be asked to process next event from their event queue.
+ *
+ * It may optionally implement [Delay] interface and support time-scheduled tasks. It is used by [runBlocking] to
+ * continue processing events when invoked from the event dispatch thread.
+ *
+ * @suppress **This an internal API and should not be used from general code.**
+ */
+internal interface EventLoop: ContinuationInterceptor {
+    /**
+     * Processes next event in this event loop.
+     *
+     * The result of this function is to be interpreted like this:
+     * * `<= 0` -- there are potentially more events for immediate processing;
+     * * `> 0` -- a number of nanoseconds to wait for next scheduled event;
+     * * [Long.MAX_VALUE] -- no more events, or was invoked from the wrong thread.
+     */
+    public fun processNextEvent(): Long
+}
+
+@NativeThreadLocal
+internal object ThreadEventLoop {
+    @JvmField
+    internal val ref = CommonThreadLocal<EventLoop?>()
+}
+
+

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

@@ -8,6 +8,7 @@ package kotlinx.coroutines.internal
 @UseExperimental(ExperimentalMultiplatform::class)
 internal expect annotation class NativeThreadLocal()
 
-internal expect class CommonThreadLocal<T>(supplier: () -> T) {
+internal expect class CommonThreadLocal<T>() {
     fun get(): T
+    fun set(value: T)
 }

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

@@ -35,35 +35,42 @@ import kotlin.coroutines.*
 public fun <T> runBlocking(context: CoroutineContext = EmptyCoroutineContext, block: suspend CoroutineScope.() -> T): T {
     val currentThread = Thread.currentThread()
     val contextInterceptor = context[ContinuationInterceptor]
-    val privateEventLoop = contextInterceptor == null // create private event loop if no dispatcher is specified
-    val eventLoop = if (privateEventLoop) BlockingEventLoop(currentThread) else contextInterceptor as? EventLoop
-    val newContext = GlobalScope.newCoroutineContext(
-        if (privateEventLoop) context + (eventLoop as ContinuationInterceptor) else context
-    )
-    val coroutine = BlockingCoroutine<T>(newContext, currentThread, eventLoop, privateEventLoop)
+    val eventLoop: EventLoop?
+    val topLevelEventLoop: Boolean
+    val newContext: CoroutineContext
+    if (contextInterceptor == null) {
+        // create/use private event loop if no dispatcher is specified
+        val currentEventLoop: EventLoop? = ThreadEventLoop.ref.get()
+        val usedEventEventLoop = currentEventLoop ?: BlockingEventLoop(currentThread).also {
+            ThreadEventLoop.ref.set(it)
+        }
+        eventLoop = usedEventEventLoop
+        topLevelEventLoop = currentEventLoop == null
+        newContext = GlobalScope.newCoroutineContext(context + usedEventEventLoop)
+    } else {
+        eventLoop = contextInterceptor as? EventLoop
+        topLevelEventLoop = false
+        newContext = GlobalScope.newCoroutineContext(context)
+    }
+    val coroutine = BlockingCoroutine<T>(newContext, currentThread, eventLoop)
     coroutine.start(CoroutineStart.DEFAULT, coroutine, block)
-    return coroutine.joinBlocking()
+    return coroutine.joinBlocking(topLevelEventLoop)
 }
 
 private class BlockingCoroutine<T>(
     parentContext: CoroutineContext,
     private val blockedThread: Thread,
-    private val eventLoop: EventLoop?,
-    private val privateEventLoop: Boolean
+    private val eventLoop: EventLoop?
 ) : AbstractCoroutine<T>(parentContext, true) {
-    init {
-        if (privateEventLoop) require(eventLoop is BlockingEventLoop)
-    }
-
     override fun onCompletionInternal(state: Any?, mode: Int, suppressed: Boolean) {
         // wake up blocked thread
         if (Thread.currentThread() != blockedThread)
             LockSupport.unpark(blockedThread)
     }
 
     @Suppress("UNCHECKED_CAST")
-    fun joinBlocking(): T {
-        timeSource.registerTimeLoopThread()
+    fun joinBlocking(topLevelEventLoop: Boolean): T {
+        if (topLevelEventLoop) timeSource.registerTimeLoopThread()
         while (true) {
             @Suppress("DEPRECATION")
             if (Thread.interrupted()) throw InterruptedException().also { cancel(it) }
@@ -72,14 +79,18 @@ private class BlockingCoroutine<T>(
             if (isCompleted) break
             timeSource.parkNanos(this, parkNanos)
         }
-        // process queued events (that could have been added after last processNextEvent and before cancel
-        if (privateEventLoop) (eventLoop as BlockingEventLoop).apply {
-            // We exit the "while" loop above when this coroutine's state "isCompleted",
-            // Here we should signal that BlockingEventLoop should not accept any more tasks
-            isCompleted = true
-            shutdown()
+        if (topLevelEventLoop) {
+            // Clean up reference here -- this event loop is shutting down
+            ThreadEventLoop.ref.set(null)
+            // process queued events (that could have been added after last processNextEvent and before cancel
+            (eventLoop as BlockingEventLoop).apply {
+                // We exit the "while" loop above when this coroutine's state "isCompleted",
+                // Here we should signal that BlockingEventLoop should not accept any more tasks
+                isCompleted = true
+                shutdown()
+            }
+            timeSource.unregisterTimeLoopThread()
         }
-        timeSource.unregisterTimeLoopThread()
         // now return result
         val state = this.state.unboxState()
         (state as? CompletedExceptionally)?.let { throw it.cause }

core/kotlinx-coroutines-core/src/DefaultExecutor.kt

@@ -10,7 +10,6 @@ internal actual val DefaultDelay: Delay = DefaultExecutor
 
 @Suppress("PLATFORM_CLASS_MAPPED_TO_KOTLIN")
 internal object DefaultExecutor : EventLoopBase(), Runnable {
-
     override val isCompleted: Boolean get() = false
 
     private const val DEFAULT_KEEP_ALIVE = 1000L // in milliseconds
@@ -26,6 +25,9 @@ internal object DefaultExecutor : EventLoopBase(), Runnable {
     @Volatile
     private var _thread: Thread? = null
 
+    override val thread: Thread
+        get() = _thread ?: createThreadSync()
+
     private const val FRESH = 0
     private const val ACTIVE = 1
     private const val SHUTDOWN_REQ = 2
@@ -81,13 +83,10 @@ internal object DefaultExecutor : EventLoopBase(), Runnable {
             acknowledgeShutdownIfNeeded()
             timeSource.unregisterTimeLoopThread()
             // recheck if queues are empty after _thread reference was set to null (!!!)
-            if (!isEmpty) thread() // recreate thread if it is needed
+            if (!isEmpty) thread // recreate thread if it is needed
         }
     }
 
-    // ensure that thread is there
-    private fun thread(): Thread = _thread ?: createThreadSync()
-
     @Synchronized
     private fun createThreadSync() = _thread ?:
         Thread(this, "kotlinx.coroutines.DefaultExecutor").apply {
@@ -96,12 +95,6 @@ internal object DefaultExecutor : EventLoopBase(), Runnable {
             start()
         }
 
-    override fun unpark() {
-        timeSource.unpark(thread()) // as a side effect creates thread if it is not there
-    }
-
-    override fun isCorrectThread(): Boolean = true
-
     // used for tests
     @Synchronized
     internal fun ensureStarted() {

core/kotlinx-coroutines-core/src/EventLoop.kt

@@ -8,27 +8,6 @@ import kotlinx.atomicfu.*
 import kotlinx.coroutines.internal.*
 import kotlin.coroutines.*
 
-/**
- * Implemented by [CoroutineDispatcher] implementations that have event loop inside and can
- * be asked to process next event from their event queue.
- *
- * It may optionally implement [Delay] interface and support time-scheduled tasks. It is used by [runBlocking] to
- * continue processing events when invoked from the event dispatch thread.
- *
- * @suppress **This an internal API and should not be used from general code.**
- */
-internal interface EventLoop: ContinuationInterceptor {
-    /**
-     * Processes next event in this event loop.
-     *
-     * The result of this function is to be interpreted like this:
-     * * `<= 0` -- there are potentially more events for immediate processing;
-     * * `> 0` -- a number of nanoseconds to wait for next scheduled event;
-     * * [Long.MAX_VALUE] -- no more events, or was invoked from the wrong thread.
-     */
-    public fun processNextEvent(): Long
-}
-
 private val DISPOSED_TASK = Symbol("REMOVED_TASK")
 
 // results for scheduleImpl
@@ -61,8 +40,8 @@ internal abstract class EventLoopBase: CoroutineDispatcher(), Delay, EventLoop {
     private val _delayed = atomic<ThreadSafeHeap<DelayedTask>?>(null)
 
     protected abstract val isCompleted: Boolean
-    protected abstract fun unpark()
-    protected abstract fun isCorrectThread(): Boolean
+
+    protected abstract val thread: Thread
 
     protected val isEmpty: Boolean
         get() = isQueueEmpty && isDelayedEmpty
@@ -95,14 +74,20 @@ internal abstract class EventLoopBase: CoroutineDispatcher(), Delay, EventLoop {
             return (nextDelayedTask.nanoTime - timeSource.nanoTime()).coerceAtLeast(0)
         }
 
+    private fun unpark() {
+        val thread = thread
+        if (Thread.currentThread() !== thread)
+            timeSource.unpark(thread)
+    }
+
     override fun dispatch(context: CoroutineContext, block: Runnable) =
         execute(block)
 
     override fun scheduleResumeAfterDelay(timeMillis: Long, continuation: CancellableContinuation<Unit>) =
         schedule(DelayedResumeTask(timeMillis, continuation))
 
     override fun processNextEvent(): Long {
-        if (!isCorrectThread()) return Long.MAX_VALUE
+        if (Thread.currentThread() !== thread) return Long.MAX_VALUE
         // queue all delayed tasks that are due to be executed
         val delayed = _delayed.value
         if (delayed != null && !delayed.isEmpty) {
@@ -318,27 +303,17 @@ internal abstract class EventLoopBase: CoroutineDispatcher(), Delay, EventLoop {
     }
 }
 
-internal abstract class ThreadEventLoop(
-    private val thread: Thread
+internal class BlockingEventLoop(
+    override val thread: Thread
 ) : EventLoopBase() {
-    override fun isCorrectThread(): Boolean = Thread.currentThread() === thread
-
-    override fun unpark() {
-        if (Thread.currentThread() !== thread)
-            timeSource.unpark(thread)
-    }
+    @Volatile
+    public override var isCompleted: Boolean = false
 
     fun shutdown() {
         closeQueue()
-        assert(isCorrectThread())
         // complete processing of all queued tasks
         while (processNextEvent() <= 0) { /* spin */ }
         // reschedule the rest of delayed tasks
         rescheduleAllDelayed()
     }
-}
-
-internal class BlockingEventLoop(thread: Thread) : ThreadEventLoop(thread) {
-    @Volatile
-    public override var isCompleted: Boolean = false
-}
+}

core/kotlinx-coroutines-core/src/internal/ThreadLocal.kt

@@ -6,8 +6,5 @@ package kotlinx.coroutines.internal
 
 import java.lang.ThreadLocal
 
-internal actual typealias CommonThreadLocal<T> = ThreadLocalWithInitialValue<T>
-
-internal class ThreadLocalWithInitialValue<T>(private val supplier: () -> T) : ThreadLocal<T>() {
-    override fun initialValue(): T = supplier()
-}
+@Suppress("ACTUAL_WITHOUT_EXPECT") // internal visibility
+internal actual typealias CommonThreadLocal<T> = ThreadLocal<T>

core/kotlinx-coroutines-core/test/EventLoopsTest.kt

@@ -0,0 +1,20 @@
+/*
+ * Copyright 2016-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package kotlinx.coroutines
+
+import kotlinx.coroutines.channels.*
+import org.junit.*
+
+class EventLoopsTest {
+    @Test
+    fun testNestedRunBlocking() {
+        runBlocking { // outer event loop
+            // Produce string "OK"
+            val ch = produce { send("OK") }
+            // try receive this string in a blocking way:
+            println(runBlocking { ch.receive() }) // it hangs here !!!
+        }
+    }
+}

js/kotlinx-coroutines-core-js/src/internal/ThreadLocal.kt

@@ -4,7 +4,9 @@
 
 package kotlinx.coroutines.internal
 
-internal actual class CommonThreadLocal<T> actual constructor(supplier: () -> T) {
-    private val value = supplier()
-    actual fun get(): T = value
+internal actual class CommonThreadLocal<T> actual constructor() {
+    private var value: T? = null
+    @Suppress("UNCHECKED_CAST")
+    actual fun get(): T = value as T
+    actual fun set(value: T) { this.value = value }
 }

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

@@ -11,25 +11,6 @@ import platform.posix.*
 import kotlin.coroutines.*
 import kotlin.system.*
 
-/**
- * Implemented by [CoroutineDispatcher] implementations that have event loop inside and can
- * be asked to process next event from their event queue.
- *
- * It may optionally implement [Delay] interface and support time-scheduled tasks. It is used by [runBlocking] to
- * continue processing events when invoked from the event dispatch thread.
- */
-internal interface EventLoop {
-    /**
-     * Processes next event in this event loop.
-     *
-     * The result of this function is to be interpreted like this:
-     * * `<= 0` -- there are potentially more events for immediate processing;
-     * * `> 0` -- a number of nanoseconds to wait for next scheduled event;
-     * * [Long.MAX_VALUE] -- no more events, or was invoked from the wrong thread.
-     */
-    public fun processNextEvent(): Long
-}
-
 /**
  * Creates a new event loop.
  */

native/kotlinx-coroutines-core-native/src/internal/ThreadLocal.kt

@@ -8,7 +8,9 @@ import kotlin.native.concurrent.*
 @Suppress("ACTUAL_WITHOUT_EXPECT")
 internal actual typealias NativeThreadLocal = kotlin.native.ThreadLocal
 
-internal actual class CommonThreadLocal<T> actual constructor(supplier: () -> T) {
-    private val value = supplier()
-    actual fun get(): T = value
-}
+internal actual class CommonThreadLocal<T> actual constructor() {
+    private var value: T? = null
+    @Suppress("UNCHECKED_CAST")
+    actual fun get(): T = value as T
+    actual fun set(value: T) { this.value = value }
+}