Merge pull request #1652 from Kotlin/scheduler-changes

CoroutineScheduler rework

Fixes #840
Fixes #1046
Fixes #1286

Author: qwwdfsad

benchmarks/src/jmh/kotlin/benchmarks/ChannelProducerConsumerBenchmark.kt

@@ -1,3 +1,7 @@
+/*
+ * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
 package benchmarks
 
 import kotlinx.coroutines.*

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

@@ -4,7 +4,7 @@
 
 package benchmarks
 
-import benchmarks.actors.CORES_COUNT
+import benchmarks.akka.CORES_COUNT
 import kotlinx.coroutines.*
 import kotlinx.coroutines.scheduling.*
 import org.openjdk.jmh.annotations.Param
@@ -22,14 +22,14 @@ abstract class ParametrizedDispatcherBase : CoroutineScope {
 
     abstract var dispatcher: String
     override lateinit var coroutineContext: CoroutineContext
-    var closeable: Closeable? = null
+    private var closeable: Closeable? = null
 
-    @UseExperimental(InternalCoroutinesApi::class)
     @Setup
+    @UseExperimental(InternalCoroutinesApi::class)
     open fun setup() {
         coroutineContext = when {
             dispatcher == "fjp" -> ForkJoinPool.commonPool().asCoroutineDispatcher()
-            dispatcher == "experimental" -> {
+            dispatcher == "scheduler" -> {
                 ExperimentalCoroutineDispatcher(CORES_COUNT).also { closeable = it }
             }
             dispatcher.startsWith("ftp") -> {

benchmarks/src/jmh/kotlin/benchmarks/actors/PingPongAkkaBenchmark.kt renamed to benchmarks/src/jmh/kotlin/benchmarks/akka/PingPongAkkaBenchmark.kt

@@ -1,8 +1,8 @@
 /*
- * Copyright 2016-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
  */
 
-package benchmarks.actors
+package benchmarks.akka
 
 import akka.actor.ActorRef
 import akka.actor.ActorSystem
@@ -13,7 +13,6 @@ import org.openjdk.jmh.annotations.*
 import scala.concurrent.Await
 import scala.concurrent.duration.Duration
 import java.util.concurrent.CountDownLatch
-import java.util.concurrent.TimeUnit
 
 const val N_MESSAGES = 100_000
 
@@ -29,12 +28,12 @@ class Stop
  * PingPongAkkaBenchmark.singlePingPong             default-dispatcher  avgt   10  173.742 ± 41.984  ms/op
  * PingPongAkkaBenchmark.singlePingPong       single-thread-dispatcher  avgt   10   24.181 ±  0.730  ms/op
  */
-@Warmup(iterations = 5, time = 1, timeUnit = TimeUnit.SECONDS)
-@Measurement(iterations = 5, time = 1, timeUnit = TimeUnit.SECONDS)
-@Fork(value = 2)
-@BenchmarkMode(Mode.AverageTime)
-@OutputTimeUnit(TimeUnit.MILLISECONDS)
-@State(Scope.Benchmark)
+//@Warmup(iterations = 5, time = 1, timeUnit = TimeUnit.SECONDS)
+//@Measurement(iterations = 5, time = 1, timeUnit = TimeUnit.SECONDS)
+//@Fork(value = 2)
+//@BenchmarkMode(Mode.AverageTime)
+//@OutputTimeUnit(TimeUnit.MILLISECONDS)
+//@State(Scope.Benchmark)
 open class PingPongAkkaBenchmark {
 
     lateinit var system: ActorSystem
@@ -62,12 +61,12 @@ open class PingPongAkkaBenchmark {
         Await.ready(system.terminate(), Duration.Inf())
     }
 
-    @Benchmark
+//    @Benchmark
     fun singlePingPong() {
         runPingPongs(1)
     }
 
-    @Benchmark
+//    @Benchmark
     fun coresCountPingPongs() {
         runPingPongs(Runtime.getRuntime().availableProcessors())
     }

benchmarks/src/jmh/kotlin/benchmarks/actors/StatefulActorAkkaBenchmark.kt renamed to benchmarks/src/jmh/kotlin/benchmarks/akka/StatefulActorAkkaBenchmark.kt

@@ -1,8 +1,8 @@
 /*
- * Copyright 2016-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
  */
 
-package benchmarks.actors
+package benchmarks.akka
 
 import akka.actor.ActorRef
 import akka.actor.ActorSystem
@@ -14,7 +14,6 @@ import scala.concurrent.Await
 import scala.concurrent.duration.Duration
 import java.util.concurrent.CountDownLatch
 import java.util.concurrent.ThreadLocalRandom
-import java.util.concurrent.TimeUnit
 
 const val ROUNDS = 10_000
 const val STATE_SIZE = 1024
@@ -38,12 +37,12 @@ val CORES_COUNT = Runtime.getRuntime().availableProcessors()
  * StatefulActorAkkaBenchmark.singleComputationSingleRequestor              default-dispatcher  avgt   14   39.964 ±  2.343  ms/op
  * StatefulActorAkkaBenchmark.singleComputationSingleRequestor        single-thread-dispatcher  avgt   14   10.214 ±  2.152  ms/op
  */
-@Warmup(iterations = 7, time = 1, timeUnit = TimeUnit.SECONDS)
-@Measurement(iterations = 7, time = 1, timeUnit = TimeUnit.SECONDS)
-@Fork(value = 2)
-@BenchmarkMode(Mode.AverageTime)
-@OutputTimeUnit(TimeUnit.MILLISECONDS)
-@State(Scope.Benchmark)
+//@Warmup(iterations = 7, time = 1, timeUnit = TimeUnit.SECONDS)
+//@Measurement(iterations = 7, time = 1, timeUnit = TimeUnit.SECONDS)
+//@Fork(value = 2)
+//@BenchmarkMode(Mode.AverageTime)
+//@OutputTimeUnit(TimeUnit.MILLISECONDS)
+//@State(Scope.Benchmark)
 open class StatefulActorAkkaBenchmark {
 
     lateinit var system: ActorSystem
@@ -72,22 +71,22 @@ open class StatefulActorAkkaBenchmark {
         Await.ready(system.terminate(), Duration.Inf())
     }
 
-    @Benchmark
+//    @Benchmark
     fun singleComputationSingleRequestor() {
         run(1, 1)
     }
 
-    @Benchmark
+//    @Benchmark
     fun singleComputationMultipleRequestors() {
         run(1, CORES_COUNT)
     }
 
-    @Benchmark
+//    @Benchmark
     fun multipleComputationsSingleRequestor() {
         run(CORES_COUNT, 1)
     }
 
-    @Benchmark
+//    @Benchmark
     fun multipleComputationsMultipleRequestors() {
         run(CORES_COUNT, CORES_COUNT)
     }
@@ -120,7 +119,8 @@ open class StatefulActorAkkaBenchmark {
 
     private fun createComputationActors(initLatch: CountDownLatch, count: Int): List<ActorRef> {
         return (0 until count).map {
-            system.actorOf(Props.create(ComputationActor::class.java,
+            system.actorOf(Props.create(
+                ComputationActor::class.java,
                     LongArray(STATE_SIZE) { ThreadLocalRandom.current().nextLong(0, 100) }, initLatch)
                     .withDispatcher("akka.actor.$dispatcher"))
         }

benchmarks/src/jmh/kotlin/benchmarks/scheduler/DispatchersContextSwitchBenchmark.kt

@@ -0,0 +1,73 @@
+/*
+ * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package benchmarks.scheduler
+
+import benchmarks.akka.*
+import kotlinx.coroutines.*
+import org.openjdk.jmh.annotations.*
+import org.openjdk.jmh.annotations.State
+import java.lang.Thread.*
+import java.util.concurrent.*
+import kotlin.concurrent.*
+import kotlin.coroutines.*
+
+@Warmup(iterations = 5, time = 1, timeUnit = TimeUnit.SECONDS)
+@Measurement(iterations = 5, time = 1, timeUnit = TimeUnit.SECONDS)
+@Fork(value = 1)
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.MILLISECONDS)
+@State(Scope.Thread)
+open class DispatchersContextSwitchBenchmark {
+    private val nCoroutines = 10000
+    private val delayTimeMs = 1L
+    private val nRepeatDelay = 10
+
+    private val fjp = ForkJoinPool.commonPool().asCoroutineDispatcher()
+    private val ftp = Executors.newFixedThreadPool(CORES_COUNT - 1).asCoroutineDispatcher()
+
+    @TearDown
+    fun teardown() {
+        ftp.close()
+        (ftp.executor as ExecutorService).awaitTermination(1, TimeUnit.SECONDS)
+    }
+
+    @Benchmark
+    fun coroutinesIoDispatcher() =  runBenchmark(Dispatchers.IO)
+
+    @Benchmark
+    fun coroutinesDefaultDispatcher() = runBenchmark(Dispatchers.Default)
+
+    @Benchmark
+    fun coroutinesFjpDispatcher()  = runBenchmark(fjp)
+
+    @Benchmark
+    fun coroutinesFtpDispatcher()  = runBenchmark(ftp)
+
+    @Benchmark
+    fun coroutinesBlockingDispatcher() = runBenchmark(EmptyCoroutineContext)
+
+    @Benchmark
+    fun threads() {
+        val threads = List(nCoroutines) {
+            thread(start = true) {
+                repeat(nRepeatDelay) {
+                    sleep(delayTimeMs)
+                }
+            }
+        }
+        threads.forEach { it.join() }
+    }
+
+    private fun runBenchmark(dispatcher: CoroutineContext)  = runBlocking {
+        repeat(nCoroutines) {
+            launch(dispatcher) {
+                repeat(nRepeatDelay) {
+                    delay(delayTimeMs)
+                }
+            }
+        }
+    }
+}
+

benchmarks/src/jmh/kotlin/benchmarks/ForkJoinBenchmark.kt renamed to benchmarks/src/jmh/kotlin/benchmarks/scheduler/ForkJoinBenchmark.kt

@@ -1,9 +1,10 @@
 /*
- * Copyright 2016-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
  */
 
-package benchmarks
+package benchmarks.scheduler
 
+import benchmarks.*
 import kotlinx.coroutines.*
 import org.openjdk.jmh.annotations.*
 import java.util.concurrent.*
@@ -44,7 +45,7 @@ open class ForkJoinBenchmark : ParametrizedDispatcherBase() {
     }
 
     lateinit var coefficients: LongArray
-    override var dispatcher: String = "experimental"
+    override var dispatcher: String = "scheduler"
 
     @Setup
     override fun setup() {
@@ -129,8 +130,18 @@ open class ForkJoinBenchmark : ParametrizedDispatcherBase() {
             } else {
                 pendingCount = 2
                 // One may fork only once here and executing second task here with looping over firstComplete to be even more efficient
-                first = RecursiveAction(coefficients, start, start + (end - start) / 2, parent = this).fork()
-                second = RecursiveAction(coefficients, start + (end - start) / 2, end, parent = this).fork()
+                first = RecursiveAction(
+                    coefficients,
+                    start,
+                    start + (end - start) / 2,
+                    parent = this
+                ).fork()
+                second = RecursiveAction(
+                    coefficients,
+                    start + (end - start) / 2,
+                    end,
+                    parent = this
+                ).fork()
             }
 
             tryComplete()

benchmarks/src/jmh/kotlin/benchmarks/LaunchBenchmark.kt renamed to benchmarks/src/jmh/kotlin/benchmarks/scheduler/LaunchBenchmark.kt

@@ -1,9 +1,10 @@
 /*
- * Copyright 2016-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
  */
 
-package benchmarks
+package benchmarks.scheduler
 
+import benchmarks.*
 import kotlinx.coroutines.*
 import org.openjdk.jmh.annotations.*
 import java.util.concurrent.*
@@ -21,7 +22,7 @@ import java.util.concurrent.*
 @State(Scope.Benchmark)
 open class LaunchBenchmark : ParametrizedDispatcherBase() {
 
-    @Param("experimental", "fjp")
+    @Param("scheduler", "fjp")
     override var dispatcher: String = "fjp"
 
     private val jobsToLaunch = 100

benchmarks/src/jmh/kotlin/benchmarks/StatefulAwaitsBenchmark.kt renamed to benchmarks/src/jmh/kotlin/benchmarks/scheduler/StatefulAwaitsBenchmark.kt

@@ -1,9 +1,10 @@
 /*
- * Copyright 2016-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
  */
 
-package benchmarks
+package benchmarks.scheduler
 
+import benchmarks.*
 import kotlinx.coroutines.*
 import kotlinx.coroutines.channels.*
 import org.openjdk.jmh.annotations.*
@@ -52,7 +53,7 @@ open class StatefulAsyncBenchmark : ParametrizedDispatcherBase() {
     @Param("1", "8", "16")
     var jobsCount = 1
 
-    @Param("fjp", "ftp_1", "ftp_8")
+    @Param("fjp", "ftp_1", "dispatcher")
     override var dispatcher: String = "fjp"
 
     @Volatile

benchmarks/src/jmh/kotlin/benchmarks/actors/ConcurrentStatefulActorBenchmark.kt renamed to benchmarks/src/jmh/kotlin/benchmarks/scheduler/actors/ConcurrentStatefulActorBenchmark.kt

@@ -2,10 +2,11 @@
  * Copyright 2016-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
  */
 
-package benchmarks.actors
+package benchmarks.scheduler.actors
 
 import benchmarks.*
-import benchmarks.actors.StatefulActorBenchmark.*
+import benchmarks.akka.*
+import benchmarks.scheduler.actors.StatefulActorBenchmark.*
 import kotlinx.coroutines.*
 import kotlinx.coroutines.channels.*
 import org.openjdk.jmh.annotations.*
@@ -57,18 +58,18 @@ import java.util.concurrent.*
 @State(Scope.Benchmark)
 open class ConcurrentStatefulActorBenchmark : ParametrizedDispatcherBase() {
 
-    @Param("1024", "8192", "262144")
+    @Param("1024", "8192")
     var stateSize: Int = -1
 
-    @Param("fjp", "ftp_1", "ftp_8", "experimental")
+    @Param("fjp", "scheduler")
     override var dispatcher: String = "fjp"
 
     @Benchmark
     fun multipleComputationsUnfair() = runBlocking {
         val resultChannel: Channel<Unit> = Channel(1)
         val computations = (0 until CORES_COUNT).map { computationActor(stateSize) }
         val requestor = requestorActorUnfair(computations, resultChannel)
-        requestor.send(Letter(Start(), Channel(0)))
+        requestor.send(Letter(Start(), requestor))
         resultChannel.receive()
     }
 
@@ -77,7 +78,7 @@ open class ConcurrentStatefulActorBenchmark : ParametrizedDispatcherBase() {
         val resultChannel: Channel<Unit> = Channel(1)
         val computations = (0 until CORES_COUNT).map { computationActor(stateSize) }
         val requestor = requestorActorFair(computations, resultChannel)
-        requestor.send(Letter(Start(), Channel(0)))
+        requestor.send(Letter(Start(), requestor))
         resultChannel.receive()
     }
 
@@ -95,6 +96,7 @@ open class ConcurrentStatefulActorBenchmark : ParametrizedDispatcherBase() {
                     }
                     is Long -> {
                         if (++received >= ROUNDS * 8) {
+                            computations.forEach { it.close() }
                             stopChannel.send(Unit)
                             return@actor
                         } else {
@@ -122,6 +124,7 @@ open class ConcurrentStatefulActorBenchmark : ParametrizedDispatcherBase() {
                     }
                     is Long -> {
                         if (++receivedTotal >= ROUNDS * computations.size) {
+                            computations.forEach { it.close() }
                             stopChannel.send(Unit)
                             return@actor
                         } else {
@@ -136,4 +139,4 @@ open class ConcurrentStatefulActorBenchmark : ParametrizedDispatcherBase() {
                 }
             }
         }
-}
+}