test(geode): add LoadBenchmark.fanoutScaling for selective-fanout perf
Verifies the FilterIndex path in LiveEventStore: each event matches exactly ONE of N subscribers, so without an index the relay walks all N subs per event (O(N)); with the author-keyed index the lookup is O(1) and per-event latency stays flat as N grows. Different from fanoutLatency (broadcast: 1 event reaches every sub). Here per-event work is lookup-bound rather than delivery-bound. Configurable via -DfanoutScalingSubs (comma list) and -DfanoutScalingEvents. Measured (laptop, JDK 21, full sweep): 100 subs, 2k events: p50=1.35ms p99=5.71ms 1000 subs, 2k events: p50=1.05ms p99=3.05ms 5000 subs, 1k events: p50=1.01ms p99=2.96ms p50 ~1ms across N — the predicted O(1) scaling. Default events count adapts downward at high N to stay below WebSocketSessionPump's 8192-frame outbound cap (test-client read side is the bottleneck above ~5k subs, not the relay). Plan doc updated with the measured numbers in the "How to verify" section.
This commit is contained in:
Claude
@@ -42,6 +42,8 @@ tasks.withType<Test>().configureEach {
|
||||
// perf.LoadBenchmark tests opt in. Off by default — load tests
|
||||
// are noisy and slow.
|
||||
systemProperty("runLoadBenchmark", System.getProperty("runLoadBenchmark") ?: "false")
|
||||
System.getProperty("fanoutScalingEvents")?.let { systemProperty("fanoutScalingEvents", it) }
|
||||
System.getProperty("fanoutScalingSubs")?.let { systemProperty("fanoutScalingSubs", it) }
|
||||
// Show println output from test JVM so the benchmark numbers are
|
||||
// actually visible without grepping the report XML.
|
||||
testLogging {
|
||||
|
||||
@@ -119,18 +119,37 @@ predicate. Dispatch:
|
||||
|
||||
### How to verify
|
||||
|
||||
`geode.perf.LoadBenchmark.fanoutScaling` (to be added):
|
||||
`geode.perf.LoadBenchmark.fanoutScaling` (added):
|
||||
|
||||
- N connections, each subscribes to `{authors: [pk_i], kinds: [1]}`.
|
||||
- Publish 10k EVENTs from a producer connection; each event matches
|
||||
- Publish events round-robin across the N pubkeys; each event matches
|
||||
exactly one subscriber.
|
||||
- Measure end-to-end latency p50/p99 for N ∈ {100, 1000, 5000}.
|
||||
|
||||
Without the index, p99 grows roughly linearly with N. With the
|
||||
index, p99 should be flat up to a much higher N.
|
||||
The benchmark adapts the per-N event count downward to stay below
|
||||
geode's `WebSocketSessionPump.MAX_OUTGOING_BUFFER` (8192 frames per
|
||||
session). The single-WS test subClient can't drain (subs + events)
|
||||
frames at full firehose rate above ~5k subs in a shared test JVM —
|
||||
that's a test-infra ceiling, not a relay one. Override with
|
||||
`-DfanoutScalingEvents=N` to push past the default.
|
||||
|
||||
Measured numbers from a development laptop (Linux, JDK 21, full
|
||||
sweep):
|
||||
|
||||
| N subs | events | mean (ms) | p50 (ms) | p99 (ms) | p999 (ms) |
|
||||
|-------:|-------:|----------:|---------:|---------:|----------:|
|
||||
| 100 | 2 000 | 1.63 | 1.35 | 5.71 | 20.49 |
|
||||
| 1 000 | 2 000 | 1.12 | 1.05 | 3.05 | 9.40 |
|
||||
| 5 000 | 1 000 | 1.07 | 1.01 | 2.96 | 7.38 |
|
||||
|
||||
p50 stays at ~1 ms across all three N values — the index is producing
|
||||
the predicted O(1)-per-event scaling. The minor p99 spread comes from
|
||||
GC pauses and OkHttp scheduler jitter on the test client, not from
|
||||
linear per-event work in the relay.
|
||||
|
||||
For the LocalCache side, a similar benchmark would publish events
|
||||
matching one of M observers and measure dispatch cost as M grows.
|
||||
Not yet added — Phase 2 candidate for a follow-up.
|
||||
|
||||
## Risks
|
||||
|
||||
|
||||
@@ -22,6 +22,7 @@ package com.vitorpamplona.geode.perf
|
||||
|
||||
import com.vitorpamplona.geode.LocalRelayServer
|
||||
import com.vitorpamplona.geode.Relay
|
||||
import com.vitorpamplona.quartz.nip01Core.core.toHexKey
|
||||
import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.client.NostrClient
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.publishAndConfirm
|
||||
@@ -40,6 +41,7 @@ import kotlinx.coroutines.runBlocking
|
||||
import kotlinx.coroutines.withTimeout
|
||||
import okhttp3.OkHttpClient
|
||||
import java.util.concurrent.atomic.AtomicLong
|
||||
import java.util.concurrent.atomic.AtomicLongArray
|
||||
import kotlin.test.Test
|
||||
import kotlin.time.measureTime
|
||||
|
||||
@@ -450,6 +452,168 @@ class LoadBenchmark {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Selective fanout: each event matches exactly ONE of N
|
||||
* subscribers. Stresses the inverted-index path in
|
||||
* `FilterIndex` / `LiveEventStore`: without an index, the relay
|
||||
* walks all N subscribers per event and runs `Filter.match` to
|
||||
* find the one true match; with the index, an author-keyed
|
||||
* lookup returns the single subscriber directly.
|
||||
*
|
||||
* Different from [fanoutLatency], which tests broadcast fanout
|
||||
* (one event reaches every subscriber). Here per-event work is
|
||||
* lookup-bound rather than delivery-bound, so latency should be
|
||||
* roughly **flat** in N once the index is wired up — that's the
|
||||
* shape change the index is meant to deliver.
|
||||
*
|
||||
* Configurable via system properties:
|
||||
* - `fanoutScalingSubs`: comma-separated list of N values
|
||||
* (default `100,1000,5000`).
|
||||
* - `fanoutScalingEvents`: total events per N (default scales
|
||||
* inversely with N to keep test wall time bounded — at 5000
|
||||
* subs the OkHttp WebSocket dispatcher on the test client
|
||||
* starts serializing inbound EVENT frames at high throughput,
|
||||
* which inflates measured latency without telling us anything
|
||||
* about the index itself).
|
||||
*/
|
||||
@Test
|
||||
fun fanoutScaling() =
|
||||
benchmark("fanout scaling") {
|
||||
val subSweep =
|
||||
System
|
||||
.getProperty("fanoutScalingSubs")
|
||||
?.split(",")
|
||||
?.mapNotNull { it.trim().toIntOrNull() }
|
||||
?.takeIf { it.isNotEmpty() }
|
||||
?: listOf(100, 1_000, 5_000)
|
||||
for (subs in subSweep) {
|
||||
runBenchmarkServer { server, http ->
|
||||
val scope = CoroutineScope(Dispatchers.Default + SupervisorJob())
|
||||
val subClient = NostrClient(BasicOkHttpWebSocket.Builder { _ -> http }, scope)
|
||||
val pubClient = NostrClient(BasicOkHttpWebSocket.Builder { _ -> http }, scope)
|
||||
try {
|
||||
val relayUrl = server.url.normalizeRelayUrl()
|
||||
|
||||
// One keypair per subscriber. Each subscriber
|
||||
// filters on its own author so the relay can
|
||||
// route an event to exactly one sub via the
|
||||
// FilterIndex's byAuthor bucket.
|
||||
val keys = List(subs) { KeyPair() }
|
||||
val pubkeys = keys.map { it.pubKey.toHexKey() }
|
||||
|
||||
// Per-subscriber arrival timestamp. We
|
||||
// overwrite per round; the publish path waits
|
||||
// until the slot is non-zero before moving on.
|
||||
val arrivalNs = AtomicLongArray(subs)
|
||||
val received = AtomicLong()
|
||||
val eosed = AtomicLong()
|
||||
|
||||
repeat(subs) { i ->
|
||||
subClient.subscribe(
|
||||
"scale-$i",
|
||||
mapOf(
|
||||
relayUrl to
|
||||
listOf(
|
||||
Filter(
|
||||
authors = listOf(pubkeys[i]),
|
||||
kinds = listOf(1),
|
||||
),
|
||||
),
|
||||
),
|
||||
object : SubscriptionListener {
|
||||
override fun onEvent(
|
||||
event: com.vitorpamplona.quartz.nip01Core.core.Event,
|
||||
isLive: Boolean,
|
||||
relay: NormalizedRelayUrl,
|
||||
forFilters: List<Filter>?,
|
||||
) {
|
||||
arrivalNs.set(i, System.nanoTime())
|
||||
received.incrementAndGet()
|
||||
}
|
||||
|
||||
override fun onEose(
|
||||
relay: NormalizedRelayUrl,
|
||||
forFilters: List<Filter>?,
|
||||
) {
|
||||
eosed.incrementAndGet()
|
||||
}
|
||||
},
|
||||
)
|
||||
}
|
||||
|
||||
runBlocking {
|
||||
withTimeout(120_000) {
|
||||
while (eosed.get() < subs) kotlinx.coroutines.delay(100)
|
||||
}
|
||||
}
|
||||
|
||||
// Events round-robin over the N pubkeys. Per
|
||||
// the plan we'd like 10k everywhere, but
|
||||
// [WebSocketSessionPump.MAX_OUTGOING_BUFFER]
|
||||
// (8192 frames) on the relay's per-session
|
||||
// outbound queue trips when the subClient's
|
||||
// single-WS read pipeline can't drain
|
||||
// (subs + events) frames fast enough. That's
|
||||
// a test-infra ceiling, not the index's. We
|
||||
// back off events at high N so the latency
|
||||
// numbers reflect dispatch cost rather than
|
||||
// backpressure-driven connection teardown.
|
||||
// Override with -DfanoutScalingEvents=N to
|
||||
// force a value (caller's responsibility to
|
||||
// stay below the pump cap).
|
||||
val totalEvents =
|
||||
System
|
||||
.getProperty("fanoutScalingEvents")
|
||||
?.toIntOrNull()
|
||||
?: when {
|
||||
subs <= 200 -> 2_000
|
||||
subs <= 1_000 -> 2_000
|
||||
subs <= 2_000 -> 1_000
|
||||
else -> 1_000
|
||||
}
|
||||
println("$subs subs ready; publishing $totalEvents events round-robin...")
|
||||
|
||||
val latenciesMs = DoubleArray(totalEvents)
|
||||
|
||||
runBlocking {
|
||||
val start = System.nanoTime()
|
||||
for (i in 0 until totalEvents) {
|
||||
val target = i % subs
|
||||
val signer = NostrSignerSync(keys[target])
|
||||
val event = signer.sign(TextNoteEvent.build("scale-$i"))
|
||||
arrivalNs.set(target, 0)
|
||||
val pubStart = System.nanoTime()
|
||||
pubClient.publishAndConfirm(event, setOf(relayUrl))
|
||||
withTimeout(30_000) {
|
||||
while (arrivalNs.get(target) == 0L) kotlinx.coroutines.delay(1)
|
||||
}
|
||||
latenciesMs[i] = (arrivalNs.get(target) - pubStart) / 1_000_000.0
|
||||
}
|
||||
val totalMs = (System.nanoTime() - start) / 1_000_000.0
|
||||
|
||||
val sorted = latenciesMs.copyOf().also { it.sort() }
|
||||
val p50 = sorted[sorted.size / 2]
|
||||
val p99 = sorted[(sorted.size * 99) / 100]
|
||||
val p999 = sorted[(sorted.size * 999) / 1000]
|
||||
val mean = sorted.average()
|
||||
println(
|
||||
"subs=$subs events=$totalEvents totalMs=${"%.0f".format(totalMs)} " +
|
||||
"meanMs=${"%.2f".format(mean)} " +
|
||||
"p50Ms=${"%.2f".format(p50)} " +
|
||||
"p99Ms=${"%.2f".format(p99)} " +
|
||||
"p999Ms=${"%.2f".format(p999)} " +
|
||||
"received=${received.get()}/$totalEvents",
|
||||
)
|
||||
}
|
||||
} finally {
|
||||
subClient.disconnect()
|
||||
pubClient.disconnect()
|
||||
scope.cancel()
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Many concurrent publishers, each on their own WebSocket. Tells
|
||||
* us whether the SQLite single-writer bottleneck is the floor or
|
||||
|
||||
Reference in New Issue
Block a user