Adaptive spinning, parking and load balancing mechanism:

Spin and park adaptively to reduce CPU consumption
Provide mechanism to work offloading when worker detects it's overloaded
Implement hand-rolled thread-local xorshift random, which appears to be 15% faster than ThreadLocal<Random> (ThreadLocalRandom is unavailable on Java 1.6)
Use queue size as metrics for work stealing in addition to task deadline

Author: qwwdfsad

benchmarks/src/jmh/kotlin/benchmarks/LaunchBenchmark.kt

@@ -1,25 +1,24 @@
 package benchmarks
 
 import kotlinx.coroutines.experimental.launch
-import kotlinx.coroutines.experimental.scheduling.ForkedMarker
 import org.openjdk.jmh.annotations.*
 import java.util.concurrent.CyclicBarrier
 import java.util.concurrent.TimeUnit
 
 /*
  * Benchmark to measure scheduling overhead in comparison with FJP.
- * LaunchBenchmark.massiveLaunch  experimental  avgt   30  187.342 ± 20.244  us/op
+ * LaunchBenchmark.massiveLaunch  experimental  avgt   30  328.662 ± 52.789  us/op
  * LaunchBenchmark.massiveLaunch           fjp  avgt   30  179.762 ±  3.931  us/op
  */
 @Warmup(iterations = 10, time = 1, timeUnit = TimeUnit.SECONDS)
-@Measurement(iterations = 100000, time = 1, timeUnit = TimeUnit.SECONDS)
-@Fork(value = 3, jvmArgsAppend = ["-XX:+PreserveFramePointer"])
+@Measurement(iterations = 10, time = 1, timeUnit = TimeUnit.SECONDS)
+@Fork(value = 2)
 @BenchmarkMode(Mode.AverageTime)
 @OutputTimeUnit(TimeUnit.MICROSECONDS)
 @State(Scope.Benchmark)
 open class LaunchBenchmark : ParametrizedDispatcherBase() {
 
-    @Param("experimental")
+    @Param("experimental", "fjp")
     override var dispatcher: String = "fjp"
 
     private val jobsToLaunch = 100
@@ -31,7 +30,7 @@ open class LaunchBenchmark : ParametrizedDispatcherBase() {
     @Benchmark
     fun massiveLaunch() {
         repeat(submitters) {
-            launch(benchmarkContext + ForkedMarker) {
+            launch(benchmarkContext) {
                 // Wait until all cores are occupied
                 allLaunched.await()
                 allLaunched.reset()

benchmarks/src/jmh/kotlin/benchmarks/ParametrizedDispatcherBase.kt

@@ -18,15 +18,15 @@ import kotlin.coroutines.experimental.CoroutineContext
 abstract class ParametrizedDispatcherBase {
 
     abstract var dispatcher: String
-    lateinit var benchmarkContext: CoroutineContext // coroutinesContext clashes with scope parameter
+    lateinit var benchmarkContext: CoroutineContext // coroutineContext clashes with scope parameter
     var closeable: Closeable? = null
 
     @Setup
     open fun setup() {
         benchmarkContext = when {
             dispatcher == "fjp" -> CommonPool
             dispatcher == "experimental" -> {
-                ExperimentalCoroutineDispatcher(CORES_COUNT)
+                ExperimentalCoroutineDispatcher(CORES_COUNT).also { closeable = it }
             }
             dispatcher.startsWith("ftp") -> {
                 newFixedThreadPoolContext(dispatcher.substring(4).toInt(), dispatcher).also { closeable = it }

benchmarks/src/jmh/kotlin/benchmarks/actors/PingPongActorBenchmark.kt

@@ -11,11 +11,13 @@ import kotlin.coroutines.experimental.CoroutineContext
 
 /*
  * Benchmark                                   (dispatcher)  Mode  Cnt    Score    Error  Units
+ * PingPongActorBenchmark.coresCountPingPongs  experimental  avgt   10  185.066 ± 21.692  ms/op
  * PingPongActorBenchmark.coresCountPingPongs           fjp  avgt   10  200.581 ± 22.669  ms/op
  * PingPongActorBenchmark.coresCountPingPongs         ftp_1  avgt   10  494.334 ± 27.450  ms/op
  * PingPongActorBenchmark.coresCountPingPongs         ftp_2  avgt   10  498.754 ± 27.743  ms/op
  * PingPongActorBenchmark.coresCountPingPongs         ftp_8  avgt   10  804.498 ± 69.826  ms/op
  *
+ * PingPongActorBenchmark.singlePingPong       experimental  avgt   10   45.521 ±  3.281  ms/op
  * PingPongActorBenchmark.singlePingPong                fjp  avgt   10  217.005 ± 18.693  ms/op
  * PingPongActorBenchmark.singlePingPong              ftp_1  avgt   10   57.632 ±  1.835  ms/op
  * PingPongActorBenchmark.singlePingPong              ftp_2  avgt   10  112.723 ±  5.280  ms/op
@@ -30,7 +32,7 @@ import kotlin.coroutines.experimental.CoroutineContext
 open class PingPongActorBenchmark : ParametrizedDispatcherBase() {
     data class Letter(val message: Any?, val sender: SendChannel<Letter>)
 
-    @Param("fjp", "ftp_1", "ftp_2", "ftp_8", "experimental")
+    @Param("experimental")
     override var dispatcher: String = "fjp"
 
     @Benchmark

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

@@ -1,10 +1,12 @@
 package kotlinx.coroutines.experimental.scheduling
 
+import kotlinx.atomicfu.atomic
 import kotlinx.coroutines.experimental.Runnable
 import java.io.Closeable
 import java.util.*
 import java.util.concurrent.ConcurrentLinkedQueue
 import java.util.concurrent.Executor
+import java.util.concurrent.TimeUnit
 import java.util.concurrent.locks.LockSupport
 
 /**
@@ -13,9 +15,27 @@ import java.util.concurrent.locks.LockSupport
 class CoroutineScheduler(private val corePoolSize: Int) : Executor, Closeable {
 
     private val workers: Array<PoolWorker>
-    private val globalWorkQueue: Queue<Task> = ConcurrentLinkedQueue<Task>()
+    private val globalWorkQueue = ConcurrentLinkedQueue<Task>()
+    private val parkedWorkers = atomic(0)
+    private val random = Random()
+
     @Volatile
-    private var isClosed = false
+    private var isTerminated = false
+
+    companion object {
+        private const val STEAL_ATTEMPTS = 4
+        private const val MAX_SPINS = 1000L
+        private const val MAX_YIELDS = 500L
+        @JvmStatic
+        private val MAX_PARK_TIME_NS = TimeUnit.SECONDS.toNanos(1)
+        @JvmStatic
+        private val MIN_PARK_TIME_NS = (WORK_STEALING_TIME_RESOLUTION_NS / 4).coerceAtLeast(10).coerceAtMost(MAX_PARK_TIME_NS)
+
+        // Local queue 'offer' results
+        private const val ADDED = -1
+        private const val ADDED_WITH_OFFLOADING = 0 // Added to the local queue, but pool requires additional worker to keep up
+        private const val NOT_ADDED = 1
+    }
 
     init {
         require(corePoolSize >= 1, { "Expected positive core pool size, but was $corePoolSize" })
@@ -26,41 +46,111 @@ class CoroutineScheduler(private val corePoolSize: Int) : Executor, Closeable {
     override fun execute(command: Runnable) = dispatch(command)
 
     override fun close() {
-        isClosed = true
+        isTerminated = true
     }
 
-    fun dispatch(command: Runnable, intensive: Boolean = false) {
-        val task = TimedTask(System.nanoTime(), command)
-        if (!submitToLocalQueue(task, intensive)) {
+    fun dispatch(command: Runnable) {
+        val task = TimedTask(schedulerTimeSource.nanoTime(), command)
+
+        val offerResult = submitToLocalQueue(task)
+        if (offerResult == ADDED) {
+            return
+        }
+
+        if (offerResult == NOT_ADDED) {
             globalWorkQueue.add(task)
         }
+
+        unparkIdleWorker()
     }
 
-    private fun submitToLocalQueue(task: Task, intensive: Boolean): Boolean {
-        val worker = Thread.currentThread() as? PoolWorker ?: return false
-        if (intensive && worker.localQueue.bufferSize > FORKED_TASK_OFFLOAD_THRESHOLD) return false
-        worker.localQueue.offer(task, globalWorkQueue)
-        return true
+    private fun unparkIdleWorker() {
+        // If no threads are parked don't try to wake anyone
+        val parked = parkedWorkers.value
+        if (parked == 0) {
+            return
+        }
+
+        // Try to wake one worker
+        repeat(STEAL_ATTEMPTS) {
+            val victim = workers[random.nextInt(workers.size)]
+            if (victim.isParking) {
+                /*
+                 * Benign data race, victim can wake up after this check, but before 'unpark' call succeeds,
+                 * making first 'park' in next idle period a no-op
+                 */
+                LockSupport.unpark(victim)
+                return
+            }
+        }
+    }
+
+
+    private fun submitToLocalQueue(task: Task): Int {
+        val worker = Thread.currentThread() as? PoolWorker ?: return NOT_ADDED
+        if (worker.localQueue.offer(task, globalWorkQueue)) {
+            // We're close to queue capacity, wakeup anyone to steal
+            if (worker.localQueue.bufferSize > QUEUE_SIZE_OFFLOAD_THRESHOLD) {
+                return ADDED_WITH_OFFLOADING
+            }
+
+            return ADDED
+        }
+
+        return ADDED_WITH_OFFLOADING
+    }
+
+    /**
+     * Returns a string identifying state of this scheduler for nicer debugging
+     */
+    override fun toString(): String {
+        var parkedWorkers = 0
+        val queueSizes = arrayListOf<Int>()
+        for (worker in workers) {
+            if (worker.isParking) {
+                ++parkedWorkers
+            } else {
+                queueSizes += worker.localQueue.bufferSize
+            }
+        }
+
+        return "${super.toString()}[core pool size = ${workers.size}, " +
+                "parked workers = $parkedWorkers, " +
+                "active workers buffer sizes = $queueSizes, " +
+                "global queue size = ${globalWorkQueue.size}]"
     }
 
-    private inner class PoolWorker(index: Int) : Thread("CoroutinesScheduler-worker-$index") {
+
+    internal inner class PoolWorker(index: Int) : Thread("CoroutinesScheduler-worker-$index") {
         init {
             isDaemon = true
         }
 
         val localQueue: WorkQueue = WorkQueue()
+        /**
+         * Time of last call to [unparkIdleWorker] due to missing tasks deadlines.
+         * Used as throttling mechanism to avoid unparking multiple threads when it's not really necessary.
+         */
+        private var lastExhaustionTime = 0L
 
         @Volatile
-        var yields = 0
+        var isParking = false
+        @Volatile
+        private var spins = 0L
+        private var yields = 0L
+        private var parkTimeNs = MIN_PARK_TIME_NS
+        private var rngState = random.nextInt()
 
         override fun run() {
-            while (!isClosed) {
+            while (!isTerminated) {
                 try {
                     val job = findTask()
                     if (job == null) {
-                        awaitWork()
+                        // Wait for a job with potential park
+                        idle()
                     } else {
-                        yields = 0
+                        idleReset()
+                        checkExhaustion(job)
                         job.task.run()
                     }
                 } catch (e: Throwable) {
@@ -69,15 +159,77 @@ class CoroutineScheduler(private val corePoolSize: Int) : Executor, Closeable {
             }
         }
 
-        private fun awaitWork() {
-            // Temporary solution
-            if (++yields > 100000) {
-                LockSupport.parkNanos(WORK_STEALING_TIME_RESOLUTION / 2)
+        private fun checkExhaustion(job: Task) {
+            val parked = parkedWorkers.value
+            if (parked == 0) {
+                return
+            }
+
+            // Check last exhaustion time to avoid the race between steal and next task execution
+            val now = schedulerTimeSource.nanoTime()
+            if (now - job.submissionTime >= WORK_STEALING_TIME_RESOLUTION_NS && now - lastExhaustionTime >= WORK_STEALING_TIME_RESOLUTION_NS * 5) {
+                lastExhaustionTime = now
+                unparkIdleWorker()
+            }
+        }
+
+        /*
+         * Marsaglia xorshift RNG with period 2^32-1 for work stealing purposes.
+         * ThreadLocalRandom cannot be used to support Android and ThreadLocal<Random> is up to 15% slower on ktor benchmarks
+         */
+        internal fun nextInt(upperBound: Int): Int {
+            rngState = rngState xor (rngState shl 13)
+            rngState = rngState xor (rngState shr 17)
+            rngState = rngState xor (rngState shl 5)
+            val mask = upperBound - 1
+            // Fast path for power of two bound
+            if (mask and upperBound == 0) {
+                return rngState and mask
             }
+
+            return (rngState and Int.MAX_VALUE) % upperBound
+        }
+
+        private fun idle() {
+            /*
+             * Simple adaptive await of work:
+             * Spin on the volatile field with an empty loop in hope that new work will arrive,
+             * then start yielding to reduce CPU pressure, and finally start adaptive parking.
+             *
+             * 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 {
+                spins < MAX_SPINS -> ++spins
+                ++yields <= MAX_YIELDS -> Thread.yield()
+                else -> {
+                    if (!isParking) {
+                        isParking = true
+                        parkedWorkers.incrementAndGet()
+                    }
+
+                    if (parkTimeNs < MAX_PARK_TIME_NS) {
+                        parkTimeNs = (parkTimeNs * 1.5).toLong().coerceAtMost(MAX_PARK_TIME_NS)
+                    }
+
+                    LockSupport.parkNanos(parkTimeNs)
+                }
+            }
+        }
+
+        private fun idleReset() {
+            if (isParking) {
+                isParking = false
+                parkTimeNs = MIN_PARK_TIME_NS
+                parkedWorkers.decrementAndGet()
+            }
+
+            spins = 0
+            yields = 0
         }
 
         private fun findTask(): Task? {
-            // TODO explain, probabilistic check with park counter?
+            // TODO probabilistic check if thread is not idle?
             var task: Task? = globalWorkQueue.poll()
             if (task != null) return task
 
@@ -92,16 +244,17 @@ class CoroutineScheduler(private val corePoolSize: Int) : Executor, Closeable {
                 return null
             }
 
-            while (true) {
-                val worker = workers[RANDOM_PROVIDER().nextInt(workers.size)]
+            // Probe a couple of workers
+            repeat(STEAL_ATTEMPTS) {
+                val worker = workers[nextInt(workers.size)]
                 if (worker !== this) {
-                    worker.localQueue.offloadWork(true) {
-                        localQueue.offer(it, globalWorkQueue)
+                    if (localQueue.trySteal(worker.localQueue, globalWorkQueue)) {
+                        return@repeat
                     }
-
-                    return localQueue.poll()
                 }
             }
+
+            return localQueue.poll()
         }
     }
 }

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

@@ -3,29 +3,23 @@ package kotlinx.coroutines.experimental.scheduling
 import kotlinx.coroutines.experimental.*
 import java.io.Closeable
 import java.util.concurrent.TimeUnit
-import kotlin.coroutines.experimental.AbstractCoroutineContextElement
 import kotlin.coroutines.experimental.CoroutineContext
 
-/**
- * Unstable API and subject to change.
- * Context marker which gives scheduler a hint that submitted jobs can be distributed among cores aggressively.
- * Usually it's useful for massive jobs submission produced by single coroutine, e.g. data intensive fork-join tasks
- * or fan-out notifications for a large number of listeners.
- */
-object ForkedMarker : AbstractCoroutineContextElement(ForkedKey)
-
-private object ForkedKey : CoroutineContext.Key<ForkedMarker>
 
 class ExperimentalCoroutineDispatcher(threads: Int = Runtime.getRuntime().availableProcessors()) : CoroutineDispatcher(), Delay, Closeable {
 
     private val coroutineScheduler = CoroutineScheduler(threads)
 
     override fun dispatch(context: CoroutineContext, block: Runnable) {
-        coroutineScheduler.dispatch(block, context[ForkedKey] != null)
+        coroutineScheduler.dispatch(block)
     }
 
     override fun scheduleResumeAfterDelay(time: Long, unit: TimeUnit, continuation: CancellableContinuation<Unit>) =
             DefaultExecutor.scheduleResumeAfterDelay(time, unit, continuation)
 
     override fun close() = coroutineScheduler.close()
+    override fun toString(): String {
+        return "${super.toString()}[scheduler = $coroutineScheduler]"
+    }
+
 }