diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt index f55bd953b..6436a32a0 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt @@ -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 = 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 diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionDriver.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionDriver.kt index 5adedc9f1..f400498e3 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionDriver.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionDriver.kt @@ -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") diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt index 20ec31821..89a8b9cfa 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt @@ -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( diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/QuicStream.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/QuicStream.kt index 1b96485e0..97974fec1 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/QuicStream.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/QuicStream.kt @@ -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 = 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 = AtomicReference(null) @Volatile internal var stopSendingEmitPending: Boolean = false