mirror of
https://github.com/vitorpamplona/amethyst.git
synced 2026-10-05 19:28:25 +00:00
fix(relay): stripe the subscription-state lock per relay to end an ANR convoy
A production ANR on a Pixel 8 (Amethyst 1.13.1, anr_2026-08-03-12-55-26-256)
showed the app burning 596% CPU — 6 of 9 cores — with the main thread stuck in
WaitingForGcToComplete. 37 of 52 runnable DefaultDispatcher workers sat at ONE
program point inside PoolRequests.onIncomingMessage and 12 more at one point in
syncState, all state=R, while the single thread actually holding the lock was
itself parked in GC.
Root cause: RequestSubscriptionState.withLock was a raw busy-wait
(`while (lock.exchange(true)) { while (lock.load()) {} }`) with no yield or
backoff, and being `inline` it disappeared into its callers' frames. The lock is
per subId, but one subId spans every relay it runs on — 191 live sockets on that
device — so dozens of relay-dispatch threads piled onto a single AtomicBoolean.
Spinning is only correct when the holder cannot be descheduled; on Android it
always can.
The fix stripes the lock per (subId, relay) rather than making waiting cheaper.
All 11 withLock bodies in PoolRequests are already scoped to exactly one relay,
and every field of RequestSubscriptionState is keyed by relay, so the sharing was
purely an artifact of mutableMapOf not being thread-safe. State moves into a
ConcurrentMap<T, RelayState>; locks live in a fixed 32-entry stripe array that is
never mutated, so lock identity stays stable — if locks lived inside the map
values, a thread holding one while another dropped and re-created that entry
would leave both inside the critical section excluding nothing.
A suspending Mutex was measured and rejected: it needs 262 method overrides and
110 call sites to become suspend, and ran at 0.35-0.63x the current throughput.
Measured (LockDesignComparisonBenchmark, 191 relays / 64 dispatcher threads):
striped vs per-sub lock 1.5-2.8x throughput, bystander p50 halved
On device (SM-T220, playBenchmark, same account, n=3 per design):
DefaultDispatcher CPU -35% mean / -30% median vs the spin lock,
with non-overlapping ranges; GC -18%
Plus 10 min of driven UI stress (feed, profiles, chat, notifications,
communities): no ANRs, no crashes, thread pools stable.
Also here:
- PlatformLock: new expect/actual parking lock (ReentrantLock on jvmAndroid,
NSRecursiveLock on Apple, spin only on linuxX64 which is a CI target). quartz
cannot use commons' equivalent KmpLock because commons depends on quartz.
- LiveNegentropyIndex had the identical busy-wait with a full list SORT inside
the critical section; switched to PlatformLock.
- ConcurrentMap.remove (+ tests), with a caution that a removable value must not
own a lock callers acquire.
- SpinLockConvoyBenchmark: regression guard asserting contended waiters PARK
rather than spin (fails-before / passes-after). Pure benchmarks are gated
behind -PprodRelayBench=1, so CI cost is 0.3s rather than 51.5s.
Analysis and measurements: quartz/plans/2026-08-03-poolrequests-lock-contention.md
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Vitor Pamplona
co-authored by
Claude Opus 5
Claude Opus 5
parent
fdc68e801a
commit
5896ae5eff
@@ -0,0 +1,221 @@
|
||||
# PoolRequests subscription-state lock: what a suspending Mutex would cost
|
||||
|
||||
**Date:** 2026-08-03
|
||||
**Status:** analysis + measurements; recommendation is NOT to use `Mutex`
|
||||
**Harness:** `quartz/src/jvmTest/.../prodbench/LockDesignComparisonBenchmark.kt`
|
||||
(`./gradlew :quartz:jvmTest --tests "*.LockDesignComparisonBenchmark"`)
|
||||
**Context:** follows the Pixel 8 ANR (`anr_2026-08-03-12-55-26-256`) that replaced
|
||||
`RequestSubscriptionState`'s busy-wait with a parking `PlatformLock`.
|
||||
|
||||
## The question
|
||||
|
||||
The parking lock removed the CPU burn, but it still **blocks** rather than suspends.
|
||||
Relay consumers run as coroutines on `Dispatchers.IO` (limitedParallelism 64) and
|
||||
production has ~191 live relays, so a blocked waiter occupies one of 64 dispatcher
|
||||
threads. kotlinx's scheduler treats IO tasks as blocking and grows the pool when they
|
||||
block — which is why the on-device fix moved CPU but left thread count flat (+7%).
|
||||
|
||||
Would `kotlinx.coroutines.sync.Mutex` (a waiter *suspends*, freeing its thread) be
|
||||
better?
|
||||
|
||||
## What Mutex actually costs
|
||||
|
||||
The lock sits at the bottom of a call chain that is **entirely non-suspend**:
|
||||
|
||||
```
|
||||
OkHttpWebSocket: scope.launch { for (m in incomingMessages) out.onMessage(m) } <- coroutine
|
||||
WebSocketListener.onMessage (non-suspend)
|
||||
BasicRelayClient.MyWebsocketListener.onMessage
|
||||
RelayPool.onIncomingMessage
|
||||
NostrClient.onIncomingMessage (RelayConnectionListener)
|
||||
PoolRequests.onIncomingMessage
|
||||
RequestSubscriptionState.withLock <- the lock
|
||||
SubscriptionListener.onEvent (non-suspend, fans out to the app)
|
||||
```
|
||||
|
||||
`Mutex.lock()` is `suspend`, so every frame above it must become `suspend`. Measured
|
||||
blast radius:
|
||||
|
||||
| interface / API | count |
|
||||
|---|---|
|
||||
| `override fun onEvent(` | 59 |
|
||||
| `override fun onEose(` | 52 |
|
||||
| `override fun onIncomingMessage(` | 35 |
|
||||
| `override fun onCannotConnect(` | 32 |
|
||||
| `override fun onClosed(` | 25 |
|
||||
| `override fun onDisconnected(` | 17 |
|
||||
| `override fun onConnected(` | 16 |
|
||||
| `override fun onSent(` | 14 |
|
||||
| `override fun onConnecting(` | 11 |
|
||||
| others (`onSubscriptionStarted`, …) | 1 |
|
||||
| **total overrides to convert** | **262** |
|
||||
| `.subscribe(` / `.unsubscribe(` call sites | **110** |
|
||||
|
||||
Worse than the count: the *entry points* are not all coroutines. `INostrClient` is
|
||||
non-suspend by design — `subscribe`, `unsubscribe`, `publish`, `syncFilters`,
|
||||
`connect` — and is called from ViewModels, filter assemblers and Compose effects.
|
||||
`PoolRequests.addOrUpdate` / `remove` / `sendToRelayIfChanged` reach the lock from
|
||||
those paths. Making them suspend pushes coroutine scoping into every call site that
|
||||
today just calls `subscribe(...)` synchronously.
|
||||
|
||||
## The cheaper alternative: stripe the lock per relay
|
||||
|
||||
**Enabling invariant (verified by reading all 11 `withLock` bodies):** *every*
|
||||
critical section in `PoolRequests` is scoped to exactly one relay. Each one takes
|
||||
`url` / `relay` / `relay.url` as its key:
|
||||
|
||||
| line | body | key |
|
||||
|---|---|---|
|
||||
| 204 | `state.connecting(url)` | url |
|
||||
| 218 | `state.onOpenReq(relay, cmd.filters)` | relay |
|
||||
| 228 | `state.onSubscriptionClosed(relay)` | relay |
|
||||
| 249 | `onNewEvent` / `currentState` / `lastKnownFilterStates` | relay.url |
|
||||
| 267 | `onEose` + `decideCommandLocked` | relay.url |
|
||||
| 296 | `onClosed` + `recordRefusalIfStructural` + `decideCommandLocked` | relay.url |
|
||||
| 325 | `state.disconnected(url)` | url |
|
||||
| 354 | `isStructurallyRefused` + `onOpenReq` | relay |
|
||||
| 382 | `lastKnownFilterStates(url)` | url |
|
||||
| 403 | `decideCommandLocked(state, subId, relay)` | relay |
|
||||
|
||||
Every field in `RequestSubscriptionState` is a `Map<T, …>` keyed by relay
|
||||
(`subStates`, `filterStates`, `lastKnownFilterStates`, `refusedFilters`,
|
||||
`refusalCounts`). The single cross-relay accessor, `currentFilters()` (no-arg,
|
||||
returns the whole map), has **zero usages** — dead code, delete it.
|
||||
|
||||
So the lock is only shared across relays as an artifact of `mutableMapOf` not being
|
||||
thread-safe. One lock per `(subId, relay)` is semantically equivalent and drops
|
||||
contention from ~191 threads to ~1–2 (that relay's consumer, plus the occasional app
|
||||
thread in `sendToRelayIfChanged`).
|
||||
|
||||
Blast radius: `RequestSubscriptionState` + `PoolRequests` only. Used by 4 production
|
||||
files (`PoolRequests`, `RelayActiveRequestStates`, `RelayReqRefusals`) and 2 tests.
|
||||
**No public API change.**
|
||||
|
||||
## Measurements
|
||||
|
||||
191 relay coroutines on a 64-thread limited-parallelism dispatcher, 3s windows. Each
|
||||
iteration yields, modelling the real per-message suspension point of
|
||||
`for (message in incomingMessages)` — without it the tight loops monopolise the
|
||||
dispatcher and the harness measures itself, not the lock.
|
||||
|
||||
`bystander` = a task that never touches the lock, on the same dispatcher. Its latency
|
||||
answers "is the lock stealing dispatcher threads from unrelated work?"
|
||||
|
||||
| subs | design | ops/s | bystander p50 | p99 |
|
||||
|---|---|---|---|---|
|
||||
| 1 | PER_SUB_BLOCKING (today) | 548,608 | 196.5µs | 770.9µs |
|
||||
| 1 | PER_SUB_MUTEX | 190,672 | **1.9µs** | 30.4µs |
|
||||
| 1 | **STRIPED** | **1,543,037** | 88.8µs | 237.6µs |
|
||||
| 4 | PER_SUB_BLOCKING | 833,090 | 136.1µs | 507.6µs |
|
||||
| 4 | PER_SUB_MUTEX | 523,440 | **3.1µs** | 26.0µs |
|
||||
| 4 | **STRIPED** | **1,499,757** | 92.3µs | 337.5µs |
|
||||
| 16 | PER_SUB_BLOCKING | 1,198,581 | 94.3µs | 444.3µs |
|
||||
| 16 | PER_SUB_MUTEX | 749,487 | **5.0µs** | 16.5µs |
|
||||
| 16 | **STRIPED** | **1,789,509** | 85.7µs | 219.8µs |
|
||||
|
||||
Reading:
|
||||
|
||||
- **STRIPED gives the most throughput** — 2.8× / 1.8× / 1.5× over today — and roughly
|
||||
halves bystander p50. It removes the contention rather than tolerating it.
|
||||
- **MUTEX is the slowest of the three** (0.35× / 0.63× / 0.63× of today): per-acquisition
|
||||
suspend/resume is not free at these rates.
|
||||
- **MUTEX does win bystander latency decisively** (1.9–5.0µs vs 85–196µs, and it collected
|
||||
626k vs 24k samples). But read that honestly — part of the win is that its coroutines
|
||||
spend their time *suspended waiting for the mutex instead of doing work*. Better
|
||||
thread-yielding is partly a symptom of lower throughput, not purely a win.
|
||||
|
||||
## On-device result (SM-T220, playBenchmark, same account, n=3 per design)
|
||||
|
||||
Cold-start burst, 75s window, exact per-thread CPU accounting from `/proc/<tid>/stat`.
|
||||
Warmups matched (~220-255 established sockets, ~200 relay reader threads each time).
|
||||
|
||||
| design | DefaultDispatcher CPU (ms/75s) mean / median / range | GC ms | threads |
|
||||
|---|---|---|---|
|
||||
| spin lock (pre-fix) | 136,743 / 117,750 / 112,230–180,250 | 33,777 | 475 |
|
||||
| parking, one lock per sub | 103,780 / 99,160 / 86,120–130,680 | 30,235 | 510 |
|
||||
| **striped, per (sub, relay)** | **88,953 / 82,160 / 72,860–111,840** | **27,687** | 489 |
|
||||
|
||||
- **Striped vs spin: −35% mean, −30% median CPU, −18% GC — and the ranges do not
|
||||
overlap** (striped max 111,840 < spin min 112,230). That separation is what the
|
||||
parking-only change could not show; its range overlapped the baseline heavily.
|
||||
- **Striped vs parking: −14% mean / −17% median**, ranges still overlap — directional,
|
||||
not conclusive at n=3.
|
||||
- **Thread count is unchanged across all three** (475 / 510 / 489). Expected: the lock
|
||||
is blocking, and kotlinx marks IO tasks blocking and grows the pool. Striping reduces
|
||||
how *often* threads block, not the fact that they can.
|
||||
- Not a false win from broken subscriptions: the feed rendered live notes 3-15 min old
|
||||
with avatars and reaction counts, on 223 sockets / 200 relay reader threads.
|
||||
|
||||
## Recommendation
|
||||
|
||||
**Do the striping; do not convert to `Mutex`.**
|
||||
|
||||
Striping attacks the cause — the lock is contended only because it is shared across
|
||||
relays that touch disjoint keys. Once contention is ~0, the blocking-vs-suspending
|
||||
question is moot: *a lock that is never contended never blocks a thread*. It also
|
||||
happens to be the fastest option and is contained to two files, versus 262 overrides
|
||||
and 110 call sites for a design that measures slower.
|
||||
|
||||
`Mutex` only becomes the right answer if a future critical section must genuinely span
|
||||
relays (or do I/O), which none does today.
|
||||
|
||||
## The lock must not live inside a removable entry
|
||||
|
||||
The obvious shape — `ConcurrentMap<T, PerRelayState>` where `PerRelayState` owns both
|
||||
the fields *and* its lock — is **wrong as soon as entries can be removed**.
|
||||
`connecting()` / `disconnected()` genuinely mean "forget this relay's wire state", so
|
||||
they want removal, and then:
|
||||
|
||||
```
|
||||
T1: getOrPut(R) -> stateA ; stateA.lock.lock() // in a critical section
|
||||
T2: disconnected(R) -> remove(R) // stateA is now orphaned
|
||||
T3: getOrPut(R) -> stateB (NEW lock) ; stateB.lock.lock() // acquires immediately
|
||||
-> T1 and T3 are both "in" relay R's critical section, excluding nothing.
|
||||
```
|
||||
|
||||
Two ways out:
|
||||
|
||||
- **(A) never remove; null the fields.** Lock identity is stable, but entries
|
||||
accumulate for every relay a sub has *ever* seen. Bounded but monotonic — and slow
|
||||
monotonic growth in a long-lived process is precisely the class of bug this whole
|
||||
investigation was about.
|
||||
- **(B) stable stripe locks + removable state.** A fixed `Array(N) { PlatformLock() }`
|
||||
indexed by `relay.hashCode()`, never mutated, so lock identity can't change; the
|
||||
`ConcurrentMap<T, RelayState>` entries are then free to be added and removed with
|
||||
exact `connecting`/`disconnected` semantics and no growth.
|
||||
|
||||
**(B) is the recommendation.** With N = 32 and ~191 relays, ~6 relays share a stripe —
|
||||
still a ~32x contention reduction versus today's single lock per sub, with none of the
|
||||
lock-lifetime hazard. `ConcurrentMap.remove` (added 2026-08-03, with tests in
|
||||
`ConcurrentCollectionsTest`) is what makes (B) possible.
|
||||
|
||||
## Implementation sketch
|
||||
|
||||
1. Delete the dead `currentFilters()` (no-arg).
|
||||
2. In `RequestSubscriptionState`, replace the five relay-keyed maps with a single
|
||||
`ConcurrentMap<T, RelayState>` holding the five fields — **no lock inside**.
|
||||
3. Add a fixed `private val stripes = Array(32) { PlatformLock() }` and
|
||||
`withLock(reference)` = `stripes[reference.hashCode().absoluteValue % 32].withLock { }`.
|
||||
The array is never mutated, so lock identity is stable for the object's life.
|
||||
4. `connecting()` / `disconnected()` become `map.remove(reference)` — exact current
|
||||
semantics, no residue.
|
||||
5. Update the 11 `withLock` call sites in `PoolRequests` to pass the relay.
|
||||
`decideCommandLocked`'s "MUST hold the lock" contract becomes "MUST hold *that
|
||||
relay's stripe*" — tighten the kdoc.
|
||||
6. Extend `PoolRequestsRefusalTest` with concurrent multi-relay access on one subId,
|
||||
and add a striped variant to `LockDesignComparisonBenchmark` to confirm the
|
||||
measured win survives stripe collisions.
|
||||
|
||||
## Risks
|
||||
|
||||
- **Stripe collisions serialize unrelated relays.** With 32 stripes and 191 relays this
|
||||
is ~6-way sharing; it is a contention *reduction*, not elimination. If a future
|
||||
profile shows it mattering, raise N — it is a one-line change with no semantic effect.
|
||||
- **Two relays on one stripe must never be locked simultaneously by one thread** — that
|
||||
would self-deadlock (`PlatformLock` is reentrant on JVM/Apple, so same-thread
|
||||
re-entry is survivable, but the invariant should be stated). `PoolRequests` already
|
||||
locks one relay at a time.
|
||||
- The state machine's atomicity comments are load-bearing; the per-relay scoping must
|
||||
be re-verified against any new `withLock` body added between now and the change.
|
||||
- Striping is NOT a fix for a critical section that does I/O or spans relays. If one is
|
||||
ever added, this analysis must be redone.
|
||||
+36
@@ -0,0 +1,36 @@
|
||||
/*
|
||||
* 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.concurrent
|
||||
|
||||
import platform.Foundation.NSRecursiveLock
|
||||
|
||||
// NSRecursiveLock parks contended waiters in the kernel, matching the
|
||||
// ReentrantLock semantics of the jvmAndroid actual (and the same choice
|
||||
// commons' KmpLock made for iOS). This must NOT be a spin lock: quartz's
|
||||
// relay client runs here too, and spinning is what produced the Android
|
||||
// ANR documented on the expect declaration.
|
||||
actual class PlatformLock {
|
||||
private val delegate = NSRecursiveLock()
|
||||
|
||||
actual fun lock() = delegate.lock()
|
||||
|
||||
actual fun unlock() = delegate.unlock()
|
||||
}
|
||||
+28
-27
@@ -84,25 +84,26 @@ class PoolRequests(
|
||||
/*
|
||||
* Locking model: every compound access to a subscription's state machine
|
||||
* ([RequestSubscriptionState]) — including the check-then-send decision in
|
||||
* [decideCommandLocked] — runs inside THAT subscription's own lock
|
||||
* ([RequestSubscriptionState.withLock]).
|
||||
* [decideCommandLocked] — runs inside the stripe for THAT (subscription, relay)
|
||||
* pair ([RequestSubscriptionState.withLock], which takes the relay).
|
||||
*
|
||||
* A single subscription can span many relays, and each relay's
|
||||
* socket-reader thread delivers messages into this class concurrently
|
||||
* while the app thread adds/removes subscriptions — so the plain maps
|
||||
* inside [RequestSubscriptionState] are written from several threads at
|
||||
* once. That is both a memory hazard (concurrent map mutation) and a
|
||||
* logic hazard: two threads must never both observe "no REQ in flight"
|
||||
* and both send a REQ for the same sub id.
|
||||
* A single subscription can span many relays, and each relay's socket-reader
|
||||
* thread delivers messages into this class concurrently while the app thread
|
||||
* adds/removes subscriptions. Two threads must never both observe "no REQ in
|
||||
* flight" and both send a REQ for the same (sub, relay).
|
||||
*
|
||||
* The lock is per subscription, not global, because different subIds
|
||||
* share no wire state: EVENT frames for different subs coming from
|
||||
* different relay consumer threads must not serialize on each other (a
|
||||
* global lock here measured negative scaling under 4 concurrent relay
|
||||
* feeders). Listener callbacks and the actual socket sends are ALWAYS
|
||||
* performed outside the lock — they re-enter this class through
|
||||
* [onSent], so holding the lock across them would self-deadlock, and the
|
||||
* lock is non-reentrant. Never hold two subscriptions' locks at once.
|
||||
* The lock is striped PER RELAY rather than per subscription: every critical
|
||||
* section below touches only one relay's slice of the state, so EVENT frames
|
||||
* arriving for the same subId from different relays no longer serialize on each
|
||||
* other. With ~191 relays that previously made a single per-sub lock contended
|
||||
* ~191 threads deep — the production ANR in
|
||||
* `quartz/plans/2026-08-03-poolrequests-lock-contention.md`.
|
||||
*
|
||||
* Two rules this file must keep:
|
||||
* - Never hold two relays' stripes (or two subscriptions') at once. Every loop
|
||||
* below locks exactly one relay at a time.
|
||||
* - Listener callbacks and socket sends are ALWAYS performed outside the lock —
|
||||
* they re-enter this class through [onSent].
|
||||
*/
|
||||
|
||||
/**
|
||||
@@ -201,7 +202,7 @@ class PoolRequests(
|
||||
fun onConnecting(url: NormalizedRelayUrl) {
|
||||
// Change states to connecting. One sub's lock at a time.
|
||||
relayState.forEach { subId, state ->
|
||||
state.withLock { state.connecting(url) }
|
||||
state.withLock(url) { state.connecting(url) }
|
||||
}
|
||||
}
|
||||
|
||||
@@ -215,7 +216,7 @@ class PoolRequests(
|
||||
when (cmd) {
|
||||
is ReqCmd -> {
|
||||
subState(cmd.subId).let { state ->
|
||||
state.withLock { state.onOpenReq(relay, cmd.filters) }
|
||||
state.withLock(relay) { state.onOpenReq(relay, cmd.filters) }
|
||||
}
|
||||
desiredSubListeners.get(cmd.subId)?.onSubscriptionStarted(
|
||||
relay = relay.url,
|
||||
@@ -225,7 +226,7 @@ class PoolRequests(
|
||||
|
||||
is CloseCmd -> {
|
||||
subState(cmd.subId).let { state ->
|
||||
state.withLock { state.onSubscriptionClosed(relay) }
|
||||
state.withLock(relay) { state.onSubscriptionClosed(relay) }
|
||||
}
|
||||
desiredSubListeners.get(cmd.subId)?.onSubscriptionClosed(
|
||||
relay = relay.url,
|
||||
@@ -246,7 +247,7 @@ class PoolRequests(
|
||||
var isLive = false
|
||||
var forFilters: List<Filter>? = null
|
||||
relayState.get(msg.subId)?.let { state ->
|
||||
state.withLock {
|
||||
state.withLock(relay.url) {
|
||||
state.onNewEvent(relay.url)
|
||||
isLive = state.currentState(relay.url) == ReqSubStatus.LIVE
|
||||
forFilters = state.lastKnownFilterStates(relay.url)
|
||||
@@ -264,7 +265,7 @@ class PoolRequests(
|
||||
var forFilters: List<Filter>? = null
|
||||
val cmd =
|
||||
relayState.get(msg.subId)?.let { state ->
|
||||
state.withLock {
|
||||
state.withLock(relay.url) {
|
||||
state.onEose(relay.url)
|
||||
forFilters = state.lastKnownFilterStates(relay.url)
|
||||
// Decide (and pre-mark) the resend while still holding the
|
||||
@@ -293,7 +294,7 @@ class PoolRequests(
|
||||
var forFilters: List<Filter>? = null
|
||||
val cmd =
|
||||
relayState.get(msg.subId)?.let { state ->
|
||||
state.withLock {
|
||||
state.withLock(relay.url) {
|
||||
state.onClosed(relay.url)
|
||||
forFilters = state.lastKnownFilterStates(relay.url)
|
||||
recordRefusalIfStructural(state, relay.url, msg.message, forFilters)
|
||||
@@ -322,7 +323,7 @@ class PoolRequests(
|
||||
*/
|
||||
fun onDisconnected(url: NormalizedRelayUrl) {
|
||||
relayState.forEach { subId, state ->
|
||||
state.withLock { state.disconnected(url) }
|
||||
state.withLock(url) { state.disconnected(url) }
|
||||
}
|
||||
}
|
||||
|
||||
@@ -351,7 +352,7 @@ class PoolRequests(
|
||||
if (!filters.isNullOrEmpty()) {
|
||||
val send =
|
||||
subState(subId).let { state ->
|
||||
state.withLock {
|
||||
state.withLock(relay) {
|
||||
if (isStructurallyRefused(state, relay, filters)) {
|
||||
false
|
||||
} else {
|
||||
@@ -379,7 +380,7 @@ class PoolRequests(
|
||||
// These are all my subs.. need to figure out which relays have them
|
||||
val subs = desiredSubs.get(subId)
|
||||
if (subs != null && url in subs.keys) {
|
||||
toNotify.add(subId to state.withLock { state.lastKnownFilterStates(url) })
|
||||
toNotify.add(subId to state.withLock(url) { state.lastKnownFilterStates(url) })
|
||||
}
|
||||
}
|
||||
|
||||
@@ -400,7 +401,7 @@ class PoolRequests(
|
||||
val state = subState(subId)
|
||||
relaysToUpdate.forEach { relay ->
|
||||
// Decide + pre-mark atomically under the sub's lock, then send outside it.
|
||||
val cmd = state.withLock { decideCommandLocked(state, subId, relay) }
|
||||
val cmd = state.withLock(relay) { decideCommandLocked(state, subId, relay) }
|
||||
if (cmd != null) {
|
||||
sync(relay, cmd)
|
||||
}
|
||||
|
||||
+139
-89
@@ -21,84 +21,116 @@
|
||||
package com.vitorpamplona.quartz.nip01Core.relay.client.reqs
|
||||
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
|
||||
import kotlin.concurrent.atomics.AtomicBoolean
|
||||
import kotlin.concurrent.atomics.ExperimentalAtomicApi
|
||||
import com.vitorpamplona.quartz.utils.concurrent.ConcurrentMap
|
||||
import com.vitorpamplona.quartz.utils.concurrent.PlatformLock
|
||||
|
||||
/**
|
||||
* Manages the State of Subscriptions by logging states as the
|
||||
* subscription progresses.
|
||||
*
|
||||
* Thread-safety: the plain maps below are only touched inside [withLock].
|
||||
* The lock lives HERE — one per subscription — instead of a single global
|
||||
* lock in PoolRequests, because every mutation is scoped to one subId:
|
||||
* relays delivering EVENTs for *different* subscriptions have no shared
|
||||
* state and must not serialize on each other. (A single global spin lock
|
||||
* measured *negative* scaling: 4 relay consumer threads pushed less
|
||||
* aggregate throughput through it than 1 — see
|
||||
* quartz/plans/2026-07-02-nostrclient-receiver-perf.md.)
|
||||
* **Thread-safety: the lock is striped per reference (per relay), not per
|
||||
* subscription.** Every operation below is scoped to a single [reference] — all state
|
||||
* lives in [RelayState] objects held in a [ConcurrentMap] keyed by it — so two relays
|
||||
* delivering EVENTs for the SAME subscription touch disjoint state and no longer
|
||||
* serialize on each other.
|
||||
*
|
||||
* That mattered in production: one subId spans every relay it is subscribed on, so a
|
||||
* single per-subscription lock was contended ~191 threads deep on the Pixel 8 that
|
||||
* produced `anr_2026-08-03-12-55-26-256` (it was a *spin* lock then, which burned 6 of
|
||||
* 9 cores — see [PlatformLock]). Striping removes the contention instead of making
|
||||
* waiting cheaper: measured 1.5-2.8x the throughput of one lock per sub in
|
||||
* `quartz/src/jvmTest/.../prodbench/LockDesignComparisonBenchmark.kt`. Full analysis,
|
||||
* including why a suspending `Mutex` was rejected, is in
|
||||
* `quartz/plans/2026-08-03-poolrequests-lock-contention.md`.
|
||||
*
|
||||
* The stripe array is allocated once and NEVER mutated, so a stripe's identity is
|
||||
* stable for this object's whole life. That is load-bearing: if locks lived inside the
|
||||
* per-relay values, a thread holding one while another thread dropped and re-created
|
||||
* that entry would leave both "inside" the critical section excluding nothing.
|
||||
*
|
||||
* Callers MUST NOT hold two references' stripes at once ([PoolRequests] locks one relay
|
||||
* at a time, including inside its all-subs iterations), and MUST keep socket sends and
|
||||
* listener callbacks outside the critical section — they re-enter this class through
|
||||
* `onSent`, and the point of a lock is to be held briefly.
|
||||
*/
|
||||
@OptIn(ExperimentalAtomicApi::class)
|
||||
class RequestSubscriptionState<T> {
|
||||
class RequestSubscriptionState<T : Any> {
|
||||
/**
|
||||
* Tiny non-reentrant spin lock (same primitive as BasicRelayClient's
|
||||
* connecting mutex). Critical sections are a handful of map operations,
|
||||
* never I/O — callers MUST NOT re-enter and MUST NOT hold two
|
||||
* subscriptions' locks at once (PoolRequests locks one sub at a time,
|
||||
* including inside its all-subs iterations).
|
||||
* One reference's (relay's) slice of this subscription's state. Plain `var`s: every
|
||||
* field is written and read under that reference's stripe by [PoolRequests].
|
||||
*
|
||||
* `@PublishedApi internal` only because [withLock] is inline (this sits
|
||||
* on the per-EVENT hot path; inlining avoids a closure allocation per
|
||||
* message) — treat it as private.
|
||||
* Note `RelayActiveRequestStates` uses this class WITHOUT locking. There the fields
|
||||
* may be read stale — but the backing [ConcurrentMap] can no longer be structurally
|
||||
* corrupted the way the plain `HashMap`s this replaced could.
|
||||
*/
|
||||
private class RelayState {
|
||||
/** Null == no REQ state on this relay (fresh, or wiped by connecting/disconnected). */
|
||||
var status: ReqSubStatus? = null
|
||||
|
||||
/** Filters of the REQ currently believed to be in flight. */
|
||||
var filters: List<Filter>? = null
|
||||
|
||||
/**
|
||||
* Survives connect/disconnect so that if new events still arrive we can link
|
||||
* them with the filters the relay was processing.
|
||||
*/
|
||||
var lastKnownFilters: List<Filter>? = null
|
||||
|
||||
/**
|
||||
* Refused-filter memory. Unlike [status]/[filters] — per-connection wire state
|
||||
* wiped by [connecting]/[disconnected] — this SURVIVES reconnects on purpose: a
|
||||
* relay that structurally refuses a filter (a search-only relay CLOSING a plain
|
||||
* kinds REQ, a relay that "does not accept REQs", "too many filters", …) refuses
|
||||
* it again on every new socket, so [PoolRequests.syncState] replaying it each
|
||||
* reconnect is pure waste. [refusalCount] accumulates repeated refusals of the
|
||||
* same shape so a one-off (transient) close isn't mistaken for a structural one.
|
||||
* Cleared on a successful REQ ([onEose]/[onNewEvent]) or when the caller observes
|
||||
* the desired filter meaningfully changed.
|
||||
*/
|
||||
var refusedFilters: List<Filter>? = null
|
||||
|
||||
var refusalCount: Int = 0
|
||||
}
|
||||
|
||||
private val states = ConcurrentMap<T, RelayState>()
|
||||
|
||||
/**
|
||||
* Fixed stripe array — allocated once, never mutated, so lock identity is stable.
|
||||
* [STRIPE_COUNT] stripes over ~191 relays is roughly 6-way sharing: a ~32x
|
||||
* contention reduction versus one lock per subscription. References colliding on a
|
||||
* stripe merely serialize; correctness never depends on N.
|
||||
*/
|
||||
@PublishedApi
|
||||
internal val lock = AtomicBoolean(false)
|
||||
internal val stripes = Array(STRIPE_COUNT) { PlatformLock() }
|
||||
|
||||
inline fun <R> withLock(block: () -> R): R {
|
||||
while (lock.exchange(true)) {
|
||||
// Test-and-test-and-set: spin-read until it looks free (cheaper
|
||||
// on the cache line than hammering exchange), then retry above.
|
||||
while (lock.load()) { }
|
||||
}
|
||||
@PublishedApi
|
||||
internal fun stripeFor(reference: T): PlatformLock = stripes[(reference.hashCode() and 0x7FFFFFFF) % STRIPE_COUNT]
|
||||
|
||||
/**
|
||||
* Runs [block] holding [reference]'s stripe. Inline so the per-EVENT hot path
|
||||
* allocates no closure.
|
||||
*/
|
||||
inline fun <R> withLock(
|
||||
reference: T,
|
||||
block: () -> R,
|
||||
): R {
|
||||
val lock = stripeFor(reference)
|
||||
lock.lock()
|
||||
try {
|
||||
return block()
|
||||
} finally {
|
||||
lock.store(false)
|
||||
lock.unlock()
|
||||
}
|
||||
}
|
||||
|
||||
// Logs the state of each channel to:
|
||||
// 1. inform when an event is received as live
|
||||
// 2. to block REQs being sent before finished (receiving an EOSE or Closed)
|
||||
//
|
||||
// If 2 happens, the relay might send multiple EOSEs in sequence
|
||||
// for the same sub and we won't know which REQ was it for.
|
||||
private val subStates = mutableMapOf<T, ReqSubStatus>()
|
||||
private val filterStates = mutableMapOf<T, List<Filter>>()
|
||||
/** Read-only lookup — never creates an entry. */
|
||||
private fun peek(reference: T): RelayState? = states[reference]
|
||||
|
||||
/**
|
||||
* This cache is used to make sure we know what the relay was processing
|
||||
* before a close or disconnect so that if new events still arrive
|
||||
* we can link them with the appropriate filters.
|
||||
*/
|
||||
private val lastKnownFilterStates = mutableMapOf<T, List<Filter>>()
|
||||
/** Write lookup — creates the entry on first use. */
|
||||
private fun mutable(reference: T): RelayState = states.getOrPut(reference) { RelayState() }
|
||||
|
||||
/**
|
||||
* Refused-filter memory. Unlike [subStates]/[filterStates] above — per-connection
|
||||
* wire state wiped by [connecting]/[disconnected] — this SURVIVES reconnects on
|
||||
* purpose: a relay that structurally refuses a filter (a search-only relay CLOSING
|
||||
* a plain kinds REQ, a relay that "does not accept REQs", "too many filters", …)
|
||||
* refuses it again on every new socket, so [PoolRequests.syncState] replaying it
|
||||
* each reconnect is pure waste. [refusalCounts] accumulates repeated refusals of
|
||||
* the same shape so a one-off (transient) close isn't mistaken for a structural
|
||||
* one. Cleared on a successful REQ ([onEose]/[onNewEvent]) or when the caller
|
||||
* observes the desired filter meaningfully changed.
|
||||
*/
|
||||
private val refusedFilters = mutableMapOf<T, List<Filter>>()
|
||||
private val refusalCounts = mutableMapOf<T, Int>()
|
||||
fun refusedFilters(reference: T) = peek(reference)?.refusedFilters
|
||||
|
||||
fun refusedFilters(reference: T) = refusedFilters[reference]
|
||||
|
||||
fun refusalCount(reference: T) = refusalCounts[reference] ?: 0
|
||||
fun refusalCount(reference: T) = peek(reference)?.refusalCount ?: 0
|
||||
|
||||
/**
|
||||
* Records that [reference] refused [filters]. [sameAsLastRefusal] must be true when
|
||||
@@ -111,73 +143,91 @@ class RequestSubscriptionState<T> {
|
||||
filters: List<Filter>,
|
||||
sameAsLastRefusal: Boolean,
|
||||
) {
|
||||
val state = mutable(reference)
|
||||
if (sameAsLastRefusal) {
|
||||
refusalCounts[reference] = refusalCount(reference) + 1
|
||||
state.refusalCount += 1
|
||||
} else {
|
||||
refusedFilters[reference] = filters
|
||||
refusalCounts[reference] = 1
|
||||
state.refusedFilters = filters
|
||||
state.refusalCount = 1
|
||||
}
|
||||
}
|
||||
|
||||
fun clearRefusal(reference: T) {
|
||||
refusedFilters.remove(reference)
|
||||
refusalCounts.remove(reference)
|
||||
peek(reference)?.let {
|
||||
it.refusedFilters = null
|
||||
it.refusalCount = 0
|
||||
}
|
||||
}
|
||||
|
||||
fun currentFilters() = filterStates
|
||||
fun currentFilters(reference: T) = peek(reference)?.filters
|
||||
|
||||
fun currentFilters(reference: T) = filterStates[reference]
|
||||
fun lastKnownFilterStates(reference: T) = peek(reference)?.lastKnownFilters
|
||||
|
||||
fun lastKnownFilterStates(reference: T) = lastKnownFilterStates[reference]
|
||||
|
||||
fun currentState(reference: T) = subStates[reference]
|
||||
fun currentState(reference: T) = peek(reference)?.status
|
||||
|
||||
fun onNewEvent(reference: T) {
|
||||
val state = mutable(reference)
|
||||
// The relay is serving this REQ (it matched an event), so any past refusal
|
||||
// no longer applies — let it be tried freely again.
|
||||
clearRefusal(reference)
|
||||
if (subStates[reference] == ReqSubStatus.SENT) {
|
||||
subStates[reference] = ReqSubStatus.QUERYING_PAST
|
||||
state.refusedFilters = null
|
||||
state.refusalCount = 0
|
||||
if (state.status == ReqSubStatus.SENT) {
|
||||
state.status = ReqSubStatus.QUERYING_PAST
|
||||
}
|
||||
}
|
||||
|
||||
fun onEose(reference: T) {
|
||||
val state = mutable(reference)
|
||||
// Reaching EOSE means the relay accepted and finished the REQ; clear any refusal.
|
||||
clearRefusal(reference)
|
||||
subStates[reference] = ReqSubStatus.LIVE
|
||||
state.refusedFilters = null
|
||||
state.refusalCount = 0
|
||||
state.status = ReqSubStatus.LIVE
|
||||
}
|
||||
|
||||
fun onClosed(reference: T) {
|
||||
subStates[reference] = ReqSubStatus.CLOSED
|
||||
// Closed messages are usually relays refusing to process a REQ
|
||||
// This message keeps the state of filterStates intact to
|
||||
// avoid sending the same filter, and getting immediately closed,
|
||||
// over and over again.
|
||||
|
||||
// filterStates.remove(reference)
|
||||
// Closed messages are usually relays refusing to process a REQ. This keeps
|
||||
// [RelayState.filters] intact to avoid sending the same filter, and getting
|
||||
// immediately closed, over and over again.
|
||||
mutable(reference).status = ReqSubStatus.CLOSED
|
||||
}
|
||||
|
||||
fun onOpenReq(
|
||||
reference: T,
|
||||
filters: List<Filter>,
|
||||
) {
|
||||
subStates[reference] = ReqSubStatus.SENT
|
||||
filterStates[reference] = filters
|
||||
lastKnownFilterStates[reference] = filters
|
||||
val state = mutable(reference)
|
||||
state.status = ReqSubStatus.SENT
|
||||
state.filters = filters
|
||||
state.lastKnownFilters = filters
|
||||
}
|
||||
|
||||
fun onSubscriptionClosed(reference: T) {
|
||||
subStates[reference] = ReqSubStatus.CLOSED
|
||||
filterStates.remove(reference)
|
||||
val state = mutable(reference)
|
||||
state.status = ReqSubStatus.CLOSED
|
||||
state.filters = null
|
||||
}
|
||||
|
||||
fun connecting(reference: T) {
|
||||
subStates.remove(reference)
|
||||
filterStates.remove(reference)
|
||||
// Wipes per-connection wire state only; lastKnownFilters and the refusal memory
|
||||
// deliberately survive (see [RelayState]).
|
||||
peek(reference)?.let {
|
||||
it.status = null
|
||||
it.filters = null
|
||||
}
|
||||
}
|
||||
|
||||
fun disconnected(reference: T) {
|
||||
subStates.remove(reference)
|
||||
filterStates.remove(reference)
|
||||
peek(reference)?.let {
|
||||
it.status = null
|
||||
it.filters = null
|
||||
}
|
||||
}
|
||||
|
||||
companion object {
|
||||
/**
|
||||
* Comfortably above the IO dispatcher's thread count (64 / 2) so collisions stay
|
||||
* rare even when many workers are inside this class at once.
|
||||
*/
|
||||
const val STRIPE_COUNT = 32
|
||||
}
|
||||
}
|
||||
|
||||
+12
-11
@@ -22,8 +22,7 @@ package com.vitorpamplona.quartz.nip77Negentropy
|
||||
|
||||
import com.vitorpamplona.negentropy.storage.IStorage
|
||||
import com.vitorpamplona.quartz.nip01Core.store.IdAndTime
|
||||
import kotlin.concurrent.atomics.AtomicBoolean
|
||||
import kotlin.concurrent.atomics.ExperimentalAtomicApi
|
||||
import com.vitorpamplona.quartz.utils.concurrent.PlatformLock
|
||||
|
||||
/**
|
||||
* Always-current `(created_at, id)` index for NIP-77 negentropy — the
|
||||
@@ -54,13 +53,12 @@ import kotlin.concurrent.atomics.ExperimentalAtomicApi
|
||||
* sealed storage, so one snapshot backs any number of concurrent
|
||||
* sessions and stays valid even if the live index mutates after.
|
||||
*
|
||||
* Thread-safety: all operations take a short spin lock (same pattern as
|
||||
* `LiveEventStore`'s replay dedup). Mutations arrive from the store's
|
||||
* single writer; snapshots from any REQ coroutine.
|
||||
* Thread-safety: all operations take a short parking lock ([PlatformLock]).
|
||||
* Mutations arrive from the store's single writer; snapshots from any REQ
|
||||
* coroutine.
|
||||
*/
|
||||
@OptIn(ExperimentalAtomicApi::class)
|
||||
class LiveNegentropyIndex {
|
||||
private val lock = AtomicBoolean(false)
|
||||
private val lock = PlatformLock()
|
||||
|
||||
/** Sorted by (createdAt, id). Only touched under [locked]. */
|
||||
private var entries = ArrayList<IdAndTime>()
|
||||
@@ -74,14 +72,17 @@ class LiveNegentropyIndex {
|
||||
private var cachedSnapshot: IStorage? = null
|
||||
private var cachedGeneration = -1L
|
||||
|
||||
// Parks rather than busy-waits. This lock's critical sections are FAR longer than a
|
||||
// handful of map ops — [rebuild] sorts the whole entry list and snapshotting builds a
|
||||
// storage — so a spinning waiter would burn a core for the duration of a sort while
|
||||
// concurrent ingest threads pile up. Same defect that produced the client-side ANR
|
||||
// documented on PlatformLock; see quartz/.../prodbench/SpinLockConvoyBenchmark.kt.
|
||||
private inline fun <R> locked(block: () -> R): R {
|
||||
while (lock.exchange(true)) {
|
||||
while (lock.load()) { }
|
||||
}
|
||||
lock.lock()
|
||||
try {
|
||||
return block()
|
||||
} finally {
|
||||
lock.store(false)
|
||||
lock.unlock()
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
+11
@@ -61,6 +61,17 @@ expect class ConcurrentMap<K : Any, V : Any>() {
|
||||
remap: (old: V, new: V) -> V,
|
||||
): V
|
||||
|
||||
/**
|
||||
* Removes [key] and returns the value it held, or null when absent.
|
||||
*
|
||||
* CAUTION: if the removed value owns a lock that callers acquire, removal
|
||||
* breaks mutual exclusion — a thread holding the old value's lock and a
|
||||
* thread that re-created the entry are no longer excluding each other. Keep
|
||||
* such locks in a structure whose identity is stable (see
|
||||
* `quartz/plans/2026-08-03-poolrequests-lock-contention.md`).
|
||||
*/
|
||||
fun remove(key: K): V?
|
||||
|
||||
fun size(): Int
|
||||
|
||||
/** A point-in-time copy of the entries — safe to iterate without holding a lock. */
|
||||
|
||||
+75
@@ -0,0 +1,75 @@
|
||||
/*
|
||||
* 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.concurrent
|
||||
|
||||
/**
|
||||
* A blocking mutual-exclusion lock whose waiters **park** (yield the core to the
|
||||
* scheduler) instead of busy-waiting.
|
||||
*
|
||||
* Why this exists: `commonMain` has no `java.util.concurrent.locks.Lock`, so the
|
||||
* relay client previously hand-rolled a spin lock over an `AtomicBoolean`. A
|
||||
* busy-wait is only ever correct when the holder cannot be descheduled while
|
||||
* holding the lock — and on Android that assumption is false. A production ANR on
|
||||
* a Pixel 8 (`anr_2026-08-03-12-55-26-256`, Amethyst 1.13.1) caught the exact
|
||||
* failure: the thread holding the lock was parked in `WaitingForGcToComplete`
|
||||
* while **51 of 52** runnable relay-dispatch threads sat in the inlined spin loop,
|
||||
* burning 596% CPU (6 of the phone's 9 cores) waiting for a holder that could not
|
||||
* be scheduled to release it. The UI thread, needing to allocate, then waited on
|
||||
* the same GC for over 5s and Android killed the frame with
|
||||
* "Input dispatching timed out".
|
||||
*
|
||||
* Measured on `SpinLockConvoyBenchmark` (12-core dev machine — a phone is worse):
|
||||
* the spin lock delivered 138M critical sections/s uncontended but only 1.2M/s
|
||||
* with 52 contenders (0.86%, a 116x collapse), and an *unrelated* allocating
|
||||
* thread's p90 latency went from 22µs to 10ms. Parking removes the CPU burn: a
|
||||
* waiter costs one context switch instead of a whole core.
|
||||
*
|
||||
* Splits the same way [ConcurrentMap] does:
|
||||
* - JVM / Android → `ReentrantLock` (real parking via `AbstractQueuedSynchronizer`).
|
||||
* - Apple → `NSRecursiveLock`, which also parks. The relay client genuinely runs
|
||||
* on iOS, so this must not spin — same choice commons' `KmpLock` already made.
|
||||
* - Linux → a spin, since Kotlin/Native ships no parking lock and there is no
|
||||
* Foundation; linuxX64 is a build/CI target, not a host for the many-relay
|
||||
* workload. Swap in a pthread mutex if that changes.
|
||||
*
|
||||
* (`commons` has an equivalent `KmpLock`, but `commons` depends on `quartz` and not
|
||||
* the reverse, so quartz cannot use it — keep the two in sync by hand.)
|
||||
*
|
||||
* Unlike the primitive it replaces, the JVM/Android actual is **reentrant**, so an
|
||||
* accidental re-entry degrades into a no-op rather than a self-deadlock. Callers
|
||||
* should still keep I/O and listener callbacks outside the critical section — that
|
||||
* discipline is about holding the lock briefly, not about avoiding a hang.
|
||||
*/
|
||||
expect class PlatformLock() {
|
||||
fun lock()
|
||||
|
||||
fun unlock()
|
||||
}
|
||||
|
||||
/** Runs [block] holding [this]. Inline so hot paths allocate no closure. */
|
||||
inline fun <R> PlatformLock.withLock(block: () -> R): R {
|
||||
lock()
|
||||
try {
|
||||
return block()
|
||||
} finally {
|
||||
unlock()
|
||||
}
|
||||
}
|
||||
+34
@@ -71,6 +71,40 @@ class ConcurrentCollectionsTest {
|
||||
assertEquals(4, m["k"])
|
||||
}
|
||||
|
||||
@Test
|
||||
fun mapRemoveReturnsOldValueAndDeletes() {
|
||||
val map = ConcurrentMap<String, Int>()
|
||||
map["a"] = 1
|
||||
map["b"] = 2
|
||||
|
||||
assertEquals(1, map.remove("a"))
|
||||
assertNull(map["a"])
|
||||
assertEquals(1, map.size())
|
||||
assertEquals(2, map["b"])
|
||||
}
|
||||
|
||||
@Test
|
||||
fun mapRemoveAbsentKeyIsNullAndNoOp() {
|
||||
val map = ConcurrentMap<String, Int>()
|
||||
map["b"] = 2
|
||||
|
||||
assertNull(map.remove("missing"))
|
||||
assertEquals(1, map.size())
|
||||
assertEquals(2, map["b"])
|
||||
}
|
||||
|
||||
@Test
|
||||
fun mapRemoveThenGetOrPutRecreates() {
|
||||
// The lifecycle PoolRequests needs: connecting()/disconnected() drop a relay's
|
||||
// wire state, and the next REQ re-creates it. A stale value must never survive.
|
||||
val map = ConcurrentMap<String, Int>()
|
||||
map.getOrPut("relay") { 1 }
|
||||
map.remove("relay")
|
||||
|
||||
assertEquals(9, map.getOrPut("relay") { 9 })
|
||||
assertEquals(9, map["relay"])
|
||||
}
|
||||
|
||||
@Test
|
||||
fun mapSnapshotIsDetached() {
|
||||
val m = ConcurrentMap<String, Int>()
|
||||
|
||||
+2
@@ -49,6 +49,8 @@ actual class ConcurrentMap<K : Any, V : Any> {
|
||||
remap: (old: V, new: V) -> V,
|
||||
): V = map.merge(key, value) { old, new -> remap(old, new) }!!
|
||||
|
||||
actual fun remove(key: K): V? = map.remove(key)
|
||||
|
||||
actual fun size(): Int = map.size
|
||||
|
||||
actual fun snapshot(): Map<K, V> = HashMap(map)
|
||||
|
||||
+35
@@ -0,0 +1,35 @@
|
||||
/*
|
||||
* 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.concurrent
|
||||
|
||||
import java.util.concurrent.locks.ReentrantLock
|
||||
|
||||
// ReentrantLock parks contended waiters through AbstractQueuedSynchronizer, so a
|
||||
// thread waiting on a holder that lost its core (GC, preemption) costs one context
|
||||
// switch rather than a spinning core. Non-fair on purpose: fairness would add a
|
||||
// handoff per acquisition and the critical sections here are microseconds long.
|
||||
actual class PlatformLock {
|
||||
private val lock = ReentrantLock()
|
||||
|
||||
actual fun lock() = lock.lock()
|
||||
|
||||
actual fun unlock() = lock.unlock()
|
||||
}
|
||||
+316
@@ -0,0 +1,316 @@
|
||||
/*
|
||||
* 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.nip01Core.relay.prodbench
|
||||
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.RequestSubscriptionState
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
|
||||
import kotlinx.coroutines.Dispatchers
|
||||
import kotlinx.coroutines.launch
|
||||
import kotlinx.coroutines.runBlocking
|
||||
import kotlinx.coroutines.sync.Mutex
|
||||
import kotlinx.coroutines.sync.withLock
|
||||
import org.junit.Test
|
||||
import java.util.concurrent.ConcurrentHashMap
|
||||
import java.util.concurrent.atomic.AtomicBoolean
|
||||
import java.util.concurrent.atomic.AtomicLong
|
||||
import java.util.concurrent.locks.ReentrantLock
|
||||
|
||||
/**
|
||||
* Compares the three candidate designs for `PoolRequests`' subscription-state lock,
|
||||
* under the real production topology: R relay-consumer coroutines on a
|
||||
* limited-parallelism dispatcher (`Dispatchers.IO` = 64) all delivering EVENTs for a
|
||||
* small number of subscription ids.
|
||||
*
|
||||
* - **PER_SUB_BLOCKING** — today: one blocking lock per subId, shared by every relay
|
||||
* that sub runs on. Waiters park, which fixed the CPU burn, but a parked waiter
|
||||
* still OCCUPIES its dispatcher thread.
|
||||
* - **PER_SUB_MUTEX** — kotlinx `Mutex`: a waiter *suspends* and releases its thread.
|
||||
* Requires making the whole listener chain `suspend` (262 overrides, 110 call sites).
|
||||
* - **STRIPED** — one lock per (subId, relay). Every critical section in PoolRequests
|
||||
* is already scoped to a single relay, so this removes the contention outright and
|
||||
* needs no API change.
|
||||
*
|
||||
* The metric that matters is NOT lock throughput — it is whether unrelated work can
|
||||
* still get a thread while the lock is contended. `bystanderLatency` models that: a
|
||||
* task that never touches the lock, submitted to the same dispatcher.
|
||||
*/
|
||||
class LockDesignComparisonBenchmark {
|
||||
private val relays = 191
|
||||
private val dispatcherThreads = 64
|
||||
private val durationMs = 3000L
|
||||
|
||||
/** One relay's slice of a subscription's state — all real fields are relay-keyed. */
|
||||
private class PerRelay {
|
||||
val lock = ReentrantLock()
|
||||
var status: Int = 0
|
||||
var filters: List<String>? = null
|
||||
var lastKnown: List<String>? = null
|
||||
}
|
||||
|
||||
private class Striped {
|
||||
val perRelay = ConcurrentHashMap<Int, PerRelay>()
|
||||
|
||||
inline fun <R> withLock(
|
||||
relay: Int,
|
||||
block: (PerRelay) -> R,
|
||||
): R {
|
||||
val s = perRelay.computeIfAbsent(relay) { PerRelay() }
|
||||
s.lock.lock()
|
||||
try {
|
||||
return block(s)
|
||||
} finally {
|
||||
s.lock.unlock()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private class PerSubBlocking {
|
||||
val lock = ReentrantLock()
|
||||
val status = HashMap<Int, Int>()
|
||||
val filters = HashMap<Int, List<String>>()
|
||||
}
|
||||
|
||||
private class PerSubMutex {
|
||||
val mutex = Mutex()
|
||||
val status = HashMap<Int, Int>()
|
||||
val filters = HashMap<Int, List<String>>()
|
||||
}
|
||||
|
||||
private fun report(
|
||||
name: String,
|
||||
subs: Int,
|
||||
ops: Long,
|
||||
bystanderSamples: List<Long>,
|
||||
) {
|
||||
val sorted = bystanderSamples.sorted()
|
||||
val p50 = sorted[sorted.size / 2] / 1000.0
|
||||
val p99 = sorted[(sorted.size * 99) / 100] / 1000.0
|
||||
val max = sorted.last() / 1000.0
|
||||
println(
|
||||
"%-18s subs=%-3d ops/s=%,10d bystander p50=%8.1fus p99=%9.1fus max=%9.1fus (n=%d)".format(
|
||||
name,
|
||||
subs,
|
||||
ops * 1000 / durationMs,
|
||||
p50,
|
||||
p99,
|
||||
max,
|
||||
sorted.size,
|
||||
),
|
||||
)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun compareDesigns() {
|
||||
if (System.getenv("PROD_RELAY_BENCH") == null && System.getProperty("prodRelayBench") == null) {
|
||||
println("compareDesigns skipped. Run with -PprodRelayBench=1 to enable.")
|
||||
return
|
||||
}
|
||||
println("relays=$relays dispatcherThreads=$dispatcherThreads window=${durationMs}ms")
|
||||
println("bystander = a task that NEVER touches the lock, on the same dispatcher.")
|
||||
println("Lower bystander latency = the lock is not stealing dispatcher threads.\n")
|
||||
for (subs in listOf(1, 4, 16)) {
|
||||
runPerSubBlocking(subs)
|
||||
runPerSubMutex(subs)
|
||||
runStriped(subs)
|
||||
runRealStriped(subs)
|
||||
println()
|
||||
}
|
||||
}
|
||||
|
||||
private fun runPerSubBlocking(subs: Int) =
|
||||
runBlocking {
|
||||
@Suppress("DEPRECATION")
|
||||
val dispatcher = Dispatchers.IO.limitedParallelism(dispatcherThreads)
|
||||
val states = Array(subs) { PerSubBlocking() }
|
||||
val stop = AtomicBoolean(false)
|
||||
val ops = AtomicLong(0)
|
||||
val bystander = ArrayList<Long>()
|
||||
val jobs =
|
||||
(0 until relays).map { relay ->
|
||||
launch(dispatcher) {
|
||||
var n = 0L
|
||||
while (!stop.get()) {
|
||||
val s = states[relay % subs]
|
||||
s.lock.lock()
|
||||
try {
|
||||
s.status[relay] = 1
|
||||
s.filters[relay] = SAMPLE
|
||||
s.status[relay]
|
||||
} finally {
|
||||
s.lock.unlock()
|
||||
}
|
||||
n++
|
||||
// Models `for (message in incomingMessages)`: every real
|
||||
// iteration suspends, letting the dispatcher multiplex.
|
||||
kotlinx.coroutines.yield()
|
||||
}
|
||||
ops.addAndGet(n)
|
||||
}
|
||||
}
|
||||
val by =
|
||||
launch(dispatcher) {
|
||||
while (!stop.get()) {
|
||||
val t = System.nanoTime()
|
||||
kotlinx.coroutines.yield()
|
||||
bystander.add(System.nanoTime() - t)
|
||||
}
|
||||
}
|
||||
Thread.sleep(durationMs)
|
||||
stop.set(true)
|
||||
jobs.forEach { it.join() }
|
||||
by.join()
|
||||
report("PER_SUB_BLOCKING", subs, ops.get(), bystander.ifEmpty { listOf(0L) })
|
||||
}
|
||||
|
||||
private fun runPerSubMutex(subs: Int) =
|
||||
runBlocking {
|
||||
@Suppress("DEPRECATION")
|
||||
val dispatcher = Dispatchers.IO.limitedParallelism(dispatcherThreads)
|
||||
val states = Array(subs) { PerSubMutex() }
|
||||
val stop = AtomicBoolean(false)
|
||||
val ops = AtomicLong(0)
|
||||
val bystander = ArrayList<Long>()
|
||||
val jobs =
|
||||
(0 until relays).map { relay ->
|
||||
launch(dispatcher) {
|
||||
var n = 0L
|
||||
while (!stop.get()) {
|
||||
val s = states[relay % subs]
|
||||
s.mutex.withLock {
|
||||
s.status[relay] = 1
|
||||
s.filters[relay] = SAMPLE
|
||||
s.status[relay]
|
||||
}
|
||||
n++
|
||||
// Models `for (message in incomingMessages)`: every real
|
||||
// iteration suspends, letting the dispatcher multiplex.
|
||||
kotlinx.coroutines.yield()
|
||||
}
|
||||
ops.addAndGet(n)
|
||||
}
|
||||
}
|
||||
val by =
|
||||
launch(dispatcher) {
|
||||
while (!stop.get()) {
|
||||
val t = System.nanoTime()
|
||||
kotlinx.coroutines.yield()
|
||||
bystander.add(System.nanoTime() - t)
|
||||
}
|
||||
}
|
||||
Thread.sleep(durationMs)
|
||||
stop.set(true)
|
||||
jobs.forEach { it.join() }
|
||||
by.join()
|
||||
report("PER_SUB_MUTEX", subs, ops.get(), bystander.ifEmpty { listOf(0L) })
|
||||
}
|
||||
|
||||
private fun runStriped(subs: Int) =
|
||||
runBlocking {
|
||||
@Suppress("DEPRECATION")
|
||||
val dispatcher = Dispatchers.IO.limitedParallelism(dispatcherThreads)
|
||||
val states = Array(subs) { Striped() }
|
||||
val stop = AtomicBoolean(false)
|
||||
val ops = AtomicLong(0)
|
||||
val bystander = ArrayList<Long>()
|
||||
val jobs =
|
||||
(0 until relays).map { relay ->
|
||||
launch(dispatcher) {
|
||||
var n = 0L
|
||||
while (!stop.get()) {
|
||||
states[relay % subs].withLock(relay) { s ->
|
||||
s.status = 1
|
||||
s.filters = SAMPLE
|
||||
s.lastKnown = SAMPLE
|
||||
}
|
||||
n++
|
||||
// Models `for (message in incomingMessages)`: every real
|
||||
// iteration suspends, letting the dispatcher multiplex.
|
||||
kotlinx.coroutines.yield()
|
||||
}
|
||||
ops.addAndGet(n)
|
||||
}
|
||||
}
|
||||
val by =
|
||||
launch(dispatcher) {
|
||||
while (!stop.get()) {
|
||||
val t = System.nanoTime()
|
||||
kotlinx.coroutines.yield()
|
||||
bystander.add(System.nanoTime() - t)
|
||||
}
|
||||
}
|
||||
Thread.sleep(durationMs)
|
||||
stop.set(true)
|
||||
jobs.forEach { it.join() }
|
||||
by.join()
|
||||
report("STRIPED", subs, ops.get(), bystander.ifEmpty { listOf(0L) })
|
||||
}
|
||||
|
||||
/**
|
||||
* Same topology, but driving the REAL shipped [RequestSubscriptionState] (striped per
|
||||
* relay) instead of a prototype — so the measured win is a property of the code that
|
||||
* ships, not of this file.
|
||||
*/
|
||||
private fun runRealStriped(subs: Int) =
|
||||
runBlocking {
|
||||
@Suppress("DEPRECATION")
|
||||
val dispatcher = Dispatchers.IO.limitedParallelism(dispatcherThreads)
|
||||
val states = Array(subs) { RequestSubscriptionState<Int>() }
|
||||
val stop = AtomicBoolean(false)
|
||||
val ops = AtomicLong(0)
|
||||
val bystander = ArrayList<Long>()
|
||||
val filters = listOf(Filter(kinds = listOf(1)))
|
||||
val jobs =
|
||||
(0 until relays).map { relay ->
|
||||
launch(dispatcher) {
|
||||
var n = 0L
|
||||
val state = states[relay % subs]
|
||||
while (!stop.get()) {
|
||||
state.withLock(relay) {
|
||||
state.onNewEvent(relay)
|
||||
state.currentState(relay)
|
||||
state.onOpenReq(relay, filters)
|
||||
state.lastKnownFilterStates(relay)
|
||||
}
|
||||
n++
|
||||
kotlinx.coroutines.yield()
|
||||
}
|
||||
ops.addAndGet(n)
|
||||
}
|
||||
}
|
||||
val by =
|
||||
launch(dispatcher) {
|
||||
while (!stop.get()) {
|
||||
val t = System.nanoTime()
|
||||
kotlinx.coroutines.yield()
|
||||
bystander.add(System.nanoTime() - t)
|
||||
}
|
||||
}
|
||||
Thread.sleep(durationMs)
|
||||
stop.set(true)
|
||||
jobs.forEach { it.join() }
|
||||
by.join()
|
||||
report("REAL_STRIPED", subs, ops.get(), bystander.ifEmpty { listOf(0L) })
|
||||
}
|
||||
|
||||
companion object {
|
||||
private val SAMPLE = listOf("kinds:1", "authors:abc")
|
||||
}
|
||||
}
|
||||
+242
@@ -0,0 +1,242 @@
|
||||
/*
|
||||
* 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.nip01Core.relay.prodbench
|
||||
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.RequestSubscriptionState
|
||||
import org.junit.Assert.assertTrue
|
||||
import org.junit.Test
|
||||
import java.util.concurrent.CountDownLatch
|
||||
import java.util.concurrent.atomic.AtomicBoolean
|
||||
import java.util.concurrent.atomic.AtomicLong
|
||||
import kotlin.concurrent.thread
|
||||
|
||||
/**
|
||||
* Reproduces the production ANR seen on a Pixel 8 (2026-08-03, anr_2026-08-03-12-55-26-256):
|
||||
* 191 live relay sockets feed EVENT frames for the same handful of subscription ids, so
|
||||
* dozens of DefaultDispatcher workers pile onto ONE [RequestSubscriptionState] busy-wait
|
||||
* lock. In that trace 51 of 52 runnable workers sat in the inlined spin loop while the
|
||||
* single lock holder was parked in `WaitingForGcToComplete`.
|
||||
*
|
||||
* Two things are measured:
|
||||
* - [contendedThroughput]: aggregate critical sections/s as the contender count grows past
|
||||
* the core count (the "negative scaling" the 2026-07-02 plan measured with 4 feeders,
|
||||
* re-run at production fan-out).
|
||||
* - [victimLatencyUnderSpin]: what an UNRELATED thread (stand-in for the UI thread) sees
|
||||
* while the spinners run — this is the ANR mechanism, not the lock throughput.
|
||||
*/
|
||||
class SpinLockConvoyBenchmark {
|
||||
private val cores = Runtime.getRuntime().availableProcessors()
|
||||
|
||||
/**
|
||||
* Every waiter targets ONE reference on purpose. The lock is striped per relay, so
|
||||
* spreading threads over distinct relays would put them on different stripes and
|
||||
* this benchmark would measure nothing — the point here is the primitive's behaviour
|
||||
* when contention DOES land on a single stripe.
|
||||
*/
|
||||
private val hotRelay = 0
|
||||
|
||||
/** Mirrors the real critical section: a handful of map reads/writes, no I/O. */
|
||||
private fun criticalSection(
|
||||
state: RequestSubscriptionState<Int>,
|
||||
relay: Int,
|
||||
) {
|
||||
state.onNewEvent(relay)
|
||||
state.currentState(relay)
|
||||
state.lastKnownFilterStates(relay)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun contendedThroughput() {
|
||||
if (System.getenv("PROD_RELAY_BENCH") == null && System.getProperty("prodRelayBench") == null) {
|
||||
println("contendedThroughput skipped. Run with -PprodRelayBench=1 to enable.")
|
||||
return
|
||||
}
|
||||
println("cores = $cores")
|
||||
println("threads | ops/s | vs 1 thread")
|
||||
var baseline = 0.0
|
||||
for (threads in listOf(1, 2, 4, 8, 16, 32, 52)) {
|
||||
val state = RequestSubscriptionState<Int>()
|
||||
val stop = AtomicBoolean(false)
|
||||
val ops = AtomicLong(0)
|
||||
val start = CountDownLatch(1)
|
||||
val workers =
|
||||
(0 until threads).map { id ->
|
||||
thread {
|
||||
start.await()
|
||||
var local = 0L
|
||||
while (!stop.get()) {
|
||||
state.withLock(hotRelay) { criticalSection(state, hotRelay) }
|
||||
local++
|
||||
}
|
||||
ops.addAndGet(local)
|
||||
}
|
||||
}
|
||||
val t0 = System.nanoTime()
|
||||
start.countDown()
|
||||
Thread.sleep(2000)
|
||||
stop.set(true)
|
||||
workers.forEach { it.join() }
|
||||
val secs = (System.nanoTime() - t0) / 1e9
|
||||
val rate = ops.get() / secs
|
||||
if (threads == 1) baseline = rate
|
||||
println("%7d | %9.0f | %.2fx".format(threads, rate, rate / baseline))
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* REGRESSION GUARD: contended waiters on [RequestSubscriptionState.withLock] must PARK,
|
||||
* never busy-wait. Fails if the lock is ever turned back into a spin lock.
|
||||
*
|
||||
* This is the signature that identified the production ANR: in
|
||||
* `anr_2026-08-03-12-55-26-256`, 37 of 52 runnable workers sat at one obfuscated line of
|
||||
* `PoolRequests.onIncomingMessage` and 12 more at one line of `syncState$lambda$0` — all
|
||||
* `state=R`, `sCount=0`, burning 596% CPU while the lock holder was stuck in
|
||||
* `WaitingForGcToComplete`. With a spin lock this test observes ~40/40 waiters RUNNABLE;
|
||||
* with a parking lock it observes 0.
|
||||
*/
|
||||
@Test
|
||||
fun contendedWaitersParkInsteadOfSpinning() {
|
||||
val spinners = 40
|
||||
val state = RequestSubscriptionState<Int>()
|
||||
val stop = AtomicBoolean(false)
|
||||
val start = CountDownLatch(1)
|
||||
|
||||
val holder =
|
||||
thread(name = "holder") {
|
||||
start.await()
|
||||
while (!stop.get()) {
|
||||
state.withLock(hotRelay) {
|
||||
val t = System.nanoTime()
|
||||
while (System.nanoTime() - t < 5_000_000) { /* hold 5ms */ }
|
||||
}
|
||||
}
|
||||
}
|
||||
val threads =
|
||||
(0 until spinners).map { id ->
|
||||
thread(name = "spinner-$id") {
|
||||
start.await()
|
||||
while (!stop.get()) {
|
||||
state.withLock(hotRelay) { criticalSection(state, hotRelay) }
|
||||
}
|
||||
}
|
||||
}
|
||||
start.countDown()
|
||||
Thread.sleep(300)
|
||||
|
||||
// Sample every spinner's stack, exactly like an ANR dump would.
|
||||
val points = HashMap<String, Int>()
|
||||
var runnable = 0
|
||||
threads.forEach { t ->
|
||||
if (t.state == Thread.State.RUNNABLE) runnable++
|
||||
val top = t.stackTrace.firstOrNull { it.className.contains("SpinLockConvoyBenchmark") || it.className.contains("RequestSubscriptionState") }
|
||||
if (top != null) {
|
||||
val key = "${top.className.substringAfterLast('.')}.${top.methodName}:${top.lineNumber}"
|
||||
points[key] = (points[key] ?: 0) + 1
|
||||
}
|
||||
}
|
||||
stop.set(true)
|
||||
threads.forEach { it.join() }
|
||||
holder.join()
|
||||
|
||||
println("sampled $spinners waiters: RUNNABLE=$runnable, distinct program points=${points.size}")
|
||||
points.entries.sortedByDescending { it.value }.forEach { println(" ${it.value}x ${it.key}") }
|
||||
|
||||
// A parked waiter reports WAITING, so the parking lock measures ~0 here while a
|
||||
// spin lock measures ~100%. The line sits at 50% deliberately: it separates the
|
||||
// two cases by a mile and leaves headroom on a loaded CI box, where a few waiters
|
||||
// can legitimately be mid-acquire when we sample.
|
||||
assertTrue(
|
||||
"Expected contended waiters to park, but $runnable/$spinners were RUNNABLE — " +
|
||||
"RequestSubscriptionState.withLock looks like it is busy-waiting again. " +
|
||||
"See PlatformLock's kdoc: this is what caused anr_2026-08-03-12-55-26-256.",
|
||||
runnable <= spinners / 2,
|
||||
)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun victimLatencyUnderSpin() {
|
||||
if (System.getenv("PROD_RELAY_BENCH") == null && System.getProperty("prodRelayBench") == null) {
|
||||
println("victimLatencyUnderSpin skipped. Run with -PprodRelayBench=1 to enable.")
|
||||
return
|
||||
}
|
||||
// One holder that briefly stalls inside the critical section (in production: a GC
|
||||
// pause, or simply being descheduled). Everyone else spins.
|
||||
val spinners = 52
|
||||
val state = RequestSubscriptionState<Int>()
|
||||
val stop = AtomicBoolean(false)
|
||||
|
||||
fun measureVictim(label: String) {
|
||||
// The "UI thread": allocates and measures its own scheduling latency.
|
||||
val samples = ArrayList<Long>()
|
||||
repeat(200) {
|
||||
val t = System.nanoTime()
|
||||
// trivial allocation work, like a Compose semantics traversal step
|
||||
val junk = ArrayList<String>(64)
|
||||
repeat(64) { i -> junk.add("node$i") }
|
||||
Thread.yield()
|
||||
samples.add(System.nanoTime() - t)
|
||||
}
|
||||
samples.sort()
|
||||
println(
|
||||
"%-22s p50=%6.0fus p90=%7.0fus p99=%8.0fus max=%8.0fus".format(
|
||||
label,
|
||||
samples[samples.size / 2] / 1000.0,
|
||||
samples[(samples.size * 90) / 100] / 1000.0,
|
||||
samples[(samples.size * 99) / 100] / 1000.0,
|
||||
samples.last() / 1000.0,
|
||||
),
|
||||
)
|
||||
}
|
||||
|
||||
measureVictim("idle (no spinners)")
|
||||
|
||||
val start = CountDownLatch(1)
|
||||
val holder =
|
||||
thread {
|
||||
start.await()
|
||||
while (!stop.get()) {
|
||||
state.withLock(hotRelay) {
|
||||
// Simulate the holder losing its core / waiting on GC mid-section.
|
||||
val t = System.nanoTime()
|
||||
while (System.nanoTime() - t < 2_000_000) { /* 2ms stall */ }
|
||||
}
|
||||
Thread.sleep(1)
|
||||
}
|
||||
}
|
||||
val threads =
|
||||
(0 until spinners).map { id ->
|
||||
thread {
|
||||
start.await()
|
||||
while (!stop.get()) {
|
||||
state.withLock(hotRelay) { criticalSection(state, hotRelay) }
|
||||
}
|
||||
}
|
||||
}
|
||||
start.countDown()
|
||||
Thread.sleep(500)
|
||||
measureVictim("$spinners spinners")
|
||||
stop.set(true)
|
||||
threads.forEach { it.join() }
|
||||
holder.join()
|
||||
|
||||
measureVictim("after (no spinners)")
|
||||
}
|
||||
}
|
||||
+44
@@ -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.quartz.utils.concurrent
|
||||
|
||||
import kotlin.concurrent.atomics.AtomicBoolean
|
||||
import kotlin.concurrent.atomics.ExperimentalAtomicApi
|
||||
|
||||
// Linux/JVM-less target: Kotlin/Native's stdlib ships no parking lock and there is
|
||||
// no Foundation here, so this keeps a test-and-test-and-set spin. Correct but not
|
||||
// scalable. Acceptable ONLY because linuxX64 is a build/CI target for quartz, not a
|
||||
// host for the many-relay client workload whose contention motivated the parking
|
||||
// actuals on jvmAndroid and Apple. If that ever changes, swap in a pthread mutex.
|
||||
@OptIn(ExperimentalAtomicApi::class)
|
||||
actual class PlatformLock {
|
||||
private val held = AtomicBoolean(false)
|
||||
|
||||
actual fun lock() {
|
||||
while (held.exchange(true)) {
|
||||
while (held.load()) { }
|
||||
}
|
||||
}
|
||||
|
||||
actual fun unlock() {
|
||||
held.store(false)
|
||||
}
|
||||
}
|
||||
+10
@@ -74,6 +74,16 @@ actual class ConcurrentMap<K : Any, V : Any> {
|
||||
}
|
||||
}
|
||||
|
||||
actual fun remove(key: K): V? {
|
||||
while (true) {
|
||||
val cur = ref.load()
|
||||
val old = cur[key] ?: return null
|
||||
val copy = HashMap(cur)
|
||||
copy.remove(key)
|
||||
if (ref.compareAndSet(cur, copy)) return old
|
||||
}
|
||||
}
|
||||
|
||||
actual fun size(): Int = ref.load().size
|
||||
|
||||
actual fun snapshot(): Map<K, V> = HashMap(ref.load())
|
||||
|
||||
Reference in New Issue
Block a user