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

fix(quic): WebTransport peer-stream demux + flow-control + perf cleanup

Browse Source 
Round-2 audit found that without prefix-stripping on peer-initiated streams,
the server's HTTP/3 control stream would deliver its SETTINGS frame to the
application, breaking MoQ framing. Plus several flow-control + perf items
the audit flagged.

WtPeerStreamDemux:
  Spawned per WT session in QuicWebTransportSessionState.init. Routes peer
  streams by their leading varint(s):
    - HTTP/3 CONTROL stream (0x00) → drains internally, captures peer
      SETTINGS frame for inspection (peerSettings property).
    - QPACK encoder/decoder streams (0x02/0x03) → drained to /dev/null
      (we run with QPACK_MAX_TABLE_CAPACITY=0).
    - WebTransport unidirectional (0x54) + matching quarter session id →
      surfaces to app as a StrippedWtStream with the prefix stripped.
    - WebTransport bidi signal (0x41) + matching quarter session id → same.
    - Anything else → dropped per RFC 9114 §9.

  pollIncomingPeerStream is now @Deprecated; the recommended path is the
  incomingStrippedStreams flow on the session state. The nestsClient adapter
  now consumes the stripped flow and emits StrippedWtReadStreamAdapter
  through the existing WebTransportSession.incomingUniStreams API.

WT_CLOSE_SESSION decoder + CapsuleReader:
  Added stateful capsule reader that yields parsed WtCloseSession
  (errorCode + reason) from the CONNECT bidi. Wires into the future close
  notification path; not yet bound to status flip but the parser is in
  place.

Flow-control fixes from the audit:
  - Per-direction receive window in getOrCreatePeerStreamLocked. SERVER_UNI
    streams now use initialMaxStreamDataUni instead of bidi-remote.
  - appendFlowControlUpdates picks per-direction window matching the
    stream's StreamId.kindOf.
  - MAX_DATA tracking via QuicConnection.advertisedMaxData high-water mark.
    Previously the writer emitted a fresh MAX_DATA on every outbound
    packet once the threshold was crossed (spam). Now: only when the new
    value strictly exceeds advertisedMaxData.

Stream iteration round-robin:
  Writer was iterating streamsLocked() in insertion order, starving streams
  created later under MTU pressure. Added streamRoundRobinStart counter on
  QuicConnection that advances after each drain.

SendBuffer alias defense:
  takeChunk's fast path used to return the head ByteArray reference directly
  when consuming the whole chunk — caller-owned byte arrays could be mutated
  in-place. Now always copies, since downstream encoders pass the bytes to
  AEAD.seal which assumes immutability.

UdpSocket cleanup:
  - readBuf shrunk from 65 KiB to 2 KiB. QUIC datagrams cap at MTU; the old
    65 KiB allocation was per-connection waste with no benefit.
  - Removed pointless synchronized(readBuf) — only the read loop touches it.

All :quic:jvmTest + :nestsClient:jvmTest pass.

https://claude.ai/code/session_01EC1tfXfap8k8GyKvrxkxZx
This commit is contained in:
Claude
2026-04-25 22:49:32 +00:00
parent 94a3d32a6d
commit 8f5280c9c3
8 changed files with 462 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nestsClient/src/jvmAndroid/kotlin/com/vitorpamplona/nestsclient/transport/QuicWebTransportFactory.kt
+15 -4
View File
@@ -219,10 +219,14 @@ class QuicWebTransportSession(
override fun incomingUniStreams(): Flow<WebTransportReadStream> =
flow {
while (state.isOpen) {
val s = state.pollIncomingPeerStream()
if (s != null) emit(QuicReadStreamAdapter(s))
kotlinx.coroutines.delay(10)
// Surface only unidirectional WT streams whose prefix bytes
// (0x54 + quarter session id) have been stripped. The H3 control
// stream and QPACK encoder/decoder streams are drained internally
// by the demux and never reach here.
state.incomingStrippedStreams.collect { stripped ->
if (stripped.isUnidirectional) {
emit(StrippedWtReadStreamAdapter(stripped))
}
}
}
@@ -270,3 +274,10 @@ private class QuicReadStreamAdapter(
) : WebTransportReadStream {
override fun incoming(): Flow<ByteArray> = stream.incoming
}
/** Adapter for a WT peer-initiated uni stream whose prefix has been stripped. */
private class StrippedWtReadStreamAdapter(
private val stripped: com.vitorpamplona.quic.webtransport.StrippedWtStream,
) : WebTransportReadStream {
override fun incoming(): Flow<ByteArray> = stripped.data
}
quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt
+24 -1
View File
@@ -102,6 +102,21 @@ class QuicConnection(
private val streams = mutableMapOf<Long, QuicStream>()
private var nextLocalBidiIndex: Long = 0L
private var nextLocalUniIndex: Long = 0L
/**
* The connection-level receive limit we've currently advertised to the
* peer. Tracks the high-water mark of the most recent MAX_DATA frame we
* sent. The writer only emits a new MAX_DATA when the new value exceeds
* this — prevents spamming the peer with redundant updates.
*/
internal var advertisedMaxData: Long = config.initialMaxData
/**
* Round-robin starting index for the writer's stream-drain iteration.
* Without rotation, streams created earlier always drain first under MTU
* pressure, starving later streams indefinitely.
*/
internal var streamRoundRobinStart: Int = 0
private val pendingDatagrams = ArrayDeque<ByteArray>()
private val incomingDatagrams = ArrayDeque<ByteArray>()
private var sendConnectionFlowCredit: Long = 0L
@@ -293,7 +308,15 @@ class QuicConnection(
}
val stream = QuicStream(id, direction)
stream.sendCredit = 0L
stream.receiveLimit = config.initialMaxStreamDataBidiRemote
// Pick the local receive-limit appropriate for the stream's direction
// — peer-bidi → we advertised initialMaxStreamDataBidiRemote;
// peer-uni → we advertised initialMaxStreamDataUni.
stream.receiveLimit =
when (StreamId.kindOf(id)) {
StreamId.Kind.SERVER_UNI, StreamId.Kind.CLIENT_UNI -> config.initialMaxStreamDataUni
StreamId.Kind.SERVER_BIDI -> config.initialMaxStreamDataBidiRemote
StreamId.Kind.CLIENT_BIDI -> config.initialMaxStreamDataBidiLocal
}
streams[id] = stream
newPeerStreams.addLast(stream)
return stream
quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt
+40 -17
View File
@@ -257,20 +257,26 @@ private fun buildApplicationPacket(
if (frames.size >= 16) break
}
// Drain stream send buffers — round-robin keeping packet under MTU and
// honoring per-stream + connection-level send credit (RFC 9000 §4).
// Drain stream send buffers — round-robin starting from a rotating index
// so streams created earlier don't starve streams created later under MTU
// pressure. Honors per-stream send credit (RFC 9000 §4).
var packetBudget = 1100
for ((id, stream) in conn.streamsLocked()) {
if (packetBudget <= 64) break
// Per-stream credit: how many more bytes is the peer willing to receive on this stream?
val streamRemaining = (stream.sendCredit - stream.send.sentOffset).coerceAtLeast(0L)
if (streamRemaining <= 0L && !stream.send.finPending) continue
val maxBytes = minOf(packetBudget - 32, streamRemaining.coerceAtMost(Int.MAX_VALUE.toLong()).toInt())
val chunk = stream.send.takeChunk(maxBytes = maxBytes) ?: continue
if (chunk.data.isNotEmpty() || chunk.fin) {
frames += StreamFrame(streamId = id, offset = chunk.offset, data = chunk.data, fin = chunk.fin, explicitLength = true)
packetBudget -= chunk.data.size + 32
val streamsList = conn.streamsLocked().entries.toList()
if (streamsList.isNotEmpty()) {
val start = conn.streamRoundRobinStart % streamsList.size
for (i in streamsList.indices) {
if (packetBudget <= 64) break
val (id, stream) = streamsList[(start + i) % streamsList.size]
val streamRemaining = (stream.sendCredit - stream.send.sentOffset).coerceAtLeast(0L)
if (streamRemaining <= 0L && !stream.send.finPending) continue
val maxBytes = minOf(packetBudget - 32, streamRemaining.coerceAtMost(Int.MAX_VALUE.toLong()).toInt())
val chunk = stream.send.takeChunk(maxBytes = maxBytes) ?: continue
if (chunk.data.isNotEmpty() || chunk.fin) {
frames += StreamFrame(streamId = id, offset = chunk.offset, data = chunk.data, fin = chunk.fin, explicitLength = true)
packetBudget -= chunk.data.size + 32
}
}
conn.streamRoundRobinStart = (start + 1) % streamsList.size
}
if (frames.isEmpty()) return null
@@ -302,8 +308,20 @@ private fun appendFlowControlUpdates(
for ((id, stream) in conn.streamsLocked()) {
val rcv = stream.receive.contiguousEnd()
if (rcv == 0L) continue
// Pick the per-direction window matching the stream kind so a
// uni-only deployment with bidi=0 doesn't accidentally use the bidi
// window and vice versa.
val window =
when (
com.vitorpamplona.quic.stream.StreamId
.kindOf(id)
) {
com.vitorpamplona.quic.stream.StreamId.Kind.SERVER_UNI -> cfg.initialMaxStreamDataUni
com.vitorpamplona.quic.stream.StreamId.Kind.SERVER_BIDI -> cfg.initialMaxStreamDataBidiRemote
com.vitorpamplona.quic.stream.StreamId.Kind.CLIENT_BIDI -> cfg.initialMaxStreamDataBidiLocal
com.vitorpamplona.quic.stream.StreamId.Kind.CLIENT_UNI -> cfg.initialMaxStreamDataUni
}
// Re-credit when consumed > half the advertised window.
val window = cfg.initialMaxStreamDataBidiRemote.coerceAtLeast(cfg.initialMaxStreamDataUni)
if (rcv >= stream.receiveLimit - window / 2) {
val newLimit = rcv + window
if (newLimit > stream.receiveLimit) {
@@ -313,9 +331,14 @@ private fun appendFlowControlUpdates(
}
totalRecvAdvanced += rcv
}
// Connection-level credit: when sum of contiguousEnd across all streams
// exceeds half the connection-level limit, re-grant.
if (totalRecvAdvanced > 0L && totalRecvAdvanced >= cfg.initialMaxData / 2) {
frames += MaxDataFrame(totalRecvAdvanced + cfg.initialMaxData)
// Connection-level: only re-grant when the new total would exceed our
// currently-advertised limit. Without this we'd emit a fresh MaxDataFrame
// on every outbound packet once the threshold was crossed.
if (totalRecvAdvanced > 0L && totalRecvAdvanced + cfg.initialMaxData / 2 >= conn.advertisedMaxData) {
val newLimit = totalRecvAdvanced + cfg.initialMaxData
if (newLimit > conn.advertisedMaxData) {
conn.advertisedMaxData = newLimit
frames += MaxDataFrame(newLimit)
}
}
}
quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/SendBuffer.kt
+11 -2
View File
@@ -83,8 +83,17 @@ class SendBuffer {
val head = chunks.first()
val available = head.size - headOffset
if (available <= cap) {
// Hand out the rest of the head chunk.
data = if (headOffset == 0) head else head.copyOfRange(headOffset, head.size)
// Hand out the rest of the head chunk. Always copy: the caller's
// ByteArray (passed to enqueue) MUST stay opaque to the rest of
// the stack, since downstream encoders eventually pass it to
// AEAD.seal which assumes immutability for the duration of the
// encryption call.
data =
if (headOffset == 0 && head.size == available) {
head.copyOf()
} else {
head.copyOfRange(headOffset, head.size)
}
chunks.removeFirst()
headOffset = 0
pendingBytes -= available
quic/src/commonMain/kotlin/com/vitorpamplona/quic/webtransport/QuicWebTransportSessionState.kt
+44
View File
@@ -23,6 +23,11 @@ package com.vitorpamplona.quic.webtransport
import com.vitorpamplona.quic.connection.QuicConnection
import com.vitorpamplona.quic.connection.QuicConnectionDriver
import com.vitorpamplona.quic.stream.QuicStream
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.launch
/**
* Container that bundles a [QuicConnection], its [QuicConnectionDriver], and
@@ -36,11 +41,42 @@ class QuicWebTransportSessionState(
val connection: QuicConnection,
val driver: QuicConnectionDriver,
val connectStreamId: Long,
/**
* Scope used to spawn the peer-stream demux. Defaults to a SupervisorJob
* so a single misbehaving stream doesn't tear down the session.
*/
private val scope: CoroutineScope =
CoroutineScope(SupervisorJob() + Dispatchers.IO),
) {
/** True until [close] is called or the underlying QUIC connection terminates. */
val isOpen: Boolean
get() = connection.status == QuicConnection.Status.CONNECTED
private val demux: WtPeerStreamDemux = WtPeerStreamDemux(connectStreamId, scope)
init {
// Spawn the dispatcher that routes peer-initiated streams through the
// demux. Without this, the server's CONTROL stream (carrying SETTINGS)
// would be handed to the application, which would interpret SETTINGS
// bytes as MoQ frames and break framing.
scope.launch {
while (connection.status != QuicConnection.Status.CLOSED) {
val s = connection.pollIncomingPeerStream()
if (s != null) {
demux.process(s)
} else {
kotlinx.coroutines.delay(5)
}
}
}
}
/** Server SETTINGS once received on the H3 control stream; null until then. */
val peerSettings get() = demux.peerSettings
/** Flow of peer-initiated WT streams whose framing prefix has been stripped. */
val incomingStrippedStreams: Flow<StrippedWtStream> get() = demux.incomingStrippedStreams
/** Open a new client-initiated bidirectional WebTransport stream. */
suspend fun openBidiStream(): QuicStream {
val s = connection.openBidiStream()
@@ -72,6 +108,14 @@ class QuicWebTransportSessionState(
return decoded.payload
}
/**
* @deprecated Use [incomingStrippedStreams] which yields streams whose
* framing prefix (CONTROL/QPACK/WT type bytes + quarter session id) has
* already been stripped. Direct callers of this would receive raw peer
* streams including the server's CONTROL stream, with SETTINGS bytes
* interpreted as application data.
*/
@Deprecated("Use incomingStrippedStreams instead", ReplaceWith("incomingStrippedStreams"))
suspend fun pollIncomingPeerStream(): QuicStream? = connection.pollIncomingPeerStream()
suspend fun close(
quic/src/commonMain/kotlin/com/vitorpamplona/quic/webtransport/WtCapsule.kt
+71
View File
@@ -53,3 +53,74 @@ fun encodeCloseSessionCapsule(
body.writeBytes(reason.encodeToByteArray())
return encodeCapsule(WtCapsuleType.WT_CLOSE_SESSION, body.toByteArray())
}
/** Parsed WT_CLOSE_SESSION capsule. */
data class WtCloseSession(
val errorCode: Int,
val reason: String,
)
/**
* Stateful capsule reader. Capsules on the WT CONNECT bidi stream are
* `(varint type)(varint length)(body)`. Feed bytes via [push]; drain
* complete capsules via [next].
*/
class CapsuleReader {
private var buf: ByteArray = ByteArray(0)
private var pos: Int = 0
fun push(bytes: ByteArray) {
if (bytes.isEmpty()) return
if (pos > 0) {
buf = buf.copyOfRange(pos, buf.size)
pos = 0
}
val combined = ByteArray(buf.size + bytes.size)
buf.copyInto(combined, 0)
bytes.copyInto(combined, buf.size)
buf = combined
}
/**
* Pop the next complete capsule, or null if the buffer doesn't contain
* one yet. Returns either a [WtCloseSession] or a raw `(type, body)`
* pair for unknown capsule types.
*/
fun next(): Any? {
val typeRes =
com.vitorpamplona.quic.Varint
.decode(buf, pos) ?: return null
val typeEnd = pos + typeRes.bytesConsumed
val lenRes =
com.vitorpamplona.quic.Varint
.decode(buf, typeEnd) ?: return null
val bodyStart = typeEnd + lenRes.bytesConsumed
val len = lenRes.value
if (len < 0 || len > Int.MAX_VALUE.toLong()) {
throw com.vitorpamplona.quic.QuicCodecException("capsule length out of range: $len")
}
val bodyEnd = bodyStart + len.toInt()
if (bodyEnd > buf.size) return null
val body = buf.copyOfRange(bodyStart, bodyEnd)
pos = bodyEnd
return when (typeRes.value) {
WtCapsuleType.WT_CLOSE_SESSION -> {
if (body.size < 4) {
WtCloseSession(0, "")
} else {
val errorCode =
((body[0].toInt() and 0xFF) shl 24) or
((body[1].toInt() and 0xFF) shl 16) or
((body[2].toInt() and 0xFF) shl 8) or
(body[3].toInt() and 0xFF)
val reason = body.copyOfRange(4, body.size).decodeToString()
WtCloseSession(errorCode, reason)
}
}
else -> {
typeRes.value to body
}
}
}
}
quic/src/commonMain/kotlin/com/vitorpamplona/quic/webtransport/WtPeerStreamDemux.kt
+244
View File
@@ -0,0 +1,244 @@
/*
* Copyright (c) 2025 Vitor Pamplona
*
* Permission is hereby granted, free of charge, to any person obtaining a copy of
* this software and associated documentation files (the "Software"), to deal in
* the Software without restriction, including without limitation the rights to use,
* copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
* Software, and to permit persons to whom the Software is furnished to do so,
* subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
* FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
* COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
* 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.
*/
package com.vitorpamplona.quic.webtransport
import com.vitorpamplona.quic.Varint
import com.vitorpamplona.quic.http3.Http3Frame
import com.vitorpamplona.quic.http3.Http3FrameReader
import com.vitorpamplona.quic.http3.Http3Settings
import com.vitorpamplona.quic.http3.Http3StreamType
import com.vitorpamplona.quic.stream.QuicStream
import com.vitorpamplona.quic.stream.StreamId
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.consumeAsFlow
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.launch
/**
* A peer-initiated WebTransport stream whose framing prefix has already been
* stripped. The [data] flow yields only application-level bytes.
*/
class StrippedWtStream(
val streamId: Long,
val isUnidirectional: Boolean,
val data: Flow<ByteArray>,
)
/**
* Demultiplexes peer-initiated streams by their leading varint(s):
*
* - HTTP/3 CONTROL stream (0x00) — drained internally, SETTINGS captured.
* - QPACK encoder/decoder streams (0x02 / 0x03) — drained (we run with
* dynamic table off).
* - WT unidirectional (0x54) + matching quarter session id — surfaced.
* - WT bidirectional signal (0x41) + matching quarter session id — surfaced.
* - Anything else — dropped per RFC 9114 §9.
*
* Without this, the server's CONTROL stream would deliver its SETTINGS frame
* bytes directly to MoQ as application data — a live-interop break.
*/
class WtPeerStreamDemux(
private val expectedConnectStreamId: Long,
private val scope: CoroutineScope,
) {
private val readyStreams = Channel<StrippedWtStream>(Channel.UNLIMITED)
@Volatile
var peerSettings: Http3Settings? = null
private set
val incomingStrippedStreams: Flow<StrippedWtStream> = readyStreams.consumeAsFlow()
/**
* Begin processing a peer-initiated stream. Caller must invoke this for
* every stream returned by [com.vitorpamplona.quic.connection.QuicConnection.pollIncomingPeerStream].
*/
fun process(stream: QuicStream) {
scope.launch { route(stream) }
}
private suspend fun route(stream: QuicStream) {
// Build a buffered chunk source: keeps unconsumed bytes until we
// know what kind of stream we're looking at.
val pending = ArrayDeque<ByteArray>()
val flowIterator = stream.incoming
val chunkChannel = Channel<ByteArray>(Channel.UNLIMITED)
scope.launch {
try {
flowIterator.collect { chunkChannel.send(it) }
} finally {
chunkChannel.close()
}
}
// Helper: read the next available bytes; returns null on stream close
// before enough bytes are present.
suspend fun moreBytes(): Boolean {
val chunk = chunkChannel.receiveCatching().getOrNull() ?: return false
pending.addLast(chunk)
return true
}
suspend fun readVarintFromPending(): Long? {
while (true) {
val flat = flatten(pending)
val res = Varint.decode(flat, 0)
if (res != null) {
consumeFromPending(pending, res.bytesConsumed)
return res.value
}
if (!moreBytes()) return null
}
}
try {
if (StreamId.isUnidirectional(stream.streamId)) {
val streamType = readVarintFromPending() ?: return
when (streamType) {
Http3StreamType.CONTROL -> {
drainControlStream(pending, chunkChannel)
}
Http3StreamType.QPACK_ENCODER, Http3StreamType.QPACK_DECODER -> {
drainBlackHole(chunkChannel)
}
Http3StreamType.WEBTRANSPORT_UNI_STREAM -> {
val quarter = readVarintFromPending() ?: return
if (quarter * 4L != expectedConnectStreamId) {
drainBlackHole(chunkChannel) // not our session
return
}
emitStripped(stream, pending, chunkChannel, isUni = true)
}
else -> {
drainBlackHole(chunkChannel)
} // unknown — drop per RFC 9114 §9
}
} else {
// Server-initiated bidi: per draft-ietf-webtrans-http3, prefixed
// with WT_BIDI_STREAM (0x41) varint then quarter session id.
val signal = readVarintFromPending() ?: return
if (signal != WtStreamType.WT_BIDI_STREAM) {
drainBlackHole(chunkChannel)
return
}
val quarter = readVarintFromPending() ?: return
if (quarter * 4L != expectedConnectStreamId) {
drainBlackHole(chunkChannel)
return
}
emitStripped(stream, pending, chunkChannel, isUni = false)
}
} catch (_: Throwable) {
// peer closed mid-prefix or framing error — drop quietly
}
}
private suspend fun drainControlStream(
pending: ArrayDeque<ByteArray>,
chunkChannel: Channel<ByteArray>,
) {
val reader = Http3FrameReader()
// Push whatever we already buffered.
while (pending.isNotEmpty()) reader.push(pending.removeFirst())
consumeFrames(reader)
for (chunk in chunkChannel) {
reader.push(chunk)
consumeFrames(reader)
}
}
private fun consumeFrames(reader: Http3FrameReader) {
while (true) {
val frame = reader.next() ?: return
when (frame) {
is Http3Frame.Settings -> peerSettings = frame.settings
is Http3Frame.Goaway -> Unit
// no new requests; we don't enforce yet
else -> Unit
}
}
}
private suspend fun drainBlackHole(chunkChannel: Channel<ByteArray>) {
@Suppress("UNUSED_VARIABLE")
for (discarded in chunkChannel) {
// intentionally discarded — stream type is one we don't process
}
}
private fun emitStripped(
stream: QuicStream,
pending: ArrayDeque<ByteArray>,
chunkChannel: Channel<ByteArray>,
isUni: Boolean,
) {
val prebuffered = pending.toList()
pending.clear()
val data: Flow<ByteArray> =
flow {
for (b in prebuffered) if (b.isNotEmpty()) emit(b)
for (chunk in chunkChannel) if (chunk.isNotEmpty()) emit(chunk)
}
readyStreams.trySend(
StrippedWtStream(
streamId = stream.streamId,
isUnidirectional = isUni,
data = data,
),
)
}
}
private fun flatten(chunks: ArrayDeque<ByteArray>): ByteArray {
var total = 0
for (c in chunks) total += c.size
val out = ByteArray(total)
var pos = 0
for (c in chunks) {
c.copyInto(out, pos)
pos += c.size
}
return out
}
private fun consumeFromPending(
pending: ArrayDeque<ByteArray>,
bytes: Int,
) {
var remaining = bytes
while (remaining > 0 && pending.isNotEmpty()) {
val head = pending.first()
if (head.size <= remaining) {
pending.removeFirst()
remaining -= head.size
} else {
pending[0] = head.copyOfRange(remaining, head.size)
remaining = 0
}
}
}
quic/src/jvmAndroid/kotlin/com/vitorpamplona/quic/transport/UdpSocket.kt
+13 -11
View File
@@ -43,7 +43,11 @@ actual class UdpSocket private constructor(
private val remote: InetSocketAddress,
) {
private val closed = AtomicBoolean(false)
private val readBuf = ByteBuffer.allocate(MAX_DATAGRAM_SIZE)
// Sized to typical Ethernet MTU + a bit; QUIC tops out at ~1500 in practice
// and any larger inbound frame is dropped as malformed anyway. The previous
// 64 KiB buffer was wasteful per connection.
private val readBuf = ByteBuffer.allocate(2048)
actual val localPort: Int
get() = (channel.localAddress as InetSocketAddress).port
@@ -59,14 +63,14 @@ actual class UdpSocket private constructor(
withContext(Dispatchers.IO) {
if (closed.get()) return@withContext null
try {
synchronized(readBuf) {
readBuf.clear()
channel.receive(readBuf) ?: return@withContext null
readBuf.flip()
val out = ByteArray(readBuf.remaining())
readBuf.get(out)
out
}
// No synchronized — only the read loop touches readBuf, by
// contract. The previous synchronized block was pointless.
readBuf.clear()
channel.receive(readBuf) ?: return@withContext null
readBuf.flip()
val out = ByteArray(readBuf.remaining())
readBuf.get(out)
out
} catch (_: ClosedChannelException) {
null
}
@@ -83,8 +87,6 @@ actual class UdpSocket private constructor(
}
actual companion object {
private const val MAX_DATAGRAM_SIZE: Int = 65535
actual suspend fun connect(
host: String,
port: Int,