Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
5 changed files
with
162 additions
and
14 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,75 @@ | ||
# To run this script run the command 'python3 scripts/generate_plots_flow_flatten_merge.py' in the /benchmarks folder | ||
|
||
|
||
import pandas as pd | ||
import sys | ||
import locale | ||
import matplotlib.pyplot as plt | ||
from matplotlib.ticker import FormatStrFormatter | ||
|
||
input_file = "build/reports/jmh/results.csv" | ||
output_file = "out/flow-flatten-merge.svg" | ||
# Please change the value of this variable according to the FlowFlattenMergeBenchmarkKt.ELEMENTS | ||
elements = 100000 | ||
benchmark_name = "benchmarks.flow.FlowFlattenMergeBenchmark.flattenMerge" | ||
csv_columns = ["Benchmark", "Score", "Unit", "Param: concurrency", "Param: flowsNumberStrategy"] | ||
rename_columns = {"Benchmark": "benchmark", "Score" : "score", "Unit" : "unit", | ||
"Param: concurrency" : "concurrency", "Param: flowsNumberStrategy" : "flows"} | ||
|
||
markers = ['.', 'v', '^', '1', '2', '8', 'p', 'P', 'x', 'D', 'd', 's'] | ||
colours = ['red', 'gold', 'sienna', 'olivedrab', 'lightseagreen', 'navy', 'blue', 'm', 'crimson', 'yellow', 'orangered', 'slateblue', 'aqua', 'black', 'silver'] | ||
|
||
def next_colour(): | ||
i = 0 | ||
while True: | ||
yield colours[i % len(colours)] | ||
i += 1 | ||
|
||
def next_marker(): | ||
i = 0 | ||
while True: | ||
yield markers[i % len(markers)] | ||
i += 1 | ||
|
||
def draw(data, plt): | ||
plt.xscale('log', basex=2) | ||
plt.gca().xaxis.set_major_formatter(FormatStrFormatter('%0.f')) | ||
plt.grid(linewidth='0.5', color='lightgray') | ||
if data.unit.unique()[0] != "ops/s": | ||
print("Unexpected time unit: " + data.unit.unique()[0]) | ||
sys.exit(1) | ||
plt.ylabel("elements / ms") | ||
plt.xlabel('concurrency') | ||
plt.xticks(data.concurrency.unique()) | ||
|
||
colour_gen = next_colour() | ||
marker_gen = next_marker() | ||
for flows in data.flows.unique(): | ||
gen_colour = next(colour_gen) | ||
gen_marker = next(marker_gen) | ||
res = data[(data.flows == flows)] | ||
# plt.plot(res.concurrency, res.score*elements/1000, label="flows={}".format(flows), color=gen_colour, marker=gen_marker) | ||
plt.errorbar(x=res.concurrency, y=res.score*elements/1000, yerr=res.score_error*elements/1000, solid_capstyle='projecting', | ||
label="flows={}".format(flows), capsize=4, color=gen_colour, linewidth=2.2) | ||
|
||
langlocale = locale.getdefaultlocale()[0] | ||
locale.setlocale(locale.LC_ALL, langlocale) | ||
dp = locale.localeconv()['decimal_point'] | ||
if dp == ",": | ||
csv_columns.append("Score Error (99,9%)") | ||
rename_columns["Score Error (99,9%)"] = "score_error" | ||
elif dp == ".": | ||
csv_columns.append("Score Error (99.9%)") | ||
rename_columns["Score Error (99.9%)"] = "score_error" | ||
else: | ||
print("Unexpected locale delimeter: " + dp) | ||
sys.exit(1) | ||
data = pd.read_csv(input_file, sep=",", decimal=dp) | ||
data = data[csv_columns].rename(columns=rename_columns) | ||
data = data[(data.benchmark == benchmark_name)] | ||
plt.rcParams.update({'font.size': 15}) | ||
plt.figure(figsize=(12.5, 10)) | ||
draw(data, plt) | ||
plt.legend(loc='upper center', borderpad=0, bbox_to_anchor=(0.5, 1.3), ncol=2, frameon=False, borderaxespad=2, prop={'size': 15}) | ||
plt.tight_layout(pad=12, w_pad=2, h_pad=1) | ||
plt.savefig(output_file, bbox_inches='tight') |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
63 changes: 63 additions & 0 deletions
63
benchmarks/src/jmh/kotlin/benchmarks/flow/FlowFlattenMergeBenchmark.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,63 @@ | ||
/* | ||
* Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license. | ||
*/ | ||
|
||
package benchmarks.flow | ||
|
||
import benchmarks.common.* | ||
import kotlinx.coroutines.* | ||
import kotlinx.coroutines.flow.* | ||
import org.openjdk.jmh.annotations.* | ||
import java.util.concurrent.* | ||
|
||
/** | ||
* 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 = 5, time = 1) | ||
@Measurement(iterations = 5, time = 1) | ||
@BenchmarkMode(Mode.Throughput) | ||
@OutputTimeUnit(TimeUnit.SECONDS) | ||
@State(Scope.Benchmark) | ||
@Fork(1) | ||
open class FlowFlattenMergeBenchmark { | ||
@Param | ||
private var flowsNumberStrategy: FlowsNumberStrategy = FlowsNumberStrategy.`10xConcurrency flows` | ||
|
||
@Param("1", "2", "4", "8") | ||
private var concurrency: Int = 0 | ||
|
||
private lateinit var flow: Flow<Flow<Int>> | ||
|
||
@Setup | ||
fun setup() { | ||
val n = flowsNumberStrategy.get(concurrency) | ||
val flowElementsToProcess = ELEMENTS / n | ||
|
||
flow = (1..n).asFlow().map { | ||
flow { | ||
repeat(flowElementsToProcess) { | ||
doGeomDistrWork(WORK) | ||
emit(it) | ||
} | ||
} | ||
} | ||
} | ||
|
||
@Benchmark | ||
fun flattenMerge() = runBlocking(Dispatchers.Default) { | ||
flow.flattenMerge(concurrency = concurrency).collect() | ||
} | ||
} | ||
|
||
enum class FlowsNumberStrategy(val get: (concurrency: Int) -> Int) { | ||
`10xConcurrency flows`({ concurrency -> concurrency * 10 }), | ||
`1xConcurrency flows`({ it }), | ||
`100 flows`({ 100 }), | ||
`500 flows`({ 500 }) | ||
} | ||
|
||
// If you change this variable please be sure that you change variable elements in the generate_plots_flow_flatten_merge.py | ||
// python script as well | ||
private const val ELEMENTS = 100_000 | ||
private const val WORK = 100 |
17 changes: 17 additions & 0 deletions
17
benchmarks/src/main/kotlin/benchmarks/common/BenchmarkUtils.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,17 @@ | ||
/* | ||
* Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license. | ||
*/ | ||
|
||
package benchmarks.common | ||
|
||
import java.util.concurrent.* | ||
|
||
fun doGeomDistrWork(work: Int) { | ||
// We use geometric distribution here. We also checked on macbook pro 13" (2017) that the resulting work times | ||
// are distributed geometrically, see https://github.com/Kotlin/kotlinx.coroutines/pull/1464#discussion_r355705325 | ||
val p = 1.0 / work | ||
val r = ThreadLocalRandom.current() | ||
while (true) { | ||
if (r.nextDouble() < p) break | ||
} | ||
} |