Documentation improvements, naming, proper spurious wakeup check

Author: qwwdfsad

kotlinx-coroutines-core/common/src/internal/LockFreeTaskQueue.kt

@@ -175,7 +175,6 @@ internal class LockFreeTaskQueueCore<E : Any>(
                 }
                 // element == Placeholder can only be when add has not finished yet
                 if (element is Placeholder) return null // consider it not added yet
-                // now we tentative know element to remove -- check predicate
                 // we cannot put null into array here, because copying thread could replace it with Placeholder and that is a disaster
                 val newHead = (head + 1) and MAX_CAPACITY_MASK
                 if (_state.compareAndSet(state, state.updateHead(newHead))) {

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

@@ -25,7 +25,7 @@ import kotlin.random.*
  * ### Structural overview
  *
  * Scheduler consists of [corePoolSize] worker threads to execute CPU-bound tasks and up to [maxPoolSize] lazily created threads
- * to execute blocking tasks. Every worker a has local queue in addition to a global scheduler queue and the global queue
+ * to execute blocking tasks. Every worker has a local queue in addition to a global scheduler queue and the global queue
  * has priority over local queue to avoid starvation of externally-submitted (e.g. from Android UI thread) tasks.
  * Work-stealing is implemented on top of that queues to provide even load distribution and illusion of centralized run queue.
  *
@@ -245,7 +245,7 @@ internal class CoroutineScheduler(
      */
     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() = (controlState.value and CPU_PERMITS_MASK shr CPU_PERMITS_SHIFT).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()
@@ -261,14 +261,11 @@ internal class CoroutineScheduler(
         controlState.addAndGet(-(1L shl BLOCKING_SHIFT))
     }
 
-    private inline fun tryAcquireCpuPermit(): Boolean {
-        while (true) {
-            val state = controlState.value
-            val available = availableCpuPermits(state)
-            if (available == 0) return false
-            val update = state - (1L shl CPU_PERMITS_SHIFT)
-            if (controlState.compareAndSet(state, update)) return true
-        }
+    private inline fun tryAcquireCpuPermit(): Boolean = controlState.loop { state ->
+        val available = availableCpuPermits(state)
+        if (available == 0) return false
+        val update = state - (1L shl CPU_PERMITS_SHIFT)
+        if (controlState.compareAndSet(state, update)) return true
     }
 
     private inline fun releaseCpuPermit() = controlState.addAndGet(1L shl CPU_PERMITS_SHIFT)
@@ -283,9 +280,12 @@ internal class CoroutineScheduler(
         val NOT_IN_STACK = Symbol("NOT_IN_STACK")
 
         // Worker termination states
-        private const val FORBIDDEN = -1
-        private const val ALLOWED = 0
+        private const val TERMINATION_FORBIDDEN = -1
+        private const val TERMINATION_ALLOWED = 0
         private const val TERMINATED = 1
+        // Worker parking states
+        private const val PARKING_FORBIDDEN = -1
+        private const val PARKING_ALLOWED = 0
         private const val PARKED = 1
 
         // Masks of control state
@@ -419,7 +419,7 @@ internal class CoroutineScheduler(
     private fun tryUnpark(): Boolean {
         while (true) {
             val worker = parkedWorkersStackPop() ?: return false
-            if (!worker.parkingState.compareAndSet(ALLOWED, FORBIDDEN)) {
+            if (!worker.parkingState.compareAndSet(PARKING_ALLOWED, PARKING_FORBIDDEN)) {
                 LockSupport.unpark(worker)
             }
             if (worker.tryForbidTermination()) return true
@@ -525,7 +525,7 @@ internal class CoroutineScheduler(
                     "CPU = $cpuWorkers, " +
                     "blocking = $blockingWorkers, " +
                     "parked = $parkedWorkers, " +
-                    "retired = $dormant, " +
+                    "dormant = $dormant, " +
                     "terminated = $terminated}, " +
                 "running workers queues = $queueSizes, "+
                 "global CPU queue size = ${globalCpuQueue.size}, " +
@@ -581,16 +581,16 @@ internal class CoroutineScheduler(
         /**
          * Small state machine for termination.
          * Followed states are allowed:
-         * [ALLOWED] -- worker can wake up and terminate itself
-         * [FORBIDDEN] -- worker is not allowed to terminate (because it was chosen by another thread to help)
+         * [TERMINATION_ALLOWED] -- worker can wake up and terminate itself
+         * [TERMINATION_FORBIDDEN] -- worker is not allowed to terminate (because it was chosen by another thread to help)
          * [TERMINATED] -- final state, thread is terminating and cannot be resurrected
          *
          * Allowed transitions:
-         * [ALLOWED] -> [FORBIDDEN]
-         * [ALLOWED] -> [TERMINATED]
-         * [FORBIDDEN] -> [ALLOWED]
+         * [TERMINATION_ALLOWED] -> [TERMINATION_FORBIDDEN]
+         * [TERMINATION_ALLOWED] -> [TERMINATED]
+         * [TERMINATION_FORBIDDEN] -> [TERMINATION_ALLOWED]
          */
-        private val terminationState = atomic(ALLOWED)
+        private val terminationState = atomic(TERMINATION_ALLOWED)
 
         /**
          * It is set to the termination deadline when started doing [park] and it reset
@@ -610,22 +610,22 @@ internal class CoroutineScheduler(
          * The delay until at least one task in other worker queues will  become stealable.
          */
         private var minDelayUntilStealableTaskNs = 0L
-        // ALLOWED | PARKED | FORBIDDEN
-        val parkingState = atomic(ALLOWED)
+        // PARKING_ALLOWED | PARKING_FORBIDDEN | PARKED
+        val parkingState = atomic(PARKING_ALLOWED)
 
         private var rngState = Random.nextInt()
         /**
-         * Tries to set [terminationState] to [FORBIDDEN], returns `false` if this attempt fails.
+         * Tries to set [terminationState] to [TERMINATION_FORBIDDEN], returns `false` if this attempt fails.
          * This attempt may fail either because worker terminated itself or because someone else
          * claimed this worker (though this case is rare, because require very bad timings)
          */
         fun tryForbidTermination(): Boolean =
             when (val state = terminationState.value) {
                 TERMINATED -> false // already terminated
-                FORBIDDEN -> false // already forbidden, someone else claimed this worker
-                ALLOWED -> terminationState.compareAndSet(
-                    ALLOWED,
-                    FORBIDDEN
+                TERMINATION_FORBIDDEN -> false // already forbidden, someone else claimed this worker
+                TERMINATION_ALLOWED -> terminationState.compareAndSet(
+                    TERMINATION_ALLOWED,
+                    TERMINATION_FORBIDDEN
                 )
                 else -> error("Invalid terminationState = $state")
             }
@@ -679,42 +679,49 @@ internal class CoroutineScheduler(
                  *  While it could potentially lead to short (up to WORK_STEALING_TIME_RESOLUTION_NS ns) starvations,
                  * excess unparks and managing "one unpark per signalling" invariant become unfeasible, instead we are going to resolve
                  * it with "spinning via scans" mechanism.
+                 * NB: this short potential parking does not interfere with `tryUnpark`
                  */
                 if (minDelayUntilStealableTaskNs != 0L) {
                     if (!rescanned) {
                         rescanned = true
                         continue
                     } else {
                         tryReleaseCpu(WorkerState.PARKING)
+                        interrupted()
                         LockSupport.parkNanos(minDelayUntilStealableTaskNs)
                         minDelayUntilStealableTaskNs = 0L
                     }
                 }
                 /*
                  * 2) No tasks available, time to park and, potentially, shut down the thread.
-                 *
                  * Add itself to the stack of parked workers, re-scans all the queues
                  * to avoid missing wake-up (requestCpuWorker) and either starts executing discovered tasks or parks itself awaiting for new tasks.
                  */
-                parkingState.value = ALLOWED
-                if (parkedWorkersStackPush(this)) {
-                    continue
-                } else {
-                    assert { localQueue.size == 0 }
-                    tryReleaseCpu(WorkerState.PARKING)
-                    interrupted() // Cleanup interruptions
-                    while (inStack()) { // Prevent spurious wakeups
-                        if (isTerminated) break
-                        if (!parkingState.compareAndSet(ALLOWED, PARKED)) {
-                            break
-                        }
-                        park()
-                    }
-                }
+                tryPark()
             }
             tryReleaseCpu(WorkerState.TERMINATED)
         }
 
+        // Counterpart to "tryUnpark"
+        private fun tryPark() {
+            parkingState.value = PARKING_ALLOWED
+            if (parkedWorkersStackPush(this)) {
+                return
+            } else {
+                assert { localQueue.size == 0 }
+                tryReleaseCpu(WorkerState.PARKING)
+                interrupted() // Cleanup interruptions
+                // Failed to get a parking permit, bailout
+                if (!parkingState.compareAndSet(PARKING_ALLOWED, PARKED)) {
+                    return
+                }
+                while (inStack()) { // Prevent spurious wakeups
+                    if (isTerminated) break
+                    park()
+                }
+            }
+        }
+
         private fun inStack(): Boolean = nextParkedWorker !== NOT_IN_STACK
 
         private fun executeTask(task: Task) {
@@ -763,7 +770,7 @@ internal class CoroutineScheduler(
         }
 
         private fun park() {
-            terminationState.value = ALLOWED
+            terminationState.value = TERMINATION_ALLOWED
             // set termination deadline the first time we are here (it is reset in idleReset)
             if (terminationDeadline == 0L) terminationDeadline = System.nanoTime() + idleWorkerKeepAliveNs
             // actually park
@@ -789,7 +796,7 @@ internal class CoroutineScheduler(
                  * See tryUnpark for state reasoning.
                  * If this CAS fails, then we were successfully unparked by other worker and cannot terminate.
                  */
-                if (!terminationState.compareAndSet(ALLOWED, TERMINATED)) return
+                if (!terminationState.compareAndSet(TERMINATION_ALLOWED, TERMINATED)) return
                 /*
                  * At this point this thread is no longer considered as usable for scheduling.
                  * We need multi-step choreography to reindex workers.