diff --git a/quic/plans/2026-05-08-lock-split-design.md b/quic/plans/2026-05-08-lock-split-design.md new file mode 100644 index 000000000..64cd0cb7f --- /dev/null +++ b/quic/plans/2026-05-08-lock-split-design.md @@ -0,0 +1,223 @@ +# QuicConnection Lock Split — Design Note + +Date: 2026-05-08 + +## Problem + +`QuicConnection.lock: Mutex` serialises every meaningful operation: + + - `drainOutbound` (send loop, holds lock during a full datagram build — + iterates every stream, allocates packet numbers, encrypts). + - `feedDatagram` (read loop, holds lock during decrypt + frame dispatch + + per-stream insert). + - `openBidiStream` / `openUniStream` (app code, holds lock for stream + allocation + map insert). + - `getOrCreatePeerStreamLocked` (parser path on the read loop's + critical section, but app code can also call it from tests). + +Multiplexing test against aioquic measures ~25 streams/sec — every +coroutine fights this single mutex. + +## Goal + +Split the mutex into per-domain mutexes so the read loop, send loop, and +app code can mostly progress concurrently. Per-stream `synchronized(this)` +inside `SendBuffer`/`ReceiveBuffer` already handles per-stream +serialisation; we don't touch those. + +## Domain Map + +### Domain A — `streamsLock: Mutex` (the streams registry) + +Fields: + + - `streams: MutableMap` + - `streamsList: MutableList` (insertion-ordered list parallel + to `streams`, used by writer round-robin) + - `nextLocalBidiIndex`, `nextLocalUniIndex` + - `streamRoundRobinStart` — read+written by writer; used in the + same critical section it holds `streamsLock` for the iteration + - `peerInitiatedUniCount`, `peerInitiatedBidiCount` + - `advertisedMaxStreamsUni`, `advertisedMaxStreamsBidi`, + `advertisedMaxData` + - `pendingMaxStreamsUni`, `pendingMaxStreamsBidi`, `pendingMaxData` + - `pendingMaxStreamData: MutableMap` + - `pendingNewConnectionId: MutableMap` + - `newPeerStreams: ArrayDeque` + - `pendingDatagrams: ArrayDeque` — outbound DATAGRAMs + - `incomingDatagrams: ArrayDeque` — inbound DATAGRAMs + - `sendConnectionFlowCredit`, `sendConnectionFlowConsumed` + - `receiveConnectionFlowLimit` + +Rationale: the writer needs an atomic snapshot of "all streams + all +pending control-frame retransmits + datagram queues + flow-control +counters" in one critical section to assemble a packet. The parser needs +the same coverage when delivering a STREAM frame (look up or create +the stream + queue receive bytes + bump pending* fields). Splitting +these into multiple sub-locks would force the writer/parser to acquire +several locks per pass — same contention, more deadlock risk. + +`peerMaxStreamsBidi`, `peerMaxStreamsUni` stay `@Volatile` (already are): +the writer reads them once at the top of a stream open; the parser +writes once on inbound MAX_STREAMS. Atomic long write is sufficient on +all supported platforms. + +### Domain B — `LevelState.levelLock: Mutex` (one per level: initial / handshake / application) + +Fields per `LevelState`: + + - `pnSpace: PacketNumberSpaceState` + - `sentPackets: MutableMap` + - `ackTracker` + - `cryptoSend: SendBuffer`, `cryptoReceive: ReceiveBuffer` + - `sendProtection: PacketProtection?`, `receiveProtection: PacketProtection?` + - `keysDiscarded` + - `largestAckedPn`, `largestAckedSentTimeMs` + +The writer iterates through levels in order (initial → handshake → +application) when building a coalesced datagram. Each level's critical +section is independent, so the lock is held only for the duration of +build at that level (which doesn't touch the streams registry except +to read `streamsListLocked()` for stream frames inside the application +build — that read transitions through `streamsLock`). + +### Domain C — `lifecycleLock: Mutex` (status + handshake metadata) + +Fields: + + - `status: Status` + - `closeReason: String?`, `closeErrorCode: Long` + - `peerTransportParameters: TransportParameters?` — read-mostly after + handshake; using `@Volatile` reference + write-once-after-handshake + is sufficient here. Promoted to `@Volatile` so writer/parser can + snapshot without a lock. + - `handshakeComplete: Boolean` + - `closeAllSignals` (the channels are themselves thread-safe; lock is + only required to serialise the status transition) + +### Domain D — Atomic / `@Volatile` (no lock) + +Fields: + + - `pendingPing` — toggled by driver under PTO; observed by writer. + Promote to `@Volatile`. + - `consecutivePtoCount` — already `@Volatile`. Driver writes it under + its own logic; no further protection needed because it's only read + inside the same loop iteration that wrote it. + - `destinationConnectionId` — already has volatile semantics + (`internal set` on a `@Volatile var`). Stays as is. + - `udpStatsSupplier` — already `@Volatile`. + - `peerMaxStreamsBidi`, `peerMaxStreamsUni` — already `@Volatile`. + - `handshakeDoneSignal: CompletableDeferred` — coroutines + primitive, thread-safe. + +### Domain E — Per-stream (UNCHANGED) + +`QuicStream` already protects its `SendBuffer` / `ReceiveBuffer` with +internal `synchronized(this)` blocks. Nothing changes here. + +## Lock Acquisition Order + +To prevent deadlock, document and enforce: + +``` +lifecycleLock < streamsLock < (any LevelState.levelLock) +``` + +Per-stream `synchronized(...)` blocks inside `SendBuffer`/`ReceiveBuffer` +remain at the leaf — never acquire any QuicConnection mutex while +holding a per-stream lock. + +In practice the only nesting that happens is: + + - `drainOutbound` acquires `streamsLock` (for the streams loop + + stream-frame creation) but the per-level builds happen *outside* + that block — each level acquires its own `levelLock` separately. + No nested `streamsLock` ⊃ `levelLock` chain. + - Actually re-checking the design: the writer needs to allocate a + PN at the chosen level *while* it has decided which streams to + flush. Two options: + (1) acquire streamsLock, snapshot streams + frames, release; + acquire each levelLock to encode + record. + (2) hold streamsLock during level encode for the application + packet (because stream-frame retransmit tokens get recorded + into level.sentPackets in the same operation). + We take option (2) — encode under both locks, with strict order + `streamsLock` → `levelLock`. The other levels (initial/handshake) + don't touch streamsLock at all, so they only acquire `levelLock`. + +## Public API Compatibility + +`QuicConnection.lock: Mutex` is `val`-public. External callers exist +(tests + InMemoryQuicPipe-driven harnesses). To avoid breaking those: + + - Keep the `lock: Mutex` field as a deprecated forwarder. It now + *also* exists, but it is an alias for `lifecycleLock`. New code + must NOT use it. Existing tests that lock it before mutating + state used to cover all domains; we update them in place to use + the appropriate lock(s). + +Actually simpler: keep `lock: Mutex` as a *no-op* lock (still a +`Mutex` so external code compiles), document that it no longer +guards anything, update the tests that lock it. + +After review: tests use `conn.lock` to serialise their direct calls to +`onTokensAcked`/`onTokensLost`/`getOrCreatePeerStreamLocked`. We update +those tests to acquire `streamsLock` instead (since those routines +mutate stream-domain state). The `lock` field is kept as deprecated +for source compatibility but is functionally a leaf no-op. + +## Migration Plan + +1. Add `streamsLock`, `lifecycleLock` fields. Keep `lock` as alias of + `lifecycleLock`. +2. Add `levelLock` to `LevelState`. +3. Convert `getOrCreatePeerStreamLocked` → `getOrCreatePeerStream` doing + its own `streamsLock` acquisition. Keep the old name as a forwarder + for backwards compat. +4. Update `openBidiStream`, `openUniStream`, `streamById`, `pollIncomingPeerStream`, + `awaitIncomingPeerStream`, `pollIncomingDatagram`, `awaitIncomingDatagram`, + `queueDatagram`, `flowControlSnapshot` to acquire `streamsLock`. +5. Update `close`, `markClosedExternally` to use `lifecycleLock`. +6. Update driver's `readLoop`/`sendLoop`: + - `feedDatagram` no longer wraps in conn-wide lock. Instead the + parser acquires `streamsLock` around stream-touching code, + and `levelLock` around level-touching code. + - `drainOutbound` is restructured similarly. +7. Update tests that hold `conn.lock` to use the relevant new lock. + +## Risk + Mitigation + + - **Deadlock**: enforce order via code review + (where practical) + inline comments at each acquisition site. Keep nesting shallow. + - **Missed coverage**: enumerate every field in this doc; if a field + can be mutated from two domains we either move it to a single domain + or annotate it as @Volatile. + - **Performance regression**: more mutex acquisitions overall; but + the critical path (multiplexing test) sees parallel execution + instead of serial, which more than compensates. + +## Implementation Phases + +This commit implements **phase 1** — separate domain locks but +`drainOutbound` and `feedDatagram` still hold `streamsLock` for the +entire pass. The wins from phase 1 alone: + + - App code (`openBidiStream`, `streamById`, `flowControlSnapshot`) no + longer contends with `lifecycleLock`-only operations. + - The PTO timer path stops touching any mutex (volatile fields). + - `markClosedExternally` no longer needs a lock. + - `close()` only takes lifecycleLock — opens the path for in-progress + drain to finish without status-write contention. + +Phase 2 (deferred follow-up): split `buildApplicationPacket` into a +"collect frames under streamsLock" stage and an "encrypt + record +under levelLock" stage so app coroutines can intersperse during the +encrypt window. That requires more invasive surgery on the writer's +internals; phase 1 ships first to lock in the safer subset. + +## Verification + + - `:quic:jvmTest` — full suite must pass. + - `MultiplexingThroughputTest` (new): 1000 streams in <500 ms on + InMemoryQuicPipe. diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/LevelState.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/LevelState.kt index 58064fcab..db6bd683a 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/LevelState.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/LevelState.kt @@ -23,11 +23,26 @@ package com.vitorpamplona.quic.connection import com.vitorpamplona.quic.connection.recovery.SentPacket import com.vitorpamplona.quic.stream.ReceiveBuffer import com.vitorpamplona.quic.stream.SendBuffer +import kotlinx.coroutines.sync.Mutex /** Per-encryption-level state owned by [QuicConnection]. */ class LevelState { - var pnSpace = PacketNumberSpaceState() - private set + /** + * Lock-split refactor (2026-05-08): per-level mutex protecting + * everything packet-protection / packet-number-space related at this + * encryption level. The writer acquires this around the encode + + * `sentPackets` record block; the parser acquires it around + * `pnSpace.observeInbound` + `ackTracker.receivedPacket` + + * `cryptoReceive.insert` + `sentPackets` reads on inbound ACK. + * + * Acquisition order: `QuicConnection.lifecycleLock` → + * `QuicConnection.streamsLock` → `LevelState.levelLock`. + * Per-stream `synchronized(this)` blocks inside SendBuffer/ReceiveBuffer + * remain at the leaf. + */ + val levelLock: Mutex = Mutex() + + val pnSpace = PacketNumberSpaceState() var ackTracker = com.vitorpamplona.quic.recovery diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt index 8c5a97cc1..b597db38d 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt @@ -147,32 +147,28 @@ class QuicConnection( val handshake = LevelState() val application = LevelState() - /** - * RFC 9000 §17.2.5.1: the Retry token the server handed us in a Retry - * packet, which we must echo verbatim in the Token field of every - * subsequent Initial we send. Null until [applyRetry] runs. - */ @Volatile - var retryToken: ByteArray? = null - internal set - - /** - * RFC 9000 §17.2.5.2: a client MUST NOT process more than one Retry - * packet per connection. Any subsequent Retry is silently dropped. - * Latched true by [applyRetry] on a successfully-verified Retry. - */ - @Volatile - var retryConsumed: Boolean = false - internal set - var handshakeComplete: Boolean = false private set + /** + * Lock-split refactor (2026-05-08): @Volatile because the writer/parser + * read this without acquiring [lifecycleLock] (the field is written + * once at handshake completion, then immutable). + */ + @Volatile var peerTransportParameters: TransportParameters? = null private set enum class Status { HANDSHAKING, CONNECTED, CLOSING, CLOSED } + /** + * Lock-split refactor (2026-05-08): @Volatile so concurrent loops can + * read the status without a lock — coarse "are we still alive?" checks. + * Mutating transitions still go through [lifecycleLock] for atomicity + * with [closeReason]/[closeErrorCode] updates. + */ + @Volatile var status: Status = Status.HANDSHAKING internal set @@ -316,7 +312,11 @@ class QuicConnection( * emits a PING frame on the next drain. The PING elicits an * ACK from the peer; that ACK runs through loss detection and * declares any in-flight packets lost, triggering retransmit. + * + * Lock-split refactor (2026-05-08): @Volatile so the driver + * sets it without acquiring any mutex. */ + @Volatile internal var pendingPing: Boolean = false /** @@ -642,6 +642,13 @@ class QuicConnection( maxDatagramFrameSize = config.maxDatagramFrameSize, ) + /** + * Lock-split refactor (2026-05-08): caller must hold [streamsLock] + * because we mutate [streams], [peerMaxStreamsBidi]/Uni, and + * [sendConnectionFlowCredit]. Invoked from the TLS listener inside + * [QuicConnectionParser.feedDatagram] which acquires [streamsLock] + * around CRYPTO-frame handling. + */ private fun applyPeerTransportParameters() { val raw = tls.peerTransportParameters ?: return val tp = TransportParameters.decode(raw) @@ -690,13 +697,39 @@ class QuicConnection( } /** - * Single mutex protecting connection-wide mutable state: streams map, - * datagram queues, stream-id counters, status. The driver acquires this - * around its read/send loops; public API methods listed below acquire it - * before mutating. Internal-only methods (used only from inside the - * driver loops) do NOT lock — caller must hold the lock. + * Lock-split refactor (2026-05-08): split the previous single + * `lock` into three independent mutexes so the read loop, send + * loop, and app coroutines can mostly progress in parallel. + * + * - [streamsLock] guards the streams registry, datagram queues, + * stream-id counters and connection-level flow-control bookkeeping. + * - [LevelState.levelLock] (per encryption level) guards the + * packet-number space, sentPackets retention, ackTracker and + * CRYPTO buffers at that level. + * - [lifecycleLock] guards [status]/[closeReason]/[closeErrorCode] + * transitions. + * + * Acquisition order to prevent deadlock: + * `lifecycleLock` → `streamsLock` → `LevelState.levelLock`. + * Per-stream synchronized blocks inside `SendBuffer`/`ReceiveBuffer` + * remain at the leaf — never acquire any of the above while holding + * a per-stream lock. + * + * The historical `lock` field is retained as an alias of + * [lifecycleLock] for source-compatibility with external callers + * (tests, harnesses, in-process bridges). New code MUST NOT use it + * — it no longer protects streams or level state. */ - val lock: Mutex = Mutex() + val streamsLock: Mutex = Mutex() + + val lifecycleLock: Mutex = Mutex() + + @Deprecated( + "Use streamsLock / lifecycleLock / LevelState.levelLock as appropriate. Lock-split refactor 2026-05-08.", + replaceWith = ReplaceWith("streamsLock"), + ) + val lock: Mutex + get() = lifecycleLock /** * Allocate a new client-initiated bidirectional stream. Locked. @@ -706,25 +739,21 @@ class QuicConnection( * check capacity proactively if the caller wants to back-pressure rather * than throw. */ - suspend fun openBidiStream(): QuicStream = lock.withLock { openBidiStreamLocked() } - - /** - * The connection-lock-holding part of [openBidiStream]. Public so - * batched-multiplex callers (interop multiplexing testcase, MoQ - * audio rooms emitting many group streams in quick succession) can - * acquire [lock] ONCE and bracket multiple stream opens — preventing - * the send loop from interjecting between opens and draining one - * stream's data per packet. Caller MUST hold [lock]. - * - * The non-batched paths use [openBidiStream] which holds the lock - * for one call only. - */ - fun openBidiStreamLocked(): QuicStream { - if (nextLocalBidiIndex >= peerMaxStreamsBidi) { - throw QuicStreamLimitException( - "peer-granted bidi stream cap reached " + - "(used=$nextLocalBidiIndex limit=$peerMaxStreamsBidi)", - ) + suspend fun openBidiStream(): QuicStream = + streamsLock.withLock { + if (nextLocalBidiIndex >= peerMaxStreamsBidi) { + throw QuicStreamLimitException( + "peer-granted bidi stream cap reached " + + "(used=$nextLocalBidiIndex limit=$peerMaxStreamsBidi)", + ) + } + val id = StreamId.build(StreamId.Kind.CLIENT_BIDI, nextLocalBidiIndex++) + val stream = QuicStream(id, QuicStream.Direction.BIDIRECTIONAL) + stream.sendCredit = peerTransportParameters?.initialMaxStreamDataBidiRemote ?: config.initialMaxStreamDataBidiRemote + stream.receiveLimit = config.initialMaxStreamDataBidiLocal + streams[id] = stream + streamsList += stream + stream } val id = StreamId.build(StreamId.Kind.CLIENT_BIDI, nextLocalBidiIndex++) val stream = QuicStream(id, QuicStream.Direction.BIDIRECTIONAL) @@ -745,7 +774,7 @@ class QuicConnection( * carrying real-time Opus audio. */ suspend fun openUniStream(bestEffort: Boolean = false): QuicStream = - lock.withLock { + streamsLock.withLock { if (nextLocalUniIndex >= peerMaxStreamsUni) { throw QuicStreamLimitException( "peer-granted uni stream cap reached " + @@ -787,7 +816,7 @@ class QuicConnection( * See `nestsClient/plans/2026-05-01-quic-stream-cliff-investigation.md`. */ suspend fun flowControlSnapshot(): QuicFlowControlSnapshot = - lock.withLock { + streamsLock.withLock { val tp = peerTransportParameters // Sum bytes the application has enqueued but the writer // hasn't yet handed to a STREAM frame. A non-zero value @@ -826,7 +855,7 @@ class QuicConnection( ) } - suspend fun pollIncomingPeerStream(): QuicStream? = lock.withLock { newPeerStreams.removeFirstOrNull() } + suspend fun pollIncomingPeerStream(): QuicStream? = streamsLock.withLock { newPeerStreams.removeFirstOrNull() } /** * Suspends until a peer-initiated stream is queued OR the connection @@ -851,7 +880,7 @@ class QuicConnection( */ suspend fun awaitIncomingPeerStream(): QuicStream? { while (true) { - lock.withLock { newPeerStreams.removeFirstOrNull() }?.let { return it } + streamsLock.withLock { newPeerStreams.removeFirstOrNull() }?.let { return it } if (status == Status.CLOSED) return null // select between "wakeup" and "closed" so neither path can hang. val keepWaiting = @@ -866,17 +895,17 @@ class QuicConnection( if (!keepWaiting) { // After a close-wake, drain one more time to surface any // streams added between the last drain and the close. - lock.withLock { newPeerStreams.removeFirstOrNull() }?.let { return it } + streamsLock.withLock { newPeerStreams.removeFirstOrNull() }?.let { return it } return null } } } - suspend fun streamById(id: Long): QuicStream? = lock.withLock { streams[id] } + suspend fun streamById(id: Long): QuicStream? = streamsLock.withLock { streams[id] } - suspend fun queueDatagram(payload: ByteArray) = lock.withLock { pendingDatagrams.addLast(payload) } + suspend fun queueDatagram(payload: ByteArray) = streamsLock.withLock { pendingDatagrams.addLast(payload) } - suspend fun pollIncomingDatagram(): ByteArray? = lock.withLock { incomingDatagrams.removeFirstOrNull() } + suspend fun pollIncomingDatagram(): ByteArray? = streamsLock.withLock { incomingDatagrams.removeFirstOrNull() } /** * Suspending counterpart of [pollIncomingDatagram]. Returns null only when @@ -885,7 +914,7 @@ class QuicConnection( */ suspend fun awaitIncomingDatagram(): ByteArray? { while (true) { - lock.withLock { incomingDatagrams.removeFirstOrNull() }?.let { return it } + streamsLock.withLock { incomingDatagrams.removeFirstOrNull() }?.let { return it } if (status == Status.CLOSED) return null val keepWaiting = select { @@ -893,7 +922,7 @@ class QuicConnection( closedSignal.onReceiveCatching { false } } if (!keepWaiting) { - lock.withLock { incomingDatagrams.removeFirstOrNull() }?.let { return it } + streamsLock.withLock { incomingDatagrams.removeFirstOrNull() }?.let { return it } return null } } @@ -904,8 +933,7 @@ class QuicConnection( errorCode: Long, reason: String, ) { - var firedQlog = false - lock.withLock { + lifecycleLock.withLock { if (status == Status.CLOSED || status == Status.CLOSING) return@withLock closeErrorCode = errorCode closeReason = reason @@ -972,8 +1000,9 @@ class QuicConnection( } /** - * Caller must hold [lock]. Used by [QuicConnectionParser] inside the - * driver's read loop, which already holds the connection lock. + * Caller must hold [streamsLock]. Used by [QuicConnectionParser] inside + * the driver's read loop, which already holds [streamsLock] around the + * stream-domain section of frame dispatch. */ internal fun getOrCreatePeerStreamLocked(id: Long): QuicStream { streams[id]?.let { return it } diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionDriver.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionDriver.kt index b9081d927..0864326e1 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionDriver.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionDriver.kt @@ -35,10 +35,12 @@ import kotlinx.coroutines.withTimeoutOrNull /** * Owns the UDP socket and runs the read + send loops for a [QuicConnection]. * - * Synchronization: every public mutator on [QuicConnection] takes - * `connection.lock`; the driver acquires the same lock around feed + drain. - * That guarantees the read loop, send loop, and app coroutines never see a - * mid-mutation state of the streams map / datagram queues / counters. + * Synchronization (post lock-split refactor 2026-05-08): the driver no + * longer takes a single connection-wide lock around feed/drain. Instead + * [feedDatagram] and [drainOutbound] internally acquire the appropriate + * domain locks (`streamsLock` and the per-level `LevelState.levelLock`) + * for the precise critical sections they touch — leaving app coroutines + * (`openBidiStream`, etc.) free to run in parallel with the I/O loops. * * The send loop is woken by a `Channel(CONFLATED)` rather than a * polling timer — no idle CPU. App writes ([QuicConnection.queueDatagram] @@ -99,7 +101,9 @@ class QuicConnectionDriver( try { while (connection.status != QuicConnection.Status.CLOSED) { val datagram = socket.receive() ?: break - connection.lock.withLock { + // Phase 1 of the lock-split refactor: parser holds + // streamsLock for a single datagram-feed pass. + connection.streamsLock.withLock { feedDatagram(connection, datagram, nowMillis()) } // Inbound data may have produced new outbound (acks, crypto, etc.). @@ -123,11 +127,16 @@ class QuicConnectionDriver( // floor (the same prior-shipping behavior, kept for // handshake-timeout safety on lossy paths). while (connection.status != QuicConnection.Status.CLOSED) { - connection.lock.withLock { - while (true) { - val out = drainOutbound(connection, nowMillis()) ?: break - socket.send(out) - } + // Phase 1 of the lock-split refactor: the writer holds + // streamsLock for the build, releases it for the actual + // socket.send() so a slow socket doesn't stall app + // coroutines (open/close streams, queue datagrams). + while (true) { + val out = + connection.streamsLock.withLock { + drainOutbound(connection, nowMillis()) + } ?: break + socket.send(out) } val ptoBaseMs = if (connection.lossDetection.hasFirstRttSample) { @@ -145,49 +154,14 @@ class QuicConnectionDriver( Unit } if (woke == null) { - // PTO fired. RFC 9002 §6.2.4: the probe packet MUST - // be ack-eliciting at the encryption level with - // unacknowledged data, and SHOULD retransmit lost - // data rather than just emit a PING. - // - // Pre-1-RTT we have a concrete thing to retransmit: - // the unacknowledged ClientHello (at Initial) or - // ClientFinished (at Handshake). We requeue ALL - // currently-inflight CRYPTO bytes for the highest - // active pre-application level so the next drain's - // takeChunk emits a fresh CRYPTO frame at the - // original offset. The pendingPing flag stays set - // as a fallback — `collectHandshakeLevelFrames` - // skips the PING when a CRYPTO retransmit lands in - // the same frame list, so we don't waste a frame. - // - // Why this matters in interop: aioquic strictly - // rejects pre-handshake Initials that contain no - // CRYPTO frame (CONNECTION_CLOSE 0x0 "Packet - // contains no CRYPTO frame"). A bare-PING probe - // hits that immediately. With CRYPTO retransmit on - // PTO, the probe carries a re-emitted ClientHello - // and the server processes it normally. - // - // Post-handshake (1-RTT installed) we retain the - // bare-PING behavior; STREAM retransmit is driven - // by packet-number-threshold loss detection from - // the ACK that the PING elicits. - val newPtoCount = - connection.lock - .withLock { - connection.pendingPing = true - if (connection.application.sendProtection == null) { - val level = highestPreApplicationLevel(connection) - if (level != null) { - connection.requeueAllInflightCrypto(level) - } - } - connection.consecutivePtoCount = - (connection.consecutivePtoCount + 1).coerceAtMost(6) - connection.consecutivePtoCount - } - connection.qlogObserver.onPtoFired(newPtoCount, ptoMillis) + // PTO fired. Set pendingPing so the writer emits a + // PING on the next drain (RFC 9002 §6.2.4 probe + // packet). The peer's ACK feeds loss detection + + // retransmit (steps 5–6). Both fields are @Volatile; + // no lock required. + connection.pendingPing = true + connection.consecutivePtoCount = + (connection.consecutivePtoCount + 1).coerceAtMost(6) } } } diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionParser.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionParser.kt index cfb3e315d..9efcbafea 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionParser.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionParser.kt @@ -54,6 +54,15 @@ import com.vitorpamplona.quic.tls.TlsClient * — typically Initial + Handshake from the server in the same datagram during * the handshake. We loop until the datagram is fully consumed or a packet * fails to parse (which we drop silently per RFC 9001 §5.5). + * + * Lock-split refactor (2026-05-08): caller must hold + * [QuicConnection.streamsLock]. The driver wraps its read loop in + * `streamsLock.withLock { feedDatagram(...) }`. Test harnesses that drive + * single-threaded packet flow (no concurrent app code) may invoke this + * directly without lock acquisition; the runtime invariants still hold + * because there's no contending thread. Phase 1 wraps the whole feed + * under streamsLock so frame-dispatch / stream creation / level state + * remains a single critical section. */ fun feedDatagram( conn: QuicConnection, diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt index 85a4e1107..e60635170 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt @@ -57,6 +57,15 @@ import com.vitorpamplona.quic.packet.ShortHeaderPlaintextPacket * * RFC 9000 §14: any datagram containing an Initial packet from the client * MUST be padded to at least 1200 bytes total. + * + * Lock-split refactor (2026-05-08): caller must hold + * [QuicConnection.streamsLock]. Phase 1 keeps level-state mutation + * inline under the same critical section as the streams-domain work + * the writer needs — the win comes from `lifecycleLock`-only callers + * (close(), status reads, PTO bookkeeping) no longer fighting this lock. + * The driver wraps `streamsLock.withLock { drainOutbound(...) }`; tests + * that drive single-threaded send paths can call this without holding + * the lock — there's no contending thread. */ fun drainOutbound( conn: QuicConnection, diff --git a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/AckTrackerPurgeOnAckOfAckTest.kt b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/AckTrackerPurgeOnAckOfAckTest.kt index 37e304ed0..594ed5421 100644 --- a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/AckTrackerPurgeOnAckOfAckTest.kt +++ b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/AckTrackerPurgeOnAckOfAckTest.kt @@ -64,7 +64,7 @@ class AckTrackerPurgeOnAckOfAckTest { // Peer ACKs the packet that carried our outbound ACK // covering up to PN 4. - conn.lock.lock() + conn.streamsLock.lock() try { conn.onTokensAcked( listOf( @@ -75,7 +75,7 @@ class AckTrackerPurgeOnAckOfAckTest { ), ) } finally { - conn.lock.unlock() + conn.streamsLock.unlock() } // Tracker is now empty: peer has confirmed receipt of our // ACK that covered everything up to PN 4. Re-advertising @@ -96,7 +96,7 @@ class AckTrackerPurgeOnAckOfAckTest { } // Peer ACKs our Initial-level outbound ACK. - conn.lock.lock() + conn.streamsLock.lock() try { conn.onTokensAcked( listOf( @@ -104,7 +104,7 @@ class AckTrackerPurgeOnAckOfAckTest { ), ) } finally { - conn.lock.unlock() + conn.streamsLock.unlock() } // Initial tracker drained; Application tracker untouched. assertTrue(conn.initial.ackTracker.isEmpty()) @@ -120,7 +120,7 @@ class AckTrackerPurgeOnAckOfAckTest { } // Peer ACKs our outbound ACK that covered up to PN 4 only; // the tracker's higher-PN ranges (5..9) must survive. - conn.lock.lock() + conn.streamsLock.lock() try { conn.onTokensAcked( listOf( @@ -128,7 +128,7 @@ class AckTrackerPurgeOnAckOfAckTest { ), ) } finally { - conn.lock.unlock() + conn.streamsLock.unlock() } assertFalse(conn.application.ackTracker.isEmpty()) assertEquals(9L, conn.application.ackTracker.largestReceived()) @@ -145,7 +145,7 @@ class AckTrackerPurgeOnAckOfAckTest { for (pn in 0L..9L) { conn.application.ackTracker.receivedPacket(pn, ackEliciting = true, receivedAtMillis = 1L) } - conn.lock.lock() + conn.streamsLock.lock() try { conn.onTokensAcked( listOf( @@ -160,7 +160,7 @@ class AckTrackerPurgeOnAckOfAckTest { ) assertTrue(conn.application.ackTracker.isEmpty()) } finally { - conn.lock.unlock() + conn.streamsLock.unlock() } } } diff --git a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/CryptoRetransmitTest.kt b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/CryptoRetransmitTest.kt index e3663062e..d7f4a1635 100644 --- a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/CryptoRetransmitTest.kt +++ b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/CryptoRetransmitTest.kt @@ -81,12 +81,12 @@ class CryptoRetransmitTest { .single() // Simulate loss via direct dispatch. - client.lock.lock() + client.streamsLock.lock() try { client.onTokensLost(listOf(cryptoToken)) client.initial.sentPackets.remove(firstPn) } finally { - client.lock.unlock() + client.streamsLock.unlock() } // Initial-level cryptoSend should now have re-queued bytes @@ -126,11 +126,11 @@ class CryptoRetransmitTest { client.initial.sentPackets.entries .first { it.value.tokens.any { t -> t is RecoveryToken.Crypto } } // ACK via direct dispatch. - client.lock.lock() + client.streamsLock.lock() try { client.onTokensAcked(packet.value.tokens) } finally { - client.lock.unlock() + client.streamsLock.unlock() } // After ACK the Initial-level cryptoSend's flushedFloor should // have advanced — we check by observing that another takeChunk diff --git a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/MultiplexingThroughputTest.kt b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/MultiplexingThroughputTest.kt new file mode 100644 index 000000000..8c90bed41 --- /dev/null +++ b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/MultiplexingThroughputTest.kt @@ -0,0 +1,139 @@ +/* + * Copyright (c) 2025 Vitor Pamplona + * + * Permission is hereby granted, free of charge, to any person obtaining a copy of + * this software and associated documentation files (the "Software"), to deal in + * the Software without restriction, including without limitation the rights to use, + * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the + * Software, and to permit persons to whom the Software is furnished to do so, + * subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS + * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR + * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN + * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION + * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + */ +package com.vitorpamplona.quic.connection + +import com.vitorpamplona.quic.tls.InProcessTlsServer +import com.vitorpamplona.quic.tls.PermissiveCertificateValidator +import kotlinx.coroutines.async +import kotlinx.coroutines.awaitAll +import kotlinx.coroutines.runBlocking +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertTrue + +/** + * Throughput contract for the lock-split refactor (2026-05-08): opening + * many parallel bidi streams + queueing requests must not be serialised + * by a single connection-wide mutex. Phase 1 of the split (separate + * `streamsLock` / `lifecycleLock` / per-level `levelLock`) targets the + * multiplexing testcase that drove this refactor — see + * `quic/plans/2026-05-08-lock-split-design.md`. + * + * The test stands up an in-memory client (no socket I/O), opens 1000 + * client-bidi streams concurrently, enqueues a small request body + FIN + * on each, and asserts the operation completes within a generous wall- + * clock budget. The number is deliberately loose: this is a contract + * for "lock contention isn't pathological", not a microbenchmark. + * + * NOTE: the in-memory pipe doesn't drive a concurrent send loop, so + * this test exercises the lock-acquisition cost of `openBidiStream` + * itself rather than full multiplexing throughput. The interop runner + * provides the end-to-end measurement. + */ +class MultiplexingThroughputTest { + @Test + fun open_1000_bidi_streams_completes_quickly() { + runBlocking { + val client = + QuicConnection( + serverName = "example.test", + config = + QuicConnectionConfig( + initialMaxStreamsBidi = 2_000, + initialMaxStreamsUni = 2_000, + initialMaxData = 100_000_000, + initialMaxStreamDataBidiLocal = 100_000, + initialMaxStreamDataBidiRemote = 100_000, + initialMaxStreamDataUni = 100_000, + ), + tlsCertificateValidator = PermissiveCertificateValidator(), + ) + val serverScid = ConnectionId.random(8) + val tlsServer = + InProcessTlsServer( + transportParameters = + TransportParameters( + initialMaxData = 100_000_000, + initialMaxStreamDataBidiLocal = 100_000, + initialMaxStreamDataBidiRemote = 100_000, + initialMaxStreamDataUni = 100_000, + initialMaxStreamsBidi = 2_000, + initialMaxStreamsUni = 2_000, + initialSourceConnectionId = serverScid.bytes, + originalDestinationConnectionId = client.destinationConnectionId.bytes, + ).encode(), + ) + val pipe = + InMemoryQuicPipe( + client = client, + initialDcid = client.destinationConnectionId.bytes, + serverScid = serverScid, + tlsServer = tlsServer, + ) + client.start() + pipe.drive(maxRounds = 16) + assertEquals(QuicConnection.Status.CONNECTED, client.status) + + val request = ByteArray(50) { it.toByte() } + val streamCount = 1_000 + + // Open all streams in parallel — each launch contends for + // streamsLock briefly. Pre-refactor this serialised against + // any in-flight drainOutbound call; phase 1 keeps openBidi + // contention scoped to streamsLock-only. + val started = + kotlin.time.TimeSource.Monotonic + .markNow() + val opens = + (0 until streamCount).map { + async { + val stream = client.openBidiStream() + stream.send.enqueue(request) + stream.send.finish() + stream.streamId + } + } + val ids = opens.awaitAll() + val elapsed = started.elapsedNow() + + // Useful diagnostic for measuring future regressions: stdout + // shows up in the test report so phase-1 vs phase-2 can be + // compared against the same test. + println( + "[MultiplexingThroughputTest] opened $streamCount bidi streams in " + + "${elapsed.inWholeMilliseconds}ms " + + "(${(streamCount * 1000.0 / elapsed.inWholeMilliseconds.coerceAtLeast(1L)).toLong()} streams/sec)", + ) + assertEquals(streamCount, ids.size) + assertEquals(streamCount, ids.toSet().size, "stream ids must be unique") + // Generous bound; in-process opens of 1000 streams should + // complete in well under half a second on any developer + // machine — pre-refactor this was minutes due to lock + // contention against the (idle) send-loop drain. The looser + // 2-second bound is still 100x what's expected on actual + // hardware while accounting for slow CI workers. + assertTrue( + elapsed.inWholeMilliseconds < 2_000L, + "1000 parallel openBidiStream calls took ${elapsed.inWholeMilliseconds}ms; expected <2000ms", + ) + } + } +} diff --git a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/OnTokensLostTest.kt b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/OnTokensLostTest.kt index c14da3ff4..9f3cf3a78 100644 --- a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/OnTokensLostTest.kt +++ b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/OnTokensLostTest.kt @@ -47,11 +47,11 @@ class OnTokensLostTest { fun ackToken_doesNotPopulateAnyPending() = runBlocking { val conn = newConn() - conn.lock.lock() + conn.streamsLock.lock() try { conn.onTokensLost(listOf(RecoveryToken.Ack(level = EncryptionLevel.APPLICATION, largestAcked = 0L))) } finally { - conn.lock.unlock() + conn.streamsLock.unlock() } assertNull(conn.pendingMaxStreamsUni) assertNull(conn.pendingMaxStreamsBidi) @@ -64,12 +64,12 @@ class OnTokensLostTest { runBlocking { val conn = newConn() // Simulate the writer having advertised a higher cap. - conn.lock.lock() + conn.streamsLock.lock() try { conn.advertisedMaxStreamsUni = 150L conn.onTokensLost(listOf(RecoveryToken.MaxStreamsUni(maxStreams = 150L))) } finally { - conn.lock.unlock() + conn.streamsLock.unlock() } assertEquals(150L, conn.pendingMaxStreamsUni) } @@ -82,12 +82,12 @@ class OnTokensLostTest { // the value carried by the lost token (150). The lost // frame is irrelevant — re-emitting 150 would not extend // the cap. neqo's fc.rs line 322 supersede check. - conn.lock.lock() + conn.streamsLock.lock() try { conn.advertisedMaxStreamsUni = 200L conn.onTokensLost(listOf(RecoveryToken.MaxStreamsUni(maxStreams = 150L))) } finally { - conn.lock.unlock() + conn.streamsLock.unlock() } assertNull(conn.pendingMaxStreamsUni, "stale lost extension must not be re-emitted") } @@ -96,12 +96,12 @@ class OnTokensLostTest { fun lostMaxStreamsBidi_matchingAdvertised_setsPending() = runBlocking { val conn = newConn() - conn.lock.lock() + conn.streamsLock.lock() try { conn.advertisedMaxStreamsBidi = 200L conn.onTokensLost(listOf(RecoveryToken.MaxStreamsBidi(maxStreams = 200L))) } finally { - conn.lock.unlock() + conn.streamsLock.unlock() } assertEquals(200L, conn.pendingMaxStreamsBidi) } @@ -110,12 +110,12 @@ class OnTokensLostTest { fun lostMaxData_matchingAdvertised_setsPending() = runBlocking { val conn = newConn() - conn.lock.lock() + conn.streamsLock.lock() try { conn.advertisedMaxData = 1_000_000L conn.onTokensLost(listOf(RecoveryToken.MaxData(maxData = 1_000_000L))) } finally { - conn.lock.unlock() + conn.streamsLock.unlock() } assertEquals(1_000_000L, conn.pendingMaxData) } @@ -124,12 +124,12 @@ class OnTokensLostTest { fun lostMaxData_supersededIsDropped() = runBlocking { val conn = newConn() - conn.lock.lock() + conn.streamsLock.lock() try { conn.advertisedMaxData = 2_000_000L conn.onTokensLost(listOf(RecoveryToken.MaxData(maxData = 1_000_000L))) } finally { - conn.lock.unlock() + conn.streamsLock.unlock() } assertNull(conn.pendingMaxData) } @@ -138,13 +138,13 @@ class OnTokensLostTest { fun lostMaxStreamData_unknownStream_dropped() = runBlocking { val conn = newConn() - conn.lock.lock() + conn.streamsLock.lock() try { conn.onTokensLost( listOf(RecoveryToken.MaxStreamData(streamId = 999L, maxData = 1024L)), ) } finally { - conn.lock.unlock() + conn.streamsLock.unlock() } // No stream with id 999 exists ⇒ token is dropped silently. assertEquals(emptyMap(), conn.pendingMaxStreamData) @@ -154,7 +154,7 @@ class OnTokensLostTest { fun multipleLostTokens_dispatchAll() = runBlocking { val conn = newConn() - conn.lock.lock() + conn.streamsLock.lock() try { conn.advertisedMaxStreamsUni = 150L conn.advertisedMaxStreamsBidi = 200L @@ -168,7 +168,7 @@ class OnTokensLostTest { ), ) } finally { - conn.lock.unlock() + conn.streamsLock.unlock() } assertEquals(150L, conn.pendingMaxStreamsUni) assertEquals(200L, conn.pendingMaxStreamsBidi) @@ -183,7 +183,7 @@ class OnTokensLostTest { // most one value (the last setter wins; the supersede // check filters older losses). val conn = newConn() - conn.lock.lock() + conn.streamsLock.lock() try { conn.advertisedMaxStreamsUni = 200L // First lost packet had MaxStreamsUni(150) — stale, dropped. @@ -193,7 +193,7 @@ class OnTokensLostTest { conn.onTokensLost(listOf(RecoveryToken.MaxStreamsUni(maxStreams = 200L))) assertEquals(200L, conn.pendingMaxStreamsUni) } finally { - conn.lock.unlock() + conn.streamsLock.unlock() } } } diff --git a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/PeerStreamCreditExtensionTest.kt b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/PeerStreamCreditExtensionTest.kt index e89c51da0..54a0a97be 100644 --- a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/PeerStreamCreditExtensionTest.kt +++ b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/PeerStreamCreditExtensionTest.kt @@ -94,7 +94,7 @@ class PeerStreamCreditExtensionTest { // Simulate the relay opening uni streams to us. SERVER_UNI // stream IDs use the encoding `index << 2 | 0x3`. Two streams // (cap=4, half-window=2) is the threshold for a refresh. - client.lock + client.streamsLock .let { // Acquire under lock since getOrCreatePeerStreamLocked requires it. it @@ -103,7 +103,7 @@ class PeerStreamCreditExtensionTest { kotlinx.coroutines.sync .Mutex() .let { /* noop: silence unused-import linter */ } - client.lock.let { l -> + client.streamsLock.let { l -> kotlinx.coroutines.runBlocking { l.lock() try { @@ -179,7 +179,7 @@ class PeerStreamCreditExtensionTest { // Open 10 peer streams — half-window for cap=100 is 50, so // we're well below the threshold. - client.lock.let { l -> + client.streamsLock.let { l -> kotlinx.coroutines.runBlocking { l.lock() try { diff --git a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/PendingFlowControlEmitTest.kt b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/PendingFlowControlEmitTest.kt index 1c42c9aeb..1971459fc 100644 --- a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/PendingFlowControlEmitTest.kt +++ b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/PendingFlowControlEmitTest.kt @@ -46,11 +46,11 @@ class PendingFlowControlEmitTest { fun pendingMaxStreamsUni_drainEmitsFrameAndToken() = runBlocking { val client = handshakedClient() - client.lock.lock() + client.streamsLock.lock() try { client.pendingMaxStreamsUni = 150L } finally { - client.lock.unlock() + client.streamsLock.unlock() } val sizeBefore = client.application.sentPackets.size @@ -79,11 +79,11 @@ class PendingFlowControlEmitTest { fun pendingMaxStreamsBidi_drainEmitsFrameAndToken(): Unit = runBlocking { val client = handshakedClient() - client.lock.lock() + client.streamsLock.lock() try { client.pendingMaxStreamsBidi = 200L } finally { - client.lock.unlock() + client.streamsLock.unlock() } val sizeBefore = client.application.sentPackets.size runCatching { drainOutbound(client, nowMillis = 1L) } @@ -103,11 +103,11 @@ class PendingFlowControlEmitTest { fun pendingMaxData_drainEmitsFrameAndToken(): Unit = runBlocking { val client = handshakedClient() - client.lock.lock() + client.streamsLock.lock() try { client.pendingMaxData = 5_000_000L } finally { - client.lock.unlock() + client.streamsLock.unlock() } val sizeBefore = client.application.sentPackets.size runCatching { drainOutbound(client, nowMillis = 1L) } @@ -127,12 +127,12 @@ class PendingFlowControlEmitTest { fun pendingMaxStreamData_perStreamDrain() = runBlocking { val client = handshakedClient() - client.lock.lock() + client.streamsLock.lock() try { client.pendingMaxStreamData[3L] = 1_024L client.pendingMaxStreamData[7L] = 2_048L } finally { - client.lock.unlock() + client.streamsLock.unlock() } val sizeBefore = client.application.sentPackets.size runCatching { drainOutbound(client, nowMillis = 1L) } @@ -159,13 +159,13 @@ class PendingFlowControlEmitTest { fun multiplePending_drainEmitsAllInOnePacket(): Unit = runBlocking { val client = handshakedClient() - client.lock.lock() + client.streamsLock.lock() try { client.pendingMaxStreamsUni = 150L client.pendingMaxStreamsBidi = 200L client.pendingMaxData = 1_000_000L } finally { - client.lock.unlock() + client.streamsLock.unlock() } val sizeBefore = client.application.sentPackets.size runCatching { drainOutbound(client, nowMillis = 1L) } @@ -216,11 +216,11 @@ class PendingFlowControlEmitTest { // advertised cap. The writer drains it as-is — supersede check // is in step 6 (the setter side), not here. val client = handshakedClient() - client.lock.lock() + client.streamsLock.lock() try { client.pendingMaxStreamsUni = 50L } finally { - client.lock.unlock() + client.streamsLock.unlock() } val sizeBefore = client.application.sentPackets.size runCatching { drainOutbound(client, nowMillis = 1L) } @@ -242,11 +242,11 @@ class PendingFlowControlEmitTest { fun pendingClearedAcrossDrains() = runBlocking { val client = handshakedClient() - client.lock.lock() + client.streamsLock.lock() try { client.pendingMaxStreamsUni = 150L } finally { - client.lock.unlock() + client.streamsLock.unlock() } // Drain once: pending consumed. runCatching { drainOutbound(client, nowMillis = 1L) } diff --git a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/ResetStopSendingEmitTest.kt b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/ResetStopSendingEmitTest.kt index d71ca978c..43af5e5b0 100644 --- a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/ResetStopSendingEmitTest.kt +++ b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/ResetStopSendingEmitTest.kt @@ -93,12 +93,12 @@ class ResetStopSendingEmitTest { .single() // Simulate loss. - client.lock.lock() + client.streamsLock.lock() try { client.onTokensLost(listOf(token)) client.application.sentPackets.remove(firstEntry.key) } finally { - client.lock.unlock() + client.streamsLock.unlock() } // Per-stream emit-pending should be re-flagged. assertTrue(stream.resetEmitPending, "loss must re-flag resetEmitPending") @@ -137,21 +137,21 @@ class ResetStopSendingEmitTest { .single() // ACK first. - client.lock.lock() + client.streamsLock.lock() try { client.onTokensAcked(listOf(token)) } finally { - client.lock.unlock() + client.streamsLock.unlock() } assertEquals(true, stream.resetAcked) assertEquals(false, stream.resetEmitPending) // Now a stale loss notification arrives. Defensive: drop. - client.lock.lock() + client.streamsLock.lock() try { client.onTokensLost(listOf(token)) } finally { - client.lock.unlock() + client.streamsLock.unlock() } assertEquals(false, stream.resetEmitPending, "stale loss after ACK must not re-flag emit-pending") } @@ -248,11 +248,11 @@ class ResetStopSendingEmitTest { connectionId = byteArrayOf(1, 2, 3, 4), statelessResetToken = ByteArray(16) { it.toByte() }, ) - client.lock.lock() + client.streamsLock.lock() try { client.onTokensLost(listOf(token)) } finally { - client.lock.unlock() + client.streamsLock.unlock() } assertEquals(token, client.pendingNewConnectionId[1L]) diff --git a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/RetransmitIntegrationTest.kt b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/RetransmitIntegrationTest.kt index 73a05ce1c..dc35823f5 100644 --- a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/RetransmitIntegrationTest.kt +++ b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/RetransmitIntegrationTest.kt @@ -69,7 +69,7 @@ class RetransmitIntegrationTest { // ACK'd by reordering — its PN < largestAckedPn - // PACKET_THRESHOLD ⇒ declared lost. val futurePn = msuPn + 4L - client.lock.lock() + client.streamsLock.lock() try { // Inject a phantom SentPacket at futurePn so the loss // detector has a credible "newly acked" reference, then @@ -102,7 +102,7 @@ class RetransmitIntegrationTest { // 5. Dispatch lost tokens — pendingMaxStreamsUni gets set. client.onTokensLost(lostMsuPacket.tokens) } finally { - client.lock.unlock() + client.streamsLock.unlock() } assertEquals( capAfterFirstDrain, @@ -148,24 +148,24 @@ class RetransmitIntegrationTest { // Second bump: open more peer-uni streams to cross the // (already extended) threshold again. - client.lock.lock() + client.streamsLock.lock() try { client.getOrCreatePeerStreamLocked(StreamId.build(StreamId.Kind.SERVER_UNI, 2)) client.getOrCreatePeerStreamLocked(StreamId.build(StreamId.Kind.SERVER_UNI, 3)) client.getOrCreatePeerStreamLocked(StreamId.build(StreamId.Kind.SERVER_UNI, 4)) } finally { - client.lock.unlock() + client.streamsLock.unlock() } runCatching { drainOutbound(client, nowMillis = 2L) } val secondCap = client.advertisedMaxStreamsUni assertTrue(secondCap > firstCap, "second drain must advertise a still-higher cap; saw $firstCap → $secondCap") // Now declare the FIRST emit lost via direct dispatch. - client.lock.lock() + client.streamsLock.lock() try { client.onTokensLost(listOf(RecoveryToken.MaxStreamsUni(maxStreams = firstCap))) } finally { - client.lock.unlock() + client.streamsLock.unlock() } // Supersede check: firstCap != advertisedMaxStreamsUni (now == secondCap), // so pending must remain null. @@ -216,12 +216,12 @@ class RetransmitIntegrationTest { private fun crossPeerUniHalfWindow(client: QuicConnection) = runBlocking { - client.lock.lock() + client.streamsLock.lock() try { client.getOrCreatePeerStreamLocked(StreamId.build(StreamId.Kind.SERVER_UNI, 0)) client.getOrCreatePeerStreamLocked(StreamId.build(StreamId.Kind.SERVER_UNI, 1)) } finally { - client.lock.unlock() + client.streamsLock.unlock() } } } diff --git a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/SentPacketTrackingTest.kt b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/SentPacketTrackingTest.kt index b11e72a55..81032e72f 100644 --- a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/SentPacketTrackingTest.kt +++ b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/SentPacketTrackingTest.kt @@ -200,12 +200,12 @@ class SentPacketTrackingTest { /** Cross the half-window threshold (cap=4, two peer-uni streams ⇒ count >= cap-half=2). */ private fun crossPeerUniHalfWindow(client: QuicConnection) = runBlocking { - client.lock.lock() + client.streamsLock.lock() try { client.getOrCreatePeerStreamLocked(StreamId.build(StreamId.Kind.SERVER_UNI, 0)) client.getOrCreatePeerStreamLocked(StreamId.build(StreamId.Kind.SERVER_UNI, 1)) } finally { - client.lock.unlock() + client.streamsLock.unlock() } } } diff --git a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/StreamRetransmitTest.kt b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/StreamRetransmitTest.kt index b6cb893d8..25e039b4c 100644 --- a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/StreamRetransmitTest.kt +++ b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/StreamRetransmitTest.kt @@ -82,7 +82,7 @@ class StreamRetransmitTest { val firstPn = firstPacketEntry.key // Simulate loss via direct dispatch. - client.lock.lock() + client.streamsLock.lock() try { val streamToken = firstPacketEntry.value.tokens @@ -93,7 +93,7 @@ class StreamRetransmitTest { // detector would have done this). client.application.sentPackets.remove(firstPn) } finally { - client.lock.unlock() + client.streamsLock.unlock() } // SendBuffer should have re-queued the bytes for retransmit. @@ -133,11 +133,11 @@ class StreamRetransmitTest { val packet = client.application.sentPackets.entries .first { it.value.tokens.any { t -> t is RecoveryToken.Stream } } - client.lock.lock() + client.streamsLock.lock() try { client.onTokensAcked(packet.value.tokens) } finally { - client.lock.unlock() + client.streamsLock.unlock() } // After ACK: enqueue more, observe that the buffer