perf(quic): outbound AEAD allocation, scratch reuse, sort skip + key-update + flow
Round 11 — six follow-ups across perf, correctness, and API hygiene.
* QuicStream.incoming: replace `consumeAsFlow()` with `flow { for (c in
incomingChannel) emit(c) }`. Pre-fix the consume-style flow
cancelled the underlying Channel when the collector terminated,
which coupled "application stopped reading" with "parser
INTERNAL_ERRORs the connection on next delivery". The new emit-only
flow leaves the channel open across collector cancellation, so
applications can stop reading temporarily and resume later
(sequential collects, not concurrent — that's still a race on the
channel iterator).
* RFC 9001 §6.1 client-initiated key update: the existing
[QuicConnection.initiateKeyUpdate] no longer requires the caller
to enforce spec invariants. Returns false if the handshake isn't
yet complete (§6.5: MUST NOT initiate before HANDSHAKE_DONE) or if
a previous rotation is still in flight (§6.4: MUST NOT initiate a
subsequent update until the previous one is confirmed). The parser
clears the [keyUpdateInProgress] flag on the first inbound packet
that AEAD-decrypts under the post-rotation live keys — the
confirmation signal that the peer has rolled forward.
* Aead.sealInto: new range + in-place seal that writes ciphertext+tag
DIRECTLY into a caller-supplied output buffer at a given offset.
Default impl falls back to sealRange + copy; JcaAesGcmAead overrides
to use Cipher.doFinal(input, inOff, inLen, output, outOff).
LongHeaderPacket.build / ShortHeaderPacket.build now pre-allocate
the final packet buffer in a single shot and have the AEAD write
ciphertext+tag in-place. Pre-fix every outbound packet allocated
4 ByteArrays (headerBytes, paddedPlaintext, ciphertext, concat
buffer); now ~2 (the final packet + the AEAD provider's internal
scratch).
* QuicConnectionWriter.drainOutbound: skip the
`active.sortedByDescending { priority }` allocation when EVERY
active stream shares the same priority — not just the
default-zero case. A homogeneous priority-7 workload now keeps
insertion order at no cost.
* QuicConnection.scratchAppFrames / scratchAppTokens: per-connection
reusable lists for buildApplicationPacket. Cleared at function
entry, re-used across drains under streamsLock's single-writer
guarantee. The tokens list is `.toList()`-snapshotted into the
SentPacket record before reuse, so retransmit dispatch is
unaffected. NOT applied to collectHandshakeLevelFrames because
its returned [HandshakeLevelContents] is held across two
buildLongHeaderFromFrames calls (natural-size + padded rebuild)
in drainOutbound.
* Removed dead `parts = mutableListOf<ByteArray>()` declaration at
the top of drainOutbound — never referenced; the actual datagram
assembly uses inline `listOfNotNull(...)` instead.
All 269 :quic:jvmTest tests pass. BUILD SUCCESSFUL in 43s.
https://claude.ai/code/session_01AhGvbMV8uPRse3TmAGaddM
This commit is contained in:
Claude
@@ -253,6 +253,18 @@ class QuicConnection(
|
||||
@Volatile
|
||||
internal var previousReceiveProtection: PacketProtection? = null
|
||||
|
||||
/**
|
||||
* RFC 9001 §6.4: an endpoint MUST NOT initiate a subsequent key update
|
||||
* unless the previous one has been confirmed by the peer. We treat
|
||||
* "confirmed" as having received any application-level packet whose
|
||||
* decrypt succeeded under the post-rotation (current) live keys —
|
||||
* meaning the peer has rolled forward in lockstep. Set true by
|
||||
* [initiateKeyUpdate], cleared by [QuicConnectionParser] on the
|
||||
* first short-header packet that AEAD-passes after the rotation.
|
||||
*/
|
||||
@Volatile
|
||||
internal var keyUpdateInProgress: Boolean = false
|
||||
|
||||
/**
|
||||
* RFC 9001 §4.10 — 0-RTT send-side packet protection. Installed when
|
||||
* the TLS layer derives `client_early_traffic_secret` (right after
|
||||
@@ -572,6 +584,22 @@ class QuicConnection(
|
||||
com.vitorpamplona.quic.connection.recovery
|
||||
.QuicLossDetection()
|
||||
|
||||
/**
|
||||
* Reusable scratch lists for [QuicConnectionWriter.buildApplicationPacket].
|
||||
* The function clears them at entry, fills them, encodes the
|
||||
* packet, snapshots the [scratchAppTokens] via `.toList()` for
|
||||
* the [SentPacket] record, and returns. The next drain (single-
|
||||
* threaded by [streamsLock]) then re-clears and re-uses.
|
||||
*
|
||||
* Skipped for [collectHandshakeLevelFrames]: that function returns
|
||||
* a [com.vitorpamplona.quic.connection.HandshakeLevelContents]
|
||||
* wrapping the lists, which the caller holds across two
|
||||
* `buildLongHeaderFromFrames` calls (natural-size + padded
|
||||
* rebuild). Reusing those would corrupt the rebuild path.
|
||||
*/
|
||||
internal val scratchAppFrames: MutableList<com.vitorpamplona.quic.frame.Frame> = mutableListOf()
|
||||
internal val scratchAppTokens: MutableList<com.vitorpamplona.quic.connection.recovery.RecoveryToken> = mutableListOf()
|
||||
|
||||
/**
|
||||
* RFC 9002 §6.2 Probe Timeout signalling. When the driver loop's
|
||||
* PTO timer fires (no ack-eliciting packet has been ACK'd in
|
||||
@@ -1845,12 +1873,14 @@ class QuicConnection(
|
||||
* outbound packet carries `KEY_PHASE = 1` and the peer is expected
|
||||
* to mirror back in the same phase.
|
||||
*
|
||||
* RFC 9001 §6.5 says an endpoint MUST NOT initiate a key update
|
||||
* before the handshake is confirmed (HANDSHAKE_DONE received). The
|
||||
* caller is responsible for that check; this method just performs
|
||||
* the rotation. §6.4 also forbids initiating a second update before
|
||||
* the current one has been confirmed (peer responds in matching
|
||||
* phase) — same caller contract.
|
||||
* Spec invariants enforced here (caller no longer responsible):
|
||||
* - RFC 9001 §6.5: returns false if [handshakeComplete] is not
|
||||
* yet true. Initiating before handshake confirmation is a
|
||||
* PROTOCOL_VIOLATION.
|
||||
* - RFC 9001 §6.4: returns false if a previous rotation is still
|
||||
* in flight ([keyUpdateInProgress]). The parser clears that
|
||||
* flag on the first inbound packet that AEAD-decrypts under
|
||||
* the post-rotation keys, signalling the peer rolled forward.
|
||||
*
|
||||
* Header-protection key is unchanged (RFC 9001 §6.1: HP key is NOT
|
||||
* rotated when keys are updated).
|
||||
@@ -1864,6 +1894,16 @@ class QuicConnection(
|
||||
* and server".
|
||||
*/
|
||||
fun initiateKeyUpdate(): Boolean {
|
||||
// RFC 9001 §6.5: handshake MUST be confirmed before initiating
|
||||
// a key update. We use [handshakeComplete] as the proxy —
|
||||
// application keys are installed and the peer's HANDSHAKE_DONE
|
||||
// has been processed.
|
||||
if (!handshakeComplete) return false
|
||||
// RFC 9001 §6.4: MUST NOT initiate a subsequent rotation until
|
||||
// the previous one is confirmed. The parser clears this flag
|
||||
// when it observes an inbound packet that AEAD-decrypts under
|
||||
// the post-rotation live keys.
|
||||
if (keyUpdateInProgress) return false
|
||||
val cs = appCipherSuite.takeIf { it != 0 } ?: return false
|
||||
val curRx = appReceiveSecret ?: return false
|
||||
val curTx = appSendSecret ?: return false
|
||||
@@ -1907,6 +1947,7 @@ class QuicConnection(
|
||||
appSendSecret = nextTxSecret
|
||||
currentReceiveKeyPhase = !currentReceiveKeyPhase
|
||||
currentSendKeyPhase = !currentSendKeyPhase
|
||||
keyUpdateInProgress = true
|
||||
qlogObserver.onKeyUpdated("client", EncryptionLevel.APPLICATION)
|
||||
qlogObserver.onKeyUpdated("server", EncryptionLevel.APPLICATION)
|
||||
return true
|
||||
|
||||
@@ -436,6 +436,16 @@ private fun feedShortHeaderPacket(
|
||||
if (rotateOnSuccess != null && parsed.packet.packetNumber > state.pnSpace.largestReceived) {
|
||||
conn.commitKeyUpdate(rotateOnSuccess)
|
||||
}
|
||||
// RFC 9001 §6.4: clear the in-flight-rotation gate once an inbound
|
||||
// packet decrypts under the live (post-rotation) keys. That confirms
|
||||
// the peer has rolled forward in lockstep, so it's safe to permit
|
||||
// a subsequent [QuicConnection.initiateKeyUpdate]. The `rotateOnSuccess
|
||||
// == null` branch above is the live-keys path (peer's key_phase bit
|
||||
// matched our [currentReceiveKeyPhase]); reaching that with a
|
||||
// successful parse is the confirmation signal.
|
||||
if (rotateOnSuccess == null && conn.keyUpdateInProgress) {
|
||||
conn.keyUpdateInProgress = false
|
||||
}
|
||||
state.pnSpace.observeInbound(parsed.packet.packetNumber, nowMillis)
|
||||
if (conn.qlogObserver !== com.vitorpamplona.quic.observability.QlogObserver.NoOp) {
|
||||
conn.qlogObserver.onPacketReceived(
|
||||
|
||||
+18
-5
@@ -74,8 +74,6 @@ fun drainOutbound(
|
||||
conn: QuicConnection,
|
||||
nowMillis: Long,
|
||||
): ByteArray? {
|
||||
val parts = mutableListOf<ByteArray>()
|
||||
|
||||
// Closing — emit a CONNECTION_CLOSE at the highest available level. The
|
||||
// datagram-build paths below MUST satisfy two RFC 9000 constraints we
|
||||
// got wrong before:
|
||||
@@ -601,7 +599,12 @@ private fun buildApplicationPacket(
|
||||
// settled mid-handshake (peer hasn't ACK'd anything yet), so the
|
||||
// walk is a no-op.
|
||||
conn.retireFullyDoneStreamsLocked()
|
||||
val frames = mutableListOf<Frame>()
|
||||
// Reuse per-connection scratch lists (cleared at entry) instead of
|
||||
// allocating fresh `mutableListOf<Frame>` / `mutableListOf<RecoveryToken>`
|
||||
// every drain — single-writer by [streamsLock], so the next drain's
|
||||
// clear() runs after this drain has already snapshotted the tokens
|
||||
// via `.toList()` into the SentPacket record.
|
||||
val frames = conn.scratchAppFrames.also { it.clear() }
|
||||
// Tokens collected in lock-step with [frames]: each retransmittable
|
||||
// frame contributes a [RecoveryToken] so the [SentPacket] recorded
|
||||
// at the bottom of this function can drive RFC 9002 loss
|
||||
@@ -610,7 +613,7 @@ private fun buildApplicationPacket(
|
||||
// are not retransmittable (DatagramFrame, StreamFrame for now) do
|
||||
// not contribute a token; the packet is still ack-eliciting and
|
||||
// tracked for loss-detection timing.
|
||||
val tokens = mutableListOf<RecoveryToken>()
|
||||
val tokens = conn.scratchAppTokens.also { it.clear() }
|
||||
|
||||
// RFC 9000 §17.2.3 — 0-RTT packets MUST NOT contain ACK frames. The
|
||||
// server cannot ACK 0-RTT-level packets because it has no way to
|
||||
@@ -725,7 +728,17 @@ private fun buildApplicationPacket(
|
||||
val sorted =
|
||||
when {
|
||||
active.size <= 1 -> active
|
||||
active.all { it.priority == 0 } -> active
|
||||
|
||||
// Single-pass uniform-priority check: if EVERY stream
|
||||
// shares the same priority (whether default 0 or
|
||||
// anything else), the sort is a no-op and we keep
|
||||
// insertion order for round-robin fairness. Pre-fix
|
||||
// we only short-circuited on `priority == 0`, so a
|
||||
// homogeneous `priority == 7` workload paid for an
|
||||
// O(N log N) sort despite the result being
|
||||
// observationally identical to insertion order.
|
||||
active.all { it.priority == active[0].priority } -> active
|
||||
|
||||
else -> active.sortedByDescending { it.priority }
|
||||
}
|
||||
val rotation = conn.streamRoundRobinStart
|
||||
|
||||
@@ -86,6 +86,41 @@ abstract class Aead {
|
||||
return seal(key, nonce, a, p)
|
||||
}
|
||||
|
||||
/**
|
||||
* Range + in-place [seal]: read `aad` and `plaintext` from sub-ranges
|
||||
* and write ciphertext+tag DIRECTLY into [output] at [outputOffset].
|
||||
* Returns the number of bytes written (== plaintextLength + tagLength).
|
||||
*
|
||||
* Saves another ByteArray allocation per outbound packet vs.
|
||||
* [sealRange] — the build path can pre-allocate one final packet
|
||||
* buffer and have the ciphertext land in-place rather than copying
|
||||
* a fresh `seal()` result. Combined with the AAD+plaintext range
|
||||
* inputs already handled by [sealRange], a complete outbound packet
|
||||
* goes from ~4 allocations (headerBytes / paddedPlaintext /
|
||||
* ciphertext / final concat) down to ~2 (the final packet buffer
|
||||
* and the AEAD provider's internal scratch).
|
||||
*
|
||||
* Default impl falls back to [sealRange] + copy; the JCA-backed
|
||||
* [com.vitorpamplona.quic.crypto.JcaAesGcmAead] overrides to use
|
||||
* `Cipher.doFinal(input, inOff, inLen, output, outOff)`.
|
||||
*/
|
||||
open fun sealInto(
|
||||
key: ByteArray,
|
||||
nonce: ByteArray,
|
||||
aad: ByteArray,
|
||||
aadOffset: Int,
|
||||
aadLength: Int,
|
||||
plaintext: ByteArray,
|
||||
plaintextOffset: Int,
|
||||
plaintextLength: Int,
|
||||
output: ByteArray,
|
||||
outputOffset: Int,
|
||||
): Int {
|
||||
val ct = sealRange(key, nonce, aad, aadOffset, aadLength, plaintext, plaintextOffset, plaintextLength)
|
||||
ct.copyInto(output, outputOffset)
|
||||
return ct.size
|
||||
}
|
||||
|
||||
/**
|
||||
* Range-based [open] — same semantics as [open] but reads `aad` and
|
||||
* `ciphertext` from sub-ranges. Default impl slices and delegates.
|
||||
|
||||
@@ -109,14 +109,28 @@ object LongHeaderPacket {
|
||||
}
|
||||
val headerBytes = w.toByteArray()
|
||||
|
||||
// Encrypt padded payload
|
||||
// Pre-allocate the final packet buffer in one shot and have the
|
||||
// AEAD seal write ciphertext+tag directly into it. Pre-fix this
|
||||
// path allocated `headerBytes`, `paddedPlaintext`, the
|
||||
// `aead.seal` return ByteArray, and the final concat buffer —
|
||||
// four ByteArrays per outbound packet. The single-buffer +
|
||||
// [Aead.sealInto] form below collapses the seal output and
|
||||
// concat into the same allocation.
|
||||
val nonce = aeadNonce(iv, plain.packetNumber)
|
||||
val ciphertext = aead.seal(key, nonce, headerBytes, paddedPlaintext)
|
||||
|
||||
// Concatenate header + ciphertext
|
||||
val packet = ByteArray(headerBytes.size + ciphertext.size)
|
||||
val packet = ByteArray(headerBytes.size + paddedPlaintext.size + aead.tagLength)
|
||||
headerBytes.copyInto(packet, 0)
|
||||
ciphertext.copyInto(packet, headerBytes.size)
|
||||
aead.sealInto(
|
||||
key = key,
|
||||
nonce = nonce,
|
||||
aad = packet,
|
||||
aadOffset = 0,
|
||||
aadLength = headerBytes.size,
|
||||
plaintext = paddedPlaintext,
|
||||
plaintextOffset = 0,
|
||||
plaintextLength = paddedPlaintext.size,
|
||||
output = packet,
|
||||
outputOffset = headerBytes.size,
|
||||
)
|
||||
|
||||
// Apply header protection. Sample is 16 bytes starting 4 bytes after pnOffset.
|
||||
val sampleStart = pnOffset + 4
|
||||
|
||||
@@ -75,12 +75,26 @@ object ShortHeaderPacket {
|
||||
// Trailing 0x00 bytes decode as PADDING frames (RFC 9000 §19.1).
|
||||
val paddedPlaintext = padForHeaderProtectionSample(plain.payload, pnLen)
|
||||
|
||||
// Same single-buffer + sealInto pattern as LongHeaderPacket.build:
|
||||
// pre-allocate the final packet and have the AEAD write
|
||||
// ciphertext+tag directly into it instead of allocating a fresh
|
||||
// `seal()` return + a concat buffer. Saves 2 ByteArrays per
|
||||
// outbound short-header packet.
|
||||
val nonce = aeadNonce(iv, plain.packetNumber)
|
||||
val ciphertext = aead.seal(key, nonce, headerBytes, paddedPlaintext)
|
||||
|
||||
val packet = ByteArray(headerBytes.size + ciphertext.size)
|
||||
val packet = ByteArray(headerBytes.size + paddedPlaintext.size + aead.tagLength)
|
||||
headerBytes.copyInto(packet, 0)
|
||||
ciphertext.copyInto(packet, headerBytes.size)
|
||||
aead.sealInto(
|
||||
key = key,
|
||||
nonce = nonce,
|
||||
aad = packet,
|
||||
aadOffset = 0,
|
||||
aadLength = headerBytes.size,
|
||||
plaintext = paddedPlaintext,
|
||||
plaintextOffset = 0,
|
||||
plaintextLength = paddedPlaintext.size,
|
||||
output = packet,
|
||||
outputOffset = headerBytes.size,
|
||||
)
|
||||
|
||||
val sampleStart = pnOffset + 4
|
||||
require(sampleStart + 16 <= packet.size) { "packet too short for HP sample" }
|
||||
|
||||
@@ -22,7 +22,7 @@ package com.vitorpamplona.quic.stream
|
||||
|
||||
import kotlinx.coroutines.channels.Channel
|
||||
import kotlinx.coroutines.flow.Flow
|
||||
import kotlinx.coroutines.flow.consumeAsFlow
|
||||
import kotlinx.coroutines.flow.flow
|
||||
|
||||
/**
|
||||
* One QUIC stream (bidirectional or unidirectional). Application code
|
||||
@@ -78,33 +78,35 @@ class QuicStream(
|
||||
* was discarded, leaving a hole in the stream that the application could
|
||||
* never know about.
|
||||
*
|
||||
* **Single-collector contract.** [consumeAsFlow] cancels the underlying
|
||||
* [Channel] when its collector terminates (either by exception or
|
||||
* explicit cancellation). After that point any subsequent
|
||||
* `trySend` from the parser fails, sets [overflowed] = true, and
|
||||
* the parser tears the whole connection down with
|
||||
* `INTERNAL_ERROR: stream … consumer overflowed`. So:
|
||||
* - Application code MUST collect [incoming] **at most once** per
|
||||
* stream and MUST hold the collect open until the stream
|
||||
* terminates (FIN, peer RESET_STREAM, or local
|
||||
* [stopSending] / [resetStream] decision).
|
||||
* - Cancelling the collector early — e.g. wrapping in
|
||||
* `withTimeout(...)` — is equivalent to telling the connection
|
||||
* to drop. If the application wants to stop receiving without
|
||||
* killing the connection, call [stopSending] FIRST, then let
|
||||
* the parser's RESET_STREAM-style teardown drain the channel
|
||||
* cleanly.
|
||||
* **Resilience to collector cancellation.** Pre-fix this exposed
|
||||
* [incomingChannel] via `consumeAsFlow()`, which cancels the
|
||||
* underlying [Channel] when its collector terminates. That coupled
|
||||
* "application stopped collecting" to "parser INTERNAL_ERRORs the
|
||||
* whole connection on next delivery" — cancelling the collector
|
||||
* early (e.g. `withTimeout(...)`) was effectively a request to
|
||||
* drop the connection. Rebuilt as `flow { for (c in
|
||||
* incomingChannel) emit(c) }`: a fresh emit-only Flow over the
|
||||
* channel iterator, with no `consume`-style ownership transfer.
|
||||
* Now collector cancellation just exits the flow without touching
|
||||
* the channel; the channel stays open, the parser keeps
|
||||
* delivering, and the application can re-collect later (a fresh
|
||||
* collector picks up at `channel.receive()` — i.e. from the
|
||||
* current head, not the start).
|
||||
*
|
||||
* This is a fragile coupling we accept for now because (a) every
|
||||
* production caller already follows the single-collector pattern,
|
||||
* and (b) widening the contract would require swapping
|
||||
* `consumeAsFlow` for a `MutableSharedFlow` with replay/buffer
|
||||
* semantics, which has its own back-pressure pitfalls. The
|
||||
* comment is here so a future refactor doesn't accidentally
|
||||
* loosen the contract.
|
||||
* Producer back-pressure unchanged: the channel buffer is still
|
||||
* 64 chunks, [trySend] still sets [overflowed] on saturation,
|
||||
* and the parser still closes the connection if the application
|
||||
* fails to drain. The looser contract just permits the previously-
|
||||
* disallowed pattern of "stop reading temporarily, then resume".
|
||||
*
|
||||
* Concurrent collectors are still NOT supported — two simultaneous
|
||||
* collects would race the channel iterator and each chunk goes to
|
||||
* exactly one of them non-deterministically. Sequential collects
|
||||
* (one finishes / cancels, then another starts) are the new
|
||||
* supported pattern.
|
||||
*/
|
||||
private val incomingChannel = Channel<ByteArray>(capacity = 64)
|
||||
val incoming: Flow<ByteArray> get() = incomingChannel.consumeAsFlow()
|
||||
val incoming: Flow<ByteArray> = flow { for (chunk in incomingChannel) emit(chunk) }
|
||||
|
||||
/**
|
||||
* True once a [deliverIncoming] call failed because the channel was
|
||||
|
||||
@@ -206,6 +206,56 @@ class JcaAesGcmAead(
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* In-place range [seal]. Same semantics as the parent's [sealInto],
|
||||
* but takes the JCA's native `Cipher.doFinal(input, inOff, inLen,
|
||||
* output, outOff)` path so ciphertext+tag land directly in
|
||||
* [output] with no intermediate allocation. Caller is expected to
|
||||
* have sized `output` large enough to hold
|
||||
* `plaintextLength + tagLength` bytes starting at `outputOffset`.
|
||||
*/
|
||||
override fun sealInto(
|
||||
key: ByteArray,
|
||||
nonce: ByteArray,
|
||||
aad: ByteArray,
|
||||
aadOffset: Int,
|
||||
aadLength: Int,
|
||||
plaintext: ByteArray,
|
||||
plaintextOffset: Int,
|
||||
plaintextLength: Int,
|
||||
output: ByteArray,
|
||||
outputOffset: Int,
|
||||
): Int =
|
||||
synchronized(this) {
|
||||
val reuse = recentEncryptNonces.any { it.contentEquals(nonce) }
|
||||
val cipher: Cipher
|
||||
if (reuse) {
|
||||
cipher = Cipher.getInstance("AES/GCM/NoPadding")
|
||||
cipher.init(Cipher.ENCRYPT_MODE, keySpec, GCMParameterSpec(128, nonce))
|
||||
} else {
|
||||
cipher = encryptCipher
|
||||
try {
|
||||
cipher.init(Cipher.ENCRYPT_MODE, keySpec, GCMParameterSpec(128, nonce))
|
||||
} catch (t: Throwable) {
|
||||
if (recentEncryptNonces.lastOrNull()?.contentEquals(nonce) == true) {
|
||||
recentEncryptNonces.removeLast()
|
||||
}
|
||||
throw t
|
||||
}
|
||||
}
|
||||
try {
|
||||
cipher.updateAAD(aad, aadOffset, aadLength)
|
||||
val written = cipher.doFinal(plaintext, plaintextOffset, plaintextLength, output, outputOffset)
|
||||
if (!reuse) rememberEncryptNonce(nonce)
|
||||
written
|
||||
} catch (t: Throwable) {
|
||||
if (!reuse && recentEncryptNonces.lastOrNull()?.contentEquals(nonce) == true) {
|
||||
recentEncryptNonces.removeLast()
|
||||
}
|
||||
throw t
|
||||
}
|
||||
}
|
||||
|
||||
private companion object {
|
||||
private const val NONCE_HISTORY_LIMIT = 8
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user