perf(benchmark): refine JMH benchmark suite

wow-benchmarks/gradle/jmh-packaging.gradle.kts

@@ -0,0 +1,189 @@
+import groovy.json.JsonOutput
+import groovy.json.JsonSlurper
+import java.util.zip.ZipFile
+
+val jmhRuntimeClasspath = configurations.named("jmhRuntimeClasspath")
+
+/**
+ * Merge all META-INF/wow-metadata.json from the runtime classpath into a single valid JSON.
+ * Without this, the JMH JAR contains duplicate entries that concatenate into invalid JSON,
+ * causing MetadataSearcher to fail silently at runtime.
+ *
+ * Aggregate entries are deep-merged (non-null values take precedence) to avoid the API module's
+ * type=null overwriting the domain module's type=full.class.Name.
+ */
+val mergedWowMetadata = layout.buildDirectory.file("tmp/wow-metadata-merged/META-INF/wow-metadata.json")
+val mergeWowMetadata = tasks.register("mergeWowMetadata") {
+    description = "Merge all wow-metadata.json from classpath into a single file."
+    outputs.file(mergedWowMetadata)
+    inputs.files(jmhRuntimeClasspath)
+    doLast {
+        val parser = JsonSlurper()
+        val merged = mutableMapOf<String, Any>()
+        val metadataContents = mutableListOf<String>()
+
+        fun deepMutable(value: Any?): Any? = when (value) {
+            is Map<*, *> -> value.entries.associate { entry ->
+                entry.key.toString() to deepMutable(entry.value)
+            }.toMutableMap()
+
+            is List<*> -> value.map { deepMutable(it) }.toMutableList()
+            else -> value
+        }
+
+        jmhRuntimeClasspath.get().resolve()
+            .filter { it.name.endsWith(".jar") || it.isDirectory }
+            .forEach { file ->
+                if (file.isDirectory) {
+                    val meta = file.resolve("META-INF/wow-metadata.json")
+                    if (meta.exists()) {
+                        metadataContents.add(meta.readText())
+                    }
+                } else {
+                    ZipFile(file).use { zip ->
+                        zip.getEntry("META-INF/wow-metadata.json")?.let { entry ->
+                            metadataContents.add(zip.getInputStream(entry).bufferedReader().readText())
+                        }
+                    }
+                }
+            }
+
+        for (text in metadataContents) {
+            try {
+                @Suppress("UNCHECKED_CAST")
+                val next = parser.parseText(text) as Map<String, Any>
+                @Suppress("UNCHECKED_CAST")
+                val contexts = next["contexts"] as? Map<String, Any> ?: continue
+                @Suppress("UNCHECKED_CAST")
+                val mergedContexts = merged.getOrPut("contexts") { mutableMapOf<String, Any>() } as MutableMap<String, Any>
+                for ((ctxName, ctxValue) in contexts) {
+                    val existing = mergedContexts[ctxName]
+                    if (existing == null) {
+                        mergedContexts[ctxName] = deepMutable(ctxValue)!!
+                    } else {
+                        @Suppress("UNCHECKED_CAST")
+                        val existingMap = existing as MutableMap<String, Any>
+                        @Suppress("UNCHECKED_CAST")
+                        val newMap = ctxValue as Map<String, Any>
+                        // Deep-merge aggregates (non-null values win)
+                        @Suppress("UNCHECKED_CAST")
+                        val existingAggregates = existingMap.getOrPut("aggregates") { mutableMapOf<String, Any>() } as MutableMap<String, Any>
+                        @Suppress("UNCHECKED_CAST")
+                        val newAggregates = newMap["aggregates"] as? Map<String, Any> ?: emptyMap()
+                        for ((aggName, aggValue) in newAggregates) {
+                            val existingAgg = existingAggregates[aggName]
+                            if (existingAgg == null) {
+                                existingAggregates[aggName] = deepMutable(aggValue)!!
+                            } else {
+                                @Suppress("UNCHECKED_CAST")
+                                val existingAggMap = existingAgg as MutableMap<String, Any>
+                                @Suppress("UNCHECKED_CAST")
+                                val newAggMap = aggValue as Map<String, Any?>
+                                for ((key, value) in newAggMap) {
+                                    if (value != null) {
+                                        val existingVal = existingAggMap[key]
+                                        if (existingVal == null || value is String) {
+                                            existingAggMap[key] = value
+                                        } else if (value is List<*>) {
+                                            @Suppress("UNCHECKED_CAST")
+                                            val existingList = (existingVal as? List<String> ?: emptyList()).toMutableList()
+                                            @Suppress("UNCHECKED_CAST")
+                                            val newList = value as List<String>
+                                            existingList.addAll(newList.filter { it !in existingList })
+                                            existingAggMap[key] = existingList
+                                        }
+                                    }
+                                }
+                            }
+                        }
+                        // Merge scopes (union)
+                        @Suppress("UNCHECKED_CAST")
+                        val existingScopes = existingMap.getOrPut("scopes") { mutableListOf<String>() } as MutableList<String>
+                        @Suppress("UNCHECKED_CAST")
+                        val newScopes = newMap["scopes"] as? List<String> ?: emptyList()
+                        existingScopes.addAll(newScopes.filter { it !in existingScopes })
+                    }
+                }
+            } catch (_: Exception) {
+                // skip invalid metadata
+            }
+        }
+
+        val outputFile = mergedWowMetadata.get().asFile
+        outputFile.parentFile.mkdirs()
+        outputFile.writeText(JsonOutput.prettyPrint(JsonOutput.toJson(merged)))
+        logger.lifecycle("Merged ${metadataContents.size} wow-metadata.json files into ${outputFile.absolutePath}")
+    }
+}
+
+val mergedJmhServicesRoot = layout.buildDirectory.dir("tmp/jmh-services-merged")
+val jmhServiceFilesToMerge = listOf(
+    "META-INF/services/tools.jackson.databind.JacksonModule",
+    "META-INF/services/me.ahoo.wow.id.GlobalIdGeneratorFactory",
+)
+val mergeJmhServices = tasks.register("mergeJmhServices") {
+    description = "Merge critical SPI service files from the JMH runtime classpath."
+    outputs.dir(mergedJmhServicesRoot)
+    inputs.files(jmhRuntimeClasspath)
+    inputs.files(jmhServiceFilesToMerge.map { layout.projectDirectory.file("src/jmh/resources/$it") })
+    doLast {
+        val outputRoot = mergedJmhServicesRoot.get().asFile
+        outputRoot.deleteRecursively()
+
+        fun MutableSet<String>.addServiceProviders(text: String) {
+            text.lineSequence()
+                .map { it.substringBefore('#').trim() }
+                .filter { it.isNotEmpty() }
+                .forEach { add(it) }
+        }
+
+        for (servicePath in jmhServiceFilesToMerge) {
+            val providers = linkedSetOf<String>()
+            jmhRuntimeClasspath.get().resolve()
+                .filter { it.name.endsWith(".jar") || it.isDirectory }
+                .forEach { file ->
+                    if (file.isDirectory) {
+                        val serviceFile = file.resolve(servicePath)
+                        if (serviceFile.exists()) {
+                            providers.addServiceProviders(serviceFile.readText())
+                        }
+                    } else {
+                        ZipFile(file).use { zip ->
+                            zip.getEntry(servicePath)?.let { entry ->
+                                providers.addServiceProviders(zip.getInputStream(entry).bufferedReader().readText())
+                            }
+                        }
+                    }
+                }
+
+            val localServiceFile = file("src/jmh/resources/$servicePath")
+            if (localServiceFile.exists()) {
+                providers.addServiceProviders(localServiceFile.readText())
+            }
+
+            if (providers.isNotEmpty()) {
+                val outputFile = outputRoot.resolve(servicePath)
+                outputFile.parentFile.mkdirs()
+                outputFile.writeText(providers.joinToString(System.lineSeparator(), postfix = System.lineSeparator()))
+            }
+        }
+    }
+}
+
+tasks.named<Jar>("jmhJar") {
+    isZip64 = true
+    dependsOn(mergeWowMetadata)
+    dependsOn(mergeJmhServices)
+    // Exclude all wow-metadata.json from dependency JARs (they're duplicated)
+    exclude("META-INF/wow-metadata.json")
+    // Add the merged metadata file
+    from(mergedWowMetadata) {
+        into("META-INF")
+    }
+    eachFile {
+        if (path in jmhServiceFilesToMerge && !file.absolutePath.startsWith(mergedJmhServicesRoot.get().asFile.absolutePath)) {
+            exclude()
+        }
+    }
+    from(mergedJmhServicesRoot)
+}

wow-benchmarks/src/jmh/kotlin/me/ahoo/wow/BenchmarkAggregateSchedulerSupplier.kt

@@ -0,0 +1,38 @@
+/*
+ * Copyright [2021-present] [ahoo wang <ahoowang@qq.com> (https://github.com/Ahoo-Wang)].
+ * 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 me.ahoo.wow
+
+import me.ahoo.wow.api.modeling.NamedAggregate
+import me.ahoo.wow.modeling.MaterializedNamedAggregate
+import me.ahoo.wow.modeling.materialize
+import me.ahoo.wow.scheduler.AggregateSchedulerSupplier
+import reactor.core.publisher.Mono
+import reactor.core.scheduler.Scheduler
+import reactor.core.scheduler.Schedulers
+import java.util.concurrent.ConcurrentHashMap
+
+class BenchmarkAggregateSchedulerSupplier : AggregateSchedulerSupplier {
+    private val schedulers: MutableMap<MaterializedNamedAggregate, Scheduler> = ConcurrentHashMap()
+
+    override fun getOrInitialize(namedAggregate: NamedAggregate): Scheduler =
+        schedulers.computeIfAbsent(namedAggregate.materialize()) {
+            Schedulers.newParallel("CommandDispatcherBenchmark-${it.aggregateName}", Schedulers.DEFAULT_POOL_SIZE)
+        }
+
+    override fun stopGracefully(): Mono<Void> =
+        Mono.fromRunnable {
+            schedulers.values.forEach(Scheduler::dispose)
+            schedulers.clear()
+        }
+}

wow-benchmarks/src/jmh/kotlin/me/ahoo/wow/BenchmarkGlobalIdGeneratorFactory.kt

@@ -0,0 +1,25 @@
+/*
+ * Copyright [2021-present] [ahoo wang <ahoowang@qq.com> (https://github.com/Ahoo-Wang)].
+ * 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 me.ahoo.wow
+
+import me.ahoo.cosid.cosid.ClockSyncCosIdGenerator
+import me.ahoo.cosid.cosid.CosIdGenerator
+import me.ahoo.cosid.cosid.Radix62CosIdGenerator
+import me.ahoo.wow.id.GlobalIdGeneratorFactory
+
+class BenchmarkGlobalIdGeneratorFactory : GlobalIdGeneratorFactory {
+    override fun create(): CosIdGenerator {
+        return ClockSyncCosIdGenerator(Radix62CosIdGenerator(0))
+    }
+}

wow-benchmarks/src/jmh/kotlin/me/ahoo/wow/command/BloomFilterIdempotencyCheckerBenchmark.kt

@@ -35,4 +35,4 @@ open class BloomFilterIdempotencyCheckerBenchmark {
         val result = idempotencyChecker.check(generateGlobalId()).block()
         blackhole.consume(result)
     }
-}
+}

wow-benchmarks/src/jmh/kotlin/me/ahoo/wow/command/Commands.kt

@@ -19,18 +19,17 @@ import me.ahoo.wow.api.command.CommandMessage
 import me.ahoo.wow.example.api.cart.AddCartItem
 import me.ahoo.wow.example.domain.cart.Cart
 import me.ahoo.wow.example.domain.cart.CartState
+import me.ahoo.wow.id.generateGlobalId
 import me.ahoo.wow.infra.idempotency.BloomFilterIdempotencyChecker
 import me.ahoo.wow.modeling.MaterializedNamedAggregate
 import me.ahoo.wow.modeling.annotation.aggregateMetadata
 import java.time.Duration
-import java.util.concurrent.atomic.AtomicLong
 
 val cartAggregateMetadata by lazy {
     aggregateMetadata<Cart, CartState>()
 }
 
 private val benchmarkCart = MaterializedNamedAggregate("example-service", "cart")
-private val benchmarkIdSequence = AtomicLong()
 const val FIXED_AGGREGATE_ID = "benchmark-cart-fixed-id"
 
 fun createCommandMessage(): CommandMessage<AddCartItem> {
@@ -78,7 +77,7 @@ private fun createCommandMessage(
     )
 }
 
-private fun nextBenchmarkId(): String = "benchmark-${benchmarkIdSequence.incrementAndGet()}"
+private fun nextBenchmarkId(): String = generateGlobalId()
 
 fun createBloomFilterIdempotencyChecker(): BloomFilterIdempotencyChecker {
     return BloomFilterIdempotencyChecker(Duration.ofMinutes(1)) {

wow-benchmarks/src/jmh/kotlin/me/ahoo/wow/eventsourcing/EventStreamFactoryBenchmark.kt

@@ -26,4 +26,10 @@ open class EventStreamFactoryBenchmark {
         val eventStream = createEventStream()
         blackhole.consume(eventStream)
     }
-}
+
+    @Benchmark
+    fun createSingleEventStream(blackhole: Blackhole) {
+        val eventStream = createSingleEventStream()
+        blackhole.consume(eventStream)
+    }
+}

wow-benchmarks/src/jmh/kotlin/me/ahoo/wow/eventsourcing/Events.kt

@@ -26,4 +26,11 @@ fun createEventStream(): DomainEventStream {
     return listOf<Any>(event).toDomainEventStream(
         upstream = GivenInitializationCommand(cartAggregateMetadata.aggregateId()),
     )
-}
+}
+
+fun createSingleEventStream(): DomainEventStream {
+    val event = CartItemAdded(CartItem("productId"))
+    return event.toDomainEventStream(
+        upstream = GivenInitializationCommand(cartAggregateMetadata.aggregateId()),
+    )
+}

wow-benchmarks/src/jmh/kotlin/me/ahoo/wow/eventsourcing/InMemoryEventStoreBenchmark.kt

@@ -14,30 +14,32 @@
 package me.ahoo.wow.eventsourcing
 
 import org.openjdk.jmh.annotations.Benchmark
+import org.openjdk.jmh.annotations.OperationsPerInvocation
 import org.openjdk.jmh.annotations.Scope
 import org.openjdk.jmh.annotations.Setup
 import org.openjdk.jmh.annotations.State
-import org.openjdk.jmh.annotations.TearDown
 
 @State(Scope.Benchmark)
 open class InMemoryEventStoreBenchmark : AbstractEventStoreBenchmark() {
+    private companion object {
+        const val APPENDS_PER_INVOCATION = 1024
+    }
 
     @Setup
     override fun setup() {
         super.setup()
     }
 
-    @TearDown
-    fun tearDown() {
-        setup()
-    }
-
     override fun createEventStore(): EventStore {
         return InMemoryEventStore()
     }
 
     @Benchmark
+    @OperationsPerInvocation(APPENDS_PER_INVOCATION)
     override fun append() {
-        super.append()
+        repeat(APPENDS_PER_INVOCATION) {
+            super.append()
+        }
+        setup()
     }
-}
+}

wow-benchmarks/src/jmh/kotlin/me/ahoo/wow/eventsourcing/NoopEventStoreBenchmark.kt

@@ -15,10 +15,16 @@ package me.ahoo.wow.eventsourcing
 
 import org.openjdk.jmh.annotations.Benchmark
 import org.openjdk.jmh.annotations.Scope
+import org.openjdk.jmh.annotations.Setup
 import org.openjdk.jmh.annotations.State
 
 @State(Scope.Benchmark)
 open class NoopEventStoreBenchmark : AbstractEventStoreBenchmark() {
+    @Setup
+    override fun setup() {
+        super.setup()
+    }
+
     override fun createEventStore(): EventStore {
         return NoopEventStore
     }
@@ -28,4 +34,4 @@ open class NoopEventStoreBenchmark : AbstractEventStoreBenchmark() {
     override fun append() {
         super.append()
     }
-}
+}