diff --git a/quic/interop/src/main/kotlin/com/vitorpamplona/quic/interop/runner/HqInteropGetClient.kt b/quic/interop/src/main/kotlin/com/vitorpamplona/quic/interop/runner/HqInteropGetClient.kt index 39299ad57..74e143adc 100644 --- a/quic/interop/src/main/kotlin/com/vitorpamplona/quic/interop/runner/HqInteropGetClient.kt +++ b/quic/interop/src/main/kotlin/com/vitorpamplona/quic/interop/runner/HqInteropGetClient.kt @@ -23,7 +23,6 @@ package com.vitorpamplona.quic.interop.runner import com.vitorpamplona.quic.connection.QuicConnection import com.vitorpamplona.quic.connection.QuicConnectionDriver import kotlinx.coroutines.flow.toList -import kotlinx.coroutines.sync.withLock /** * HQ-interop (HTTP/0.9 over QUIC) GET client. quic-interop-runner convention @@ -57,19 +56,14 @@ class HqInteropGetClient( @Suppress("UNUSED_PARAMETER") authority: String, paths: List, ): List { - // Pre-format requests outside the lock — see Http3GetClient - // for the full story. + // Pre-format outside the lock — see Http3GetClient. Then + // openBidiStreamsBatch atomically opens all N streams under + // streamsLock so the writer's next drain coalesces them. val encoded = paths.map { "GET $it\r\n".encodeToByteArray() } - // streamsLock, NOT lifecycleLock (the deprecated `conn.lock` - // alias). Holding the wrong lock lets the send-loop drain - // between opens and emits one STREAM per packet. - return conn.streamsLock.withLock { - encoded.map { request -> - val stream = conn.openBidiStreamLocked() - stream.send.enqueue(request) - stream.send.finish() - HqRequestHandle(stream) - } + return conn.openBidiStreamsBatch(encoded) { stream, request -> + stream.send.enqueue(request) + stream.send.finish() + HqRequestHandle(stream) } } diff --git a/quic/interop/src/main/kotlin/com/vitorpamplona/quic/interop/runner/Http3GetClient.kt b/quic/interop/src/main/kotlin/com/vitorpamplona/quic/interop/runner/Http3GetClient.kt index 9a4152fab..a3dc638b8 100644 --- a/quic/interop/src/main/kotlin/com/vitorpamplona/quic/interop/runner/Http3GetClient.kt +++ b/quic/interop/src/main/kotlin/com/vitorpamplona/quic/interop/runner/Http3GetClient.kt @@ -33,7 +33,6 @@ import com.vitorpamplona.quic.qpack.QpackDecoder import com.vitorpamplona.quic.qpack.QpackEncoder import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.flow.collect -import kotlinx.coroutines.sync.withLock /** Common shape for the two interop GET clients (HTTP/3 and HQ-interop). * @@ -165,25 +164,16 @@ class Http3GetClient( // non-trivial; doing it under streamsLock would stall the // send loop for the full encode time across every chunk. val encoded = paths.map { encodeRequest(authority, it) } - // CRITICAL: streamsLock — the lock the writer's drainOutbound - // takes. Holding lifecycleLock (the deprecated `conn.lock` - // alias) here did NOT block the send loop, so the writer - // interjected between every openBidiStreamLocked call and - // emitted ONE STREAM frame per packet. aioquic interop - // 2026-05-06 qlog: 2898 packets sent in 60s, each carrying - // exactly one stream frame; server processed streams strictly - // sequentially at ~1 RTT per stream → 1421/2000 files in 60s - // before timeout. Holding streamsLock blocks the drain so - // all N opens land before any drain runs, the writer then - // packs many STREAM frames into each datagram, and the - // server sees the burst on the wire. - return conn.streamsLock.withLock { - encoded.map { request -> - val stream = conn.openBidiStreamLocked() - stream.send.enqueue(request) - stream.send.finish() - Http3RequestHandle(stream) - } + // openBidiStreamsBatch holds streamsLock for the entire + // batch — the send loop can't interject between opens so + // the writer's next drain finds all N streams' frames ready + // and packs them into coalesced packets (vs. the regressed + // shape that emitted one STREAM per packet against aioquic + // on 2026-05-06). + return conn.openBidiStreamsBatch(encoded) { stream, request -> + stream.send.enqueue(request) + stream.send.finish() + Http3RequestHandle(stream) } } 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 2bdf355cd..87fe408cc 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/LevelState.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/LevelState.kt @@ -23,25 +23,20 @@ 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]. */ +/** + * Per-encryption-level state owned by [QuicConnection]. + * + * Concurrency: [cryptoSend] / [cryptoReceive] use their internal + * [SendBuffer] / [ReceiveBuffer] `synchronized(this)` blocks for + * thread safety — the writer's `takeChunk`, the parser's `markAcked`, + * and PTO-driven `requeueAllInflight` are all serialized through + * those leaf locks. [sentPackets] is currently mutated by the writer + * (under [QuicConnection.streamsLock]) and read by the parser without + * synchronization; that race is pre-existing audit-tracked and not + * fixed by [LevelState] today. + */ class LevelState { - /** - * 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 = 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 45b8ef8db..9a6efe315 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt @@ -717,22 +717,24 @@ class QuicConnection( /** * Lock-split refactor (2026-05-08): split the previous single - * `lock` into three independent mutexes so the read loop, send + * `lock` into two 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. + * stream-id counters, connection-level flow-control bookkeeping, + * packet-number space + sentPackets retention + CRYPTO buffer + * mutations at every encryption level. The writer's drain and + * the parser's feed both take it. + * - [lifecycleLock] guards [status] / [closeReason] / + * [closeErrorCode] transitions. + * + * Per-stream and per-level buffer mutations serialize through + * `synchronized(this)` inside `SendBuffer` / `ReceiveBuffer` / + * `AckTracker` — those leaf locks are safe to take with or + * without an outer mutex held. * * 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. + * `lifecycleLock` → `streamsLock`. Never go the other way. * * The historical `lock` field is retained as an alias of * [lifecycleLock] for source-compatibility with external callers @@ -744,7 +746,7 @@ class QuicConnection( val lifecycleLock: Mutex = Mutex() @Deprecated( - "Use streamsLock / lifecycleLock / LevelState.levelLock as appropriate. Lock-split refactor 2026-05-08.", + "Use streamsLock or lifecycleLock as appropriate. Lock-split refactor 2026-05-08.", replaceWith = ReplaceWith("streamsLock"), ) val lock: Mutex @@ -761,11 +763,38 @@ class QuicConnection( suspend fun openBidiStream(): QuicStream = streamsLock.withLock { openBidiStreamLocked() } /** - * The streamsLock-holding part of [openBidiStream]. Public so - * batched-multiplex callers (prepareRequests) can acquire - * [streamsLock] ONCE and bracket multiple stream opens — without - * yielding to the send loop between opens. Caller MUST hold - * [streamsLock]. + * Atomically open one bidi stream per [items] entry under a single + * [streamsLock] hold and run [init] for each (stream, item) inside + * the lock. The send loop cannot interject between opens — when it + * next drains it sees ALL N streams' frames ready and packs them + * into coalesced packets instead of emitting one tiny packet per + * stream. + * + * This is the bug-resistant API for the prepareRequests pattern. + * The previous shape (caller manually wraps `streamsLock.withLock` + * around a loop of [openBidiStreamLocked]) regressed twice: once + * by holding the wrong lock, and once by skipping the wrapper + * entirely. Both shapes failed silently as "one STREAM per packet" + * under multiplex load, while the unit tests passed. + * + * Callers that just need a single stream should still use + * [openBidiStream]. [openBidiStreamLocked] remains public for the + * rare custom-batching scenarios that need finer control, but + * those callers should generally migrate to this API. + */ + suspend fun openBidiStreamsBatch( + items: List, + init: (QuicStream, I) -> R, + ): List = + streamsLock.withLock { + items.map { init(openBidiStreamLocked(), it) } + } + + /** + * The streamsLock-holding primitive used by [openBidiStream] and + * [openBidiStreamsBatch]. Public so callers that need a custom + * batching shape (e.g. mixed bidi+uni opens) can compose it under + * a manual [streamsLock] hold. Caller MUST hold [streamsLock]. */ fun openBidiStreamLocked(): QuicStream { // Mutex.isLocked is the only check we have — kotlinx.coroutines @@ -808,21 +837,50 @@ class QuicConnection( * [QuicStream.bestEffort]). Used for moq-lite group streams * carrying real-time Opus audio. */ - suspend fun openUniStream(bestEffort: Boolean = false): QuicStream = + suspend fun openUniStream(bestEffort: Boolean = false): QuicStream = streamsLock.withLock { openUniStreamLocked(bestEffort) } + + /** + * The streamsLock-holding primitive used by [openUniStream] and + * [openUniStreamsBatch]. Caller MUST hold [streamsLock]. + */ + fun openUniStreamLocked(bestEffort: Boolean = false): QuicStream { + check(streamsLock.isLocked) { + "openUniStreamLocked requires streamsLock to be held" + } + if (nextLocalUniIndex >= peerMaxStreamsUni) { + throw QuicStreamLimitException( + "peer-granted uni stream cap reached " + + "(used=$nextLocalUniIndex limit=$peerMaxStreamsUni)", + ) + } + val id = StreamId.build(StreamId.Kind.CLIENT_UNI, nextLocalUniIndex++) + val stream = QuicStream(id, QuicStream.Direction.UNIDIRECTIONAL_LOCAL_TO_REMOTE, bestEffort = bestEffort) + stream.sendCredit = peerTransportParameters?.initialMaxStreamDataUni ?: config.initialMaxStreamDataUni + stream.receiveLimit = 0L // can't receive + streams[id] = stream + streamsList += stream + return stream + } + + /** + * Bug-resistant counterpart to [openBidiStreamsBatch] for uni + * streams. Atomically open one client-uni stream per [items] + * entry under a single [streamsLock] hold and run [init] for + * each (stream, item). + * + * Use this for moq audio-rooms and any other path that opens many + * uni streams in burst — without batching, each open releases the + * lock and the send loop can interject, emitting one stream per + * packet (the same shape that broke bidi multiplexing on + * 2026-05-06). + */ + suspend fun openUniStreamsBatch( + items: List, + bestEffort: Boolean = false, + init: (QuicStream, I) -> R, + ): List = streamsLock.withLock { - if (nextLocalUniIndex >= peerMaxStreamsUni) { - throw QuicStreamLimitException( - "peer-granted uni stream cap reached " + - "(used=$nextLocalUniIndex limit=$peerMaxStreamsUni)", - ) - } - val id = StreamId.build(StreamId.Kind.CLIENT_UNI, nextLocalUniIndex++) - val stream = QuicStream(id, QuicStream.Direction.UNIDIRECTIONAL_LOCAL_TO_REMOTE, bestEffort = bestEffort) - stream.sendCredit = peerTransportParameters?.initialMaxStreamDataUni ?: config.initialMaxStreamDataUni - stream.receiveLimit = 0L // can't receive - streams[id] = stream - streamsList += stream - stream + items.map { init(openUniStreamLocked(bestEffort), it) } } /** Snapshot of peer-granted bidi cap. Reads do not need the lock — long writes are atomic on every supported platform. */ 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 d20e62781..f0be55c39 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionDriver.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionDriver.kt @@ -37,10 +37,11 @@ import kotlinx.coroutines.withTimeoutOrNull * * 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. + * [feedDatagram] and [drainOutbound] internally acquire `streamsLock` + * for the precise critical sections they touch — leaving app + * coroutines (`openBidiStream`, etc.) free to run in parallel with the + * I/O loops. Per-stream and per-level buffers serialize through their + * leaf `synchronized(this)` blocks. * * The send loop is woken by a `Channel(CONFLATED)` rather than a * polling timer — no idle CPU. App writes ([QuicConnection.queueDatagram] @@ -253,20 +254,21 @@ class QuicConnectionDriver( * (commits c0d7b6031, then again in the lock-split refactor) without * any test breaking. * - * `pendingPing` and `consecutivePtoCount` are `@Volatile` so we set - * them outside any external lock. `requeueAllInflightCrypto` mutates - * the level's `cryptoSend` buffer; we take `levelLock` for the - * requeue so the writer's `takeChunk` can't observe a half-mutated - * inflight queue mid-build. + * Concurrency: `pendingPing` and `consecutivePtoCount` are `@Volatile`. + * [QuicConnection.requeueAllInflightCrypto] delegates to + * [com.vitorpamplona.quic.stream.SendBuffer.requeueAllInflight] which + * is `synchronized(this)` internally, so it's safe to call without + * an external lock — even concurrent with the writer's `takeChunk`. + * If the parser concurrently runs `discardKeys` on the same level, + * `requeueAllInflight` operates on the buffer reference we captured + * (or the fresh one — both are valid) and is at worst a no-op. */ -internal suspend fun handlePtoFired(conn: QuicConnection) { +internal fun handlePtoFired(conn: QuicConnection) { conn.pendingPing = true if (conn.application.sendProtection == null) { val level = highestPreApplicationLevel(conn) if (level != null) { - conn.levelState(level).levelLock.withLock { - conn.requeueAllInflightCrypto(level) - } + conn.requeueAllInflightCrypto(level) } } conn.consecutivePtoCount = (conn.consecutivePtoCount + 1).coerceAtMost(6) diff --git a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/BatchedOpenLockContractTest.kt b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/BatchedOpenLockContractTest.kt index 1bc0d4348..9f2b4909d 100644 --- a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/BatchedOpenLockContractTest.kt +++ b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/BatchedOpenLockContractTest.kt @@ -116,6 +116,51 @@ class BatchedOpenLockContractTest { } } + @Test + fun `openBidiStreamsBatch is the bug-resistant API for the prepareRequests pattern`() { + runBlocking { + val client = handshakedClient() + // The new high-level API — caller can't hold the wrong + // lock because it doesn't take any. Encapsulates the + // streamsLock acquisition + per-item init under the lock. + val payloads = (0 until 64).map { "req-$it".encodeToByteArray() } + val streams = + client.openBidiStreamsBatch(payloads) { stream, payload -> + stream.send.enqueue(payload) + stream.send.finish() + stream + } + assertEquals(64, streams.size) + assertEquals(64, streams.map { it.streamId }.toSet().size) + } + } + + @Test + fun `openUniStreamLocked throws when streamsLock is not held`() { + runBlocking { + val client = handshakedClient() + assertFailsWith { + client.openUniStreamLocked() + } + } + } + + @Test + fun `openUniStreamsBatch holds streamsLock for the whole batch`() { + runBlocking { + val client = handshakedClient() + // moq audio-rooms shape: many uni streams in burst. Without + // the batched API, each open releases the lock and the + // send loop interjects (the same shape that broke bidi + // multiplexing on 2026-05-06). + val items = List(16) { it } + val streams = + client.openUniStreamsBatch(items) { stream, _ -> stream } + assertEquals(16, streams.size) + assertEquals(16, streams.map { it.streamId }.toSet().size) + } + } + private fun handshakedClient(): QuicConnection = runBlocking { val client =