Reduce contention on worker's lock in CoroutineScheduler (#3686)

The previous implementation was prone to a non-trivial contention that caused EDT freezes and, potentially, Android's ANRs.

Two root causes were identified:

1) Thread() constructor that has a non-trivial complexity along with JVM upcalls and is significantly slower than any other regular allocation
2) Thread.start() is on itself a JVM upcall that ends up on a global JVM lock[s]

Thread.start() is now invoked when the lock is released to reduce contention.
The first root cause is not addressed as optimistic thread allocation may lead to a potential CPU waste due to how optimistically lock-less detection is and because Thread.start() is an order of magnitude slower anyway

Fixes #3652

Author: qwwdfsad

kotlinx-coroutines-core/common/src/JobSupport.kt

@@ -208,7 +208,7 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
         assert { state.isCompleting } // consistency check -- must be marked as completing
         val proposedException = (proposedUpdate as? CompletedExceptionally)?.cause
         // Create the final exception and seal the state so that no more exceptions can be added
-        var wasCancelling = false // KLUDGE: we cannot have contract for our own expect fun synchronized
+        val wasCancelling: Boolean
         val finalException = synchronized(state) {
             wasCancelling = state.isCancelling
             val exceptions = state.sealLocked(proposedException)

kotlinx-coroutines-core/common/src/flow/StateFlow.kt

@@ -319,8 +319,8 @@ private class StateFlowImpl<T>(
         updateState(expect ?: NULL, update ?: NULL)
 
     private fun updateState(expectedState: Any?, newState: Any): Boolean {
-        var curSequence = 0
-        var curSlots: Array<StateFlowSlot?>? = this.slots // benign race, we will not use it
+        var curSequence: Int
+        var curSlots: Array<StateFlowSlot?>? // benign race, we will not use it
         synchronized(this) {
             val oldState = _state.value
             if (expectedState != null && oldState != expectedState) return false // CAS support

kotlinx-coroutines-core/common/src/flow/internal/AbstractSharedFlow.kt

@@ -41,7 +41,7 @@ internal abstract class AbstractSharedFlow<S : AbstractSharedFlowSlot<*>> : Sync
     @Suppress("UNCHECKED_CAST")
     protected fun allocateSlot(): S {
         // Actually create slot under lock
-        var subscriptionCount: SubscriptionCountStateFlow? = null
+        val subscriptionCount: SubscriptionCountStateFlow?
         val slot = synchronized(this) {
             val slots = when (val curSlots = slots) {
                 null -> createSlotArray(2).also { slots = it }
@@ -72,7 +72,7 @@ internal abstract class AbstractSharedFlow<S : AbstractSharedFlowSlot<*>> : Sync
     @Suppress("UNCHECKED_CAST")
     protected fun freeSlot(slot: S) {
         // Release slot under lock
-        var subscriptionCount: SubscriptionCountStateFlow? = null
+        val subscriptionCount: SubscriptionCountStateFlow?
         val resumes = synchronized(this) {
             nCollectors--
             subscriptionCount = _subscriptionCount // retrieve under lock if initialized

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

@@ -5,6 +5,7 @@
 package kotlinx.coroutines.internal
 
 import kotlinx.coroutines.*
+import kotlin.contracts.*
 
 /**
  * @suppress **This an internal API and should not be used from general code.**
@@ -16,4 +17,16 @@ public expect open class SynchronizedObject() // marker abstract class
  * @suppress **This an internal API and should not be used from general code.**
  */
 @InternalCoroutinesApi
-public expect inline fun <T> synchronized(lock: SynchronizedObject, block: () -> T): T
+public expect inline fun <T> synchronizedImpl(lock: SynchronizedObject, block: () -> T): T
+
+/**
+ * @suppress **This an internal API and should not be used from general code.**
+ */
+@OptIn(ExperimentalContracts::class)
+@InternalCoroutinesApi
+public inline fun <T> synchronized(lock: SynchronizedObject, block: () -> T): T {
+    contract {
+        callsInPlace(block, InvocationKind.EXACTLY_ONCE)
+    }
+    return synchronizedImpl(lock, block)
+}

kotlinx-coroutines-core/js/src/internal/Synchronized.kt

@@ -16,5 +16,4 @@ public actual typealias SynchronizedObject = Any
  * @suppress **This an internal API and should not be used from general code.**
  */
 @InternalCoroutinesApi
-public actual inline fun <T> synchronized(lock: SynchronizedObject, block: () -> T): T =
-    block()
+public actual inline fun <T> synchronizedImpl(lock: SynchronizedObject, block: () -> T): T = block()

kotlinx-coroutines-core/jvm/src/internal/ResizableAtomicArray.kt

@@ -28,11 +28,12 @@ internal class ResizableAtomicArray<T>(initialLength: Int) {
         val curLen = curArray.length()
         if (index < curLen) {
             curArray[index] = value
-        } else {
-            val newArray = AtomicReferenceArray<T>((index + 1).coerceAtLeast(2 * curLen))
-            for (i in 0 until curLen) newArray[i] = curArray[i]
-            newArray[index] = value
-            array = newArray // copy done
+            return
         }
+        // It would be nice to copy array in batch instead of 1 by 1, but it seems like Java has no API for that
+        val newArray = AtomicReferenceArray<T>((index + 1).coerceAtLeast(2 * curLen))
+        for (i in 0 until curLen) newArray[i] = curArray[i]
+        newArray[index] = value
+        array = newArray // copy done
     }
 }

kotlinx-coroutines-core/jvm/src/internal/Synchronized.kt

@@ -16,5 +16,5 @@ public actual typealias SynchronizedObject = Any
  * @suppress **This an internal API and should not be used from general code.**
  */
 @InternalCoroutinesApi
-public actual inline fun <T> synchronized(lock: SynchronizedObject, block: () -> T): T =
+public actual inline fun <T> synchronizedImpl(lock: SynchronizedObject, block: () -> T): T =
     kotlin.synchronized(lock, block)

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

@@ -252,16 +252,21 @@ internal class CoroutineScheduler(
 
     /**
      * State of worker threads.
-     * [workers] is array of lazily created workers up to [maxPoolSize] workers.
+     * [workers] is a dynamic array of lazily created workers up to [maxPoolSize] workers.
      * [createdWorkers] is count of already created workers (worker with index lesser than [createdWorkers] exists).
-     * [blockingTasks] is count of pending (either in the queue or being executed) tasks
+     * [blockingTasks] is count of pending (either in the queue or being executed) blocking tasks.
+     *
+     * Workers array is also used as a lock for workers' creation and termination sequence.
      *
      * **NOTE**: `workers[0]` is always `null` (never used, works as sentinel value), so
      * workers are 1-indexed, code path in [Worker.trySteal] is a bit faster and index swap during termination
-     * works properly
+     * works properly.
+     *
+     * Initial size is `Dispatchers.Default` size * 2 to prevent unnecessary resizes for slightly or steadily loaded
+     * applications.
      */
     @JvmField
-    val workers = ResizableAtomicArray<Worker>(corePoolSize + 1)
+    val workers = ResizableAtomicArray<Worker>((corePoolSize + 1) * 2)
 
     /**
      * The `Long` value describing the state of workers in this pool.
@@ -464,12 +469,13 @@ internal class CoroutineScheduler(
         }
     }
 
-    /*
+    /**
      * Returns the number of CPU workers after this function (including new worker) or
      * 0 if no worker was created.
      */
     private fun createNewWorker(): Int {
-        synchronized(workers) {
+        val worker: Worker
+        return synchronized(workers) {
             // Make sure we're not trying to resurrect terminated scheduler
             if (isTerminated) return -1
             val state = controlState.value
@@ -487,12 +493,11 @@ internal class CoroutineScheduler(
              * 2) Make it observable by increment created workers count
              * 3) Only then start the worker, otherwise it may miss its own creation
              */
-            val worker = Worker(newIndex)
+            worker = Worker(newIndex)
             workers.setSynchronized(newIndex, worker)
             require(newIndex == incrementCreatedWorkers())
-            worker.start()
-            return cpuWorkers + 1
-        }
+            cpuWorkers + 1
+        }.also { worker.start() } // Start worker when the lock is released to reduce contention, see #3652
     }
 
     /**

kotlinx-coroutines-core/native/src/internal/Synchronized.kt

@@ -17,4 +17,4 @@ public actual typealias SynchronizedObject = kotlinx.atomicfu.locks.Synchronized
  * @suppress **This an internal API and should not be used from general code.**
  */
 @InternalCoroutinesApi
-public actual inline fun <T> synchronized(lock: SynchronizedObject, block: () -> T): T = lock.withLock2(block)
+public actual inline fun <T> synchronizedImpl(lock: SynchronizedObject, block: () -> T): T = lock.withLock2(block)