fix(quic): tier-local round-robin so priority survives every drain
The previous priority-then-round-robin shape applied the rotating start-index globally over the sorted list, so the cross-tier order flipped on alternate drains: drain N had high-priority first, drain N+1 advanced streamRoundRobinStart and had low-priority first. The priority hint was silently defeated under any sustained traffic. Replace it with strict priority across tiers + round-robin only within each same-priority tier. Higher tiers always drain ahead of lower ones; same-priority peers continue to take turns via the existing rotating start. Default-priority callers see no behaviour change (single tier, identical rotation semantics). Tighten the test: drain twice and assert the higher-priority stream emits first on BOTH drains — the regression case that the single- drain version of the test missed. Add a regression guard for the same-priority round-robin so a future refactor can't silently serialise on the first stream. https://claude.ai/code/session_01KWdr4RjVvyYZfEuPVaQfUa
This commit is contained in:
Claude
+64
-48
@@ -410,58 +410,74 @@ private fun buildApplicationPacket(
|
|||||||
// insertion-ordered and stays in sync with the streams map.
|
// insertion-ordered and stays in sync with the streams map.
|
||||||
val streamsView = conn.streamsListLocked()
|
val streamsView = conn.streamsListLocked()
|
||||||
if (streamsView.isNotEmpty()) {
|
if (streamsView.isNotEmpty()) {
|
||||||
// Priority-then-round-robin: stable sortedByDescending preserves
|
// Strict priority across tiers, round-robin within each tier.
|
||||||
// insertion order within a tier, so same-priority streams keep
|
// Higher-priority streams (e.g. moq-lite newer-sequence group
|
||||||
// the rotating start-index round-robin behaviour. Higher-priority
|
// streams) ALWAYS drain ahead of lower-priority ones; the
|
||||||
// streams (e.g. moq-lite newer-sequence group streams) drain
|
// rotating start-index only rotates among same-priority peers.
|
||||||
// first under congestion. Default priority is 0; if every stream
|
// This is the spec-aligned shape — applying the rotation
|
||||||
// is at the default, iteration order matches pre-priority code.
|
// globally over the sorted list would flip cross-tier order on
|
||||||
// Cost: O(N log N) per drain pass and one transient list
|
// alternating drains, defeating the priority hint entirely.
|
||||||
// allocation. N is small (1–10 in the moq-lite audio path); if
|
//
|
||||||
// it ever grows enough to matter, switch to an indirect index
|
// Default priority is 0; if every stream is at the default, all
|
||||||
// sort or maintain an incrementally-sorted view on setPriority.
|
// streams form a single tier and iteration order matches the
|
||||||
|
// pre-priority round-robin behaviour exactly.
|
||||||
|
//
|
||||||
|
// Cost: O(N log N) per drain pass plus one transient sorted
|
||||||
|
// list. N is small (1–10 in the moq-lite audio path); if it
|
||||||
|
// ever grows enough to matter, switch to an indirect index sort
|
||||||
|
// or maintain an incrementally-sorted view on setPriority.
|
||||||
val sorted =
|
val sorted =
|
||||||
if (streamsView.size > 1) streamsView.sortedByDescending { it.priority } else streamsView
|
if (streamsView.size > 1) streamsView.sortedByDescending { it.priority } else streamsView
|
||||||
val start = conn.streamRoundRobinStart % sorted.size
|
val rotation = conn.streamRoundRobinStart
|
||||||
for (i in sorted.indices) {
|
var tierStart = 0
|
||||||
if (packetBudget <= 64) break
|
outer@ while (tierStart < sorted.size) {
|
||||||
val stream = sorted[(start + i) % sorted.size]
|
// Walk the contiguous run of same-priority streams.
|
||||||
val streamRemaining = (stream.sendCredit - stream.send.sentOffset).coerceAtLeast(0L)
|
val tierPriority = sorted[tierStart].priority
|
||||||
// Skip if both stream and connection have no credit; FIN-only
|
var tierEnd = tierStart + 1
|
||||||
// (zero-byte) chunks may still go through because they don't
|
while (tierEnd < sorted.size && sorted[tierEnd].priority == tierPriority) tierEnd++
|
||||||
// consume credit.
|
val tierSize = tierEnd - tierStart
|
||||||
if (streamRemaining <= 0L && connBudget <= 0L && !stream.send.finPending) continue
|
val tierRotation = if (tierSize > 1) rotation % tierSize else 0
|
||||||
val effectiveCap = minOf(streamRemaining, connBudget)
|
for (k in 0 until tierSize) {
|
||||||
val maxBytes =
|
if (packetBudget <= 64) break@outer
|
||||||
minOf(packetBudget - 32, effectiveCap.coerceAtMost(Int.MAX_VALUE.toLong()).toInt())
|
val stream = sorted[tierStart + ((tierRotation + k) % tierSize)]
|
||||||
val chunk = stream.send.takeChunk(maxBytes = maxBytes) ?: continue
|
val streamRemaining = (stream.sendCredit - stream.send.sentOffset).coerceAtLeast(0L)
|
||||||
if (chunk.data.isNotEmpty() || chunk.fin) {
|
// Skip if both stream and connection have no credit; FIN-only
|
||||||
frames +=
|
// (zero-byte) chunks may still go through because they don't
|
||||||
StreamFrame(
|
// consume credit.
|
||||||
streamId = stream.streamId,
|
if (streamRemaining <= 0L && connBudget <= 0L && !stream.send.finPending) continue
|
||||||
offset = chunk.offset,
|
val effectiveCap = minOf(streamRemaining, connBudget)
|
||||||
data = chunk.data,
|
val maxBytes =
|
||||||
fin = chunk.fin,
|
minOf(packetBudget - 32, effectiveCap.coerceAtMost(Int.MAX_VALUE.toLong()).toInt())
|
||||||
explicitLength = true,
|
val chunk = stream.send.takeChunk(maxBytes = maxBytes) ?: continue
|
||||||
)
|
if (chunk.data.isNotEmpty() || chunk.fin) {
|
||||||
// Step C of the deferred-follow-ups pass: track this
|
frames +=
|
||||||
// STREAM emission so RFC 9002 retransmit can re-queue
|
StreamFrame(
|
||||||
// the byte range on loss. SendBuffer.markLost (commit B)
|
streamId = stream.streamId,
|
||||||
// moves the range from in-flight back to the retransmit
|
offset = chunk.offset,
|
||||||
// queue, and the next takeChunk replays it.
|
data = chunk.data,
|
||||||
tokens +=
|
fin = chunk.fin,
|
||||||
RecoveryToken.Stream(
|
explicitLength = true,
|
||||||
streamId = stream.streamId,
|
)
|
||||||
offset = chunk.offset,
|
// Step C of the deferred-follow-ups pass: track this
|
||||||
length = chunk.data.size.toLong(),
|
// STREAM emission so RFC 9002 retransmit can re-queue
|
||||||
fin = chunk.fin,
|
// the byte range on loss. SendBuffer.markLost (commit B)
|
||||||
)
|
// moves the range from in-flight back to the retransmit
|
||||||
packetBudget -= chunk.data.size + 32
|
// queue, and the next takeChunk replays it.
|
||||||
connBudget -= chunk.data.size
|
tokens +=
|
||||||
conn.sendConnectionFlowConsumed += chunk.data.size
|
RecoveryToken.Stream(
|
||||||
|
streamId = stream.streamId,
|
||||||
|
offset = chunk.offset,
|
||||||
|
length = chunk.data.size.toLong(),
|
||||||
|
fin = chunk.fin,
|
||||||
|
)
|
||||||
|
packetBudget -= chunk.data.size + 32
|
||||||
|
connBudget -= chunk.data.size
|
||||||
|
conn.sendConnectionFlowConsumed += chunk.data.size
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
tierStart = tierEnd
|
||||||
}
|
}
|
||||||
conn.streamRoundRobinStart = (start + 1) % streamsView.size
|
conn.streamRoundRobinStart = (rotation + 1) % streamsView.size
|
||||||
}
|
}
|
||||||
|
|
||||||
if (frames.isEmpty()) return null
|
if (frames.isEmpty()) return null
|
||||||
|
|||||||
+68
-32
@@ -112,45 +112,81 @@ class QuicConnectionWriterTest {
|
|||||||
|
|
||||||
@Test
|
@Test
|
||||||
fun writer_drains_higher_priority_streams_before_lower_priority() {
|
fun writer_drains_higher_priority_streams_before_lower_priority() {
|
||||||
// T11.3 follow-up: the writer's drain loop must iterate streams
|
// T11.3 follow-up: priority must dominate iteration order on
|
||||||
// in descending priority order so moq-lite group streams with a
|
// EVERY drain, not just the first. The naive "sort by priority
|
||||||
// higher sequence number drain ahead of older ones under
|
// then apply the existing rotating start globally" shape looks
|
||||||
// congestion. Pre-fix this test, the writer iterated in stable
|
// right at a glance but the rotating start advances on every
|
||||||
// round-robin order regardless of priority — so a backlog of
|
// drain, so cross-tier ordering flips on alternate drains and
|
||||||
// retransmits on an older group could starve the listener of
|
// the priority hint is silently defeated. The correct shape is
|
||||||
// fresh frames. We pin the load-bearing invariant by inspecting
|
// strict priority across tiers + round-robin only within a
|
||||||
// StreamFrame order in a single emitted packet: low-priority
|
// tier, which we pin here by draining TWICE and asserting the
|
||||||
// stream is opened FIRST (so insertion order would normally win
|
// higher-priority stream's StreamFrame lands first in BOTH
|
||||||
// round-robin), but the high-priority stream's bytes must land
|
// packets. Low-priority stream is opened FIRST so insertion-
|
||||||
// earlier in the packet.
|
// order can't accidentally pass for priority ordering.
|
||||||
runBlocking {
|
runBlocking {
|
||||||
val (client, pipe) = connectedClient()
|
val (client, pipe) = connectedClient()
|
||||||
val low = client.openBidiStream()
|
val low = client.openBidiStream()
|
||||||
val high = client.openBidiStream()
|
val high = client.openBidiStream()
|
||||||
low.priority = 0
|
low.priority = 0
|
||||||
high.priority = 10
|
high.priority = 10
|
||||||
// Distinct payloads small enough to coexist in one packet.
|
|
||||||
val lowPayload = ByteArray(200) { 0xAA.toByte() }
|
|
||||||
val highPayload = ByteArray(200) { 0xBB.toByte() }
|
|
||||||
low.send.enqueue(lowPayload)
|
|
||||||
high.send.enqueue(highPayload)
|
|
||||||
|
|
||||||
val datagram = drainOutbound(client, nowMillis = 0L)
|
repeat(2) { round ->
|
||||||
assertNotNull(datagram, "drain must emit a packet")
|
low.send.enqueue(ByteArray(200) { 0xAA.toByte() })
|
||||||
val frames = pipe.decryptClientApplicationFrames(datagram)
|
high.send.enqueue(ByteArray(200) { 0xBB.toByte() })
|
||||||
assertNotNull(frames, "decrypt must succeed at the application level")
|
|
||||||
val streamFrames = frames.filterIsInstance<StreamFrame>()
|
val datagram = drainOutbound(client, nowMillis = 0L)
|
||||||
assertEquals(
|
assertNotNull(datagram, "drain on round $round must emit a packet")
|
||||||
2,
|
val frames = pipe.decryptClientApplicationFrames(datagram)
|
||||||
streamFrames.size,
|
assertNotNull(frames, "decrypt must succeed on round $round")
|
||||||
"expected one StreamFrame per stream in this drain, got $streamFrames",
|
val streamFrames = frames.filterIsInstance<StreamFrame>()
|
||||||
)
|
assertEquals(
|
||||||
assertEquals(
|
2,
|
||||||
high.streamId,
|
streamFrames.size,
|
||||||
streamFrames[0].streamId,
|
"expected one StreamFrame per stream on round $round, got $streamFrames",
|
||||||
"higher-priority stream must drain first; saw ${streamFrames.map { it.streamId }}",
|
)
|
||||||
)
|
assertEquals(
|
||||||
assertEquals(low.streamId, streamFrames[1].streamId)
|
high.streamId,
|
||||||
|
streamFrames[0].streamId,
|
||||||
|
"higher-priority stream must drain first on round $round; " +
|
||||||
|
"saw ${streamFrames.map { it.streamId }}",
|
||||||
|
)
|
||||||
|
assertEquals(low.streamId, streamFrames[1].streamId, "low second on round $round")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
fun writer_round_robins_within_a_priority_tier() {
|
||||||
|
// Regression guard for the tier-local round-robin: same-priority
|
||||||
|
// streams must still rotate so an early-opened stream doesn't
|
||||||
|
// monopolise a packet's stream-frame slot indefinitely. We open
|
||||||
|
// three streams at the default (0) priority and verify the
|
||||||
|
// rotating start advances by one per drain, matching the
|
||||||
|
// pre-priority behaviour.
|
||||||
|
runBlocking {
|
||||||
|
val (client, pipe) = connectedClient()
|
||||||
|
val a = client.openBidiStream()
|
||||||
|
val b = client.openBidiStream()
|
||||||
|
val c = client.openBidiStream()
|
||||||
|
// All default priority — single tier, three streams.
|
||||||
|
val expectedRotation =
|
||||||
|
listOf(
|
||||||
|
listOf(a.streamId, b.streamId, c.streamId),
|
||||||
|
listOf(b.streamId, c.streamId, a.streamId),
|
||||||
|
listOf(c.streamId, a.streamId, b.streamId),
|
||||||
|
)
|
||||||
|
for ((round, expected) in expectedRotation.withIndex()) {
|
||||||
|
a.send.enqueue(ByteArray(64) { 0xA1.toByte() })
|
||||||
|
b.send.enqueue(ByteArray(64) { 0xB2.toByte() })
|
||||||
|
c.send.enqueue(ByteArray(64) { 0xC3.toByte() })
|
||||||
|
|
||||||
|
val datagram = drainOutbound(client, nowMillis = 0L)
|
||||||
|
assertNotNull(datagram, "drain $round must emit a packet")
|
||||||
|
val frames = pipe.decryptClientApplicationFrames(datagram)
|
||||||
|
assertNotNull(frames, "decrypt must succeed on round $round")
|
||||||
|
val ids = frames.filterIsInstance<StreamFrame>().map { it.streamId }
|
||||||
|
assertEquals(expected, ids, "round $round round-robin order")
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user