Added context parameter to CoroutineDispatcher methods, implemented Executor.toCoroutineDispatcher

Author: elizarov

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

@@ -4,14 +4,15 @@ import java.util.concurrent.Executor
 import java.util.concurrent.Executors
 import java.util.concurrent.ForkJoinPool
 import java.util.concurrent.atomic.AtomicInteger
+import kotlin.coroutines.CoroutineContext
 
 /**
  * Represents common pool of shared threads as coroutine dispatcher for compute-intensive tasks.
  * It uses [ForkJoinPool] when available, which implements efficient work-stealing algorithm for its queues, so every
  * 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(), Yield {
+object CommonPool : CoroutineDispatcher() {
     private val pool: Executor = findPool()
 
     private inline fun <T> Try(block: () -> T) = try { block() } catch (e: Throwable) { null }
@@ -35,10 +36,6 @@ object CommonPool : CoroutineDispatcher(), Yield {
 
     private fun defaultParallelism() = (Runtime.getRuntime().availableProcessors() - 1).coerceAtLeast(1)
 
-    override fun isDispatchNeeded(): Boolean = true
-    override fun dispatch(block: Runnable) = pool.execute(block)
-
-    override fun scheduleResume(continuation: CancellableContinuation<Unit>) {
-        pool.execute { continuation.resume(Unit) }
-    }
+    override fun isDispatchNeeded(context: CoroutineContext): Boolean = true
+    override fun dispatch(context: CoroutineContext, block: Runnable) = pool.execute(block)
 }

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

@@ -3,6 +3,7 @@ package kotlinx.coroutines.experimental
 import kotlin.coroutines.AbstractCoroutineContextElement
 import kotlin.coroutines.Continuation
 import kotlin.coroutines.ContinuationInterceptor
+import kotlin.coroutines.CoroutineContext
 
 /**
  * Base class that shall be extended by all coroutine dispatcher implementations.
@@ -30,12 +31,12 @@ public abstract class CoroutineDispatcher :
     /**
      * Return `true` if execution shall be dispatched onto another thread.
      */
-    public abstract fun isDispatchNeeded(): Boolean
+    public abstract fun isDispatchNeeded(context: CoroutineContext): Boolean
 
     /**
-     * Dispatches execution of a runnable [block] onto another thread.
+     * Dispatches execution of a runnable [block] onto another thread in the given [context].
      */
-    public abstract fun dispatch(block: Runnable)
+    public abstract fun dispatch(context: CoroutineContext, block: Runnable)
 
     override fun <T> interceptContinuation(continuation: Continuation<T>): Continuation<T> =
             DispatchedContinuation<T>(this, continuation)
@@ -46,27 +47,29 @@ private class DispatchedContinuation<T>(
         val continuation: Continuation<T>
 ): Continuation<T> by continuation {
     override fun resume(value: T) {
-        if (dispatcher.isDispatchNeeded())
-            dispatcher.dispatch(Runnable {
-                withDefaultCoroutineContext(continuation.context) {
+        val context = continuation.context
+        if (dispatcher.isDispatchNeeded(context))
+            dispatcher.dispatch(context, Runnable {
+                withDefaultCoroutineContext(context) {
                     continuation.resume(value)
                 }
             })
         else
-            withDefaultCoroutineContext(continuation.context) {
+            withDefaultCoroutineContext(context) {
                 continuation.resume(value)
             }
     }
 
     override fun resumeWithException(exception: Throwable) {
-        if (dispatcher.isDispatchNeeded())
-            dispatcher.dispatch(Runnable {
-                withDefaultCoroutineContext(continuation.context) {
+        val context = continuation.context
+        if (dispatcher.isDispatchNeeded(context))
+            dispatcher.dispatch(context, Runnable {
+                withDefaultCoroutineContext(context) {
                     continuation.resumeWithException(exception)
                 }
             })
         else
-            withDefaultCoroutineContext(continuation.context) {
+            withDefaultCoroutineContext(context) {
                 continuation.resumeWithException(exception)
             }
     }

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

@@ -19,8 +19,8 @@ internal val CURRENT_CONTEXT = ThreadLocal<CoroutineContext>()
  * mandating any specific threading policy.
  */
 public object Here : CoroutineDispatcher() {
-    override fun isDispatchNeeded(): Boolean = false
-    override fun dispatch(block: Runnable) { throw UnsupportedOperationException() }
+    override fun isDispatchNeeded(context: CoroutineContext): Boolean = false
+    override fun dispatch(context: CoroutineContext, block: Runnable) { throw UnsupportedOperationException() }
 }
 
 /**

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

@@ -43,7 +43,6 @@ suspend fun delay(time: Long, unit: TimeUnit = TimeUnit.MILLISECONDS) {
             scheduleResumeAfterDelay(time, unit, cont)
             return@sc
         }
-        val timeout = scheduledExecutor.schedule({ cont.resume(Unit) }, time, unit)
-        cont.cancelFutureOnCompletion(timeout)
+        scheduledExecutor.scheduleResumeAfterDelay(time, unit, cont)
     }
 }

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

@@ -4,6 +4,7 @@ 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.CoroutineContext
 
 /**
  * Implemented by [CoroutineDispatcher] implementations that have event loop inside and can
@@ -49,9 +50,9 @@ internal class EventLoopImpl(
         this.parentJob = coroutine
     }
 
-    override fun isDispatchNeeded(): Boolean = Thread.currentThread() != thread
+    override fun isDispatchNeeded(context: CoroutineContext): Boolean = Thread.currentThread() != thread
 
-    override fun dispatch(block: Runnable) {
+    override fun dispatch(context: CoroutineContext, block: Runnable) {
         schedule(Dispatch(block))
     }
 

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

@@ -0,0 +1,32 @@
+package kotlinx.coroutines.experimental
+
+import java.util.concurrent.Executor
+import java.util.concurrent.ExecutorService
+import java.util.concurrent.ScheduledExecutorService
+import java.util.concurrent.TimeUnit
+import kotlin.coroutines.CoroutineContext
+
+/**
+ * Converts an instance of [Executor] to an implementation of [CoroutineDispatcher].
+ */
+public fun Executor.toCoroutineDispatcher(): CoroutineDispatcher =
+    ExecutorCoroutineDispatcher(this)
+
+internal open class ExecutorCoroutineDispatcher(val executor: Executor) : CoroutineDispatcher(), Delay {
+    override fun isDispatchNeeded(context: CoroutineContext): Boolean = true
+    override fun dispatch(context: CoroutineContext, block: Runnable) = executor.execute(block)
+
+    override fun scheduleResumeAfterDelay(time: Long, unit: TimeUnit, continuation: CancellableContinuation<Unit>) {
+        (executor as? ScheduledExecutorService ?: scheduledExecutor).scheduleResumeAfterDelay(time, unit, continuation)
+    }
+}
+
+internal fun ExecutorService.scheduleResume(cont: CancellableContinuation<Unit>) {
+    val future = submit { cont.resume(Unit) }
+    cont.cancelFutureOnCompletion(future)
+}
+
+internal fun ScheduledExecutorService.scheduleResumeAfterDelay(time: Long, unit: TimeUnit, cont: CancellableContinuation<Unit>) {
+    val timeout = schedule({ cont.resume(Unit) }, time, unit)
+    cont.cancelFutureOnCompletion(timeout)
+}

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

@@ -5,11 +5,10 @@ import java.util.concurrent.ScheduledExecutorService
 import java.util.concurrent.TimeUnit
 import java.util.concurrent.atomic.AtomicInteger
 import kotlin.concurrent.thread
-import kotlin.coroutines.ContinuationInterceptor
 import kotlin.coroutines.CoroutineContext
 
 /**
- * Creates new coroutine execution context with the a single thread and built-in [delay] support.
+ * Creates new coroutine execution context with the a single thread and built-in [yield] and [delay] support.
  * All continuations are dispatched immediately when invoked inside the thread of this context.
  * Resources of this pool (its thread) are reclaimed when job of this context is cancelled.
  * The specified [name] defines the name of the new thread.
@@ -19,16 +18,16 @@ fun newSingleThreadContext(name: String, parent: Job? = null): CoroutineContext
     newFixedThreadPoolContext(1, name, parent)
 
 /**
- * Creates new coroutine execution context with the fixed-size thread-pool and built-in [delay] support.
+ * Creates new coroutine execution context with the fixed-size thread-pool and built-in [yield] and [delay] support.
  * All continuations are dispatched immediately when invoked inside the threads of this context.
  * Resources of this pool (its threads) are reclaimed when job of this context is cancelled.
  * The specified [name] defines the names of the threads.
  * An optional [parent] job may be specified upon creation.
  */
 fun newFixedThreadPoolContext(nThreads: Int, name: String, parent: Job? = null): CoroutineContext {
     require(nThreads >= 1) { "Expected at least one thread, but $nThreads specified" }
-    val lifetime = Job(parent)
-    return lifetime + ThreadPoolDispatcher(nThreads, name, lifetime)
+    val job = Job(parent)
+    return job + ThreadPoolDispatcher(nThreads, name, job)
 }
 
 private val thisThreadContext = ThreadLocal<ThreadPoolDispatcher>()
@@ -37,7 +36,7 @@ private class ThreadPoolDispatcher(
         nThreads: Int,
         name: String,
         val job: Job
-) : CoroutineDispatcher(), ContinuationInterceptor, Yield, Delay {
+) : CoroutineDispatcher(), Yield, Delay {
     val threadNo = AtomicInteger()
     val executor: ScheduledExecutorService = Executors.newScheduledThreadPool(nThreads) { target ->
         thread(start = false, isDaemon = true,
@@ -51,16 +50,15 @@ private class ThreadPoolDispatcher(
         job.onCompletion { executor.shutdown() }
     }
 
-    override fun isDispatchNeeded(): Boolean = thisThreadContext.get() != this
+    override fun isDispatchNeeded(context: CoroutineContext): Boolean = thisThreadContext.get() != this
 
-    override fun dispatch(block: Runnable) = executor.execute(block)
+    override fun dispatch(context: CoroutineContext, block: Runnable) = executor.execute(block)
 
     override fun scheduleResume(continuation: CancellableContinuation<Unit>) {
-        executor.execute { continuation.resume(Unit) }
+        executor.scheduleResume(continuation)
     }
 
     override fun scheduleResumeAfterDelay(time: Long, unit: TimeUnit, continuation: CancellableContinuation<Unit>) {
-        val timeout = executor.schedule({ continuation.resume(Unit) }, time, unit)
-        continuation.cancelFutureOnCompletion(timeout)
+        executor.scheduleResumeAfterDelay(time, unit, continuation)
     }
 }

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

@@ -4,7 +4,8 @@ import kotlin.coroutines.ContinuationInterceptor
 
 /**
  * This dispatcher _feature_ is implemented by [CoroutineDispatcher] implementations that
- * natively support [yield] function.
+ * natively support [yield] function. It shall be implemented only by [CoroutineDispatcher]
+ * classes with non-trivial [CoroutineDispatcher.isDispatchNeeded] implementations.
  */
 public interface Yield {
     /**

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

@@ -0,0 +1,85 @@
+package kotlinx.coroutines.experimental
+
+import org.junit.After
+import org.junit.Before
+import org.junit.Test
+import java.util.concurrent.Executors
+
+class ExecutorsTest {
+    fun threadNames(): Set<String> {
+        val arrayOfThreads = Array<Thread?>(Thread.activeCount()) { null }
+        val n = Thread.enumerate(arrayOfThreads)
+        val names = hashSetOf<String>()
+        for (i in 0 until n)
+            names.add(arrayOfThreads[i]!!.name)
+        return names
+    }
+
+    lateinit var threadNamesBefore: Set<String>
+
+    @Before
+    fun before() {
+        threadNamesBefore = threadNames()
+    }
+
+    @After
+    fun after() {
+        // give threads some time to shutdown
+        val waitTill = System.currentTimeMillis() + 1000L
+        var diff: Set<String>
+        do {
+            val threadNamesAfter = threadNames()
+            diff = threadNamesAfter - threadNamesBefore
+            if (diff.isEmpty()) break
+        } while (System.currentTimeMillis() <= waitTill)
+        diff.forEach { println("Lost thread '$it'") }
+        check(diff.isEmpty()) { "Lost ${diff.size} threads"}
+    }
+
+    fun checkThreadName(prefix: String) {
+        val name = Thread.currentThread().name
+        check(name.startsWith(prefix)) { "Expected thread name to start with '$prefix', found: '$name'" }
+    }
+
+    @Test
+    fun testSingleThread() {
+        val context = newSingleThreadContext("TestThread")
+        runBlocking(context) {
+            checkThreadName("TestThread")
+        }
+        context[Job]!!.cancel()
+    }
+
+    @Test
+    fun testFixedThreadPool() {
+        val context = newFixedThreadPoolContext(2, "TestPool")
+        runBlocking(context) {
+            checkThreadName("TestPool")
+        }
+        context[Job]!!.cancel()
+    }
+
+    @Test
+    fun testToExecutor() {
+        val executor = Executors.newSingleThreadExecutor { r -> Thread(r, "TestExecutor") }
+        runBlocking(executor.toCoroutineDispatcher()) {
+            checkThreadName("TestExecutor")
+        }
+        executor.shutdown()
+    }
+
+    @Test
+    fun testTwoThreads() {
+        val ctx1 = newSingleThreadContext("Ctx1")
+        val ctx2 = newSingleThreadContext("Ctx2")
+        runBlocking(ctx1) {
+            checkThreadName("Ctx1")
+            run(ctx2) {
+                checkThreadName("Ctx2")
+            }
+            checkThreadName("Ctx1")
+        }
+        ctx1[Job]!!.cancel()
+        ctx2[Job]!!.cancel()
+    }
+}

kotlinx-coroutines-javafx/src/main/kotlin/kotlinx/coroutines/experimental/javafx/JavaFx.kt

@@ -11,6 +11,7 @@ import kotlinx.coroutines.experimental.*
 import kotlinx.coroutines.experimental.javafx.JavaFx.delay
 import java.util.concurrent.CopyOnWriteArrayList
 import java.util.concurrent.TimeUnit
+import kotlin.coroutines.CoroutineContext
 
 
 /**
@@ -21,8 +22,8 @@ object JavaFx : CoroutineDispatcher(), Yield, Delay {
         PulseTimer().apply { start() }
     }
 
-    override fun isDispatchNeeded(): Boolean = !Platform.isFxApplicationThread()
-    override fun dispatch(block: Runnable) = Platform.runLater(block)
+    override fun isDispatchNeeded(context: CoroutineContext): Boolean = !Platform.isFxApplicationThread()
+    override fun dispatch(context: CoroutineContext, block: Runnable) = Platform.runLater(block)
 
     /**
      * Suspends coroutine until next JavaFx pulse and returns time of the pulse on resumption.