diff --git a/geode/config.example.toml b/geode/config.example.toml index 5835a5527..3d706999b 100644 --- a/geode/config.example.toml +++ b/geode/config.example.toml @@ -25,6 +25,20 @@ contact = "admin@example.com" host = "0.0.0.0" port = 7447 path = "/" +# Ktor CIO event-loop pool sizing. Leave commented-out for sensible +# per-CPU defaults (typical for <2k concurrent connections). Lift on +# big-VM deployments targeting 10k+ connections — over-threading at +# low connection counts hurts L1/L2 cache locality, so always +# benchmark before/after when tuning these. +# +# Operators targeting >1k concurrent WebSockets should also raise the +# OS file-descriptor limit: `ulimit -n 65536` (or higher) before +# launching, plus a matching `LimitNOFILE=` in any systemd unit. The +# default of 1024 on most distros caps the relay well below 1k FDs +# (one per WS plus DB and listening sockets). +# connection_group_size = 4 +# worker_group_size = 16 +# call_group_size = 64 [database] # True keeps an in-memory SQLite db (events vanish on restart). Useful diff --git a/geode/src/main/kotlin/com/vitorpamplona/geode/LocalRelayServer.kt b/geode/src/main/kotlin/com/vitorpamplona/geode/LocalRelayServer.kt index 3840bf482..56ef2f6da 100644 --- a/geode/src/main/kotlin/com/vitorpamplona/geode/LocalRelayServer.kt +++ b/geode/src/main/kotlin/com/vitorpamplona/geode/LocalRelayServer.kt @@ -32,8 +32,10 @@ import io.ktor.http.ContentType import io.ktor.http.HttpHeaders import io.ktor.http.HttpStatusCode import io.ktor.server.application.install +import io.ktor.server.application.serverConfig import io.ktor.server.cio.CIO import io.ktor.server.cio.CIOApplicationEngine +import io.ktor.server.engine.connector import io.ktor.server.engine.embeddedServer import io.ktor.server.request.header import io.ktor.server.response.respondText @@ -109,6 +111,16 @@ class LocalRelayServer( * RPC payload. */ val maxAdminBodyBytes: Int = 1 shl 20, + /** + * Ktor CIO acceptor-thread count. `null` keeps Ktor's default. + * Lift on machines with many cores when targeting 10k+ + * concurrent connections — see `[network]` config docs. + */ + val connectionGroupSize: Int? = null, + /** Ktor CIO worker-thread count. `null` keeps Ktor's default. */ + val workerGroupSize: Int? = null, + /** Ktor CIO call-handling thread count. `null` keeps Ktor's default. */ + val callGroupSize: Int? = null, ) { private val infoHolder = object : Nip86Server.InfoHolder { @@ -167,51 +179,78 @@ class LocalRelayServer( * [url] is safe to read on the very next line. */ fun start(): LocalRelayServer { + // Snapshot the constructor-supplied overrides into locals so + // the `configure` lambda below can assign to its receiver + // without the names colliding with outer properties. + val connGrp = connectionGroupSize + val workGrp = workerGroupSize + val callGrp = callGroupSize + val bindHost = host + val bindPort = port val server = - embeddedServer(CIO, host = host, port = port) { - install(WebSockets) { - maxFrameBytes?.let { maxFrameSize = it } - } - routing { - // NIP-11: GET on the relay URL with Accept: - // application/nostr+json returns the relay info doc. - // We mount this *before* the webSocket route so Ktor - // serves NIP-11 for plain HTTP GETs and only upgrades - // to a WebSocket when the request is a WS upgrade. - get(path) { - val accept = call.request.header(HttpHeaders.Accept).orEmpty() - if (accept.contains("application/nostr+json")) { - call.response.headers.append("Access-Control-Allow-Origin", "*") - call.respondText( - relay.info.json, - ContentType.parse("application/nostr+json"), - ) - } else { - call.respondText( - "Use a Nostr client (NIP-01 WebSocket) or send Accept: application/nostr+json (NIP-11).", - ContentType.Text.Plain, - HttpStatusCode.UpgradeRequired, - ) + embeddedServer( + factory = CIO, + rootConfig = + serverConfig { + module { + install(WebSockets) { + maxFrameBytes?.let { maxFrameSize = it } + } + routing { + // NIP-11: GET on the relay URL with Accept: + // application/nostr+json returns the relay info doc. + // We mount this *before* the webSocket route so Ktor + // serves NIP-11 for plain HTTP GETs and only upgrades + // to a WebSocket when the request is a WS upgrade. + get(path) { + val accept = call.request.header(HttpHeaders.Accept).orEmpty() + if (accept.contains("application/nostr+json")) { + call.response.headers.append("Access-Control-Allow-Origin", "*") + call.respondText( + relay.info.json, + ContentType.parse("application/nostr+json"), + ) + } else { + call.respondText( + "Use a Nostr client (NIP-01 WebSocket) or send Accept: application/nostr+json (NIP-11).", + ContentType.Text.Plain, + HttpStatusCode.UpgradeRequired, + ) + } + } + // NIP-86: POST application/nostr+json+rpc with a NIP-98 + // signed Authorization header → JSON-RPC dispatch. + post(path) { + nip86Route.handle(call) + } + webSocket(path) { + if (shuttingDown) { + // Just return — Ktor closes the WS for us. + return@webSocket + } + WebSocketSessionPump(this).pump( + server = relay.server, + registerSession = activeSessions::add, + unregisterSession = activeSessions::remove, + ) + } + } } + }, + configure = { + connector { + host = bindHost + port = bindPort } - // NIP-86: POST application/nostr+json+rpc with a NIP-98 - // signed Authorization header → JSON-RPC dispatch. - post(path) { - nip86Route.handle(call) - } - webSocket(path) { - if (shuttingDown) { - // Just return — Ktor closes the WS for us. - return@webSocket - } - WebSocketSessionPump(this).pump( - server = relay.server, - registerSession = activeSessions::add, - unregisterSession = activeSessions::remove, - ) - } - } - } + // Keep Ktor defaults unless the operator overrode + // them — Ktor's per-CPU sizing is sensible for + // most deployments, and over-threading hurts L1/L2 + // locality at low connection counts. + connGrp?.let { connectionGroupSize = it } + workGrp?.let { workerGroupSize = it } + callGrp?.let { callGroupSize = it } + }, + ) server.start(wait = false) engine = server.engine // Ktor 3.x made resolvedConnectors() suspend. We block here so diff --git a/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt b/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt index 228c40709..e7b7a97b0 100644 --- a/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt +++ b/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt @@ -117,6 +117,9 @@ fun main(args: Array) { maxFrameBytes = frameLimit, adminPubkeys = config.admin.pubkeys.toSet(), publicUrl = config.admin.public_url, + connectionGroupSize = config.network.connection_group_size, + workerGroupSize = config.network.worker_group_size, + callGroupSize = config.network.call_group_size, ).start() Runtime.getRuntime().addShutdownHook( diff --git a/geode/src/main/kotlin/com/vitorpamplona/geode/config/RelayConfig.kt b/geode/src/main/kotlin/com/vitorpamplona/geode/config/RelayConfig.kt index b1b99634d..1417ec61f 100644 --- a/geode/src/main/kotlin/com/vitorpamplona/geode/config/RelayConfig.kt +++ b/geode/src/main/kotlin/com/vitorpamplona/geode/config/RelayConfig.kt @@ -96,6 +96,29 @@ data class RelayConfig( val host: String = "0.0.0.0", val port: Int = 7447, val path: String = "/", + /** + * Ktor CIO acceptor-thread count. `null` (default) keeps Ktor's + * default sizing — fine up to a few thousand concurrent + * connections. On big-VM deployments targeting 10k+ + * connections, lift this to roughly half the available cores + * so the acceptor doesn't starve workers. + */ + val connection_group_size: Int? = null, + /** + * Ktor CIO worker-thread count (handles socket I/O). `null` + * keeps Ktor's default. Each connection's WebSocket read/write + * is dispatched onto this pool; for many idle long-lived + * connections the pool can stay small, but 10k+ connections + * benefit from sizing this to the full CPU count. + */ + val worker_group_size: Int? = null, + /** + * Ktor CIO call-handling thread count. `null` keeps Ktor's + * default. Sized higher than [worker_group_size] because each + * call (incl. WebSocket upgrade) may suspend on I/O — at + * 10k+ connections, ~4× cores is a reasonable starting point. + */ + val call_group_size: Int? = null, ) data class DatabaseSection( diff --git a/geode/src/main/kotlin/com/vitorpamplona/geode/server/WebSocketSessionPump.kt b/geode/src/main/kotlin/com/vitorpamplona/geode/server/WebSocketSessionPump.kt index 7cd700a33..f5319779d 100644 --- a/geode/src/main/kotlin/com/vitorpamplona/geode/server/WebSocketSessionPump.kt +++ b/geode/src/main/kotlin/com/vitorpamplona/geode/server/WebSocketSessionPump.kt @@ -29,6 +29,7 @@ import kotlinx.coroutines.channels.Channel import kotlinx.coroutines.channels.ClosedSendChannelException import kotlinx.coroutines.channels.consumeEach import kotlinx.coroutines.launch +import java.util.concurrent.atomic.AtomicInteger /** * Per-WebSocket pump that owns the bounded outbound queue and the @@ -44,15 +45,39 @@ import kotlinx.coroutines.launch * 3. `finally`-style teardown closes the queue, cancels the * writer, unregisters the session, and closes it. * - * Slow-client policy: when [outQueue] fills, [SESSION_OUTGOING_BUFFER] - * frames behind, the connection is dropped rather than silently - * losing EVENT/EOSE — silent drop would corrupt NIP-01. + * Slow-client policy: once the outbound backlog reaches + * [MAX_OUTGOING_BUFFER] frames, the connection is dropped rather + * than silently losing EVENT/EOSE — silent drop would corrupt + * NIP-01. + * + * Memory model: the outbound queue is `Channel.UNLIMITED`, which in + * kotlinx.coroutines allocates segments lazily — an idle connection + * pays only a small head-segment cost. The cap is enforced via + * [outstanding] rather than the channel's own capacity so we don't + * reserve a fixed-size buffer up-front for every connection. At + * 5 000+ idle connections this matters: an 8 192-slot fixed buffer + * per connection would otherwise dominate JVM heap usage even + * though the vast majority of connections never fan out. */ internal class WebSocketSessionPump( private val ws: DefaultWebSocketServerSession, ) { - private val outQueue = Channel(capacity = SESSION_OUTGOING_BUFFER) - private var droppedForBackpressure = false + /** + * Unbounded channel — bounded by [outstanding] above, not by the + * channel's own capacity. See class kdoc for memory rationale. + */ + private val outQueue = Channel(capacity = Channel.UNLIMITED) + + /** + * Number of frames queued but not yet written to the socket. + * Producer increments before [Channel.trySend]; writer decrements + * after the frame is handed to Ktor. When this would cross + * [MAX_OUTGOING_BUFFER] we treat the client as slow and close + * the queue. + */ + private val outstanding = AtomicInteger(0) + + @Volatile private var droppedForBackpressure = false suspend fun pump( server: NostrServer, @@ -64,6 +89,7 @@ internal class WebSocketSessionPump( try { for (json in outQueue) { ws.outgoing.send(Frame.Text(json)) + outstanding.decrementAndGet() } } catch (_: ClosedSendChannelException) { // socket closed — outer handler runs normal teardown. @@ -71,13 +97,22 @@ internal class WebSocketSessionPump( } val session = server.connect { json -> - val res = outQueue.trySend(json) - if (!res.isSuccess && !res.isClosed) { - // Buffer is full → slow client. Mark + close the - // queue; the writer drains, then the outer handler - // closes the WS session. + // The channel itself is UNLIMITED, so trySend can't + // report "full". Enforce the cap explicitly: increment + // first, refuse if we'd cross the bound, otherwise + // enqueue. + val depth = outstanding.incrementAndGet() + if (depth > MAX_OUTGOING_BUFFER) { + outstanding.decrementAndGet() droppedForBackpressure = true outQueue.close() + return@connect + } + val res = outQueue.trySend(json) + if (!res.isSuccess) { + // Channel was closed concurrently (e.g. teardown). + // Roll back the counter; nothing more to do. + outstanding.decrementAndGet() } } registerSession(session) @@ -98,17 +133,19 @@ internal class WebSocketSessionPump( companion object { /** - * Per-session outbound buffer size. When a slow client falls + * Per-session outbound backlog cap. When a slow client falls * this many frames behind, we close their connection rather * than silently dropping further frames (which would corrupt * NIP-01 by missing EVENT/EOSE messages). * * Sized to hold fan-out for a connection holding several * thousand subscriptions when one event matches all of them - * — the realistic upper bound for a relay client. At ~250B - * per frame this caps per-session memory at ~2 MiB before - * we drop the connection. + * — the realistic upper bound for a relay client. At ~250 B + * per frame this caps per-session worst-case memory at + * ~2 MiB before we drop the connection. Idle connections + * pay only the small head-segment cost of an unlimited + * channel (≈ a few hundred bytes), not the full cap. */ - const val SESSION_OUTGOING_BUFFER: Int = 8192 + const val MAX_OUTGOING_BUFFER: Int = 8192 } } diff --git a/geode/src/test/kotlin/com/vitorpamplona/geode/perf/LoadBenchmark.kt b/geode/src/test/kotlin/com/vitorpamplona/geode/perf/LoadBenchmark.kt index a083dd812..839d22f58 100644 --- a/geode/src/test/kotlin/com/vitorpamplona/geode/perf/LoadBenchmark.kt +++ b/geode/src/test/kotlin/com/vitorpamplona/geode/perf/LoadBenchmark.kt @@ -139,6 +139,192 @@ class LoadBenchmark { } } + /** + * Holds 10 000 idle WebSocket connections open against a single + * relay. Verifies that the adaptive outQueue (sketch A in + * [connection-scaling plan][1]) lets us cross the ~2 000-connection + * floor measured by [connectionsHeldOpen] without FD exhaustion or + * runaway RSS. + * + * RUN PREREQ: requires a process FD limit ≥ ~12 000 (each WS uses + * one FD on each side plus margin). On Linux: `ulimit -n 32768` + * before launching the test JVM. + * + * [1]: geode/plans/2026-05-07-connection-scaling.md + */ + @Test + fun connectionsHeldOpen10k() = + benchmark("connections held open 10k") { + val target = 10_000 + runBenchmarkServer { server, http -> + val httpUrl = + okhttp3.Request + .Builder() + .url(server.url.replace("ws://", "http://")) + .build() + val sockets = java.util.concurrent.CopyOnWriteArrayList() + val opened = AtomicLong() + val gotEose = AtomicLong() + val opens = + measureTime { + repeat(target) { + val ws = + http.newWebSocket( + httpUrl, + object : okhttp3.WebSocketListener() { + override fun onOpen( + webSocket: okhttp3.WebSocket, + response: okhttp3.Response, + ) { + opened.incrementAndGet() + webSocket.send( + """["REQ","s",{"kinds":[1],"limit":1}]""", + ) + } + + override fun onMessage( + webSocket: okhttp3.WebSocket, + text: String, + ) { + if (text.startsWith("[\"EOSE\"")) { + gotEose.incrementAndGet() + } + } + }, + ) + sockets += ws + } + val deadline = System.currentTimeMillis() + 120_000 + while (gotEose.get() < target && System.currentTimeMillis() < deadline) { + Thread.sleep(50) + } + } + val rt = Runtime.getRuntime() + val rssMb = (rt.totalMemory() - rt.freeMemory()) / (1024 * 1024) + println( + "target=$target opened=${opened.get()} eosed=${gotEose.get()} " + + "active=${server.activeSessionCount} elapsedMs=${opens.inWholeMilliseconds} " + + "heapMb=$rssMb", + ) + sockets.forEach { runCatching { it.cancel() } } + check(gotEose.get() == target.toLong()) { + "expected $target EOSE but got ${gotEose.get()} — connection scaling regression" + } + check(rssMb < 1024) { + "heap usage $rssMb MiB exceeded 1 GiB ceiling for $target idle connections" + } + } + } + + /** + * 5 000 idle subscribers, one publisher emitting 10 EPS for 10 s. + * Measures fan-out latency at scale — exercises the queue path + * for a connection that *does* fan out, not just an idle one. + * + * Each subscriber matches every published event (`kinds:[1]`), + * so a single EVENT generates 5 000 outbound frames per tick. + */ + @Test + fun connectionsHeldOpenWithFanout() = + benchmark("connections held open with fanout") { + val subs = 5_000 + val durationSeconds = 10 + val targetEps = 10 + runBenchmarkServer { server, http -> + val scope = CoroutineScope(Dispatchers.Default + SupervisorJob()) + val subClient = NostrClient(BasicOkHttpWebSocket.Builder { _ -> http }, scope) + val pubClient = NostrClient(BasicOkHttpWebSocket.Builder { _ -> http }, scope) + try { + val relayUrl = server.url.normalizeRelayUrl() + val received = AtomicLong() + val eosed = AtomicLong() + // Per-event fan-out latency, capped at the # of + // events we expect (durationSeconds * targetEps). + val fanoutLatenciesNs = + java.util.concurrent.ConcurrentHashMap() + val firstSeenNs = + java.util.concurrent.ConcurrentHashMap() + + repeat(subs) { i -> + subClient.subscribe( + "fanout-$i", + mapOf(relayUrl to listOf(Filter(kinds = listOf(1)))), + object : SubscriptionListener { + override fun onEvent( + event: com.vitorpamplona.quartz.nip01Core.core.Event, + isLive: Boolean, + relay: NormalizedRelayUrl, + forFilters: List?, + ) { + val now = System.nanoTime() + firstSeenNs + .computeIfAbsent(event.id) { AtomicLong(now) } + fanoutLatenciesNs + .computeIfAbsent(event.id) { AtomicLong(now) } + .set(now) + received.incrementAndGet() + } + + override fun onEose( + relay: NormalizedRelayUrl, + forFilters: List?, + ) { + eosed.incrementAndGet() + } + }, + ) + } + + runBlocking { + withTimeout(120_000) { + while (eosed.get() < subs) kotlinx.coroutines.delay(100) + } + } + println("$subs subs ready; publishing ${targetEps * durationSeconds} events at $targetEps EPS...") + + val signer = NostrSignerSync(KeyPair()) + val publishedAt = java.util.concurrent.ConcurrentHashMap() + val totalEvents = targetEps * durationSeconds + val tickIntervalMs = 1000L / targetEps + + runBlocking { + repeat(totalEvents) { i -> + val event = signer.sign(TextNoteEvent.build("fanout-$i")) + publishedAt[event.id] = System.nanoTime() + pubClient.publishAndConfirm(event, setOf(relayUrl)) + kotlinx.coroutines.delay(tickIntervalMs) + } + } + + // Wait for fan-out completion (or 30s, whichever first). + runBlocking { + withTimeout(30_000) { + while (received.get() < subs.toLong() * totalEvents) { + kotlinx.coroutines.delay(100) + } + } + } + + val perEventLastMs = + fanoutLatenciesNs.entries + .mapNotNull { (id, last) -> + publishedAt[id]?.let { (last.get() - it) / 1_000_000.0 } + }.sorted() + val p50 = perEventLastMs.getOrNull(perEventLastMs.size / 2) ?: -1.0 + val p99 = perEventLastMs.getOrNull((perEventLastMs.size * 99) / 100) ?: -1.0 + println( + "subs=$subs events=$totalEvents received=${received.get()}/${subs.toLong() * totalEvents} " + + "p50LastFanoutMs=${"%.1f".format(p50)} " + + "p99LastFanoutMs=${"%.1f".format(p99)}", + ) + } finally { + subClient.disconnect() + pubClient.disconnect() + scope.cancel() + } + } + } + /** * One publisher sends 10k events serially. Measures the round-trip * `EVENT` → `OK true` time, which is dominated by SQLite write