nick/amethyst
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

fix(quic-interop): batch enqueue + single wakeup per multiplex chunk

Browse Source 
Diagnosis from qlog timeline pattern: send packets clustered ~37ms
apart (sim RTT) but each cluster contained only ONE stream's data
(80-byte packets despite 1452-byte capacity). 1457 GETs in 58s, ~25
streams/sec.

Root cause: race between client.get()'s per-call driver.wakeup() and
the dispatcher scheduling the OTHER 63 coroutines. Sequence:
  1. c1 acquires conn lock, enqueues request, wakes send loop, releases
  2. Send loop wakes, queues for lock — other 63 coroutines haven't
     started yet (dispatcher hasn't picked them up)
  3. Send loop acquires lock alone, drains c1's data into one tiny
     packet, releases
  4. c2 finally starts, acquires, enqueues, wakes...
  → one stream per packet, no coalescing

Fix: split GetClient.get() into prepareRequest (open + enqueue + FIN,
synchronous, no wake) and awaitResponse (collect, async). Multiplex
chunk loop now:
  Phase 1: serial prepareRequest for every URL in the chunk (64 in
           sequence, each adding to send buffers)
  Phase 2: SINGLE driver.wakeup() — by now all 64 streams have data
           queued; send loop drains them all in coalesced packets
  Phase 3: parallel awaitResponse with per-stream timeout

Predicted throughput jump: 25 streams/sec → ~1000+/sec (sim RTT-bound
at ~30ms per round trip = 64 streams per RTT = 2100/sec ceiling).

Single-request paths (transfer / chacha20 / etc) keep using the
default GetClient.get() which still wraps prepare+await; no behavioral
change there.

https://claude.ai/code/session_01HcvfQq1ttPV9PkRoJb4nyT
This commit is contained in:
Claude
2026-05-07 01:57:47 +00:00
parent 0cc577f0fb
commit bc19e90c17
3 changed files with 81 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files
quic/interop/src/main/kotlin/com/vitorpamplona/quic/interop/runner/HqInteropGetClient.kt
+11 -5
View File
@@ -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
quic/interop/src/main/kotlin/com/vitorpamplona/quic/interop/runner/Http3GetClient.kt
+37 -13
View File
@@ -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<ByteArray>()
@@ -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
quic/interop/src/main/kotlin/com/vitorpamplona/quic/interop/runner/InteropClient.kt
+33 -12
View File
@@ -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<Pair<URI, GetResponse>>()
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)))
}