Move tests around

also fix wrong package names

Author: vsalavatov

kotlinx-coroutines-core/jvm/test/scheduling/BlockingCoroutineDispatcherLivenessStressTest.kt

@@ -4,10 +4,6 @@ import kotlinx.coroutines.testing.*
 import kotlinx.coroutines.*
 import org.junit.*
 import org.junit.Test
-import org.junit.runner.*
-import org.junit.runners.*
-import java.util.LinkedList
-import java.util.concurrent.*
 import java.util.concurrent.atomic.*
 import kotlin.test.*
 

kotlinx-coroutines-core/jvm/test/scheduling/BlockingCoroutineDispatcherTest.kt

@@ -1,10 +1,10 @@
 package kotlinx.coroutines.scheduling
 
+import kotlinx.coroutines.testing.*
 import kotlinx.coroutines.*
 import org.junit.*
 import org.junit.rules.*
 import java.util.concurrent.*
-import java.util.concurrent.atomic.*
 
 class BlockingCoroutineDispatcherTest : SchedulerTestBase() {
 
@@ -171,49 +171,4 @@ class BlockingCoroutineDispatcherTest : SchedulerTestBase() {
     fun testZeroParallelism() {
         blockingDispatcher(0)
     }
-
-    @Test
-    fun testNoCpuStarvationRunBlockingOnDefaultDispatcherThread() {
-        val maxDepth = CORES_COUNT * 3 + 3
-        fun body(depth: Int) {
-            if (depth == maxDepth) return
-            runBlocking(dispatcher) {
-                launch(dispatcher) {
-                    body(depth + 1)
-                }
-            }
-        }
-
-        body(1)
-        checkPoolThreadsCreated(maxDepth..maxDepth + 1)
-    }
-
-    @Test
-    fun testNoStarvationOfLimitedDispatcherWithRunBlocking() {
-        val taskCount = 5
-        val dispatcher = blockingDispatcher(2)
-        val barrier = CompletableDeferred<Unit>()
-        val count = AtomicInteger(0)
-        fun blockingCode() {
-            runBlocking {
-                count.incrementAndGet()
-                barrier.await()
-                count.decrementAndGet()
-            }
-        }
-        runBlocking {
-            repeat(taskCount) {
-                launch(dispatcher) {
-                    blockingCode()
-                }
-            }
-            while (count.get() != taskCount) {
-                Thread.sleep(1)
-            }
-            barrier.complete(Unit)
-            while (count.get() != 0) {
-                Thread.sleep(1)
-            }
-        }
-    }
 }

kotlinx-coroutines-core/jvm/test/scheduling/RunBlockingCoroutineDispatcherTest.kt

@@ -0,0 +1,59 @@
+package kotlinx.coroutines.scheduling
+
+import kotlinx.coroutines.*
+import org.junit.*
+import java.util.concurrent.atomic.*
+
+class RunBlockingCoroutineDispatcherTest : SchedulerTestBase() {
+    @Test
+    fun testRecursiveRunBlockingCanExceedDefaultDispatcherLimit() {
+        val maxDepth = CORES_COUNT * 3 + 3
+        fun body(depth: Int) {
+            if (depth == maxDepth) return
+            runBlocking(Dispatchers.Default) {
+                launch(Dispatchers.Default) {
+                    body(depth + 1)
+                }
+            }
+        }
+
+        body(1)
+        checkPoolThreadsCreated(maxDepth..maxDepth + 1)
+    }
+
+    @Test
+    fun testNoDefaultDispatcherStarvationWithRunBlocking() = testRunBlockingCanExceedDispatchersLimit(dispatcher, CORE_POOL_SIZE * 3 + 3)
+
+    @Test
+    fun testNoIoDispatcherStarvationWithRunBlocking() = testRunBlockingCanExceedDispatchersLimit(blockingDispatcher(2), 5)
+
+    private fun testRunBlockingCanExceedDispatchersLimit(targetDispatcher: CoroutineDispatcher, threadsToReach: Int) {
+        val barrier = CompletableDeferred<Unit>()
+        val count = AtomicInteger(0)
+        fun blockingCode() {
+            runBlocking {
+                count.incrementAndGet()
+                barrier.await()
+                count.decrementAndGet()
+            }
+        }
+        runBlocking {
+            repeat(threadsToReach) {
+                launch(targetDispatcher) {
+                    blockingCode()
+                }
+            }
+            while (count.get() != threadsToReach) {
+                Thread.sleep(1)
+            }
+            async(targetDispatcher) {
+                yield()
+                42
+            }.join()
+            barrier.complete(Unit)
+            while (count.get() != 0) {
+                Thread.sleep(1)
+            }
+        }
+    }
+}

kotlinx-coroutines-core/jvm/test/scheduling/RunBlockingCoroutineSchedulerLivenessTestBase.kt renamed to kotlinx-coroutines-core/jvm/test/scheduling/RunBlockingCoroutineSchedulerLivenessStressTest.kt

@@ -1,13 +1,46 @@
-package scheduling
+package kotlinx.coroutines.scheduling
 
 import kotlinx.coroutines.*
-import kotlinx.coroutines.scheduling.*
 import kotlinx.coroutines.testing.*
+import org.junit.*
+import org.junit.runner.*
+import org.junit.runners.*
 import java.util.*
 import java.util.concurrent.*
 
-open class RunBlockingCoroutineSchedulerLivenessTestBase : SchedulerTestBase() {
-    protected fun testSchedulerLiveness(targetDispatcher: CoroutineDispatcher, yieldMask: Int = 0b1111): Unit = runBlocking {
+@RunWith(Parameterized::class)
+class RunBlockingCoroutineSchedulerLivenessStressTest(private val yieldMask: Int) : SchedulerTestBase() {
+    init {
+        corePoolSize = 1
+    }
+
+    companion object {
+        @JvmStatic
+        @Parameterized.Parameters
+        fun data(): Array<Array<Any?>> {
+            return Array(16 * stressTestMultiplierSqrt) { arrayOf(it) }
+        }
+    }
+
+    @Test
+    fun testLivenessOfDefaultDispatcher(): Unit = testSchedulerLiveness(dispatcher, yieldMask)
+
+    @Test
+    fun testLivenessOfIoDispatcher(): Unit = testSchedulerLiveness(blockingDispatcher(1), yieldMask)
+
+    @Test
+    fun testLivenessOfLimitedDispatcherOnTopOfDefaultDispatcher() =
+        testSchedulerLiveness(dispatcher.limitedParallelism(1), yieldMask)
+
+    @Test
+    fun testLivenessOfLimitedDispatcherOnTopOfIoDispatcher() = testSchedulerLiveness(
+        // Important: inner limitedDispatcher will be on top of this LimitedDispatcher, so there are two Workers from
+        // two different LimitedDispatchers that must coordinate their permits, not just one.
+        // In other words, LimitedDispatcher's Worker should also respect BlockingDispatchAware on its inner tasks
+        blockingDispatcher.value.limitedParallelism(1), yieldMask
+    )
+
+    private fun testSchedulerLiveness(targetDispatcher: CoroutineDispatcher, yieldMask: Int = 0b1111): Unit = runBlocking {
         val oldRunBlockings = LinkedList<Job>()
         var maxOldRunBlockings = 0
         var busyWaits = 0