From 57ba23519d0baf85c478f6e717f3c85717ea9969 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 7 May 2026 02:06:58 +0000 Subject: [PATCH] fix(quic): batch openBidiStream under one lock hold for multiplexing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Diagnosis from yet another qlog round: streams/packet still ~1 even with the prepareRequest/awaitResponse split. Root cause: openBidiStream is suspend due to lock.withLock, and each call releases the lock between iterations. The send loop is queued on the lock; it grabs it the moment we release, drains the one stream of data we just enqueued, and the next prepareRequest call has to re-acquire after the send loop releases. Net: one stream per drain per packet, same useless coalescing as before. Fix is structural: - QuicConnection.openBidiStreamLocked() — public, lock-not-acquired version of openBidiStream. Caller MUST hold conn.lock. - GetClient.prepareRequests(authority, paths) — batch API that holds conn.lock once, opens + enqueues all N streams in a single critical section, releases. Send loop can't interject; when it next drains it sees ALL N streams' data ready and packs them into coalesced packets. - Http3GetClient + HqInteropGetClient: implement prepareRequests using openBidiStreamLocked under conn.lock.withLock { ... }. - InteropClient's chunked-multiplex loop: uses prepareRequests (batch) instead of N x prepareRequest. Single-stream paths still use prepareRequest / get(); behavior unchanged. The fundamental architectural improvement (per-stream / per-level lock split, or actor-model dispatch) is a follow-up; this commit gets us the throughput we need from the existing single-mutex shape by holding the lock for the full chunk's worth of work. https://claude.ai/code/session_01HcvfQq1ttPV9PkRoJb4nyT --- .../quic/interop/runner/HqInteropGetClient.kt | 15 +++++++ .../quic/interop/runner/Http3GetClient.kt | 31 ++++++++++++++ .../quic/interop/runner/InteropClient.kt | 12 +++--- .../quic/connection/QuicConnection.kt | 42 ++++++++++++------- 4 files changed, 80 insertions(+), 20 deletions(-) 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 95af75432..cef596ede 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,6 +23,7 @@ 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 @@ -52,6 +53,20 @@ class HqInteropGetClient( return HqRequestHandle(stream) } + override suspend fun prepareRequests( + @Suppress("UNUSED_PARAMETER") authority: String, + paths: List, + ): List = + conn.lock.withLock { + paths.map { path -> + 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 val chunks = stream.incoming.toList() 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 0a0fd1ef6..db68d2cb7 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,6 +33,7 @@ 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). * @@ -49,6 +50,23 @@ interface GetClient { path: String, ): RequestHandle + /** + * Atomically open + enqueue + FIN N streams under a single hold of + * the connection lock. The send loop cannot interject between + * opens, so when it next drains it sees ALL N streams' data ready + * and packs them into coalesced packets instead of emitting one + * tiny packet per stream. + * + * Without this, the equivalent serial loop of [prepareRequest] + * yields between calls (lock release → send loop wakes → drains + * one stream → next prepareRequest acquires...) and we send one + * stream per packet — what cratered the multiplexing testcase. + */ + suspend fun prepareRequests( + authority: String, + paths: List, + ): List + /** Suspend until the server FINs the response stream associated with * [handle]. Returns the assembled response. */ suspend fun awaitResponse(handle: RequestHandle): GetResponse @@ -137,6 +155,19 @@ class Http3GetClient( return Http3RequestHandle(stream) } + override suspend fun prepareRequests( + authority: String, + paths: List, + ): List = + conn.lock.withLock { + paths.map { path -> + val stream = conn.openBidiStreamLocked() + stream.send.enqueue(encodeRequest(authority, path)) + stream.send.finish() + Http3RequestHandle(stream) + } + } + override suspend fun awaitResponse(handle: RequestHandle): GetResponse { val stream = (handle as Http3RequestHandle).stream val reader = Http3FrameReader() diff --git a/quic/interop/src/main/kotlin/com/vitorpamplona/quic/interop/runner/InteropClient.kt b/quic/interop/src/main/kotlin/com/vitorpamplona/quic/interop/runner/InteropClient.kt index 477d0b9db..39112ec8a 100644 --- a/quic/interop/src/main/kotlin/com/vitorpamplona/quic/interop/runner/InteropClient.kt +++ b/quic/interop/src/main/kotlin/com/vitorpamplona/quic/interop/runner/InteropClient.kt @@ -349,14 +349,16 @@ private fun runTransferTest( // by batching enqueues + single wake. val collected = mutableListOf>() urls.chunked(MULTIPLEX_PARALLELISM).forEach { chunk -> - val prepared = - chunk.map { url -> - url to client.prepareRequest(authority, url.path) - } + // Single lock-held batch open + enqueue. + // Without this, openBidiStream's per-call + // lock acquire / release lets the send loop + // interject between opens and drain one + // stream per packet. + val handles = client.prepareRequests(authority, chunk.map { it.path }) driver.wakeup() coroutineScope { val deferreds = - prepared.map { (url, handle) -> + chunk.zip(handles).map { (url, handle) -> async { val resp = withTimeoutOrNull(PER_STREAM_TIMEOUT_SEC * 1_000L) { 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 77dc85ca8..8c5a97cc1 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt @@ -706,22 +706,34 @@ class QuicConnection( * check capacity proactively if the caller wants to back-pressure rather * than throw. */ - suspend fun openBidiStream(): QuicStream = - lock.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 + 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)", + ) } + 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 + return stream + } /** * Allocate a new client-initiated unidirectional (write-only) stream.