diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/filters/FilterIndex.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/filters/FilterIndex.kt index 1cae5b471d..a5c058ec6d 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/filters/FilterIndex.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/filters/FilterIndex.kt @@ -22,6 +22,11 @@ package com.vitorpamplona.quartz.nip01Core.relay.filters import com.vitorpamplona.quartz.nip01Core.core.Event import com.vitorpamplona.quartz.nip01Core.core.HexKey +import kotlinx.collections.immutable.PersistentMap +import kotlinx.collections.immutable.PersistentSet +import kotlinx.collections.immutable.persistentHashMapOf +import kotlinx.collections.immutable.persistentHashSetOf +import kotlinx.collections.immutable.toPersistentHashSet import kotlin.concurrent.atomics.AtomicReference import kotlin.concurrent.atomics.ExperimentalAtomicApi @@ -62,9 +67,10 @@ import kotlin.concurrent.atomics.ExperimentalAtomicApi * copy-on-write CAS loops, mirroring the * `nip86RelayManagement.server.BanStore` pattern. Reads in * [candidatesFor] and [forEach] are wait-free single-load atomic. - * Writes (subscription register / unregister) copy the inner maps - * — fine for this workload because writes are subscription-rate - * (rare) while reads are event-rate (frequent). + * Writes (subscription register / unregister) build the next snapshot + * from persistent (HAMT) maps — O(keys × log S) with structural + * sharing rather than a full O(S) copy of both maps, since on a relay + * a write happens on every REQ open and close. * * ## What the index does NOT cover * @@ -113,10 +119,16 @@ class FilterIndex { * subscribers registered under it; [assignments] is the reverse * map used by [unregister] to find a subscriber's keys without * scanning every bucket. + * + * Persistent (HAMT) maps/sets: a register/unregister produces the + * next snapshot in O(keys × log S) with structural sharing, instead + * of copying both full maps — registration happens on every REQ + * open/close, so with S live subscriptions the full copy was + * O(S) work and O(S) allocation per REQ. */ private data class State( - val buckets: Map> = emptyMap(), - val assignments: Map> = emptyMap(), + val buckets: PersistentMap> = persistentHashMapOf(), + val assignments: PersistentMap> = persistentHashMapOf(), ) private val state: AtomicReference> = AtomicReference(State()) @@ -181,17 +193,18 @@ class FilterIndex { while (true) { val current = state.load() val keys = current.assignments[subscriber] ?: return - val newBuckets = current.buckets.toMutableMap() + var newBuckets = current.buckets for (key in keys) { val cur = newBuckets[key] ?: continue - val next = cur - subscriber - if (next.isEmpty()) { - newBuckets.remove(key) - } else { - newBuckets[key] = next - } + val next = cur.remove(subscriber) + newBuckets = + if (next.isEmpty()) { + newBuckets.remove(key) + } else { + newBuckets.put(key, next) + } } - val newAssignments = current.assignments - subscriber + val newAssignments = current.assignments.remove(subscriber) if (state.compareAndSet(current, State(newBuckets, newAssignments))) return } } @@ -235,18 +248,18 @@ class FilterIndex { keys: List, ) { if (keys.isEmpty()) return - val keySet = keys.toSet() + val keySet = keys.toPersistentHashSet() while (true) { val current = state.load() - val newBuckets = current.buckets.toMutableMap() + var newBuckets = current.buckets for (key in keySet) { - val cur = newBuckets[key] ?: emptySet() - if (subscriber in cur) continue - newBuckets[key] = cur + subscriber + val cur = newBuckets[key] ?: persistentHashSetOf() + val next = cur.add(subscriber) + if (next !== cur) newBuckets = newBuckets.put(key, next) } val existing = current.assignments[subscriber] - val merged = if (existing == null) keySet else existing + keySet - val newAssignments = current.assignments + (subscriber to merged) + val merged = existing?.addAll(keySet) ?: keySet + val newAssignments = current.assignments.put(subscriber, merged) if (state.compareAndSet(current, State(newBuckets, newAssignments))) return } } diff --git a/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/SmallReqFloorBenchmark.kt b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/SmallReqFloorBenchmark.kt index 9fe3a6ec33..cd6eb4525b 100644 --- a/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/SmallReqFloorBenchmark.kt +++ b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/SmallReqFloorBenchmark.kt @@ -32,6 +32,7 @@ import com.vitorpamplona.quartz.nip01Core.store.sqlite.EventStore import com.vitorpamplona.quartz.utils.EventFactory import kotlinx.coroutines.CompletableDeferred import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.CoroutineStart import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.cancel @@ -65,6 +66,7 @@ class SmallReqFloorBenchmark { const val AUTHORS = 2_500 // ~20 events per author, matching author-archive const val ROUNDS = 400 const val WARMUP = 100 + const val IDLE_SUBS = 1_000 } private fun hexId(seed: Int): String = seed.toString(16).padStart(64, '0') @@ -122,11 +124,14 @@ class SmallReqFloorBenchmark { } // --- B: backend queryRaw to EOSE (live machinery included) --- + // UNDISPATCHED mirrors the production path (RelaySession.handleReq + // starts the query coroutine undispatched), so B−A is the live + // machinery itself, not a benchmark-only scheduler hop. suspend fun timeBackend(round: Int): Long { val eose = CompletableDeferred() val t0 = System.nanoTime() val job = - scope.launch { + scope.launch(start = CoroutineStart.UNDISPATCHED) { live.queryRaw( ctx = ctx, filters = listOf(filterFor(round)), @@ -143,6 +148,33 @@ class SmallReqFloorBenchmark { val b = LongArray(ROUNDS) repeat(ROUNDS) { b[it] = timeBackend(it) } + // --- B@1k: same, with 1000 idle live subscriptions parked --- + // Register/unregister cost scales with the live population + // (FilterIndex mutates a shared snapshot per REQ open/close), + // which the single-sub stage can't see. Each idle sub filters + // on an author absent from the corpus: 0-row replay, then parks + // at the live tail and stays registered. + val idleJobs = + (0 until IDLE_SUBS).map { i -> + val ready = CompletableDeferred() + val job = + scope.launch(start = CoroutineStart.UNDISPATCHED) { + live.queryRaw( + ctx = ctx, + filters = listOf(Filter(authors = listOf(hexId(1_000_000 + i)), kinds = listOf(1), limit = 1)), + onEachStored = {}, + onEachLive = {}, + onEose = { ready.complete(Unit) }, + ) + } + ready.await() + job + } + repeat(WARMUP) { timeBackend(it) } + val b1k = LongArray(ROUNDS) + repeat(ROUNDS) { b1k[it] = timeBackend(it) } + idleJobs.forEach { it.cancel() } + // --- C: full session dispatch, REQ json in → EOSE frame out --- suspend fun timeSession(round: Int): Long { val eose = CompletableDeferred() @@ -164,10 +196,12 @@ class SmallReqFloorBenchmark { val mA = median(a) val mB = median(b) + val mB1k = median(b1k) val mC = median(c) println("SmallReqFloorBenchmark @ ${EVENTS / 1000}k events, ~${rowsA / ROUNDS} rows/req, medians of $ROUNDS") println(" A raw store query: ${"%6.3f".format(mA)} ms") println(" B backend queryRaw→EOSE: ${"%6.3f".format(mB)} ms (live machinery +${"%6.3f".format(mB - mA)})") + println(" B@${IDLE_SUBS} idle subs: ${"%6.3f".format(mB1k)} ms (population cost +${"%6.3f".format(mB1k - mB)})") println(" C session REQ→EOSE: ${"%6.3f".format(mC)} ms (dispatch+frames +${"%6.3f".format(mC - mB)})") server.close()