Move victim argument in WorkQieie into receiver position to simplify the overall code structure

Author: qwwdfsad

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

@@ -265,15 +265,15 @@ internal class CoroutineScheduler(
 
     /**
      * Long describing state of workers in this pool.
-     * Currently includes created, CPU-acquired and blocking workers each occupying [BLOCKING_SHIFT] bits.
+     * Currently, includes created, CPU-acquired and blocking workers each occupying [BLOCKING_SHIFT] bits.
      */
     private val controlState = atomic(corePoolSize.toLong() shl CPU_PERMITS_SHIFT)
     private val createdWorkers: Int inline get() = (controlState.value and CREATED_MASK).toInt()
     private val availableCpuPermits: Int inline get() = availableCpuPermits(controlState.value)
 
     private inline fun createdWorkers(state: Long): Int = (state and CREATED_MASK).toInt()
     private inline fun blockingTasks(state: Long): Int = (state and BLOCKING_MASK shr BLOCKING_SHIFT).toInt()
-    public inline fun availableCpuPermits(state: Long): Int = (state and CPU_PERMITS_MASK shr CPU_PERMITS_SHIFT).toInt()
+    inline fun availableCpuPermits(state: Long): Int = (state and CPU_PERMITS_MASK shr CPU_PERMITS_SHIFT).toInt()
 
     // Guarded by synchronization
     private inline fun incrementCreatedWorkers(): Int = createdWorkers(controlState.incrementAndGet())
@@ -927,9 +927,9 @@ internal class CoroutineScheduler(
                 if (worker !== null && worker !== this) {
                     assert { localQueue.size == 0 }
                     val stealResult = if (blockingOnly) {
-                        localQueue.tryStealBlockingFrom(victim = worker.localQueue, stolenTask)
+                        worker.localQueue.tryStealBlocking(stolenTask)
                     } else {
-                        localQueue.tryStealFrom(victim = worker.localQueue, stolenTask)
+                        worker.localQueue.trySteal(stolenTask)
                     }
                     if (stealResult == TASK_STOLEN) {
                         val result = stolenTask.element

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

@@ -101,40 +101,37 @@ internal class WorkQueue {
     }
 
     /**
-     * Tries stealing from [victim] queue into the [stolenTaskRef] argument.
+     * Tries stealing from this queue into the [stolenTaskRef] argument.
      *
      * 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.
      */
-    fun tryStealFrom(victim: WorkQueue, stolenTaskRef: ObjectRef<Task?>): Long {
-        assert { bufferSize == 0 }
-        val task  = victim.pollBuffer()
+    fun trySteal(stolenTaskRef: ObjectRef<Task?>): Long {
+        val task  = pollBuffer()
         if (task != null) {
             stolenTaskRef.element = task
             return TASK_STOLEN
         }
-        return tryStealLastScheduled(victim, stolenTaskRef, blockingOnly = false)
+        return tryStealLastScheduled(stolenTaskRef, blockingOnly = false)
     }
 
-    fun tryStealBlockingFrom(victim: WorkQueue, stolenTaskRef: ObjectRef<Task?>): Long {
-        assert { bufferSize == 0 }
-        var start = victim.consumerIndex.value
-        val end = victim.producerIndex.value
-        val buffer = victim.buffer
+    fun tryStealBlocking(stolenTaskRef: ObjectRef<Task?>): Long {
+        var start = consumerIndex.value
+        val end = producerIndex.value
 
         while (start != end) {
             val index = start and MASK
-            if (victim.blockingTasksInBuffer.value == 0) break
+            if (blockingTasksInBuffer.value == 0) break
             val value = buffer[index]
             if (value != null && value.isBlocking && buffer.compareAndSet(index, value, null)) {
-                victim.blockingTasksInBuffer.decrementAndGet()
+                blockingTasksInBuffer.decrementAndGet()
                 stolenTaskRef.element = value
                 return TASK_STOLEN
             } else {
                 ++start
             }
         }
-        return tryStealLastScheduled(victim, stolenTaskRef, blockingOnly = true)
+        return tryStealLastScheduled(stolenTaskRef, blockingOnly = true)
     }
 
     fun offloadAllWorkTo(globalQueue: GlobalQueue) {
@@ -145,11 +142,11 @@ internal class WorkQueue {
     }
 
     /**
-     * Contract on return value is the same as for [tryStealFrom]
+     * Contract on return value is the same as for [trySteal]
      */
-    private fun tryStealLastScheduled(victim: WorkQueue, stolenTaskRef: ObjectRef<Task?>, blockingOnly: Boolean): Long {
+    private fun tryStealLastScheduled(stolenTaskRef: ObjectRef<Task?>, blockingOnly: Boolean): Long {
         while (true) {
-            val lastScheduled = victim.lastScheduledTask.value ?: return NOTHING_TO_STEAL
+            val lastScheduled = lastScheduledTask.value ?: return NOTHING_TO_STEAL
             if (blockingOnly && !lastScheduled.isBlocking) return NOTHING_TO_STEAL
 
             // TODO time wraparound ?
@@ -163,7 +160,7 @@ internal class WorkQueue {
              * If CAS has failed, either someone else had stolen this task or the owner executed this task
              * and dispatched another one. In the latter case we should retry to avoid missing task.
              */
-            if (victim.lastScheduledTask.compareAndSet(lastScheduled, null)) {
+            if (lastScheduledTask.compareAndSet(lastScheduled, null)) {
                 stolenTaskRef.element = lastScheduled
                 return TASK_STOLEN
             }

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

@@ -58,12 +58,12 @@ class WorkQueueStressTest : TestBase() {
                 startLatch.await()
                 while (!producerFinished || producerQueue.size != 0) {
                     stolenTasks[i].addAll(myQueue.drain(ref).map { task(it) })
-                    myQueue.tryStealFrom(victim = producerQueue, ref)
+                    producerQueue.trySteal(ref)
                 }
 
                 // Drain last element which is not counted in buffer
                 stolenTasks[i].addAll(myQueue.drain(ref).map { task(it) })
-                myQueue.tryStealFrom(producerQueue, ref)
+                producerQueue.trySteal(ref)
                 stolenTasks[i].addAll(myQueue.drain(ref).map { task(it) })
             }
         }
@@ -94,7 +94,7 @@ class WorkQueueStressTest : TestBase() {
             val ref = Ref.ObjectRef<Task?>()
             startLatch.await()
             while (stolen.size != offerIterations) {
-                if (myQueue.tryStealFrom(producerQueue, ref) != NOTHING_TO_STEAL) {
+                if (producerQueue.trySteal(ref) != NOTHING_TO_STEAL) {
                     stolen.addAll(myQueue.drain(ref).map { task(it) })
                 }
             }

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

@@ -63,10 +63,10 @@ class WorkQueueTest : TestBase() {
 
         val stealer = WorkQueue()
         val ref = ObjectRef<Task?>()
-        assertEquals(TASK_STOLEN, stealer.tryStealFrom(victim, ref))
+        assertEquals(TASK_STOLEN, victim.trySteal(ref))
         assertEquals(arrayListOf(1L), stealer.drain(ref))
 
-        assertEquals(TASK_STOLEN, stealer.tryStealFrom(victim, ref))
+        assertEquals(TASK_STOLEN, victim.trySteal(ref))
         assertEquals(arrayListOf(2L), stealer.drain(ref))
     }
 }