Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 24 additions & 0 deletions packages/pytcp/pytcp/protocols/tcp/session/tcp__session.py
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,7 @@

from __future__ import annotations

import collections
import errno
import os
import threading
Expand Down Expand Up @@ -561,6 +562,12 @@ def __init__(
# Out of order packet buffer.
self._ooo_packet_queue: dict[int, TcpMetadata] = {}

# Segments released from '_ooo_packet_queue' by a gap-fill,
# waiting for the outermost 'tcp_fsm' call to dispatch them
# one after another ('_ooo_draining' marks that loop).
self._ooo_ready: collections.deque[TcpMetadata] = collections.deque()
self._ooo_draining: bool = False

# Used to report cause of connection failure.
self._connection_error: ConnError = ConnError.NONE

Expand Down Expand Up @@ -2286,3 +2293,20 @@ def tcp_fsm(
tcp_fsm_dispatch_timer(self)

self._pump_tail(state_at_entry, packet_rx_md is not None or syscall is not None)

# RFC 9293 §3.10.7.4: segments held for later processing
# are processed once RCV.NXT reaches them. A gap-fill
# releases them one at a time into '_ooo_ready'; process
# them here in a loop rather than by re-entering the FSM
# from inside the dispatch that released them, which
# nested one dispatch per queued segment and raised
# RecursionError on a long OOO run. The nested
# 'tcp_fsm' calls below see '_ooo_draining' and leave
# the next released segment to this loop.
if self._ooo_ready and not self._ooo_draining:
self._ooo_draining = True
try:
while self._ooo_ready:
self.tcp_fsm(packet_rx_md=self._ooo_ready.popleft())
finally:
self._ooo_draining = False
Original file line number Diff line number Diff line change
Expand Up @@ -823,10 +823,13 @@ def _phase5_consume_segment_and_postprocess(self, packet_rx_md: TcpMetadata) ->
"tcp-ss",
f"[{session}] - Purged expired TX packet retransmit request counter for {seq}",
)
# Bring next packet from ooo_packet_queue if available.
# Bring next packet from ooo_packet_queue if available. The
# outermost 'tcp_fsm' call processes it once this dispatch
# has finished, so a long OOO run does not nest one FSM
# dispatch per queued segment.
if ooo_packet := session._ooo_packet_queue.pop(session._rcv_seq.nxt, None):
__debug__ and log(
"tcp-ss",
f"[{session}] - <lg>Retrieving packet {session._rcv_seq.nxt} from Out of Order queue</>",
)
session.tcp_fsm(ooo_packet)
session._ooo_ready.append(ooo_packet)
Original file line number Diff line number Diff line change
Expand Up @@ -597,3 +597,56 @@ def test__data_transfer_out_of_order__five_ooo_segments_emit_at_least_three_dup_
FsmState.ESTABLISHED,
msg="OOO-burst handling must not transition the session out of ESTABLISHED.",
)

def test__data_transfer_out_of_order__long_queue_drains_without_recursion(self) -> None:
"""
Ensure that a gap-fill in front of a long run of
queued out-of-order segments delivers the whole run
in order without nesting one FSM dispatch per queued
segment. Draining by re-entering the FSM from inside
the dispatch that is still processing the previous
segment exhausts the interpreter stack (RecursionError)
once the run is a few hundred segments long, which
kills the packet-handler thread.

Reference: RFC 9293 §3.10.7.4 (segments with higher sequence numbers held for later processing).
"""

session = self._drive_handshake_to_established(iss=LOCAL__ISS, peer_iss=PEER__ISS)

segment_count = 300
payload_size = 10

def _build(index: int) -> bytes:
"""Helper: build the peer's 'index'-th data segment (0-based)."""
return build_tcp4(
sport=PEER__PORT,
dport=STACK__PORT,
seq=PEER__ISS + 1 + index * payload_size,
ack=LOCAL__ISS + 1,
flags=("ACK",),
payload=bytes([index % 256]) * payload_size,
win=PEER__WIN,
)

for index in range(1, segment_count):
self._drive_rx(frame=_build(index))
self.assertEqual(
len(session._ooo_packet_queue),
segment_count - 1,
msg="Setup precondition: every segment after the gap must be held in the OOO queue.",
)

self._drive_rx(frame=_build(0))

self.assertEqual(
bytes(session._rx_buffer),
b"".join(bytes([index % 256]) * payload_size for index in range(segment_count)),
msg="The gap-fill must deliver all queued segments to '_rx_buffer' in sequence order.",
)
self.assertEqual(
session._rcv_seq.nxt,
PEER__ISS + 1 + segment_count * payload_size,
msg="RCV.NXT must cover every drained segment.",
)
self.assertEqual(len(session._ooo_packet_queue), 0, msg="The OOO queue must be empty after the drain.")