Fixed according to the code review

Author: fmixing

benchmarks/build.gradle

@@ -74,4 +74,6 @@ dependencies {
     compile "org.openjdk.jmh:jmh-core:1.21"
     compile 'com.typesafe.akka:akka-actor_2.12:2.5.0'
     compile project(':kotlinx-coroutines-core')
+    // add jmh dependency on main
+    jmh sourceSets.main.runtimeClasspath
 }

benchmarks/plots-generators/generate-flatten-merge-flow-plots.py renamed to benchmarks/scripts/generate_plots_flow_flatten_merge.py

@@ -3,9 +3,9 @@
 from matplotlib.ticker import FormatStrFormatter
 
 inputFile = "jmh-result.csv"
-outputFile = "flatten-merge-plots.svg"
-elements = 10000
-benchmarkName="benchmarks.flow.FlattenMergeBenchmark.flattenMerge"
+outputFile = "flow-flatten-merge.svg"
+elements = 100000
+benchmarkName="benchmarks.flow.FlowFlattenMergeBenchmark.flattenMerge"
 
 markers = ['.', 'v', '^', '1', '2', '8', 'p', 'P', 'x', 'D', 'd', 's']
 colours = ['black', 'silver', 'red', 'gold', 'sienna', 'olivedrab', 'lightseagreen', 'navy', 'blue', 'm', 'crimson', 'yellow', 'orangered', 'slateblue', 'aqua']
@@ -29,7 +29,7 @@ def draw(data, plt):
     plt.gca().xaxis.set_major_formatter(FormatStrFormatter('%0.f'))
     plt.grid(linewidth='0.5', color='lightgray')
     plt.ylabel(data.unit.unique()[0])
-    plt.xlabel('parallelism')
+    plt.xlabel('concurrency')
     plt.xticks(data.concurrency.unique())
 
     colourGen = next_colour()
@@ -39,13 +39,11 @@ def draw(data, plt):
         genMarker = next(markerGen)
         res = data[(data.flows == flows)]
         plt.plot(res.concurrency, res.score*elements, label="flows={}".format(flows), color=genColour, marker=genMarker)
-
-def genFile():
-    data = pd.read_table(inputFile, sep=",", skiprows=1, names=["benchmark","mode","threads","samples","score","scoreError","unit","concurrency","flows"])
-    plt.figure(figsize=(20, 20))
-    draw(data, plt)
-    plt.legend(loc='upper center', borderpad=0, ncol=4, frameon=False, borderaxespad=4, prop={'size': 8})
-    plt.tight_layout(pad=12, w_pad=2, h_pad=1)
-    plt.savefig(outputFile, bbox_inches='tight')
-
-genFile()
+        plt.errorbar(x=res.concurrency, y=res.score*elements, yerr=res.scoreError*elements, solid_capstyle='projecting', capsize=5)
+
+data = pd.read_table(inputFile, sep=",", skiprows=1, names=["benchmark","mode","threads","samples","score","scoreError","unit","concurrency","flows"])
+plt.figure(figsize=(20, 20))
+draw(data, plt)
+plt.legend(loc='upper center', borderpad=0, ncol=4, frameon=False, borderaxespad=4, prop={'size': 8})
+plt.tight_layout(pad=12, w_pad=2, h_pad=1)
+plt.savefig(outputFile, bbox_inches='tight')

benchmarks/src/jmh/kotlin/benchmarks/SemaphoreBenchmark.kt

@@ -1,13 +1,13 @@
 package benchmarks
 
+import doWork
 import kotlinx.coroutines.*
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.scheduling.ExperimentalCoroutineDispatcher
 import kotlinx.coroutines.sync.Semaphore
 import kotlinx.coroutines.sync.withPermit
 import org.openjdk.jmh.annotations.*
 import java.util.concurrent.ForkJoinPool
-import java.util.concurrent.ThreadLocalRandom
 import java.util.concurrent.TimeUnit
 
 @Warmup(iterations = 3, time = 500, timeUnit = TimeUnit.MICROSECONDS)
@@ -83,15 +83,7 @@ enum class SemaphoreBenchDispatcherCreator(val create: (parallelism: Int) -> Cor
     EXPERIMENTAL({ parallelism -> ExperimentalCoroutineDispatcher(corePoolSize = parallelism, maxPoolSize = parallelism) })
 }
 
-fun doWork(work: Int) {
-    // We use geometric distribution here
-    val p = 1.0 / work
-    val r = ThreadLocalRandom.current()
-    while (true) {
-        if (r.nextDouble() < p) break
-    }
-}
-
 private const val WORK_INSIDE = 80
 private const val WORK_OUTSIDE = 40
+// If you change this variable please be sure that you change variable elements in the corresponding python script as well
 private const val BATCH_SIZE = 1000000

benchmarks/src/jmh/kotlin/benchmarks/flow/FlattenMergeBenchmark.kt renamed to benchmarks/src/jmh/kotlin/benchmarks/flow/FlowFlattenMergeBenchmark.kt

@@ -1,23 +1,18 @@
 package benchmarks.flow
 
-import benchmarks.doWork
-import kotlinx.coroutines.*
+import doWork
 import kotlinx.coroutines.flow.*
+import kotlinx.coroutines.runBlocking
 import org.openjdk.jmh.annotations.*
-import java.util.concurrent.ThreadLocalRandom
 import java.util.concurrent.TimeUnit
-import kotlin.math.ceil
 
-/**
- * This benchmark can be considered as a macro benchmark for the [kotlinx.coroutines.sync.Semaphore]
- */
-@Warmup(iterations = 5)
+@Warmup(iterations = 3)
 @Measurement(iterations = 10)
 @BenchmarkMode(Mode.Throughput)
 @OutputTimeUnit(TimeUnit.MICROSECONDS)
 @State(Scope.Benchmark)
 @Fork(1)
-open class FlattenMergeBenchmark {
+open class FlowFlattenMergeBenchmark {
 
     /**
      * Number of flows that are merged in this benchmark. Negative number means that number of flows
@@ -27,20 +22,19 @@ open class FlattenMergeBenchmark {
     private var flows: Int = 0
 
     @Param("1", "2", "4") // local machine
-//    @Param("1", "2", "4", "8", "16", "32", "64", "128", "144") // dasquad
 //    @Param("1", "2", "4", "8", "16", "32", "64", "96") // Google Cloud
     private var concurrency: Int = 0
 
     private lateinit var flow: Flow<Flow<Int>>
 
     @Setup
     fun setup() {
-        val flowsNumber = if (flows >= 0) flows else -(flows * concurrency)
-        val flowElementsToProcess = ceil(ELEMENTS / flowsNumber.toDouble())
+        val n = if (flows >= 0) flows else -(flows * concurrency)
+        val flowElementsToProcess = ELEMENTS / n
 
-        flow = (1..flowsNumber).asFlow().map {
+        flow = (1..n).asFlow().map {
             flow {
-                repeat(flowElementsToProcess.toInt()) {
+                repeat(flowElementsToProcess) {
                     doWork(WORK)
                     emit(it)
                 }

benchmarks/src/main/kotlin/BenchmarkUtils.kt

@@ -0,0 +1,14 @@
+/*
+* Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+*/
+
+import java.util.concurrent.ThreadLocalRandom
+
+fun doWork(work: Int) {
+    // We use geometric distribution here
+    val p = 1.0 / work
+    val r = ThreadLocalRandom.current()
+    while (true) {
+        if (r.nextDouble() < p) break
+    }
+}