diff --git a/geode/src/test/kotlin/com/vitorpamplona/geode/OkHttpDispatcherWebSocketCapTest.kt b/geode/src/test/kotlin/com/vitorpamplona/geode/OkHttpDispatcherWebSocketCapTest.kt new file mode 100644 index 0000000000..98d798d495 --- /dev/null +++ b/geode/src/test/kotlin/com/vitorpamplona/geode/OkHttpDispatcherWebSocketCapTest.kt @@ -0,0 +1,143 @@ +/* + * 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.geode + +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.normalizeRelayUrl +import okhttp3.Dispatcher +import okhttp3.OkHttpClient +import okhttp3.Request +import okhttp3.Response +import okhttp3.WebSocket +import okhttp3.WebSocketListener +import java.util.concurrent.CountDownLatch +import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicInteger +import kotlin.test.AfterTest +import kotlin.test.BeforeTest +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertFalse + +/** + * Does OkHttp's `Dispatcher.maxRequests` cap *concurrent handshakes*, or does a live + * WebSocket hold its dispatcher slot for the socket's whole lifetime? + * + * This decides whether lowering `maxRequests` on the relay `OkHttpClient` + * (`OkHttpClientFactoryForRelays`, currently 1024) is a safe way to throttle the + * cold-start dial burst — or a change that would silently cap the app at N relays + * forever. The app connects to ~190 relays at once and spikes to ~640 threads, so + * this knob looks tempting; the difference between the two behaviours is the + * difference between a one-line fix and an outage. + * + * The answer is not something to take from memory: `RealWebSocket.connect()` goes + * through `newCall().enqueue()`, so the upgrade *is* dispatched, and whether the + * call is retired when the 101 response arrives is an implementation detail that + * has changed across OkHttp versions. + */ +class OkHttpDispatcherWebSocketCapTest { + private lateinit var relay: RelayEngine + private lateinit var server: KtorRelay + + companion object { + const val CAP = 4 + const val SOCKETS = 20 + } + + @BeforeTest + fun setup() { + relay = RelayEngine(url = "ws://127.0.0.1:7771/".normalizeRelayUrl()) + server = KtorRelay(relay, host = "127.0.0.1", port = 0).start() + } + + @AfterTest + fun teardown() { + server.stop() + relay.close() + } + + @Test + fun liveWebSocketsHoldTheirDispatcherSlotForever() { + val client = + OkHttpClient + .Builder() + .dispatcher( + Dispatcher().apply { + maxRequests = CAP + // all sockets share one host here, so this would bind too + maxRequestsPerHost = CAP + }, + ).build() + + val opened = AtomicInteger() + val failed = AtomicInteger() + val latch = CountDownLatch(SOCKETS) + val sockets = mutableListOf() + + val url = server.url.replace("ws://", "http://") + repeat(SOCKETS) { + sockets += + client.newWebSocket( + Request.Builder().url(url).build(), + object : WebSocketListener() { + override fun onOpen( + webSocket: WebSocket, + response: Response, + ) { + opened.incrementAndGet() + latch.countDown() + } + + override fun onFailure( + webSocket: WebSocket, + t: Throwable, + response: Response?, + ) { + failed.incrementAndGet() + latch.countDown() + } + }, + ) + } + + val settled = latch.await(20, TimeUnit.SECONDS) + // snapshot BEFORE cancelling: cancel() drives onFailure on every queued dial + val openedCount = opened.get() + val failedCount = failed.get() + println( + "\nmaxRequests=$CAP, dialed $SOCKETS websockets -> " + + "opened=$openedCount failed=$failedCount allSettled=$settled", + ) + + sockets.forEach { it.cancel() } + client.dispatcher.executorService.shutdown() + + // ANSWER: a live WebSocket holds its dispatcher slot for the socket's whole + // lifetime. Exactly CAP open; the remaining SOCKETS-CAP sit queued forever and + // never fail, so nothing surfaces the stall. + // + // Therefore `maxRequests` must NEVER be used to throttle the relay dial burst: + // setting it to N would cap the app at N relays permanently. Throttling has to + // happen above OkHttp, in RelayPool, where a settled dial can release its permit. + assertEquals(CAP, openedCount, "only maxRequests websockets ever open") + assertEquals(0, failedCount, "the queued dials never fail - they just hang") + assertFalse(settled, "the remaining dials never settle") + } +} diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/NostrClient.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/NostrClient.kt index bc281cfd3e..30da4b8992 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/NostrClient.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/NostrClient.kt @@ -91,7 +91,7 @@ class NostrClient( * another subscription or another relay reuses the already-parsed Event; * dispatch semantics are unchanged). */ - decoder: MessageDecoder = MessageDecoder.Default, + private val decoder: MessageDecoder = MessageDecoder.Default, ) : INostrClient, RelayConnectionListener, AutoCloseable { @@ -156,6 +156,11 @@ class NostrClient( override fun disconnect() { isActiveFlow.value = false relayPool.disconnect() + // The host is putting us down (typically the app backgrounding). Whatever the + // decoder cached has no dedup value left across that gap, but it would keep + // costing memory for the whole background stretch, so release it now rather + // than leaving a timer running to do it later. + decoder.clearCache() } override fun isActive() = isActiveFlow.value @@ -208,8 +213,52 @@ class NostrClient( } } + /** + * Ages out the decoder's cache on a clock. + * + * A caching decoder bounds what it holds by capacity, which caps the cost during a + * burst but never ends it: its generations only rotate inside an insert, so a + * client that drops to a trickle keeps everything from the initial burst forever. + * + * Deliberately NOT idle-triggered. Measured on device: relays keep pushing events + * down open subscriptions indefinitely, so the decoder is never idle in the sense + * of "saw no frames" — an idle check returned false on every tick across 240s of + * a flat heap. Aging unconditionally is what actually releases the memory. + * + * Ids survive one tick and die on the next, so [DECODER_AGE_OUT_MS] bounds their + * lifetime at ~2x that. Suspends on [isActiveFlow] like [keepAliveJob], so no timer + * fires while the client is down — [disconnect] clears outright on the way there. + * + * 30s is measured, not guessed. Sweeping the interval on device and comparing hit + * rate at a matched ~19k frames (runs pull different volumes, so raw endpoints are + * not comparable): + * + * ``` + * tick hit rate parses ids still cached at rest + * 10s 44.9% 10,663 ~2 + * 30s 58.5% 7,569 ~2 + * 60s 60.4% 7,520 ~3 + * 120s 60.6% 7,555 7,438 (one tick in 3 min -> never releases) + * ``` + * + * Hit rate saturates by 30s, so a shorter tick only buys re-parses (10s costs 41% + * more), and a longer one only holds memory. The knee is where the tick stops being + * the binding constraint and [CachingEventDecoder.capacity] takes over: at the + * observed ~400 frames/s that crossover is 8192/400 ~= 20s, so 30s sits just past + * it. Re-measure if capacity or typical frame rates change. + */ + private val decoderTrimJob = + scope.launch { + while (true) { + isActiveFlow.first { it } + delay(DECODER_AGE_OUT_MS) + decoder.ageOutCache() + } + } + companion object { private const val KEEP_ALIVE_INTERVAL_MS = 60_000L + private const val DECODER_AGE_OUT_MS = 30_000L } override fun reconnect( diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/commands/toClient/CachingEventDecoder.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/commands/toClient/CachingEventDecoder.kt index b3517a3f5e..f2ea54e695 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/commands/toClient/CachingEventDecoder.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/commands/toClient/CachingEventDecoder.kt @@ -71,7 +71,22 @@ import kotlin.concurrent.atomics.incrementAndFetch */ @OptIn(ExperimentalAtomicApi::class) class CachingEventDecoder( - private val capacity: Int = 2048, + /** + * Ids kept per generation, so up to `2 * capacity` are addressable. + * + * Sized from the measured *reuse distance* — how many frames separate two + * deliveries of the same event — because a duplicate is only caught when its + * reuse distance still fits in the cache. Against the recorded multi-relay + * startup capture (150k frames, 82% duplicates, median reuse distance 1,430), + * the old 2048 caught just 58.3% of duplicates; 8192 catches 80.5% and halves + * decode time (354ms -> 169ms). Beyond 8192 the curve flattens (32768 buys + * 96.5% for 4x the entries), so this trades the last 16% for memory the + * client would otherwise hold on top of `LocalCache`. + * + * See `DedupCapacityBenchmark`. Re-measure it before changing this: the right + * value follows the duplication of real traffic, not intuition. + */ + private val capacity: Int = 8192, private val fullParser: MessageDecoder = MessageDecoder.Default, ) : MessageDecoder { @Volatile private var live = ConcurrentHashCache() @@ -108,6 +123,36 @@ class CachingEventDecoder( return msg } + /** + * Retires the live generation, so entries age out on the caller's clock instead of + * only when the cache fills. + * + * [capacity] bounds what the cache costs *during* a burst; this bounds how long it + * costs anything *after* one. Without it the maps only rotate inside an insert, so + * a client receiving a slow trickle keeps everything it saw during the initial + * burst forever — memory whose dedup value is long gone. + * + * An id survives one call and dies on the next, so ticking every 30s keeps nothing + * older than ~60s. Two consecutive calls with no traffic in between leave the cache + * empty, which is what releases memory when the client really does go quiet. + * + * Racy by the same design as capacity rotation (see class kdoc): retiring a + * generation concurrently with an insert can only lose an id, and a lost id costs a + * re-parse, never a wrong message. + */ + override fun ageOutCache() { + previous = live + live = ConcurrentHashCache() + } + + override fun clearCache() { + previous = ConcurrentHashCache() + live = ConcurrentHashCache() + } + + /** Ids currently addressable across both generations. Observability + tests. */ + val cachedCount: Int get() = live.size() + previous.size() + private class ScannedEvent( val subId: String, val eventId: HexKey, diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/commands/toClient/MessageDecoder.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/commands/toClient/MessageDecoder.kt index 4754ccf62f..6362040b18 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/commands/toClient/MessageDecoder.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/commands/toClient/MessageDecoder.kt @@ -33,6 +33,23 @@ import com.vitorpamplona.quartz.nip01Core.core.OptimizedJsonMapper fun interface MessageDecoder { fun decode(text: String): Message + /** + * Ages out whatever the decoder is holding. Call periodically: an entry must + * survive at most two calls, so a 30s tick keeps nothing older than ~60s. + * + * A caching decoder's value decays with time — a duplicate that has not arrived + * within seconds is not going to — while its memory does not decay at all. Aging + * on a clock rather than on idleness is deliberate: relays keep pushing a trickle + * of events down open subscriptions forever, so a decoder is never idle in the + * sense of "saw no frames", and an idle-triggered release would never fire. + * + * Never affects correctness: dropping a cached id can only cost a re-parse. + */ + fun ageOutCache() = Unit + + /** Releases everything cached, e.g. when the host is shutting the client down. */ + fun clearCache() = Unit + companion object { /** Plain full-JSON parse of every frame. */ val Default: MessageDecoder = MessageDecoder { OptimizedJsonMapper.fromJsonToMessage(it) } diff --git a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/commands/toClient/CachingEventDecoderTrimTest.kt b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/commands/toClient/CachingEventDecoderTrimTest.kt new file mode 100644 index 0000000000..28d3ff7ea0 --- /dev/null +++ b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/commands/toClient/CachingEventDecoderTrimTest.kt @@ -0,0 +1,157 @@ +/* + * 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.commands.toClient + +import com.vitorpamplona.quartz.nip01Core.core.Event +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertTrue + +/** + * [CachingEventDecoder.ageOutCache] bounds how long the cache costs memory, which + * [CachingEventDecoder.capacity] does not: the generations otherwise only rotate + * inside an insert, so a client that drops to a trickle keeps everything from its + * initial burst forever. + * + * Aging is on a clock, not on idleness, and that distinction is load-bearing — + * measured on device, relays keep pushing events down open subscriptions + * indefinitely, so an idle-triggered release returned false on every tick across + * 240s of a flat heap. The rules pinned here are the lifetime guarantee ("an entry + * survives one call and dies on the next"), that traffic between calls is kept, and + * that aging only ever costs a re-parse — never a different message. + * + * In `commonTest`: `commonMain` logic, no platform surface, no clock. + */ +class CachingEventDecoderTrimTest { + private fun hexId(seed: Int) = seed.toString(16).padStart(64, '0') + + private fun event(seed: Int) = + Event( + id = hexId(seed), + pubKey = hexId(seed + 500_000), + createdAt = seed.toLong(), + kind = 1, + tags = arrayOf(arrayOf("t", "trim")), + content = "trim test $seed", + sig = "f".repeat(128), + ) + + private fun frame( + seed: Int, + subId: String = "s", + ) = """["EVENT","$subId",${event(seed).toJson()}]""" + + @Test + fun entriesSurviveExactlyOneAgeOut() { + val decoder = CachingEventDecoder(capacity = 1_000) + decoder.decode(frame(1)) + assertEquals(1, decoder.parsedCount.toInt()) + + // first tick: retired into `previous`, still addressable + decoder.ageOutCache() + decoder.decode(frame(1)) + assertEquals(1, decoder.parsedCount.toInt(), "one age-out must not drop the id") + assertEquals(1, decoder.reusedCount.toInt()) + } + + @Test + fun twoAgeOutsWithNoTrafficEmptyTheCache() { + val decoder = CachingEventDecoder(capacity = 1_000) + (1..50).forEach { decoder.decode(frame(it)) } + assertEquals(50, decoder.cachedCount) + + decoder.ageOutCache() + decoder.ageOutCache() + + assertEquals(0, decoder.cachedCount, "nothing survives two ticks without traffic") + decoder.decode(frame(1)) + assertEquals(51, decoder.parsedCount.toInt(), "the id is gone, so it re-parses") + assertEquals(0, decoder.reusedCount.toInt()) + } + + @Test + fun trafficBetweenTicksIsKept() { + // the trickle case: events keep arriving, so the cache must keep serving them + val decoder = CachingEventDecoder(capacity = 1_000) + decoder.decode(frame(1)) + decoder.ageOutCache() + decoder.decode(frame(2)) + decoder.ageOutCache() + + // 1 was inserted two ticks ago -> gone; 2 was inserted one tick ago -> kept + decoder.decode(frame(2)) + assertEquals(2, decoder.parsedCount.toInt()) + assertEquals(1, decoder.reusedCount.toInt(), "the recent event is still cached") + + decoder.decode(frame(1)) + assertEquals(3, decoder.parsedCount.toInt(), "the older event aged out") + } + + @Test + fun agingABusyCacheKeepsItSmallInsteadOfUnbounded() { + // what the device measurement is about: a burst, then a trickle. Without aging + // the burst's ids stay resident forever. + val decoder = CachingEventDecoder(capacity = 100_000) + (1..5_000).forEach { decoder.decode(frame(it)) } + assertEquals(5_000, decoder.cachedCount) + + decoder.ageOutCache() + (5_001..5_010).forEach { decoder.decode(frame(it)) } // the trickle + decoder.ageOutCache() + + assertEquals(10, decoder.cachedCount, "only the trickle survives, not the burst") + } + + @Test + fun clearCacheReleasesEverythingAtOnce() { + // what NostrClient.disconnect() uses when the host backgrounds the app + val decoder = CachingEventDecoder(capacity = 1_000) + (1..10).forEach { decoder.decode(frame(it)) } + assertTrue(decoder.cachedCount > 0) + + decoder.clearCache() + assertEquals(0, decoder.cachedCount) + } + + @Test + fun agingNeverChangesWhatADecodeReturns() { + val decoder = CachingEventDecoder(capacity = 1_000) + val before = decoder.decode(frame(7, subId = "a")) as EventMessage + decoder.ageOutCache() + decoder.ageOutCache() + val after = decoder.decode(frame(7, subId = "a")) as EventMessage + + assertEquals(before.subId, after.subId) + assertEquals(before.event.id, after.event.id) + assertEquals(before.event.pubKey, after.event.pubKey) + assertEquals(before.event.content, after.event.content) + assertEquals(before.event.createdAt, after.event.createdAt) + // only the parse count differs: aging cost one re-parse, nothing else + assertEquals(2, decoder.parsedCount.toInt()) + } + + @Test + fun theDefaultDecoderIgnoresBoth() { + // stateless: nothing to age or release, and neither call may throw + MessageDecoder.Default.ageOutCache() + MessageDecoder.Default.clearCache() + } +} diff --git a/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/prodbench/IngestAllocationBenchmark.kt b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/prodbench/IngestAllocationBenchmark.kt new file mode 100644 index 0000000000..0369af8a30 --- /dev/null +++ b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/prodbench/IngestAllocationBenchmark.kt @@ -0,0 +1,205 @@ +/* + * 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.prodbench + +import com.fasterxml.jackson.core.JsonToken +import com.sun.management.ThreadMXBean +import com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.crypto.verify +import com.vitorpamplona.quartz.nip01Core.jackson.JacksonMapper +import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.Message +import com.vitorpamplona.quartz.nip10Notes.content.findHashtags +import com.vitorpamplona.quartz.nip10Notes.content.findIndexTagsWithPeople +import com.vitorpamplona.quartz.nip19Bech32.Nip19Parser +import java.lang.management.ManagementFactory +import java.util.zip.GZIPInputStream +import kotlin.test.Test +import kotlin.test.assertTrue + +/** + * Allocation-RATE profile of the per-event ingest path. + * + * Every previous memory measurement in this codebase looked at *retained* heap (heap + * dumps, histograms, the intern tradeoff). That answers "what is alive", which is the + * wrong question for the observed problem: on an SM-T220, ingest saturates all 8 cores + * and the driver is GC read barriers charged to the mutator threads, not any single + * consume stage. Read-barrier cost tracks the *allocation rate* — bytes created and + * discarded per second — which retained-heap analysis is blind to. + * + * `com.sun.management.ThreadMXBean.getThreadAllocatedBytes` reports exactly that, per + * thread, with no sampling error. It counts TLAB bumps, so it sees garbage that never + * survives a single GC — precisely the allocations a heap dump cannot show. + * + * The headline number to look for is **bytes allocated per byte of wire JSON**. An event + * arrives as N bytes of text; if turning it into an `Event` costs 10N, the ingest path + * is a garbage pump and bounding concurrency will not help, because every added thread + * multiplies the same waste. + * + * Runs on the JVM, so the absolute numbers are HotSpot's. Allocation *counts* come from + * the same bytecode ART runs and the object layouts are close, so the per-stage ranking + * and the ratios transfer; treat the absolute MB/s as indicative and confirm any fix + * on device. + */ +class IngestAllocationBenchmark { + companion object { + /** The real multi-relay capture of an account cold start, checked into quartz. */ + const val CORPUS = "nostr_vitor_startup_data.json.gz" + const val EVENTS = 20_000 + + private val threads = ManagementFactory.getThreadMXBean() as ThreadMXBean + + /** Anything the measured loops produce, parked here so JIT cannot elide them. */ + var keep: Any? = null + + /** + * Bytes allocated by running [block] over [reps] iterations, minus the loop's own + * overhead. + * + * Warmed hard on purpose: escape analysis only kicks in after C2 compiles the + * method, and a cold measurement reports allocations the JIT would have scalarised + * away — i.e. it over-reports exactly the transient garbage this is looking for. + */ + fun allocBytes( + reps: Int, + warmup: Int = reps, + block: (Int) -> Any?, + ): Long { + var sink: Any? = null + for (i in 0 until warmup) sink = block(i % reps) + keep = sink + + val id = Thread.currentThread().threadId() + val before = threads.getThreadAllocatedBytes(id) + for (i in 0 until reps) sink = block(i) + val after = threads.getThreadAllocatedBytes(id) + keep = sink + + // the empty loop itself allocates nothing measurable, but the accessor does + val idle = threads.getThreadAllocatedBytes(id) + val overhead = idle - after + return (after - before) - overhead + } + + fun used(): Long { + val rt = Runtime.getRuntime() + return rt.totalMemory() - rt.freeMemory() + } + + fun retained(build: () -> Any): Pair { + repeat(3) { + System.gc() + Thread.sleep(60) + } + val before = used() + val held = build() + repeat(3) { + System.gc() + Thread.sleep(60) + } + return (used() - before) to held + } + } + + /** + * The real capture of Vitor's account startup — the exact traffic this profile is + * about — streamed rather than slurped, because the file is 247 MB decompressed and + * only the first [EVENTS] are needed. + * + * Real signatures, so `verify()` does its real work, and a real kind/size/tag mix + * rather than a guess at one. + */ + private fun corpus(): List { + val stream = + javaClass.classLoader?.getResourceAsStream(CORPUS) + ?: throw IllegalStateException("corpus $CORPUS missing from the test classpath") + + val out = ArrayList(EVENTS) + val seen = HashSet(EVENTS * 2) + GZIPInputStream(stream).use { gz -> + val parser = JacksonMapper.mapper.factory.createParser(gz) + check(parser.nextToken() == JsonToken.START_ARRAY) { "$CORPUS is not a JSON array" } + while (out.size < EVENTS && parser.nextToken() == JsonToken.START_OBJECT) { + val event: Event = JacksonMapper.mapper.readValue(parser, JacksonMapper.eventTypeInstance) + // the capture spans many relays, so the same event arrives several times + if (seen.add(event.id)) out.add(event) + } + } + return out + } + + @Test + fun allocationPerIngestStage() { + val events = corpus() + val eventJson = events.map { it.toJson() } + val frames = eventJson.mapIndexed { i, j -> "[\"EVENT\",\"sub$i\",$j]" } + val frameBytes = frames.map { it.toByteArray(Charsets.UTF_8) } + + val count = events.size + val wireBytes = frames.sumOf { it.length.toLong() } + val n = count.toLong() + + assertTrue(count > EVENTS / 2, "corpus yielded only $count events") + assertTrue(eventJson.all { it.length > 50 }, "corpus produced empty events") + + println("\n=== corpus: $count events, ${wireBytes / 1024} KB of wire JSON, mean ${wireBytes / n} B/event ===") + + // ---- what surviving the parse actually costs to keep ---- + val (retainedEvents, held) = retained { frames.map { Message.fromJson(it) } } + keep = held + + // ---- per stage ---- + val decode = allocBytes(count) { i -> String(frameBytes[i], Charsets.UTF_8) } + val msgParse = allocBytes(count) { i -> Message.fromJson(frames[i]) } + val evtParse = allocBytes(count) { i -> JacksonMapper.fromJson(eventJson[i]) } + val verify = allocBytes(count) { i -> events[i].verify() } + val scans = + allocBytes(count) { i -> + val e = events[i] + findHashtags(e.content) + findIndexTagsWithPeople(e.content, e.tags) + Nip19Parser.parseAll(e.content) + } + + fun row( + label: String, + bytes: Long, + ) = println( + " %-22s %8.1f MB total %8d B/event %5.2fx wire".format( + label, + bytes / 1048576.0, + bytes / n, + bytes.toDouble() / wireBytes, + ), + ) + + println("\nALLOCATED (transient + surviving)") + row("utf8 decode", decode) + row("frame -> Message", msgParse) + row("event json -> Event", evtParse) + row("verify()", verify) + row("content scans", scans) + row("TOTAL decode+parse+verify", decode + msgParse + verify) + + println("\nRETAINED after parse") + println(" parsed events %8.1f MB total %8d B/event %5.2fx wire".format(retainedEvents / 1048576.0, retainedEvents / n, retainedEvents.toDouble() / wireBytes)) + println(" garbage per kept byte %8.1fx".format((decode + msgParse + verify - retainedEvents).toDouble() / retainedEvents)) + } +} diff --git a/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/prodbench/InternTradeoffBenchmark.kt b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/prodbench/InternTradeoffBenchmark.kt new file mode 100644 index 0000000000..9095a17bcf --- /dev/null +++ b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/prodbench/InternTradeoffBenchmark.kt @@ -0,0 +1,170 @@ +/* + * 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.prodbench + +import java.util.concurrent.ConcurrentHashMap +import kotlin.test.Test + +/** + * `String.intern()` is applied to every event `id`, `pubKey` (EventDeserializer) and + * every tag value (TagArrayDeserializer). An on-device profile of a release build + * during ingest put `art::InternTable::InternWeak` at **3.5% of the ingest workers' + * CPU** — the second largest symbol after memcpy. + * + * Interning is a MEMORY optimisation, not a speed one: Nostr repeats strings heavily, + * so collapsing duplicates to one instance is why it is there. Removing it could + * therefore cost more memory than the CPU it saves. This measures both axes, split by + * how duplicated each field actually is, because the answer differs per field: + * + * - tag names (`p`, `e`, `a`) — a handful of values repeated on every event + * - relay URLs — tens of values repeated across everything + * - pubkeys — measured on device: 1,880 distinct for 11,338 notes + * - event ids — one per event, but every `e` tag that references + * that event repeats it, so they dedup too + * + * Cardinalities come from the device probe cited above, scaled to a corpus big enough + * for the heap delta to clear GC noise. + * + * Results (JVM, 60k events / 840k strings): + * + * ``` + * retained parse CPU + * no interning 69.1 MB 146 ms + * intern everything 17.8 MB 354 ms + * app-level pool 23.3 MB 192 ms + * intern all but ids 35.8 MB 257 ms + * ``` + * + * So interning is a ~4x memory win and worth its CPU. Skipping the hex fields is not a + * shortcut — it gives up half the memory for half the CPU, because those hex strings + * repeat too. An app-level pool is the only variant that beats interning on CPU, but it + * holds *strong* references where ART's intern table is weak, so it would never release + * a string whose event was pruned; the 5.5 MB it loses to interning is its own node + * headers. Not worth trading a self-clearing table for an unbounded one to reclaim ~2.7% + * of ingest CPU on a device that is already memory-bound. + * + * Measures the interning primitive on realistic string distributions — not the full + * Jackson parse path — so it isolates the tradeoff without needing a second + * deserializer wired in. Note this is HotSpot's `intern()`; ART's `InternWeak` is a + * different implementation, so treat the ordering as transferable and the magnitudes as + * indicative. + */ +class InternTradeoffBenchmark { + companion object { + const val EVENTS = 60_000 + const val AUTHORS = 10_000 // ~6 events per author, as measured on device + const val RELAYS = 60 + val TAG_NAMES = listOf("p", "e", "a", "t", "r", "alt") + + fun hex(seed: Int): String = "%064x".format(seed) + + fun used(): Long { + val rt = Runtime.getRuntime() + return rt.totalMemory() - rt.freeMemory() + } + + /** Retained bytes of whatever [build] returns, with GC settled either side. */ + fun retained(build: () -> Any): Pair { + repeat(3) { + System.gc() + Thread.sleep(60) + } + val before = used() + val held = build() + repeat(3) { + System.gc() + Thread.sleep(60) + } + val after = used() + return (after - before) to held + } + } + + /** The strings one event contributes, as fresh instances (as if just parsed off the wire). */ + private fun eventStrings(i: Int): List = + buildList { + add(String(hex(i).toCharArray())) // id — unique per event + add(String(hex(i % AUTHORS).toCharArray())) // pubKey — repeats + repeat(4) { t -> + add(String(TAG_NAMES[(i + t) % TAG_NAMES.size].toCharArray())) // tag name — tiny set + add(String("wss://relay${(i + t) % RELAYS}.example.com/".toCharArray())) // relay url + add(String(hex((i + t) * 31).toCharArray())) // tag value: a referenced id + } + } + + @Test + fun internCostAndBenefit() { + var sink: Any? = null + + println("\n=== corpus: $EVENTS events, $AUTHORS authors, $RELAYS relays ===") + + // ---- memory ---- + val (memNone, holdA) = retained { (0 until EVENTS).flatMap { eventStrings(it) } } + sink = holdA + val (memAll, holdB) = retained { (0 until EVENTS).flatMap { eventStrings(it).map { s -> s.intern() } } } + sink = holdB + // intern everything EXCEPT the two unique-hex classes (event id, tag id refs) + val (memSel, holdC) = + retained { + (0 until EVENTS).flatMap { i -> + eventStrings(i).mapIndexed { idx, s -> + // index 0 is the id; every 3rd tag slot is a referenced id + if (idx == 0 || (idx > 1 && (idx - 2) % 3 == 2)) s else s.intern() + } + } + } + sink = holdC + + println("\nRETAINED HEAP") + println(" no interning %6.1f MB".format(memNone / 1048576.0)) + println(" intern everything %6.1f MB (%+.1f MB vs none)".format(memAll / 1048576.0, (memAll - memNone) / 1048576.0)) + println(" intern all but ids %6.1f MB (%+.1f MB vs none)".format(memSel / 1048576.0, (memSel - memNone) / 1048576.0)) + + // ---- cpu ---- + fun timeIt( + label: String, + op: (String, Int) -> String, + ) { + repeat(2) { i -> (0 until 5_000).forEach { eventStrings(it).forEachIndexed { idx, s -> op(s, idx) } } } + val t0 = System.nanoTime() + var n = 0 + (0 until EVENTS).forEach { i -> eventStrings(i).forEachIndexed { idx, s -> if (op(s, idx) !== s) n++ } } + val ms = (System.nanoTime() - t0) / 1_000_000 + println(" %-20s %5d ms (%d strings)".format(label, ms, EVENTS * 14)) + } + + // An app-level dedup table gives the same collapsing without going through + // ART's global InternTable, which is what showed up in the device profile. + val pool = ConcurrentHashMap(1 shl 16) + val (memPool, holdD) = retained { (0 until EVENTS).flatMap { eventStrings(it).map { s -> pool.putIfAbsent(s, s) ?: s } } } + sink = holdD + println(" app-level pool %6.1f MB (%+.1f MB vs none, pool holds %d)".format(memPool / 1048576.0, (memPool - memNone) / 1048576.0, pool.size)) + + println("\nPARSE-SIDE CPU (build + optionally intern)") + timeIt("no interning") { s, _ -> s } + timeIt("intern everything") { s, _ -> s.intern() } + timeIt("intern all but ids") { s, idx -> if (idx == 0 || (idx > 1 && (idx - 2) % 3 == 2)) s else s.intern() } + val pool2 = ConcurrentHashMap(1 shl 16) + timeIt("app-level pool") { s, _ -> pool2.putIfAbsent(s, s) ?: s } + + check(sink != null) + } +} diff --git a/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/DedupCapacityBenchmark.kt b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/DedupCapacityBenchmark.kt new file mode 100644 index 0000000000..6f8b43017c --- /dev/null +++ b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/DedupCapacityBenchmark.kt @@ -0,0 +1,206 @@ +/* + * 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.quartz.nip01Core.relay.commands.toClient.CachingEventDecoder +import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.MessageDecoder +import java.util.zip.GZIPInputStream +import kotlin.test.Test +import kotlin.test.assertTrue + +/** + * Is the production [CachingEventDecoder] capacity big enough for a cold start? + * + * The app connects to every relay it knows and pulls everything at once, by design — + * that redundancy is the point, and throttling it would defeat the product. What is + * NOT wanted is paying a full JSON parse for the same event once per relay that + * serves it. [CachingEventDecoder] exists precisely to avoid that, and it is wired in + * (`AppModules.kt`, `NostrClient(websocketBuilder, applicationIOScope, + * CachingEventDecoder())`) — at its **default capacity of 2048**. + * + * `DedupDecodeBenchmark` proves the mechanism, but with `capacity = UNIQUE * 2` + * (40,000) and a frame order that spaces each duplicate 20,000 frames apart. Both + * choices flatter the cache. This measures the setting that actually ships, against + * the real multi-relay capture (248,241 frames / 31,161 unique -- ~8x duplication) in + * the order it was recorded. + * + * The decisive statistic is the **reuse distance**: how many frames pass between two + * deliveries of the same event. A capacity only catches duplicates whose reuse + * distance falls inside it, so the distance histogram gives the hit rate for every + * candidate capacity at once, without re-running the decoder for each. + */ +class DedupCapacityBenchmark { + companion object { + const val CORPUS = "nostr_vitor_startup_data.json.gz" + const val FRAMES = 150_000 + val CAPACITIES = listOf(2_048, 8_192, 32_768, 131_072) + } + + /** Raw event JSON in recorded order, wrapped as EVENT frames. */ + private fun frames(): List { + val stream = + javaClass.classLoader?.getResourceAsStream(CORPUS) + ?: throw IllegalStateException("corpus $CORPUS missing from the test classpath") + + val out = ArrayList(FRAMES) + GZIPInputStream(stream).bufferedReader().use { reader -> + // one giant single-line JSON array; walk it and slice out each top-level object + val buf = CharArray(1 shl 16) + val sb = StringBuilder() + var depth = 0 + var inStr = false + var esc = false + var n = reader.read(buf) + while (n > 0 && out.size < FRAMES) { + for (i in 0 until n) { + val c = buf[i] + if (depth > 0) sb.append(c) + when { + esc -> esc = false + c == '\\' && inStr -> esc = true + c == '"' -> inStr = !inStr + inStr -> {} + c == '{' -> { + if (depth == 0) sb.append(c) + depth++ + } + c == '}' -> { + depth-- + if (depth == 0) { + out.add("[\"EVENT\",\"s\",$sb]") + sb.setLength(0) + if (out.size >= FRAMES) break + } + } + } + } + if (out.size >= FRAMES) break + n = reader.read(buf) + } + } + return out + } + + private fun idOf(frame: String): String? { + val k = frame.indexOf("\"id\":\"") + if (k < 0 || k + 6 + 64 > frame.length) return null + return frame.substring(k + 6, k + 6 + 64) + } + + private fun used(): Long { + val rt = Runtime.getRuntime() + return rt.totalMemory() - rt.freeMemory() + } + + /** Retained bytes of whatever [build] returns, with GC settled either side. */ + private fun retained(build: () -> Any): Pair { + repeat(3) { + System.gc() + Thread.sleep(60) + } + val before = used() + val held = build() + repeat(3) { + System.gc() + Thread.sleep(60) + } + return (used() - before) to held + } + + /** + * What the cache actually costs to hold. + * + * This is the WORST case: the decoder is the only holder, so every cached Event + * counts. In the app most of them are also in `LocalCache` (the decoder hands out + * the very same instance), and for those the true incremental cost is just the map + * node. The gap between the two is the memory the decoder pins for events + * `LocalCache` chose to drop. + */ + @Test + fun retainedMemoryPerCapacity() { + val frames = frames() + var sink: Any? = null + println("\n=== retained heap held by the decoder (worst case: sole holder) ===") + (CAPACITIES + listOf(2_048)).distinct().sorted().forEach { cap -> + val (bytes, held) = + retained { + val d = CachingEventDecoder(capacity = cap) + frames.forEach { d.decode(it) } + d + } + sink = held + val entries = minOf(cap * 2, 25_571) + println( + " capacity %7d %6.1f MB retained (<= %d cached events, %.0f B/entry)".format( + cap, + bytes / 1048576.0, + entries, + bytes.toDouble() / entries, + ), + ) + } + checkNotNull(sink) + } + + @Test + fun productionCapacityAgainstRealArrivalOrder() { + val frames = frames() + assertTrue(frames.size > FRAMES / 2, "corpus yielded only ${frames.size} frames") + + // ---- reuse-distance histogram: hit rate for ANY capacity, in one pass ---- + val lastSeen = HashMap(frames.size) + val distances = ArrayList(frames.size) + var unique = 0 + frames.forEachIndexed { i, f -> + val id = idOf(f) ?: return@forEachIndexed + val prev = lastSeen.put(id, i) + if (prev == null) unique++ else distances.add(i - prev) + } + val dupes = distances.size + println("\n=== real capture: ${frames.size} frames, $unique unique, $dupes duplicates (${dupes * 100 / frames.size}%) ===") + + println("\nREUSE DISTANCE -> share of duplicates a cache of that size can catch") + CAPACITIES.forEach { cap -> + val caught = distances.count { it <= cap } + val label = if (cap == 2_048) " <- PRODUCTION" else "" + println(" capacity %7d catches %5.1f%% of duplicates%s".format(cap, caught * 100.0 / dupes, label)) + } + val med = distances.sorted()[dupes / 2] + println(" median reuse distance: $med frames") + + // ---- confirm with the real decoder at the shipping capacity ---- + println("\nACTUAL DECODER (parse time over ${frames.size} frames)") + (listOf(0) + CAPACITIES).forEach { cap -> + val decoder = if (cap == 0) MessageDecoder.Default else CachingEventDecoder(capacity = cap) + repeat(2_000) { decoder.decode(frames[it]) } // warm + val t0 = System.nanoTime() + frames.forEach { decoder.decode(it) } + val ms = (System.nanoTime() - t0) / 1_000_000 + val reuse = + if (decoder is CachingEventDecoder) { + " reused=${decoder.reusedCount} parsed=${decoder.parsedCount}" + } else { + "" + } + println(" capacity %7s %6d ms%s".format(if (cap == 0) "none" else "$cap", ms, reuse)) + } + } +}