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 a3a81fe3d..39299ad57 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 @@ -56,20 +56,22 @@ class HqInteropGetClient( override suspend fun prepareRequests( @Suppress("UNUSED_PARAMETER") authority: String, paths: List, - ): List = + ): List { + // Pre-format requests outside the lock — see Http3GetClient + // for the full story. + val encoded = paths.map { "GET $it\r\n".encodeToByteArray() } // streamsLock, NOT lifecycleLock (the deprecated `conn.lock` - // alias) — see Http3GetClient.prepareRequests for the full - // story. tl;dr: holding the wrong lock lets the send-loop - // drain between opens and emits one STREAM per packet. - conn.streamsLock.withLock { - paths.map { path -> + // 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() - val request = "GET $path\r\n".encodeToByteArray() stream.send.enqueue(request) stream.send.finish() HqRequestHandle(stream) } } + } override suspend fun awaitResponse(handle: RequestHandle): GetResponse { val stream = (handle as 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 e48b87f79..9a4152fab 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 @@ -158,7 +158,13 @@ class Http3GetClient( override suspend fun prepareRequests( authority: String, paths: List, - ): List = + ): List { + // Pre-encode all requests OUTSIDE the lock so QPACK encoding + // (allocations, hashing, varint emit) doesn't extend the + // critical section. With 1999 paths the encoding cost is + // 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 @@ -171,14 +177,15 @@ class Http3GetClient( // 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. - conn.streamsLock.withLock { - paths.map { path -> + return conn.streamsLock.withLock { + encoded.map { request -> val stream = conn.openBidiStreamLocked() - stream.send.enqueue(encodeRequest(authority, path)) + stream.send.enqueue(request) stream.send.finish() Http3RequestHandle(stream) } } + } override suspend fun awaitResponse(handle: RequestHandle): GetResponse { val stream = (handle as Http3RequestHandle).stream diff --git a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/MultiplexingRoundTripTest.kt b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/MultiplexingRoundTripTest.kt index 147e79213..016872499 100644 --- a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/MultiplexingRoundTripTest.kt +++ b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/MultiplexingRoundTripTest.kt @@ -30,6 +30,7 @@ import kotlinx.coroutines.runBlocking import kotlinx.coroutines.withTimeoutOrNull import kotlin.test.Test import kotlin.test.assertEquals +import kotlin.test.assertNotNull import kotlin.test.assertTrue /** @@ -94,7 +95,7 @@ class MultiplexingRoundTripTest { // is supposed to coalesce many streams' frames into one datagram // (MultiplexingCoalescingTest pins that contract); here we just // walk every emitted datagram until the client has nothing left. - val seenRequests = mutableMapOf() + val seenRequests = mutableMapOf>() val seenFin = mutableSetOf() var totalDatagrams = 0 while (true) { @@ -104,13 +105,10 @@ class MultiplexingRoundTripTest { val frames = pipe.decryptClientApplicationFrames(out) ?: break for (frame in frames) { if (frame is StreamFrame) { - // Concatenate by offset (matrix reqs are tiny; offset - // is always 0, but real flows might split — accept both). - val existing = seenRequests[frame.streamId] ?: ByteArray(0) - val merged = ByteArray(existing.size + frame.data.size) - existing.copyInto(merged, 0) - frame.data.copyInto(merged, existing.size) - seenRequests[frame.streamId] = merged + // Append to per-stream chunk list — joined once at the + // end. Earlier shape merged eagerly per frame which is + // O(N²) over byte count. + seenRequests.getOrPut(frame.streamId) { mutableListOf() } += frame.data if (frame.fin) seenFin += frame.streamId } } @@ -128,10 +126,30 @@ class MultiplexingRoundTripTest { "server saw FINs from only ${seenFin.size}/$n streams — client failed " + "to deliver request FIN on every stream", ) + // End-to-end coalescing contract: 64 streams × ~10-byte + // requests fit in a handful of datagrams. The aioquic + // 2026-05-06 regression hit when the writer emitted ONE + // STREAM per datagram (so 64 datagrams for this batch). + // Threshold of 12 leaves headroom for the writer's choice + // of when to coalesce + any per-drain ACK frames. + assertTrue( + totalDatagrams <= 12, + "expected ≤12 datagrams for 64 streams, got $totalDatagrams — " + + "writer regressed to one-stream-per-datagram (the aioquic " + + "interop 2026-05-06 multiplexing failure mode)", + ) // Sanity-check the request payloads match what each stream sent. for ((i, stream) in streams.withIndex()) { val expected = "req-$i" - val actual = seenRequests[stream.streamId]?.decodeToString() + val chunks = seenRequests[stream.streamId] + assertNotNull(chunks, "stream ${stream.streamId} (i=$i): no STREAM frames received") + val joined = ByteArray(chunks.sumOf { it.size }) + var p = 0 + for (c in chunks) { + c.copyInto(joined, p) + p += c.size + } + val actual = joined.decodeToString() assertEquals( expected, actual,