From 385123277fcf8dc9ac72ece48b504a75e12995bc Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 30 Sep 2026 00:00:32 +0200 Subject: [PATCH] Ack Cut 1: one watermark owns ordered delivery in all four runtimes 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 --- .../cultnet-py/src/cultnet_py/transport.py | 120 +++------ packages/cultnet-rs/src/rudp.rs | 246 ++++++------------ packages/cultnet-ts/src/rudp.ts | 139 ++++------ src/GameCult.Networking/CultNetTransport.cs | 151 ++++------- 4 files changed, 223 insertions(+), 433 deletions(-) diff --git a/packages/cultnet-py/src/cultnet_py/transport.py b/packages/cultnet-py/src/cultnet_py/transport.py index 0f378613..61e06da2 100644 --- a/packages/cultnet-py/src/cultnet_py/transport.py +++ b/packages/cultnet-py/src/cultnet_py/transport.py @@ -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 diff --git a/packages/cultnet-rs/src/rudp.rs b/packages/cultnet-rs/src/rudp.rs index 3b348fe3..9952dc8f 100644 --- a/packages/cultnet-rs/src/rudp.rs +++ b/packages/cultnet-rs/src/rudp.rs @@ -156,12 +156,6 @@ struct PendingReliablePacket { expires_at_ms: Option, } -#[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, queued_reliable: VecDeque<(CultNetRudpPacket, Option)>, reliable_packets_expired: u64, - ordered_next_sequence_by_channel: BTreeMap, - ordered_buffers: BTreeMap>, + /// 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, + /// 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, 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> { - 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> { + 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::() + > 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 { - 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 { diff --git a/packages/cultnet-ts/src/rudp.ts b/packages/cultnet-ts/src/rudp.ts index ae7ea838..e87c9dd3 100644 --- a/packages/cultnet-ts/src/rudp.ts +++ b/packages/cultnet-ts/src/rudp.ts @@ -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(); readonly #pendingReliable = new Map(); readonly #queuedReliable: CultNetRudpPacket[] = []; - readonly #orderedNextSequenceByChannel = new Map(); - readonly #orderedBuffers = new Map>(); + /// 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(); readonly #fragmentBuffers = new Map(); /// 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 { diff --git a/src/GameCult.Networking/CultNetTransport.cs b/src/GameCult.Networking/CultNetTransport.cs index e98fdc03..1a9db1ce 100644 --- a/src/GameCult.Networking/CultNetTransport.cs +++ b/src/GameCult.Networking/CultNetTransport.cs @@ -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 _pendingReliable = new Dictionary(); private readonly Queue _queuedReliable = new Queue(); - private readonly Dictionary _orderedNextSequenceByChannel = new Dictionary(StringComparer.Ordinal); - private readonly Dictionary> _orderedBuffers = - new Dictionary>(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 _orderedHeld = + new SortedDictionary(); private readonly Dictionary _fragmentBuffers = new Dictionary(StringComparer.Ordinal); /// @@ -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(); 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 }; } /// @@ -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 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(); - } - - if (frame.Sequence > next) - { - if (!_orderedBuffers.TryGetValue(frame.ChannelId, out var buffer)) - { - buffer = new SortedDictionary(); - _orderedBuffers[frame.ChannelId] = buffer; - } - - buffer[frame.Sequence] = new PendingOrderedFrame { Frame = frame, NextSequence = nextSequenceAfterFrame }; - return Array.Empty(); - } - - _orderedNextSequenceByChannel[frame.ChannelId] = nextSequenceAfterFrame; - return new[] { frame }.Concat(DrainOrdered(frame.ChannelId)).ToArray(); - } - - private IReadOnlyList DrainOrdered(string channelId) + /// + /// 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. + /// + private IReadOnlyList DrainOrdered() { var delivered = new List(); - 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;