Coroutine scheduler is used by default instead of deprecated CommonPool

* Documentation and guide are updated correspondingly
* "DefaultDispatcher" is used as a public name of the default impl
* Implementation is integrated with virtual time-source
* Shutdown sequence is reimplemented in a safe way for tests,
  makes "close" safe to use on custom instances.
* "close" on DefaultDispatcher throws exception just in case
* -Dkotlinx.coroutines.scheduler=off can be used to switch back to
  CommonPool

Fixes #198

Author: elizarov

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

@@ -16,10 +16,8 @@ public object Dispatchers {
      * [launch][CoroutineScope.launch], [async][CoroutineScope.async], etc
      * if no dispatcher nor any other [ContinuationInterceptor] is specified in their context.
      *
-     * It is backed by a shared pool of threads on JVM.
-     * You can set system property "`kotlinx.coroutines.scheduler`" (either no value or to the value of "`on`")
-     * to use an experimental coroutine dispatcher that shares threads with [Dispatchers.IO] and thus can switch to
-     * context without performing an actual thread context switch.
+     * It is backed by a shared pool of threads on JVM. By default, the maximal number of threads used
+     * by this dispatcher is equal to the number CPU cores, but is at least two.
      */
     @JvmField
     public val Default: CoroutineDispatcher =

core/kotlinx-coroutines-core/build.gradle

@@ -32,6 +32,7 @@ tasks.withType(Test) {
 
 test {
     exclude '**/*LFTest.*'
+    systemProperty 'kotlinx.coroutines.scheduler.keep.alive.sec', '100000' // any unpark problem hangs test
 }
 
 task lockFreedomTest(type: Test, dependsOn: testClasses) {
@@ -48,18 +49,14 @@ task jdk16Test(type: Test, dependsOn: [testClasses, checkJdk16]) {
     exclude '**/exceptions/.*'
 }
 
-task schedulerTest(type: Test, dependsOn: testClasses) {
-    systemProperty 'kotlinx.coroutines.scheduler', ''
-    systemProperty 'kotlinx.coroutines.scheduler.keep.alive.sec', '100000' // any unpark problem hangs test
-}
-
 // Always run those tests
 task moreTest(dependsOn: [lockFreedomTest])
 
-build.dependsOn moreTest
-
 // Run these tests only during nightly stress test
-task extraTest(dependsOn: [jdk16Test, schedulerTest])
+task extraTest(dependsOn: [jdk16Test])
+
+
+build.dependsOn moreTest
 
 extraTest.onlyIf { project.properties['stressTest'] != null }
 

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

@@ -106,7 +106,7 @@ object CommonPool : ExecutorCoroutineDispatcher() {
         pool ?: createPool().also { pool = it }
 
     override fun dispatch(context: CoroutineContext, block: Runnable) =
-        try { (pool ?: getOrCreatePoolSync()).execute(timeSource.trackTask(block)) }
+        try { (pool ?: getOrCreatePoolSync()).execute(timeSource.wrapTask(block)) }
         catch (e: RejectedExecutionException) {
             timeSource.unTrackTask()
             DefaultExecutor.execute(block)

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

@@ -20,8 +20,8 @@ internal const val COROUTINES_SCHEDULER_PROPERTY_NAME = "kotlinx.coroutines.sche
 
 internal val useCoroutinesScheduler = systemProp(COROUTINES_SCHEDULER_PROPERTY_NAME).let { value ->
     when (value) {
-        null -> false
-        "", "on" -> true
+        null, "", "on" -> true
+        "off" -> false
         else -> error("System property '$COROUTINES_SCHEDULER_PROPERTY_NAME' has unrecognized value '$value'")
     }
 }
@@ -39,7 +39,7 @@ public actual val DefaultDispatcher: CoroutineDispatcher
     get() = Dispatchers.Default
 
 internal actual fun createDefaultDispatcher(): CoroutineDispatcher =
-    if (useCoroutinesScheduler) BackgroundDispatcher else CommonPool
+    if (useCoroutinesScheduler) DefaultScheduler else CommonPool
 
 /**
  * The [CoroutineDispatcher] that is designed for offloading blocking IO tasks to a shared pool of threads.

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

@@ -20,6 +20,10 @@ public const val IO_PARALLELISM_PROPERTY_NAME = "kotlinx.coroutines.io.paralleli
  * The number of threads used by this dispatcher is limited by the value of
  * "`kotlinx.coroutines.io.parallelism`" ([IO_PARALLELISM_PROPERTY_NAME]) system property.
  * It defaults to the limit of 64 threads or the number of cores (whichever is larger).
+ *
+ * This dispatcher shares threads with a [Default][Dispatchers.Default] dispatcher, so using
+ * `withContext(Dispatchers.IO) { ... }` does not lead to an actual switching to another thread —
+ * typically execution continues in the same thread.
  */
 public val Dispatchers.IO: CoroutineDispatcher
-    get() = BackgroundDispatcher.IO
+    get() = DefaultScheduler.IO

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

@@ -83,7 +83,7 @@ private class ExecutorCoroutineDispatcherImpl(override val executor: Executor) :
 public abstract class ExecutorCoroutineDispatcherBase : ExecutorCoroutineDispatcher(), Delay {
 
     override fun dispatch(context: CoroutineContext, block: Runnable) =
-        try { executor.execute(timeSource.trackTask(block)) }
+        try { executor.execute(timeSource.wrapTask(block)) }
         catch (e: RejectedExecutionException) {
             timeSource.unTrackTask()
             DefaultExecutor.execute(block)

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

@@ -9,7 +9,8 @@ import java.util.concurrent.locks.LockSupport
 internal interface TimeSource {
     fun currentTimeMillis(): Long
     fun nanoTime(): Long
-    fun trackTask(block: Runnable): Runnable
+    fun wrapTask(block: Runnable): Runnable
+    fun trackTask()
     fun unTrackTask()
     fun registerTimeLoopThread()
     fun unregisterTimeLoopThread()
@@ -20,7 +21,8 @@ internal interface TimeSource {
 internal object DefaultTimeSource : TimeSource {
     override fun currentTimeMillis(): Long = System.currentTimeMillis()
     override fun nanoTime(): Long = System.nanoTime()
-    override fun trackTask(block: Runnable): Runnable = block
+    override fun wrapTask(block: Runnable): Runnable = block
+    override fun trackTask() {}
     override fun unTrackTask() {}
     override fun registerTimeLoopThread() {}
     override fun unregisterTimeLoopThread() {}

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

@@ -290,35 +290,33 @@ internal class CoroutineScheduler(
 
     override fun execute(command: Runnable) = dispatch(command)
 
-    override fun close() = shutdown(1000L)
+    override fun close() = shutdown(10_000L)
 
-    /*
-     * Shuts down current scheduler and waits until all threads are stopped.
-     * This method uses unsafe API (does unconditional unparks)
-     * and intended to be used only for testing. Invocation has no additional effect if already closed.
-     */
+    // Shuts down current scheduler and waits until all work is done and all threads are stopped.
     fun shutdown(timeout: Long) {
         // atomically set termination flag which is checked when workers are added or removed
         if (!isTerminated.compareAndSet(false, true)) return
-
-        /*
-         * Shutdown current thread. Note that shutdown is testing utility,
-         * so we don't do anything special to properly verify that no tasks are submitted after close()
-         */
-        val thread = Thread.currentThread()
-        (thread as? Worker)?.tryReleaseCpu(WorkerState.TERMINATED)
-
+        // make sure we are not waiting for the current thread
+        val currentWorker = Thread.currentThread() as? Worker
         // Capture # of created workers that cannot change anymore (mind the synchronized block!)
         val created = synchronized(workers) { createdWorkers }
         for (i in 1..created) {
             val worker = workers[i]!!
-            if (worker.isAlive) {
-                // Unparking alive thread is unsafe in general, but acceptable for testing purposes
+            if (worker.isAlive && worker !== currentWorker) {
                 LockSupport.unpark(worker)
                 worker.join(timeout)
+                worker.localQueue.offloadAllWork(globalQueue)
             }
+
+        }
+        // Finish processing tasks from globalQueue and/or from this worker's local queue
+        while (true) {
+            val task = currentWorker?.findTask() ?: globalQueue.removeFirstOrNull() ?: break
+            runSafely(task)
         }
-        // cleanup state to make sure that tryUnpark tries to create new threads and crashes because it isTerminated
+        // Shutdown current thread
+        currentWorker?.tryReleaseCpu(WorkerState.TERMINATED)
+        // cleanup state to make sure that tryUnpark tries to create new threads and fails because isTerminated
         assert(cpuPermits.availablePermits() == corePoolSize)
         parkedWorkersStack.value = 0L
         controlState.value = 0L
@@ -333,6 +331,7 @@ internal class CoroutineScheduler(
      * @param fair whether the task should be dispatched fairly (strict FIFO) or not (semi-FIFO)
      */
     fun dispatch(block: Runnable, taskContext: TaskContext = NonBlockingContext, fair: Boolean = false) {
+        timeSource.trackTask() // this is needed for virtual time support
         // TODO at some point make DispatchTask extend Task and make its field settable to save an allocation
         val task = Task(block, schedulerTimeSource.nanoTime(), taskContext)
         // try to submit the task to the local queue and act depending on the result
@@ -439,7 +438,7 @@ internal class CoroutineScheduler(
     private fun createNewWorker(): Int {
         synchronized(workers) {
             // Make sure we're not trying to resurrect terminated scheduler
-            if (isTerminated.value) throw ShutdownException()
+            if (isTerminated.value) throw RejectedExecutionException("$schedulerName was terminated")
             val state = controlState.value
             val created = createdWorkers(state)
             val blocking = blockingWorkers(state)
@@ -456,9 +455,6 @@ internal class CoroutineScheduler(
         }
     }
 
-    // Is thrown when attempting to create new worker, but this scheduler isTerminated
-    private class ShutdownException : RuntimeException()
-
     /**
      * Returns [ADDED], or [NOT_ADDED], or [ADDED_REQUIRES_HELP].
      */
@@ -565,6 +561,17 @@ internal class CoroutineScheduler(
                 "]"
     }
 
+    private fun runSafely(task: Task) {
+        try {
+            task.run()
+        } catch (e: Throwable) {
+            val thread = Thread.currentThread()
+            thread.uncaughtExceptionHandler.uncaughtException(thread, e)
+        } finally {
+            timeSource.unTrackTask()
+        }
+    }
+
     internal inner class Worker private constructor() : Thread() {
         init {
             isDaemon = true
@@ -685,41 +692,28 @@ internal class CoroutineScheduler(
         private var lastStealIndex = 0 // try in order repeated, reset when unparked
 
         override fun run() {
-            try {
-                var wasIdle = false // local variable to avoid extra idleReset invocations when tasks repeatedly arrive
-                while (!isTerminated.value && state != WorkerState.TERMINATED) {
-                    val task = findTask()
-                    if (task == null) {
-                        // Wait for a job with potential park
-                        if (state == WorkerState.CPU_ACQUIRED) {
-                            cpuWorkerIdle()
-                        } else {
-                            blockingWorkerIdle()
-                        }
-                        wasIdle = true
+            var wasIdle = false // local variable to avoid extra idleReset invocations when tasks repeatedly arrive
+            while (!isTerminated.value && state != WorkerState.TERMINATED) {
+                val task = findTask()
+                if (task == null) {
+                    // Wait for a job with potential park
+                    if (state == WorkerState.CPU_ACQUIRED) {
+                        cpuWorkerIdle()
                     } else {
-                        if (wasIdle) {
-                            idleReset(task.mode)
-                            wasIdle = false
-                        }
-                        beforeTask(task)
-                        runSafely(task)
-                        afterTask(task)
+                        blockingWorkerIdle()
                     }
+                    wasIdle = true
+                } else {
+                    if (wasIdle) {
+                        idleReset(task.mode)
+                        wasIdle = false
+                    }
+                    beforeTask(task)
+                    runSafely(task)
+                    afterTask(task)
                 }
-            } catch (e: ShutdownException) {
-                // race with shutdown -- ignore exception and don't print it on the console
-            } finally {
-                tryReleaseCpu(WorkerState.TERMINATED)
-            }
-        }
-
-        private fun runSafely(task: Task) {
-            try {
-                task.run()
-            } catch (t: Throwable) {
-                uncaughtExceptionHandler.uncaughtException(this, t)
             }
+            tryReleaseCpu(WorkerState.TERMINATED)
         }
 
         private fun beforeTask(task: Task) {
@@ -823,7 +817,7 @@ internal class CoroutineScheduler(
         private fun tryTerminateWorker() {
             synchronized(workers) {
                 // Make sure we're not trying race with termination of scheduler
-                if (isTerminated.value) throw ShutdownException()
+                if (isTerminated.value) return
                 // Someone else terminated, bail out
                 if (createdWorkers <= corePoolSize) return
                 // Try to find blocking task before termination
@@ -906,7 +900,7 @@ internal class CoroutineScheduler(
             spins = 0 // Volatile write, should be written last
         }
 
-        private fun findTask(): Task? {
+        internal fun findTask(): Task? {
             if (tryAcquireCpuPermit()) return findTaskWithCpuPermit()
             /*
              * If the local queue is empty, try to extract blocking task from global queue.

core/kotlinx-coroutines-core/src/scheduling/ExperimentalCoroutineDispatcher.kt renamed to core/kotlinx-coroutines-core/src/scheduling/Dispatcher.kt

@@ -7,14 +7,19 @@ package kotlinx.coroutines.experimental.scheduling
 import kotlinx.atomicfu.*
 import kotlinx.coroutines.experimental.*
 import kotlinx.coroutines.experimental.internal.*
+import java.lang.UnsupportedOperationException
 import java.util.concurrent.*
 import kotlin.coroutines.experimental.*
 
 /**
- * Default instance of coroutine dispatcher for background coroutines (as opposed to UI coroutines).
+ * Default instance of coroutine dispatcher.
  */
-internal object BackgroundDispatcher : ExperimentalCoroutineDispatcher() {
+internal object DefaultScheduler : ExperimentalCoroutineDispatcher() {
     val IO = blocking(systemProp(IO_PARALLELISM_PROPERTY_NAME, 64.coerceAtLeast(AVAILABLE_PROCESSORS)))
+
+    override fun close() {
+        throw UnsupportedOperationException("$DEFAULT_SCHEDULER_NAME cannot be closed")
+    }
 }
 
 /**
@@ -39,15 +44,14 @@ open class ExperimentalCoroutineDispatcher(
         get() = coroutineScheduler
 
     // This is variable for test purposes, so that we can reinitialize from clean state
-    private var coroutineScheduler = CoroutineScheduler(corePoolSize, maxPoolSize, idleWorkerKeepAliveNs)
+    private var coroutineScheduler = createScheduler()
 
     override fun dispatch(context: CoroutineContext, block: Runnable): Unit =
         coroutineScheduler.dispatch(block)
 
     override fun dispatchYield(context: CoroutineContext, block: Runnable): Unit =
         coroutineScheduler.dispatch(block, fair = true)
 
-    // 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 {
@@ -82,16 +86,23 @@ open class ExperimentalCoroutineDispatcher(
     internal fun dispatchWithContext(block: Runnable, context: TaskContext, fair: Boolean): Unit =
         coroutineScheduler.dispatch(block, context, fair)
 
+    private fun createScheduler() = CoroutineScheduler(corePoolSize, maxPoolSize, idleWorkerKeepAliveNs)
+
     // fot tests only
+    @Synchronized
     internal fun usePrivateScheduler() {
-        coroutineScheduler.shutdown(1000L)
-        coroutineScheduler = CoroutineScheduler(corePoolSize, maxPoolSize, idleWorkerKeepAliveNs)
+        coroutineScheduler.shutdown(10_000L)
+        coroutineScheduler = createScheduler()
     }
 
     // for tests only
+    @Synchronized
     internal fun shutdown(timeout: Long) {
         coroutineScheduler.shutdown(timeout)
     }
+
+    // for tests only
+    internal fun restore() = usePrivateScheduler() // recreate scheduler
 }
 
 private class LimitingDispatcher(

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

@@ -11,7 +11,7 @@ import java.util.concurrent.*
 
 // TODO most of these fields will be moved to 'object ExperimentalDispatcher'
 
-internal const val DEFAULT_SCHEDULER_NAME = "CoroutineScheduler"
+internal const val DEFAULT_SCHEDULER_NAME = "DefaultDispatcher"
 
 // 100us as default
 @JvmField

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

@@ -144,6 +144,13 @@ internal class WorkQueue {
         }
     }
 
+    internal fun offloadAllWork(globalQueue: GlobalQueue) {
+        while (true) {
+            val task = pollExternal() ?: return
+            globalQueue.addLast(task)
+        }
+    }
+
     /**
      * [poll] for external (not owning this queue) workers
      */