Merge ack Cut 3: the 4 MiB send bound counts every byte above the lowest unacknowledged packet
F1: acknowledged sequences above a gap still count until the lowest advances, in Rust, TS, Python and C#. Rust direct send purges expired packets before the window check. Soul passes: F1 held; batch 2 held with 13 own mutants killed and interop 27/27 (ack1d-interop:2). Only docs differ from the verified tree. Follow-up: cultmesh-rs Cargo.lock stale. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens
commit
1210fdc7ae
10 changed files with 1098 additions and 44 deletions
|
|
@ -271,6 +271,15 @@ Current progress:
|
|||
flush where the runtime supports one, so callers that require confirmed
|
||||
transport delivery do not substitute a fixed sleep. TCP-framed profiles do
|
||||
not publish the field because they do not own the RUDP pending-reliable queue.
|
||||
- The receive window is a contract, not a discovery. A receiver must hold 1,023
|
||||
sequences and 4 MiB of payload beyond its contiguous received watermark, and
|
||||
a conforming sender never exceeds either. The sender's flow window enforces
|
||||
it without a wire field: a reliable packet is admitted only while its
|
||||
sequence is at most 1,023 above the lowest unacked sequence and the payload
|
||||
above that sequence stays within 4 MiB. The first packet is always
|
||||
admissible, so a payload larger than 4 MiB goes out alone, and packets wait
|
||||
in the FIFO in order, so a small packet never overtakes a large one. C#,
|
||||
Rust, TypeScript, and Python enforce it; Kotlin does not yet.
|
||||
- TypeScript, C#, Rust, Python, and Kotlin now share a portable
|
||||
`cultnet.reconnect_policy.v0` document and deterministic exponential delay
|
||||
helper, and RUDP transport profiles advertise that policy under
|
||||
|
|
|
|||
|
|
@ -438,6 +438,10 @@ def create_rudp_transport_profile(
|
|||
class CultNetRudpSession:
|
||||
RELIABLE_SEND_WINDOW_PACKETS = 32
|
||||
RECEIVED_SEQUENCE_WINDOW = 4_096
|
||||
# Flow window: a reliable packet is admitted only while its sequence is at most FLOW_WINDOW_SEQUENCES above the
|
||||
# lowest unacked one and the payload above that sequence stays within FLOW_WINDOW_BYTES.
|
||||
FLOW_WINDOW_SEQUENCES = 1_023
|
||||
FLOW_WINDOW_BYTES = 4 * 1024 * 1024
|
||||
|
||||
def __init__(self, options: CultNetRudpSessionOptions) -> None:
|
||||
self.connection_id = _uint32(options.connection_id, "connection_id")
|
||||
|
|
@ -478,6 +482,9 @@ class CultNetRudpSession:
|
|||
self._received_sequences: set[int] = set()
|
||||
self._latest_sequenced_by_channel: dict[str, int] = {}
|
||||
self._pending_reliable: dict[int, _PendingReliablePacket] = {}
|
||||
# Payload sizes of acknowledged reliable sequences above the lowest unacknowledged one. The receiver still
|
||||
# holds those bytes behind the gap, so the flow window keeps counting them until the lowest sequence passes them.
|
||||
self._acked_above_lowest: dict[int, int] = {}
|
||||
self._queued_reliable: deque[CultNetRudpPacket] = deque()
|
||||
# Every reliable sequence up to and including this one has been received
|
||||
# since the peer state was last reset. Only the handshake seeds it: the
|
||||
|
|
@ -540,6 +547,7 @@ class CultNetRudpSession:
|
|||
self._awaiting_accept = False
|
||||
self._generation += 1
|
||||
self._pending_reliable.clear()
|
||||
self._acked_above_lowest.clear()
|
||||
self._queued_reliable.clear()
|
||||
|
||||
def reset_peer_state(self) -> None:
|
||||
|
|
@ -656,7 +664,7 @@ class CultNetRudpSession:
|
|||
options: CultNetRudpSendOptions | None = None,
|
||||
) -> CultNetRudpPacket:
|
||||
resolved = options or CultNetRudpSendOptions()
|
||||
if resolved.reliable and len(self._pending_reliable) >= self.RELIABLE_SEND_WINDOW_PACKETS:
|
||||
if resolved.reliable and (self._queued_reliable or not self._window_admits(self._next_sequence, len(payload))):
|
||||
raise ValueError("RUDP reliable send window is full; receive acknowledgements before sending")
|
||||
return self.send_many(channel_id, payload, resolved)[0]
|
||||
|
||||
|
|
@ -935,22 +943,39 @@ class CultNetRudpSession:
|
|||
last_sent_at_ms=now_ms,
|
||||
)
|
||||
|
||||
def _window_admits(self, sequence: int, payload_length: int) -> bool:
|
||||
"""Whether a reliable packet may go on the wire now: the window has a slot, and its sequence and the
|
||||
payload above the lowest unacked sequence stay inside the flow window. With nothing pending, any packet
|
||||
is admissible."""
|
||||
if len(self._pending_reliable) >= self.RELIABLE_SEND_WINDOW_PACKETS:
|
||||
return False
|
||||
if not self._pending_reliable:
|
||||
return True
|
||||
lowest = min(self._pending_reliable)
|
||||
bytes_above = sum(
|
||||
len(pending.packet.payload) for pending_sequence, pending in self._pending_reliable.items() if pending_sequence > lowest
|
||||
) + sum(self._acked_above_lowest.values())
|
||||
return sequence - lowest <= self.FLOW_WINDOW_SEQUENCES and bytes_above + payload_length <= self.FLOW_WINDOW_BYTES
|
||||
|
||||
def _admit_reliable_packets(
|
||||
self,
|
||||
packets: tuple[CultNetRudpPacket, ...],
|
||||
now_ms: int,
|
||||
) -> tuple[CultNetRudpPacket, ...]:
|
||||
available = max(0, self.RELIABLE_SEND_WINDOW_PACKETS - len(self._pending_reliable))
|
||||
ready = packets[:available]
|
||||
for packet in ready:
|
||||
self._track_reliable(packet, now_ms)
|
||||
self._queued_reliable.extend(packets[available:])
|
||||
return ready
|
||||
ready: list[CultNetRudpPacket] = []
|
||||
for packet in packets:
|
||||
if not self._queued_reliable and self._window_admits(packet.sequence, len(packet.payload)):
|
||||
self._track_reliable(packet, now_ms)
|
||||
ready.append(packet)
|
||||
else:
|
||||
self._queued_reliable.append(packet)
|
||||
return tuple(ready)
|
||||
|
||||
def _promote_queued_reliable(self, now_ms: int) -> tuple[CultNetRudpPacket, ...]:
|
||||
available = max(0, self.RELIABLE_SEND_WINDOW_PACKETS - len(self._pending_reliable))
|
||||
ready: list[CultNetRudpPacket] = []
|
||||
while len(ready) < available and self._queued_reliable:
|
||||
while self._queued_reliable and self._window_admits(
|
||||
self._queued_reliable[0].sequence, len(self._queued_reliable[0].payload)
|
||||
):
|
||||
packet = self._queued_reliable.popleft()
|
||||
self._track_reliable(packet, now_ms)
|
||||
ready.append(packet)
|
||||
|
|
@ -963,10 +988,21 @@ class CultNetRudpSession:
|
|||
raise ValueError("RUDP reliable send queue is full")
|
||||
|
||||
def _apply_acknowledgements(self, packet: CultNetRudpPacket) -> None:
|
||||
self._pending_reliable.pop(packet.ack, None)
|
||||
self._acknowledge(packet.ack)
|
||||
for bit in range(32):
|
||||
if packet.ack_mask & (1 << bit):
|
||||
self._pending_reliable.pop(packet.ack - bit - 1, None)
|
||||
self._acknowledge(packet.ack - bit - 1)
|
||||
# Acknowledged sizes stop counting once the lowest unacknowledged sequence passes them, and all of them stop
|
||||
# when nothing is unacknowledged.
|
||||
lowest = min(self._pending_reliable, default=None)
|
||||
self._acked_above_lowest = (
|
||||
{} if lowest is None else {sequence: length for sequence, length in self._acked_above_lowest.items() if sequence > lowest}
|
||||
)
|
||||
|
||||
def _acknowledge(self, sequence: int) -> None:
|
||||
pending = self._pending_reliable.pop(sequence, None)
|
||||
if pending is not None:
|
||||
self._acked_above_lowest[sequence] = len(pending.packet.payload)
|
||||
|
||||
def _seed_received(self, sequence: int) -> None:
|
||||
"""The handshake's one act on the watermark: the peer's Connect or Accept
|
||||
|
|
|
|||
|
|
@ -879,6 +879,163 @@ class CultNetTests(unittest.TestCase):
|
|||
self.assertEqual(old_ack.ack, oldest_sequence)
|
||||
self.assertEqual(old_ack.ack_mask, 0)
|
||||
|
||||
# The sender's flow window: a reliable packet goes on the wire only while its sequence is at most 1,023 above
|
||||
# the lowest unacked one and the payload above that sequence stays within 4 MiB.
|
||||
FLOW_CONNECTION_ID = 0x464C4F57
|
||||
MIB = 1024 * 1024
|
||||
|
||||
def _flow_session(self, initial_sequence: int) -> CultNetRudpSession:
|
||||
return CultNetRudpSession(
|
||||
CultNetRudpSessionOptions(connection_id=self.FLOW_CONNECTION_ID, initial_sequence=initial_sequence)
|
||||
)
|
||||
|
||||
def _connected_flow_pair(
|
||||
self, sender_initial_sequence: int, receiver_initial_sequence: int
|
||||
) -> tuple[CultNetRudpSession, CultNetRudpSession]:
|
||||
"""A sender and a receiver that have shaken hands: the receiver's watermark is seeded by the sender's
|
||||
Connect, and the sender's Connect is acknowledged, so nothing is pending on the sender."""
|
||||
sender = self._flow_session(sender_initial_sequence)
|
||||
receiver = self._flow_session(receiver_initial_sequence)
|
||||
sender.receive(receiver.accept_connect(sender.create_connect(0), 0), 0)
|
||||
return sender, receiver
|
||||
|
||||
def _connected_flow_session(self, initial_sequence: int) -> CultNetRudpSession:
|
||||
return self._connected_flow_pair(initial_sequence, 900)[0]
|
||||
|
||||
@staticmethod
|
||||
def _flow_send(session: CultNetRudpSession, payload_bytes: int) -> tuple[CultNetRudpPacket, ...]:
|
||||
return session.send_many("state", bytes(payload_bytes), CultNetRudpSendOptions(reliable=True))
|
||||
|
||||
def _flow_ack(self, sequence: int) -> CultNetRudpPacket:
|
||||
return CultNetRudpPacket(CultNetRudpPacketType.ACK, self.FLOW_CONNECTION_ID, 0, sequence, 0, "control")
|
||||
|
||||
def _lose_g_and_fill_the_span(self, session: CultNetRudpSession) -> CultNetRudpPacket:
|
||||
g = self._flow_send(session, 1)[0]
|
||||
for offset in range(1, 1024):
|
||||
self.assertEqual([p.sequence for p in self._flow_send(session, 1)], [g.sequence + offset])
|
||||
session.receive(self._flow_ack(g.sequence + offset), 1)
|
||||
return g
|
||||
|
||||
def test_cultnet_rudp_a_lost_packet_holds_the_sender_1023_sequences_ahead_and_is_still_delivered(self) -> None:
|
||||
sender, receiver = self._connected_flow_pair(1, 100)
|
||||
g = self._flow_send(sender, 1)[0]
|
||||
admitted_after_g = 0
|
||||
for _ in range(4200):
|
||||
for packet in self._flow_send(sender, 1):
|
||||
admitted_after_g += 1
|
||||
receiver.receive(packet, 1)
|
||||
sender.receive(receiver.create_ack(), 1)
|
||||
self.assertEqual(admitted_after_g, 1023)
|
||||
self.assertEqual(sender.queued_reliable_packet_count, 4200 - 1023)
|
||||
|
||||
retransmit = next(p for p in sender.due_resends(1000) if p.sequence == g.sequence)
|
||||
delivered = receiver.receive(retransmit, 1000).delivered
|
||||
self.assertEqual([frame.sequence for frame in delivered], [g.sequence])
|
||||
|
||||
promoted = sender.receive(receiver.create_ack_for_received(g.sequence), 1001).ready_to_send
|
||||
self.assertEqual(promoted[0].sequence, g.sequence + 1024)
|
||||
|
||||
def test_cultnet_rudp_a_direct_send_past_the_span_is_refused_without_consuming_a_sequence(self) -> None:
|
||||
sender = self._connected_flow_session(1)
|
||||
g = self._lose_g_and_fill_the_span(sender)
|
||||
with self.assertRaises(ValueError):
|
||||
sender.send("state", bytes(1), CultNetRudpSendOptions(reliable=True))
|
||||
self.assertEqual(sender.queued_reliable_packet_count, 0)
|
||||
sender.receive(self._flow_ack(g.sequence), 2)
|
||||
following = sender.send("state", bytes(1), CultNetRudpSendOptions(reliable=True))
|
||||
self.assertEqual(following.sequence, g.sequence + 1024)
|
||||
|
||||
def test_cultnet_rudp_payload_above_the_lowest_unacked_sequence_is_bounded_by_4_mib(self) -> None:
|
||||
sender = self._connected_flow_session(1)
|
||||
g = self._flow_send(sender, self.MIB)[0]
|
||||
# The lowest unacked packet's own bytes do not count: four more MiB fit above it.
|
||||
for _ in range(4):
|
||||
self.assertEqual(len(self._flow_send(sender, self.MIB)), 1)
|
||||
self.assertEqual(len(self._flow_send(sender, self.MIB)), 0)
|
||||
self.assertEqual(len(self._flow_send(sender, self.MIB)), 0)
|
||||
self.assertEqual(sender.queued_reliable_packet_count, 2)
|
||||
|
||||
# Acking g moves the floor up one packet: exactly one MiB more fits.
|
||||
promoted = sender.receive(self._flow_ack(g.sequence), 1).ready_to_send
|
||||
self.assertEqual([p.sequence for p in promoted], [g.sequence + 5])
|
||||
self.assertEqual(sender.queued_reliable_packet_count, 1)
|
||||
|
||||
def test_cultnet_rudp_a_packet_larger_than_the_window_goes_out_alone_and_waits_behind_anything_unacked(self) -> None:
|
||||
self.assertEqual(len(self._flow_send(self._connected_flow_session(1), 5 * self.MIB)), 1)
|
||||
sender = self._connected_flow_session(1)
|
||||
g = self._flow_send(sender, 1)[0]
|
||||
self.assertEqual(len(self._flow_send(sender, 5 * self.MIB)), 0)
|
||||
promoted = sender.receive(self._flow_ack(g.sequence), 1).ready_to_send
|
||||
self.assertEqual([p.sequence for p in promoted], [g.sequence + 1])
|
||||
|
||||
def test_cultnet_rudp_a_small_packet_does_not_overtake_a_queued_large_one(self) -> None:
|
||||
sender = self._connected_flow_session(1)
|
||||
g = self._flow_send(sender, 1)[0]
|
||||
self.assertEqual(len(self._flow_send(sender, 3 * self.MIB)), 1)
|
||||
self.assertEqual(len(self._flow_send(sender, 2 * self.MIB)), 0)
|
||||
# A direct send refuses instead of queueing behind it, and takes no sequence.
|
||||
with self.assertRaises(ValueError):
|
||||
sender.send("state", bytes(1), CultNetRudpSendOptions(reliable=True))
|
||||
self.assertEqual(sender.queued_reliable_packet_count, 1)
|
||||
# One byte would fit above g, but the 2 MiB packet is ahead of it.
|
||||
self.assertEqual(len(self._flow_send(sender, 1)), 0)
|
||||
promoted = sender.receive(self._flow_ack(g.sequence), 1).ready_to_send
|
||||
self.assertEqual([p.sequence for p in promoted], [g.sequence + 2, g.sequence + 3])
|
||||
|
||||
def test_cultnet_rudp_acknowledged_bytes_above_a_lost_packet_still_count_against_4_mib(self) -> None:
|
||||
sender = self._connected_flow_session(1)
|
||||
g = self._flow_send(sender, 1)[0]
|
||||
# g is lost. Each 1 MiB frame above it is delivered and acknowledged, and the receiver still holds it behind
|
||||
# the gap, so it keeps counting.
|
||||
for offset in range(1, 5):
|
||||
self.assertEqual([p.sequence for p in self._flow_send(sender, self.MIB)], [g.sequence + offset])
|
||||
sender.receive(self._flow_ack(g.sequence + offset), 1)
|
||||
self.assertEqual(len(self._flow_send(sender, self.MIB)), 0)
|
||||
self.assertEqual(sender.queued_reliable_packet_count, 1)
|
||||
with self.assertRaises(ValueError):
|
||||
sender.send("state", bytes(self.MIB), CultNetRudpSendOptions(reliable=True))
|
||||
|
||||
# g arrives: nothing is held behind a gap any more, and the queue drains.
|
||||
promoted = sender.receive(self._flow_ack(g.sequence), 2).ready_to_send
|
||||
self.assertEqual([p.sequence for p in promoted], [g.sequence + 5])
|
||||
# The bytes above the old gap no longer count: another MiB fits.
|
||||
self.assertEqual([p.sequence for p in self._flow_send(sender, self.MIB)], [g.sequence + 6])
|
||||
|
||||
def test_cultnet_rudp_acknowledged_bytes_count_against_the_lowest_unacked_sequence_as_it_advances(self) -> None:
|
||||
sender = self._connected_flow_session(1)
|
||||
g = self._flow_send(sender, 1)[0]
|
||||
h = self._flow_send(sender, 1)[0]
|
||||
for _ in range(3):
|
||||
self._flow_send(sender, self.MIB)
|
||||
# g and the 3 MiB above h are acknowledged; h is lost.
|
||||
for offset in (2, 3, 4, 0):
|
||||
sender.receive(self._flow_ack(g.sequence + offset), 1)
|
||||
self.assertEqual([p.sequence for p in self._flow_send(sender, self.MIB)], [g.sequence + 5])
|
||||
# 4 MiB sit above h, acknowledged or not: one more byte does not fit.
|
||||
self.assertEqual(len(self._flow_send(sender, 1)), 0)
|
||||
# h arrives, the floor moves to g + 5, and the byte fits.
|
||||
promoted = sender.receive(self._flow_ack(h.sequence), 2).ready_to_send
|
||||
self.assertEqual([p.sequence for p in promoted], [g.sequence + 6])
|
||||
|
||||
def test_cultnet_rudp_a_new_generation_on_the_server_does_not_inherit_acknowledged_bytes(self) -> None:
|
||||
# A server that accepts a fresh Connect may send before its Accept is acknowledged, so no acknowledgement of
|
||||
# the new generation has moved the floor yet. Bytes the ended generation had acknowledged above its own lost
|
||||
# packet belong to no receiver's hold any more and must not count.
|
||||
server = self._flow_session(500)
|
||||
accept = server.accept_connect(self._flow_session(1).create_connect(0), 0)
|
||||
server.receive(self._flow_ack(accept.sequence), 0)
|
||||
g = self._flow_send(server, 1)[0]
|
||||
for offset in range(1, 5):
|
||||
self.assertEqual([p.sequence for p in self._flow_send(server, self.MIB)], [g.sequence + offset])
|
||||
server.receive(self._flow_ack(g.sequence + offset), 1)
|
||||
self.assertEqual(len(self._flow_send(server, self.MIB)), 0)
|
||||
|
||||
next_accept = server.accept_connect(self._flow_session(200_000).create_connect(2), 2)
|
||||
self.assertEqual(next_accept.packet_type, CultNetRudpPacketType.ACCEPT)
|
||||
# The Accept is unacknowledged and the only thing pending: 1 MiB fits above it.
|
||||
sent = server.send("state", bytes(self.MIB), CultNetRudpSendOptions(reliable=True))
|
||||
self.assertEqual(sent.sequence, next_accept.sequence + 1)
|
||||
|
||||
def test_cultnet_rudp_socket_transport_handshakes_and_carries_reliable_ordered_schema_frames(self) -> None:
|
||||
server_socket = bind_udp_socket()
|
||||
client_socket = bind_udp_socket()
|
||||
|
|
|
|||
|
|
@ -53,6 +53,13 @@ const RUDP_RECEIVE_AHEAD_WINDOW: u32 = 1_024;
|
|||
const RUDP_CONNECT_ATTEMPT_MS: u64 = 3_000;
|
||||
const RUDP_MAX_ORDERED_BUFFERED_FRAMES: usize = 1_024;
|
||||
const RUDP_MAX_ORDERED_BUFFERED_BYTES: usize = 4 * 1024 * 1024;
|
||||
/// The sender's flow window: a reliable packet goes on the wire only while its
|
||||
/// sequence is at most this far above the lowest unacked one, and the payload
|
||||
/// above that sequence stays within the byte bound. A receiver holds
|
||||
/// `RUDP_RECEIVE_AHEAD_WINDOW - 1` sequences and `RUDP_MAX_ORDERED_BUFFERED_BYTES`
|
||||
/// beyond its watermark, so a sender that obeys this never overruns it.
|
||||
const RUDP_FLOW_WINDOW_SEQUENCES: u32 = RUDP_RECEIVE_AHEAD_WINDOW - 1;
|
||||
const RUDP_FLOW_WINDOW_BYTES: usize = RUDP_MAX_ORDERED_BUFFERED_BYTES;
|
||||
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
||||
pub enum CultNetRudpPacketType {
|
||||
|
|
@ -244,6 +251,11 @@ pub struct CultNetRudpSession {
|
|||
received_sequences: BTreeSet<u32>,
|
||||
latest_sequenced_by_channel: BTreeMap<String, u32>,
|
||||
pending_reliable: BTreeMap<u32, PendingReliablePacket>,
|
||||
/// Payload sizes of acknowledged reliable sequences above the lowest
|
||||
/// unacknowledged one. The receiver still holds those bytes behind the gap,
|
||||
/// so the flow window keeps counting them until the lowest sequence
|
||||
/// advances past them.
|
||||
acked_above_lowest: BTreeMap<u32, usize>,
|
||||
queued_reliable: VecDeque<(CultNetRudpPacket, Option<u64>)>,
|
||||
reliable_packets_expired: u64,
|
||||
/// Every reliable sequence up to and including this one has been received
|
||||
|
|
@ -284,6 +296,7 @@ impl CultNetRudpSession {
|
|||
received_sequences: BTreeSet::new(),
|
||||
latest_sequenced_by_channel: BTreeMap::new(),
|
||||
pending_reliable: BTreeMap::new(),
|
||||
acked_above_lowest: BTreeMap::new(),
|
||||
queued_reliable: VecDeque::new(),
|
||||
reliable_packets_expired: 0,
|
||||
received_through: None,
|
||||
|
|
@ -368,6 +381,7 @@ impl CultNetRudpSession {
|
|||
self.awaiting_accept = false;
|
||||
self.generation += 1;
|
||||
self.pending_reliable.clear();
|
||||
self.acked_above_lowest.clear();
|
||||
self.queued_reliable.clear();
|
||||
}
|
||||
|
||||
|
|
@ -560,8 +574,14 @@ impl CultNetRudpSession {
|
|||
payload: Vec<u8>,
|
||||
options: CultNetRudpSendOptions,
|
||||
) -> Result<CultNetRudpPacket> {
|
||||
// Measure the window as `send_many` will: without the deadline-passed
|
||||
// sends it is about to reclaim.
|
||||
if options.reliable {
|
||||
self.purge_expired_reliable(options.now_ms);
|
||||
}
|
||||
if options.reliable
|
||||
&& self.pending_reliable.len() >= CULTNET_RUDP_RELIABLE_SEND_WINDOW_PACKETS
|
||||
&& (!self.queued_reliable.is_empty()
|
||||
|| !self.window_admits(self.next_sequence, payload.len()))
|
||||
{
|
||||
return Err(anyhow!(
|
||||
"RUDP reliable send window is full; receive acknowledgements before sending"
|
||||
|
|
@ -1073,22 +1093,47 @@ impl CultNetRudpSession {
|
|||
self.queued_reliable
|
||||
.retain(|(_, expires_at_ms)| expires_at_ms.is_none_or(|at| now_ms <= at));
|
||||
let after = self.pending_reliable.len() + self.queued_reliable.len();
|
||||
if after != before {
|
||||
self.forget_acked_below_lowest();
|
||||
}
|
||||
self.reliable_packets_expired = self
|
||||
.reliable_packets_expired
|
||||
.saturating_add((before - after) as u64);
|
||||
}
|
||||
|
||||
/// Whether a reliable packet may go on the wire now: the window has a slot,
|
||||
/// and its sequence and the payload above the lowest unacked sequence stay
|
||||
/// inside the flow window. With nothing pending, any packet is admissible.
|
||||
fn window_admits(&self, sequence: u32, payload_len: usize) -> bool {
|
||||
if self.pending_reliable.len() >= CULTNET_RUDP_RELIABLE_SEND_WINDOW_PACKETS {
|
||||
return false;
|
||||
}
|
||||
let Some((&lowest, _)) = self.pending_reliable.first_key_value() else {
|
||||
return true;
|
||||
};
|
||||
let pending_bytes: usize = self
|
||||
.pending_reliable
|
||||
.iter()
|
||||
.filter(|(pending_sequence, _)| **pending_sequence > lowest)
|
||||
.map(|(_, pending)| pending.packet.payload.len())
|
||||
.sum();
|
||||
let acked_bytes: usize = self.acked_above_lowest.values().sum();
|
||||
let bytes_above = pending_bytes + acked_bytes;
|
||||
sequence.saturating_sub(lowest) <= RUDP_FLOW_WINDOW_SEQUENCES
|
||||
&& bytes_above + payload_len <= RUDP_FLOW_WINDOW_BYTES
|
||||
}
|
||||
|
||||
fn admit_reliable_packets(
|
||||
&mut self,
|
||||
packets: Vec<CultNetRudpPacket>,
|
||||
now_ms: u64,
|
||||
expires_at_ms: Option<u64>,
|
||||
) -> Vec<CultNetRudpPacket> {
|
||||
let available =
|
||||
CULTNET_RUDP_RELIABLE_SEND_WINDOW_PACKETS.saturating_sub(self.pending_reliable.len());
|
||||
let mut ready = Vec::with_capacity(available.min(packets.len()));
|
||||
let mut ready = Vec::new();
|
||||
for packet in packets {
|
||||
if ready.len() < available {
|
||||
if self.queued_reliable.is_empty()
|
||||
&& self.window_admits(packet.sequence, packet.payload.len())
|
||||
{
|
||||
self.track_reliable(packet.clone(), now_ms, expires_at_ms);
|
||||
ready.push(packet);
|
||||
} else {
|
||||
|
|
@ -1099,10 +1144,12 @@ impl CultNetRudpSession {
|
|||
}
|
||||
|
||||
fn promote_queued_reliable(&mut self, now_ms: u64) -> Vec<CultNetRudpPacket> {
|
||||
let available =
|
||||
CULTNET_RUDP_RELIABLE_SEND_WINDOW_PACKETS.saturating_sub(self.pending_reliable.len());
|
||||
let mut ready = Vec::with_capacity(available.min(self.queued_reliable.len()));
|
||||
for _ in 0..available {
|
||||
let mut ready = Vec::new();
|
||||
while self
|
||||
.queued_reliable
|
||||
.front()
|
||||
.is_some_and(|(packet, _)| self.window_admits(packet.sequence, packet.payload.len()))
|
||||
{
|
||||
let Some((packet, expires_at_ms)) = self.queued_reliable.pop_front() else {
|
||||
break;
|
||||
};
|
||||
|
|
@ -1133,12 +1180,29 @@ impl CultNetRudpSession {
|
|||
}
|
||||
|
||||
fn apply_acknowledgements(&mut self, packet: &CultNetRudpPacket) {
|
||||
self.pending_reliable.remove(&packet.ack);
|
||||
self.acknowledge(packet.ack);
|
||||
for bit in 0..32 {
|
||||
if (packet.ack_mask & (1_u32 << bit)) != 0 && packet.ack > bit {
|
||||
self.pending_reliable.remove(&(packet.ack - bit - 1));
|
||||
self.acknowledge(packet.ack - bit - 1);
|
||||
}
|
||||
}
|
||||
self.forget_acked_below_lowest();
|
||||
}
|
||||
|
||||
fn acknowledge(&mut self, sequence: u32) {
|
||||
if let Some(pending) = self.pending_reliable.remove(&sequence) {
|
||||
self.acked_above_lowest
|
||||
.insert(sequence, pending.packet.payload.len());
|
||||
}
|
||||
}
|
||||
|
||||
/// Acknowledged sizes stop counting once the lowest unacknowledged
|
||||
/// sequence passes them, and all of them stop when nothing is unacknowledged.
|
||||
fn forget_acked_below_lowest(&mut self) {
|
||||
match self.pending_reliable.first_key_value() {
|
||||
Some((&lowest, _)) => self.acked_above_lowest.retain(|sequence, _| *sequence > lowest),
|
||||
None => self.acked_above_lowest.clear(),
|
||||
}
|
||||
}
|
||||
|
||||
/// The handshake's one act on the watermark: the peer's Connect or Accept
|
||||
|
|
|
|||
|
|
@ -575,9 +575,30 @@ fn a_flush_that_a_timeout_ended_names_the_timeout() -> Result<()> {
|
|||
// Hold-buffer boundaries: 1,024 frames and 64 channels are held; one more is
|
||||
// refused.
|
||||
|
||||
/// An acknowledgement that names only `sequence`.
|
||||
fn ack_naming(sequence: u32) -> CultNetRudpPacket {
|
||||
CultNetRudpPacket {
|
||||
packet_type: CultNetRudpPacketType::Ack,
|
||||
connection_id: CONNECTION_ID,
|
||||
sequence: 0,
|
||||
ack: sequence,
|
||||
ack_mask: 0,
|
||||
channel_id: "control".into(),
|
||||
reliable: false,
|
||||
ordered: false,
|
||||
sequenced: false,
|
||||
fragment_id: 0,
|
||||
fragment_index: 0,
|
||||
fragment_count: 0,
|
||||
payload: Vec::new(),
|
||||
}
|
||||
}
|
||||
|
||||
/// A connected pair with one sequence the receiver will never see, so every
|
||||
/// later ordered frame waits behind it. The sender's window is kept open by
|
||||
/// acknowledging what the receiver has.
|
||||
/// later ordered frame waits behind it. The sender's flow window would stop it
|
||||
/// 1,023 sequences above its lowest unacked packet, so the missing one is
|
||||
/// acknowledged by hand (the receiver never saw it) and the rest by what the
|
||||
/// receiver has: the receiver's hold buffer is what this probes, not the sender.
|
||||
fn held_behind_a_gap(
|
||||
frames: usize,
|
||||
channel_of: impl Fn(usize) -> String,
|
||||
|
|
@ -585,7 +606,8 @@ fn held_behind_a_gap(
|
|||
let mut sender = session(1);
|
||||
let mut receiver = session(500);
|
||||
handshake(&mut sender, &mut receiver)?;
|
||||
let _missing = sender.send("gap", b"lost".to_vec(), ordered())?;
|
||||
let missing = sender.send("gap", b"lost".to_vec(), ordered())?;
|
||||
sender.receive(&ack_naming(missing.sequence), 1)?;
|
||||
let mut held = 0;
|
||||
for index in 0..frames {
|
||||
let packet = sender.send(&channel_of(index), vec![0], ordered())?;
|
||||
|
|
|
|||
310
packages/cultnet-rs/tests/rudp_flow_window.rs
Normal file
310
packages/cultnet-rs/tests/rudp_flow_window.rs
Normal file
|
|
@ -0,0 +1,310 @@
|
|||
//! The sender's flow window: a reliable packet goes on the wire only while its
|
||||
//! sequence is at most 1,023 above the lowest unacked one and the payload above
|
||||
//! that sequence stays within 4 MiB. Everything else waits in order in the queue.
|
||||
|
||||
use anyhow::Result;
|
||||
use cultnet_rs::*;
|
||||
|
||||
const CONNECTION_ID: u32 = 0x464c_4f57;
|
||||
const MIB: usize = 1024 * 1024;
|
||||
|
||||
fn session(initial_sequence: u32) -> CultNetRudpSession {
|
||||
CultNetRudpSession::new(CultNetRudpSessionOptions {
|
||||
connection_id: CONNECTION_ID,
|
||||
initial_sequence: Some(initial_sequence),
|
||||
resend_delay_ms: 25,
|
||||
max_pending_reliable_packets: None,
|
||||
})
|
||||
}
|
||||
|
||||
/// A sender and a receiver that have shaken hands: the receiver's watermark is
|
||||
/// seeded by the sender's Connect, and the sender's Connect is acknowledged, so
|
||||
/// nothing is pending on the sender.
|
||||
fn connected_pair(
|
||||
sender_initial_sequence: u32,
|
||||
receiver_initial_sequence: u32,
|
||||
) -> Result<(CultNetRudpSession, CultNetRudpSession)> {
|
||||
let mut sender = session(sender_initial_sequence);
|
||||
let mut receiver = session(receiver_initial_sequence);
|
||||
let connect = sender.create_connect(0, Vec::new())?;
|
||||
let accept = receiver.accept_connect(&connect, 0, Vec::new())?;
|
||||
sender.receive(&accept, 0)?;
|
||||
Ok((sender, receiver))
|
||||
}
|
||||
|
||||
fn connected(initial_sequence: u32) -> Result<CultNetRudpSession> {
|
||||
Ok(connected_pair(initial_sequence, 900)?.0)
|
||||
}
|
||||
|
||||
fn reliable() -> CultNetRudpSendOptions {
|
||||
CultNetRudpSendOptions {
|
||||
reliable: true,
|
||||
..Default::default()
|
||||
}
|
||||
}
|
||||
|
||||
/// What `send_many` puts on the wire now; the rest of the packet waits queued.
|
||||
fn send(session: &mut CultNetRudpSession, payload_bytes: usize) -> Result<Vec<CultNetRudpPacket>> {
|
||||
session.send_many("state", vec![7; payload_bytes], reliable(), None)
|
||||
}
|
||||
|
||||
fn ack_for(sequence: u32) -> CultNetRudpPacket {
|
||||
CultNetRudpPacket {
|
||||
packet_type: CultNetRudpPacketType::Ack,
|
||||
connection_id: CONNECTION_ID,
|
||||
sequence: 0,
|
||||
ack: sequence,
|
||||
ack_mask: 0,
|
||||
channel_id: "control".into(),
|
||||
reliable: false,
|
||||
ordered: false,
|
||||
sequenced: false,
|
||||
fragment_id: 0,
|
||||
fragment_index: 0,
|
||||
fragment_count: 0,
|
||||
payload: Vec::new(),
|
||||
}
|
||||
}
|
||||
|
||||
fn sequences(packets: &[CultNetRudpPacket]) -> Vec<u32> {
|
||||
packets.iter().map(|packet| packet.sequence).collect()
|
||||
}
|
||||
|
||||
/// Sends `g`, then acks every packet up to and including the one 1,023 above
|
||||
/// it, leaving `g` the only unacked sequence. Returns `g`.
|
||||
fn lose_g_and_fill_the_span(session: &mut CultNetRudpSession) -> Result<CultNetRudpPacket> {
|
||||
let g = send(session, 1)?.remove(0);
|
||||
for offset in 1..=1_023u32 {
|
||||
let sent = send(session, 1)?;
|
||||
assert_eq!(sequences(&sent), vec![g.sequence + offset]);
|
||||
session.receive(&ack_for(g.sequence + offset), 1)?;
|
||||
}
|
||||
Ok(g)
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_lost_packet_holds_the_sender_1023_sequences_ahead_and_is_still_delivered() -> Result<()> {
|
||||
let (mut sender, mut receiver) = connected_pair(1, 100)?;
|
||||
let g = send(&mut sender, 1)?.remove(0);
|
||||
// g is lost. Everything after it is delivered and acked, until the sender stops.
|
||||
let mut admitted_after_g = 0u32;
|
||||
for _ in 0..4_200 {
|
||||
for packet in send(&mut sender, 1)? {
|
||||
admitted_after_g += 1;
|
||||
receiver.receive(&packet, 1)?;
|
||||
sender.receive(&receiver.create_ack(), 1)?;
|
||||
}
|
||||
}
|
||||
assert_eq!(admitted_after_g, 1_023);
|
||||
assert_eq!(sender.queued_reliable_packet_count(), 4_200 - 1_023);
|
||||
|
||||
let retransmit = sender
|
||||
.due_resends(1_000)
|
||||
.into_iter()
|
||||
.find(|packet| packet.sequence == g.sequence)
|
||||
.expect("g is still pending and due");
|
||||
let delivered = receiver.receive(&retransmit, 1_000)?.delivered;
|
||||
assert_eq!(delivered.len(), 1);
|
||||
assert_eq!(delivered[0].sequence, g.sequence);
|
||||
|
||||
let promoted = sender
|
||||
.receive(&receiver.create_ack_for_received(g.sequence), 1_001)?
|
||||
.ready_to_send;
|
||||
assert_eq!(promoted.first().map(|packet| packet.sequence), Some(g.sequence + 1_024));
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_direct_send_past_the_span_is_refused_without_consuming_a_sequence() -> Result<()> {
|
||||
let mut sender = connected(1)?;
|
||||
let g = lose_g_and_fill_the_span(&mut sender)?;
|
||||
assert!(sender.send("state", vec![7], reliable()).is_err());
|
||||
assert_eq!(sender.queued_reliable_packet_count(), 0);
|
||||
sender.receive(&ack_for(g.sequence), 2)?;
|
||||
let next = sender.send("state", vec![7], reliable())?;
|
||||
assert_eq!(next.sequence, g.sequence + 1_024);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn payload_above_the_lowest_unacked_sequence_is_bounded_by_4_mib() -> Result<()> {
|
||||
let mut sender = connected(1)?;
|
||||
let g = send(&mut sender, MIB)?.remove(0);
|
||||
// The lowest unacked packet's own bytes do not count: four more MiB fit above it.
|
||||
for _ in 0..4 {
|
||||
assert_eq!(send(&mut sender, MIB)?.len(), 1);
|
||||
}
|
||||
assert!(send(&mut sender, MIB)?.is_empty());
|
||||
assert!(send(&mut sender, MIB)?.is_empty());
|
||||
assert_eq!(sender.queued_reliable_packet_count(), 2);
|
||||
|
||||
// Acking g moves the floor up one packet: exactly one MiB more fits.
|
||||
let promoted = sender.receive(&ack_for(g.sequence), 1)?.ready_to_send;
|
||||
assert_eq!(sequences(&promoted), vec![g.sequence + 5]);
|
||||
assert_eq!(sender.queued_reliable_packet_count(), 1);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_packet_larger_than_the_window_goes_out_alone_and_waits_behind_anything_unacked() -> Result<()> {
|
||||
let mut sender = connected(1)?;
|
||||
assert_eq!(send(&mut sender, 5 * MIB)?.len(), 1);
|
||||
let mut sender = connected(1)?;
|
||||
let g = send(&mut sender, 1)?.remove(0);
|
||||
assert!(send(&mut sender, 5 * MIB)?.is_empty());
|
||||
let promoted = sender.receive(&ack_for(g.sequence), 1)?.ready_to_send;
|
||||
assert_eq!(sequences(&promoted), vec![g.sequence + 1]);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_small_packet_does_not_overtake_a_queued_large_one() -> Result<()> {
|
||||
let mut sender = connected(1)?;
|
||||
let g = send(&mut sender, 1)?.remove(0);
|
||||
assert_eq!(send(&mut sender, 3 * MIB)?.len(), 1);
|
||||
assert!(send(&mut sender, 2 * MIB)?.is_empty());
|
||||
// A direct send refuses instead of queueing behind it, and takes no sequence.
|
||||
assert!(sender.send("state", vec![7], reliable()).is_err());
|
||||
assert_eq!(sender.queued_reliable_packet_count(), 1);
|
||||
// One byte would fit above g, but the 2 MiB packet is ahead of it.
|
||||
assert!(send(&mut sender, 1)?.is_empty());
|
||||
let promoted = sender.receive(&ack_for(g.sequence), 1)?.ready_to_send;
|
||||
assert_eq!(sequences(&promoted), vec![g.sequence + 2, g.sequence + 3]);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn acknowledged_bytes_above_a_lost_packet_still_count_against_4_mib() -> Result<()> {
|
||||
let mut sender = connected(1)?;
|
||||
let g = send(&mut sender, 1)?.remove(0);
|
||||
// g is lost. Each 1 MiB frame above it is delivered and acknowledged, and
|
||||
// the receiver still holds it behind the gap, so it keeps counting.
|
||||
for offset in 1..=4u32 {
|
||||
let sent = send(&mut sender, MIB)?;
|
||||
assert_eq!(sequences(&sent), vec![g.sequence + offset]);
|
||||
sender.receive(&ack_for(g.sequence + offset), 1)?;
|
||||
}
|
||||
assert!(send(&mut sender, MIB)?.is_empty());
|
||||
assert_eq!(sender.queued_reliable_packet_count(), 1);
|
||||
assert!(sender.send("state", vec![7; MIB], reliable()).is_err());
|
||||
|
||||
// g arrives: nothing is held behind a gap any more, and the queue drains.
|
||||
let promoted = sender.receive(&ack_for(g.sequence), 2)?.ready_to_send;
|
||||
assert_eq!(sequences(&promoted), vec![g.sequence + 5]);
|
||||
// The bytes above the old gap no longer count: another MiB fits.
|
||||
assert_eq!(sequences(&send(&mut sender, MIB)?), vec![g.sequence + 6]);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn acknowledged_bytes_count_against_the_lowest_unacked_sequence_as_it_advances() -> Result<()> {
|
||||
let mut sender = connected(1)?;
|
||||
let g = send(&mut sender, 1)?.remove(0);
|
||||
let h = send(&mut sender, 1)?.remove(0);
|
||||
for _ in 0..3 {
|
||||
send(&mut sender, MIB)?;
|
||||
}
|
||||
// g and the 3 MiB above h are acknowledged; h is lost.
|
||||
for offset in [2, 3, 4, 0] {
|
||||
sender.receive(&ack_for(g.sequence + offset), 1)?;
|
||||
}
|
||||
assert_eq!(sequences(&send(&mut sender, MIB)?), vec![g.sequence + 5]);
|
||||
// 4 MiB sit above h, acknowledged or not: one more byte does not fit.
|
||||
assert!(send(&mut sender, 1)?.is_empty());
|
||||
// h arrives, the floor moves to g + 5, and the byte fits.
|
||||
let promoted = sender.receive(&ack_for(h.sequence), 2)?.ready_to_send;
|
||||
assert_eq!(sequences(&promoted), vec![g.sequence + 6]);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn acknowledged_bytes_stop_counting_when_the_lowest_unacked_packet_expires() -> Result<()> {
|
||||
let mut sender = connected(1)?;
|
||||
let expiring = CultNetRudpSendOptions {
|
||||
reliable: true,
|
||||
now_ms: 0,
|
||||
reliable_expire_after_ms: Some(100),
|
||||
..Default::default()
|
||||
};
|
||||
let g = sender
|
||||
.send_many("state", vec![7], expiring, None)?
|
||||
.remove(0);
|
||||
for offset in 1..=4u32 {
|
||||
assert_eq!(send(&mut sender, MIB)?.len(), 1);
|
||||
sender.receive(&ack_for(g.sequence + offset), 1)?;
|
||||
}
|
||||
assert!(sender.send("state", vec![7; MIB], reliable()).is_err());
|
||||
|
||||
// g's deadline passes and it leaves the window; nothing is unacked, so the
|
||||
// acknowledged bytes no longer sit above any gap.
|
||||
let at = |now_ms| CultNetRudpSendOptions {
|
||||
reliable: true,
|
||||
now_ms,
|
||||
..Default::default()
|
||||
};
|
||||
let first = sender.send_many("state", vec![7], at(200), None)?;
|
||||
assert_eq!(first.len(), 1);
|
||||
for _ in 0..4 {
|
||||
assert_eq!(sender.send_many("state", vec![7; MIB], at(200), None)?.len(), 1);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_direct_send_is_not_refused_for_bytes_whose_lowest_packet_has_expired() -> Result<()> {
|
||||
let mut sender = connected(1)?;
|
||||
let expiring = CultNetRudpSendOptions {
|
||||
reliable: true,
|
||||
now_ms: 0,
|
||||
reliable_expire_after_ms: Some(100),
|
||||
..Default::default()
|
||||
};
|
||||
let g = sender
|
||||
.send_many("state", vec![7], expiring, None)?
|
||||
.remove(0);
|
||||
for offset in 1..=4u32 {
|
||||
assert_eq!(send(&mut sender, MIB)?.len(), 1);
|
||||
sender.receive(&ack_for(g.sequence + offset), 1)?;
|
||||
}
|
||||
assert!(sender.send("state", vec![7; MIB], reliable()).is_err());
|
||||
|
||||
// The first call after g's deadline is a direct send: it measures the
|
||||
// window without g, as `send_many` would, and goes on the wire.
|
||||
let at_200 = CultNetRudpSendOptions {
|
||||
reliable: true,
|
||||
now_ms: 200,
|
||||
..Default::default()
|
||||
};
|
||||
let sent = sender.send("state", vec![7; MIB], at_200)?;
|
||||
assert_eq!(sent.sequence, g.sequence + 5);
|
||||
assert_eq!(sender.outstanding_reliable_packet_count(), 1);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// A server that accepts a fresh Connect may send before its Accept is
|
||||
/// acknowledged, so no acknowledgement of the new generation has moved the
|
||||
/// floor yet. Bytes the ended generation had acknowledged above its own lost
|
||||
/// packet belong to no receiver's hold any more and must not count.
|
||||
#[test]
|
||||
fn a_new_generation_on_the_server_does_not_inherit_acknowledged_bytes() -> Result<()> {
|
||||
let mut server = session(500);
|
||||
let mut first_client = session(1);
|
||||
let connect = first_client.create_connect(0, Vec::new())?;
|
||||
let accept = server.accept_connect(&connect, 0, Vec::new())?;
|
||||
server.receive(&ack_for(accept.sequence), 0)?;
|
||||
let g = send(&mut server, 1)?.remove(0);
|
||||
for offset in 1..=4u32 {
|
||||
assert_eq!(sequences(&send(&mut server, MIB)?), vec![g.sequence + offset]);
|
||||
server.receive(&ack_for(g.sequence + offset), 1)?;
|
||||
}
|
||||
assert!(send(&mut server, MIB)?.is_empty());
|
||||
|
||||
let mut second_client = session(200_000);
|
||||
let connect = second_client.create_connect(2, Vec::new())?;
|
||||
let accept = server.accept_connect(&connect, 2, Vec::new())?;
|
||||
assert_eq!(accept.packet_type, CultNetRudpPacketType::Accept);
|
||||
// The Accept is unacknowledged and the only thing pending: 1 MiB fits above it.
|
||||
let sent = server.send("state", vec![7; MIB], reliable())?;
|
||||
assert_eq!(sent.sequence, accept.sequence + 1);
|
||||
Ok(())
|
||||
}
|
||||
|
|
@ -33,6 +33,11 @@ const RUDP_FIXED_HEADER_BYTES = 36;
|
|||
const MAX_CHANNEL_ID_BYTES = 255;
|
||||
export const CULTNET_RUDP_RELIABLE_SEND_WINDOW_PACKETS = 32;
|
||||
const RUDP_RECEIVED_SEQUENCE_WINDOW = 4_096;
|
||||
// Flow window: a reliable packet is admitted only while its sequence is at most RUDP_FLOW_WINDOW_SEQUENCES above the
|
||||
// lowest unacked one and the payload above that sequence stays within RUDP_FLOW_WINDOW_BYTES.
|
||||
const RUDP_FLOW_WINDOW_SEQUENCES = 1_023;
|
||||
const RUDP_FLOW_WINDOW_BYTES = 4 * 1024 * 1024;
|
||||
|
||||
/// How long a client keeps retransmitting a Connect nobody answered before it
|
||||
/// abandons that attempt for a fresh one. A Connect the server answers with an
|
||||
/// Ack, never an Accept, is one the server judged stale (see `acceptConnect`);
|
||||
|
|
@ -212,6 +217,9 @@ export class CultNetRudpSession {
|
|||
readonly #receivedSequences = new Set<number>();
|
||||
readonly #latestSequencedByChannel = new Map<string, number>();
|
||||
readonly #pendingReliable = new Map<number, PendingReliablePacket>();
|
||||
/// Payload sizes of acknowledged reliable sequences above the lowest unacknowledged one. The receiver still holds
|
||||
/// those bytes behind the gap, so the flow window keeps counting them until the lowest sequence passes them.
|
||||
readonly #ackedAboveLowest = new Map<number, number>();
|
||||
readonly #queuedReliable: CultNetRudpPacket[] = [];
|
||||
/// Every reliable sequence up to and including this one has been received
|
||||
/// since the peer state was last reset. Only the handshake seeds it: the
|
||||
|
|
@ -294,6 +302,7 @@ export class CultNetRudpSession {
|
|||
this.#awaitingAccept = false;
|
||||
this.#generation += 1;
|
||||
this.#pendingReliable.clear();
|
||||
this.#ackedAboveLowest.clear();
|
||||
this.#queuedReliable.splice(0);
|
||||
}
|
||||
|
||||
|
|
@ -420,7 +429,7 @@ export class CultNetRudpSession {
|
|||
payload: Uint8Array,
|
||||
options: { reliable?: boolean; ordered?: boolean; sequenced?: boolean; nowMs?: number } = {},
|
||||
): CultNetRudpPacket {
|
||||
if (options.reliable && this.#pendingReliable.size >= CULTNET_RUDP_RELIABLE_SEND_WINDOW_PACKETS) {
|
||||
if (options.reliable && (this.#queuedReliable.length > 0 || !this.#windowAdmits(this.#nextSequence, payload.length))) {
|
||||
throw new Error("RUDP reliable send window is full; receive acknowledgements before sending.");
|
||||
}
|
||||
return this.sendMany(channelId, payload, options)[0]!;
|
||||
|
|
@ -753,18 +762,44 @@ export class CultNetRudpSession {
|
|||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Whether a reliable packet may go on the wire now: the window has a slot, and its sequence and the payload above
|
||||
* the lowest unacked sequence stay inside the flow window. With nothing pending, any packet is admissible.
|
||||
*/
|
||||
#windowAdmits(sequence: number, payloadLength: number): boolean {
|
||||
if (this.#pendingReliable.size >= CULTNET_RUDP_RELIABLE_SEND_WINDOW_PACKETS) return false;
|
||||
if (this.#pendingReliable.size === 0) return true;
|
||||
const lowest = Math.min(...this.#pendingReliable.keys());
|
||||
let bytesAbove = 0;
|
||||
for (const [pendingSequence, pending] of this.#pendingReliable) {
|
||||
if (pendingSequence > lowest) bytesAbove += pending.packet.payload?.length ?? 0;
|
||||
}
|
||||
for (const length of this.#ackedAboveLowest.values()) bytesAbove += length;
|
||||
return sequence - lowest <= RUDP_FLOW_WINDOW_SEQUENCES && bytesAbove + payloadLength <= RUDP_FLOW_WINDOW_BYTES;
|
||||
}
|
||||
|
||||
#admitReliablePackets(packets: CultNetRudpPacket[], nowMs: number): CultNetRudpPacket[] {
|
||||
const available = Math.max(0, CULTNET_RUDP_RELIABLE_SEND_WINDOW_PACKETS - this.#pendingReliable.size);
|
||||
const ready = packets.slice(0, available);
|
||||
for (const packet of ready) this.#trackReliable(packet, nowMs);
|
||||
this.#queuedReliable.push(...packets.slice(available));
|
||||
const ready: CultNetRudpPacket[] = [];
|
||||
for (const packet of packets) {
|
||||
if (this.#queuedReliable.length === 0 && this.#windowAdmits(packet.sequence, packet.payload?.length ?? 0)) {
|
||||
this.#trackReliable(packet, nowMs);
|
||||
ready.push(packet);
|
||||
} else {
|
||||
this.#queuedReliable.push(packet);
|
||||
}
|
||||
}
|
||||
return ready;
|
||||
}
|
||||
|
||||
#promoteQueuedReliable(nowMs: number): CultNetRudpPacket[] {
|
||||
const available = Math.max(0, CULTNET_RUDP_RELIABLE_SEND_WINDOW_PACKETS - this.#pendingReliable.size);
|
||||
const ready = this.#queuedReliable.splice(0, available);
|
||||
for (const packet of ready) this.#trackReliable(packet, nowMs);
|
||||
const ready: CultNetRudpPacket[] = [];
|
||||
while (this.#queuedReliable.length > 0) {
|
||||
const packet = this.#queuedReliable[0];
|
||||
if (!this.#windowAdmits(packet.sequence, packet.payload?.length ?? 0)) break;
|
||||
this.#queuedReliable.shift();
|
||||
this.#trackReliable(packet, nowMs);
|
||||
ready.push(packet);
|
||||
}
|
||||
return ready;
|
||||
}
|
||||
|
||||
|
|
@ -778,12 +813,33 @@ export class CultNetRudpSession {
|
|||
}
|
||||
|
||||
#applyAcknowledgements(packet: CultNetRudpPacket): void {
|
||||
this.#pendingReliable.delete(packet.ack);
|
||||
this.#acknowledge(packet.ack);
|
||||
for (let bit = 0; bit < 32; bit += 1) {
|
||||
if ((packet.ackMask & (1 << bit)) !== 0) {
|
||||
this.#pendingReliable.delete(packet.ack - bit - 1);
|
||||
this.#acknowledge(packet.ack - bit - 1);
|
||||
}
|
||||
}
|
||||
this.#forgetAckedBelowLowest();
|
||||
}
|
||||
|
||||
#acknowledge(sequence: number): void {
|
||||
const pending = this.#pendingReliable.get(sequence);
|
||||
if (pending === undefined) return;
|
||||
this.#pendingReliable.delete(sequence);
|
||||
this.#ackedAboveLowest.set(sequence, pending.packet.payload?.length ?? 0);
|
||||
}
|
||||
|
||||
/// Acknowledged sizes stop counting once the lowest unacknowledged sequence passes them, and all of them stop
|
||||
/// when nothing is unacknowledged.
|
||||
#forgetAckedBelowLowest(): void {
|
||||
if (this.#pendingReliable.size === 0) {
|
||||
this.#ackedAboveLowest.clear();
|
||||
return;
|
||||
}
|
||||
const lowest = Math.min(...this.#pendingReliable.keys());
|
||||
for (const sequence of this.#ackedAboveLowest.keys()) {
|
||||
if (sequence <= lowest) this.#ackedAboveLowest.delete(sequence);
|
||||
}
|
||||
}
|
||||
|
||||
/// The handshake's one act on the watermark: the peer's Connect or Accept is
|
||||
|
|
|
|||
|
|
@ -1417,6 +1417,167 @@ test("rudp session advances large fragment sets through a bounded reliable windo
|
|||
assert.equal(oldAck.ackMask, 0);
|
||||
});
|
||||
|
||||
// The sender's flow window: a reliable packet goes on the wire only while its sequence is at most 1,023 above the
|
||||
// lowest unacked one and the payload above that sequence stays within 4 MiB.
|
||||
const FLOW_CONNECTION_ID = 0x464c4f57;
|
||||
const MIB = 1024 * 1024;
|
||||
|
||||
function flowSession(initialSequence: number): CultNetRudpSession {
|
||||
return new CultNetRudpSession({ connectionId: FLOW_CONNECTION_ID, initialSequence, resendDelayMs: 25 });
|
||||
}
|
||||
|
||||
// A sender and a receiver that have shaken hands: the receiver's watermark is seeded by the sender's Connect, and the
|
||||
// sender's Connect is acknowledged, so nothing is pending on the sender.
|
||||
function connectedFlowPair(senderInitialSequence: number, receiverInitialSequence: number): [CultNetRudpSession, CultNetRudpSession] {
|
||||
const sender = flowSession(senderInitialSequence);
|
||||
const receiver = flowSession(receiverInitialSequence);
|
||||
sender.receive(receiver.acceptConnect(sender.createConnect(0), 0), 0);
|
||||
return [sender, receiver];
|
||||
}
|
||||
|
||||
function connectedFlowSession(initialSequence: number): CultNetRudpSession {
|
||||
return connectedFlowPair(initialSequence, 900)[0];
|
||||
}
|
||||
|
||||
function flowSend(session: CultNetRudpSession, payloadBytes: number): CultNetRudpPacket[] {
|
||||
return session.sendMany("state", new Uint8Array(payloadBytes), { reliable: true });
|
||||
}
|
||||
|
||||
function flowAck(sequence: number): CultNetRudpPacket {
|
||||
return { packetType: "ack", connectionId: FLOW_CONNECTION_ID, sequence: 0, ack: sequence, ackMask: 0, channelId: "control" };
|
||||
}
|
||||
|
||||
const sequencesOf = (packets: CultNetRudpPacket[]): number[] => packets.map((packet) => packet.sequence);
|
||||
|
||||
function loseGAndFillTheSpan(session: CultNetRudpSession): CultNetRudpPacket {
|
||||
const g = flowSend(session, 1)[0]!;
|
||||
for (let offset = 1; offset <= 1023; offset++) {
|
||||
assert.deepEqual(sequencesOf(flowSend(session, 1)), [g.sequence + offset]);
|
||||
session.receive(flowAck(g.sequence + offset), 1);
|
||||
}
|
||||
return g;
|
||||
}
|
||||
|
||||
test("a lost packet holds the sender 1023 sequences ahead and is still delivered", () => {
|
||||
const [sender, receiver] = connectedFlowPair(1, 100);
|
||||
const g = flowSend(sender, 1)[0]!;
|
||||
let admittedAfterG = 0;
|
||||
for (let index = 0; index < 4200; index++) {
|
||||
for (const packet of flowSend(sender, 1)) {
|
||||
admittedAfterG++;
|
||||
receiver.receive(packet, 1);
|
||||
sender.receive(receiver.createAck(), 1);
|
||||
}
|
||||
}
|
||||
assert.equal(admittedAfterG, 1023);
|
||||
assert.equal(sender.queuedReliablePacketCount, 4200 - 1023);
|
||||
|
||||
const retransmit = sender.dueResends(1000).find((packet) => packet.sequence === g.sequence)!;
|
||||
const delivered = receiver.receive(retransmit, 1000).delivered;
|
||||
assert.deepEqual(delivered.map((frame) => frame.sequence), [g.sequence]);
|
||||
|
||||
const promoted = sender.receive(receiver.createAckForReceived(g.sequence), 1001).readyToSend ?? [];
|
||||
assert.equal(promoted[0]!.sequence, g.sequence + 1024);
|
||||
});
|
||||
|
||||
test("a direct send past the span is refused without consuming a sequence", () => {
|
||||
const sender = connectedFlowSession(1);
|
||||
const g = loseGAndFillTheSpan(sender);
|
||||
assert.throws(() => sender.send("state", new Uint8Array(1), { reliable: true }));
|
||||
assert.equal(sender.queuedReliablePacketCount, 0);
|
||||
sender.receive(flowAck(g.sequence), 2);
|
||||
assert.equal(sender.send("state", new Uint8Array(1), { reliable: true }).sequence, g.sequence + 1024);
|
||||
});
|
||||
|
||||
test("payload above the lowest unacked sequence is bounded by 4 MiB", () => {
|
||||
const sender = connectedFlowSession(1);
|
||||
const g = flowSend(sender, MIB)[0]!;
|
||||
// The lowest unacked packet's own bytes do not count: four more MiB fit above it.
|
||||
for (let index = 0; index < 4; index++) assert.equal(flowSend(sender, MIB).length, 1);
|
||||
assert.equal(flowSend(sender, MIB).length, 0);
|
||||
assert.equal(flowSend(sender, MIB).length, 0);
|
||||
assert.equal(sender.queuedReliablePacketCount, 2);
|
||||
|
||||
// Acking g moves the floor up one packet: exactly one MiB more fits.
|
||||
const promoted = sender.receive(flowAck(g.sequence), 1).readyToSend ?? [];
|
||||
assert.deepEqual(sequencesOf(promoted), [g.sequence + 5]);
|
||||
assert.equal(sender.queuedReliablePacketCount, 1);
|
||||
});
|
||||
|
||||
test("a packet larger than the window goes out alone and waits behind anything unacked", () => {
|
||||
assert.equal(flowSend(connectedFlowSession(1), 5 * MIB).length, 1);
|
||||
const sender = connectedFlowSession(1);
|
||||
const g = flowSend(sender, 1)[0]!;
|
||||
assert.equal(flowSend(sender, 5 * MIB).length, 0);
|
||||
assert.deepEqual(sequencesOf(sender.receive(flowAck(g.sequence), 1).readyToSend ?? []), [g.sequence + 1]);
|
||||
});
|
||||
|
||||
test("a small packet does not overtake a queued large one", () => {
|
||||
const sender = connectedFlowSession(1);
|
||||
const g = flowSend(sender, 1)[0]!;
|
||||
assert.equal(flowSend(sender, 3 * MIB).length, 1);
|
||||
assert.equal(flowSend(sender, 2 * MIB).length, 0);
|
||||
// A direct send refuses instead of queueing behind it, and takes no sequence.
|
||||
assert.throws(() => sender.send("state", new Uint8Array(1), { reliable: true }));
|
||||
assert.equal(sender.queuedReliablePacketCount, 1);
|
||||
// One byte would fit above g, but the 2 MiB packet is ahead of it.
|
||||
assert.equal(flowSend(sender, 1).length, 0);
|
||||
assert.deepEqual(sequencesOf(sender.receive(flowAck(g.sequence), 1).readyToSend ?? []), [g.sequence + 2, g.sequence + 3]);
|
||||
});
|
||||
|
||||
test("acknowledged bytes above a lost packet still count against 4 MiB", () => {
|
||||
const sender = connectedFlowSession(1);
|
||||
const g = flowSend(sender, 1)[0]!;
|
||||
// g is lost. Each 1 MiB frame above it is delivered and acknowledged, and the receiver still holds it behind the
|
||||
// gap, so it keeps counting.
|
||||
for (let offset = 1; offset <= 4; offset++) {
|
||||
assert.deepEqual(sequencesOf(flowSend(sender, MIB)), [g.sequence + offset]);
|
||||
sender.receive(flowAck(g.sequence + offset), 1);
|
||||
}
|
||||
assert.equal(flowSend(sender, MIB).length, 0);
|
||||
assert.equal(sender.queuedReliablePacketCount, 1);
|
||||
assert.throws(() => sender.send("state", new Uint8Array(MIB), { reliable: true }));
|
||||
|
||||
// g arrives: nothing is held behind a gap any more, and the queue drains.
|
||||
assert.deepEqual(sequencesOf(sender.receive(flowAck(g.sequence), 2).readyToSend ?? []), [g.sequence + 5]);
|
||||
// The bytes above the old gap no longer count: another MiB fits.
|
||||
assert.deepEqual(sequencesOf(flowSend(sender, MIB)), [g.sequence + 6]);
|
||||
});
|
||||
|
||||
test("acknowledged bytes count against the lowest unacked sequence as it advances", () => {
|
||||
const sender = connectedFlowSession(1);
|
||||
const g = flowSend(sender, 1)[0]!;
|
||||
const h = flowSend(sender, 1)[0]!;
|
||||
for (let index = 0; index < 3; index++) flowSend(sender, MIB);
|
||||
// g and the 3 MiB above h are acknowledged; h is lost.
|
||||
for (const offset of [2, 3, 4, 0]) sender.receive(flowAck(g.sequence + offset), 1);
|
||||
assert.deepEqual(sequencesOf(flowSend(sender, MIB)), [g.sequence + 5]);
|
||||
// 4 MiB sit above h, acknowledged or not: one more byte does not fit.
|
||||
assert.equal(flowSend(sender, 1).length, 0);
|
||||
// h arrives, the floor moves to g + 5, and the byte fits.
|
||||
assert.deepEqual(sequencesOf(sender.receive(flowAck(h.sequence), 2).readyToSend ?? []), [g.sequence + 6]);
|
||||
});
|
||||
|
||||
// A server that accepts a fresh Connect may send before its Accept is acknowledged, so no acknowledgement of the new
|
||||
// generation has moved the floor yet. Bytes the ended generation had acknowledged above its own lost packet belong to
|
||||
// no receiver's hold any more and must not count.
|
||||
test("a new generation on the server does not inherit acknowledged bytes", () => {
|
||||
const server = flowSession(500);
|
||||
const accept = server.acceptConnect(flowSession(1).createConnect(0), 0);
|
||||
server.receive(flowAck(accept.sequence), 0);
|
||||
const g = flowSend(server, 1)[0]!;
|
||||
for (let offset = 1; offset <= 4; offset++) {
|
||||
assert.deepEqual(sequencesOf(flowSend(server, MIB)), [g.sequence + offset]);
|
||||
server.receive(flowAck(g.sequence + offset), 1);
|
||||
}
|
||||
assert.equal(flowSend(server, MIB).length, 0);
|
||||
|
||||
const nextAccept = server.acceptConnect(flowSession(200_000).createConnect(2), 2);
|
||||
assert.equal(nextAccept.packetType, "accept");
|
||||
// The Accept is unacknowledged and the only thing pending: 1 MiB fits above it.
|
||||
assert.equal(server.send("state", new Uint8Array(MIB), { reliable: true }).sequence, nextAccept.sequence + 1);
|
||||
});
|
||||
|
||||
test("CultNet contracts encode legacy bytes without Node Buffer authority", () => {
|
||||
const originalBuffer = globalThis.Buffer;
|
||||
try {
|
||||
|
|
|
|||
|
|
@ -885,6 +885,10 @@ namespace GameCult.Networking
|
|||
{
|
||||
/// <summary>Maximum reliable packets admitted to the wire before acknowledgements advance the window.</summary>
|
||||
public const int ReliableSendWindowPackets = 32;
|
||||
// Flow window: a reliable packet is admitted only while its sequence is at most FlowWindowSequences above the
|
||||
// lowest unacked one and the payload above that sequence stays within FlowWindowBytes.
|
||||
private const uint FlowWindowSequences = 1023;
|
||||
private const int FlowWindowBytes = 4 * 1024 * 1024;
|
||||
private const int ReceivedSequenceWindow = 4096;
|
||||
// How long a client keeps retransmitting a Connect nobody answered before it abandons that
|
||||
// attempt for a fresh one. A Connect the server answers with an Ack, never an Accept, is one
|
||||
|
|
@ -935,6 +939,9 @@ namespace GameCult.Networking
|
|||
private readonly Dictionary<string, uint> _latestSequencedByChannel = new Dictionary<string, uint>(StringComparer.Ordinal);
|
||||
private readonly object _pendingReliableGate = new object();
|
||||
private readonly Dictionary<uint, PendingReliablePacket> _pendingReliable = new Dictionary<uint, PendingReliablePacket>();
|
||||
// Payload sizes of acknowledged reliable sequences above the lowest unacknowledged one. The receiver still
|
||||
// holds those bytes behind the gap, so the flow window keeps counting them until the lowest sequence passes them.
|
||||
private readonly Dictionary<uint, int> _ackedAboveLowest = new Dictionary<uint, int>();
|
||||
private readonly Queue<CultNetRudpPacket> _queuedReliable = new Queue<CultNetRudpPacket>();
|
||||
// Every reliable sequence up to and including this one has been received since the peer state
|
||||
// was last reset. Only the handshake seeds it: the peer's Connect on the accepting side, the
|
||||
|
|
@ -1156,7 +1163,7 @@ namespace GameCult.Networking
|
|||
options ??= new CultNetRudpSendOptions();
|
||||
lock (_pendingReliableGate)
|
||||
{
|
||||
if (options.Reliable && _pendingReliable.Count >= ReliableSendWindowPackets)
|
||||
if (options.Reliable && (_queuedReliable.Count > 0 || !WindowAdmits(_nextSequence, payload.Length)))
|
||||
throw new InvalidOperationException("RUDP reliable send window is full; receive acknowledgements before sending.");
|
||||
}
|
||||
return SendMany(channelId, payload, options).First();
|
||||
|
|
@ -1420,6 +1427,7 @@ namespace GameCult.Networking
|
|||
lock (_pendingReliableGate)
|
||||
{
|
||||
_pendingReliable.Clear();
|
||||
_ackedAboveLowest.Clear();
|
||||
_queuedReliable.Clear();
|
||||
}
|
||||
}
|
||||
|
|
@ -1573,17 +1581,39 @@ namespace GameCult.Networking
|
|||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Whether a reliable packet may go on the wire now: the window has a slot, and its sequence and the
|
||||
/// payload above the lowest unacked sequence stay inside the flow window. With nothing pending, any
|
||||
/// packet is admissible. Callers hold the pending gate.
|
||||
/// </summary>
|
||||
private bool WindowAdmits(uint sequence, int payloadLength)
|
||||
{
|
||||
if (_pendingReliable.Count >= ReliableSendWindowPackets)
|
||||
return false;
|
||||
if (_pendingReliable.Count == 0)
|
||||
return true;
|
||||
var lowest = _pendingReliable.Keys.Min();
|
||||
long bytesAbove = 0;
|
||||
foreach (var pending in _pendingReliable)
|
||||
{
|
||||
if (pending.Key > lowest)
|
||||
bytesAbove += pending.Value.Packet.Payload.Length;
|
||||
}
|
||||
foreach (var length in _ackedAboveLowest.Values)
|
||||
bytesAbove += length;
|
||||
return sequence - lowest <= FlowWindowSequences && bytesAbove + payloadLength <= FlowWindowBytes;
|
||||
}
|
||||
|
||||
private IReadOnlyList<CultNetRudpPacket> AdmitReliablePackets(
|
||||
IReadOnlyList<CultNetRudpPacket> packets,
|
||||
long nowMs)
|
||||
{
|
||||
lock (_pendingReliableGate)
|
||||
{
|
||||
var available = Math.Max(0, ReliableSendWindowPackets - _pendingReliable.Count);
|
||||
var ready = new List<CultNetRudpPacket>(Math.Min(available, packets.Count));
|
||||
var ready = new List<CultNetRudpPacket>();
|
||||
foreach (var packet in packets)
|
||||
{
|
||||
if (ready.Count < available)
|
||||
if (_queuedReliable.Count == 0 && WindowAdmits(packet.Sequence, packet.Payload.Length))
|
||||
{
|
||||
_pendingReliable[packet.Sequence] = new PendingReliablePacket
|
||||
{
|
||||
|
|
@ -1605,9 +1635,9 @@ namespace GameCult.Networking
|
|||
{
|
||||
lock (_pendingReliableGate)
|
||||
{
|
||||
var available = Math.Max(0, ReliableSendWindowPackets - _pendingReliable.Count);
|
||||
var ready = new List<CultNetRudpPacket>(Math.Min(available, _queuedReliable.Count));
|
||||
while (ready.Count < available && _queuedReliable.Count > 0)
|
||||
var ready = new List<CultNetRudpPacket>();
|
||||
while (_queuedReliable.Count > 0
|
||||
&& WindowAdmits(_queuedReliable.Peek().Sequence, _queuedReliable.Peek().Payload.Length))
|
||||
{
|
||||
var packet = _queuedReliable.Dequeue();
|
||||
_pendingReliable[packet.Sequence] = new PendingReliablePacket
|
||||
|
|
@ -1641,17 +1671,36 @@ namespace GameCult.Networking
|
|||
{
|
||||
lock (_pendingReliableGate)
|
||||
{
|
||||
_pendingReliable.Remove(packet.Ack);
|
||||
Acknowledge(packet.Ack);
|
||||
for (var bit = 0; bit < 32; bit++)
|
||||
{
|
||||
if ((packet.AckMask & (1u << bit)) != 0 && packet.Ack > bit)
|
||||
{
|
||||
_pendingReliable.Remove(packet.Ack - (uint)bit - 1);
|
||||
Acknowledge(packet.Ack - (uint)bit - 1);
|
||||
}
|
||||
}
|
||||
// Acknowledged sizes stop counting once the lowest unacknowledged sequence passes them, and all
|
||||
// of them stop when nothing is unacknowledged.
|
||||
if (_pendingReliable.Count == 0)
|
||||
{
|
||||
_ackedAboveLowest.Clear();
|
||||
}
|
||||
else
|
||||
{
|
||||
var lowest = _pendingReliable.Keys.Min();
|
||||
foreach (var sequence in _ackedAboveLowest.Keys.Where(sequence => sequence <= lowest).ToArray())
|
||||
_ackedAboveLowest.Remove(sequence);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Callers hold the pending gate.
|
||||
private void Acknowledge(uint sequence)
|
||||
{
|
||||
if (_pendingReliable.Remove(sequence, out var pending))
|
||||
_ackedAboveLowest[sequence] = pending.Packet.Payload.Length;
|
||||
}
|
||||
|
||||
private static bool PacketAcknowledges(CultNetRudpPacket packet, uint sequence)
|
||||
{
|
||||
if (packet.Ack == sequence)
|
||||
|
|
|
|||
|
|
@ -1141,6 +1141,196 @@ namespace GameCult.Networking.Tests
|
|||
Assert.That(delivered[0].Payload, Is.EqualTo(payload));
|
||||
}
|
||||
|
||||
// The sender's flow window: a reliable packet goes on the wire only while its sequence is at most 1,023
|
||||
// above the lowest unacked one and the payload above that sequence stays within 4 MiB.
|
||||
private const int Mib = 1024 * 1024;
|
||||
|
||||
private static CultNetRudpSession FlowSession(uint initialSequence) =>
|
||||
new CultNetRudpSession(new CultNetRudpSessionOptions
|
||||
{
|
||||
ConnectionId = 0x464c4f57,
|
||||
InitialSequence = initialSequence,
|
||||
ResendDelayMs = 25
|
||||
});
|
||||
|
||||
// A sender and a receiver that have shaken hands: the receiver's watermark is seeded by the sender's
|
||||
// Connect, and the sender's Connect is acknowledged, so nothing is pending on the sender.
|
||||
private static (CultNetRudpSession Sender, CultNetRudpSession Receiver) ConnectedFlowPair(uint senderInitialSequence, uint receiverInitialSequence)
|
||||
{
|
||||
var sender = FlowSession(senderInitialSequence);
|
||||
var receiver = FlowSession(receiverInitialSequence);
|
||||
sender.Receive(receiver.AcceptConnect(sender.CreateConnect(0), 0), 0);
|
||||
return (sender, receiver);
|
||||
}
|
||||
|
||||
private static CultNetRudpSession ConnectedFlowSession(uint initialSequence) => ConnectedFlowPair(initialSequence, 900).Sender;
|
||||
|
||||
private static IReadOnlyList<CultNetRudpPacket> FlowSend(CultNetRudpSession session, int payloadBytes) =>
|
||||
session.SendMany("state", new byte[payloadBytes], new CultNetRudpSendOptions { Reliable = true });
|
||||
|
||||
private static CultNetRudpPacket FlowAck(uint sequence) =>
|
||||
new CultNetRudpPacket { PacketType = CultNetRudpPacketType.Ack, ConnectionId = 0x464c4f57, Ack = sequence, ChannelId = "control" };
|
||||
|
||||
private static CultNetRudpPacket LoseGAndFillTheSpan(CultNetRudpSession session)
|
||||
{
|
||||
var g = FlowSend(session, 1).Single();
|
||||
for (uint offset = 1; offset <= 1023; offset++)
|
||||
{
|
||||
Assert.That(FlowSend(session, 1).Select(packet => packet.Sequence), Is.EqualTo(new[] { g.Sequence + offset }));
|
||||
session.Receive(FlowAck(g.Sequence + offset), 1);
|
||||
}
|
||||
return g;
|
||||
}
|
||||
|
||||
[Test]
|
||||
public void RudpSession_ALostPacketHoldsTheSender1023SequencesAheadAndIsStillDelivered()
|
||||
{
|
||||
var (sender, receiver) = ConnectedFlowPair(1, 100);
|
||||
var g = FlowSend(sender, 1).Single();
|
||||
var admittedAfterG = 0;
|
||||
for (var index = 0; index < 4200; index++)
|
||||
{
|
||||
foreach (var packet in FlowSend(sender, 1))
|
||||
{
|
||||
admittedAfterG++;
|
||||
receiver.Receive(packet, 1);
|
||||
sender.Receive(receiver.CreateAck(), 1);
|
||||
}
|
||||
}
|
||||
Assert.That(admittedAfterG, Is.EqualTo(1023));
|
||||
Assert.That(sender.QueuedReliablePacketCount, Is.EqualTo(4200 - 1023));
|
||||
|
||||
var retransmit = sender.DueResends(1000).Single(packet => packet.Sequence == g.Sequence);
|
||||
var delivered = receiver.Receive(retransmit, 1000).Delivered;
|
||||
Assert.That(delivered.Select(frame => frame.Sequence), Is.EqualTo(new[] { g.Sequence }));
|
||||
|
||||
var promoted = sender.Receive(receiver.CreateAckForReceived(g.Sequence), 1001).ReadyToSend;
|
||||
Assert.That(promoted.First().Sequence, Is.EqualTo(g.Sequence + 1024));
|
||||
}
|
||||
|
||||
[Test]
|
||||
public void RudpSession_ADirectSendPastTheSpanIsRefusedWithoutConsumingASequence()
|
||||
{
|
||||
var sender = ConnectedFlowSession(1);
|
||||
var g = LoseGAndFillTheSpan(sender);
|
||||
Assert.Throws<InvalidOperationException>(() => sender.Send("state", new byte[1], new CultNetRudpSendOptions { Reliable = true }));
|
||||
Assert.That(sender.QueuedReliablePacketCount, Is.Zero);
|
||||
sender.Receive(FlowAck(g.Sequence), 2);
|
||||
var next = sender.Send("state", new byte[1], new CultNetRudpSendOptions { Reliable = true });
|
||||
Assert.That(next.Sequence, Is.EqualTo(g.Sequence + 1024));
|
||||
}
|
||||
|
||||
[Test]
|
||||
public void RudpSession_PayloadAboveTheLowestUnackedSequenceIsBoundedBy4MiB()
|
||||
{
|
||||
var sender = ConnectedFlowSession(1);
|
||||
var g = FlowSend(sender, Mib).Single();
|
||||
// The lowest unacked packet's own bytes do not count: four more MiB fit above it.
|
||||
for (var index = 0; index < 4; index++)
|
||||
Assert.That(FlowSend(sender, Mib), Has.Count.EqualTo(1));
|
||||
Assert.That(FlowSend(sender, Mib), Is.Empty);
|
||||
Assert.That(FlowSend(sender, Mib), Is.Empty);
|
||||
Assert.That(sender.QueuedReliablePacketCount, Is.EqualTo(2));
|
||||
|
||||
// Acking g moves the floor up one packet: exactly one MiB more fits.
|
||||
var promoted = sender.Receive(FlowAck(g.Sequence), 1).ReadyToSend;
|
||||
Assert.That(promoted.Select(packet => packet.Sequence), Is.EqualTo(new[] { g.Sequence + 5 }));
|
||||
Assert.That(sender.QueuedReliablePacketCount, Is.EqualTo(1));
|
||||
}
|
||||
|
||||
[Test]
|
||||
public void RudpSession_APacketLargerThanTheWindowGoesOutAloneAndWaitsBehindAnythingUnacked()
|
||||
{
|
||||
Assert.That(FlowSend(ConnectedFlowSession(1), 5 * Mib), Has.Count.EqualTo(1));
|
||||
var sender = ConnectedFlowSession(1);
|
||||
var g = FlowSend(sender, 1).Single();
|
||||
Assert.That(FlowSend(sender, 5 * Mib), Is.Empty);
|
||||
var promoted = sender.Receive(FlowAck(g.Sequence), 1).ReadyToSend;
|
||||
Assert.That(promoted.Select(packet => packet.Sequence), Is.EqualTo(new[] { g.Sequence + 1 }));
|
||||
}
|
||||
|
||||
[Test]
|
||||
public void RudpSession_ASmallPacketDoesNotOvertakeAQueuedLargeOne()
|
||||
{
|
||||
var sender = ConnectedFlowSession(1);
|
||||
var g = FlowSend(sender, 1).Single();
|
||||
Assert.That(FlowSend(sender, 3 * Mib), Has.Count.EqualTo(1));
|
||||
Assert.That(FlowSend(sender, 2 * Mib), Is.Empty);
|
||||
// A direct send refuses instead of queueing behind it, and takes no sequence.
|
||||
Assert.Throws<InvalidOperationException>(() => sender.Send("state", new byte[1], new CultNetRudpSendOptions { Reliable = true }));
|
||||
Assert.That(sender.QueuedReliablePacketCount, Is.EqualTo(1));
|
||||
// One byte would fit above g, but the 2 MiB packet is ahead of it.
|
||||
Assert.That(FlowSend(sender, 1), Is.Empty);
|
||||
var promoted = sender.Receive(FlowAck(g.Sequence), 1).ReadyToSend;
|
||||
Assert.That(promoted.Select(packet => packet.Sequence), Is.EqualTo(new[] { g.Sequence + 2, g.Sequence + 3 }));
|
||||
}
|
||||
|
||||
[Test]
|
||||
public void RudpSession_AcknowledgedBytesAboveALostPacketStillCountAgainst4MiB()
|
||||
{
|
||||
var sender = ConnectedFlowSession(1);
|
||||
var g = FlowSend(sender, 1).Single();
|
||||
// g is lost. Each 1 MiB frame above it is delivered and acknowledged, and the receiver still holds it
|
||||
// behind the gap, so it keeps counting.
|
||||
for (uint offset = 1; offset <= 4; offset++)
|
||||
{
|
||||
Assert.That(FlowSend(sender, Mib).Select(packet => packet.Sequence), Is.EqualTo(new[] { g.Sequence + offset }));
|
||||
sender.Receive(FlowAck(g.Sequence + offset), 1);
|
||||
}
|
||||
Assert.That(FlowSend(sender, Mib), Is.Empty);
|
||||
Assert.That(sender.QueuedReliablePacketCount, Is.EqualTo(1));
|
||||
Assert.Throws<InvalidOperationException>(() => sender.Send("state", new byte[Mib], new CultNetRudpSendOptions { Reliable = true }));
|
||||
|
||||
// g arrives: nothing is held behind a gap any more, and the queue drains.
|
||||
var promoted = sender.Receive(FlowAck(g.Sequence), 2).ReadyToSend;
|
||||
Assert.That(promoted.Select(packet => packet.Sequence), Is.EqualTo(new[] { g.Sequence + 5 }));
|
||||
// The bytes above the old gap no longer count: another MiB fits.
|
||||
Assert.That(FlowSend(sender, Mib).Select(packet => packet.Sequence), Is.EqualTo(new[] { g.Sequence + 6 }));
|
||||
}
|
||||
|
||||
[Test]
|
||||
public void RudpSession_AcknowledgedBytesCountAgainstTheLowestUnackedSequenceAsItAdvances()
|
||||
{
|
||||
var sender = ConnectedFlowSession(1);
|
||||
var g = FlowSend(sender, 1).Single();
|
||||
var h = FlowSend(sender, 1).Single();
|
||||
for (var index = 0; index < 3; index++)
|
||||
FlowSend(sender, Mib);
|
||||
// g and the 3 MiB above h are acknowledged; h is lost.
|
||||
foreach (uint offset in new uint[] { 2, 3, 4, 0 })
|
||||
sender.Receive(FlowAck(g.Sequence + offset), 1);
|
||||
Assert.That(FlowSend(sender, Mib).Select(packet => packet.Sequence), Is.EqualTo(new[] { g.Sequence + 5 }));
|
||||
// 4 MiB sit above h, acknowledged or not: one more byte does not fit.
|
||||
Assert.That(FlowSend(sender, 1), Is.Empty);
|
||||
// h arrives, the floor moves to g + 5, and the byte fits.
|
||||
var promoted = sender.Receive(FlowAck(h.Sequence), 2).ReadyToSend;
|
||||
Assert.That(promoted.Select(packet => packet.Sequence), Is.EqualTo(new[] { g.Sequence + 6 }));
|
||||
}
|
||||
|
||||
// A server that accepts a fresh Connect may send before its Accept is acknowledged, so no acknowledgement of
|
||||
// the new generation has moved the floor yet. Bytes the ended generation had acknowledged above its own lost
|
||||
// packet belong to no receiver's hold any more and must not count.
|
||||
[Test]
|
||||
public void RudpSession_ANewGenerationOnTheServerDoesNotInheritAcknowledgedBytes()
|
||||
{
|
||||
var server = FlowSession(500);
|
||||
var accept = server.AcceptConnect(FlowSession(1).CreateConnect(0), 0);
|
||||
server.Receive(FlowAck(accept.Sequence), 0);
|
||||
var g = FlowSend(server, 1).Single();
|
||||
for (uint offset = 1; offset <= 4; offset++)
|
||||
{
|
||||
Assert.That(FlowSend(server, Mib).Select(packet => packet.Sequence), Is.EqualTo(new[] { g.Sequence + offset }));
|
||||
server.Receive(FlowAck(g.Sequence + offset), 1);
|
||||
}
|
||||
Assert.That(FlowSend(server, Mib), Is.Empty);
|
||||
|
||||
var nextAccept = server.AcceptConnect(FlowSession(200_000).CreateConnect(2), 2);
|
||||
Assert.That(nextAccept.PacketType, Is.EqualTo(CultNetRudpPacketType.Accept));
|
||||
// The Accept is unacknowledged and the only thing pending: 1 MiB fits above it.
|
||||
var sent = server.Send("state", new byte[Mib], new CultNetRudpSendOptions { Reliable = true });
|
||||
Assert.That(sent.Sequence, Is.EqualTo(nextAccept.Sequence + 1));
|
||||
}
|
||||
|
||||
[Test]
|
||||
public void RudpSocketTransport_HandshakesAndCarriesReliableOrderedSchemaFrames()
|
||||
{
|
||||
|
|
|
|||
Loading…
Reference in a new issue