Fixed according to the code review

Author: fmixing

benchmarks/scripts/generate_plots_flow_flatten_merge.py

@@ -40,7 +40,7 @@ def draw(data, plt):
         gen_marker = next(marker_gen)
         res = data[(data.flows == flows)]
         plt.plot(res.concurrency, res.score*elements, label="flows={}".format(flows), color=gen_colour, marker=gen_marker)
-        plt.errorbar(x=res.concurrency, y=res.score*elements, yerr=res.score_error*elements, solid_capstyle='projecting', capsize=5)
+        plt.errorbar(x=res.concurrency, y=res.score*elements, yerr=res.score_error*elements, solid_capstyle='projecting', capsize=5, color=gen_colour)
 
 data = pd.read_table(input_file, sep=",")
 data = data[csv_columns].rename(columns=rename_columns)

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

@@ -1,6 +1,6 @@
 package benchmarks
 
-import doWork
+import doGeomDistrWork
 import kotlinx.coroutines.*
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.scheduling.ExperimentalCoroutineDispatcher
@@ -50,9 +50,9 @@ open class SemaphoreBenchmark {
             jobs += GlobalScope.launch {
                 repeat(n) {
                     semaphore.withPermit {
-                        doWork(WORK_INSIDE)
+                        doGeomDistrWork(WORK_INSIDE)
                     }
-                    doWork(WORK_OUTSIDE)
+                    doGeomDistrWork(WORK_OUTSIDE)
                 }
             }
         }
@@ -68,9 +68,9 @@ open class SemaphoreBenchmark {
             jobs += GlobalScope.launch {
                 repeat(n) {
                     semaphore.send(Unit) // acquire
-                    doWork(WORK_INSIDE)
+                    doGeomDistrWork(WORK_INSIDE)
                     semaphore.receive() // release
-                    doWork(WORK_OUTSIDE)
+                    doGeomDistrWork(WORK_OUTSIDE)
                 }
             }
         }
@@ -85,5 +85,4 @@ enum class SemaphoreBenchDispatcherCreator(val create: (parallelism: Int) -> Cor
 
 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/FlowFlattenMergeBenchmark.kt

@@ -1,15 +1,23 @@
+/*
+ * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
 package benchmarks.flow
 
-import doWork
+import doGeomDistrWork
+import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.flow.*
 import kotlinx.coroutines.runBlocking
 import org.openjdk.jmh.annotations.*
 import java.util.concurrent.TimeUnit
 
-@Warmup(iterations = 3)
-@Measurement(iterations = 10)
+/**
+ * Benchmark to measure performance of [kotlinx.coroutines.flow.FlowKt.flattenMerge].
+ * In addition to that, it can be considered as a macro benchmark for the [kotlinx.coroutines.sync.Semaphore]
+ */
+@Warmup(iterations = 3, time = 1)
+@Measurement(iterations = 10, time = 1)
 @BenchmarkMode(Mode.Throughput)
-@OutputTimeUnit(TimeUnit.MICROSECONDS)
+@OutputTimeUnit(TimeUnit.SECONDS)
 @State(Scope.Benchmark)
 @Fork(1)
 open class FlowFlattenMergeBenchmark {
@@ -35,18 +43,19 @@ open class FlowFlattenMergeBenchmark {
         flow = (1..n).asFlow().map {
             flow {
                 repeat(flowElementsToProcess) {
-                    doWork(WORK)
+                    doGeomDistrWork(WORK)
                     emit(it)
                 }
             }
         }
     }
 
     @Benchmark
-    fun flattenMerge() = runBlocking {
+    fun flattenMerge() = runBlocking(Dispatchers.Default) {
         flow.flattenMerge(concurrency = concurrency).collect()
     }
 }
 
+// If you change this variable please be sure that you change variable elements in the corresponding python script as well
 private const val ELEMENTS = 100_000
 private const val WORK = 80

benchmarks/src/main/kotlin/BenchmarkUtils.kt

@@ -4,7 +4,7 @@
 
 import java.util.concurrent.ThreadLocalRandom
 
-fun doWork(work: Int) {
+fun doGeomDistrWork(work: Int) {
     // We use geometric distribution here
     val p = 1.0 / work
     val r = ThreadLocalRandom.current()