nick/amethyst
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

perf(quic): drop synchronized from QuicStream + replace close() polling with CompletableDeferred

Browse Source 
Round 2 of the blocking-code audit follow-ups. Both changes remove
commonMain synchronized blocks and replace polling with event-driven
suspension; :quic:jvmTest stays green on JVM and :quic:compileAndroidMain
passes for Android.

QuicStream.resetStream / stopSending
  - Replace synchronized(this) double-checked init with
    AtomicReference<ResetState?>.compareAndSet(null, …) (and same for
    stopSendingState). The pattern is "first call wins"; CAS expresses
    that directly without a lock.
  - Backing fields converted from `internal var … = null` to
    `internal val … = AtomicReference(null)`. The two readers in
    QuicConnectionWriter switch to .load(); QuicConnectionWriter gains a
    file-level @OptIn(ExperimentalAtomicApi::class).
  - The @Volatile resetEmitPending / stopSendingEmitPending writes happen
    only on the winning CAS path, so the writer reading the flag still
    sees the populated state via volatile happens-before.

QuicConnectionDriver.close polling loop
  - Replace `while (status == CLOSING) delay(1)` polling with a
    CompletableDeferred<Unit> on QuicConnection (closingDrainSignal).
    The deferred is completed at both transitions to CLOSED:
    (1) drainOutbound after building the CONNECTION_CLOSE datagram, and
    (2) markClosedExternally on its synchronized first-call-wins path.
  - close() now awaits the deferred under the existing
    CLOSE_FLUSH_TIMEOUT_MILLIS bound — same upper-bound semantics, no
    1 ms wakeups during teardown. complete(Unit) is idempotent, so the
    two transition sites racing each other is safe.

https://claude.ai/code/session_01CXTjnuHKCNXDmpfyKxgG3V
This commit is contained in:
Claude
2026-05-09 02:25:53 +00:00
parent 0e4b12654d
commit 2bb55ff9ec
4 changed files with 67 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files
quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt
+14
View File
@@ -324,6 +324,16 @@ class QuicConnection(
*/
private val closeStateMonitor = Any()
/**
* Single-shot signal completed when [status] transitions to CLOSED —
* either by the writer flushing CONNECTION_CLOSE in `drainOutbound`,
* or by [markClosedExternally] forcing the state. Replaces an earlier
* 1 ms polling loop in `QuicConnectionDriver.close()`. Idempotent:
* `complete(Unit)` returns false on subsequent firers, so racing
* teardown paths are safe.
*/
internal val closingDrainSignal: CompletableDeferred<Unit> = CompletableDeferred()
/** App-level error code for graceful close. */
var closeReason: String? = null
private set
@@ -1522,6 +1532,10 @@ class QuicConnection(
closeReason = reason
true
}
// Wake any teardown coroutine waiting for the close to land. Safe
// to call after the monitor — complete() is idempotent and the
// CLOSED transition has already been published via @Volatile.
if (firstClose) closingDrainSignal.complete(Unit)
if (firstClose) {
// "remote" covers both peer-initiated CONNECTION_CLOSE and
// local invariant violations (CID mismatch, frame decode
quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionDriver.kt
+7 -8
View File
@@ -380,15 +380,14 @@ class QuicConnectionDriver(
wakeup()
val send = sendJob
// Bounded wait for the send loop to flush CONNECTION_CLOSE.
// We don't want to hang forever if the writer is wedged —
// the timeout is the upper bound on how long close() blocks.
// Event-driven via [QuicConnection.closingDrainSignal] —
// both `drainOutbound` (after building the close datagram)
// and `markClosedExternally` (forced transition) complete
// the deferred. Replaces an earlier 1 ms polling loop.
// The timeout is the upper bound on how long close() blocks
// if the writer is wedged.
withTimeoutOrNull(CLOSE_FLUSH_TIMEOUT_MILLIS) {
// Spin until the writer has actually drained the queued
// close. The CLOSING-status check transitions to CLOSED
// once drainOutbound builds the CONNECTION_CLOSE packet.
while (connection.status == QuicConnection.Status.CLOSING) {
kotlinx.coroutines.delay(1)
}
connection.closingDrainSignal.await()
}
// Now flip to CLOSED so both loops exit their while-guards.
connection.markClosedExternally("driver close requested")
quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt
+7 -2
View File
@@ -18,6 +18,8 @@
* 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.
*/
@file:OptIn(kotlin.concurrent.atomics.ExperimentalAtomicApi::class)
package com.vitorpamplona.quic.connection
import com.vitorpamplona.quartz.utils.Log
@@ -86,6 +88,9 @@ fun drainOutbound(
if (conn.status == QuicConnection.Status.CLOSING) {
val datagram = buildClosingDatagram(conn, nowMillis)
conn.status = QuicConnection.Status.CLOSED
// Signal the teardown coroutine that the close datagram is built;
// the driver awaits this instead of polling status.
conn.closingDrainSignal.complete(Unit)
return datagram
}
@@ -952,7 +957,7 @@ private fun appendFlowControlUpdates(
// resetAcked / stopSendingAcked so subsequent stale loss tokens
// are dropped.
for (stream in conn.streamsListLocked()) {
val resetState = stream.resetState
val resetState = stream.resetState.load()
if (resetState != null && stream.resetEmitPending && !stream.resetAcked) {
frames +=
ResetStreamFrame(
@@ -968,7 +973,7 @@ private fun appendFlowControlUpdates(
)
stream.resetEmitPending = false
}
val stopSendingState = stream.stopSendingState
val stopSendingState = stream.stopSendingState.load()
if (stopSendingState != null && stream.stopSendingEmitPending && !stream.stopSendingAcked) {
frames +=
StopSendingFrame(
quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/QuicStream.kt
+39 -25
View File
@@ -23,6 +23,8 @@ package com.vitorpamplona.quic.stream
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.flow
import kotlin.concurrent.atomics.AtomicReference
import kotlin.concurrent.atomics.ExperimentalAtomicApi
/**
* One QUIC stream (bidirectional or unidirectional). Application code
@@ -30,6 +32,7 @@ import kotlinx.coroutines.flow.flow
* APIs; the [QuicConnection] owns the underlying buffers and drains them
* into STREAM frames on the wire.
*/
@OptIn(ExperimentalAtomicApi::class)
class QuicStream(
val streamId: Long,
val direction: Direction,
@@ -255,24 +258,28 @@ class QuicStream(
* (two app threads racing the writer's clear-after-emit).
*/
fun resetStream(errorCode: Long) {
// Synchronized atomic compare-and-set: pre-fix the
// `if (resetState != null) return` plus the assignment was
// racy. Two concurrent callers (e.g. the application aborting
// a request while STOP_SENDING from the peer triggers our own
// resetStream from the parser) could both observe null and
// both write — the second write would clobber the first
// errorCode while [resetEmitPending] was already set. The
// writer would then emit a RESET_STREAM with whichever
// errorCode landed last, possibly different from what the
// application asked for. The synchronized block makes
// first-call-wins genuinely first-call-wins.
synchronized(this) {
if (resetState != null) return
resetState =
ResetState(
errorCode = errorCode,
finalSize = send.nextOffset,
)
// Lock-free first-call-wins. The previous synchronized block
// existed because the naive `if (resetState != null) return;
// resetState = …` had a write-write race: two concurrent
// callers (e.g. the application aborting a request while
// STOP_SENDING from the peer triggers our own resetStream
// from the parser) could both observe null and both write,
// letting the second clobber the first errorCode while
// resetEmitPending was already set. The writer would then
// emit a RESET_STREAM with whichever errorCode landed last.
//
// compareAndSet collapses that to a single CAS: only the
// first writer succeeds, all others observe non-null and
// bail out. `resetEmitPending = true` happens only on the
// winning path, so its @Volatile write happens-after the
// resetState publication — the writer reading the flag sees
// the populated state.
val newState =
ResetState(
errorCode = errorCode,
finalSize = send.nextOffset,
)
if (resetState.compareAndSet(null, newState)) {
resetEmitPending = true
}
}
@@ -292,10 +299,8 @@ class QuicStream(
* original frame already on the wire).
*/
fun stopSending(errorCode: Long) {
// Same atomic-CAS rationale as [resetStream].
synchronized(this) {
if (stopSendingState != null) return
stopSendingState = StopSendingState(errorCode = errorCode)
// Same lock-free first-call-wins rationale as [resetStream].
if (stopSendingState.compareAndSet(null, StopSendingState(errorCode = errorCode))) {
stopSendingEmitPending = true
}
}
@@ -304,8 +309,13 @@ class QuicStream(
* RFC 9000 §3.5 send-side reset state. Set once by [resetStream]
* (subsequent calls no-op); read by the writer + loss/ACK
* dispatchers. Once set, contents are immutable.
*
* Held in an [AtomicReference] so [resetStream] can use
* `compareAndSet(null, …)` to win the first-call-wins race
* without acquiring a lock. Readers use `.load()` — the published
* state is immutable after the CAS so a single load is enough.
*/
internal var resetState: ResetState? = null
internal val resetState: AtomicReference<ResetState?> = AtomicReference(null)
/**
* True while a RESET_STREAM emit is pending. Cleared after the
@@ -331,8 +341,12 @@ class QuicStream(
@Volatile
internal var resetAcked: Boolean = false
/** Receive-side stop-sending state. Set by [stopSending]. */
internal var stopSendingState: StopSendingState? = null
/**
* Receive-side stop-sending state. Set once by [stopSending]
* (subsequent calls no-op); read by the writer + loss/ACK
* dispatchers. Atomic for the same reason as [resetState].
*/
internal val stopSendingState: AtomicReference<StopSendingState?> = AtomicReference(null)
@Volatile
internal var stopSendingEmitPending: Boolean = false