audit(geode): drop dead firstSeenNs + clarify heap-vs-RSS in LoadBenchmark

Pre-merge audit findings on the connection-scaling perf changes:

- connectionsHeldOpenWithFanout populated firstSeenNs but never read
  it — only lastReceiveNs feeds the p50/p99 latency metric. Drop the
  unused map and rename the latency map to lastReceiveNs to match
  what it actually holds.
- connectionsHeldOpen10k printed/asserted on a variable named rssMb
  that was actually JVM heap (`Runtime.totalMemory - freeMemory`).
  Rename to heapMb, force a GC + 200 ms settle before reading so the
  number reflects retained bytes rather than connect-ramp churn, and
  update the assertion message to say "JVM heap" not "heap usage".
This commit is contained in:
Claude
2026-05-07 20:50:10 +00:00
parent 64624bdfdb
commit 6c792fdf44
@@ -199,19 +199,25 @@ class LoadBenchmark {
Thread.sleep(50) Thread.sleep(50)
} }
} }
// JVM heap usage, not OS RSS — we can only measure
// what the JVM itself has allocated. Force a GC first
// so the reading reflects retained bytes, not in-flight
// allocation churn from the connect ramp-up.
val rt = Runtime.getRuntime() val rt = Runtime.getRuntime()
val rssMb = (rt.totalMemory() - rt.freeMemory()) / (1024 * 1024) System.gc()
Thread.sleep(200)
val heapMb = (rt.totalMemory() - rt.freeMemory()) / (1024 * 1024)
println( println(
"target=$target opened=${opened.get()} eosed=${gotEose.get()} " + "target=$target opened=${opened.get()} eosed=${gotEose.get()} " +
"active=${server.activeSessionCount} elapsedMs=${opens.inWholeMilliseconds} " + "active=${server.activeSessionCount} elapsedMs=${opens.inWholeMilliseconds} " +
"heapMb=$rssMb", "heapMb=$heapMb",
) )
sockets.forEach { runCatching { it.cancel() } } sockets.forEach { runCatching { it.cancel() } }
check(gotEose.get() == target.toLong()) { check(gotEose.get() == target.toLong()) {
"expected $target EOSE but got ${gotEose.get()} — connection scaling regression" "expected $target EOSE but got ${gotEose.get()} — connection scaling regression"
} }
check(rssMb < 1024) { check(heapMb < 1024) {
"heap usage $rssMb MiB exceeded 1 GiB ceiling for $target idle connections" "JVM heap $heapMb MiB exceeded 1 GiB ceiling for $target idle connections"
} }
} }
} }
@@ -238,11 +244,11 @@ class LoadBenchmark {
val relayUrl = server.url.normalizeRelayUrl() val relayUrl = server.url.normalizeRelayUrl()
val received = AtomicLong() val received = AtomicLong()
val eosed = AtomicLong() val eosed = AtomicLong()
// Per-event fan-out latency, capped at the # of // Last-receive timestamp per event id. The N-th
// events we expect (durationSeconds * targetEps). // subscriber to deliver wins; combined with the
val fanoutLatenciesNs = // publish timestamp this gives us the full fan-out
java.util.concurrent.ConcurrentHashMap<String, AtomicLong>() // duration to the slowest subscriber.
val firstSeenNs = val lastReceiveNs =
java.util.concurrent.ConcurrentHashMap<String, AtomicLong>() java.util.concurrent.ConcurrentHashMap<String, AtomicLong>()
repeat(subs) { i -> repeat(subs) { i ->
@@ -256,12 +262,9 @@ class LoadBenchmark {
relay: NormalizedRelayUrl, relay: NormalizedRelayUrl,
forFilters: List<Filter>?, forFilters: List<Filter>?,
) { ) {
val now = System.nanoTime() lastReceiveNs
firstSeenNs .computeIfAbsent(event.id) { AtomicLong() }
.computeIfAbsent(event.id) { AtomicLong(now) } .set(System.nanoTime())
fanoutLatenciesNs
.computeIfAbsent(event.id) { AtomicLong(now) }
.set(now)
received.incrementAndGet() received.incrementAndGet()
} }
@@ -306,7 +309,7 @@ class LoadBenchmark {
} }
val perEventLastMs = val perEventLastMs =
fanoutLatenciesNs.entries lastReceiveNs.entries
.mapNotNull { (id, last) -> .mapNotNull { (id, last) ->
publishedAt[id]?.let { (last.get() - it) / 1_000_000.0 } publishedAt[id]?.let { (last.get() - it) / 1_000_000.0 }
}.sorted() }.sorted()