From 7e61d24962f572d364d70716c0f2f1793ce69826 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 2 Jul 2026 16:09:49 +0000 Subject: [PATCH 1/4] perf: remove needless blocking on two per-packet/per-event hot paths Part of the dispatchers/thread-caps audit. Two low-risk fixes that remove thread-blocking work from paths hit on every event / every packet: - LargeCache (iOS actual): drop the runBlocking wrapper around createIfAbsent. The block contained only synchronous CacheMap ops (the same get/put getOrCreate already calls without runBlocking), so it was pure dispatcher-blocking overhead on the per-event ingest path. Now mirrors the JVM actual's plain-function shape. - QUIC JCA AEADs (AES-GCM + ChaCha20-Poly1305): split the single `synchronized(this)` monitor into disjoint encryptLock / decryptLock. seal-family touches only encryptCipher + recentEncryptNonces; open-family touches only decryptCipher, so a connection's send loop and read loop no longer serialize against each other through crypto on every packet. The documented defence-in-depth against cross-coroutine Cipher corruption is preserved per direction. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01ANuUziXKRafSTBxbh4SMoq --- .../quartz/utils/cache/LargeCache.apple.kt | 20 +++++------ .../quic/crypto/JcaAesGcmAead.kt | 33 ++++++++++++------- .../quic/crypto/JcaChaCha20Poly1305Aead.kt | 27 ++++++++++----- 3 files changed, 49 insertions(+), 31 deletions(-) diff --git a/quartz/src/appleMain/kotlin/com/vitorpamplona/quartz/utils/cache/LargeCache.apple.kt b/quartz/src/appleMain/kotlin/com/vitorpamplona/quartz/utils/cache/LargeCache.apple.kt index c64e575b35..c1a71e0709 100644 --- a/quartz/src/appleMain/kotlin/com/vitorpamplona/quartz/utils/cache/LargeCache.apple.kt +++ b/quartz/src/appleMain/kotlin/com/vitorpamplona/quartz/utils/cache/LargeCache.apple.kt @@ -22,7 +22,6 @@ package com.vitorpamplona.quartz.utils.cache import io.github.charlietap.cachemap.CacheMap import io.github.charlietap.cachemap.cacheMapOf -import kotlinx.coroutines.runBlocking // An implementation of a Threadsafe map, using CacheMap. // Investigating a Swift-based alternative(for now) @@ -70,17 +69,16 @@ actual class LargeCache : ICacheOperations { actual fun createIfAbsent( key: K, builder: (key: K) -> V, - ): Boolean = - runBlocking { - val value = concurrentMap.get(key) - if (value != null) { - false - } else { - val newObject = builder(key) - concurrentMap.put(key, newObject) - concurrentMap[key] != null - } + ): Boolean { + val value = concurrentMap.get(key) + return if (value != null) { + false + } else { + val newObject = builder(key) + concurrentMap.put(key, newObject) + concurrentMap[key] != null } + } actual override fun size(): Int = concurrentMap.size diff --git a/quic/src/jvmAndroid/kotlin/com/vitorpamplona/quic/crypto/JcaAesGcmAead.kt b/quic/src/jvmAndroid/kotlin/com/vitorpamplona/quic/crypto/JcaAesGcmAead.kt index 501e9166bc..f51386dfc9 100644 --- a/quic/src/jvmAndroid/kotlin/com/vitorpamplona/quic/crypto/JcaAesGcmAead.kt +++ b/quic/src/jvmAndroid/kotlin/com/vitorpamplona/quic/crypto/JcaAesGcmAead.kt @@ -34,12 +34,18 @@ import javax.crypto.spec.SecretKeySpec * (which is much cheaper than `getInstance`) plus the AEAD math itself. * * Single-thread per direction: one PacketProtection feeds either the read - * loop OR the send loop, never both. The class still synchronizes on a - * private monitor as a defence-in-depth: the lock-split refactor in - * `QuicConnectionDriver` keeps each side single-threaded by design, but - * a future caller (test harness, key-update path) sharing the instance - * across coroutines would otherwise corrupt the cached `Cipher` state - * silently. The JCA `Cipher` itself is documented as not thread-safe. + * loop OR the send loop, never both. The class still synchronizes as + * defence-in-depth: the lock-split refactor in `QuicConnectionDriver` + * keeps each side single-threaded by design, but a future caller (test + * harness, key-update path) sharing the instance across coroutines would + * otherwise corrupt the cached `Cipher` state silently. The JCA `Cipher` + * itself is documented as not thread-safe. + * + * Two independent monitors — [encryptLock] guards the encrypt-side state + * ([encryptCipher] + [recentEncryptNonces]); [decryptLock] guards + * [decryptCipher]. The two state groups are disjoint, so seal (send loop) + * and open (read loop) — which genuinely run on different coroutines — + * no longer serialize against each other per packet. */ class JcaAesGcmAead( key: ByteArray, @@ -67,6 +73,11 @@ class JcaAesGcmAead( private val encryptCipher: Cipher = Cipher.getInstance("AES/GCM/NoPadding") private val decryptCipher: Cipher = Cipher.getInstance("AES/GCM/NoPadding") + // Disjoint monitors so encrypt and decrypt don't serialize against each + // other. seal-family holds encryptLock; open-family holds decryptLock. + private val encryptLock = Any() + private val decryptLock = Any() + /** * Last nonce successfully consumed by [seal]. We use a fresh * [Cipher.getInstance] when the caller asks us to seal under the @@ -89,7 +100,7 @@ class JcaAesGcmAead( aad: ByteArray, plaintext: ByteArray, ): ByteArray = - synchronized(this) { + synchronized(encryptLock) { val reuse = recentEncryptNonces.any { it.contentEquals(nonce) } if (reuse) { val fresh = Cipher.getInstance("AES/GCM/NoPadding") @@ -132,7 +143,7 @@ class JcaAesGcmAead( aad: ByteArray, ciphertext: ByteArray, ): ByteArray? = - synchronized(this) { + synchronized(decryptLock) { try { decryptCipher.init(Cipher.DECRYPT_MODE, keySpec, GCMParameterSpec(128, nonce)) decryptCipher.updateAAD(aad) @@ -161,7 +172,7 @@ class JcaAesGcmAead( ciphertextOffset: Int, ciphertextLength: Int, ): ByteArray? = - synchronized(this) { + synchronized(decryptLock) { try { decryptCipher.init(Cipher.DECRYPT_MODE, keySpec, GCMParameterSpec(128, nonce)) decryptCipher.updateAAD(aad, aadOffset, aadLength) @@ -187,7 +198,7 @@ class JcaAesGcmAead( plaintextOffset: Int, plaintextLength: Int, ): ByteArray = - synchronized(this) { + synchronized(encryptLock) { val reuse = recentEncryptNonces.any { it.contentEquals(nonce) } if (reuse) { val fresh = Cipher.getInstance("AES/GCM/NoPadding") @@ -230,7 +241,7 @@ class JcaAesGcmAead( output: ByteArray, outputOffset: Int, ): Int = - synchronized(this) { + synchronized(encryptLock) { val reuse = recentEncryptNonces.any { it.contentEquals(nonce) } val cipher: Cipher if (reuse) { diff --git a/quic/src/jvmAndroid/kotlin/com/vitorpamplona/quic/crypto/JcaChaCha20Poly1305Aead.kt b/quic/src/jvmAndroid/kotlin/com/vitorpamplona/quic/crypto/JcaChaCha20Poly1305Aead.kt index e66e606957..52d8b4a889 100644 --- a/quic/src/jvmAndroid/kotlin/com/vitorpamplona/quic/crypto/JcaChaCha20Poly1305Aead.kt +++ b/quic/src/jvmAndroid/kotlin/com/vitorpamplona/quic/crypto/JcaChaCha20Poly1305Aead.kt @@ -35,10 +35,14 @@ import javax.crypto.spec.SecretKeySpec * allocations the pure-Kotlin [ChaCha20Poly1305Aead] requires. * * Single-thread per direction (one PacketProtection per side, one - * direction per side). Synchronization on a private monitor is - * defence-in-depth — a future caller (test harness, key-update path) - * sharing the instance across coroutines would otherwise corrupt - * the cached `Cipher` state silently. + * direction per side). Synchronization is defence-in-depth — a future + * caller (test harness, key-update path) sharing the instance across + * coroutines would otherwise corrupt the cached `Cipher` state silently. + * + * Two independent monitors — [encryptLock] guards the encrypt-side state + * ([encryptCipher] + [recentEncryptNonces]); [decryptLock] guards + * [decryptCipher]. Disjoint state, so seal and open don't serialize + * against each other per packet. */ class JcaChaCha20Poly1305Aead( key: ByteArray, @@ -55,6 +59,11 @@ class JcaChaCha20Poly1305Aead( private val encryptCipher: Cipher = Cipher.getInstance("ChaCha20-Poly1305") private val decryptCipher: Cipher = Cipher.getInstance("ChaCha20-Poly1305") + // Disjoint monitors so encrypt and decrypt don't serialize against each + // other. seal-family holds encryptLock; open-family holds decryptLock. + private val encryptLock = Any() + private val decryptLock = Any() + /** * Last-N nonces successfully consumed by [seal] / [sealRange] / * [sealInto]. JCA's ChaCha20-Poly1305 (like AES-GCM) refuses to @@ -72,7 +81,7 @@ class JcaChaCha20Poly1305Aead( aad: ByteArray, plaintext: ByteArray, ): ByteArray = - synchronized(this) { + synchronized(encryptLock) { sealCommon(nonce, aad, 0, aad.size, plaintext, 0, plaintext.size, output = null, outputOffset = 0) .first } @@ -87,7 +96,7 @@ class JcaChaCha20Poly1305Aead( plaintextOffset: Int, plaintextLength: Int, ): ByteArray = - synchronized(this) { + synchronized(encryptLock) { sealCommon(nonce, aad, aadOffset, aadLength, plaintext, plaintextOffset, plaintextLength, output = null, outputOffset = 0) .first } @@ -104,7 +113,7 @@ class JcaChaCha20Poly1305Aead( output: ByteArray, outputOffset: Int, ): Int = - synchronized(this) { + synchronized(encryptLock) { sealCommon(nonce, aad, aadOffset, aadLength, plaintext, plaintextOffset, plaintextLength, output, outputOffset) .second } @@ -177,7 +186,7 @@ class JcaChaCha20Poly1305Aead( aad: ByteArray, ciphertext: ByteArray, ): ByteArray? = - synchronized(this) { + synchronized(decryptLock) { try { decryptCipher.init(Cipher.DECRYPT_MODE, keySpec, IvParameterSpec(nonce)) decryptCipher.updateAAD(aad) @@ -197,7 +206,7 @@ class JcaChaCha20Poly1305Aead( ciphertextOffset: Int, ciphertextLength: Int, ): ByteArray? = - synchronized(this) { + synchronized(decryptLock) { try { decryptCipher.init(Cipher.DECRYPT_MODE, keySpec, IvParameterSpec(nonce)) decryptCipher.updateAAD(aad, aadOffset, aadLength) From ca5ae978fbeebea88862788bbf7869c1f32a8727 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 2 Jul 2026 16:21:27 +0000 Subject: [PATCH 2/4] perf: add lock-free-read ConcurrentLruCache, use on two hot read paths MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Part C of the dispatchers/thread-caps audit. Both LnurlEndpointCache and DesktopCachedRichTextParser were bounded caches backed by a LinkedHashMap behind a single monitor (@Synchronized / Collections.synchronizedMap with accessOrder). An access-order map structurally mutates on get, so every read took the lock — serializing all readers on paths that are hot (kind-9735 zap-receipt validation; feed rich-text rendering). Add ConcurrentLruCache in quartz utils: storage is a ConcurrentHashMap so get is lock-free; writes + eviction run under a small write lock that is off the read path. Eviction is least-recently-put order (get does not refresh recency) — exactly what LnurlEndpointCache already did, and fine for the deterministic rich-text parse cache. Point both caches at the shared helper. Covered by a new ConcurrentLruCacheTest (round-trip, eviction order, re-put recency refresh, get-does-not-refresh, clear, and a concurrent size-bound smoke test); the existing LnurlEndpointCacheTest still passes unchanged. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01ANuUziXKRafSTBxbh4SMoq --- .../service/DesktopCachedRichTextParser.kt | 14 +- .../nip57Zaps/validate/LnurlEndpointCache.kt | 30 ++-- .../quartz/utils/cache/ConcurrentLruCache.kt | 91 ++++++++++++ .../utils/cache/ConcurrentLruCacheTest.kt | 136 ++++++++++++++++++ 4 files changed, 242 insertions(+), 29 deletions(-) create mode 100644 quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/utils/cache/ConcurrentLruCache.kt create mode 100644 quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/utils/cache/ConcurrentLruCacheTest.kt diff --git a/desktopApp/src/jvmMain/kotlin/com/vitorpamplona/amethyst/desktop/service/DesktopCachedRichTextParser.kt b/desktopApp/src/jvmMain/kotlin/com/vitorpamplona/amethyst/desktop/service/DesktopCachedRichTextParser.kt index c478f7f6ad..0d17dd3db0 100644 --- a/desktopApp/src/jvmMain/kotlin/com/vitorpamplona/amethyst/desktop/service/DesktopCachedRichTextParser.kt +++ b/desktopApp/src/jvmMain/kotlin/com/vitorpamplona/amethyst/desktop/service/DesktopCachedRichTextParser.kt @@ -23,25 +23,23 @@ package com.vitorpamplona.amethyst.desktop.service import com.vitorpamplona.amethyst.commons.model.ImmutableListOfLists import com.vitorpamplona.amethyst.commons.richtext.RichTextParser import com.vitorpamplona.amethyst.commons.richtext.RichTextViewerState +import com.vitorpamplona.quartz.utils.cache.ConcurrentLruCache object DesktopCachedRichTextParser { private const val MAX_CACHE_SIZE = 50 - private val cache = - java.util.Collections.synchronizedMap( - object : LinkedHashMap(64, 0.75f, true) { - override fun removeEldestEntry(eldest: Map.Entry) = size > MAX_CACHE_SIZE - }, - ) + // Lock-free get on the feed rich-text render path; the previous access-order + // synchronizedMap took a monitor even on reads. + private val cache = ConcurrentLruCache(MAX_CACHE_SIZE) fun parseText( content: String, tags: ImmutableListOfLists, callbackUri: String? = null, ): RichTextViewerState { - cache[content]?.let { return it } + cache.get(content)?.let { return it } val state = RichTextParser().parseText(content, tags, callbackUri) - cache[content] = state + cache.put(content, state) return state } diff --git a/quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/nip57Zaps/validate/LnurlEndpointCache.kt b/quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/nip57Zaps/validate/LnurlEndpointCache.kt index e9f2da0987..cf5ac7f093 100644 --- a/quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/nip57Zaps/validate/LnurlEndpointCache.kt +++ b/quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/nip57Zaps/validate/LnurlEndpointCache.kt @@ -20,6 +20,8 @@ */ package com.vitorpamplona.quartz.nip57Zaps.validate +import com.vitorpamplona.quartz.utils.cache.ConcurrentLruCache + /** * Process-wide cache of LNURL-pay endpoint metadata, keyed by the canonical * `/.well-known/lnurlp/` URL the recipient resolves to. @@ -37,37 +39,23 @@ package com.vitorpamplona.quartz.nip57Zaps.validate object LnurlEndpointCache { private const val MAX_ENTRIES = 1000 - // Insertion-ordered map so we can evict the oldest entry once we hit the cap. - // Synchronized externally — every mutating call holds the monitor. - private val cache: LinkedHashMap = LinkedHashMap() + // Bounded cache with a lock-free get — hot on the zap-validation read path. + // Eviction is least-recently-put (a get does not refresh recency), matching + // the previous LinkedHashMap-based behaviour where only put reordered. + private val cache = ConcurrentLruCache(MAX_ENTRIES) - @Synchronized - fun get(url: String): LnurlEndpointInfo? = cache[LnurlForm.normalizeUrl(url)] + fun get(url: String): LnurlEndpointInfo? = cache.get(LnurlForm.normalizeUrl(url)) - @Synchronized fun put( url: String, info: LnurlEndpointInfo, ) { - val key = LnurlForm.normalizeUrl(url) - // Re-insert so the entry becomes "youngest" in iteration order. - cache.remove(key) - cache[key] = info - if (cache.size > MAX_ENTRIES) { - val oldest = - cache.entries - .iterator() - .next() - .key - cache.remove(oldest) - } + cache.put(LnurlForm.normalizeUrl(url), info) } - @Synchronized fun clear() { cache.clear() } - @Synchronized - internal fun size(): Int = cache.size + internal fun size(): Int = cache.size() } diff --git a/quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/utils/cache/ConcurrentLruCache.kt b/quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/utils/cache/ConcurrentLruCache.kt new file mode 100644 index 0000000000..94896edde8 --- /dev/null +++ b/quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/utils/cache/ConcurrentLruCache.kt @@ -0,0 +1,91 @@ +/* + * 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.utils.cache + +import java.util.concurrent.ConcurrentHashMap + +/** + * A bounded, thread-safe cache with a **lock-free [get]**. + * + * The common alternative — a `Collections.synchronizedMap` wrapping an + * access-order `LinkedHashMap`, or an `@Synchronized`-guarded `LinkedHashMap` — + * takes a single monitor on *every* operation, including reads (an access-order + * map structurally mutates on `get`, so it cannot be read without the lock). + * On hot read paths (zap-receipt validation, feed rich-text rendering) that one + * monitor serializes every reader across all dispatcher threads. + * + * Here storage is a [ConcurrentHashMap], so [get] never takes a lock. [put] and + * [clear] hold a small monitor that keeps the map and the recency order + * consistent; that lock is off the read path entirely. Eviction is + * **least-recently-put** order (a `get` does not refresh recency); re-putting a + * key moves it to the youngest position. This matches the semantics the previous + * monitor-based caches relied on and keeps the read path contention-free. + * + * Because writes and eviction happen atomically under the monitor while readers + * see the [ConcurrentHashMap] directly, an external [size] can transiently + * observe at most `maxSize + 1` (the instant a new entry is inserted before the + * over-cap entry is evicted, within a single locked section). It settles to + * `<= maxSize` once writers quiesce. + * + * Keys and values must be non-null (a [ConcurrentHashMap] constraint). + */ +class ConcurrentLruCache( + private val maxSize: Int, +) { + init { + require(maxSize > 0) { "maxSize must be > 0, was $maxSize" } + } + + private val map = ConcurrentHashMap() + + // Guards writes + eviction so the map and the recency order stay consistent. + // Reads never touch it. Writes are the cold path here, so serializing them + // is fine; the point is a lock-free [get]. + private val writeLock = Any() + private val order = ArrayDeque() + + fun get(key: K): V? = map[key] + + fun put( + key: K, + value: V, + ) { + synchronized(writeLock) { + // Re-inserting an existing key makes it the youngest again. + val existed = map.put(key, value) != null + if (existed) order.remove(key) + order.addLast(key) + while (order.size > maxSize) { + val oldest = order.removeFirst() + map.remove(oldest) + } + } + } + + fun clear() { + synchronized(writeLock) { + order.clear() + map.clear() + } + } + + fun size(): Int = map.size +} diff --git a/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/utils/cache/ConcurrentLruCacheTest.kt b/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/utils/cache/ConcurrentLruCacheTest.kt new file mode 100644 index 0000000000..23a0ee7a66 --- /dev/null +++ b/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/utils/cache/ConcurrentLruCacheTest.kt @@ -0,0 +1,136 @@ +/* + * 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.utils.cache + +import java.util.concurrent.CountDownLatch +import java.util.concurrent.Executors +import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicInteger +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertNull +import kotlin.test.assertTrue + +class ConcurrentLruCacheTest { + @Test + fun `get returns put value`() { + val cache = ConcurrentLruCache(4) + cache.put("a", 1) + assertEquals(1, cache.get("a")) + assertNull(cache.get("missing")) + } + + @Test + fun `re-put overwrites value`() { + val cache = ConcurrentLruCache(4) + cache.put("a", 1) + cache.put("a", 2) + assertEquals(2, cache.get("a")) + assertEquals(1, cache.size()) + } + + @Test + fun `evicts least-recently-put when over capacity`() { + val cache = ConcurrentLruCache(3) + cache.put("a", 1) + cache.put("b", 2) + cache.put("c", 3) + cache.put("d", 4) // pushes out "a" + + assertNull(cache.get("a")) + assertEquals(2, cache.get("b")) + assertEquals(3, cache.get("c")) + assertEquals(4, cache.get("d")) + assertEquals(3, cache.size()) + } + + @Test + fun `re-putting a key refreshes its recency`() { + val cache = ConcurrentLruCache(3) + cache.put("a", 1) + cache.put("b", 2) + cache.put("c", 3) + // Touch "a" via put so it is no longer the oldest. + cache.put("a", 11) + cache.put("d", 4) // oldest is now "b", not "a" + + assertNull(cache.get("b")) + assertEquals(11, cache.get("a")) + assertEquals(3, cache.get("c")) + assertEquals(4, cache.get("d")) + } + + @Test + fun `get does not refresh recency`() { + val cache = ConcurrentLruCache(3) + cache.put("a", 1) + cache.put("b", 2) + cache.put("c", 3) + // A read must NOT save "a" from eviction (least-recently-put semantics). + assertEquals(1, cache.get("a")) + cache.put("d", 4) + assertNull(cache.get("a")) + } + + @Test + fun `clear empties the cache`() { + val cache = ConcurrentLruCache(4) + cache.put("a", 1) + cache.put("b", 2) + cache.clear() + assertEquals(0, cache.size()) + assertNull(cache.get("a")) + } + + @Test + fun `size never exceeds capacity under concurrent puts`() { + val cap = 100 + val cache = ConcurrentLruCache(cap) + val threads = 8 + val perThread = 5_000 + val pool = Executors.newFixedThreadPool(threads) + val start = CountDownLatch(1) + val done = CountDownLatch(threads) + // Writes+eviction are atomic under one lock, so an external reader can see + // at most one over-cap entry (new inserted before old evicted). + val overBound = AtomicInteger(0) + + repeat(threads) { t -> + pool.execute { + start.await() + for (i in 0 until perThread) { + cache.put(t * perThread + i, i) + // Interleave reads; they must never throw or take a lock. + cache.get((t * perThread + i) - 1) + if (cache.size() > cap + 1) overBound.incrementAndGet() + } + done.countDown() + } + } + start.countDown() + assertTrue(done.await(30, TimeUnit.SECONDS), "workers did not finish in time") + pool.shutdown() + + assertEquals(0, overBound.get(), "cache size exceeded cap+1 during concurrent puts") + // Once writers quiesce it must settle to <= cap. + assertTrue(cache.size() <= cap, "final size ${cache.size()} exceeds cap $cap") + } +} From 49e0fee162a3ca7270d3689304f38619867c991f Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 2 Jul 2026 16:53:43 +0000 Subject: [PATCH 3/4] perf: add expect/actual ConcurrentSet, use for EventDeduplicator MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Part B of the dispatchers/thread-caps audit. EventDeduplicator is fed from relay subscription callbacks (AdvancedSearchBarState.trackRelayEvent), which arrive on multiple threads concurrently. It backed a plain mutableSet with a single KmpLock, so every delivery from every relay thread serialized on one monitor. Add ConcurrentSet as a KMP expect/actual util: - jvmAndroid actual: ConcurrentHashMap.newKeySet() — lock-striped writes, lock-free reads, no single cross-thread monitor. - iOS actual: a KmpLock-guarded set (no lock-free set in the K/N stdlib) — same behaviour as before, no regression. The win lands on JVM/Android, which is where the high-throughput event paths run. Point EventDeduplicator at it. Covered by ConcurrentSetTest (commonTest, behaviour) and ConcurrentSetConcurrencyTest (jvmTest, exactly-one-add-per-key under 8 threads). Scope note: the other two commonMain sites the audit flagged were left as-is on purpose. The compose subscription managers' KmpLock is a deliberate, documented KMP choice on a single (main-thread) writer where the lock is effectively free; EOSECache is a bounded LRU with compound value mutation on a per-subscription (not per-event) path. Neither is a clean fit for a concurrent set, and converting them would fight a documented decision for negligible gain. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01ANuUziXKRafSTBxbh4SMoq --- .../commons/search/EventDeduplicator.kt | 21 ++++--- .../amethyst/commons/util/ConcurrentSet.kt | 53 ++++++++++++++++ .../commons/util/ConcurrentSetTest.kt | 63 +++++++++++++++++++ .../commons/util/ConcurrentSet.ios.kt | 39 ++++++++++++ .../commons/util/ConcurrentSet.jvmAndroid.kt | 38 +++++++++++ .../util/ConcurrentSetConcurrencyTest.kt | 63 +++++++++++++++++++ 6 files changed, 269 insertions(+), 8 deletions(-) create mode 100644 commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/util/ConcurrentSet.kt create mode 100644 commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/util/ConcurrentSetTest.kt create mode 100644 commons/src/iosMain/kotlin/com/vitorpamplona/amethyst/commons/util/ConcurrentSet.ios.kt create mode 100644 commons/src/jvmAndroid/kotlin/com/vitorpamplona/amethyst/commons/util/ConcurrentSet.jvmAndroid.kt create mode 100644 commons/src/jvmTest/kotlin/com/vitorpamplona/amethyst/commons/util/ConcurrentSetConcurrencyTest.kt diff --git a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/search/EventDeduplicator.kt b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/search/EventDeduplicator.kt index 24cc954eb5..3c92e92f93 100644 --- a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/search/EventDeduplicator.kt +++ b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/search/EventDeduplicator.kt @@ -20,18 +20,23 @@ */ package com.vitorpamplona.amethyst.commons.search -import com.vitorpamplona.amethyst.commons.util.KmpLock -import com.vitorpamplona.amethyst.commons.util.withLock +import com.vitorpamplona.amethyst.commons.util.ConcurrentSet +/** + * Tracks event ids already seen so duplicate deliveries can be dropped. + * + * Fed from relay subscription callbacks, which can arrive on multiple threads + * concurrently, so this uses a [ConcurrentSet] (lock-striped on JVM/Android) + * rather than a single lock that would serialize every delivery. + */ class EventDeduplicator { - private val lock = KmpLock() - private val seenIds = mutableSetOf() + private val seenIds = ConcurrentSet() - fun tryAdd(id: String): Boolean = lock.withLock { seenIds.add(id) } + fun tryAdd(id: String): Boolean = seenIds.add(id) - fun contains(id: String): Boolean = lock.withLock { id in seenIds } + fun contains(id: String): Boolean = seenIds.contains(id) - fun clear() = lock.withLock { seenIds.clear() } + fun clear() = seenIds.clear() - val size: Int get() = lock.withLock { seenIds.size } + val size: Int get() = seenIds.size } diff --git a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/util/ConcurrentSet.kt b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/util/ConcurrentSet.kt new file mode 100644 index 0000000000..519a5e1aa3 --- /dev/null +++ b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/util/ConcurrentSet.kt @@ -0,0 +1,53 @@ +/* + * 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.amethyst.commons.util + +/** + * KMP-friendly thread-safe set of non-null elements. + * + * On JVM/Android this is backed by `ConcurrentHashMap.newKeySet()`, so + * concurrent [add] / [contains] / [remove] from many threads do NOT all + * serialize on one monitor — writes are lock-striped and reads are lock-free. + * That is the difference from wrapping a plain set in a single [KmpLock], where + * every operation from every thread contends the same lock. + * + * iOS has no lock-free set in the standard library, so its actual guards a plain + * set with a reentrant lock — same behaviour as before, no regression. The + * win is on the JVM/Android app, which is where the high-throughput event + * paths run. + * + * Use this only when the access pattern is genuinely multi-threaded and hot + * (e.g. per-event dedup fed by multiple relay callback threads). A set touched + * only from one thread does not need it. + */ +expect class ConcurrentSet() { + /** Adds [element]; returns true if it was not already present. */ + fun add(element: E): Boolean + + fun contains(element: E): Boolean + + /** Removes [element]; returns true if it was present. */ + fun remove(element: E): Boolean + + fun clear() + + val size: Int +} diff --git a/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/util/ConcurrentSetTest.kt b/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/util/ConcurrentSetTest.kt new file mode 100644 index 0000000000..f6e2543476 --- /dev/null +++ b/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/util/ConcurrentSetTest.kt @@ -0,0 +1,63 @@ +/* + * 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.amethyst.commons.util + +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertFalse +import kotlin.test.assertTrue + +class ConcurrentSetTest { + @Test + fun `add returns true only the first time`() { + val set = ConcurrentSet() + assertTrue(set.add("a")) + assertFalse(set.add("a")) + assertEquals(1, set.size) + } + + @Test + fun `contains reflects membership`() { + val set = ConcurrentSet() + assertFalse(set.contains("a")) + set.add("a") + assertTrue(set.contains("a")) + } + + @Test + fun `remove returns true only when present`() { + val set = ConcurrentSet() + set.add("a") + assertTrue(set.remove("a")) + assertFalse(set.remove("a")) + assertFalse(set.contains("a")) + } + + @Test + fun `clear empties the set`() { + val set = ConcurrentSet() + set.add("a") + set.add("b") + set.clear() + assertEquals(0, set.size) + assertFalse(set.contains("a")) + } +} diff --git a/commons/src/iosMain/kotlin/com/vitorpamplona/amethyst/commons/util/ConcurrentSet.ios.kt b/commons/src/iosMain/kotlin/com/vitorpamplona/amethyst/commons/util/ConcurrentSet.ios.kt new file mode 100644 index 0000000000..5fc7921153 --- /dev/null +++ b/commons/src/iosMain/kotlin/com/vitorpamplona/amethyst/commons/util/ConcurrentSet.ios.kt @@ -0,0 +1,39 @@ +/* + * 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.amethyst.commons.util + +actual class ConcurrentSet { + // No lock-free set in the K/N standard library — guard a plain set with the + // same reentrant lock KmpLock uses. Same behaviour as the previous + // KmpLock-wrapped sets; the lock-striped win only exists on JVM/Android. + private val lock = KmpLock() + private val set = HashSet() + + actual fun add(element: E): Boolean = lock.withLock { set.add(element) } + + actual fun contains(element: E): Boolean = lock.withLock { set.contains(element) } + + actual fun remove(element: E): Boolean = lock.withLock { set.remove(element) } + + actual fun clear() = lock.withLock { set.clear() } + + actual val size: Int get() = lock.withLock { set.size } +} diff --git a/commons/src/jvmAndroid/kotlin/com/vitorpamplona/amethyst/commons/util/ConcurrentSet.jvmAndroid.kt b/commons/src/jvmAndroid/kotlin/com/vitorpamplona/amethyst/commons/util/ConcurrentSet.jvmAndroid.kt new file mode 100644 index 0000000000..e91b7196f9 --- /dev/null +++ b/commons/src/jvmAndroid/kotlin/com/vitorpamplona/amethyst/commons/util/ConcurrentSet.jvmAndroid.kt @@ -0,0 +1,38 @@ +/* + * 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.amethyst.commons.util + +import java.util.concurrent.ConcurrentHashMap + +actual class ConcurrentSet { + // Lock-striped writes, lock-free reads — no single monitor across threads. + private val set: MutableSet = ConcurrentHashMap.newKeySet() + + actual fun add(element: E): Boolean = set.add(element) + + actual fun contains(element: E): Boolean = set.contains(element) + + actual fun remove(element: E): Boolean = set.remove(element) + + actual fun clear() = set.clear() + + actual val size: Int get() = set.size +} diff --git a/commons/src/jvmTest/kotlin/com/vitorpamplona/amethyst/commons/util/ConcurrentSetConcurrencyTest.kt b/commons/src/jvmTest/kotlin/com/vitorpamplona/amethyst/commons/util/ConcurrentSetConcurrencyTest.kt new file mode 100644 index 0000000000..d54d47b470 --- /dev/null +++ b/commons/src/jvmTest/kotlin/com/vitorpamplona/amethyst/commons/util/ConcurrentSetConcurrencyTest.kt @@ -0,0 +1,63 @@ +/* + * 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.amethyst.commons.util + +import java.util.concurrent.CountDownLatch +import java.util.concurrent.Executors +import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicInteger +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertTrue + +class ConcurrentSetConcurrencyTest { + @Test + fun `exactly one add per key wins across threads`() { + val set = ConcurrentSet() + val keys = 10_000 + val threads = 8 + val pool = Executors.newFixedThreadPool(threads) + val start = CountDownLatch(1) + val done = CountDownLatch(threads) + // Every thread races to add the SAME keys; exactly one add per key must + // return true. With a single lock this would still hold — the point is + // that it also holds under the lock-striped ConcurrentHashMap backing. + val wonAdds = AtomicInteger(0) + + repeat(threads) { + pool.execute { + start.await() + var local = 0 + for (k in 0 until keys) { + if (set.add(k)) local++ + } + wonAdds.addAndGet(local) + done.countDown() + } + } + start.countDown() + assertTrue(done.await(30, TimeUnit.SECONDS), "workers did not finish in time") + pool.shutdown() + + assertEquals(keys, wonAdds.get(), "each key must be added exactly once") + assertEquals(keys, set.size) + } +} From c549bd22df5eca21acbb5552307932fc5ef2d232 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 2 Jul 2026 21:35:57 +0000 Subject: [PATCH 4/4] perf: isolate QUIC blocking socket I/O onto dedicated threads MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Part A of the dispatchers/thread-caps audit — the one genuine at-scale starvation the audit found. UdpSocket.receive() does a blocking DatagramChannel recvfrom that parks its thread for the ENTIRE life of the connection. It ran via withContext(Dispatchers.IO) from a read loop already on Dispatchers.IO, so the blocking call pinned one shared IO-pool thread per connection. Past ~64 concurrent connections that starves ALL other Dispatchers.IO work in the process — this module's and the host app's alike. Give each socket two dedicated daemon threads: recvDispatcher for the perpetually-parked receive and sendDispatcher for the send (they can't share one thread — the receive would monopolise it). QUIC's blocking socket I/O now never touches the shared pool. connect() keeps its one-shot DNS/bind on Dispatchers.IO (setup cost, not a lifetime parker). close() calls shutdownNow() on both executors: interrupting the recv worker breaks the parked recvfrom immediately (ClosedByInterruptException, caught as ClosedChannelException -> receive() returns null), so the threads exit promptly instead of leaking per closed connection. The closed-check is hoisted out of withContext so a post-close call fails fast without dispatching onto a shut-down executor. Verified: QuicConnectionDriverLifecycleTest (100 session open/close cycles, asserts thread growth <=16 and no FD leak) passes, confirming the two new threads per socket are reclaimed on teardown. New UdpSocketTest covers round-trip, the dedicated-thread isolation, thread shutdown on close, and the after-close contract. Full :quic:jvmTest suite green. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01ANuUziXKRafSTBxbh4SMoq --- .../vitorpamplona/quic/transport/UdpSocket.kt | 63 ++++++-- .../quic/transport/UdpSocketTest.kt | 146 ++++++++++++++++++ 2 files changed, 199 insertions(+), 10 deletions(-) create mode 100644 quic/src/jvmTest/kotlin/com/vitorpamplona/quic/transport/UdpSocketTest.kt diff --git a/quic/src/jvmAndroid/kotlin/com/vitorpamplona/quic/transport/UdpSocket.kt b/quic/src/jvmAndroid/kotlin/com/vitorpamplona/quic/transport/UdpSocket.kt index 72a1bd4568..6a68360b5e 100644 --- a/quic/src/jvmAndroid/kotlin/com/vitorpamplona/quic/transport/UdpSocket.kt +++ b/quic/src/jvmAndroid/kotlin/com/vitorpamplona/quic/transport/UdpSocket.kt @@ -21,6 +21,8 @@ package com.vitorpamplona.quic.transport import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.ExecutorCoroutineDispatcher +import kotlinx.coroutines.asCoroutineDispatcher import kotlinx.coroutines.withContext import java.net.InetAddress import java.net.InetSocketAddress @@ -28,17 +30,31 @@ import java.net.StandardSocketOptions import java.nio.ByteBuffer import java.nio.channels.ClosedChannelException import java.nio.channels.DatagramChannel +import java.util.concurrent.Executors import java.util.concurrent.atomic.AtomicBoolean import java.util.concurrent.atomic.AtomicLong /** - * JVM/Android UDP socket using blocking [DatagramChannel] dispatched onto - * [Dispatchers.IO]. We don't use NIO selectors because each QUIC connection - * has exactly one socket and one receive loop — Selector doesn't pay for - * itself at this scale. + * JVM/Android UDP socket using a blocking [DatagramChannel]. We don't use NIO + * selectors because each QUIC connection has exactly one socket and one receive + * loop — a Selector doesn't pay for itself at this scale. * - * The receive buffer is sized to 64 KiB (max IPv4/IPv6 datagram); QUIC packets - * cap at MTU (~1500 in practice). + * Threading: the blocking `recvfrom` parks its thread for the *entire* life of + * the connection (it only returns when a datagram arrives or the socket + * closes). If that ran on the shared [Dispatchers.IO] pool it would pin one + * pool thread per connection, and past ~64 concurrent connections it would + * starve *all* other `Dispatchers.IO` work in the process — this module's and + * the host app's alike. So each socket owns two dedicated daemon threads: + * [recvDispatcher] for the perpetually-blocked receive, and [sendDispatcher] + * for the (rarely-blocking, but still-blocking) send. The receive can't share a + * thread with send — it would monopolise it — hence two. Both are shut down in + * [close]. QUIC's blocking socket I/O therefore never touches the shared pool. + * + * [connect] still resolves DNS + binds on [Dispatchers.IO]: that's a one-shot + * setup cost, not a lifetime parker, so it doesn't need isolation. + * + * The receive buffer is sized to typical Ethernet MTU; QUIC packets cap at MTU + * (~1500 in practice). */ actual class UdpSocket private constructor( private val channel: DatagramChannel, @@ -46,6 +62,19 @@ actual class UdpSocket private constructor( ) { private val closed = AtomicBoolean(false) + // Dedicated single-thread executors so the blocking socket calls never + // occupy the shared Dispatchers.IO pool. Daemon threads so a leaked socket + // can't keep the JVM alive. Separate recv/send threads because the receive + // parks continuously and would otherwise block sends behind it. We keep the + // ExecutorService handles (not just the dispatchers) so close() can call + // shutdownNow() — an interrupt that breaks the parked recvfrom immediately + // (ClosedByInterruptException) rather than the graceful shutdown() that + // dispatcher.close() would do. + private val recvExecutor = Executors.newSingleThreadExecutor { r -> Thread(r, "quic-udp-recv").apply { isDaemon = true } } + private val sendExecutor = Executors.newSingleThreadExecutor { r -> Thread(r, "quic-udp-send").apply { isDaemon = true } } + private val recvDispatcher: ExecutorCoroutineDispatcher = recvExecutor.asCoroutineDispatcher() + private val sendDispatcher: ExecutorCoroutineDispatcher = sendExecutor.asCoroutineDispatcher() + // Sized to typical Ethernet MTU + a bit; QUIC tops out at ~1500 in practice // and any larger inbound frame is dropped as malformed anyway. The previous // 64 KiB buffer was wasteful per connection. @@ -76,15 +105,19 @@ actual class UdpSocket private constructor( actual val receiveBufferSizeBytes: Int get() = channel.getOption(StandardSocketOptions.SO_RCVBUF) - actual suspend fun send(payload: ByteArray): Int = - withContext(Dispatchers.IO) { + actual suspend fun send(payload: ByteArray): Int { + // Fail fast without dispatching onto a possibly shut-down executor. + if (closed.get()) throw ClosedChannelException() + return withContext(sendDispatcher) { if (closed.get()) throw ClosedChannelException() val buf = ByteBuffer.wrap(payload) channel.send(buf, remote) } + } - actual suspend fun receive(): ByteArray? = - withContext(Dispatchers.IO) { + actual suspend fun receive(): ByteArray? { + if (closed.get()) return null + return withContext(recvDispatcher) { if (closed.get()) return@withContext null try { // No synchronized — only the read loop touches readBuf, by @@ -101,6 +134,7 @@ actual class UdpSocket private constructor( null } } + } actual fun close() { if (closed.compareAndSet(false, true)) { @@ -109,6 +143,15 @@ actual class UdpSocket private constructor( } catch (_: Throwable) { // already closed } + // shutdownNow() interrupts the dedicated workers: a thread parked in + // a blocking recvfrom throws ClosedByInterruptException (a + // ClosedChannelException, caught below), so receive() returns null + // and both threads exit promptly instead of leaking per closed + // connection. channel.close() above would also unblock it + // (AsynchronousCloseException), but the interrupt is immediate and + // guarantees the executor terminates. + recvExecutor.shutdownNow() + sendExecutor.shutdownNow() } } diff --git a/quic/src/jvmTest/kotlin/com/vitorpamplona/quic/transport/UdpSocketTest.kt b/quic/src/jvmTest/kotlin/com/vitorpamplona/quic/transport/UdpSocketTest.kt new file mode 100644 index 0000000000..64acd1d0b2 --- /dev/null +++ b/quic/src/jvmTest/kotlin/com/vitorpamplona/quic/transport/UdpSocketTest.kt @@ -0,0 +1,146 @@ +/* + * 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.quic.transport + +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.async +import kotlinx.coroutines.delay +import kotlinx.coroutines.launch +import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.withTimeoutOrNull +import java.net.DatagramPacket +import java.net.DatagramSocket +import java.net.InetSocketAddress +import java.nio.channels.ClosedChannelException +import kotlin.test.AfterTest +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertFailsWith +import kotlin.test.assertNull +import kotlin.test.assertTrue + +class UdpSocketTest { + // A plain UDP peer on loopback that echoes one datagram back. + private val peer = DatagramSocket(InetSocketAddress("127.0.0.1", 0)) + + @AfterTest + fun tearDown() { + runCatching { peer.close() } + } + + private fun threadNames(): List = Thread.getAllStackTraces().keys.map { it.name } + + private fun hasThreadPrefixed(prefix: String) = threadNames().any { it.startsWith(prefix) } + + @Test + fun `round trips a datagram over loopback`() { + runBlocking { + val socket = UdpSocket.connect("127.0.0.1", peer.localPort) + try { + // Peer thread: receive one packet and echo it back to the sender. + val echo = + Thread { + val buf = ByteArray(2048) + val incoming = DatagramPacket(buf, buf.size) + peer.receive(incoming) + peer.send(DatagramPacket(incoming.data, incoming.length, incoming.socketAddress)) + }.apply { + isDaemon = true + start() + } + + socket.send(byteArrayOf(1, 2, 3, 4)) + val reply = withTimeoutOrNull(3_000) { socket.receive() } + echo.join(1_000) + + assertTrue(reply != null && reply.contentEquals(byteArrayOf(1, 2, 3, 4)), "should echo the datagram back") + assertEquals(1, socket.receivedDatagramCount) + } finally { + socket.close() + } + } + } + + @Test + fun `blocking receive runs on a dedicated thread, not Dispatchers-IO`() { + runBlocking { + val socket = UdpSocket.connect("127.0.0.1", peer.localPort) + try { + // Park a receive with no incoming datagram — it blocks in recvfrom + // on the socket's dedicated recv thread, not a Dispatchers.IO worker. + val pending = async(Dispatchers.IO) { socket.receive() } + // Give the receive time to reach the blocking call on its own thread. + delay(200) + + assertTrue(hasThreadPrefixed("quic-udp-recv"), "a dedicated recv thread must carry the blocking receive") + + // Closing unblocks the parked receive (returns null), proving the + // blocking call was on the dedicated thread and is released on close. + socket.close() + val result = withTimeoutOrNull(2_000) { pending.await() } + assertNull(result, "receive() must return null once the socket closes") + } finally { + socket.close() + } + } + } + + @Test + fun `close shuts down the dedicated threads`() { + runBlocking { + val socket = UdpSocket.connect("127.0.0.1", peer.localPort) + try { + // The single-thread executors spawn their thread lazily on first + // use, so touch BOTH directions to bring both threads up. + socket.send(byteArrayOf(0)) + val recv = launch(Dispatchers.IO) { socket.receive() } + delay(200) + assertTrue( + hasThreadPrefixed("quic-udp-recv") && hasThreadPrefixed("quic-udp-send"), + "both dedicated threads must be up while open", + ) + + socket.close() + recv.join() + + // The executors shut down on close; their threads must exit promptly. + val gone = + withTimeoutOrNull(2_000) { + while (hasThreadPrefixed("quic-udp-recv") || hasThreadPrefixed("quic-udp-send")) delay(25) + true + } + assertTrue(gone == true, "dedicated recv/send threads must be gone after close() — else they leak per connection") + } finally { + socket.close() + } + } + } + + @Test + fun `after close receive returns null and send throws`() { + runBlocking { + val socket = UdpSocket.connect("127.0.0.1", peer.localPort) + socket.close() + assertNull(socket.receive(), "receive() returns null after close") + assertFailsWith { socket.send(byteArrayOf(9)) } + } + } +}