From 920b36cdd6433f323cc678a8f684720ef483bb67 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 26 Apr 2026 00:49:20 +0000 Subject: [PATCH] =?UTF-8?q?perf(quic):=20round-4=20perf-audit=20fixes=20?= =?UTF-8?q?=E2=80=94=20ACK=20gating,=20flow-control=20dirty-set,=20streams?= =?UTF-8?q?=20list=20view?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Four perf wins from the round-4 audit, all in the steady-state hot path (audio rooms run at ~50 datagrams/sec/participant). Perf #1 — ACK frame gating (saves a frame + bandwidth on every drain): AckTracker.buildAckFrame now returns null when no new ack-eliciting packet has arrived since the last build. Pre-fix the writer emitted a redundant ACK frame on every outbound packet (~50/sec each direction) even when the only inbound traffic since last drain was ACK-only. RFC 9000 §13.2 only requires ACKs in response to ack-eliciting packets within max_ack_delay; gating on ackElicitingPending satisfies that without delay-timer machinery. Perf #11 — AckTracker.purgeBelow short-circuit: Common case: peer ACKs a high PN, we already pruned below it. Pre-fix triggered a full ListIterator walk anyway. Now bails out when the tail's start is already above the threshold. Perf #9 — Flow-control dirty-set: QuicStream gains a receiveDirtyForFlowControl flag set by the parser when readContiguous advances the frontier. The writer's appendFlowControlUpdates now skips per-direction-window lookup + threshold comparison for streams whose flag is unset. Big win for multi-stream sessions (audio rooms with N×M streams per participant). Perf #10 — Streams list view: QuicConnection maintains a parallel insertion-ordered List alongside the streams Map. Writer's round-robin scan reads the list directly instead of allocating `entries.toList()` per drain. No removal path exists today; the insert points (openBidi/UniStream, getOrCreatePeerStreamLocked) update both. AckTrackerGatingTest pins the new gating contract: first build returns a frame; second build without new reception returns null; subsequent ack-eliciting reception re-arms; non-ack-eliciting receptions alone don't. https://claude.ai/code/session_01EC1tfXfap8k8GyKvrxkxZx --- .../quic/connection/QuicConnection.kt | 21 +++++ .../quic/connection/QuicConnectionParser.kt | 5 ++ .../quic/connection/QuicConnectionWriter.kt | 39 ++++++-- .../vitorpamplona/quic/recovery/AckTracker.kt | 23 ++++- .../vitorpamplona/quic/stream/QuicStream.kt | 11 +++ .../quic/recovery/AckTrackerGatingTest.kt | 89 +++++++++++++++++++ 6 files changed, 180 insertions(+), 8 deletions(-) create mode 100644 quic/src/commonTest/kotlin/com/vitorpamplona/quic/recovery/AckTrackerGatingTest.kt 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 9a807f800..2cdbed14d 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt @@ -101,6 +101,16 @@ class QuicConnection( private set private val streams = mutableMapOf() + + /** + * Round-4 perf #10: parallel insertion-ordered list of streams so the + * writer's round-robin scan can index by position without + * `streams.entries.toList()` allocating per drain. Streams are only ever + * added (no removal in the current model), so the two stay in sync as + * long as `getOrCreatePeerStreamLocked` and `openBidi/UniStream` append + * to both. + */ + private val streamsList = mutableListOf() private var nextLocalBidiIndex: Long = 0L private var nextLocalUniIndex: Long = 0L @@ -340,6 +350,7 @@ class QuicConnection( stream.sendCredit = peerTransportParameters?.initialMaxStreamDataBidiRemote ?: config.initialMaxStreamDataBidiRemote stream.receiveLimit = config.initialMaxStreamDataBidiLocal streams[id] = stream + streamsList += stream stream } @@ -357,6 +368,7 @@ class QuicConnection( stream.sendCredit = peerTransportParameters?.initialMaxStreamDataUni ?: config.initialMaxStreamDataUni stream.receiveLimit = 0L // can't receive streams[id] = stream + streamsList += stream stream } @@ -492,6 +504,7 @@ class QuicConnection( StreamId.Kind.CLIENT_BIDI -> config.initialMaxStreamDataBidiLocal } streams[id] = stream + streamsList += stream newPeerStreams.addLast(stream) // Wake any awaitIncomingPeerStream caller. trySend on a CONFLATED // channel can never fail in steady state. @@ -518,6 +531,14 @@ class QuicConnection( /** Caller must hold [lock]. Snapshot of streams for the driver's send loop. */ internal fun streamsLocked(): Map = streams + /** + * Insertion-ordered list view used by the writer's round-robin scan. + * Stays in sync with [streams] because the only mutation paths + * (openBidi/UniStream, getOrCreatePeerStreamLocked) append to both. No + * remove path exists today; if/when one is added it MUST update both. + */ + internal fun streamsListLocked(): List = streamsList + /** Caller must hold [lock]. Pending datagram queue for the driver's send loop. */ internal fun pendingDatagramsLocked(): ArrayDeque = pendingDatagrams diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionParser.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionParser.kt index af8b5803c..97889159d 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionParser.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionParser.kt @@ -218,6 +218,11 @@ private fun dispatchFrames( stream.receive.insert(frame.offset, frame.data, frame.fin) val data = stream.receive.readContiguous() if (data.isNotEmpty()) { + // Round-4 perf #9: mark the stream as needing a flow- + // control re-credit check. Writer's + // appendFlowControlUpdates consults this flag instead of + // walking every open stream on every drain. + stream.receiveDirtyForFlowControl = true val delivered = stream.deliverIncoming(data) if (!delivered) { // Audit-4 #3: incoming channel saturated. Closing the diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt index 4248ffce7..70eb5cc74 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt @@ -268,12 +268,15 @@ private fun buildApplicationPacket( val connRemaining = (conn.sendConnectionFlowCredit - conn.sendConnectionFlowConsumed).coerceAtLeast(0L) var connBudget = connRemaining - val streamsList = conn.streamsLocked().entries.toList() - if (streamsList.isNotEmpty()) { - val start = conn.streamRoundRobinStart % streamsList.size - for (i in streamsList.indices) { + // Round-4 perf #10: use the connection's pre-built list view instead of + // allocating a fresh `entries.toList()` per drain. The list is + // insertion-ordered and stays in sync with the streams map. + val streamsView = conn.streamsListLocked() + if (streamsView.isNotEmpty()) { + val start = conn.streamRoundRobinStart % streamsView.size + for (i in streamsView.indices) { if (packetBudget <= 64) break - val (id, stream) = streamsList[(start + i) % streamsList.size] + val stream = streamsView[(start + i) % streamsView.size] val streamRemaining = (stream.sendCredit - stream.send.sentOffset).coerceAtLeast(0L) // Skip if both stream and connection have no credit; FIN-only // (zero-byte) chunks may still go through because they don't @@ -284,13 +287,20 @@ private fun buildApplicationPacket( minOf(packetBudget - 32, effectiveCap.coerceAtMost(Int.MAX_VALUE.toLong()).toInt()) val chunk = stream.send.takeChunk(maxBytes = maxBytes) ?: continue if (chunk.data.isNotEmpty() || chunk.fin) { - frames += StreamFrame(streamId = id, offset = chunk.offset, data = chunk.data, fin = chunk.fin, explicitLength = true) + frames += + StreamFrame( + streamId = stream.streamId, + offset = chunk.offset, + data = chunk.data, + fin = chunk.fin, + explicitLength = true, + ) packetBudget -= chunk.data.size + 32 connBudget -= chunk.data.size conn.sendConnectionFlowConsumed += chunk.data.size } } - conn.streamRoundRobinStart = (start + 1) % streamsList.size + conn.streamRoundRobinStart = (start + 1) % streamsView.size } if (frames.isEmpty()) return null @@ -319,9 +329,20 @@ private fun appendFlowControlUpdates( ) { val cfg = conn.config var totalRecvAdvanced = 0L + // Round-4 perf #9: only walk streams flagged by the parser since the last + // drain. Streams whose receive frontier hasn't advanced cannot need a + // new MAX_STREAM_DATA frame, so iterating them is wasted work. The + // connection-level totalRecvAdvanced sum still requires looking at each + // stream's contiguousEnd, but only when the dirty flag is set. for ((id, stream) in conn.streamsLocked()) { val rcv = stream.receive.contiguousEnd() if (rcv == 0L) continue + if (!stream.receiveDirtyForFlowControl) { + // Stream has data but nothing changed since last drain — skip the + // per-direction window lookup and the comparison. + totalRecvAdvanced += rcv + continue + } // Pick the per-direction window matching the stream kind so a // uni-only deployment with bidi=0 doesn't accidentally use the bidi // window and vice versa. @@ -343,6 +364,10 @@ private fun appendFlowControlUpdates( frames += MaxStreamDataFrame(id, newLimit) } } + // Clear the dirty flag once we've considered this stream — even if we + // didn't emit a new MAX_STREAM_DATA, the threshold-check work doesn't + // need to repeat until more bytes arrive. + stream.receiveDirtyForFlowControl = false totalRecvAdvanced += rcv } // Connection-level: only re-grant when the new total would exceed our diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/recovery/AckTracker.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/recovery/AckTracker.kt index 1e56ec8be..90e76d9ed 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/recovery/AckTracker.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/recovery/AckTracker.kt @@ -92,6 +92,12 @@ class AckTracker { */ fun purgeBelow(threshold: Long) { if (threshold <= 0L) return + if (ranges.isEmpty()) return + // Round-4 perf #11: short-circuit when nothing in the tail is below + // the threshold (the common case for steady receivers — purgeBelow + // is called per inbound ACK). Pre-fix every ACK triggered a full + // ListIterator walk even when there was nothing to remove. + if (ranges.last().start >= threshold) return // ranges are descending by lo; drop the tail. val it = ranges.listIterator(ranges.size) while (it.hasPrevious()) { @@ -108,12 +114,27 @@ class AckTracker { fun largestReceived(): Long = if (ranges.isEmpty()) -1L else ranges[0].endInclusive - /** Build an ACK frame covering everything we've received. */ + /** + * Build an ACK frame covering everything we've received, OR null if nothing + * new ack-eliciting has arrived since the last call. + * + * Round-4 perf #1: pre-fix this returned non-null whenever the range list + * was non-empty, so EVERY outbound packet (~50/sec for an audio room) + * carried a redundant ACK. RFC 9000 §13.2 only requires ACKs in response + * to ack-eliciting packets, within max_ack_delay. Gating on + * [ackElicitingPending] satisfies the RFC requirement and saves a varint- + * heavy frame per drain. + * + * Note: the flag is cleared by this method only when an ACK is actually + * built. Pre-fix the flag was cleared even when buildAckFrame returned a + * stale frame, which compounded the problem. + */ fun buildAckFrame( nowMillis: Long, ackDelayExponent: Int = 3, ): AckFrame? { if (ranges.isEmpty()) return null + if (!ackElicitingPending) return null val largest = ranges[0].endInclusive val firstRangeLength = largest - ranges[0].start val rest = mutableListOf() diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/QuicStream.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/QuicStream.kt index 487f92708..e9c29c985 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/QuicStream.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/QuicStream.kt @@ -69,6 +69,17 @@ class QuicStream( var receiveLimit: Long = 0L internal set + /** + * Marker the parser sets whenever [receive.contiguousEnd] advances; the + * writer's appendFlowControlUpdates consumes it to skip streams that + * haven't received any new bytes since the last MAX_STREAM_DATA emission. + * + * Pre-fix the writer iterated EVERY open stream on every drain + * (audit-4 perf #9 — O(streams) × ~50 drains/sec; significant for audio + * rooms with many WT streams). + */ + internal var receiveDirtyForFlowControl: Boolean = false + /** True once we've FIN'd our write side and the peer FIN'd theirs. */ val isClosed: Boolean get() = send.finSent && receive.finReceived diff --git a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/recovery/AckTrackerGatingTest.kt b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/recovery/AckTrackerGatingTest.kt new file mode 100644 index 000000000..3ef84776e --- /dev/null +++ b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/recovery/AckTrackerGatingTest.kt @@ -0,0 +1,89 @@ +/* + * Copyright (c) 2025 Vitor Pamplona + * + * Permission is hereby granted, free of charge, to any person obtaining a copy of + * this software and associated documentation files (the "Software"), to deal in + * the Software without restriction, including without limitation the rights to use, + * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the + * Software, and to permit persons to whom the Software is furnished to do so, + * subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS + * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR + * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN + * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION + * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + */ +package com.vitorpamplona.quic.recovery + +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertNotNull +import kotlin.test.assertNull + +/** + * Round-4 perf #1 regression: [AckTracker.buildAckFrame] must return null when + * nothing new ack-eliciting has arrived since the last call. Pre-fix every + * outbound packet (~50/sec for an audio room) carried a redundant ACK frame. + * + * The contract: + * 1. First ack-eliciting reception → buildAckFrame returns a frame; flag clears. + * 2. Second buildAckFrame without new reception → returns null. + * 3. Subsequent ack-eliciting reception → next buildAckFrame returns frame. + * 4. Non-ack-eliciting receptions (PADDING-only, ACK-only) DO NOT trigger a + * new ACK on their own. + */ +class AckTrackerGatingTest { + @Test + fun first_build_after_ack_eliciting_returns_frame() { + val tracker = AckTracker() + tracker.receivedPacket(packetNumber = 5L, ackEliciting = true, receivedAtMillis = 1000L) + assertNotNull(tracker.buildAckFrame(nowMillis = 1010L)) + } + + @Test + fun second_build_without_new_reception_returns_null() { + val tracker = AckTracker() + tracker.receivedPacket(packetNumber = 5L, ackEliciting = true, receivedAtMillis = 1000L) + tracker.buildAckFrame(nowMillis = 1010L) // first ack — clears flag + assertNull( + tracker.buildAckFrame(nowMillis = 1020L), + "no new ack-eliciting reception → no redundant ACK", + ) + } + + @Test + fun build_re_arms_after_subsequent_ack_eliciting_reception() { + val tracker = AckTracker() + tracker.receivedPacket(packetNumber = 5L, ackEliciting = true, receivedAtMillis = 1000L) + tracker.buildAckFrame(nowMillis = 1010L) + // New ack-eliciting reception arrives. + tracker.receivedPacket(packetNumber = 6L, ackEliciting = true, receivedAtMillis = 1100L) + val ack = tracker.buildAckFrame(nowMillis = 1110L) + assertNotNull(ack, "fresh ack-eliciting reception must re-arm the gate") + assertEquals(6L, ack.largestAcknowledged) + } + + @Test + fun non_ack_eliciting_reception_alone_does_not_arm_the_gate() { + // Per RFC 9000 §13.2.1 we don't have to ACK ACK-only packets. The + // tracker still records the PN (so future ACKs cover it), but the + // gate doesn't open until something ack-eliciting arrives. + val tracker = AckTracker() + tracker.receivedPacket(packetNumber = 1L, ackEliciting = false, receivedAtMillis = 1000L) + assertNull( + tracker.buildAckFrame(nowMillis = 1010L), + "non-ack-eliciting reception alone must not produce an ACK", + ) + } + + @Test + fun empty_tracker_returns_null() { + val tracker = AckTracker() + assertNull(tracker.buildAckFrame(nowMillis = 1000L)) + } +}