Scheduler cosmetics

Stealing logic cleaned up a bit (mostly style)
Update worker thread names when worker threads are terminated to maintain
consistency between their index in array and name;
Improved debugging info (toString)
Merged yield and spin counters, parkTimeNs is in Int

Author: elizarov

core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/CommonPool.kt

@@ -16,6 +16,7 @@
 
 package kotlinx.coroutines.experimental
 
+import kotlinx.coroutines.experimental.internal.*
 import kotlinx.coroutines.experimental.timeunit.TimeUnit
 import java.util.concurrent.*
 import java.util.concurrent.atomic.*
@@ -34,7 +35,6 @@ import kotlin.coroutines.experimental.*
  * privileged actions.
  */
 object CommonPool : CoroutineDispatcher() {
-
     /**
      * Name of the property that controls default parallelism level of [CommonPool].
      * If the property is not specified, `Runtime.getRuntime().availableProcessors() - 1` will be used instead (or `1` for single-core JVM).
@@ -43,18 +43,8 @@ object CommonPool : CoroutineDispatcher() {
      */
     public const val DEFAULT_PARALLELISM_PROPERTY_NAME = "kotlinx.coroutines.default.parallelism"
 
-    private val parallelism = run<Int> {
-        val property = Try { System.getProperty(DEFAULT_PARALLELISM_PROPERTY_NAME) }
-        if (property == null) {
-            (Runtime.getRuntime().availableProcessors() - 1).coerceAtLeast(1)
-        } else {
-            val parallelism = property.toIntOrNull()
-            if (parallelism == null || parallelism < 1) {
-                error("Expected positive number in $DEFAULT_PARALLELISM_PROPERTY_NAME, but has $property")
-            }
-            parallelism
-        }
-    }
+    private val parallelism = systemProp(DEFAULT_PARALLELISM_PROPERTY_NAME,
+        (AVAILABLE_PROCESSORS - 1).coerceAtLeast(1))
 
     // For debug and tests
     private var usePrivatePool = false

core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/CoroutineContext.kt

@@ -16,7 +16,7 @@
 
 package kotlinx.coroutines.experimental
 
-import sun.text.normalizer.UTF16.*
+import kotlinx.coroutines.experimental.internal.*
 import java.util.concurrent.atomic.*
 import kotlin.coroutines.experimental.*
 
@@ -40,9 +40,7 @@ public const val DEBUG_PROPERTY_VALUE_ON = "on"
  */
 public const val DEBUG_PROPERTY_VALUE_OFF = "off"
 
-internal val DEBUG = run {
-    val value = try { System.getProperty(DEBUG_PROPERTY_NAME) }
-        catch (e: SecurityException) { null }
+internal val DEBUG = systemProp(DEBUG_PROPERTY_NAME).let { value ->
     when (value) {
         DEBUG_PROPERTY_VALUE_AUTO, null -> CoroutineId::class.java.desiredAssertionStatus()
         DEBUG_PROPERTY_VALUE_ON, "" -> true

core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/internal/SystemProps.kt

@@ -0,0 +1,40 @@
+/*
+ * Copyright 2016-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package kotlinx.coroutines.experimental.internal
+
+// number of processors at startup for consistent prop initialization
+internal val AVAILABLE_PROCESSORS = Runtime.getRuntime().availableProcessors()
+
+internal fun systemProp(
+    propertyName: String
+): String? =
+    try {
+        System.getProperty(propertyName)
+    } catch (e: SecurityException) {
+        null
+    }
+
+internal fun systemProp(
+    propertyName: String,
+    defaultValue: Int,
+    minValue: Int = 1,
+    maxValue: Int = Int.MAX_VALUE
+): Int
+    = systemProp(propertyName, defaultValue.toLong(), minValue.toLong(), maxValue.toLong()).toInt()
+
+internal fun systemProp(
+    propertyName: String,
+    defaultValue: Long,
+    minValue: Long = 1,
+    maxValue: Long = Long.MAX_VALUE
+): Long {
+    val value = systemProp(propertyName) ?: return defaultValue
+    val parsed = value.toLongOrNull()
+        ?: error("System property '$propertyName' has unrecognized value '$value'")
+    if (parsed !in minValue..maxValue) {
+        error("System property '$propertyName' should be in range $minValue..$maxValue, but is '$parsed'")
+    }
+    return parsed
+}

core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/scheduling/CoroutineScheduler.kt

@@ -54,9 +54,14 @@ import java.util.concurrent.locks.*
  */
 @Suppress("NOTHING_TO_INLINE")
 internal class CoroutineScheduler(
+    private val schedulerName: String,
     private val corePoolSize: Int,
     private val maxPoolSize: Int
 ) : Closeable {
+    constructor(
+        corePoolSize: Int,
+        maxPoolSize: Int
+    ) : this("CoroutineScheduler", corePoolSize, maxPoolSize)
 
     private val globalQueue: GlobalQueue = GlobalQueue()
 
@@ -161,16 +166,16 @@ internal class CoroutineScheduler(
     private val isTerminated = atomic(false)
 
     companion object {
-        private const val MAX_SPINS = 1000L
-        private const val MAX_YIELDS = 500L
-
-        @JvmStatic
-        private val MAX_PARK_TIME_NS = TimeUnit.SECONDS.toNanos(1)
+        private const val MAX_SPINS = 1000
+        private const val MAX_YIELDS = MAX_SPINS + 500
+        
+        @JvmStatic // Note, that is fits into Int (it is is equal to 10^9)
+        private val MAX_PARK_TIME_NS = TimeUnit.SECONDS.toNanos(1).toInt()
 
         @JvmStatic
         private val MIN_PARK_TIME_NS = (WORK_STEALING_TIME_RESOLUTION_NS / 4)
             .coerceAtLeast(10)
-            .coerceAtMost(MAX_PARK_TIME_NS)
+            .coerceAtMost(MAX_PARK_TIME_NS.toLong()).toInt()
 
         // A symbol to mark workers that are not in parkedWorkersStack
         private val NOT_IN_STACK = Symbol("NOT_IN_STACK")
@@ -417,14 +422,10 @@ internal class CoroutineScheduler(
         var blockingWorkers = 0
         var cpuWorkers = 0
         var retired = 0
-        var finished = 0
-
+        var terminated = 0
         val queueSizes = arrayListOf<String>()
         for (worker in workers) {
-            if (worker == null) {
-                continue
-            }
-
+            if (worker == null) continue
             val queueSize = worker.localQueue.size()
             when (worker.state) {
                 WorkerState.PARKING -> ++parkedWorkers
@@ -440,29 +441,44 @@ internal class CoroutineScheduler(
                     ++retired
                     if (queueSize > 0) queueSizes += queueSize.toString() + "r" // Retiring
                 }
-                WorkerState.FINISHED -> ++finished
+                WorkerState.TERMINATED -> ++terminated
             }
         }
-
-        return "${super.toString()}[core pool size = $corePoolSize, " +
-                "CPU workers = $cpuWorkers, " +
-                "blocking workers = $blockingWorkers, " +
-                "parked workers = $parkedWorkers, " +
-                "retired workers = $retired, " +
-                "finished workers = $finished, " +
+        val state = controlState.value
+        return "$schedulerName@$hexAddress[" +
+                "Pool Size {" +
+                    "core = $corePoolSize, " +
+                    "max = $maxPoolSize}, " +
+                "Worker States {" +
+                    "CPU = $cpuWorkers, " +
+                    "blocking = $blockingWorkers, " +
+                    "parked = $parkedWorkers, " +
+                    "retired = $retired, " +
+                    "terminated = $terminated}, " +
                 "running workers queues = $queueSizes, "+
-                "global queue size = ${globalQueue.size}], " +
-                "control state: ${controlState.value}"
+                "global queue size = ${globalQueue.size}, " +
+                "Control State Workers {" +
+                    "created = ${createdWorkers(state)}, " +
+                    "blocking = ${blockingWorkers(state)}}" +
+                "]"
     }
 
-    // todo: make name of the pool configurable (optional parameter to CoroutineScheduler) and base thread names on it
-    internal inner class Worker(sequenceNumber: Int) : Thread("CoroutineScheduler-worker-$sequenceNumber") {
+    internal inner class Worker private constructor() : Thread() {
         init {
             isDaemon = true
         }
 
         // guarded by scheduler lock
-        private var indexInArray = sequenceNumber
+        private var indexInArray = -1
+            set(index) {
+                name = "$schedulerName-worker-${if (index < 0) "TERMINATED" else index.toString()}"
+                field = index
+            }
+
+        constructor(index: Int) : this() {
+            indexInArray = index
+        }
+
         val localQueue: WorkQueue = WorkQueue()
 
         /**
@@ -552,15 +568,16 @@ internal class CoroutineScheduler(
         private var lastExhaustionTime = 0L
 
         @Volatile // Required for concurrent idleResetBeforeUnpark
-        private var spins = 0L
-        private var yields = 0L // TODO replace with IntPair when inline classes arrive
+        private var spins = 0 // spins until MAX_SPINS, then yields until MAX_YIELDS
 
+        // Note: it is concurrently reset by idleResetBeforeUnpark
         private var parkTimeNs = MIN_PARK_TIME_NS
+        
         private var rngState = random.nextInt()
 
         override fun run() {
             var wasIdle = false // local variable to avoid extra idleReset invocations when tasks repeatedly arrive
-            while (!isTerminated.value && state != WorkerState.FINISHED) {
+            while (!isTerminated.value && state != WorkerState.TERMINATED) {
                 val task = findTask()
                 if (task == null) {
                     // Wait for a job with potential park
@@ -581,7 +598,7 @@ internal class CoroutineScheduler(
                 }
             }
 
-            tryReleaseCpu(WorkerState.FINISHED)
+            tryReleaseCpu(WorkerState.TERMINATED)
         }
 
         private fun runSafely(block: Runnable) {
@@ -655,20 +672,16 @@ internal class CoroutineScheduler(
              * The main idea is not to park while it's possible (otherwise throughput on asymmetric workloads suffers due to too frequent
              * park/unpark calls and delays between job submission and thread queue checking)
              */
-            when {
-                // Volatile read spins, shall be read first
-                spins < MAX_SPINS -> ++spins
-                yields <= MAX_YIELDS -> {
-                    ++yields
-                    yield()
-                }
-                else -> {
-                    if (parkTimeNs < MAX_PARK_TIME_NS) {
-                        parkTimeNs = (parkTimeNs * 3 shr 1).coerceAtMost(MAX_PARK_TIME_NS)
-                    }
-                    tryReleaseCpu(WorkerState.PARKING)
-                    doPark(parkTimeNs)
+            val spins = this.spins // volatile read
+            if (spins <= MAX_YIELDS) {
+                this.spins = spins + 1 // volatile write
+                if (spins >= MAX_SPINS) yield()
+            } else {
+                if (parkTimeNs < MAX_PARK_TIME_NS) {
+                    parkTimeNs = (parkTimeNs * 3 ushr 1).coerceAtMost(MAX_PARK_TIME_NS)
                 }
+                tryReleaseCpu(WorkerState.PARKING)
+                doPark(parkTimeNs.toLong())
             }
         }
 
@@ -712,12 +725,14 @@ internal class CoroutineScheduler(
                  * Now move last worker into an index in array that was previously occupied by this worker.
                  */
                 val lastWorkerIndex = decrementCreatedWorkers()
-                val worker = workers[lastWorkerIndex]!!
-                workers[indexInArray] = worker
-                worker.indexInArray = indexInArray
+                val lastWorker = workers[lastWorkerIndex]!!
+                workers[indexInArray] = lastWorker
+                lastWorker.indexInArray = indexInArray
                 workers[lastWorkerIndex] = null
+                // Cleanup index of this worker for debugging purposes
+                indexInArray = -1
             }
-            state = WorkerState.FINISHED
+            state = WorkerState.TERMINATED
         }
 
         /**
@@ -741,14 +756,12 @@ internal class CoroutineScheduler(
                 state = WorkerState.BLOCKING
                 parkTimeNs = MIN_PARK_TIME_NS
             }
-            yields = 0
             spins = 0
         }
 
         // It is invoked by other thread before this worker is unparked
         fun idleResetBeforeUnpark() {
             parkTimeNs = MIN_PARK_TIME_NS
-            yields = 0
             spins = 0 // Volatile write, should be written last
         }
 
@@ -816,6 +829,6 @@ internal class CoroutineScheduler(
         /**
          * Terminal state, will no longer be used
          */
-        FINISHED
+        TERMINATED
     }
 }

core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/scheduling/ExperimentalCoroutineDispatcher.kt

@@ -14,7 +14,6 @@ class ExperimentalCoroutineDispatcher(
     corePoolSize: Int = CORE_POOL_SIZE,
     maxPoolSize: Int = MAX_POOL_SIZE
 ) : CoroutineDispatcher(), Delay, Closeable {
-
     private val coroutineScheduler = CoroutineScheduler(corePoolSize, maxPoolSize)
 
     override fun dispatch(context: CoroutineContext, block: Runnable): Unit = coroutineScheduler.dispatch(block)
@@ -45,7 +44,11 @@ class ExperimentalCoroutineDispatcher(
     internal fun dispatchBlocking(block: Runnable, context: TaskMode, fair: Boolean): Unit = coroutineScheduler.dispatch(block, context, fair)
 }
 
-private class LimitingBlockingDispatcher(val parallelism: Int, val taskContext: TaskMode, val dispatcher: ExperimentalCoroutineDispatcher) : CoroutineDispatcher(), Delay {
+private class LimitingBlockingDispatcher(
+    val parallelism: Int,
+    val taskContext: TaskMode,
+    val dispatcher: ExperimentalCoroutineDispatcher
+) : CoroutineDispatcher(), Delay {
 
     private val queue = ConcurrentLinkedQueue<Runnable>()
     private val inFlightTasks = atomic(0)
@@ -145,5 +148,6 @@ private class LimitingBlockingDispatcher(val parallelism: Int, val taskContext:
         }
     }
 
-    override fun scheduleResumeAfterDelay(time: Long, unit: TimeUnit, continuation: CancellableContinuation<Unit>) = dispatcher.scheduleResumeAfterDelay(time, unit, continuation)
+    override fun scheduleResumeAfterDelay(time: Long, unit: TimeUnit, continuation: CancellableContinuation<Unit>) =
+        dispatcher.scheduleResumeAfterDelay(time, unit, continuation)
 }