Scheduler tests infrastructure

Support "kotlinx.coroutines.scheduler" property to change DefaultDispatcher
Updated scheduler shrink test with custom idleWorkerKeepAliveNs config,
so that they execute faster
Fix submissions from another instance of CoroutineScheduler
Run JDK16 tests and tests under new scheduler tests only during nightly stress test

Author: elizarov

core/kotlinx-coroutines-core/build.gradle

@@ -44,10 +44,21 @@ task jdk16Test(type: Test, dependsOn: [testClasses, checkJdk16]) {
     exclude '**/*CancellableContinuationExceptionHandlingTest.*'
 }
 
-task moreTest(dependsOn: [lockFreedomTest, jdk16Test])
+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])
+
+extraTest.onlyIf { project.properties['stressTest'] != null }
+
 task testsJar(type: Jar, dependsOn: testClasses) {
     classifier = 'tests'
     from sourceSets.test.output

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

@@ -19,6 +19,7 @@ package kotlinx.coroutines.experimental
 import kotlinx.coroutines.experimental.internal.*
 import java.util.concurrent.atomic.*
 import kotlin.coroutines.experimental.*
+import kotlinx.coroutines.experimental.scheduling.*
 
 /**
  * Name of the property that control coroutine debugging. See [newCoroutineContext].
@@ -56,14 +57,25 @@ internal fun resetCoroutineId() {
     COROUTINE_ID.set(0)
 }
 
+internal const val COROUTINES_SCHEDULER_PROPERTY_NAME = "kotlinx.coroutines.scheduler"
+
+internal val useCoroutinesScheduler = systemProp(COROUTINES_SCHEDULER_PROPERTY_NAME).let { value ->
+    when (value) {
+        null -> false
+        "", "on" -> true
+        else -> error("System property '$COROUTINES_SCHEDULER_PROPERTY_NAME' has unrecognized value '$value'")
+    }
+}
+
 /**
  * This is the default [CoroutineDispatcher] that is used by all standard builders like
  * [launch], [async], etc if no dispatcher nor any other [ContinuationInterceptor] is specified in their context.
  *
  * It is currently equal to [CommonPool], but the value is subject to change in the future.
  */
 @Suppress("PropertyName")
-public actual val DefaultDispatcher: CoroutineDispatcher = CommonPool
+public actual val DefaultDispatcher: CoroutineDispatcher =
+    if (useCoroutinesScheduler) ExperimentalCoroutineDispatcher() else CommonPool
 
 /**
  * Creates context for the new coroutine. It installs [DefaultDispatcher] when no other dispatcher nor

core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/scheduling/CoroutineScheduler.kt

@@ -2,7 +2,6 @@ package kotlinx.coroutines.experimental.scheduling
 
 import kotlinx.atomicfu.*
 import kotlinx.coroutines.experimental.*
-import kotlinx.coroutines.experimental.channels.*
 import kotlinx.coroutines.experimental.internal.*
 import java.io.Closeable
 import java.util.*
@@ -55,15 +54,11 @@ import java.util.concurrent.locks.*
  */
 @Suppress("NOTHING_TO_INLINE")
 internal class CoroutineScheduler(
-    private val schedulerName: String,
     private val corePoolSize: Int,
-    private val maxPoolSize: Int
+    private val maxPoolSize: Int,
+    private val idleWorkerKeepAliveNs: Long = IDLE_WORKER_KEEP_ALIVE_NS,
+    private val schedulerName: String = DEFAULT_SCHEDULER_NAME
 ) : Closeable {
-    constructor(
-        corePoolSize: Int,
-        maxPoolSize: Int
-    ) : this("CoroutineScheduler", corePoolSize, maxPoolSize)
-
     init {
         require(corePoolSize >= MIN_SUPPORTED_POOL_SIZE) {
             "Core pool size $corePoolSize should be at least $MIN_SUPPORTED_POOL_SIZE"
@@ -74,6 +69,9 @@ internal class CoroutineScheduler(
         require(maxPoolSize <= MAX_SUPPORTED_POOL_SIZE) {
             "Max pool size $maxPoolSize should not exceed maximal supported number of threads $MAX_SUPPORTED_POOL_SIZE"
         }
+        require(idleWorkerKeepAliveNs > 0) {
+            "Idle worker keep alive time $idleWorkerKeepAliveNs must be postiive"
+        }
     }
 
     private val globalQueue: GlobalQueue = GlobalQueue()
@@ -454,6 +452,8 @@ internal class CoroutineScheduler(
      */
     private fun submitToLocalQueue(task: Task, fair: Boolean): Int {
         val worker = Thread.currentThread() as? Worker ?: return NOT_ADDED
+        if (worker.scheduler !== this) return NOT_ADDED // different scheduler's worker (!!!)
+
         var result = ADDED
 
         if (task.mode == TaskMode.NON_BLOCKING) {
@@ -570,6 +570,8 @@ internal class CoroutineScheduler(
             indexInArray = index
         }
 
+        val scheduler get() = this@CoroutineScheduler
+
         val localQueue: WorkQueue = WorkQueue()
 
         /**
@@ -780,9 +782,9 @@ internal class CoroutineScheduler(
             if (!blockingQuiescence()) return
             terminationState.value = ALLOWED
             // set termination deadline the first time we are here (it is reset in idleReset)
-            if (terminationDeadline == 0L) terminationDeadline = System.nanoTime() + IDLE_WORKER_KEEP_ALIVE_NS
+            if (terminationDeadline == 0L) terminationDeadline = System.nanoTime() + idleWorkerKeepAliveNs
             // actually park
-            doPark(IDLE_WORKER_KEEP_ALIVE_NS)
+            doPark(idleWorkerKeepAliveNs)
             // try terminate when we are idle past termination deadline
             // note, that comparison is written like this to protect against potential nanoTime wraparound
             if (System.nanoTime() - terminationDeadline >= 0) {
@@ -813,15 +815,15 @@ internal class CoroutineScheduler(
                 /*
                  * At this point this thread is no longer considered as usable for scheduling.
                  * We need multi-step choreography to reindex workers.
-                 * 
+                 *
                  * 1) Read current worker's index and reset it to zero.
                  */
                 val oldIndex = indexInArray
                 indexInArray = 0
                 /*
                  * Now this worker cannot become the top of parkedWorkersStack, but it can
                  * still be at the stack top via oldIndex.
-                 * 
+                 *
                  * 2) Update top of stack if it was pointing to oldIndex and make sure no
                  *    pending push/pop operation that might have already retrieved oldIndex could complete.
                  */

core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/scheduling/ExperimentalCoroutineDispatcher.kt

@@ -9,12 +9,23 @@ import kotlin.coroutines.experimental.*
 /**
  * @suppress **This is unstable API and it is subject to change.**
  */
-// TODO make internal after integration wih Ktor
+// TODO make internal (and rename) after complete integration
 class ExperimentalCoroutineDispatcher(
-    corePoolSize: Int = CORE_POOL_SIZE,
-    maxPoolSize: Int = MAX_POOL_SIZE
+    private val corePoolSize: Int,
+    private val maxPoolSize: Int,
+    private val idleWorkerKeepAliveNs: Long
 ) : CoroutineDispatcher(), Delay, Closeable {
-    private val coroutineScheduler = CoroutineScheduler(corePoolSize, maxPoolSize)
+    constructor(
+        corePoolSize: Int = CORE_POOL_SIZE,
+        maxPoolSize: Int = MAX_POOL_SIZE
+    ) : this(
+        corePoolSize,
+        maxPoolSize,
+        IDLE_WORKER_KEEP_ALIVE_NS
+    )
+
+    // This is variable for test purposes, so that we can reinitialize from clean state
+    private var coroutineScheduler = CoroutineScheduler(corePoolSize, maxPoolSize, idleWorkerKeepAliveNs)
 
     override fun dispatch(context: CoroutineContext, block: Runnable): Unit =
         coroutineScheduler.dispatch(block)
@@ -45,6 +56,17 @@ class ExperimentalCoroutineDispatcher(
 
     internal fun dispatchWithContext(block: Runnable, context: TaskContext?, fair: Boolean): Unit =
         coroutineScheduler.dispatch(block, context, fair)
+
+    // fot tests only
+    internal fun usePrivateScheduler() {
+        coroutineScheduler.shutdown(1000L)
+        coroutineScheduler = CoroutineScheduler(corePoolSize, maxPoolSize, idleWorkerKeepAliveNs)
+    }
+
+    // for tests only
+    internal fun shutdown(timeout: Long) {
+        coroutineScheduler.shutdown(timeout)
+    }
 }
 
 private class LimitingBlockingDispatcher(

core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/scheduling/Tasks.kt

@@ -7,6 +7,8 @@ import java.util.concurrent.*
 
 // TODO most of these fields will be moved to 'object ExperimentalDispatcher'
 
+internal const val DEFAULT_SCHEDULER_NAME = "CoroutineScheduler"
+
 // 100us as default
 @JvmField
 internal val WORK_STEALING_TIME_RESOLUTION_NS = systemProp(
@@ -20,9 +22,12 @@ internal val QUEUE_SIZE_OFFLOAD_THRESHOLD = systemProp(
 internal val BLOCKING_DEFAULT_PARALLELISM = systemProp(
     "kotlinx.coroutines.scheduler.blocking.parallelism", 16)
 
+// NOTE: we coerce default to at least two threads to give us chances that multi-threading problems
+// get reproduced even on a single-core machine, but support explicit setting of 1 thread scheduler if needed.
 @JvmField
 internal val CORE_POOL_SIZE = systemProp("kotlinx.coroutines.scheduler.core.pool.size",
-    AVAILABLE_PROCESSORS.coerceAtLeast(2), minValue = 2)
+    AVAILABLE_PROCESSORS.coerceAtLeast(2), // !!! at least two here
+    minValue = CoroutineScheduler.MIN_SUPPORTED_POOL_SIZE)
 
 @JvmField
 internal val MAX_POOL_SIZE = systemProp("kotlinx.coroutines.scheduler.max.pool.size",

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

@@ -16,6 +16,7 @@
 
 package kotlinx.coroutines.experimental
 
+import kotlinx.coroutines.experimental.scheduling.*
 import org.junit.After
 import org.junit.Before
 import java.util.concurrent.atomic.AtomicBoolean
@@ -105,17 +106,27 @@ public actual open class TestBase actual constructor() {
 
     @Before
     fun before() {
-        CommonPool.usePrivatePool()
+        initPoolsBeforeTest()
         threadsBefore = currentThreads()
     }
 
     @After
     fun onCompletion() {
         error.get()?.let { throw it }
         check(actionIndex.get() == 0 || finished.get()) { "Expecting that 'finish(...)' was invoked, but it was not" }
+        shutdownPoolsAfterTest()
+        checkTestThreads(threadsBefore)
+    }
+
+    fun initPoolsBeforeTest() {
+        CommonPool.usePrivatePool()
+        if (useCoroutinesScheduler) (DefaultDispatcher as ExperimentalCoroutineDispatcher).usePrivateScheduler()
+    }
+
+    fun shutdownPoolsAfterTest() {
         CommonPool.shutdown(SHUTDOWN_TIMEOUT)
+        if (useCoroutinesScheduler) (DefaultDispatcher as ExperimentalCoroutineDispatcher).shutdown(SHUTDOWN_TIMEOUT)
         DefaultExecutor.shutdown(SHUTDOWN_TIMEOUT)
-        checkTestThreads(threadsBefore)
     }
 
     @Suppress("ACTUAL_WITHOUT_EXPECT", "ACTUAL_FUNCTION_WITH_DEFAULT_ARGUMENTS")

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

@@ -21,19 +21,17 @@ import org.junit.*
 class TestBaseTest : TestBase() {
     @Test
     fun testThreadsShutdown() {
-        val SHUTDOWN_TIMEOUT = 1_000L
         repeat(1000 * stressTestMultiplier) { _ ->
-            CommonPool.usePrivatePool()
+            initPoolsBeforeTest()
             val threadsBefore = currentThreads()
             runBlocking {
-                val sub = launch(DefaultDispatcher) {
+                val sub = launch {
                     delay(10000000L)
                 }
                 sub.cancel()
                 sub.join()
             }
-            CommonPool.shutdown(SHUTDOWN_TIMEOUT)
-            DefaultExecutor.shutdown(SHUTDOWN_TIMEOUT)
+            shutdownPoolsAfterTest()
             checkTestThreads(threadsBefore)
         }
 

…xperimental/WithContextCommonPoolTest.kt …s/experimental/WithDefaultContextTest.ktcore/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/WithContextCommonPoolTest.kt renamed to core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/WithDefaultContextTest.kt core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/WithContextCommonPoolTest.kt renamed to core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/WithDefaultContextTest.kt

@@ -18,11 +18,11 @@ package kotlinx.coroutines.experimental
 
 import kotlin.test.*
 
-class WithContextCommonPoolTest : TestBase() {
+class WithDefaultContextTest : TestBase() {
     @Test
-    fun testCommonPoolNoSuspend() = runTest {
+    fun testNoSuspend() = runTest {
         expect(1)
-        val result = withContext(CommonPool) {
+        val result = withContext(DefaultDispatcher) {
             expect(2)
             "OK"
         }
@@ -31,9 +31,9 @@ class WithContextCommonPoolTest : TestBase() {
     }
 
     @Test
-    fun testCommonPoolWithSuspend() = runTest {
+    fun testWithSuspend() = runTest {
         expect(1)
-        val result = withContext(CommonPool) {
+        val result = withContext(DefaultDispatcher) {
             expect(2)
             delay(100)
             expect(3)

core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/channels/BroadcastChannelSubStressTest.kt

@@ -49,7 +49,7 @@ class BroadcastChannelSubStressTest(
     @Test
     fun testStress() = runBlocking {
         println("--- BroadcastChannelSubStressTest $kind")
-        val ctx = coroutineContext + CommonPool
+        val ctx = coroutineContext + DefaultDispatcher
         val sender =
             launch(context = ctx + CoroutineName("Sender")) {
                 while (isActive) {

core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/channels/ChannelAtomicCancelStressTest.kt

@@ -107,7 +107,7 @@ class ChannelAtomicCancelStressTest(val kind: TestChannelKind) : TestBase() {
     }
 
     fun launchSender() {
-        sender = launch(CommonPool, start = CoroutineStart.ATOMIC) {
+        sender = launch(DefaultDispatcher, start = CoroutineStart.ATOMIC) {
             val rnd = Random()
             cancellable(senderDone) {
                 var counter = 0
@@ -132,7 +132,7 @@ class ChannelAtomicCancelStressTest(val kind: TestChannelKind) : TestBase() {
     }
 
     fun launchReceiver() {
-        receiver = launch(CommonPool, start = CoroutineStart.ATOMIC) {
+        receiver = launch(DefaultDispatcher, start = CoroutineStart.ATOMIC) {
             val rnd = Random()
             cancellable(receiverDone) {
                 while (true) {