~

Author: qwwdfsad

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

@@ -103,7 +103,8 @@ internal actual inline fun platformAutoreleasePool(crossinline block: () -> Unit
  */
 @InternalCoroutinesApi
 @DelicateCoroutinesApi
-public fun runSingleTaskFromCurrentSystemDispatcher(): Long {
+@PublishedApi
+internal fun runSingleTaskFromCurrentSystemDispatcher(): Long {
     val thread = Thread.currentThread()
     if (thread !is CoroutineScheduler.Worker) throw IllegalStateException("Expected CoroutineScheduler.Worker, but got $thread")
     return thread.runSingleTask()
@@ -119,7 +120,8 @@ public fun runSingleTaskFromCurrentSystemDispatcher(): Long {
  */
 @InternalCoroutinesApi
 @DelicateCoroutinesApi
-public fun Thread.isIoDispatcherThread(): Boolean {
+@PublishedApi
+internal fun Thread.isIoDispatcherThread(): Boolean {
     if (this !is CoroutineScheduler.Worker) return false
     return isIo()
 }

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

@@ -906,12 +906,14 @@ internal class CoroutineScheduler(
             return findBlockingTask()
         }
 
+        // NB: ONLY for runSingleTask method
         private fun findBlockingTask(): Task? {
             return localQueue.pollBlocking()
                 ?: globalBlockingQueue.removeFirstOrNull()
                 ?: trySteal(STEAL_BLOCKING_ONLY)
         }
 
+        // NB: ONLY for runSingleTask method
         private fun findCpuTask(): Task? {
             return localQueue.pollCpu()
                 ?: globalBlockingQueue.removeFirstOrNull()

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

@@ -112,10 +112,12 @@ internal class WorkQueue {
      *
      * Returns [NOTHING_TO_STEAL] if queue has nothing to steal, [TASK_STOLEN] if at least task was stolen
      * or positive value of how many nanoseconds should pass until the head of this queue will be available to steal.
+     *
+     * [StealingMode] controls what tasks to steal:
+     * * [STEAL_ANY] is default mode for scheduler, task from the head (in FIFO order) is stolen
+     * * [STEAL_BLOCKING_ONLY] is mode for stealing *an arbitrary* blocking task which is used by scheduler when helping in Dispatchers.IO mode
+     * * [STEAL_CPU_ONLY] is a kludge for `runSingleTaskFromCurrentSystemDispatcher`
      */
-    // TODO move it to tests where appropriate
-    fun trySteal(stolenTaskRef: ObjectRef<Task?>): Long = trySteal(STEAL_ANY, stolenTaskRef)
-
     fun trySteal(stealingMode: StealingMode, stolenTaskRef: ObjectRef<Task?>): Long {
         val task = when (stealingMode) {
             STEAL_ANY -> pollBuffer()
@@ -168,7 +170,7 @@ internal class WorkQueue {
         return pollWithMode(onlyBlocking = false /* only cpu */)
     }
 
-    private fun pollWithMode(/* Only blocking OR only CPU */onlyBlocking: Boolean): Task? {
+    private fun pollWithMode(/* Only blocking OR only CPU */ onlyBlocking: Boolean): Task? {
         val start = consumerIndex.value
         var end = producerIndex.value
         // CPU or (BLOCKING & hasBlocking)

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

@@ -56,7 +56,7 @@ public actual typealias TestResult = Unit
  */
 public actual open class TestBase(private var disableOutCheck: Boolean)  {
 
-    actual constructor(): this(true)
+    actual constructor(): this(false)
 
     public actual val isBoundByJsTestTimeout = false
     private var actionIndex = AtomicInteger()

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

@@ -9,6 +9,7 @@ import kotlinx.coroutines.internal.AVAILABLE_PROCESSORS
 import org.junit.Test
 import java.util.*
 import java.util.concurrent.ConcurrentHashMap
+import java.util.concurrent.CountDownLatch
 import java.util.concurrent.CyclicBarrier
 import java.util.concurrent.atomic.AtomicInteger
 import kotlin.random.*
@@ -18,63 +19,76 @@ import kotlin.time.*
 
 class CoroutineSchedulerInternalApiStressTest : TestBase() {
 
-    @Test
-    fun testHelpDefaultIoIsIsolated() {
-        repeat(10) {
-            runTest {
-                val jobToComplete = Job()
-                val tasksToCompleteJob = AtomicInteger(200)
+    @Test(timeout = 120_000L)
+    fun testHelpDefaultIoIsIsolated() = repeat(100 * stressTestMultiplierSqrt) {
+        val ioTaskMarker = ThreadLocal.withInitial { false }
+        var threadThatShould = Thread.currentThread()
+        runTest {
+            val jobToComplete = Job()
+            val expectedIterations = 100
+            val completionLatch = CountDownLatch(1)
+            val tasksToCompleteJob = AtomicInteger(expectedIterations)
+            val observedIoThreads = Collections.newSetFromMap(ConcurrentHashMap<Thread, Boolean>())
+            val observedDefaultThreads = Collections.newSetFromMap(ConcurrentHashMap<Thread, Boolean>())
 
-                val observedIoThreads = Collections.newSetFromMap(ConcurrentHashMap<Thread, Boolean>())
-                val observedDefaultThreads = Collections.newSetFromMap(ConcurrentHashMap<Thread, Boolean>())
-
-                val barrier = CyclicBarrier(AVAILABLE_PROCESSORS)
-                repeat(AVAILABLE_PROCESSORS - 1) {
-                    // Launch CORES - 1 spawners
-                    launch(Dispatchers.Default) {
-                        barrier.await()
-                        while (!jobToComplete.isCompleted) {
-                            launch {
-                                observedDefaultThreads.add(Thread.currentThread())
-                                val tasksLeft = tasksToCompleteJob.decrementAndGet()
-                                if (tasksLeft == 0) {
-                                    // Verify threads first
-                                    try {
-                                        assertFalse(observedIoThreads.containsAll(observedDefaultThreads))
-                                    } finally {
-                                        jobToComplete.complete()
-                                    }
+            val barrier = CyclicBarrier(AVAILABLE_PROCESSORS)
+            val spawners = ArrayList<Job>()
+            repeat(AVAILABLE_PROCESSORS - 1) {
+                // Launch CORES - 1 spawners
+                spawners += launch(Dispatchers.Default) {
+                    barrier.await()
+                    repeat(expectedIterations) {
+                        launch {
+                            val tasksLeft = tasksToCompleteJob.decrementAndGet()
+                            if (tasksLeft < 0) return@launch // Leftovers are being executed all over the place
+                            if (threadThatShould !== Thread.currentThread()) {
+                                val a = 2
+                            }
+                            observedDefaultThreads.add(Thread.currentThread())
+                            if (tasksLeft == 0) {
+                                // Verify threads first
+                                try {
+                                    assertFalse(observedIoThreads.containsAll(observedDefaultThreads))
+                                } finally {
+                                    jobToComplete.complete()
                                 }
                             }
+                        }
 
-                            // Sometimes launch an IO task
-                            if (Random.nextInt(0..9) == 0) {
-                                launch(Dispatchers.IO) {
-                                    observedIoThreads.add(Thread.currentThread())
-                                    assertTrue(Thread.currentThread().isIoDispatcherThread())
-                                }
+                        // Sometimes launch an IO task to mess with a scheduler
+                        if (Random.nextInt(0..9) == 0) {
+                            launch(Dispatchers.IO) {
+                                ioTaskMarker.set(true)
+                                observedIoThreads.add(Thread.currentThread())
+                                assertTrue(Thread.currentThread().isIoDispatcherThread())
                             }
                         }
                     }
+                    completionLatch.await()
                 }
+            }
 
-                withContext(Dispatchers.Default) {
-                    barrier.await()
-
-                    while (!jobToComplete.isCompleted) {
-                        val result = runSingleTaskFromCurrentSystemDispatcher()
-                        if (result == 0L) {
-                            continue
-                        } else if (result >= 0L) {
-                            delay(result.toDuration(DurationUnit.NANOSECONDS))
-                        } else {
-                            delay(10)
-                        }
+            withContext(Dispatchers.Default) {
+                threadThatShould = Thread.currentThread()
+                barrier.await()
+                var timesHelped = 0
+                while (!jobToComplete.isCompleted) {
+                    val result = runSingleTaskFromCurrentSystemDispatcher()
+                    assertFalse(ioTaskMarker.get())
+                    if (result == 0L) {
+                        ++timesHelped
+                        continue
+                    } else if (result >= 0L) {
+                        Thread.sleep(result.toDuration(DurationUnit.NANOSECONDS).toDelayMillis())
+                    } else {
+                        Thread.sleep(10)
                     }
-                    assertTrue(Thread.currentThread() in observedDefaultThreads)
                 }
-                coroutineContext.job.children.toList().joinAll()
+                completionLatch.countDown()
+//                assertEquals(100, timesHelped)
+//                assertTrue(Thread.currentThread() in observedDefaultThreads, observedDefaultThreads.toString())
             }
         }
     }
 }
+

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

@@ -107,3 +107,5 @@ internal fun GlobalQueue.drain(): List<Long> {
     }
     return result
 }
+
+internal fun WorkQueue.trySteal(stolenTaskRef: ObjectRef<Task?>): Long = trySteal(STEAL_ANY, stolenTaskRef)