Introduce ExecutorCoroutineDispatcher instead of CloseableCoroutineDispatcher

Fixes #385

Author: qwwdfsad

binary-compatibility-validator/reference-public-api/kotlinx-coroutines-core.txt

@@ -86,10 +86,12 @@ public abstract class kotlinx/coroutines/experimental/CloseableCoroutineDispatch
 	public fun <init> ()V
 }
 
-public final class kotlinx/coroutines/experimental/CommonPool : kotlinx/coroutines/experimental/CoroutineDispatcher {
+public final class kotlinx/coroutines/experimental/CommonPool : kotlinx/coroutines/experimental/ExecutorCoroutineDispatcher {
 	public static final field DEFAULT_PARALLELISM_PROPERTY_NAME Ljava/lang/String;
 	public static final field INSTANCE Lkotlinx/coroutines/experimental/CommonPool;
+	public fun close ()V
 	public fun dispatch (Lkotlin/coroutines/experimental/CoroutineContext;Ljava/lang/Runnable;)V
+	public fun getExecutor ()Ljava/util/concurrent/Executor;
 	public fun toString ()Ljava/lang/String;
 }
 
@@ -269,12 +271,6 @@ public final class kotlinx/coroutines/experimental/DispatchedTask$DefaultImpls {
 	public static fun run (Lkotlinx/coroutines/experimental/DispatchedTask;)V
 }
 
-public final class kotlinx/coroutines/experimental/DisposableFutureHandle : kotlinx/coroutines/experimental/DisposableHandle {
-	public fun <init> (Ljava/util/concurrent/Future;)V
-	public fun dispose ()V
-	public fun toString ()Ljava/lang/String;
-}
-
 public abstract interface class kotlinx/coroutines/experimental/DisposableHandle {
 	public abstract fun dispose ()V
 }
@@ -294,7 +290,12 @@ public final class kotlinx/coroutines/experimental/EventLoopKt {
 	public static synthetic fun EventLoop$default (Ljava/lang/Thread;Lkotlinx/coroutines/experimental/Job;ILjava/lang/Object;)Lkotlinx/coroutines/experimental/CoroutineDispatcher;
 }
 
-public abstract class kotlinx/coroutines/experimental/ExecutorCoroutineDispatcherBase : kotlinx/coroutines/experimental/CloseableCoroutineDispatcher, kotlinx/coroutines/experimental/Delay {
+public abstract class kotlinx/coroutines/experimental/ExecutorCoroutineDispatcher : kotlinx/coroutines/experimental/CloseableCoroutineDispatcher, java/io/Closeable {
+	public fun <init> ()V
+	public abstract fun getExecutor ()Ljava/util/concurrent/Executor;
+}
+
+public abstract class kotlinx/coroutines/experimental/ExecutorCoroutineDispatcherBase : kotlinx/coroutines/experimental/ExecutorCoroutineDispatcher, kotlinx/coroutines/experimental/Delay {
 	public fun <init> ()V
 	public fun close ()V
 	public fun delay (JLjava/util/concurrent/TimeUnit;Lkotlin/coroutines/experimental/Continuation;)Ljava/lang/Object;
@@ -308,7 +309,8 @@ public abstract class kotlinx/coroutines/experimental/ExecutorCoroutineDispatche
 
 public final class kotlinx/coroutines/experimental/ExecutorsKt {
 	public static final fun asCoroutineDispatcher (Ljava/util/concurrent/Executor;)Lkotlinx/coroutines/experimental/CoroutineDispatcher;
-	public static final fun asCoroutineDispatcher (Ljava/util/concurrent/ExecutorService;)Lkotlinx/coroutines/experimental/CloseableCoroutineDispatcher;
+	public static final synthetic fun asCoroutineDispatcher (Ljava/util/concurrent/ExecutorService;)Lkotlinx/coroutines/experimental/CloseableCoroutineDispatcher;
+	public static final fun asCoroutineDispatcher (Ljava/util/concurrent/ExecutorService;)Lkotlinx/coroutines/experimental/ExecutorCoroutineDispatcher;
 	public static final fun toCoroutineDispatcher (Ljava/util/concurrent/Executor;)Lkotlinx/coroutines/experimental/CoroutineDispatcher;
 }
 
@@ -423,8 +425,9 @@ public final class kotlinx/coroutines/experimental/ScheduledKt {
 	public static synthetic fun withTimeoutOrNull$default (JLjava/util/concurrent/TimeUnit;Lkotlin/jvm/functions/Function2;Lkotlin/coroutines/experimental/Continuation;ILjava/lang/Object;)Ljava/lang/Object;
 }
 
-public final class kotlinx/coroutines/experimental/ThreadPoolDispatcher : kotlinx/coroutines/experimental/ExecutorCoroutineDispatcherBase, java/io/Closeable {
+public final class kotlinx/coroutines/experimental/ThreadPoolDispatcher : kotlinx/coroutines/experimental/ExecutorCoroutineDispatcherBase {
 	public fun close ()V
+	public fun getExecutor ()Ljava/util/concurrent/Executor;
 	public fun toString ()Ljava/lang/String;
 }
 

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

@@ -21,7 +21,7 @@ import kotlin.coroutines.experimental.*
  * pool is created. This is to work around the fact that ForkJoinPool creates threads that cannot perform
  * privileged actions.
  */
-object CommonPool : CoroutineDispatcher() {
+object CommonPool : ExecutorCoroutineDispatcher() {
 
     /**
      * Name of the property that controls default parallelism level of [CommonPool].
@@ -31,6 +31,9 @@ object CommonPool : CoroutineDispatcher() {
      */
     public const val DEFAULT_PARALLELISM_PROPERTY_NAME = "kotlinx.coroutines.default.parallelism"
 
+    override val executor: Executor
+        get() = pool ?: getOrCreatePoolSync()
+
     // Equals to -1 if not explicitly specified
     private val requestedParallelism = run<Int> {
         val property = Try { System.getProperty(DEFAULT_PARALLELISM_PROPERTY_NAME) } ?: return@run -1
@@ -132,4 +135,6 @@ object CommonPool : CoroutineDispatcher() {
     }
 
     override fun toString(): String = "CommonPool"
+
+    override fun close(): Unit = error("Close cannot be invoked on CommonPool")
 }

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

@@ -10,43 +10,66 @@ import java.util.concurrent.*
 import kotlin.coroutines.experimental.*
 
 /**
- * [CoroutineDispatcher] that implements [Closeable]
+ * [CoroutineDispatcher] that has underlying [Executor] for dispatching tasks.
+ * Instances of [ExecutorCoroutineDispatcher] should be closed by the owner of the dispatcher.
+ *
+ * This class is generally used as a bridge between coroutine-based API and
+ * asynchronous API which requires instance of the [Executor].
  */
-abstract class CloseableCoroutineDispatcher: CoroutineDispatcher(), Closeable
+public abstract class ExecutorCoroutineDispatcher: CloseableCoroutineDispatcher(), Closeable {
+
+    /**
+     * Underlying executor of current [CoroutineDispatcher].
+     */
+    public abstract val executor: Executor
+}
 
 /**
- * Converts an instance of [ExecutorService] to an implementation of [CloseableCoroutineDispatcher].
+ * [CoroutineDispatcher] that implements [Closeable].
+ *
+ * @suppress **Deprecated**: Use [ExecutorCoroutineDispatcher].
  */
-public fun ExecutorService.asCoroutineDispatcher(): CloseableCoroutineDispatcher =
+@Deprecated("Use ExecutorCoroutineDispatcher instead", replaceWith = ReplaceWith("ExecutorCoroutineDispatcher"))
+public abstract class CloseableCoroutineDispatcher: CoroutineDispatcher(), Closeable
+
+/**
+ * Converts an instance of [ExecutorService] to an implementation of [ExecutorCoroutineDispatcher].
+ */
+public fun ExecutorService.asCoroutineDispatcher(): ExecutorCoroutineDispatcher =
     // we know that an implementation of Executor.asCoroutineDispatcher actually returns a closeable one
-    (this as Executor).asCoroutineDispatcher() as CloseableCoroutineDispatcher
+    (this as Executor).asCoroutineDispatcher() as ExecutorCoroutineDispatcher
 
 /**
  * Converts an instance of [Executor] to an implementation of [CoroutineDispatcher].
- * @suppress **Deprecated**: Renamed to [asCoroutineDispatcher].
  */
-@Deprecated("Renamed to `asCoroutineDispatcher`",
-    replaceWith = ReplaceWith("asCoroutineDispatcher()"))
-public fun Executor.toCoroutineDispatcher(): CoroutineDispatcher =
-    ExecutorCoroutineDispatcher(this)
+public fun Executor.asCoroutineDispatcher(): CoroutineDispatcher =
+    ExecutorCoroutineDispatcherImpl(this)
+
+/**
+ * Converts an instance of [ExecutorService] to an implementation of [CloseableCoroutineDispatcher].
+ * @suppress **Deprecated**: Return type changed to [ExecutorCoroutineDispatcher].
+ */
+@Deprecated(level = DeprecationLevel.HIDDEN, message = "Return type changed to ExecutorCoroutineDispatcher")
+@JvmName("asCoroutineDispatcher") // for binary compatibility
+public fun ExecutorService.asCoroutineDispatcher_Deprecated(): CloseableCoroutineDispatcher =
+    asCoroutineDispatcher()
 
 /**
  * Converts an instance of [Executor] to an implementation of [CoroutineDispatcher].
+ * @suppress **Deprecated**: Renamed to [asCoroutineDispatcher].
  */
-public fun Executor.asCoroutineDispatcher(): CoroutineDispatcher =
-    ExecutorCoroutineDispatcher(this)
+@Deprecated("Renamed to `asCoroutineDispatcher`",
+    replaceWith = ReplaceWith("asCoroutineDispatcher()"))
+public fun Executor.toCoroutineDispatcher(): CoroutineDispatcher =
+    asCoroutineDispatcher()
 
-private class ExecutorCoroutineDispatcher(override val executor: Executor) : ExecutorCoroutineDispatcherBase()
+private class ExecutorCoroutineDispatcherImpl(override val executor: Executor) : ExecutorCoroutineDispatcherBase()
 
 /**
  * @suppress **This is unstable API and it is subject to change.**
  */
-public abstract class ExecutorCoroutineDispatcherBase : CloseableCoroutineDispatcher(), Delay {
-    /**
-     * @suppress **This is unstable API and it is subject to change.**
-     */
-    internal abstract val executor: Executor
-    
+public abstract class ExecutorCoroutineDispatcherBase : ExecutorCoroutineDispatcher(), Delay {
+
     override fun dispatch(context: CoroutineContext, block: Runnable) =
         try { executor.execute(timeSource.trackTask(block)) }
         catch (e: RejectedExecutionException) {
@@ -70,10 +93,10 @@ public abstract class ExecutorCoroutineDispatcherBase : CloseableCoroutineDispat
             try { (executor as? ScheduledExecutorService)
                 ?.schedule(block, time, unit) }
             catch (e: RejectedExecutionException) { null }
-        if (timeout != null)
-            return DisposableFutureHandle(timeout)
+        return if (timeout != null)
+            DisposableFutureHandle(timeout)
         else
-            return DefaultExecutor.invokeOnTimeout(time, unit, block)
+            DefaultExecutor.invokeOnTimeout(time, unit, block)
     }
 
     override fun close() {
@@ -98,7 +121,7 @@ private class ResumeUndispatchedRunnable(
  * An implementation of [DisposableHandle] that cancels the specified future on dispose.
  * @suppress **This is unstable API and it is subject to change.**
  */
-public class DisposableFutureHandle(private val future: Future<*>) : DisposableHandle {
+private class DisposableFutureHandle(private val future: Future<*>) : DisposableHandle {
     override fun dispose() {
         future.cancel(false)
     }

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

@@ -4,11 +4,9 @@
 
 package kotlinx.coroutines.experimental
 
-import java.io.Closeable
-import java.util.concurrent.Executors
-import java.util.concurrent.ScheduledExecutorService
+import java.util.concurrent.*
 import java.util.concurrent.atomic.AtomicInteger
-import kotlin.coroutines.experimental.CoroutineContext
+import kotlin.coroutines.experimental.*
 
 /**
  * Creates a new coroutine execution context using a single thread with built-in [yield] and [delay] support.
@@ -63,18 +61,18 @@ internal class PoolThread(
 public class ThreadPoolDispatcher internal constructor(
     private val nThreads: Int,
     private val name: String
-) : ExecutorCoroutineDispatcherBase(), Closeable {
+) : ExecutorCoroutineDispatcherBase() {
     private val threadNo = AtomicInteger()
 
-    internal override val executor: ScheduledExecutorService = Executors.newScheduledThreadPool(nThreads) { target ->
+    override val executor: Executor = Executors.newScheduledThreadPool(nThreads) { target ->
         PoolThread(this, target, if (nThreads == 1) name else name + "-" + threadNo.incrementAndGet())
     }
 
     /**
      * Closes this dispatcher -- shuts down all threads in this pool and releases resources.
      */
     public override fun close() {
-        executor.shutdown()
+        (executor as ExecutorService).shutdown()
     }
 
     override fun toString(): String = "ThreadPoolDispatcher[$nThreads, $name]"

core/kotlinx-coroutines-core/src/scheduling/CoroutineScheduler.kt

@@ -62,7 +62,7 @@ internal class CoroutineScheduler(
     private val maxPoolSize: Int,
     private val idleWorkerKeepAliveNs: Long = IDLE_WORKER_KEEP_ALIVE_NS,
     private val schedulerName: String = DEFAULT_SCHEDULER_NAME
-) : Closeable {
+) : Executor, Closeable {
     init {
         require(corePoolSize >= MIN_SUPPORTED_POOL_SIZE) {
             "Core pool size $corePoolSize should be at least $MIN_SUPPORTED_POOL_SIZE"
@@ -290,6 +290,8 @@ internal class CoroutineScheduler(
         private const val PARKED_VERSION_INC = 1L shl BLOCKING_SHIFT
     }
 
+    override fun execute(command: Runnable) = dispatch(command)
+
     override fun close() = shutdown(1000L)
 
     /*

core/kotlinx-coroutines-core/src/scheduling/ExperimentalCoroutineDispatcher.kt

@@ -6,7 +6,6 @@ package kotlinx.coroutines.experimental.scheduling
 
 import kotlinx.atomicfu.*
 import kotlinx.coroutines.experimental.*
-import java.io.*
 import java.util.concurrent.*
 import kotlin.coroutines.experimental.*
 
@@ -18,7 +17,7 @@ class ExperimentalCoroutineDispatcher(
     private val corePoolSize: Int,
     private val maxPoolSize: Int,
     private val idleWorkerKeepAliveNs: Long
-) : CoroutineDispatcher(), Delay, Closeable {
+) : ExecutorCoroutineDispatcher(), Delay {
     constructor(
         corePoolSize: Int = CORE_POOL_SIZE,
         maxPoolSize: Int = MAX_POOL_SIZE
@@ -28,6 +27,9 @@ class ExperimentalCoroutineDispatcher(
         IDLE_WORKER_KEEP_ALIVE_NS
     )
 
+    override val executor: Executor
+        get() = coroutineScheduler
+
     // This is variable for test purposes, so that we can reinitialize from clean state
     private var coroutineScheduler = CoroutineScheduler(corePoolSize, maxPoolSize, idleWorkerKeepAliveNs)
 
@@ -40,6 +42,7 @@ class ExperimentalCoroutineDispatcher(
     override fun scheduleResumeAfterDelay(time: Long, unit: TimeUnit, continuation: CancellableContinuation<Unit>): Unit =
             DefaultExecutor.scheduleResumeAfterDelay(time, unit, continuation)
 
+    // TODO throw error when this API becomes public and close it in tests via another method
     override fun close() = coroutineScheduler.close()
 
     override fun toString(): String {
@@ -90,11 +93,18 @@ private class LimitingDispatcher(
     val dispatcher: ExperimentalCoroutineDispatcher,
     val parallelism: Int,
     override val taskMode: TaskMode
-) : CoroutineDispatcher(), Delay, TaskContext {
+) : ExecutorCoroutineDispatcher(), Delay, TaskContext, Executor {
 
     private val queue = ConcurrentLinkedQueue<Runnable>()
     private val inFlightTasks = atomic(0)
 
+    override val executor: Executor
+        get() = this
+
+    override fun execute(command: Runnable) = dispatch(command, false)
+
+    override fun close(): Unit = error("Close cannot be invoked on LimitingBlockingDispatcher")
+
     override fun dispatch(context: CoroutineContext, block: Runnable) = dispatch(block, false)
 
     private fun dispatch(block: Runnable, fair: Boolean) {

core/kotlinx-coroutines-core/test/guide/test/TestUtil.kt

@@ -100,7 +100,7 @@ private fun shutdownDispatcherPools(timeout: Long) {
     for (i in 0 until n) {
         val thread = threads[i]
         if (thread is PoolThread)
-            thread.dispatcher.executor.apply {
+            (thread.dispatcher.executor as ExecutorService).apply {
                 shutdown()
                 awaitTermination(timeout, TimeUnit.MILLISECONDS)
                 shutdownNow().forEach { DefaultExecutor.execute(it) }

integration/kotlinx-coroutines-jdk8/test/future/FutureTest.kt

@@ -12,6 +12,7 @@ import org.junit.Assert.*
 import java.util.concurrent.*
 import java.util.concurrent.atomic.*
 import java.util.concurrent.locks.*
+import java.util.function.Supplier
 import kotlin.concurrent.*
 import kotlin.coroutines.experimental.*
 
@@ -314,6 +315,23 @@ class FutureTest : TestBase() {
         }
     }
 
+    private val threadLocal = ThreadLocal<String>()
+
+    @Test
+    fun testApiBridge() = runTest {
+        val result = newSingleThreadContext("ctx").use {
+            val future = CompletableFuture.supplyAsync(Supplier { threadLocal.set("value") }, it.executor)
+            val job = async(it) {
+                future.await()
+                threadLocal.get()
+            }
+
+            job.await()
+        }
+
+        assertEquals("value", result)
+    }
+
     class TestException(message: String) : Exception(message)
 
     private fun wrapContinuation(wrapper: (() -> Unit) -> Unit): CoroutineDispatcher = object : CoroutineDispatcher() {