perf(quic): round-4 perf-audit fixes — ACK gating, flow-control dirty-set, streams list view
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<QuicStream> 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
This commit is contained in:
Claude
@@ -101,6 +101,16 @@ class QuicConnection(
|
||||
private set
|
||||
|
||||
private val streams = mutableMapOf<Long, QuicStream>()
|
||||
|
||||
/**
|
||||
* 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<QuicStream>()
|
||||
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<Long, QuicStream> = 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<QuicStream> = streamsList
|
||||
|
||||
/** Caller must hold [lock]. Pending datagram queue for the driver's send loop. */
|
||||
internal fun pendingDatagramsLocked(): ArrayDeque<ByteArray> = pendingDatagrams
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
+32
-7
@@ -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
|
||||
|
||||
@@ -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<AckRange>()
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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))
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user