laantungir/amethyst
1
0
Fork 0
mirror of https://github.com/vitorpamplona/amethyst.git synced 2026-07-22 15:52:22 +00:00
Code Issues Packages Projects Releases Wiki Activity

perf(relay): isolate WS frame dispatch onto a dedicated pool, off Dispatchers.IO

Browse Source 
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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01MSW59hJtP4Yn8fnRUxc7F5
This commit is contained in:
Claude
2026-07-09 17:56:22 +00:00
parent 4c20522d78
commit ea1093adaf
1 changed files with 18 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/nip01Core/relay/sockets/okhttp/BasicOkHttpWebSocket.kt
View File

@@ -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