perf+test(quic): audit follow-ups for the multiplex lock fix
Audit of recent multiplex / PTO commits surfaced three concrete improvements: 1. **Pre-encode requests outside streamsLock** in both prepareRequests impls. QPACK encoding (Http3GetClient) and string formatting (HqInteropGetClient) are non-trivial under multiplex load — 1999 paths means we were holding streamsLock across all 32 chunks × ~2KB of QPACK encoding per chunk = ~64 KB of CPU work serialised against the send loop. Now done outside the lock. 2. **Fix O(N²) byte merging in MultiplexingRoundTripTest.** Previous shape allocated a new array per STREAM frame; for real-size payloads this would dominate test runtime. Switched to a per- stream MutableList<ByteArray> joined once at the end. 3. **Pin coalescing end-to-end in MultiplexingRoundTripTest.** Pre-existing test verified per-stream content arrived but said nothing about the wire shape. Added totalDatagrams ≤ 12 assertion so the matrix's "one stream per datagram" failure mode would break the test instead of passing silently. Audit also surfaced two non-actionable items, flagged for future cleanup but not changed in this commit: - LevelState.levelLock docstring claims writer/parser acquire it around sentPackets mutations; in practice neither does. The handlePtoFired call that takes levelLock currently serialises only against itself; SendBuffer's internal synchronized is what prevents the actual cryptoSend race. Kept the call (matches the documented design intent and future-proofs against the writer actually taking levelLock) but the docstring is stale. - openBidiStreamLocked's check(streamsLock.isLocked) catches "no lock" and "wrong lock" callers but not "another coroutine holds streamsLock and I'm calling without holding it" — kotlinx Mutex doesn't expose owner-aware checks without an explicit owner arg we don't pass. Acceptable since the bug we just fixed and any future regression in the same shape are caught. https://claude.ai/code/session_01HcvfQq1ttPV9PkRoJb4nyT
This commit is contained in:
+27
-9
@@ -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<Long, ByteArray>()
|
||||
val seenRequests = mutableMapOf<Long, MutableList<ByteArray>>()
|
||||
val seenFin = mutableSetOf<Long>()
|
||||
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,
|
||||
|
||||
Reference in New Issue
Block a user