nick/amethyst
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

perf(geode): adaptive outQueue + CIO pool sizing for 10k+ connections

Browse Source 
Implements geode/plans/2026-05-07-connection-scaling.md to push the
single-relay connection ceiling past the ~2k floor measured by
LoadBenchmark.connectionsHeldOpen.

- WebSocketSessionPump: switch outQueue to Channel.UNLIMITED bounded
  by an AtomicInteger backlog cap. Idle connections no longer reserve
  an 8 192-slot fixed buffer; lazy-allocated head segments cost only
  a few hundred bytes per idle session. Slow-client policy preserved:
  once outstanding > MAX_OUTGOING_BUFFER, the queue is closed and
  the connection drops, so NIP-01 ordering is never silently
  violated.
- LocalRelayServer / RelayConfig.NetworkSection: expose CIO
  connectionGroupSize / workerGroupSize / callGroupSize so big-VM
  operators can tune Ktor's event-loop pools without forking. Switch
  to the serverConfig {} + configure {} embeddedServer overload so
  CIO tunables can be set.
- LoadBenchmark: add connectionsHeldOpen10k (asserts 10 000 idle
  WebSockets settle under a 1 GiB heap ceiling) and
  connectionsHeldOpenWithFanout (5 000 subscribers, 10 EPS fan-out
  for 10 s, reports p50/p99 last-fanout latency).
- config.example.toml: document the three CIO tunables and the
  ulimit -n requirement for operators targeting >1k connections.
This commit is contained in:
Claude
2026-05-07 20:12:50 +00:00
parent 9bf17f44af
commit 0f7d8696de
6 changed files with 359 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files
geode/config.example.toml
+14
View File
@@ -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
geode/src/main/kotlin/com/vitorpamplona/geode/LocalRelayServer.kt
+81 -42
View File
@@ -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
geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt
+3
View File
@@ -117,6 +117,9 @@ fun main(args: Array<String>) {
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(
geode/src/main/kotlin/com/vitorpamplona/geode/config/RelayConfig.kt
+23
View File
@@ -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(
geode/src/main/kotlin/com/vitorpamplona/geode/server/WebSocketSessionPump.kt
+52 -15
View File
@@ -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<String>(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<String>(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
}
}
geode/src/test/kotlin/com/vitorpamplona/geode/perf/LoadBenchmark.kt
+186
View File
@@ -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<okhttp3.WebSocket>()
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<String, AtomicLong>()
val firstSeenNs =
java.util.concurrent.ConcurrentHashMap<String, AtomicLong>()
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<Filter>?,
) {
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<Filter>?,
) {
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<String, Long>()
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