Setup benchmarks

Author: whyoleg

.github/workflows/ci-benchmarks.yml

@@ -0,0 +1,36 @@
+name: Benchmarks CI
+on:
+  push:
+
+jobs:
+  benchmark:
+    runs-on: ${{ matrix.os }}
+    strategy:
+      fail-fast: false
+      matrix:
+        os: [ ubuntu-latest, macos-latest ]
+    steps:
+      - uses: actions/checkout@v4
+      - uses: ./.github/actions/setup-gradle
+
+      - run: >
+          ./gradlew
+          FastCsvLocalRequestChannelBenchmark
+          FastCsvLocalRequestResponseBenchmark
+          FastCsvLocalRequestStreamBenchmark
+          --dry-run
+
+      - run: >
+          ./gradlew
+          FastCsvLocalRequestChannelBenchmark
+          FastCsvLocalRequestResponseBenchmark
+          FastCsvLocalRequestStreamBenchmark
+          --no-parallel
+          --max-workers=1
+
+      - if: always() && !cancelled()
+        uses: actions/upload-artifact@v4
+        with:
+          name: benchmark-reports-${{ matrix.os }}
+          path: "benchmarks/**/build/reports/benchmarks/**/*.csv"
+          retention-days: 1

benchmarks/build.gradle.kts

@@ -0,0 +1,47 @@
+/*
+ * Copyright 2015-2025 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import kotlinx.benchmark.gradle.*
+import rsocketbuild.*
+
+plugins {
+    id("rsocketbuild.multiplatform-benchmarks")
+}
+
+kotlin {
+    jvmTarget()
+
+    sourceSets {
+        jvmMain.dependencies {
+            implementation(projects.benchmarksShared)
+
+            implementation(libs.kotlinx.coroutines.reactor)
+            implementation(libs.rsocket.java.transport.local)
+            implementation(libs.rsocket.java.transport.netty)
+        }
+    }
+}
+
+benchmark {
+    targets {
+        register("jvm") {
+            this as JvmBenchmarkTarget
+            jmhVersion = libs.versions.jmh.get()
+        }
+    }
+
+    registerBenchmarks("RSocketJava", listOf("local", "tcp", "ws"))
+}

benchmarks/rsocket-java/build.gradle.kts

@@ -0,0 +1,25 @@
+/*
+ * Copyright 2015-2025 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.rsocket.kotlin.benchmarks.java
+
+import io.rsocket.transport.*
+import io.rsocket.transport.local.*
+
+class LocalRSocketJavaBenchmark : RSocketJavaBenchmark() {
+    override val serverTransport: ServerTransport<*> = LocalServerTransport.create("local")
+    override val clientTransport: ClientTransport = LocalClientTransport.create("local")
+}

benchmarks/rsocket-java/src/jvmMain/kotlin/LocalRSocketJavaBenchmark.kt

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2022 the original author or authors.
+ * Copyright 2015-2025 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -14,32 +14,57 @@
  * limitations under the License.
  */
 
-package io.rsocket.kotlin.benchmarks
+package io.rsocket.kotlin.benchmarks.java
 
 import io.rsocket.*
 import io.rsocket.core.*
 import io.rsocket.frame.decoder.*
-import io.rsocket.transport.local.*
+import io.rsocket.kotlin.benchmarks.*
+import io.rsocket.transport.*
 import io.rsocket.util.*
+import kotlinx.benchmark.*
+import kotlinx.benchmark.Mode
 import kotlinx.coroutines.flow.*
 import kotlinx.coroutines.reactive.*
 import org.reactivestreams.*
 import reactor.core.publisher.*
 import kotlin.random.*
 
-class RSocketJavaBenchmark : RSocketBenchmark<Payload>() {
+@BenchmarkMode(Mode.Throughput)
+@Warmup(iterations = WARMUP, time = WARMUP_DURATION)
+@Measurement(iterations = ITERATION, time = ITERATION_DURATION)
+@State(Scope.Benchmark)
+abstract class RSocketJavaBenchmark : RSocketBenchmark<Payload, Blackhole>() {
+    protected abstract val clientTransport: ClientTransport
+    protected abstract val serverTransport: ServerTransport<*>
+
+    private lateinit var payload: Payload
+    private lateinit var payloadMono: Mono<Payload>
+    private lateinit var payloadsFlux: Flux<Payload>
+    private lateinit var payloadsFlow: Flow<Payload>
+    private lateinit var client: RSocket
+    private lateinit var server: Closeable
 
-    lateinit var client: RSocket
-    lateinit var server: Closeable
+    override fun createPayload(size: Int): Payload = if (size == 0) EmptyPayload.INSTANCE else ByteBufPayload.create(
+        ByteArray(size / 2).also { Random.nextBytes(it) },
+        ByteArray(size / 2).also { Random.nextBytes(it) }
+    )
+
+    override fun createPayloadCopy(): Payload = payload.retain()
+
+    override fun releasePayload(payload: Payload) {
+        payload.release()
+    }
 
-    lateinit var payload: Payload
-    lateinit var payloadMono: Mono<Payload>
-    lateinit var payloadsFlux: Flux<Payload>
-    lateinit var payloadsFlow: Flow<Payload>
+    override fun consumePayload(bh: Blackhole, value: Payload) = bh.consume(value)
 
+    override suspend fun doRequestResponse(): Payload = client.requestResponse(payload.retain()).awaitSingle()
+    override fun doRequestStream(): Flow<Payload> = client.requestStream(payload.retain()).asFlow()
+    override fun doRequestChannel(): Flow<Payload> = client.requestChannel(payloadsFlow.asPublisher()).asFlow()
+
+    @Setup
     override fun setup() {
         payload = createPayload(payloadSize)
-
         payloadMono = Mono.fromSupplier(payload::retain)
         payloadsFlux = Flux.range(0, 5000).map { payload.retain() }
         payloadsFlow = flow { repeat(5000) { emit(payload.retain()) } }
@@ -61,33 +86,48 @@ class RSocketJavaBenchmark : RSocketBenchmark<Payload>() {
                 })
         }
             .payloadDecoder(PayloadDecoder.ZERO_COPY)
-            .bind(LocalServerTransport.create("server"))
+            .bind(serverTransport)
             .block()!!
 
         client = RSocketConnector.create()
             .payloadDecoder(PayloadDecoder.ZERO_COPY)
-            .connect(LocalClientTransport.create("server"))
+            .connect(clientTransport)
             .block()!!
     }
 
+    @TearDown
     override fun cleanup() {
         client.dispose()
         server.dispose()
     }
 
-    override fun createPayload(size: Int): Payload = if (size == 0) EmptyPayload.INSTANCE else ByteBufPayload.create(
-        ByteArray(size / 2).also { Random.nextBytes(it) },
-        ByteArray(size / 2).also { Random.nextBytes(it) }
-    )
+    @Param("0")
+    override var payloadSize: Int = 0
 
-    override fun releasePayload(payload: Payload) {
-        payload.release()
-    }
+    @Benchmark
+    override fun requestResponseBlocking(bh: Blackhole) = super.requestResponseBlocking(bh)
 
-    override suspend fun doRequestResponse(): Payload = client.requestResponse(payload.retain()).awaitSingle()
+    @Benchmark
+    override fun requestResponseParallel(bh: Blackhole) = super.requestResponseParallel(bh)
+
+    @Benchmark
+    override fun requestResponseConcurrent(bh: Blackhole) = super.requestResponseConcurrent(bh)
+
+    @Benchmark
+    override fun requestStreamBlocking(bh: Blackhole) = super.requestStreamBlocking(bh)
+
+    @Benchmark
+    override fun requestStreamParallel(bh: Blackhole) = super.requestStreamParallel(bh)
+
+    @Benchmark
+    override fun requestStreamConcurrent(bh: Blackhole) = super.requestStreamConcurrent(bh)
 
-    override suspend fun doRequestStream(): Flow<Payload> = client.requestStream(payload.retain()).asFlow()
+    @Benchmark
+    override fun requestChannelBlocking(bh: Blackhole) = super.requestChannelBlocking(bh)
 
-    override suspend fun doRequestChannel(): Flow<Payload> = client.requestChannel(payloadsFlow.asPublisher()).asFlow()
+    @Benchmark
+    override fun requestChannelParallel(bh: Blackhole) = super.requestChannelParallel(bh)
 
+    @Benchmark
+    override fun requestChannelConcurrent(bh: Blackhole) = super.requestChannelConcurrent(bh)
 }

benchmarks/src/javaMain/kotlin/io/rsocket/kotlin/benchmarks/RSocketJavaBenchmark.kt renamed to benchmarks/rsocket-java/src/jvmMain/kotlin/RSocketJavaBenchmark.kt

@@ -0,0 +1,26 @@
+/*
+ * Copyright 2015-2025 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.rsocket.kotlin.benchmarks.java
+
+import io.rsocket.transport.*
+import io.rsocket.transport.netty.client.*
+import io.rsocket.transport.netty.server.*
+
+class TcpRSocketJavaBenchmark : RSocketJavaBenchmark() {
+    override val serverTransport: ServerTransport<*> = TcpServerTransport.create(9000)
+    override val clientTransport: ClientTransport = TcpClientTransport.create(9000)
+}