From 5c2f93f82f5a0f2eb4a0868a7cee763a1b8d6852 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 24 May 2026 21:10:35 +0000 Subject: [PATCH] refactor: replace stately with LargeCache + KmpLock MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address review feedback: the project already has LargeCache (in quartz, with jvmAndroid/appleMain/linuxMain actuals) as its KMP concurrent-map abstraction — it's used pervasively in the model layer. Adding stately duplicated that capability with an external dep. - Comparable-key maps switch to LargeCache: * ChessEventCollector.moves (String key) * ChessEventCollectorManager.collectors (String key) * ChessRelayFetchHelper.events (String key) * ChessRelayFetchHelper.relayEventCounts: LargeCache with getOrCreate { AtomicInt(0) }.addAndFetch(1) — replaces the stately .block { compute } increment idiom. getOrCreate is atomic via ConcurrentSkipListMap.putIfAbsent so all threads end up incrementing the same AtomicInt instance. * ChessLobbyLogic.recentlyLoadedGames (String key) - The SubscriptionManager pair (MutableComposeSubscriptionManager, ComposeSubscriptionManager) keeps a plain mutableMapOf — T : MutableQueryState is generic and not Comparable, so LargeCache's ConcurrentSkipListMap backing would ClassCastException at put time. Concurrency comes from a KmpLock-guarded map. - Set-shaped uses switch to KmpLock + mutableSetOf: * ChessEventCollector.processedEventIds * ChessRelayFetchHelper.eoseReceived * ChessLobbyLogic.dismissedGameIds + seenEventIds (the bounded LRU keeps insertion-order eviction; mutableSetOf returns LinkedHashSet on every KMP target). - UserRelaysCache.flow's lock: stately Lock -> KmpLock. Adds expect class KmpLock() with jvmAndroid actual that wraps ReentrantLock. iOS actual (NSLock) will land with the iOS target. Mirrors the WeakReference pattern from the previous PR. Drops stately-concurrent-collections 2.1.0 from libs.versions.toml and commons/build.gradle.kts (no remaining consumers). --- commons/build.gradle.kts | 7 -- .../commons/chess/ChessEventCollector.kt | 66 +++++++++++-------- .../amethyst/commons/chess/ChessLobbyLogic.kt | 63 ++++++++++-------- .../commons/chess/ChessRelayFetchHelper.kt | 50 ++++++++------ .../model/nip01Core/UserRelaysCache.kt | 6 +- .../ComposeSubscriptionManager.kt | 22 ++++--- .../MutableComposeSubscriptionManager.kt | 35 ++++++---- .../amethyst/commons/util/KmpLock.kt | 44 +++++++++++++ .../commons/util/KmpLock.jvmAndroid.kt | 31 +++++++++ gradle/libs.versions.toml | 2 - 10 files changed, 219 insertions(+), 107 deletions(-) create mode 100644 commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/util/KmpLock.kt create mode 100644 commons/src/jvmAndroid/kotlin/com/vitorpamplona/amethyst/commons/util/KmpLock.jvmAndroid.kt diff --git a/commons/build.gradle.kts b/commons/build.gradle.kts index acfd45e211..684213eea6 100644 --- a/commons/build.gradle.kts +++ b/commons/build.gradle.kts @@ -72,13 +72,6 @@ kotlin { // LruCache (KMP-ready) implementation(libs.androidx.collection) - // KMP-friendly concurrent map/set. Replaces ConcurrentHashMap - // for code that genuinely needs cross-thread atomic put/remove - // (chess event collector, relay subscription managers). On JVM - // it wraps mutableMapOf with synchronized; on Native it uses - // platform mutex. No new gradle plugin required. - implementation(libs.stately.concurrent.collections) - // Immutable collections api(libs.kotlinx.collections.immutable) diff --git a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/chess/ChessEventCollector.kt b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/chess/ChessEventCollector.kt index 2f5db8dffb..da643df359 100644 --- a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/chess/ChessEventCollector.kt +++ b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/chess/ChessEventCollector.kt @@ -20,14 +20,15 @@ */ package com.vitorpamplona.amethyst.commons.chess -import co.touchlab.stately.collections.ConcurrentMutableMap -import co.touchlab.stately.collections.ConcurrentMutableSet +import com.vitorpamplona.amethyst.commons.util.KmpLock +import com.vitorpamplona.amethyst.commons.util.withLock import com.vitorpamplona.quartz.nip01Core.core.Event import com.vitorpamplona.quartz.nip64Chess.jester.JesterEvent import com.vitorpamplona.quartz.nip64Chess.jester.JesterGameEvents import com.vitorpamplona.quartz.nip64Chess.jester.JesterProtocol import com.vitorpamplona.quartz.nip64Chess.jester.toJesterEvent import com.vitorpamplona.quartz.utils.Log +import com.vitorpamplona.quartz.utils.cache.LargeCache import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.flow.asStateFlow @@ -68,11 +69,15 @@ class ChessEventCollector( private val _startEvent = MutableStateFlow(null) val startEvent: StateFlow = _startEvent.asStateFlow() - // Move events (deduplicated by event ID) - private val moves = ConcurrentMutableMap() + // Move events (deduplicated by event ID). String keys are Comparable, so + // LargeCache (ConcurrentSkipListMap on JVM, CacheMap on Apple) works. + private val moves = LargeCache() - // Track all processed event IDs for fast deduplication - private val processedEventIds = ConcurrentMutableSet() + // Track all processed event IDs for fast deduplication. A plain HashSet + // guarded by KmpLock — simpler than a LargeCache for set-shaped + // membership. + private val processedEventIdsLock = KmpLock() + private val processedEventIds = mutableSetOf() // Flow that emits when any event is added (for reactive updates) private val _eventCount = MutableStateFlow(0) @@ -81,7 +86,11 @@ class ChessEventCollector( /** * Check if an event has already been processed. */ - fun hasEvent(eventId: String): Boolean = processedEventIds.contains(eventId) + fun hasEvent(eventId: String): Boolean = processedEventIdsLock.withLock { processedEventIds.contains(eventId) } + + private fun markProcessed(eventId: String): Boolean = processedEventIdsLock.withLock { processedEventIds.add(eventId) } + + private fun processedEventCount(): Int = processedEventIdsLock.withLock { processedEventIds.size } /** * Add a Jester event for this game. @@ -90,7 +99,7 @@ class ChessEventCollector( * @return true if the event was added, false if already exists or invalid */ fun addEvent(event: JesterEvent): Boolean { - if (processedEventIds.contains(event.id)) { + if (hasEvent(event.id)) { Log.d("chessdebug") { "[Collector] DEDUP: event ${event.id.take(8)} already processed for game ${startEventId.take(8)}" } return false } @@ -130,12 +139,12 @@ class ChessEventCollector( * @return true if the event was added, false if already exists or invalid */ private fun addStartEvent(event: JesterEvent): Boolean { - if (processedEventIds.contains(event.id)) return false + if (hasEvent(event.id)) return false if (!event.isStartEvent()) return false if (event.id != startEventId) return false if (_startEvent.compareAndSet(null, event)) { - processedEventIds.add(event.id) + markProcessed(event.id) incrementEventCount() Log.d("chessdebug") { "[Collector] START event added: id=${event.id.take(8)}, pubkey=${event.pubKey.take(8)}, createdAt=${event.createdAt}" } return true @@ -150,14 +159,15 @@ class ChessEventCollector( * @return true if the event was added, false if already exists or invalid */ private fun addMoveEvent(event: JesterEvent): Boolean { - if (processedEventIds.contains(event.id)) return false + if (hasEvent(event.id)) return false if (!event.isMoveEvent()) return false if (event.startEventId() != startEventId) return false - if (moves.putIfAbsent(event.id, event) == null) { - processedEventIds.add(event.id) + // createIfAbsent returns true iff the entry was added (no prior entry). + if (moves.createIfAbsent(event.id) { event }) { + markProcessed(event.id) incrementEventCount() - Log.d("chessdebug") { "[Collector] MOVE event added: id=${event.id.take(8)}, pubkey=${event.pubKey.take(8)}, move=${event.move()}, historySize=${event.history().size}, fen=${event.fen()?.take(30)}, result=${event.result()}, totalMoves=${moves.size}" } + Log.d("chessdebug") { "[Collector] MOVE event added: id=${event.id.take(8)}, pubkey=${event.pubKey.take(8)}, move=${event.move()}, historySize=${event.history().size}, fen=${event.fen()?.take(30)}, result=${event.result()}, totalMoves=${moves.size()}" } return true } return false @@ -169,7 +179,7 @@ class ChessEventCollector( fun getEvents(): JesterGameEvents = JesterGameEvents( startEvent = _startEvent.value, - moves = moves.values.toList(), + moves = moves.values().toList(), ) /** @@ -180,22 +190,22 @@ class ChessEventCollector( /** * Check if the game has any moves (indicates game is active). */ - fun hasMoves(): Boolean = moves.isNotEmpty() + fun hasMoves(): Boolean = !moves.isEmpty() /** * Check if the game has ended (has a move with result). */ - fun hasEnded(): Boolean = moves.values.any { it.result() != null } + fun hasEnded(): Boolean = moves.values().any { it.result() != null } /** * Get the number of moves collected. */ - fun moveCount(): Int = moves.size + fun moveCount(): Int = moves.size() /** * Get the latest move (with longest history). */ - fun latestMove(): JesterEvent? = moves.values.maxByOrNull { it.history().size } + fun latestMove(): JesterEvent? = moves.values().maxByOrNull { it.history().size } /** * Clear all collected events. @@ -203,12 +213,12 @@ class ChessEventCollector( fun clear() { _startEvent.value = null moves.clear() - processedEventIds.clear() + processedEventIdsLock.withLock { processedEventIds.clear() } _eventCount.value = 0 } private fun incrementEventCount() { - _eventCount.value = processedEventIds.size + _eventCount.value = processedEventCount() } } @@ -219,21 +229,21 @@ class ChessEventCollector( * such as in a chess lobby or when spectating multiple games. */ class ChessEventCollectorManager { - private val collectors = ConcurrentMutableMap() + private val collectors = LargeCache() /** * Get or create a collector for a game. * * @param startEventId The start event ID (game identifier) */ - fun getOrCreate(startEventId: String): ChessEventCollector = collectors.getOrPut(startEventId) { ChessEventCollector(startEventId) } + fun getOrCreate(startEventId: String): ChessEventCollector = collectors.getOrCreate(startEventId) { ChessEventCollector(it) } /** * Get a collector if it exists. * * @param startEventId The start event ID (game identifier) */ - fun get(startEventId: String): ChessEventCollector? = collectors[startEventId] + fun get(startEventId: String): ChessEventCollector? = collectors.get(startEventId) /** * Remove a collector for a game. @@ -245,13 +255,13 @@ class ChessEventCollectorManager { /** * Get all active game IDs (start event IDs). */ - fun activeGameIds(): Set = collectors.keys.toSet() + fun activeGameIds(): Set = collectors.keys() /** * Clear all collectors. */ fun clear() { - collectors.values.forEach { it.clear() } + collectors.values().forEach { it.clear() } collectors.clear() } @@ -276,8 +286,8 @@ class ChessEventCollectorManager { return false } val collector = - collectors[startId] ?: run { - Log.d("chessdebug") { "[CollectorMgr] REJECTED: no collector for game ${startId.take(8)} (active games: ${collectors.keys.map { it.take(8) }})" } + collectors.get(startId) ?: run { + Log.d("chessdebug") { "[CollectorMgr] REJECTED: no collector for game ${startId.take(8)} (active games: ${collectors.keys().map { it.take(8) }})" } return false } return collector.addEvent(event) diff --git a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/chess/ChessLobbyLogic.kt b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/chess/ChessLobbyLogic.kt index e49849a0f4..8d8b2791ac 100644 --- a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/chess/ChessLobbyLogic.kt +++ b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/chess/ChessLobbyLogic.kt @@ -20,8 +20,8 @@ */ package com.vitorpamplona.amethyst.commons.chess -import co.touchlab.stately.collections.ConcurrentMutableMap -import co.touchlab.stately.collections.ConcurrentMutableSet +import com.vitorpamplona.amethyst.commons.util.KmpLock +import com.vitorpamplona.amethyst.commons.util.withLock import com.vitorpamplona.quartz.nip64Chess.ChessGameEnd import com.vitorpamplona.quartz.nip64Chess.ChessMoveEvent import com.vitorpamplona.quartz.nip64Chess.Color @@ -32,6 +32,7 @@ import com.vitorpamplona.quartz.nip64Chess.jester.JesterEvent import com.vitorpamplona.quartz.nip64Chess.jester.JesterGameEvents import com.vitorpamplona.quartz.utils.Log import com.vitorpamplona.quartz.utils.TimeUtils +import com.vitorpamplona.quartz.utils.cache.LargeCache import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.delay @@ -131,20 +132,21 @@ class ChessLobbyLogic( ) { val state = ChessLobbyState(userPubkey, scope) - private val dismissedGameIds = - ConcurrentMutableSet().apply { - dismissedStorage?.load(userPubkey)?.let { addAll(it) } - } + private val dismissedGameIdsLock = KmpLock() + private val dismissedGameIds: MutableSet = + (dismissedStorage?.load(userPubkey)?.toMutableSet() ?: mutableSetOf()) // Track when games were last loaded to prevent duplicate fetches - // (e.g., discoverUserGames loads a game, then polling immediately re-fetches it) - private val recentlyLoadedGames = ConcurrentMutableMap() + // (e.g., discoverUserGames loads a game, then polling immediately re-fetches it). + // String keys are Comparable, so LargeCache works. + private val recentlyLoadedGames = LargeCache() // Dedup incoming events (same event delivered by multiple relays). - // Bounded LRU: evict oldest when exceeding capacity. ConcurrentMutableSet - // wraps mutableSetOf() (LinkedHashSet on every KMP target), so iterator - // order is insertion order — required for LRU eviction below. - private val seenEventIds = ConcurrentMutableSet() + // Bounded LRU: evict oldest when exceeding capacity. mutableSetOf returns + // LinkedHashSet on every KMP target, preserving insertion-order iteration + // required for LRU eviction below. + private val seenEventIdsLock = KmpLock() + private val seenEventIds = mutableSetOf() private val seenEventIdsMax = 500 private val pollingDelegate = @@ -212,7 +214,7 @@ class ChessLobbyLogic( // Dedup: skip if we already processed this event ID (multiple relays deliver same event) val isNew = - seenEventIds.block { + seenEventIdsLock.withLock { if (!seenEventIds.add(event.id)) { false } else { @@ -448,13 +450,13 @@ class ChessLobbyLogic( */ fun handleGameAccepted(startEventId: String) { // Skip if already loaded or in-flight (multiple move events from same game trigger this) - val lastLoaded = recentlyLoadedGames[startEventId] + val lastLoaded = recentlyLoadedGames.get(startEventId) if (lastLoaded != null && (TimeUtils.now() - lastLoaded) < 10) { Log.d("chessdebug") { "[Lobby] handleGameAccepted: SKIPPED game ${startEventId.take(8)} - loaded ${TimeUtils.now() - lastLoaded}s ago" } return } // Mark immediately to prevent concurrent launches - recentlyLoadedGames[startEventId] = TimeUtils.now() + recentlyLoadedGames.put(startEventId, TimeUtils.now()) Log.d("chessdebug") { "[Lobby] handleGameAccepted: game ${startEventId.take(8)} - fetching from relays" } scope.launch(Dispatchers.Default) { @@ -466,7 +468,7 @@ class ChessLobbyLogic( when (result) { is LoadGameResult.Success -> { - recentlyLoadedGames[startEventId] = TimeUtils.now() + recentlyLoadedGames.put(startEventId, TimeUtils.now()) Log.d("chessdebug") { "[Lobby] handleGameAccepted SUCCESS: game ${startEventId.take(8)}, role=${result.reconstructedState.viewerRole}" } state.addActiveGame(startEventId, result.liveState) pollingDelegate.addGameId(startEventId) @@ -618,7 +620,7 @@ class ChessLobbyLogic( when (result) { is LoadGameResult.Success -> { - recentlyLoadedGames[startEventId] = TimeUtils.now() + recentlyLoadedGames.put(startEventId, TimeUtils.now()) state.addSpectatingGame(startEventId, result.liveState) pollingDelegate.addGameId(startEventId) state.setBroadcastStatus(ChessBroadcastStatus.Idle) @@ -651,7 +653,7 @@ class ChessLobbyLogic( when (result) { is LoadGameResult.Success -> { - recentlyLoadedGames[startEventId] = TimeUtils.now() + recentlyLoadedGames.put(startEventId, TimeUtils.now()) if (result.liveState.isSpectator) { state.addSpectatingGame(startEventId, result.liveState) } else { @@ -688,13 +690,13 @@ class ChessLobbyLogic( private suspend fun refreshGame(startEventId: String) { // Skip if this game was just loaded (prevents duplicate fetch after discoverUserGames) - val lastLoaded = recentlyLoadedGames[startEventId] + val lastLoaded = recentlyLoadedGames.get(startEventId) if (lastLoaded != null && (TimeUtils.now() - lastLoaded) < 10) { Log.d("chessdebug") { "[Lobby] refreshGame: SKIPPED game ${startEventId.take(8)} - loaded ${TimeUtils.now() - lastLoaded}s ago" } return } // Mark immediately to prevent concurrent fetches for the same game - recentlyLoadedGames[startEventId] = TimeUtils.now() + recentlyLoadedGames.put(startEventId, TimeUtils.now()) Log.d("chessdebug") { "[Lobby] refreshGame: fetching game ${startEventId.take(8)} from relays" } val events = fetcher.fetchGameEvents(startEventId) @@ -883,16 +885,17 @@ class ChessLobbyLogic( .map { it.gameId } .toSet() + val dismissedSnapshot = dismissedGameIdsLock.withLock { dismissedGameIds.toSet() } for (startEventId in newGameIds) { if (startEventId in completedGameIds) continue - if (startEventId in dismissedGameIds) continue + if (startEventId in dismissedSnapshot) continue val events = fetcher.fetchGameEvents(startEventId) val result = ChessGameLoader.loadGame(events, userPubkey) when (result) { is LoadGameResult.Success -> { - recentlyLoadedGames[startEventId] = TimeUtils.now() + recentlyLoadedGames.put(startEventId, TimeUtils.now()) // If the discovered game is already finished, send it straight to completed val gameStatus = result.liveState.gameStatus.value if (gameStatus is GameStatus.Finished) { @@ -961,15 +964,23 @@ class ChessLobbyLogic( fun dismissCompletedGame(gameId: String) { state.removeCompletedGame(gameId) - dismissedGameIds.add(gameId) - dismissedStorage?.save(userPubkey, dismissedGameIds.toSet()) + val snapshot = + dismissedGameIdsLock.withLock { + dismissedGameIds.add(gameId) + dismissedGameIds.toSet() + } + dismissedStorage?.save(userPubkey, snapshot) } fun dismissAllCompletedGames() { val allIds = state.completedGames.value.map { it.gameId } state.clearCompletedGames() - dismissedGameIds.addAll(allIds) - dismissedStorage?.save(userPubkey, dismissedGameIds.toSet()) + val snapshot = + dismissedGameIdsLock.withLock { + dismissedGameIds.addAll(allIds) + dismissedGameIds.toSet() + } + dismissedStorage?.save(userPubkey, snapshot) } /** diff --git a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/chess/ChessRelayFetchHelper.kt b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/chess/ChessRelayFetchHelper.kt index 6cfb0971e3..7a1a945712 100644 --- a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/chess/ChessRelayFetchHelper.kt +++ b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/chess/ChessRelayFetchHelper.kt @@ -20,16 +20,19 @@ */ package com.vitorpamplona.amethyst.commons.chess -import co.touchlab.stately.collections.ConcurrentMutableMap -import co.touchlab.stately.collections.ConcurrentMutableSet +import com.vitorpamplona.amethyst.commons.util.KmpLock +import com.vitorpamplona.amethyst.commons.util.withLock import com.vitorpamplona.quartz.nip01Core.core.Event import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.SubscriptionListener import com.vitorpamplona.quartz.nip01Core.relay.client.single.newSubId import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl +import com.vitorpamplona.quartz.utils.cache.LargeCache import kotlinx.coroutines.CompletableDeferred import kotlinx.coroutines.withTimeoutOrNull +import kotlin.concurrent.atomics.AtomicInt +import kotlin.concurrent.atomics.ExperimentalAtomicApi /** * Progress callback for relay fetch operations @@ -68,6 +71,7 @@ class ChessRelayFetchHelper( * @param onProgress Optional callback for progress updates per relay * @return Deduplicated list of events received before timeout/EOSE */ + @OptIn(ExperimentalAtomicApi::class) suspend fun fetchEvents( filters: Map>, timeoutMs: Long = ChessConfig.FETCH_TIMEOUT_MS, @@ -75,16 +79,22 @@ class ChessRelayFetchHelper( ): List { if (filters.isEmpty()) return emptyList() - val events = ConcurrentMutableMap() + val events = LargeCache() val relayCount = filters.keys.size - val eoseReceived = ConcurrentMutableSet() - val relayEventCounts = ConcurrentMutableMap() + // Per-relay event counters. Each counter is independently atomic, so + // increments don't need to lock the whole map. + val relayEventCounts = LargeCache() + // Small bounded set; KmpLock-guarded plain set is fine for the size we + // expect (one entry per relay in the filter map). + val eoseLock = KmpLock() + val eoseReceived = mutableSetOf() val allEose = CompletableDeferred() val subId = newSubId() - // Initialize all relays as WAITING + // Initialize all relays as WAITING. Eagerly create AtomicInt counters + // so onEose / timeout paths can read load() without a put race. filters.keys.forEach { relay -> - relayEventCounts[relay] = 0 + relayEventCounts.getOrCreate(relay) { AtomicInt(0) } onProgress?.invoke(RelayFetchProgress(relay, RelayFetchStatus.WAITING, 0)) } @@ -96,13 +106,8 @@ class ChessRelayFetchHelper( relay: NormalizedRelayUrl, forFilters: List?, ) { - events[event.id] = event - val count = - relayEventCounts.block { - val newVal = (relayEventCounts[relay] ?: 0) + 1 - relayEventCounts[relay] = newVal - newVal - } + events.put(event.id, event) + val count = relayEventCounts.getOrCreate(relay) { AtomicInt(0) }.addAndFetch(1) onProgress?.invoke(RelayFetchProgress(relay, RelayFetchStatus.RECEIVING, count)) } @@ -110,11 +115,15 @@ class ChessRelayFetchHelper( relay: NormalizedRelayUrl, forFilters: List?, ) { - eoseReceived.add(relay) - val count = relayEventCounts[relay] ?: 0 + val newEoseSize = + eoseLock.withLock { + eoseReceived.add(relay) + eoseReceived.size + } + val count = relayEventCounts.get(relay)?.load() ?: 0 onProgress?.invoke(RelayFetchProgress(relay, RelayFetchStatus.EOSE_RECEIVED, count)) // Complete when all relays respond - if (eoseReceived.size >= relayCount) { + if (newEoseSize >= relayCount) { allEose.complete(Unit) } } @@ -125,9 +134,10 @@ class ChessRelayFetchHelper( // Mark timed-out relays if (eoseResult == null) { + val eoseSnapshot = eoseLock.withLock { eoseReceived.toSet() } filters.keys.forEach { relay -> - if (relay !in eoseReceived) { - val count = relayEventCounts[relay] ?: 0 + if (relay !in eoseSnapshot) { + val count = relayEventCounts.get(relay)?.load() ?: 0 onProgress?.invoke(RelayFetchProgress(relay, RelayFetchStatus.TIMEOUT, count)) } } @@ -135,6 +145,6 @@ class ChessRelayFetchHelper( client.unsubscribe(subId) - return events.values.toList() + return events.values().toList() } } diff --git a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/model/nip01Core/UserRelaysCache.kt b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/model/nip01Core/UserRelaysCache.kt index 70007668bf..a1c5385b37 100644 --- a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/model/nip01Core/UserRelaysCache.kt +++ b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/model/nip01Core/UserRelaysCache.kt @@ -21,9 +21,9 @@ package com.vitorpamplona.amethyst.commons.model.nip01Core import androidx.compose.runtime.Stable -import co.touchlab.stately.concurrency.Lock -import co.touchlab.stately.concurrency.withLock +import com.vitorpamplona.amethyst.commons.util.KmpLock import com.vitorpamplona.amethyst.commons.util.WeakReference +import com.vitorpamplona.amethyst.commons.util.withLock import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl import com.vitorpamplona.quartz.nip01Core.relay.normalizer.isLocalHost import kotlinx.coroutines.flow.MutableStateFlow @@ -55,7 +55,7 @@ val DefaultOrder = class UserRelaysCache { var data: Map = mapOf() private var flow: WeakReference>? = null - private val flowLock = Lock() + private val flowLock = KmpLock() fun flow() = flow?.get() ?: flowLock.withLock { diff --git a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relayClient/composeSubscriptionManagers/ComposeSubscriptionManager.kt b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relayClient/composeSubscriptionManagers/ComposeSubscriptionManager.kt index 4e66aee673..22ffa50d4e 100644 --- a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relayClient/composeSubscriptionManagers/ComposeSubscriptionManager.kt +++ b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relayClient/composeSubscriptionManagers/ComposeSubscriptionManager.kt @@ -21,7 +21,8 @@ package com.vitorpamplona.amethyst.commons.relayClient.composeSubscriptionManagers import androidx.compose.runtime.Stable -import co.touchlab.stately.collections.ConcurrentMutableMap +import com.vitorpamplona.amethyst.commons.util.KmpLock +import com.vitorpamplona.amethyst.commons.util.withLock /** * This allows composables to directly register their queries @@ -32,13 +33,16 @@ import co.touchlab.stately.collections.ConcurrentMutableMap abstract class ComposeSubscriptionManager : ComposeSubscriptionManagerControls, Subscribable { - private val composeSubscriptions = ConcurrentMutableMap() + // T has no Comparable bound — see the note in MutableComposeSubscriptionManager + // for why LargeCache isn't used here. + private val lock = KmpLock() + private val composeSubscriptions = mutableMapOf() // This is called by main. Keep it really fast. override fun subscribe(query: T?) { if (query == null) return - composeSubscriptions.put(query, query) + lock.withLock { composeSubscriptions[query] = query } invalidateKeys() } @@ -47,7 +51,7 @@ abstract class ComposeSubscriptionManager : override fun unsubscribe(query: T?) { if (query == null) return - composeSubscriptions.remove(query) + lock.withLock { composeSubscriptions.remove(query) } invalidateKeys() } @@ -55,8 +59,8 @@ abstract class ComposeSubscriptionManager : override fun subscribe(query: List) { if (query.isEmpty()) return - query.forEach { - composeSubscriptions.put(it, it) + lock.withLock { + query.forEach { composeSubscriptions[it] = it } } invalidateKeys() @@ -66,14 +70,14 @@ abstract class ComposeSubscriptionManager : override fun unsubscribe(query: List) { if (query.isEmpty()) return - query.forEach { - composeSubscriptions.remove(it) + lock.withLock { + query.forEach { composeSubscriptions.remove(it) } } invalidateKeys() } - fun allKeys() = composeSubscriptions.keys + fun allKeys(): Set = lock.withLock { composeSubscriptions.keys.toSet() } } interface Subscribable { diff --git a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relayClient/composeSubscriptionManagers/MutableComposeSubscriptionManager.kt b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relayClient/composeSubscriptionManagers/MutableComposeSubscriptionManager.kt index 61e6d46ce7..97eabca106 100644 --- a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relayClient/composeSubscriptionManagers/MutableComposeSubscriptionManager.kt +++ b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relayClient/composeSubscriptionManagers/MutableComposeSubscriptionManager.kt @@ -21,7 +21,8 @@ package com.vitorpamplona.amethyst.commons.relayClient.composeSubscriptionManagers import androidx.compose.runtime.Stable -import co.touchlab.stately.collections.ConcurrentMutableMap +import com.vitorpamplona.amethyst.commons.util.KmpLock +import com.vitorpamplona.amethyst.commons.util.withLock import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Job import kotlinx.coroutines.flow.Flow @@ -41,19 +42,26 @@ import kotlinx.coroutines.launch abstract class MutableComposeSubscriptionManager( val scope: CoroutineScope, ) : ComposeSubscriptionManagerControls { - private val composeSubscriptions = ConcurrentMutableMap() + // T is generic and not required to be Comparable, so this can't use + // LargeCache (ConcurrentSkipListMap-backed on JVM). A plain map guarded by + // KmpLock gives the same atomicity guarantees ConcurrentHashMap did + // previously, with no Comparable requirement. + private val lock = KmpLock() + private val composeSubscriptions = mutableMapOf() // This is called by main. Keep it really fast. fun subscribe(query: T?) { if (query == null) return - composeSubscriptions[query]?.cancel() - composeSubscriptions[query] = - scope.launch { - query.flow().collectLatest { - invalidateKeys() + lock.withLock { + composeSubscriptions[query]?.cancel() + composeSubscriptions[query] = + scope.launch { + query.flow().collectLatest { + invalidateKeys() + } } - } + } invalidateKeys() } @@ -62,16 +70,19 @@ abstract class MutableComposeSubscriptionManager( fun unsubscribe(query: T?) { if (query == null) return - composeSubscriptions[query]?.cancel() - composeSubscriptions.remove(query) + lock.withLock { + composeSubscriptions[query]?.cancel() + composeSubscriptions.remove(query) + } invalidateKeys() } - fun allKeys() = composeSubscriptions.keys + fun allKeys(): Set = lock.withLock { composeSubscriptions.keys.toSet() } fun forEachSubscriber(action: (T) -> Unit) { - composeSubscriptions.keys.forEach(action) + val snapshot = lock.withLock { composeSubscriptions.keys.toList() } + snapshot.forEach(action) } } diff --git a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/util/KmpLock.kt b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/util/KmpLock.kt new file mode 100644 index 0000000000..b6a5ac0a66 --- /dev/null +++ b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/util/KmpLock.kt @@ -0,0 +1,44 @@ +/* + * 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 reentrant lock. JVM/Android map to `java.util.concurrent.locks.ReentrantLock`; + * iOS will map to `NSLock` (or a thin wrapper around it) when the target is added. + * + * Use [withLock] in preference to manual lock/unlock — it guarantees release on + * exception. The class is reentrant on every platform, so a thread that already + * holds the lock can re-enter without deadlocking. + */ +expect class KmpLock() { + fun lock() + + fun unlock() +} + +inline fun KmpLock.withLock(block: () -> T): T { + lock() + try { + return block() + } finally { + unlock() + } +} diff --git a/commons/src/jvmAndroid/kotlin/com/vitorpamplona/amethyst/commons/util/KmpLock.jvmAndroid.kt b/commons/src/jvmAndroid/kotlin/com/vitorpamplona/amethyst/commons/util/KmpLock.jvmAndroid.kt new file mode 100644 index 0000000000..db64401eb5 --- /dev/null +++ b/commons/src/jvmAndroid/kotlin/com/vitorpamplona/amethyst/commons/util/KmpLock.jvmAndroid.kt @@ -0,0 +1,31 @@ +/* + * 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.locks.ReentrantLock + +actual class KmpLock { + private val delegate = ReentrantLock() + + actual fun lock() = delegate.lock() + + actual fun unlock() = delegate.unlock() +} diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml index c33aeb00e7..24ca396591 100644 --- a/gradle/libs.versions.toml +++ b/gradle/libs.versions.toml @@ -69,7 +69,6 @@ zelory = "3.0.1" zoomable = "2.11.1" vlcj = "4.8.3" commonsImaging = "1.0.0-alpha6" -statelyConcurrent = "2.1.0" zxing = "3.5.4" zxingAndroidEmbedded = "4.3.0" windowCoreAndroid = "1.5.1" @@ -101,7 +100,6 @@ androidx-compose-bom = { group = "androidx.compose", name = "compose-bom", versi androidx-compose-foundation = { group = "androidx.compose.foundation", name = "foundation" } androidx-compose-runtime-annotation = { group = "androidx.compose.runtime", name = "runtime-annotation", version.ref = "composeRuntimeAnnotation" } androidx-collection = { group = "androidx.collection", name = "collection", version.ref = "androidxCollection" } -stately-concurrent-collections = { group = "co.touchlab", name = "stately-concurrent-collections", version.ref = "statelyConcurrent" } androidx-exifinterface = { group = "androidx.exifinterface", name = "exifinterface", version.ref = "androidxExifinterface" } androidx-core-ktx = { group = "androidx.core", name = "core-ktx", version.ref = "coreKtx" } androidx-datastore-preferences = { module = "androidx.datastore:datastore-preferences", version.ref = "datastore" }