perf(quic): lock-free hot paths — ThreadLocal Cipher, AtomicReference close, @Volatile getters
Audit of blocking and synchronized code in the QUIC module surfaced four
hot-path wins, all verified against :quic:jvmTest.
- PlatformCrypto: header-protection AES-ECB now uses a per-thread cached
Cipher. Previously every inbound and outbound packet paid for
Cipher.getInstance("AES/ECB/NoPadding") provider lookup. ThreadLocal is
safe because the call is stateless — every invocation re-init's with the
caller-supplied key.
- JdkCertificateValidator: gate the SAN-side InetAddress.getByName behind
looksLikeIpLiteral so a malformed cert with a hostname in a type 7 SAN
cannot trigger DNS resolution on the TLS validation path.
- QuicConnectionDriver.close: replace synchronized(this) double-checked
init with AtomicReference<Job?> + compareAndSet on a CoroutineStart.LAZY
job. Lock-free, removes a synchronized from commonMain, preserves the
original "first caller wins, second awaits same Job" contract.
- SendBuffer: mark _nextOffset, nextSendOffset, _finPending, _finSent,
_finAcked @Volatile and drop synchronized from their single-field
getters (nextOffset, sentOffset, finPending, finSent, finAcked). The
compound-formula readableBytes getter still synchronizes.
https://claude.ai/code/session_01CXTjnuHKCNXDmpfyKxgG3V
This commit is contained in:
+48
-35
@@ -31,6 +31,8 @@ import kotlinx.coroutines.joinAll
|
|||||||
import kotlinx.coroutines.launch
|
import kotlinx.coroutines.launch
|
||||||
import kotlinx.coroutines.sync.withLock
|
import kotlinx.coroutines.sync.withLock
|
||||||
import kotlinx.coroutines.withTimeoutOrNull
|
import kotlinx.coroutines.withTimeoutOrNull
|
||||||
|
import kotlin.concurrent.atomics.AtomicReference
|
||||||
|
import kotlin.concurrent.atomics.ExperimentalAtomicApi
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Owns the UDP socket and runs the read + send loops for a [QuicConnection].
|
* Owns the UDP socket and runs the read + send loops for a [QuicConnection].
|
||||||
@@ -48,6 +50,7 @@ import kotlinx.coroutines.withTimeoutOrNull
|
|||||||
* and [QuicConnection.openBidiStream]/[com.vitorpamplona.quic.stream.SendBuffer.enqueue])
|
* and [QuicConnection.openBidiStream]/[com.vitorpamplona.quic.stream.SendBuffer.enqueue])
|
||||||
* call [wakeup] to nudge the send loop.
|
* call [wakeup] to nudge the send loop.
|
||||||
*/
|
*/
|
||||||
|
@OptIn(ExperimentalAtomicApi::class)
|
||||||
class QuicConnectionDriver(
|
class QuicConnectionDriver(
|
||||||
val connection: QuicConnection,
|
val connection: QuicConnection,
|
||||||
private val socket: UdpSocket,
|
private val socket: UdpSocket,
|
||||||
@@ -94,17 +97,20 @@ class QuicConnectionDriver(
|
|||||||
* to poll `connection.status == CLOSED` and trust that the rest of
|
* to poll `connection.status == CLOSED` and trust that the rest of
|
||||||
* the cleanup eventually settled.
|
* the cleanup eventually settled.
|
||||||
*/
|
*/
|
||||||
internal val closeTeardownJob: Job? get() = closeJob
|
internal val closeTeardownJob: Job? get() = closeJob.load()
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Round-5 concurrency #5: close() guard. A second concurrent invocation
|
* Round-5 concurrency #5: close() guard. A second concurrent invocation
|
||||||
* (e.g. session close + read-loop death close racing) used to launch a
|
* (e.g. session close + read-loop death close racing) used to launch a
|
||||||
* parallel teardown that called scope.cancel() and socket.close() while
|
* parallel teardown that called scope.cancel() and socket.close() while
|
||||||
* the first close was mid-joinAll. We now memoize the teardown Job so
|
* the first close was mid-joinAll. We memoize the teardown Job so the
|
||||||
* the second caller awaits the first's completion instead.
|
* second caller awaits the first's completion instead.
|
||||||
|
*
|
||||||
|
* Lock-free CAS replaces the previous `synchronized(this)` double-checked
|
||||||
|
* init: the close path is single-shot ("first writer wins"), which is
|
||||||
|
* exactly what `compareAndSet` expresses.
|
||||||
*/
|
*/
|
||||||
@Volatile
|
private val closeJob: AtomicReference<Job?> = AtomicReference(null)
|
||||||
private var closeJob: Job? = null
|
|
||||||
|
|
||||||
fun start() {
|
fun start() {
|
||||||
connection.start()
|
connection.start()
|
||||||
@@ -365,38 +371,45 @@ class QuicConnectionDriver(
|
|||||||
// second concurrent caller (which is common: session.close() and
|
// second concurrent caller (which is common: session.close() and
|
||||||
// read-loop death both race to close()) awaits the same Job rather
|
// read-loop death both race to close()) awaits the same Job rather
|
||||||
// than launching a parallel teardown.
|
// than launching a parallel teardown.
|
||||||
if (closeJob != null) return
|
if (closeJob.load() != null) return
|
||||||
synchronized(this) {
|
// Build the teardown coroutine LAZY so we can race-test the CAS
|
||||||
if (closeJob != null) return
|
// without paying for a launched-and-cancelled Job on the loser.
|
||||||
closeJob =
|
val teardown =
|
||||||
parentScope.launch {
|
parentScope.launch(start = kotlinx.coroutines.CoroutineStart.LAZY) {
|
||||||
connection.close(0L, "")
|
connection.close(0L, "")
|
||||||
wakeup()
|
wakeup()
|
||||||
val send = sendJob
|
val send = sendJob
|
||||||
// Bounded wait for the send loop to flush CONNECTION_CLOSE.
|
// Bounded wait for the send loop to flush CONNECTION_CLOSE.
|
||||||
// We don't want to hang forever if the writer is wedged —
|
// We don't want to hang forever if the writer is wedged —
|
||||||
// the timeout is the upper bound on how long close() blocks.
|
// the timeout is the upper bound on how long close() blocks.
|
||||||
withTimeoutOrNull(CLOSE_FLUSH_TIMEOUT_MILLIS) {
|
withTimeoutOrNull(CLOSE_FLUSH_TIMEOUT_MILLIS) {
|
||||||
// Spin until the writer has actually drained the queued
|
// Spin until the writer has actually drained the queued
|
||||||
// close. The CLOSING-status check transitions to CLOSED
|
// close. The CLOSING-status check transitions to CLOSED
|
||||||
// once drainOutbound builds the CONNECTION_CLOSE packet.
|
// once drainOutbound builds the CONNECTION_CLOSE packet.
|
||||||
while (connection.status == QuicConnection.Status.CLOSING) {
|
while (connection.status == QuicConnection.Status.CLOSING) {
|
||||||
kotlinx.coroutines.delay(1)
|
kotlinx.coroutines.delay(1)
|
||||||
}
|
|
||||||
}
|
}
|
||||||
// Now flip to CLOSED so both loops exit their while-guards.
|
|
||||||
connection.markClosedExternally("driver close requested")
|
|
||||||
wakeup()
|
|
||||||
// Wait for both loops to actually exit — joinAll won't
|
|
||||||
// return until the in-flight socket.send() completes.
|
|
||||||
withTimeoutOrNull(CLOSE_FLUSH_TIMEOUT_MILLIS) {
|
|
||||||
listOfNotNull(readJob, send).joinAll()
|
|
||||||
}
|
|
||||||
// Final teardown — cancel guarantees both jobs are done
|
|
||||||
// before we close the socket.
|
|
||||||
scope.cancel()
|
|
||||||
socket.close()
|
|
||||||
}
|
}
|
||||||
|
// Now flip to CLOSED so both loops exit their while-guards.
|
||||||
|
connection.markClosedExternally("driver close requested")
|
||||||
|
wakeup()
|
||||||
|
// Wait for both loops to actually exit — joinAll won't
|
||||||
|
// return until the in-flight socket.send() completes.
|
||||||
|
withTimeoutOrNull(CLOSE_FLUSH_TIMEOUT_MILLIS) {
|
||||||
|
listOfNotNull(readJob, send).joinAll()
|
||||||
|
}
|
||||||
|
// Final teardown — cancel guarantees both jobs are done
|
||||||
|
// before we close the socket.
|
||||||
|
scope.cancel()
|
||||||
|
socket.close()
|
||||||
|
}
|
||||||
|
if (closeJob.compareAndSet(null, teardown)) {
|
||||||
|
teardown.start()
|
||||||
|
} else {
|
||||||
|
// A concurrent close() already installed the teardown Job; drop
|
||||||
|
// ours without ever starting it. The winner's Job runs, this
|
||||||
|
// call is a no-op (matching the original idempotent contract).
|
||||||
|
teardown.cancel()
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -60,9 +60,14 @@ package com.vitorpamplona.quic.stream
|
|||||||
* send loop under the connection mutex; [markAcked] / [markLost] run on
|
* send loop under the connection mutex; [markAcked] / [markLost] run on
|
||||||
* the parser path also under the connection mutex. The two execution
|
* the parser path also under the connection mutex. The two execution
|
||||||
* paths are NOT serialised by a shared lock, so all internal state is
|
* paths are NOT serialised by a shared lock, so all internal state is
|
||||||
* mutated under `synchronized(this)`. Even the cheap getters
|
* mutated under `synchronized(this)`. Single-field reads
|
||||||
* ([readableBytes], [sentOffset], [finPending], [finSent]) take the
|
* ([nextOffset], [sentOffset], [finPending], [finSent], [finAcked])
|
||||||
* monitor so a writer pre-flight check can't observe torn state.
|
* use `@Volatile` backing fields and bypass the monitor — they cannot
|
||||||
|
* tear (Boolean is single-byte; Long writes happen inside the
|
||||||
|
* synchronized block on JVM/Android, where `@Volatile Long` is
|
||||||
|
* atomic). [readableBytes] still synchronizes because its formula
|
||||||
|
* combines two fields and would otherwise observe a transient
|
||||||
|
* negative value if read mid-`takeChunk`.
|
||||||
*
|
*
|
||||||
* # FIN
|
* # FIN
|
||||||
*
|
*
|
||||||
@@ -105,6 +110,7 @@ class SendBuffer(
|
|||||||
private var flushedFloor: Long = 0L
|
private var flushedFloor: Long = 0L
|
||||||
|
|
||||||
/** Logical offset just past the last byte. Advances on [enqueue]. */
|
/** Logical offset just past the last byte. Advances on [enqueue]. */
|
||||||
|
@Volatile
|
||||||
private var _nextOffset: Long = 0L
|
private var _nextOffset: Long = 0L
|
||||||
|
|
||||||
/**
|
/**
|
||||||
@@ -115,6 +121,7 @@ class SendBuffer(
|
|||||||
*
|
*
|
||||||
* Invariant: `flushedFloor <= nextSendOffset <= nextOffset`.
|
* Invariant: `flushedFloor <= nextSendOffset <= nextOffset`.
|
||||||
*/
|
*/
|
||||||
|
@Volatile
|
||||||
private var nextSendOffset: Long = 0L
|
private var nextSendOffset: Long = 0L
|
||||||
|
|
||||||
/**
|
/**
|
||||||
@@ -151,14 +158,19 @@ class SendBuffer(
|
|||||||
*/
|
*/
|
||||||
private var retransmitTotalBytes: Long = 0L
|
private var retransmitTotalBytes: Long = 0L
|
||||||
|
|
||||||
|
@Volatile
|
||||||
private var _finPending: Boolean = false
|
private var _finPending: Boolean = false
|
||||||
|
|
||||||
|
@Volatile
|
||||||
private var _finSent: Boolean = false
|
private var _finSent: Boolean = false
|
||||||
|
|
||||||
|
@Volatile
|
||||||
private var _finAcked: Boolean = false
|
private var _finAcked: Boolean = false
|
||||||
|
|
||||||
val nextOffset: Long get() = synchronized(this) { _nextOffset }
|
val nextOffset: Long get() = _nextOffset
|
||||||
val finPending: Boolean get() = synchronized(this) { _finPending }
|
val finPending: Boolean get() = _finPending
|
||||||
val finSent: Boolean get() = synchronized(this) { _finSent }
|
val finSent: Boolean get() = _finSent
|
||||||
val finAcked: Boolean get() = synchronized(this) { _finAcked }
|
val finAcked: Boolean get() = _finAcked
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Bytes the writer would emit on the next [takeChunk] before any
|
* Bytes the writer would emit on the next [takeChunk] before any
|
||||||
@@ -183,7 +195,7 @@ class SendBuffer(
|
|||||||
* the high-water mark of *fresh* sends only, not the cumulative
|
* the high-water mark of *fresh* sends only, not the cumulative
|
||||||
* retransmit volume.
|
* retransmit volume.
|
||||||
*/
|
*/
|
||||||
val sentOffset: Long get() = synchronized(this) { nextSendOffset }
|
val sentOffset: Long get() = nextSendOffset
|
||||||
|
|
||||||
fun enqueue(bytes: ByteArray) {
|
fun enqueue(bytes: ByteArray) {
|
||||||
if (bytes.isEmpty()) return
|
if (bytes.isEmpty()) return
|
||||||
|
|||||||
@@ -26,11 +26,23 @@ import javax.crypto.spec.SecretKeySpec
|
|||||||
|
|
||||||
/**
|
/**
|
||||||
* One-block AES-ECB encryption via JCA. Used only by QUIC header protection
|
* One-block AES-ECB encryption via JCA. Used only by QUIC header protection
|
||||||
* (one block per packet, so no need for a more elaborate API).
|
* (one block per packet). The `Cipher` instance is cached per-thread so the
|
||||||
|
* read/send loops avoid `Cipher.getInstance(...)` provider lookup on every
|
||||||
|
* packet — which is the dominant cost for a 16-byte one-shot AEAD-less call.
|
||||||
|
*
|
||||||
|
* `ThreadLocal` is safe here because the call is stateless: every invocation
|
||||||
|
* re-`init`s with the caller-supplied key before `doFinal`, so a coroutine
|
||||||
|
* that hops dispatchers between calls just lands on whichever thread's cached
|
||||||
|
* `Cipher` it ends up on. No state leaks across calls.
|
||||||
*/
|
*/
|
||||||
|
private val aesEcbCipher: ThreadLocal<Cipher> =
|
||||||
|
ThreadLocal.withInitial { Cipher.getInstance("AES/ECB/NoPadding") }
|
||||||
|
|
||||||
actual val PlatformAesOneBlock: AesOneBlockEncrypt =
|
actual val PlatformAesOneBlock: AesOneBlockEncrypt =
|
||||||
AesOneBlockEncrypt { key, block ->
|
AesOneBlockEncrypt { key, block ->
|
||||||
val cipher = Cipher.getInstance("AES/ECB/NoPadding")
|
// .get() is non-null because withInitial supplies a Cipher, but
|
||||||
|
// Kotlin sees the Java return type as platform-nullable.
|
||||||
|
val cipher = aesEcbCipher.get()!!
|
||||||
cipher.init(Cipher.ENCRYPT_MODE, SecretKeySpec(key, "AES"))
|
cipher.init(Cipher.ENCRYPT_MODE, SecretKeySpec(key, "AES"))
|
||||||
cipher.doFinal(block)
|
cipher.doFinal(block)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -191,10 +191,20 @@ class JdkCertificateValidator(
|
|||||||
// GeneralName type 2 = dNSName, type 7 = iPAddress.
|
// GeneralName type 2 = dNSName, type 7 = iPAddress.
|
||||||
if (type == 2 && dnsMatches(idnAscii(value), normalizedHost)) return true
|
if (type == 2 && dnsMatches(idnAscii(value), normalizedHost)) return true
|
||||||
if (type == 7 && hostAsIp != null) {
|
if (type == 7 && hostAsIp != null) {
|
||||||
|
// Defense-in-depth: a malformed cert could put a hostname in
|
||||||
|
// a type 7 SAN. Without this gate, InetAddress.getByName(value)
|
||||||
|
// would perform a DNS A/AAAA lookup on the validation path,
|
||||||
|
// both leaking the name in plaintext and blocking the read
|
||||||
|
// loop on the system resolver. Forcing a literal check keeps
|
||||||
|
// the JDK call to pure parsing (no I/O, no name service).
|
||||||
val sanIp =
|
val sanIp =
|
||||||
try {
|
if (looksLikeIpLiteral(value)) {
|
||||||
InetAddress.getByName(value).hostAddress
|
try {
|
||||||
} catch (_: Throwable) {
|
InetAddress.getByName(value).hostAddress
|
||||||
|
} catch (_: Throwable) {
|
||||||
|
null
|
||||||
|
}
|
||||||
|
} else {
|
||||||
null
|
null
|
||||||
}
|
}
|
||||||
if (sanIp != null && sanIp.equals(hostAsIp, ignoreCase = true)) return true
|
if (sanIp != null && sanIp.equals(hostAsIp, ignoreCase = true)) return true
|
||||||
|
|||||||
Reference in New Issue
Block a user