Dispatchers.Default backed by worker

Author: qwwdfsad

kotlinx-coroutines-core/concurrent/test/DefaultDispatcherConcurrencyTest.kt

@@ -0,0 +1,53 @@
+package kotlinx.coroutines
+
+import kotlinx.coroutines.channels.*
+import kotlin.test.*
+
+/*
+ * Copyright 2016-2021 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+class DefaultDispatcherConcurrencyTest : TestBase() {
+
+    @Test
+    fun testLaunchAndJoin() {
+        expect(1)
+        var capturedMutableState = 0
+        val job = GlobalScope.launch {
+            ++capturedMutableState
+            expect(2)
+        }
+        runBlocking { job.join() }
+        assertEquals(1, capturedMutableState)
+        finish(3)
+    }
+
+    @Test
+    fun testDefaultDispatcherIsActuallyMultithreaded() {
+        val channel = Channel<Int>()
+        GlobalScope.launch {
+            channel.send(42)
+        }
+
+        var result = ChannelResult.failure<Int>()
+        while (!result.isSuccess) {
+            result = channel.tryReceive()
+            // Block the thread, wait
+        }
+        // Delivery was successful, let's check it
+        assertEquals(42, result.getOrThrow())
+    }
+
+    @Test
+    fun testDelayInDefaultDispatcher() {
+        expect(1)
+        val job = GlobalScope.launch {
+            expect(2)
+            delay(100)
+            println("Actually resumed")
+            expect(3)
+        }
+        runBlocking { job.join() }
+        finish(4)
+    }
+}

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

@@ -7,6 +7,7 @@ package kotlinx.coroutines
 import kotlinx.cinterop.*
 import platform.posix.*
 import kotlin.coroutines.*
+import kotlin.native.concurrent.*
 
 /**
  * Runs new coroutine and **blocks** current thread _interruptibly_ until its completion.
@@ -33,7 +34,7 @@ import kotlin.coroutines.*
 public actual fun <T> runBlocking(context: CoroutineContext, block: suspend CoroutineScope.() -> T): T {
     val contextInterceptor = context[ContinuationInterceptor]
     val eventLoop: EventLoop?
-    var newContext: CoroutineContext = context // todo: kludge for data flow analysis error
+    val newContext: CoroutineContext
     if (contextInterceptor == null) {
         // create or use private event loop if no dispatcher is specified
         eventLoop = ThreadLocalEventLoop.eventLoop
@@ -57,24 +58,21 @@ private class BlockingCoroutine<T>(
     override val isScopedCoroutine: Boolean get() = true
 
     @Suppress("UNCHECKED_CAST")
-    fun joinBlocking(): T = memScoped {
+    fun joinBlocking(): T {
         try {
             eventLoop?.incrementUseCount()
-            val timespec = alloc<timespec>()
             while (true) {
                 val parkNanos = eventLoop?.processNextEvent() ?: Long.MAX_VALUE
                 // note: process next even may loose unpark flag, so check if completed before parking
                 if (isCompleted) break
-                timespec.tv_sec = (parkNanos / 1000000000L).convert() // 1e9 ns -> sec
-                timespec.tv_nsec = (parkNanos % 1000000000L).convert() // % 1e9
-                nanosleep(timespec.ptr, null)
+                Worker.current.park(parkNanos / 1000L)
             }
         } finally { // paranoia
             eventLoop?.decrementUseCount()
         }
         // now return result
         val state = state.unboxState()
         (state as? CompletedExceptionally)?.let { throw it.cause }
-        state as T
+        return state as T
     }
 }

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

@@ -8,19 +8,27 @@ import kotlinx.coroutines.internal.*
 import kotlin.coroutines.*
 import kotlin.native.concurrent.*
 
-private fun takeEventLoop(): EventLoopImpl =
-    ThreadLocalEventLoop.currentOrNull() as? EventLoopImpl ?:
-        error("There is no event loop. Use runBlocking { ... } to start one.")
 
 internal actual object DefaultExecutor : CoroutineDispatcher(), Delay {
+
+    private val worker = Worker.start(name = "Dispatchers.Default")
+
     override fun dispatch(context: CoroutineContext, block: Runnable) =
-        takeEventLoop().dispatch(context, block)
-    override fun scheduleResumeAfterDelay(timeMillis: Long, continuation: CancellableContinuation<Unit>) =
-        takeEventLoop().scheduleResumeAfterDelay(timeMillis, continuation)
-    override fun invokeOnTimeout(timeMillis: Long, block: Runnable, context: CoroutineContext): DisposableHandle =
-        takeEventLoop().invokeOnTimeout(timeMillis, block, context)
+        worker.executeAfter(0L) { block.run() }
+
+    override fun scheduleResumeAfterDelay(timeMillis: Long, continuation: CancellableContinuation<Unit>) {
+        // TODO proper toMicros
+        worker.executeAfter(timeMillis * 1000)
+        { with(continuation) { resumeUndispatched(Unit) } }
+    }
+
+    override fun invokeOnTimeout(timeMillis: Long, block: Runnable, context: CoroutineContext): DisposableHandle {
+        // No API to cancel on timeout
+        worker.executeAfter(timeMillis * 1000) { block.run() }
+        return NonDisposableHandle
+    }
 
-    actual fun enqueue(task: Runnable): Unit = loopWasShutDown()
+    actual fun enqueue(task: Runnable): Unit = worker.executeAfter(0L) { task.run() }
 }
 
 internal fun loopWasShutDown(): Nothing = error("Cannot execute task because event loop was shut down")

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

@@ -4,11 +4,21 @@
 
 package kotlinx.coroutines
 
+import kotlinx.cinterop.*
+import platform.posix.*
 import kotlin.coroutines.*
 import kotlin.system.*
+import kotlin.native.concurrent.*
 
-internal actual abstract class EventLoopImplPlatform: EventLoop() {
-    protected actual fun unpark() { /* does nothing */ }
+internal actual abstract class EventLoopImplPlatform : EventLoop() {
+
+    private val current = Worker.current
+
+    protected actual fun unpark() {
+        current.execute(TransferMode.SAFE, {}) {} // send an empty task to unpark the waiting event loop
+    }
+
+    // TODO verify loop was shut down is an okay behaviour
     protected actual fun reschedule(now: Long, delayedTask: EventLoopImplBase.DelayedTask): Unit =
         loopWasShutDown()
 }

kotlinx-coroutines-core/native/test/DelayExceptionTest.kt

@@ -8,22 +8,6 @@ import kotlin.coroutines.*
 import kotlin.test.*
 
 class DelayExceptionTest : TestBase() {
-    private object Dispatcher : CoroutineDispatcher() {
-        override fun isDispatchNeeded(context: CoroutineContext): Boolean = true
-        override fun dispatch(context: CoroutineContext, block: Runnable) { block.run() }
-    }
-
-    private lateinit var exception: Throwable
-
-
-    @Test
-    fun testThrowsTce() {
-        CoroutineScope(Dispatcher + CoroutineExceptionHandler { _, e -> exception = e }).launch {
-            delay(10)
-        }
-
-        assertTrue(exception is IllegalStateException)
-    }
 
     @Test
     fun testMaxDelay() = runBlocking {

kotlinx-coroutines-core/native/test/WorkerTest.kt

@@ -23,12 +23,12 @@ class WorkerTest : TestBase() {
     }
 
     @Test
-    fun testLaunchInWorkerTroughGlobalScope() {
+    fun testLaunchInWorkerThroughGlobalScope() {
         val worker = Worker.start()
         worker.execute(TransferMode.SAFE, { }) {
             runBlocking {
                 CoroutineScope(EmptyCoroutineContext).launch {
-                    delay(1)
+                    delay(10)
                 }.join()
             }
         }.result