GameCult/CultLib
Watch
9
0
Fork
You've already forked CultLib
 0
Code Issues Pull requests 2 Projects Releases Packages Wiki Activity Actions 2

Ack Cut 1: one watermark owns ordered delivery in all four runtimes

Browse source 
The per-channel next pointer, its first-use initialisation, the skip over
other channels' sequences, and the per-channel drain are deleted. Each
session keeps one contiguous received-through watermark, seeded by the
peer's Connect or Accept (or the first packet of an unhandshaken session),
advanced in the received-sequence bookkeeping. Ordered frames wait in one
map keyed by first sequence, and every reliable receipt drains the frames at
or below watermark + 1 in sequence order, after the unordered frame of the
same call. A frame is held only while a sequence below it is missing, so a
5 MiB ordered frame with no gap is never held. Rust bounds what is still
held after the drain (frames, bytes, channels with held frames).

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-30 00:00:32 +02:00
commit 385123277f
4 changed files with 220 additions and 430 deletions
Download patch file Download diff file Expand all files Collapse all files

120
packages/cultnet-py/src/cultnet_py/transport.py
View file

@ -421,8 +421,15 @@ class CultNetRudpSession:
self._latest_sequenced_by_channel: dict[str, int] = {}
self._pending_reliable: dict[int, _PendingReliablePacket] = {}
self._queued_reliable: deque[CultNetRudpPacket] = deque()
self._ordered_next_sequence_by_channel: dict[str, int] = {}
self._ordered_buffers: dict[str, dict[int, tuple[CultNetRudpDeliveredFrame, int]]] = {}
# Every reliable sequence up to and including this one has been received
# since the peer state was last reset. The first reliable sequence after
# a reset seeds it: the peer's Connect or Accept, or, for a session that
# never handshook, its first packet.
self._received_through: int | None = None
# Ordered frames received but not yet deliverable, keyed by first
# sequence. A frame is held for exactly one reason: a reliable sequence
# below it has not arrived.
self._ordered_held: dict[int, CultNetRudpDeliveredFrame] = {}
self._fragment_buffers: dict[tuple[str, int], dict[str, Any]] = {}
# Matches cultnet-rs max_pending_fragment_sets. A set that loses one
# fragment is never completed; without a bound it is kept for the life
@ -465,8 +472,8 @@ class CultNetRudpSession:
self._latest_sequenced_by_channel.clear()
self._pending_reliable.clear()
self._queued_reliable.clear()
self._ordered_next_sequence_by_channel.clear()
self._ordered_buffers.clear()
self._received_through = None
self._ordered_held.clear()
self._fragment_buffers.clear()
self._fragment_sets_evicted = 0
@ -592,11 +599,6 @@ class CultNetRudpSession:
self._apply_acknowledgements(packet)
ready_to_send = self._promote_queued_reliable(now_ms)
self._last_received_at_ms = now_ms
expected_sequence_if_uninitialized = (
packet.sequence
if self._highest_received_sequence is None
else self._highest_received_sequence + 1
)
if packet.packet_type == CultNetRudpPacketType.ACCEPT:
self._remember_received(packet.sequence)
@ -639,25 +641,26 @@ class CultNetRudpSession:
if duplicate:
return CultNetRudpReceiveResult(ready_to_send=ready_to_send)
delivered: list[CultNetRudpDeliveredFrame] = []
reassembled = self._reassemble(packet)
if reassembled is None:
return CultNetRudpReceiveResult(ready_to_send=ready_to_send)
frame, ordered, next_sequence = reassembled
if not ordered and packet.sequenced:
newest_sequence = next_sequence - 1
latest_sequence = self._latest_sequenced_by_channel.get(frame.channel_id)
if latest_sequence is not None and newest_sequence <= latest_sequence:
return CultNetRudpReceiveResult(ready_to_send=ready_to_send)
self._latest_sequenced_by_channel[frame.channel_id] = newest_sequence
return CultNetRudpReceiveResult(delivered=(frame,), ready_to_send=ready_to_send)
if not ordered:
return CultNetRudpReceiveResult(delivered=(frame,), ready_to_send=ready_to_send)
return CultNetRudpReceiveResult(
delivered=tuple(
self._deliver_ordered(frame, next_sequence, expected_sequence_if_uninitialized)
),
ready_to_send=ready_to_send,
)
if reassembled is not None:
frame, ordered, next_sequence = reassembled
if ordered:
self._ordered_held[frame.sequence] = frame
elif not packet.sequenced:
delivered.append(frame)
else:
newest_sequence = next_sequence - 1
latest_sequence = self._latest_sequenced_by_channel.get(frame.channel_id)
if latest_sequence is None or newest_sequence > latest_sequence:
self._latest_sequenced_by_channel[frame.channel_id] = newest_sequence
delivered.append(frame)
# Any reliable packet may have advanced the watermark, a fragment or an
# unordered frame as much as an ordered one, so the drain runs after all
# of them.
delivered.extend(self._drain_ordered())
return CultNetRudpReceiveResult(delivered=tuple(delivered), ready_to_send=ready_to_send)
def create_ack(self) -> CultNetRudpPacket:
ack, ack_mask = self._ack_state()
@ -808,6 +811,10 @@ class CultNetRudpSession:
def _remember_received(self, sequence: int) -> None:
self._received_sequences.add(sequence)
through = sequence if self._received_through is None else self._received_through
while through + 1 in self._received_sequences:
through += 1
self._received_through = through
if self._highest_received_sequence is None or sequence > self._highest_received_sequence:
self._highest_received_sequence = sequence
if len(self._received_sequences) > self.RECEIVED_SEQUENCE_WINDOW:
@ -865,58 +872,15 @@ class CultNetRudpSession:
max(sequences) + 1,
)
def _deliver_ordered(
self,
frame: CultNetRudpDeliveredFrame,
next_after_frame: int,
expected_sequence_if_uninitialized: int,
) -> list[CultNetRudpDeliveredFrame]:
next_sequence = self._ordered_next_sequence_by_channel.get(frame.channel_id)
if next_sequence is None:
next_sequence = min(expected_sequence_if_uninitialized, frame.sequence)
self._ordered_next_sequence_by_channel[frame.channel_id] = next_sequence
while (
frame.sequence > next_sequence
and next_sequence in self._received_sequences
and next_sequence not in self._ordered_buffers.get(frame.channel_id, {})
):
next_sequence += 1
self._ordered_next_sequence_by_channel[frame.channel_id] = next_sequence
if frame.sequence < next_sequence:
def _drain_ordered(self) -> list[CultNetRudpDeliveredFrame]:
"""Delivers, in sequence order, every held ordered frame whose first
sequence is at most one past the watermark. Ordered delivery has one
owner: the watermark. A frame this call does not deliver waits for the
sequence below it, and nothing else releases it."""
if self._received_through is None:
return []
if frame.sequence > next_sequence:
self._ordered_buffers.setdefault(frame.channel_id, {})[frame.sequence] = (frame, next_after_frame)
return []
self._ordered_next_sequence_by_channel[frame.channel_id] = next_after_frame
return [frame, *self._drain_ordered(frame.channel_id)]
def _drain_ordered(self, channel_id: str) -> list[CultNetRudpDeliveredFrame]:
delivered: list[CultNetRudpDeliveredFrame] = []
buffer = self._ordered_buffers.get(channel_id)
if buffer is None:
return delivered
while True:
next_sequence = self._ordered_next_sequence_by_channel[channel_id]
pending = buffer.pop(next_sequence, None)
if pending is None:
break
frame, next_after_frame = pending
delivered.append(frame)
self._ordered_next_sequence_by_channel[channel_id] = next_after_frame
self._skip_received_non_channel_sequences(channel_id)
return delivered
def _skip_received_non_channel_sequences(self, channel_id: str) -> None:
next_sequence = self._ordered_next_sequence_by_channel.get(channel_id)
while (
next_sequence is not None
and next_sequence in self._received_sequences
and next_sequence not in self._ordered_buffers.get(channel_id, {})
):
next_sequence += 1
self._ordered_next_sequence_by_channel[channel_id] = next_sequence
releasable = sorted(sequence for sequence in self._ordered_held if sequence <= self._received_through + 1)
return [self._ordered_held.pop(sequence) for sequence in releasable]
def _allocate_fragment_id(self) -> int:
fragment_id = self._next_fragment_id

246
packages/cultnet-rs/src/rudp.rs
View file

@ -156,12 +156,6 @@ struct PendingReliablePacket {
expires_at_ms: Option<u64>,
}
#[derive(Clone, Debug, PartialEq, Eq)]
struct PendingOrderedFrame {
frame: CultNetRudpDeliveredFrame,
next_sequence: u32,
}
#[derive(Clone, Debug, PartialEq, Eq)]
struct FragmentBuffer {
channel_id: String,
@ -193,8 +187,15 @@ pub struct CultNetRudpSession {
pending_reliable: BTreeMap<u32, PendingReliablePacket>,
queued_reliable: VecDeque<(CultNetRudpPacket, Option<u64>)>,
reliable_packets_expired: u64,
ordered_next_sequence_by_channel: BTreeMap<String, u32>,
ordered_buffers: BTreeMap<String, BTreeMap<u32, PendingOrderedFrame>>,
/// Every reliable sequence up to and including this one has been received
/// since the peer state was last reset. The first reliable sequence after a
/// reset seeds it: the peer's Connect or Accept, or, for a session that
/// never handshook, its first packet.
received_through: Option<u32>,
/// Ordered frames received but not yet deliverable, keyed by first
/// sequence. A frame is held for exactly one reason: a reliable sequence
/// below it has not arrived.
ordered_held: BTreeMap<u32, CultNetRudpDeliveredFrame>,
fragment_buffers: BTreeMap<(String, u16), FragmentBuffer>,
fragment_touches: u64,
fragment_sets_evicted: u64,
@ -220,8 +221,8 @@ impl CultNetRudpSession {
pending_reliable: BTreeMap::new(),
queued_reliable: VecDeque::new(),
reliable_packets_expired: 0,
ordered_next_sequence_by_channel: BTreeMap::new(),
ordered_buffers: BTreeMap::new(),
received_through: None,
ordered_held: BTreeMap::new(),
fragment_buffers: BTreeMap::new(),
fragment_touches: 0,
fragment_sets_evicted: 0,
@ -351,8 +352,8 @@ impl CultNetRudpSession {
self.latest_sequenced_by_channel.clear();
self.pending_reliable.clear();
self.queued_reliable.clear();
self.ordered_next_sequence_by_channel.clear();
self.ordered_buffers.clear();
self.received_through = None;
self.ordered_held.clear();
self.fragment_buffers.clear();
}
@ -537,10 +538,6 @@ impl CultNetRudpSession {
self.purge_expired_reliable(now_ms);
let ready_to_send = self.promote_queued_reliable(now_ms);
self.last_received_at_ms = Some(now_ms);
let expected_sequence_if_uninitialized = self
.highest_received_sequence
.map(|sequence| sequence + 1)
.unwrap_or(packet.sequence);
if packet.packet_type == CultNetRudpPacketType::Accept {
self.remember_received(packet.sequence);
@ -664,42 +661,36 @@ impl CultNetRudpSession {
});
}
let Some((frame, ordered, next_sequence)) = self.reassemble(packet)? else {
return Ok(CultNetRudpReceiveResult {
delivered: Vec::new(),
ready_to_send,
reply: None,
pong: false,
pong_payload: Vec::new(),
disconnected: false,
disconnect_reason: Vec::new(),
});
};
let delivered = if ordered {
self.deliver_ordered(frame, next_sequence, expected_sequence_if_uninitialized)?
} else if packet.sequenced {
if !self
.latest_sequenced_by_channel
.contains_key(&frame.channel_id)
&& self.latest_sequenced_by_channel.len() >= RUDP_MAX_TRACKED_CHANNELS
{
return Err(anyhow!("RUDP session has too many sequenced channels"));
}
let latest = self
.latest_sequenced_by_channel
.get(&frame.channel_id)
.copied();
let newest_sequence = next_sequence.saturating_sub(1);
if latest.is_some_and(|latest| newest_sequence <= latest) {
Vec::new()
let mut delivered = Vec::new();
if let Some((frame, ordered, next_sequence)) = self.reassemble(packet)? {
if ordered {
self.ordered_held.insert(frame.sequence, frame);
} else if packet.sequenced {
if !self
.latest_sequenced_by_channel
.contains_key(&frame.channel_id)
&& self.latest_sequenced_by_channel.len() >= RUDP_MAX_TRACKED_CHANNELS
{
return Err(anyhow!("RUDP session has too many sequenced channels"));
}
let latest = self
.latest_sequenced_by_channel
.get(&frame.channel_id)
.copied();
let newest_sequence = next_sequence.saturating_sub(1);
if !latest.is_some_and(|latest| newest_sequence <= latest) {
self.latest_sequenced_by_channel
.insert(frame.channel_id.clone(), newest_sequence);
delivered.push(frame);
}
} else {
self.latest_sequenced_by_channel
.insert(frame.channel_id.clone(), newest_sequence);
vec![frame]
delivered.push(frame);
}
} else {
vec![frame]
};
}
// Any reliable packet may have advanced the watermark, a fragment or an
// unordered frame as much as an ordered one, so the drain runs after all
// of them.
delivered.extend(self.drain_ordered()?);
Ok(CultNetRudpReceiveResult {
delivered,
ready_to_send,
@ -972,6 +963,14 @@ impl CultNetRudpSession {
fn remember_received(&mut self, sequence: u32) {
self.received_sequences.insert(sequence);
let mut through = self.received_through.unwrap_or(sequence);
while let Some(next) = through.checked_add(1) {
if !self.received_sequences.contains(&next) {
break;
}
through = next;
}
self.received_through = Some(through);
if self
.highest_received_sequence
.is_none_or(|highest| sequence > highest)
@ -1123,126 +1122,39 @@ impl CultNetRudpSession {
Ok(())
}
fn ordered_buffered_frames(&self) -> usize {
self.ordered_buffers.values().map(BTreeMap::len).sum()
}
fn ordered_buffered_bytes(&self) -> usize {
self.ordered_buffers
.values()
.flat_map(BTreeMap::values)
.map(|pending| pending.frame.payload.len())
.sum()
}
fn deliver_ordered(
&mut self,
frame: CultNetRudpDeliveredFrame,
next_sequence_after_frame: u32,
expected_sequence_if_uninitialized: u32,
) -> Result<Vec<CultNetRudpDeliveredFrame>> {
let channel_id = frame.channel_id.clone();
if !self.ordered_next_sequence_by_channel.contains_key(&channel_id)
&& self.ordered_next_sequence_by_channel.len() >= RUDP_MAX_TRACKED_CHANNELS
/// Delivers, in sequence order, every held ordered frame whose first
/// sequence is at most one past the watermark, then bounds what is still
/// held. Ordered delivery has one owner: the watermark. A frame this call
/// does not deliver waits for the sequence below it, and nothing else
/// releases it.
fn drain_ordered(&mut self) -> Result<Vec<CultNetRudpDeliveredFrame>> {
let Some(through) = self.received_through else {
return Ok(Vec::new());
};
let still_held = match through.checked_add(1) {
Some(next) => self.ordered_held.split_off(&next.saturating_add(1)),
None => BTreeMap::new(),
};
let deliverable = std::mem::replace(&mut self.ordered_held, still_held);
if self.ordered_held.len() > RUDP_MAX_ORDERED_BUFFERED_FRAMES
|| self
.ordered_held
.values()
.map(|frame| frame.payload.len())
.sum::<usize>()
> RUDP_MAX_ORDERED_BUFFERED_BYTES
{
return Err(anyhow!("RUDP ordered hold buffer is full"));
}
let held_channels: BTreeSet<&str> = self
.ordered_held
.values()
.map(|frame| frame.channel_id.as_str())
.collect();
if held_channels.len() > RUDP_MAX_TRACKED_CHANNELS {
return Err(anyhow!("RUDP session has too many ordered channels"));
}
let mut next = if let Some(next) = self
.ordered_next_sequence_by_channel
.get(&channel_id)
.copied()
{
next
} else {
self.ordered_next_sequence_by_channel.insert(
channel_id.clone(),
expected_sequence_if_uninitialized.min(frame.sequence),
);
expected_sequence_if_uninitialized.min(frame.sequence)
};
while frame.sequence > next
&& self.received_sequences.contains(&next)
&& !self
.ordered_buffers
.get(&channel_id)
.is_some_and(|buffer| buffer.contains_key(&next))
{
next = next.saturating_add(1);
self.ordered_next_sequence_by_channel
.insert(channel_id.clone(), next);
}
if frame.sequence < next {
return Ok(Vec::new());
}
if frame.sequence > next {
if self.ordered_buffered_frames() >= RUDP_MAX_ORDERED_BUFFERED_FRAMES
|| self.ordered_buffered_bytes() + frame.payload.len()
> RUDP_MAX_ORDERED_BUFFERED_BYTES
{
return Err(anyhow!("RUDP ordered hold buffer is full"));
}
self.ordered_buffers.entry(channel_id).or_default().insert(
frame.sequence,
PendingOrderedFrame {
frame,
next_sequence: next_sequence_after_frame,
},
);
return Ok(Vec::new());
}
self.ordered_next_sequence_by_channel
.insert(channel_id.clone(), next_sequence_after_frame);
let mut delivered = vec![frame];
delivered.extend(self.drain_ordered(&channel_id));
Ok(delivered)
}
fn drain_ordered(&mut self, channel_id: &str) -> Vec<CultNetRudpDeliveredFrame> {
let mut delivered = Vec::new();
loop {
let Some(next) = self
.ordered_next_sequence_by_channel
.get(channel_id)
.copied()
else {
break;
};
let Some(buffer) = self.ordered_buffers.get_mut(channel_id) else {
break;
};
let Some(pending) = buffer.remove(&next) else {
break;
};
delivered.push(pending.frame);
self.ordered_next_sequence_by_channel
.insert(channel_id.to_string(), pending.next_sequence);
self.skip_received_non_channel_sequences(channel_id);
}
delivered
}
fn skip_received_non_channel_sequences(&mut self, channel_id: &str) {
let Some(mut next) = self
.ordered_next_sequence_by_channel
.get(channel_id)
.copied()
else {
return;
};
while self.received_sequences.contains(&next)
&& !self
.ordered_buffers
.get(channel_id)
.is_some_and(|buffer| buffer.contains_key(&next))
{
next = next.saturating_add(1);
self.ordered_next_sequence_by_channel
.insert(channel_id.to_string(), next);
}
Ok(deliverable.into_values().collect())
}
fn allocate_fragment_id(&mut self) -> u16 {

139
packages/cultnet-ts/src/rudp.ts
View file

@ -72,11 +72,6 @@ type PendingReliablePacket = {
lastSentAtMs: number;
};
type PendingOrderedFrame = {
frame: CultNetRudpDeliveredFrame;
nextSequence: number;
};
type FragmentBuffer = {
channelId: string;
reliable: boolean;
@ -157,8 +152,15 @@ export class CultNetRudpSession {
readonly #latestSequencedByChannel = new Map<string, number>();
readonly #pendingReliable = new Map<number, PendingReliablePacket>();
readonly #queuedReliable: CultNetRudpPacket[] = [];
readonly #orderedNextSequenceByChannel = new Map<string, number>();
readonly #orderedBuffers = new Map<string, Map<number, PendingOrderedFrame>>();
/// Every reliable sequence up to and including this one has been received
/// since the peer state was last reset. The first reliable sequence after a
/// reset seeds it: the peer's Connect or Accept, or, for a session that never
/// handshook, its first packet.
#receivedThrough: number | undefined;
/// Ordered frames received but not yet deliverable, keyed by first sequence.
/// A frame is held for exactly one reason: a reliable sequence below it has
/// not arrived.
readonly #orderedHeld = new Map<number, CultNetRudpDeliveredFrame>();
readonly #fragmentBuffers = new Map<string, FragmentBuffer>();
/// Matches cultnet-rs's `max_pending_fragment_sets`. A fragment set is only
/// removed on successful reassembly, so a set that loses one fragment is
@ -220,8 +222,8 @@ export class CultNetRudpSession {
this.#latestSequencedByChannel.clear();
this.#pendingReliable.clear();
this.#queuedReliable.splice(0);
this.#orderedNextSequenceByChannel.clear();
this.#orderedBuffers.clear();
this.#receivedThrough = undefined;
this.#orderedHeld.clear();
this.#fragmentBuffers.clear();
this.#fragmentSetsEvicted = 0;
}
@ -372,9 +374,6 @@ export class CultNetRudpSession {
this.#applyAcknowledgements(packet);
const readyToSend = this.#promoteQueuedReliable(nowMs);
this.#lastReceivedAtMs = nowMs;
const expectedSequenceIfUninitialized = this.#highestReceivedSequence === undefined
? packet.sequence
: this.#highestReceivedSequence + 1;
if (packet.packetType === "accept") {
this.#rememberReceived(packet.sequence);
@ -427,29 +426,28 @@ export class CultNetRudpSession {
return { delivered: [], readyToSend };
}
const delivered: CultNetRudpDeliveredFrame[] = [];
const reassembled = this.#reassemble(packet);
if (!reassembled) {
return { delivered: [], readyToSend };
}
if (!reassembled.ordered && (packet.sequenced ?? false)) {
const newestSequence = reassembled.nextSequence - 1;
const latestSequence = this.#latestSequencedByChannel.get(reassembled.frame.channelId);
if (latestSequence !== undefined && newestSequence <= latestSequence) {
return { delivered: [], readyToSend };
if (reassembled) {
if (reassembled.ordered) {
this.#orderedHeld.set(reassembled.frame.sequence, reassembled.frame);
} else if (!(packet.sequenced ?? false)) {
delivered.push(reassembled.frame);
} else {
const newestSequence = reassembled.nextSequence - 1;
const latestSequence = this.#latestSequencedByChannel.get(reassembled.frame.channelId);
if (latestSequence === undefined || newestSequence > latestSequence) {
this.#latestSequencedByChannel.set(reassembled.frame.channelId, newestSequence);
delivered.push(reassembled.frame);
}
}
this.#latestSequencedByChannel.set(reassembled.frame.channelId, newestSequence);
return { delivered: [reassembled.frame], readyToSend };
}
if (!reassembled.ordered) {
return { delivered: [reassembled.frame], readyToSend };
}
return {
delivered: this.#deliverOrdered(reassembled.frame, reassembled.nextSequence, expectedSequenceIfUninitialized),
readyToSend,
};
// Any reliable packet may have advanced the watermark, a fragment or an
// unordered frame as much as an ordered one, so the drain runs after all of
// them.
delivered.push(...this.#drainOrdered());
return { delivered, readyToSend };
}
createAck(): CultNetRudpPacket {
@ -618,6 +616,9 @@ export class CultNetRudpSession {
#rememberReceived(sequence: number): void {
this.#receivedSequences.add(sequence);
let through = this.#receivedThrough ?? sequence;
while (this.#receivedSequences.has(through + 1)) through += 1;
this.#receivedThrough = through;
if (this.#highestReceivedSequence === undefined || sequence > this.#highestReceivedSequence) {
this.#highestReceivedSequence = sequence;
}
@ -734,68 +735,20 @@ export class CultNetRudpSession {
};
}
#deliverOrdered(
frame: CultNetRudpDeliveredFrame,
nextSequence: number,
expectedSequenceIfUninitialized: number,
): CultNetRudpDeliveredFrame[] {
let next = this.#orderedNextSequenceByChannel.get(frame.channelId);
if (next === undefined) {
next = Math.min(expectedSequenceIfUninitialized, frame.sequence);
this.#orderedNextSequenceByChannel.set(frame.channelId, next);
}
while (frame.sequence > next && this.#receivedSequences.has(next) && !this.#orderedBuffers.get(frame.channelId)?.has(next)) {
next += 1;
this.#orderedNextSequenceByChannel.set(frame.channelId, next);
}
if (frame.sequence < next) {
return [];
}
if (frame.sequence > next) {
let buffer = this.#orderedBuffers.get(frame.channelId);
if (!buffer) {
buffer = new Map();
this.#orderedBuffers.set(frame.channelId, buffer);
}
buffer.set(frame.sequence, { frame, nextSequence });
return [];
}
this.#orderedNextSequenceByChannel.set(frame.channelId, nextSequence);
return [
frame,
...this.#drainOrdered(frame.channelId),
];
}
#drainOrdered(channelId: string): CultNetRudpDeliveredFrame[] {
const delivered: CultNetRudpDeliveredFrame[] = [];
const buffer = this.#orderedBuffers.get(channelId);
if (!buffer) {
return delivered;
}
let next = this.#orderedNextSequenceByChannel.get(channelId);
while (next !== undefined && buffer.has(next)) {
const pending = buffer.get(next)!;
buffer.delete(next);
delivered.push(pending.frame);
next = pending.nextSequence;
this.#orderedNextSequenceByChannel.set(channelId, next);
this.#skipReceivedNonChannelSequences(channelId);
}
return delivered;
}
#skipReceivedNonChannelSequences(channelId: string): void {
let next = this.#orderedNextSequenceByChannel.get(channelId);
while (next !== undefined && this.#receivedSequences.has(next) && !this.#orderedBuffers.get(channelId)?.has(next)) {
next += 1;
this.#orderedNextSequenceByChannel.set(channelId, next);
}
/// Delivers, in sequence order, every held ordered frame whose first sequence
/// is at most one past the watermark. Ordered delivery has one owner: the
/// watermark. A frame this call does not deliver waits for the sequence below
/// it, and nothing else releases it.
#drainOrdered(): CultNetRudpDeliveredFrame[] {
if (this.#receivedThrough === undefined) return [];
const releasable = [...this.#orderedHeld.keys()]
.filter((sequence) => sequence <= this.#receivedThrough! + 1)
.sort((left, right) => left - right);
return releasable.map((sequence) => {
const frame = this.#orderedHeld.get(sequence)!;
this.#orderedHeld.delete(sequence);
return frame;
});
}
#allocateFragmentId(): number {

151
src/GameCult.Networking/CultNetTransport.cs
View file

@ -888,12 +888,6 @@ namespace GameCult.Networking
public long LastSentAtMs { get; set; }
}
private sealed class PendingOrderedFrame
{
public CultNetRudpDeliveredFrame Frame { get; set; } = new CultNetRudpDeliveredFrame();
public uint NextSequence { get; set; }
}
private sealed class FragmentBuffer
{
public string ChannelId { get; set; } = string.Empty;
@ -915,9 +909,14 @@ namespace GameCult.Networking
private readonly object _pendingReliableGate = new object();
private readonly Dictionary<uint, PendingReliablePacket> _pendingReliable = new Dictionary<uint, PendingReliablePacket>();
private readonly Queue<CultNetRudpPacket> _queuedReliable = new Queue<CultNetRudpPacket>();
private readonly Dictionary<string, uint> _orderedNextSequenceByChannel = new Dictionary<string, uint>(StringComparer.Ordinal);
private readonly Dictionary<string, SortedDictionary<uint, PendingOrderedFrame>> _orderedBuffers =
new Dictionary<string, SortedDictionary<uint, PendingOrderedFrame>>(StringComparer.Ordinal);
// Every reliable sequence up to and including this one has been received since the peer state
// was last reset. The first reliable sequence after a reset seeds it: the peer's Connect or
// Accept, or, for a session that never handshook, its first packet.
private uint? _receivedThrough;
// Ordered frames received but not yet deliverable, keyed by first sequence. A frame is held for
// exactly one reason: a reliable sequence below it has not arrived.
private readonly SortedDictionary<uint, CultNetRudpDeliveredFrame> _orderedHeld =
new SortedDictionary<uint, CultNetRudpDeliveredFrame>();
private readonly Dictionary<string, FragmentBuffer> _fragmentBuffers = new Dictionary<string, FragmentBuffer>(StringComparer.Ordinal);
/// <summary>
@ -1148,9 +1147,6 @@ namespace GameCult.Networking
ApplyAcknowledgements(packet);
var readyToSend = PromoteQueuedReliable(nowMs);
_lastReceivedAtMs = nowMs;
var expectedSequenceIfUninitialized = _highestReceivedSequence.HasValue
? _highestReceivedSequence.Value + 1
: packet.Sequence;
if (packet.PacketType == CultNetRudpPacketType.Accept)
{
@ -1217,30 +1213,34 @@ namespace GameCult.Networking
return new CultNetRudpReceiveResult { ReadyToSend = readyToSend };
}
var delivered = new List<CultNetRudpDeliveredFrame>();
var reassembled = Reassemble(packet);
if (reassembled == null)
if (reassembled != null)
{
return new CultNetRudpReceiveResult { ReadyToSend = readyToSend };
}
if (!reassembled.Ordered && packet.Sequenced)
{
var newestSequence = reassembled.NextSequence - 1;
if (_latestSequencedByChannel.TryGetValue(reassembled.Frame.ChannelId, out var latestSequence)
&& newestSequence <= latestSequence)
if (reassembled.Ordered)
{
return new CultNetRudpReceiveResult { ReadyToSend = readyToSend };
_orderedHeld[reassembled.Frame.Sequence] = reassembled.Frame;
}
else if (!packet.Sequenced)
{
delivered.Add(reassembled.Frame);
}
else
{
var newestSequence = reassembled.NextSequence - 1;
if (!_latestSequencedByChannel.TryGetValue(reassembled.Frame.ChannelId, out var latestSequence)
|| newestSequence > latestSequence)
{
_latestSequencedByChannel[reassembled.Frame.ChannelId] = newestSequence;
delivered.Add(reassembled.Frame);
}
}
_latestSequencedByChannel[reassembled.Frame.ChannelId] = newestSequence;
}
return new CultNetRudpReceiveResult
{
ReadyToSend = readyToSend,
Delivered = reassembled.Ordered
? DeliverOrdered(reassembled.Frame, reassembled.NextSequence, expectedSequenceIfUninitialized)
: new[] { reassembled.Frame }
};
// Any reliable packet may have advanced the watermark, a fragment or an unordered frame as
// much as an ordered one, so the drain runs after all of them.
delivered.AddRange(DrainOrdered());
return new CultNetRudpReceiveResult { ReadyToSend = readyToSend, Delivered = delivered };
}
/// <summary>
@ -1302,8 +1302,8 @@ namespace GameCult.Networking
_pendingReliable.Clear();
_queuedReliable.Clear();
}
_orderedNextSequenceByChannel.Clear();
_orderedBuffers.Clear();
_receivedThrough = null;
_orderedHeld.Clear();
_fragmentBuffers.Clear();
}
@ -1518,6 +1518,12 @@ namespace GameCult.Networking
private void RememberReceived(uint sequence)
{
_receivedSequences.Add(sequence);
var through = _receivedThrough ?? sequence;
while (through != uint.MaxValue && _receivedSequences.Contains(through + 1))
{
through++;
}
_receivedThrough = through;
if (!_highestReceivedSequence.HasValue || sequence > _highestReceivedSequence.Value)
{
_highestReceivedSequence = sequence;
@ -1625,80 +1631,35 @@ namespace GameCult.Networking
};
}
private IReadOnlyList<CultNetRudpDeliveredFrame> DeliverOrdered(
CultNetRudpDeliveredFrame frame,
uint nextSequenceAfterFrame,
uint expectedSequenceIfUninitialized)
{
if (!_orderedNextSequenceByChannel.TryGetValue(frame.ChannelId, out var next))
{
next = Math.Min(expectedSequenceIfUninitialized, frame.Sequence);
_orderedNextSequenceByChannel[frame.ChannelId] = next;
}
while (frame.Sequence > next
&& _receivedSequences.Contains(next)
&& (!_orderedBuffers.TryGetValue(frame.ChannelId, out var pendingBuffer) || !pendingBuffer.ContainsKey(next)))
{
next++;
_orderedNextSequenceByChannel[frame.ChannelId] = next;
}
if (frame.Sequence < next)
{
return Array.Empty<CultNetRudpDeliveredFrame>();
}
if (frame.Sequence > next)
{
if (!_orderedBuffers.TryGetValue(frame.ChannelId, out var buffer))
{
buffer = new SortedDictionary<uint, PendingOrderedFrame>();
_orderedBuffers[frame.ChannelId] = buffer;
}
buffer[frame.Sequence] = new PendingOrderedFrame { Frame = frame, NextSequence = nextSequenceAfterFrame };
return Array.Empty<CultNetRudpDeliveredFrame>();
}
_orderedNextSequenceByChannel[frame.ChannelId] = nextSequenceAfterFrame;
return new[] { frame }.Concat(DrainOrdered(frame.ChannelId)).ToArray();
}
private IReadOnlyList<CultNetRudpDeliveredFrame> DrainOrdered(string channelId)
/// <summary>
/// Delivers, in sequence order, every held ordered frame whose first sequence is at most one past
/// the watermark. Ordered delivery has one owner: the watermark. A frame this call does not
/// deliver waits for the sequence below it, and nothing else releases it.
/// </summary>
private IReadOnlyList<CultNetRudpDeliveredFrame> DrainOrdered()
{
var delivered = new List<CultNetRudpDeliveredFrame>();
if (!_orderedBuffers.TryGetValue(channelId, out var buffer))
if (!_receivedThrough.HasValue)
{
return delivered;
}
while (_orderedNextSequenceByChannel.TryGetValue(channelId, out var next) && buffer.TryGetValue(next, out var pending))
var through = _receivedThrough.Value;
foreach (var pair in _orderedHeld)
{
buffer.Remove(next);
delivered.Add(pending.Frame);
_orderedNextSequenceByChannel[channelId] = pending.NextSequence;
SkipReceivedNonChannelSequences(channelId);
if (through != uint.MaxValue && pair.Key > through + 1)
{
break;
}
delivered.Add(pair.Value);
}
foreach (var frame in delivered)
{
_orderedHeld.Remove(frame.Sequence);
}
return delivered;
}
private void SkipReceivedNonChannelSequences(string channelId)
{
if (!_orderedNextSequenceByChannel.TryGetValue(channelId, out var next))
{
return;
}
while (_receivedSequences.Contains(next)
&& (!_orderedBuffers.TryGetValue(channelId, out var buffer) || !buffer.ContainsKey(next)))
{
next++;
_orderedNextSequenceByChannel[channelId] = next;
}
}
private ushort AllocateFragmentId()
{
var fragmentId = _nextFragmentId;