introduce runBlockingWithParallelismCompensation and remove parallelism compensation from runBlocking

Author: vsalavatov

kotlinx-coroutines-core/jvm/src/Builders.kt

@@ -46,6 +46,21 @@ import kotlin.coroutines.*
  */
 @Throws(InterruptedException::class)
 public actual fun <T> runBlocking(context: CoroutineContext, block: suspend CoroutineScope.() -> T): T {
+    contract {
+        callsInPlace(block, InvocationKind.EXACTLY_ONCE)
+    }
+    return runBlocking(context, compensateParallelism = false, block)
+}
+
+@Throws(InterruptedException::class)
+internal fun <T> runBlockingWithParallelismCompensation(context: CoroutineContext = EmptyCoroutineContext, block: suspend CoroutineScope.() -> T): T {
+    contract {
+        callsInPlace(block, InvocationKind.EXACTLY_ONCE)
+    }
+    return runBlocking(context, compensateParallelism = true, block)
+}
+
+private fun <T> runBlocking(context: CoroutineContext, compensateParallelism: Boolean, block: suspend CoroutineScope.() -> T): T {
     contract {
         callsInPlace(block, InvocationKind.EXACTLY_ONCE)
     }
@@ -64,15 +79,16 @@ public actual fun <T> runBlocking(context: CoroutineContext, block: suspend Coro
             ?: ThreadLocalEventLoop.currentOrNull()
         newContext = GlobalScope.newCoroutineContext(context)
     }
-    val coroutine = BlockingCoroutine<T>(newContext, currentThread, eventLoop)
+    val coroutine = BlockingCoroutine<T>(newContext, currentThread, eventLoop, compensateParallelism)
     coroutine.start(CoroutineStart.DEFAULT, coroutine, block)
     return coroutine.joinBlocking()
 }
 
 private class BlockingCoroutine<T>(
     parentContext: CoroutineContext,
     private val blockedThread: Thread,
-    private val eventLoop: EventLoop?
+    private val eventLoop: EventLoop?,
+    private val compensateParallelism: Boolean,
 ) : AbstractCoroutine<T>(parentContext, true, true) {
 
     override val isScopedCoroutine: Boolean get() = true
@@ -96,7 +112,11 @@ private class BlockingCoroutine<T>(
                     // note: process next even may loose unpark flag, so check if completed before parking
                     if (isCompleted) break
                     if (parkNanos > 0) {
-                        withCompensatedParallelism {
+                        if (compensateParallelism) {
+                            withCompensatedParallelism {
+                                parkNanos(this, parkNanos)
+                            }
+                        } else {
                             parkNanos(this, parkNanos)
                         }
                     }

kotlinx-coroutines-core/jvm/test/scheduling/RunBlockingCoroutineDispatcherTest.kt renamed to kotlinx-coroutines-core/jvm/test/scheduling/RunBlockingWithParallelismCompensationCoroutineDispatcherTest.kt

@@ -4,13 +4,13 @@ import kotlinx.coroutines.*
 import org.junit.*
 import java.util.concurrent.atomic.*
 
-class RunBlockingCoroutineDispatcherTest : SchedulerTestBase() {
+class RunBlockingWithParallelismCompensationCoroutineDispatcherTest : SchedulerTestBase() {
     @Test
     fun testRecursiveRunBlockingCanExceedDefaultDispatcherLimit() {
         val maxDepth = CORES_COUNT * 3 + 3
         fun body(depth: Int) {
             if (depth == maxDepth) return
-            runBlocking(Dispatchers.Default) {
+            runBlockingWithParallelismCompensation(Dispatchers.Default) {
                 launch(Dispatchers.Default) {
                     body(depth + 1)
                 }
@@ -31,13 +31,13 @@ class RunBlockingCoroutineDispatcherTest : SchedulerTestBase() {
         val barrier = CompletableDeferred<Unit>()
         val count = AtomicInteger(0)
         fun blockingCode() {
-            runBlocking {
+            runBlockingWithParallelismCompensation {
                 count.incrementAndGet()
                 barrier.await()
                 count.decrementAndGet()
             }
         }
-        runBlocking {
+        runBlockingWithParallelismCompensation {
             repeat(threadsToReach) {
                 launch(targetDispatcher) {
                     blockingCode()

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

@@ -10,7 +10,7 @@ import java.util.*
 import java.util.concurrent.*
 
 @RunWith(Parameterized::class)
-class RunBlockingCoroutineSchedulerLivenessStressTest(private val yieldMask: Int) : SchedulerTestBase() {
+class RunBlockingWithParallelismCompensationCoroutineSchedulerLivenessStressTest(private val yieldMask: Int) : SchedulerTestBase() {
     init {
         corePoolSize = 1
     }
@@ -53,7 +53,7 @@ class RunBlockingCoroutineSchedulerLivenessStressTest(private val yieldMask: Int
             val barrier2 = CompletableDeferred<Unit>()
             val blocking = launch(targetDispatcher) {
                 barrier.await()
-                runBlocking {
+                runBlockingWithParallelismCompensation {
                     if ((yieldMask and 1) != 0) yield()
                     barrier2.await()
                     if ((yieldMask and 2) != 0) yield()