mirror of
https://github.com/vitorpamplona/amethyst.git
synced 2026-08-12 01:07:46 +00:00
feat: add bulk-download benchmark with negentropy vs paging test case
Measures how to download millions of events from a single relay as fast as possible, download+parse only: local geode ceilings (giant REQ vs until-cursor paging vs created_at-sharded connections, quartz stack vs raw frames), offline per-frame strategies (full parse, parallel parse, id-scan for raw archiving), and a production case syncing kind 30382 from nip85.nosfabrica.com via NIP-77 negentropy against plain paging. Headline results in the plan doc: paging beats giant REQs 26x (which also dropped frames), created_at sharding stacks ~2x on top, parse is 2-5% of the budget, and negentropy is 2.4x slower than paging for a cold download (its win is incremental re-sync). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_018saXqYfAa3RvSJoDXK591R
This commit is contained in:
Claude
@@ -169,6 +169,83 @@ Findings:
|
||||
`onNewEvent`/stats counters, so paging/EOSE bookkeeping would need the
|
||||
cheap counters kept ahead of the skip.
|
||||
|
||||
## Bulk download: N-million events from one relay (download + parse only)
|
||||
|
||||
**Harness:** `quartz/src/jvmTest/.../relay/prodbench/BulkDownloadBenchmark.kt`
|
||||
(gated: `./gradlew :quartz:jvmTest --tests "*.BulkDownloadBenchmark" -PprodRelayBench=1`)
|
||||
|
||||
### Local ceilings (geode on localhost TCP, 100k seeded events, 4 cores)
|
||||
|
||||
| strategy | events/s | wall for 100k |
|
||||
|---|---|---|
|
||||
| quartz, 1 conn, one giant REQ | 476 | 210s (and 2 events silently missing) |
|
||||
| quartz, 4 conns time-sharded, giant REQs | 2,001 | 50s |
|
||||
| quartz, 1 conn, paged 1000/page | 12,480 | 8.0s |
|
||||
| **quartz, 4 conns time-sharded + paged** | **24,210** | **4.1s** |
|
||||
| raw socket (no parse), one giant REQ | ~676 | timed out at 120s (81k/100k) |
|
||||
|
||||
- **Giant single REQs are a trap.** The raw (no-parse) variant proves it's
|
||||
server-side: geode streams a 100k-event response at ~700 events/s
|
||||
(per-frame cost in the session pump / query streaming path — needs its own
|
||||
investigation, `WebSocketSessionPump`), and the quartz run came back 2
|
||||
events short. Public relays are worse: they clamp limits and may drop
|
||||
frames or the connection under output backpressure. Paged cursors through
|
||||
the very same server ran 26× faster.
|
||||
- **Sharding the `created_at` range across connections stacks with paging:**
|
||||
4 sharded + paged connections ≈ 2× one paged connection locally (24k/s),
|
||||
and each connection gets its own receiver coroutine, so parse parallelizes
|
||||
for free.
|
||||
|
||||
### Per-frame strategies, offline (50k frames, ~580B each)
|
||||
|
||||
| strategy | events/s | µs/frame |
|
||||
|---|---|---|
|
||||
| full `fromJsonToMessage`, 1 thread | 276k | 3.6 |
|
||||
| full parse, 4 threads | 657k | 1.5 (aggregate) |
|
||||
| id-scan only (archive raw, parse lazily) | 2.96M | 0.34 |
|
||||
|
||||
Parse of small events is ~3.6µs; the earlier production capture (mixed sizes,
|
||||
damus 8.7KB events) measured 32µs. Either way **parse is not the bottleneck
|
||||
for bulk download**: one core parses 10M small events in ~36s, and a 4-core
|
||||
fan-out does it in ~15s. The wire and the relay's page cadence dominate.
|
||||
|
||||
### Production test case: kinds=[30382] on nip85.nosfabrica.com (cap 20k)
|
||||
|
||||
| strategy | wall | events/s | notes |
|
||||
|---|---|---|---|
|
||||
| paged download (until-cursor) | 5.4s | 3,711 | 41 pages ≈ 500/page (relay clamp) |
|
||||
| NIP-77 negentropy sync | 12.7s | 1,575 | full set = 31,241 ids, 9 reconcile windows |
|
||||
|
||||
Negentropy correctly enumerated the whole 31,241-id set (splitting 9 windows
|
||||
around the relay's `max_sync_events` cap) and streamed id-batch REQs — but
|
||||
for a **cold** download it was 2.4× slower than plain paging: the reconcile
|
||||
rounds and by-id lookups cost more than sequential pages when you need
|
||||
*everything anyway*. Negentropy's win is **incremental re-sync**: once the
|
||||
10M events are local, the next sync transfers only fingerprints + the diff
|
||||
instead of re-paging the world. (kind 30382 events carry empty `content` —
|
||||
all data in tags — so the MB/s column reads 0.)
|
||||
|
||||
### Answer for "10M events from one relay, fastest"
|
||||
|
||||
At nosfabrica's measured page cadence, one connection ≈ 3.7k events/s → 10M
|
||||
in ~45 min. To improve, in order:
|
||||
|
||||
1. **Page with until-cursors — never one giant REQ** (silent drops, server
|
||||
slow paths, relay clamps).
|
||||
2. **Shard the `created_at` range across K connections** to the same relay
|
||||
(each shard pages independently; no cursor dependency between shards).
|
||||
Local: 2× at K=4; WAN, where RTT dominates page turnaround, closer to
|
||||
linear until the relay rate-limits per-IP.
|
||||
3. **Don't optimize parse first** — it's 2–5% of the budget. If the goal is
|
||||
archival, id-scan + store raw frames (0.34µs/frame) and parse lazily
|
||||
in parallel later.
|
||||
4. **Use negentropy for the second sync onward**, not the first.
|
||||
5. **Memory:** the per-connection channel is UNLIMITED — at 10M events a
|
||||
sink slower than the socket accumulates heap without bound. A bulk
|
||||
downloader should bound the channel (blocking the OkHttp reader thread is
|
||||
fine — that's TCP backpressure doing its job) and stream events to disk,
|
||||
never hold the set.
|
||||
|
||||
## Recommendations (in order of value/risk)
|
||||
|
||||
1. **Move Schnorr verification off the receiver coroutine** in the app's
|
||||
|
||||
+419
@@ -0,0 +1,419 @@
|
||||
/*
|
||||
* Copyright (c) 2025 Vitor Pamplona
|
||||
*
|
||||
* Permission is hereby granted, free of charge, to any person obtaining a copy of
|
||||
* this software and associated documentation files (the "Software"), to deal in
|
||||
* the Software without restriction, including without limitation the rights to use,
|
||||
* copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
|
||||
* Software, and to permit persons to whom the Software is furnished to do so,
|
||||
* subject to the following conditions:
|
||||
*
|
||||
* The above copyright notice and this permission notice shall be included in all
|
||||
* copies or substantial portions of the Software.
|
||||
*
|
||||
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
||||
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
|
||||
* FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
|
||||
* COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
|
||||
* AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
|
||||
* WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
|
||||
*/
|
||||
package com.vitorpamplona.quartz.nip01Core.relay.prodbench
|
||||
|
||||
import com.vitorpamplona.geode.KtorRelay
|
||||
import com.vitorpamplona.geode.RelayEngine
|
||||
import com.vitorpamplona.geode.fixtures.SyntheticEvents
|
||||
import com.vitorpamplona.quartz.nip01Core.core.OptimizedJsonMapper
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.client.NostrClient
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.NegentropySyncException
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.fetchAllPages
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.negentropySync
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.normalizeRelayUrl
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.server.policies.LimitsPolicy
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.server.policies.RelayLimits
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.sockets.okhttp.BasicOkHttpWebSocket
|
||||
import kotlinx.coroutines.Dispatchers
|
||||
import kotlinx.coroutines.async
|
||||
import kotlinx.coroutines.awaitAll
|
||||
import kotlinx.coroutines.coroutineScope
|
||||
import kotlinx.coroutines.runBlocking
|
||||
import okhttp3.OkHttpClient
|
||||
import okhttp3.Request
|
||||
import okhttp3.Response
|
||||
import java.util.concurrent.CountDownLatch
|
||||
import java.util.concurrent.TimeUnit
|
||||
import java.util.concurrent.atomic.AtomicLong
|
||||
import kotlin.test.Test
|
||||
|
||||
/**
|
||||
* Answers: "we need to download N-million events from a single relay as fast
|
||||
* as possible — considering just download and parsing, what helps?"
|
||||
*
|
||||
* Three parts:
|
||||
*
|
||||
* 1. LOCAL CEILINGS — a real geode relay on a localhost TCP port, seeded with
|
||||
* [LOCAL_EVENTS] events, so network bandwidth and server latency are ~free
|
||||
* and the client stack is what's measured. Variants: 1 connection vs
|
||||
* [SHARDS] connections with the `created_at` range sharded across them;
|
||||
* single giant REQ vs realistic 1000/page cursor pagination (a second
|
||||
* server instance clamps limits to force paging); and the full quartz
|
||||
* stack (parse + dispatch) vs a raw OkHttp socket that only counts frames
|
||||
* (no parse), which isolates the parse share of the pipeline.
|
||||
*
|
||||
* 2. OFFLINE PARSE STRATEGIES — what to do with each frame: full
|
||||
* `fromJsonToMessage`, full parse fanned across cores, or an id-only scan
|
||||
* (the "archive the raw frame now, parse lazily later" strategy).
|
||||
*
|
||||
* 3. PRODUCTION TEST CASE — NIP-77 negentropy sync of `kinds:[30382]` from
|
||||
* nip85.nosfabrica.com (capped at [PROD_MAX_EVENTS]) against plain
|
||||
* until-cursor paging of the same filter. Negentropy enumerates the id set
|
||||
* server-side and downloads by id-batch with 8 concurrent REQs on one
|
||||
* connection — the protocol-level answer to serial page round-trips.
|
||||
*
|
||||
* Gated behind PROD_RELAY_BENCH (same as ProductionReceiverBenchmark):
|
||||
* ./gradlew :quartz:jvmTest --tests "*.BulkDownloadBenchmark" -PprodRelayBench=1
|
||||
*/
|
||||
class BulkDownloadBenchmark {
|
||||
companion object {
|
||||
const val LOCAL_EVENTS = 100_000
|
||||
const val SHARDS = 4
|
||||
const val PAGE_LIMIT = 1_000
|
||||
const val LOCAL_PAGE_TIMEOUT_MS = 120_000L
|
||||
|
||||
const val PROD_RELAY = "wss://nip85.nosfabrica.com"
|
||||
const val PROD_KIND = 30382
|
||||
const val PROD_MAX_EVENTS = 20_000
|
||||
}
|
||||
|
||||
private fun report(
|
||||
name: String,
|
||||
events: Long,
|
||||
bytes: Long,
|
||||
wallNanos: Long,
|
||||
) {
|
||||
println(
|
||||
" %-34s %,8d events %6.1f MB wall=%6.0fms -> %,8.0f events/s %5.1f MB/s"
|
||||
.format(
|
||||
name,
|
||||
events,
|
||||
bytes / 1e6,
|
||||
wallNanos / 1e6,
|
||||
events * 1e9 / wallNanos,
|
||||
bytes * 1e3 / wallNanos,
|
||||
),
|
||||
)
|
||||
}
|
||||
|
||||
// ------------------------------------------------------------------
|
||||
// Part 1: local ceilings against a seeded geode relay on localhost TCP
|
||||
// ------------------------------------------------------------------
|
||||
|
||||
/** Splits [1, LOCAL_EVENTS] (the seeded createdAt space) into [shards] windows. */
|
||||
private fun windows(shards: Int): List<Pair<Long, Long>> {
|
||||
val step = LOCAL_EVENTS / shards
|
||||
return (0 until shards).map { i ->
|
||||
val lo = i * step + 1L
|
||||
val hi = if (i == shards - 1) LOCAL_EVENTS.toLong() else (i + 1) * step.toLong()
|
||||
lo to hi
|
||||
}
|
||||
}
|
||||
|
||||
/** Full quartz stack: NostrClient + fetchAllPages, one client per shard. */
|
||||
private suspend fun quartzDownload(
|
||||
name: String,
|
||||
relayUrl: NormalizedRelayUrl,
|
||||
httpClient: OkHttpClient,
|
||||
shards: Int,
|
||||
) {
|
||||
val count = AtomicLong(0)
|
||||
val bytes = AtomicLong(0)
|
||||
val start = System.nanoTime()
|
||||
coroutineScope {
|
||||
windows(shards)
|
||||
.map { (lo, hi) ->
|
||||
async(Dispatchers.IO) {
|
||||
val client = NostrClient(BasicOkHttpWebSocket.Builder { httpClient })
|
||||
try {
|
||||
client.fetchAllPages(
|
||||
relay = relayUrl,
|
||||
filters = listOf(Filter(kinds = listOf(1), since = lo, until = hi)),
|
||||
timeoutMs = LOCAL_PAGE_TIMEOUT_MS,
|
||||
) { event ->
|
||||
count.incrementAndGet()
|
||||
bytes.addAndGet(event.content.length.toLong())
|
||||
}
|
||||
} finally {
|
||||
client.close()
|
||||
}
|
||||
}
|
||||
}.awaitAll()
|
||||
}
|
||||
val wall = System.nanoTime() - start
|
||||
report(name, count.get(), bytes.get(), wall)
|
||||
if (count.get() != LOCAL_EVENTS.toLong()) {
|
||||
println(" !! expected $LOCAL_EVENTS events, got ${count.get()}")
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Raw OkHttp socket per shard: counts EVENT frames, no JSON parse, no
|
||||
* quartz client. The download-only floor the parse stage sits on top of.
|
||||
*/
|
||||
private fun rawDownload(
|
||||
name: String,
|
||||
serverWsUrl: String,
|
||||
httpClient: OkHttpClient,
|
||||
shards: Int,
|
||||
) {
|
||||
val count = AtomicLong(0)
|
||||
val bytes = AtomicLong(0)
|
||||
val latch = CountDownLatch(shards)
|
||||
val start = System.nanoTime()
|
||||
|
||||
val sockets =
|
||||
windows(shards).mapIndexed { i, (lo, hi) ->
|
||||
httpClient.newWebSocket(
|
||||
Request.Builder().url(serverWsUrl).build(),
|
||||
object : okhttp3.WebSocketListener() {
|
||||
override fun onOpen(
|
||||
webSocket: okhttp3.WebSocket,
|
||||
response: Response,
|
||||
) {
|
||||
webSocket.send("""["REQ","raw$i",{"kinds":[1],"since":$lo,"until":$hi}]""")
|
||||
}
|
||||
|
||||
override fun onMessage(
|
||||
webSocket: okhttp3.WebSocket,
|
||||
text: String,
|
||||
) {
|
||||
if (text.startsWith("[\"EVENT\"")) {
|
||||
count.incrementAndGet()
|
||||
bytes.addAndGet(text.length.toLong())
|
||||
} else if (text.startsWith("[\"EOSE\"")) {
|
||||
latch.countDown()
|
||||
}
|
||||
}
|
||||
|
||||
override fun onFailure(
|
||||
webSocket: okhttp3.WebSocket,
|
||||
t: Throwable,
|
||||
response: Response?,
|
||||
) {
|
||||
latch.countDown()
|
||||
}
|
||||
},
|
||||
)
|
||||
}
|
||||
|
||||
latch.await(120, TimeUnit.SECONDS)
|
||||
val wall = System.nanoTime() - start
|
||||
sockets.forEach { it.cancel() }
|
||||
report(name, count.get(), bytes.get(), wall)
|
||||
}
|
||||
|
||||
private fun localCeilings(httpClient: OkHttpClient) {
|
||||
println("\n=== LOCAL CEILINGS: geode on localhost, $LOCAL_EVENTS seeded events ===")
|
||||
|
||||
val placeholderA = "ws://127.0.0.1:7771/".normalizeRelayUrl()
|
||||
val placeholderB = "ws://127.0.0.1:7772/".normalizeRelayUrl()
|
||||
|
||||
val engine = RelayEngine(url = placeholderA)
|
||||
val seedNanos =
|
||||
runBlocking {
|
||||
val events =
|
||||
(1..LOCAL_EVENTS).map {
|
||||
SyntheticEvents.fakeEvent(
|
||||
idSeed = it,
|
||||
kind = 1,
|
||||
pubKey = SyntheticEvents.hexId(it % 1000 + 1),
|
||||
createdAt = it.toLong(),
|
||||
content = "bulk download benchmark payload $it ".repeat(6),
|
||||
)
|
||||
}
|
||||
val t = System.nanoTime()
|
||||
events.chunked(2_000).forEach { engine.store.batchInsert(it) }
|
||||
System.nanoTime() - t
|
||||
}
|
||||
println(" (seeded in %.1fs)".format(seedNanos / 1e9))
|
||||
|
||||
// Same store, second engine that clamps every REQ to PAGE_LIMIT — forces
|
||||
// fetchAllPages into realistic until-cursor pagination.
|
||||
val pagedEngine =
|
||||
RelayEngine(
|
||||
url = placeholderB,
|
||||
store = engine.store,
|
||||
policyBuilder = { LimitsPolicy(RelayLimits(defaultLimit = PAGE_LIMIT, maxLimit = PAGE_LIMIT)) },
|
||||
)
|
||||
|
||||
val openServer = KtorRelay(engine, port = 0).start()
|
||||
val pagedServer = KtorRelay(pagedEngine, port = 0).start()
|
||||
try {
|
||||
val openUrl = openServer.url.normalizeRelayUrl()
|
||||
val pagedUrl = pagedServer.url.normalizeRelayUrl()
|
||||
|
||||
runBlocking {
|
||||
rawDownload("raw 1-conn single-REQ (no parse)", openServer.url, httpClient, shards = 1)
|
||||
rawDownload("raw $SHARDS-conn sharded (no parse)", openServer.url, httpClient, shards = SHARDS)
|
||||
quartzDownload("quartz 1-conn single-REQ", openUrl, httpClient, shards = 1)
|
||||
quartzDownload("quartz $SHARDS-conn sharded", openUrl, httpClient, shards = SHARDS)
|
||||
quartzDownload("quartz 1-conn paged $PAGE_LIMIT", pagedUrl, httpClient, shards = 1)
|
||||
quartzDownload("quartz $SHARDS-conn sharded+paged", pagedUrl, httpClient, shards = SHARDS)
|
||||
}
|
||||
} finally {
|
||||
openServer.stop(gracePeriodMillis = 200, timeoutMillis = 1_000)
|
||||
pagedServer.stop(gracePeriodMillis = 200, timeoutMillis = 1_000)
|
||||
pagedEngine.close()
|
||||
engine.close()
|
||||
}
|
||||
}
|
||||
|
||||
// ------------------------------------------------------------------
|
||||
// Part 2: offline per-frame parse strategies
|
||||
// ------------------------------------------------------------------
|
||||
|
||||
private fun offlineParseStrategies() {
|
||||
println("\n=== OFFLINE: per-frame strategies (50k synthetic frames) ===")
|
||||
val frames =
|
||||
(1..50_000).map {
|
||||
val event =
|
||||
SyntheticEvents.fakeEvent(
|
||||
idSeed = it,
|
||||
kind = 1,
|
||||
pubKey = SyntheticEvents.hexId(it % 1000 + 1),
|
||||
createdAt = it.toLong(),
|
||||
content = "bulk download benchmark payload $it ".repeat(6),
|
||||
)
|
||||
"""["EVENT","sub",${event.toJson()}]"""
|
||||
}
|
||||
val totalBytes = frames.sumOf { it.length.toLong() }
|
||||
|
||||
// warmup
|
||||
frames.take(5_000).forEach { OptimizedJsonMapper.fromJsonToMessage(it) }
|
||||
|
||||
var t = System.nanoTime()
|
||||
frames.forEach { OptimizedJsonMapper.fromJsonToMessage(it) }
|
||||
report("full parse, 1 thread", frames.size.toLong(), totalBytes, System.nanoTime() - t)
|
||||
|
||||
val cores = Runtime.getRuntime().availableProcessors()
|
||||
t = System.nanoTime()
|
||||
runBlocking(Dispatchers.Default) {
|
||||
frames
|
||||
.chunked((frames.size + cores - 1) / cores)
|
||||
.map { chunk -> async { chunk.forEach { OptimizedJsonMapper.fromJsonToMessage(it) } } }
|
||||
.awaitAll()
|
||||
}
|
||||
report("full parse, $cores threads", frames.size.toLong(), totalBytes, System.nanoTime() - t)
|
||||
|
||||
// id-only scan: what "write the raw frame to disk now, parse lazily
|
||||
// later" pays per frame to dedup/route.
|
||||
var sink = 0
|
||||
t = System.nanoTime()
|
||||
frames.forEach { frame ->
|
||||
val i = frame.indexOf("\"id\":\"")
|
||||
if (i >= 0) sink += frame[i + 6].code
|
||||
}
|
||||
report("id-scan only (archive raw)", frames.size.toLong(), totalBytes, System.nanoTime() - t)
|
||||
check(sink != 0)
|
||||
}
|
||||
|
||||
// ------------------------------------------------------------------
|
||||
// Part 3: production — negentropy vs paging for kind 30382 on nosfabrica
|
||||
// ------------------------------------------------------------------
|
||||
|
||||
private suspend fun productionNegentropy(httpClient: OkHttpClient) {
|
||||
println("\n=== PRODUCTION: $PROD_RELAY kinds=[$PROD_KIND], capped at $PROD_MAX_EVENTS events ===")
|
||||
val relay = PROD_RELAY.normalizeRelayUrl()
|
||||
|
||||
// A) NIP-77 negentropy: server enumerates the id set, we download by
|
||||
// id-batches with 8 concurrent REQs on the same connection.
|
||||
run {
|
||||
val client = NostrClient(BasicOkHttpWebSocket.Builder { httpClient })
|
||||
try {
|
||||
val count = AtomicLong(0)
|
||||
val bytes = AtomicLong(0)
|
||||
val start = System.nanoTime()
|
||||
val result =
|
||||
client.negentropySync(
|
||||
relay = relay,
|
||||
filter = Filter(kinds = listOf(PROD_KIND)),
|
||||
maxEvents = PROD_MAX_EVENTS,
|
||||
onProgress = { need, downloaded ->
|
||||
if (downloaded % 5_000 == 0 && downloaded > 0) {
|
||||
println(" … negentropy progress: need=$need downloaded=$downloaded")
|
||||
}
|
||||
},
|
||||
) { event ->
|
||||
count.incrementAndGet()
|
||||
bytes.addAndGet(event.content.length.toLong())
|
||||
}
|
||||
val wall = System.nanoTime() - start
|
||||
report("negentropy sync", count.get(), bytes.get(), wall)
|
||||
println(
|
||||
" relay reported need=${result.needCount} ids for the full set, " +
|
||||
"reconciled in ${result.windows} window(s), downloaded=${result.downloaded}",
|
||||
)
|
||||
} catch (e: NegentropySyncException) {
|
||||
println(" !! negentropy failed: ${e.reason} — ${e.message}")
|
||||
} catch (e: Exception) {
|
||||
println(" !! negentropy errored: ${e::class.simpleName} ${e.message}")
|
||||
} finally {
|
||||
client.close()
|
||||
}
|
||||
}
|
||||
|
||||
// B) Plain until-cursor paging of the same filter, same cap.
|
||||
run {
|
||||
val client = NostrClient(BasicOkHttpWebSocket.Builder { httpClient })
|
||||
try {
|
||||
val count = AtomicLong(0)
|
||||
val bytes = AtomicLong(0)
|
||||
var pages = 0
|
||||
val start = System.nanoTime()
|
||||
client.fetchAllPages(
|
||||
relay = relay,
|
||||
filters = listOf(Filter(kinds = listOf(PROD_KIND), limit = PROD_MAX_EVENTS)),
|
||||
timeoutMs = 30_000L,
|
||||
onNewPage = { pages++ },
|
||||
) { event ->
|
||||
count.incrementAndGet()
|
||||
bytes.addAndGet(event.content.length.toLong())
|
||||
}
|
||||
val wall = System.nanoTime() - start
|
||||
report("paged download", count.get(), bytes.get(), wall)
|
||||
println(" pages=${pages + 1}")
|
||||
} catch (e: Exception) {
|
||||
println(" !! paged download errored: ${e::class.simpleName} ${e.message}")
|
||||
} finally {
|
||||
client.close()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ------------------------------------------------------------------
|
||||
|
||||
@Test
|
||||
fun bulkDownloadBenchmark() {
|
||||
if (System.getenv("PROD_RELAY_BENCH") == null && System.getProperty("prodRelayBench") == null) {
|
||||
println("BulkDownloadBenchmark skipped. Run with -PprodRelayBench=1 to enable.")
|
||||
return
|
||||
}
|
||||
|
||||
val httpClient =
|
||||
OkHttpClient
|
||||
.Builder()
|
||||
.connectTimeout(15, TimeUnit.SECONDS)
|
||||
.readTimeout(120, TimeUnit.SECONDS)
|
||||
.pingInterval(30, TimeUnit.SECONDS)
|
||||
.build()
|
||||
|
||||
println("=== BULK DOWNLOAD BENCHMARK === cores=${Runtime.getRuntime().availableProcessors()}")
|
||||
|
||||
localCeilings(httpClient)
|
||||
offlineParseStrategies()
|
||||
runBlocking { productionNegentropy(httpClient) }
|
||||
|
||||
httpClient.dispatcher.executorService.shutdown()
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user