mirror of
https://github.com/vitorpamplona/amethyst.git
synced 2026-08-08 23:54:39 +00:00
feat: add dispatch-stage benchmark (post-parse, pre-verify path)
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 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_018saXqYfAa3RvSJoDXK591R
This commit is contained in:
Claude
@@ -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.
|
||||
|
||||
+310
@@ -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<Event> 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<List<Event>> = Channel(Channel.UNLIMITED)
|
||||
val handedOff = AtomicLong(0)
|
||||
|
||||
// one batch buffer per feeder thread; flushed when full
|
||||
private val buffer = ThreadLocal.withInitial { ArrayList<Event>(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<String>?) -> 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<String>? = 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<String>?) -> Unit = { _, _, _ -> }
|
||||
|
||||
private val sinkDedup: (Event, Handoff?, MutableSet<String>?) -> Unit = { event, _, seen ->
|
||||
seen!!.add(event.id)
|
||||
}
|
||||
|
||||
private val sinkDedupHandoff: (Event, Handoff?, MutableSet<String>?) -> Unit = { event, handoff, seen ->
|
||||
if (seen!!.add(event.id)) handoff!!.offer(event)
|
||||
}
|
||||
|
||||
private val sinkHandoffOnly: (Event, Handoff?, MutableSet<String>?) -> 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<IRelayClient> = 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)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user