fix(quic): batch openBidiStream under one lock hold for multiplexing
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
This commit is contained in:
+15
@@ -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<String>,
|
||||
): List<RequestHandle> =
|
||||
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()
|
||||
|
||||
@@ -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<String>,
|
||||
): List<RequestHandle>
|
||||
|
||||
/** 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<String>,
|
||||
): List<RequestHandle> =
|
||||
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()
|
||||
|
||||
@@ -349,14 +349,16 @@ private fun runTransferTest(
|
||||
// by batching enqueues + single wake.
|
||||
val collected = mutableListOf<Pair<URI, GetResponse>>()
|
||||
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) {
|
||||
|
||||
Reference in New Issue
Block a user