Redispatch local tasks before thread parking in runBlocking

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

@@ -4,6 +4,8 @@
 
 package kotlinx.coroutines
 
+import kotlinx.coroutines.scheduling.*
+import kotlinx.coroutines.scheduling.CoroutineScheduler
 import java.util.concurrent.locks.*
 import kotlin.contracts.*
 import kotlin.coroutines.*
@@ -95,6 +97,12 @@ private class BlockingCoroutine<T>(
                     val parkNanos = eventLoop?.processNextEvent() ?: Long.MAX_VALUE
                     // note: process next even may loose unpark flag, so check if completed before parking
                     if (isCompleted) break
+                    if (blockedThread is CoroutineScheduler.Worker) {
+                        val queue = blockedThread.localQueue
+                        while (queue.size > 0) {
+                            queue.poll()?.let { blockedThread.scheduler.dispatch(it) }
+                        }
+                    }
                     parkNanos(this, parkNanos)
                 }
             } finally { // paranoia

kotlinx-coroutines-core/jvm/test/RunBlockingDispatchLocalTasksTest.kt

@@ -0,0 +1,36 @@
+package kotlinx.coroutines
+
+import java.util.concurrent.*
+import kotlin.coroutines.*
+import kotlin.coroutines.intrinsics.*
+import kotlin.test.*
+
+class RunBlockingDispatchLocalTasksTest {
+
+    // A coroutine deadlock occurs if there is a task in the local queue
+    // of the blocking thread before it is parked by the nested runBlocking
+    @Test(timeout = 1000)
+    fun testEmptyLocalTasksBeforePark() {
+        runBlocking(Dispatchers.IO) {
+            val latch = CountDownLatch(1)
+            lateinit var launchContinuation: Continuation<Unit>
+            lateinit var runBlockingContinuation: Continuation<Unit>
+            CoroutineScope(Dispatchers.IO).launch {
+                suspendCoroutineUninterceptedOrReturn {
+                    launchContinuation = it
+                    latch.countDown()
+                    COROUTINE_SUSPENDED
+                }
+                yield()
+                runBlockingContinuation.resume(Unit)
+            }
+            latch.await()
+            runBlocking {
+                suspendCancellableCoroutine {
+                    runBlockingContinuation = it
+                    launchContinuation.resume(Unit)
+                }
+            }
+        }
+    }
+}