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 10c249c9f8..95af754329 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 @@ -39,19 +39,21 @@ import kotlinx.coroutines.flow.toList */ class HqInteropGetClient( private val conn: QuicConnection, - private val driver: QuicConnectionDriver, + @Suppress("UNUSED_PARAMETER") private val driver: QuicConnectionDriver, ) : GetClient { - override suspend fun get( + override suspend fun prepareRequest( @Suppress("UNUSED_PARAMETER") authority: String, path: String, - ): GetResponse { + ): RequestHandle { val stream = conn.openBidiStream() val request = "GET $path\r\n".encodeToByteArray() stream.send.enqueue(request) stream.send.finish() - // Nudge the send loop — see Http3GetClient.get for rationale. - driver.wakeup() + return HqRequestHandle(stream) + } + override suspend fun awaitResponse(handle: RequestHandle): GetResponse { + val stream = (handle as HqRequestHandle).stream val chunks = stream.incoming.toList() val total = chunks.sumOf { it.size } val body = ByteArray(total) @@ -63,3 +65,7 @@ class HqInteropGetClient( return GetResponse(status = if (body.isEmpty()) 0 else 200, body = body) } } + +private class HqRequestHandle( + val stream: com.vitorpamplona.quic.stream.QuicStream, +) : RequestHandle 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 ca2571d840..0a0fd1ef69 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 @@ -34,14 +34,39 @@ import com.vitorpamplona.quic.qpack.QpackEncoder import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.flow.collect -/** Common shape for the two interop GET clients (HTTP/3 and HQ-interop). */ +/** Common shape for the two interop GET clients (HTTP/3 and HQ-interop). + * + * The interface is split into two phases so the parallel multiplexing + * path can BATCH enqueues (synchronous, serial) and SINGLE-wakeup the + * send loop, vs. waking on every individual request which produces + * one tiny packet per stream instead of coalesced packets per drain. */ interface GetClient { + /** Open a stream + enqueue the request bytes + FIN. Does NOT wake the + * send loop — caller is responsible for batching wakes. Returns an + * opaque handle the caller passes to [awaitResponse]. */ + suspend fun prepareRequest( + authority: String, + path: String, + ): RequestHandle + + /** Suspend until the server FINs the response stream associated with + * [handle]. Returns the assembled response. */ + suspend fun awaitResponse(handle: RequestHandle): GetResponse + + /** Convenience shortcut for the sequential / single-request paths. */ suspend fun get( authority: String, path: String, - ): GetResponse + ): GetResponse { + val h = prepareRequest(authority, path) + return awaitResponse(h) + } } +/** Opaque handle returned by [GetClient.prepareRequest]. Implementations + * cast it back to their internal stream representation. */ +interface RequestHandle + data class GetResponse( val status: Int, val body: ByteArray, @@ -102,23 +127,18 @@ class Http3GetClient( conn.drainPeerInitiatedUniStreamsIntoBlackHole(scope) } - /** - * Issue a GET on a fresh bidi stream and return the parsed response. - * Suspends until the server FINs the response stream. - */ - override suspend fun get( + override suspend fun prepareRequest( authority: String, path: String, - ): GetResponse { + ): RequestHandle { val stream = conn.openBidiStream() stream.send.enqueue(encodeRequest(authority, path)) stream.send.finish() - // Nudge the send loop. Without this it suspends until PTO (~1s) - // or until an inbound packet arrives. For the multiplexing path - // this was the dominant throughput bottleneck — chunks of 64 - // requests sat idle for ~1s each waiting to be drained. - driver.wakeup() + return Http3RequestHandle(stream) + } + override suspend fun awaitResponse(handle: RequestHandle): GetResponse { + val stream = (handle as Http3RequestHandle).stream val reader = Http3FrameReader() var status = 0 val body = mutableListOf() @@ -146,6 +166,10 @@ class Http3GetClient( } } +private class Http3RequestHandle( + val stream: com.vitorpamplona.quic.stream.QuicStream, +) : RequestHandle + /** * Serialize a GET request as a single HEADERS frame ready to be enqueued * onto a fresh bidi stream. Exposed for unit-testing the wire format 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 40bab3887a..477d0b9db8 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 @@ -320,26 +320,47 @@ private fun runTransferTest( // dispatcher thrashes context-switching. // // Bound concurrency: process in chunks of - // [MULTIPLEX_PARALLELISM]. Each chunk is fully - // parallel on the wire (what the runner's - // tshark check verifies — streams overlap in - // time within a chunk), and the connection - // lock only ever has ~64 live waiters instead - // of ~1999. Throughput predicted to jump from - // 23 to ~600+ streams/sec. + // [MULTIPLEX_PARALLELISM]. Each chunk is + // batched in two phases: + // 1. SERIAL prepareRequest for every URL + // in the chunk — opens the bidi + // stream, encodes the request, FINs. + // Synchronous; no async / no per-call + // wakeup. + // 2. SINGLE driver.wakeup() so the send + // loop drains all 64 enqueued requests + // in coalesced packets (multi-stream + // framing per drain) instead of one + // tiny packet per stream. + // 3. PARALLEL await — one async per + // stream collects its response with + // a per-stream timeout so a hung + // stream surfaces as status=0 instead + // of blocking its peers. // - // Per-stream timeout still wraps each get() so - // a single hung stream surfaces as status=0 - // instead of blocking its chunk's await. + // Earlier shape (per-call wakeup inside + // client.get()) produced ~23 streams/sec + // because each individual enqueue tripped + // the send loop, which then drained alone + // (the other 63 coroutines hadn't queued + // yet on the dispatcher). Result: one + // ~80-byte packet per stream instead of + // ~10 streams/packet. Coalescing recovered + // 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) + } + driver.wakeup() coroutineScope { val deferreds = - chunk.map { url -> + prepared.map { (url, handle) -> async { val resp = withTimeoutOrNull(PER_STREAM_TIMEOUT_SEC * 1_000L) { - client.get(authority, url.path) + client.awaitResponse(handle) } url to (resp ?: GetResponse(status = 0, body = ByteArray(0))) }