Document blocking tasks machinery, get rid of an additional class

Author: qwwdfsad

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

@@ -20,49 +20,70 @@ import kotlin.random.*
  *
  * Current scheduler implementation has two optimization targets:
  * * Efficiency in the face of communication patterns (e.g., actors communicating via channel)
- * * Dynamic resizing to support blocking calls without re-dispatching coroutine to separate "blocking" thread pool
+ * * Dynamic resizing to support blocking calls without re-dispatching coroutine to separate "blocking" thread pool.
  *
  * ### 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 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.
+ * Scheduler consists of [corePoolSize] worker threads to execute CPU-bound tasks and up to
+ * [maxPoolSize] lazily created  threads 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.
  *
  * ### Scheduling policy
  *
  * When a coroutine is dispatched from within a scheduler worker, it's placed into the head of worker run queue.
- * If the head is not empty, the task from the head is moved to the tail. Though it is unfair scheduling policy,
+ * If the head is not empty, the task from the head is moved to the tail. Though it is an unfair scheduling policy,
  * it effectively couples communicating coroutines into one and eliminates scheduling latency
- * that arises from placing task to the end of the queue.
- * Placing former head to the tail is necessary to provide semi-FIFO order, otherwise queue degenerates to stack.
+ * that arises from placing tasks to the end of the queue.
+ * Placing former head to the tail is necessary to provide semi-FIFO order, otherwise, queue degenerates to stack.
  * When a coroutine is dispatched from an external thread, it's put into the global queue.
  *
  * ### Work stealing and affinity
  *
- * To provide even tasks distribution worker tries to steal tasks from other workers queues before parking when his local queue is empty.
- * A non-standard solution is implemented to provide tasks affinity: task from FIFO buffer may be stolen only if it is stale enough
- * (based on the value of [WORK_STEALING_TIME_RESOLUTION_NS]).
- * For this purpose monotonic global clock ([System.nanoTime]) is used and every task has associated with it submission time.
- * This approach shows outstanding results when coroutines are cooperative, but as downside scheduler now depends on high-resolution global clock
+ * To provide even tasks distribution worker tries to steal tasks from other workers queues
+ * before parking when his local queue is empty.
+ * A non-standard solution is implemented to provide tasks affinity: a task from FIFO buffer may be stolen
+ * only if it is stale enough based on the value of [WORK_STEALING_TIME_RESOLUTION_NS].
+ * For this purpose, monotonic global clock is used, and every task has associated with its submission time.
+ * This approach shows outstanding results when coroutines are cooperative,
+ * but as downside scheduler now depends on a high-resolution global clock,
  * which may limit scalability on NUMA machines. Tasks from LIFO buffer can be stolen on a regular basis.
  *
  * ### Thread management
- * One of the hardest parts of the scheduler is decentralized management of the threads with the progress guarantees similar
- * to the regular centralized executors. The state of the threads consists of [controlState] and [parkedWorkersStack] fields.
- * The former field incorporates the amount of created threads, CPU-tokens and blocking tasks that require a thread compensation,
+ * One of the hardest parts of the scheduler is decentralized management of the threads with the progress guarantees
+ * similar to the regular centralized executors.
+ * The state of the threads consists of [controlState] and [parkedWorkersStack] fields.
+ * The former field incorporates the amount of created threads, CPU-tokens and blocking tasks
+ * that require a thread compensation,
  * while the latter represents intrusive versioned Treiber stack of idle workers.
- * When a worker cannot find any work, he first adds itself to the stack, then re-scans the queue (to avoid missing signal)
- * and then attempts to park itself (there is additional layer of signalling against unnecessary park/unpark).
- * If worker finds a task that it cannot yet steal due to timer constraints, it stores this fact in its state
+ * When a worker cannot find any work, they first add themselves to the stack,
+ * then re-scans the queue to avoid missing signals and then attempts to park
+ * with additional rendezvous against unnecessary parking.
+ * If a worker finds a task that it cannot yet steal due to time constraints, it stores this fact in its state
  * (to be uncounted when additional work is signalled) and parks for such duration.
  *
- * When a new task arrives to the scheduler (whether it's local or global queue), either an idle worker is being signalled, or
- * a new worker is attempted to be created (only [corePoolSize] workers can be created for regular CPU tasks).
+ * When a new task arrives in the scheduler (whether it is local or global queue),
+ * either an idle worker is being signalled, or a new worker is attempted to be created.
+ * Only [corePoolSize] workers can be created for regular CPU tasks)
  *
- * ### Dynamic resizing and support of blocking tasks
+ * ### Support for blocking tasks
+ * The scheduler also supports the notion of [blocking][TaskMode.PROBABLY_BLOCKING] tasks.
+ * When executing or enqueuing blocking tasks, the scheduler notifies or creates one more worker in
+ * addition to core pool size, so at any given moment, it has [corePoolSize] threads (potentially not yet created)
+ * to serve CPU-bound tasks. To properly guarantee liveness, the scheduler maintains
+ * "CPU permits" -- [corePoolSize] special tokens that permit an arbitrary worker to execute and steal CPU-bound tasks.
+ * When worker encounters blocking tasks, it basically hands off its permit to another thread (not directly though) to
+ * keep invariant "scheduler always has at least min(pending CPU tasks, core pool size)
+ * and at most core pool size threads to execute CPU tasks".
+ * To avoid overprovision, workers without CPU permit are allowed to scan [globalBlockingQueue]
+ * and steal **only** blocking tasks from other workers.
  *
- * TODO
+ * The scheduler does not limit the count of pending blocking tasks, potentially creating up to [maxPoolSize] threads.
+ * End users do not have access to the scheduler directly and can dispatch blocking tasks only with
+ * [LimitingDispatcher] that does control concurrency level by its own mechanism.
  */
 @Suppress("NOTHING_TO_INLINE")
 internal class CoroutineScheduler(

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

@@ -138,7 +138,7 @@ internal class WorkQueue {
     }
 
     fun offloadAllWorkTo(globalQueue: GlobalQueue) {
-        lastScheduledTask.getAndSet(null)?.let { globalQueue.add(it) }
+        lastScheduledTask.getAndSet(null)?.let { globalQueue.addLast(it) }
         while (pollTo(globalQueue)) {
             // Steal everything
         }
@@ -173,7 +173,7 @@ internal class WorkQueue {
 
     private fun pollTo(queue: GlobalQueue): Boolean {
         val task = pollBuffer() ?: return false
-        queue.add(task)
+        queue.addLast(task)
         return true
     }
 
@@ -198,13 +198,3 @@ internal class WorkQueue {
         }
     }
 }
-
-private fun GlobalQueue.add(task: Task) {
-    /*
-     * globalQueue is closed as the very last step in the shutdown sequence when all worker threads had
-     * been already shutdown (with the only exception of the last worker thread that might be performing
-     * shutdown procedure itself). As a consistency check we do a [cheap!] check that it is not closed here yet.
-     */
-    val added = addLast(task)
-    assert { added }
-}