From ea1093adaf6e6e673f2b49665d9f1de814cd6ea1 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 9 Jul 2026 17:56:22 +0000 Subject: [PATCH] perf(relay): isolate WS frame dispatch onto a dedicated pool, off Dispatchers.IO MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Measured on a GrapeRank crawl, frame decode/dispatch (per-connection consumer coroutines) ran on the shared Dispatchers.IO — the same pool that runs the store's blocking SQLite inserts. During event floods, frame coroutines queued behind those inserts: mean 200ms and up to 3.5s of dispatch lag, with 43k frames waiting >1s in our pipeline. That lag also skews the relay-idle/EOSE timing the crawler reads. Give frame processing its own daemon thread pool (sized to a small multiple of cores; decode is light + CPU-bound), shared across all connections. Frame delivery stays prompt regardless of what the IO pool is doing. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01MSW59hJtP4Yn8fnRUxc7F5 --- .../sockets/okhttp/BasicOkHttpWebSocket.kt | 20 +++++++++++++++++-- 1 file changed, 18 insertions(+), 2 deletions(-) diff --git a/quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/nip01Core/relay/sockets/okhttp/BasicOkHttpWebSocket.kt b/quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/nip01Core/relay/sockets/okhttp/BasicOkHttpWebSocket.kt index 1200a264ab..4497909ce6 100644 --- a/quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/nip01Core/relay/sockets/okhttp/BasicOkHttpWebSocket.kt +++ b/quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/nip01Core/relay/sockets/okhttp/BasicOkHttpWebSocket.kt @@ -28,7 +28,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.sockets.WebsocketBuilder import com.vitorpamplona.quartz.utils.Log import kotlinx.coroutines.CoroutineExceptionHandler import kotlinx.coroutines.CoroutineScope -import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.asCoroutineDispatcher import kotlinx.coroutines.cancel import kotlinx.coroutines.channels.Channel import kotlinx.coroutines.channels.trySendBlocking @@ -36,6 +36,7 @@ import kotlinx.coroutines.launch import okhttp3.OkHttpClient import okhttp3.Request import okhttp3.Response +import java.util.concurrent.Executors import kotlin.time.TimeSource import kotlin.time.TimeSource.Monotonic.ValueTimeMark import okhttp3.WebSocket as OkHttpWebSocket @@ -52,6 +53,21 @@ class BasicOkHttpWebSocket( CoroutineExceptionHandler { _, throwable -> Log.e("BasicOkHttpWebSocket", "WebsocketListener Caught exception: ${throwable.message}", throwable) } + + // Frame decode + dispatch runs on its OWN pool, isolated from Dispatchers.IO. + // The shared IO pool also runs the store write path (blocking SQLite inserts); + // measured on a GrapeRank crawl, frame-processing coroutines were queueing + // behind those inserts, adding a 200ms MEAN and up to 3.5s TAIL of dispatch lag + // during event floods — which in turn skews the relay-idle/EOSE timing the + // crawler reads. A dedicated daemon pool keeps frame delivery prompt regardless + // of what the IO pool is doing. Shared across all connections; sized to the + // machine (decode is light + CPU-bound, so a small multiple of cores suffices). + private val frameDispatcher = + Executors + .newFixedThreadPool( + (Runtime.getRuntime().availableProcessors() * 2).coerceIn(4, 32), + ) { r -> Thread(r, "ws-frame").apply { isDaemon = true } } + .asCoroutineDispatcher() } private var socket: OkHttpWebSocket? = null @@ -63,7 +79,7 @@ class BasicOkHttpWebSocket( val listener = object : OkHttpWebSocketListener() { - val scope = CoroutineScope(Dispatchers.IO + exceptionHandler) + val scope = CoroutineScope(frameDispatcher + exceptionHandler) // UNLIMITED on purpose — do NOT bound this channel. The app // holds 2000+ relay connections; a bounded buffer under a