kitetorrent-session/io.github.yuroyami.kitetorrent.session.net/UtpStream

UtpStream

class UtpStream(socket: UdpSocket, remoteHost: String, remotePort: Int, scope: CoroutineScope, nowMicros: () -> Long = MicrosCounter(), connectionId: Int = ((nowMicros() and 0xFFFF).toInt()), minTimeoutMicros: Long = DEFAULT_MIN_TIMEOUT_MICROS, selfTick: Boolean = true, tickIntervalMs: Long = DEFAULT_TICK_INTERVAL_MS, nagleEnabled: Boolean = true) : ByteStream

uTP (Micro Transport Protocol, BEP-29) connection over a single UdpSocket to one remote (host, port). This is a faithful core port of libtorrent's utp_socket_impl (src/utp_stream.cpp), with enough state machine to actually move bytes. It implements ByteStream so a PeerConnection runs over uTP with no changes (the same seam TcpConnection.asByteStream satisfies).

What is faithful

Connection-id assignment. For an outgoing connect, libtorrent's new_utp_socket() picks send_id = random() and recv_id = send_id - 1; the SYN is then sent with connection_id = recv_id ("using recv_id here is intentional", send_syn()), and every subsequent packet we send carries send_id. Incoming packets must carry our recv_id. We reproduce this exactly.
Sequence/ack numbers are 16-bit wrap-around counters compared with compare_less_wrap (= seqLessWrap); ACK_MASK = 0xffff. seq_nr only advances for packets that carry a sequence (SYN/DATA/FIN), matching the C++ comment "m_seq_nr is only incremented when sending packets with data payload".
Handshake. connect() sends ST_SYN in state syn_sent; the first ST_STATE/ST_DATA whose ack_nr == (seq_nr-1) transitions to connected and initialises ack_nr from the peer's seq_nr (case state_t::syn_sent).
In-order reassembly + reorder buffer. consumeData mirrors consume_incoming_data(): a packet at (ack_nr+1) is delivered and drains any contiguous run from the reorder map; anything else is stashed by seq_nr.
Cumulative ack of our send buffer. Incoming ack_nr frees every in-flight packet up to it (for (ack_nr = acked+1 .. next_ack) m_outbuf.remove(...)), releasing send-window space.
FIN / RESET. ST_FIN advances ack_nr in order and signals EOF to readers; ST_RESET errors the socket. close sends ST_FIN.

What is implemented (and where it is still SIMPLIFIED vs. libtorrent):

LEDBAT / delay-based congestion control. growCwndLocked runs do_ledbat() against a 100 ms target_delay, opening/closing cwnd from one-way delay samples (timestamp_difference_microseconds), gated by the cwnd_saturated check so an unsaturated link doesn't inflate the window. Base delay is a time-bucketed rolling minimum (updateDelayLocked) so it tracks slow clock drift.
Retransmission / RTO timer + fast-resend. tick resends the unacked window on RTO (go-back-N, exponential backoff), and SACK / duplicate-ACK loss triggers an early fast-resend. A lost SYN or FIN is also retained in the out-buffer and resent.
Selective ACK (SACK). We both emit (buildSackLocked) and act on (applySackLocked) the SACK extension.
Path-MTU discovery is conservative (shrink-only). The MSS starts at a 1500-byte Ethernet estimate and shrinks toward MIN_MSS when sustained RTOs hint our packets are too large; we never actively probe upward (no ICMP feedback). Current value: currentMss. libtorrent does a full [mtu_floor, mtu_ceiling] search with MTU probes.
Nagle coalescing holds a sub-MSS write tail back to merge with the next write (disable via the nagleEnabled ctor flag); libtorrent's m_nagle_packet is finer-grained.
wnd_size flow control is honoured: bytes in flight never exceed min(cwnd, peerWindow) (see awaitWindow).

Driving the read side

uTP is connectionless at the socket layer: one UdpSocket multiplexes many streams by (connection_id). This class does not own the socket's receive loop; the owner pumps inbound datagrams in by calling onDatagram with the raw bytes of each packet addressed to this stream. UtpSocketManager is that owner in the engine. It runs the receive loop and demultiplexes by connection id, the role utp_socket_manager plays upstream.

Parameters

socket

the shared UDP socket used to send; the owner drives receive.

remoteHost

remote peer host.

remotePort

remote peer port.

scope

coroutine scope the background ack/feed work runs in.

nowMicros

injected microsecond clock. The core is clockless, so this defaults to a monotonically increasing counter (MicrosCounter). Production may pass a real monotonic clock so timestamp_difference is meaningful to peers.

connectionId

optional fixed send_id (the SYN goes out on send_id - 1); defaults to a value derived from the clock so two streams differ. Injectable for deterministic tests.

Constructors

UtpStream

constructor(socket: UdpSocket, remoteHost: String, remotePort: Int, scope: CoroutineScope, nowMicros: () -> Long = MicrosCounter(), connectionId: Int = ((nowMicros() and 0xFFFF).toInt()), minTimeoutMicros: Long = DEFAULT_MIN_TIMEOUT_MICROS, selfTick: Boolean = true, tickIntervalMs: Long = DEFAULT_TICK_INTERVAL_MS, nagleEnabled: Boolean = true)

Types

Companion

object Companion

Properties

remoteAddress

open override val remoteAddress: String

A human-readable identifier for the remote end, for logging.

Functions

fun close()

Graceful close: send ST_FIN (faithful to send_fin() → state fin_sent) and mark the stream closed locally. Idempotent. Does not wait for the peer's FIN.

congestionWindowBytes

suspend fun congestionWindowBytes(): Int

Current congestion window in bytes, for tests and observability.

connect

suspend fun connect()

Perform the active open: send ST_SYN and suspend until the peer's ST_STATE (or first ST_DATA) acknowledges it. Faithful to send_syn() + case state_t::syn_sent.

currentMss

suspend fun currentMss(): Int

The current discovered maximum payload per packet (MSS), for tests and observability.

onDatagram

suspend fun onDatagram(raw: ByteArray)

Feed one raw inbound uTP datagram (header + extensions + payload) addressed to this stream. Drives the full state machine: handshake completion, in-order reassembly, cumulative ack of our send buffer, FIN and RESET. Malformed or mis-addressed packets are dropped silently (return without effect), mirroring the many return false/true rejects in incoming_packet().

readExactly

open suspend override fun readExactly(n: Int): ByteArray

Suspend until exactly n reassembled, in-order payload bytes are available, then return them. Throws if the stream errored or hit EOF before n bytes.

tick

suspend fun tick()

One retransmission tick: the analogue of utp_socket_impl::tick(now). If the oldest unacked packet has gone unacknowledged for longer than packetTimeoutMicros, an RTO has fired. We then retransmit every in-flight packet (go-back-N) and deepen the backoff. After MAX_TIMEOUTS fruitless rounds we fail the socket, so a dead peer cannot block a reader forever. Built under lock; the resends go out after.

write

open suspend override fun write(bytes: ByteArray)

Packetise bytes into one or more ST_DATA packets and send them, blocking on the congestion+flow window so bytes-in-flight never exceed min(cwnd, peerWindow). Faithful to send_pkt() chunking (payload_size = min(write_buffer, mtu - header)) with the discovered mss, LEDBAT-driven window, and Nagle coalescing.