mirror of
https://github.com/vitorpamplona/amethyst.git
synced 2026-10-05 19:28:25 +00:00
Merge pull request #3926 from vitorpamplona/fix/dedup-cache-idle-trim
perf: bound the dedup decoder cache in size and in time
This commit is contained in:
Vitor Pamplona
@@ -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<WebSocket>()
|
||||
|
||||
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")
|
||||
}
|
||||
}
|
||||
+50
-1
@@ -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(
|
||||
|
||||
+46
-1
@@ -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<HexKey, Event>()
|
||||
@@ -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,
|
||||
|
||||
+17
@@ -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) }
|
||||
|
||||
+157
@@ -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()
|
||||
}
|
||||
}
|
||||
+205
@@ -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<Long, Any> {
|
||||
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<Event> {
|
||||
val stream =
|
||||
javaClass.classLoader?.getResourceAsStream(CORPUS)
|
||||
?: throw IllegalStateException("corpus $CORPUS missing from the test classpath")
|
||||
|
||||
val out = ArrayList<Event>(EVENTS)
|
||||
val seen = HashSet<String>(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))
|
||||
}
|
||||
}
|
||||
+170
@@ -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<Long, Any> {
|
||||
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<String> =
|
||||
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<String, String>(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<String, String>(1 shl 16)
|
||||
timeIt("app-level pool") { s, _ -> pool2.putIfAbsent(s, s) ?: s }
|
||||
|
||||
check(sink != null)
|
||||
}
|
||||
}
|
||||
+206
@@ -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<String> {
|
||||
val stream =
|
||||
javaClass.classLoader?.getResourceAsStream(CORPUS)
|
||||
?: throw IllegalStateException("corpus $CORPUS missing from the test classpath")
|
||||
|
||||
val out = ArrayList<String>(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<Long, Any> {
|
||||
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<String, Int>(frames.size)
|
||||
val distances = ArrayList<Int>(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))
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user