From a1aa79f6f47ab143871596ebad757b72d99dbc7d Mon Sep 17 00:00:00 2001 From: Dhruv Kumar Singh Date: Tue, 4 Aug 2026 14:02:35 +0530 Subject: [PATCH] Let a host connect to its own listener MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The stream registry keyed each end by (conn, host), so the two ends of a self-connection shared a key and the client overwrote the server: the connect never completed and the run deadlocked. Ends are now keyed by their own port as well, data packets address the peer's port, and the accept and refusal answer to the connector's port rather than the listener's — on one machine, port is the only thing telling the two directions apart. A connect whose drawn ephemeral port would equal the target port on the same host takes the next number instead. Packets between distinct hosts keep their keys, order and trace events, so recorded hashes stay where they were. --- CHANGELOG.md | 7 ++++++ src/simloop/_net.py | 43 ++++++++++++++++++++++---------- src/simloop/_transports.py | 7 +++++- tests/test_streams.py | 51 ++++++++++++++++++++++++++++++++++++++ 4 files changed, 94 insertions(+), 14 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 377f754..2b4c192 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -113,6 +113,13 @@ threads are real ones spawned through no loop API — a real thread racing a virtual clock ends in the cross-thread fence, a hang, or the caller's own timeout, whichever the race picks. +- A host can connect to its own listener. The stream registry used to key + each connection end by host alone, so the two ends of a self-connection + collapsed onto one entry and the run deadlocked; the key now carries the + end's own port, the connect handshake answers to the connector's port + rather than the listener's, and a loopback connect to a closed port is + refused instead of hanging. Packets between distinct hosts are keyed, + ordered and traced exactly as before, so existing hashes do not move. - `server.sockets` on a simulated server answers with an empty tuple instead of not existing, which is all aiohttp's `web.TCPSite` and websockets' `serve()` need to start; both now run their documented diff --git a/src/simloop/_net.py b/src/simloop/_net.py index 38641d0..cca06fc 100644 --- a/src/simloop/_net.py +++ b/src/simloop/_net.py @@ -413,8 +413,11 @@ def __init__(self, loop: SimLoop) -> None: self._held: list[_Packet] = [] self._datagrams: dict[tuple[str, int], _SimDatagramTransport] = {} self._listeners: dict[tuple[str, int], _Listener] = {} - self._streams: dict[tuple[int, str], _SimStreamTransport] = {} - self._inbound: dict[tuple[int, str], _InOrder] = {} + # Keyed by (conn, host, local port): a host can connect to its own + # listener, and without the port the two ends of that connection + # would collapse onto one key and overwrite each other. + self._streams: dict[tuple[int, str, int], _SimStreamTransport] = {} + self._inbound: dict[tuple[int, str, int], _InOrder] = {} self._pending: dict[int, _Connect] = {} self._next_conn = 0 self._next_uid = 0 @@ -730,7 +733,7 @@ def _deliver(self, packet: _Packet) -> None: _current_host.reset(token) def _dispatch_stream(self, packet: _Packet) -> None: - key = (packet.conn, packet.dst) + key = (packet.conn, packet.dst, packet.dst_port) queue = self._inbound.get(key) if queue is None: queue = self._inbound[key] = _InOrder(self) @@ -753,18 +756,18 @@ def _dispatch_ready(self, packet: _Packet) -> None: client = _SimStreamTransport( self, packet.conn, local=connect.local, remote=connect.remote ) - self._streams[(packet.conn, connect.local[0])] = client + self._streams[ + (packet.conn, connect.local[0], connect.local[1]) + ] = client protocol = connect.factory() client._begin(protocol) connect.fut.set_result((client, protocol)) else: connect.fut.set_exception( - ConnectionRefusedError( - f"connect to ({packet.src!r}, {packet.dst_port}) refused" - ) + ConnectionRefusedError(f"connect to {connect.remote!r} refused") ) return - transport = self._streams.get((packet.conn, packet.dst)) + transport = self._streams.get((packet.conn, packet.dst, packet.dst_port)) if transport is None: return # connection already torn down locally if packet.kind == "data": @@ -783,7 +786,8 @@ def _handle_syn(self, packet: _Packet) -> None: dst=packet.src, conn=packet.conn, seq=0, - dst_port=packet.dst_port, + src_port=packet.dst_port, + dst_port=packet.src_port, ) return transport = _SimStreamTransport( @@ -792,12 +796,21 @@ def _handle_syn(self, packet: _Packet) -> None: local=(packet.dst, packet.dst_port), remote=(packet.src, packet.src_port), ) - self._streams[(packet.conn, packet.dst)] = transport + self._streams[(packet.conn, packet.dst, packet.dst_port)] = transport # The accept is seq 0 of the server-to-client direction, so any data # the protocol writes from connection_made (seq 1+) can never arrive # ahead of the accept, whatever the latency draws say. + # The accept (and the refusal above) answers to the connector's own + # port: the answer belongs to the client end's inbound direction, and + # on a self-connection that direction is told apart by port alone. self._send_stream( - kind="accept", src=packet.dst, dst=packet.src, conn=packet.conn, seq=0 + kind="accept", + src=packet.dst, + dst=packet.src, + conn=packet.conn, + seq=0, + src_port=packet.dst_port, + dst_port=packet.src_port, ) protocol = listener.factory() transport._begin(protocol) @@ -828,8 +841,8 @@ def _send_stream( ) ) - def _drop_stream(self, conn: int, host: str) -> None: - self._streams.pop((conn, host), None) + def _drop_stream(self, conn: int, host: str, port: int) -> None: + self._streams.pop((conn, host, port), None) async def _open_connection( self, protocol_factory: Any, host: Any, port: Any @@ -842,6 +855,10 @@ async def _open_connection( conn = self._next_conn self._next_conn += 1 src_port = self._ephemeral() + if host == src and src_port == port: + # A self-connection whose ends share a port would collapse onto + # one registry key; the next ephemeral number cannot collide. + src_port = self._ephemeral() fut: asyncio.Future[tuple[_SimStreamTransport, Any]] = ( self._loop.create_future() ) diff --git a/src/simloop/_transports.py b/src/simloop/_transports.py index 8d13bef..e6cd418 100644 --- a/src/simloop/_transports.py +++ b/src/simloop/_transports.py @@ -211,6 +211,11 @@ def _send(self, kind: str, payload: bytes = b"") -> None: conn=self._conn, seq=seq, payload=payload, + # The ports address the peer's end of this connection: on a + # self-connection the two ends share a host and only the port + # tells the registry which one a packet is for. + src_port=self._local[1], + dst_port=self._remote[1], ) def write(self, data: Any) -> None: @@ -254,7 +259,7 @@ def _finish(self, exc: Exception | None) -> None: return self._closed = True self._closing = True - self._net._drop_stream(self._conn, self._local[0]) + self._net._drop_stream(self._conn, self._local[0], self._local[1]) if self._extra_socket is not None: self._extra_socket._dispose() protocol, self._protocol = self._protocol, None diff --git a/tests/test_streams.py b/tests/test_streams.py index 1845257..05025af 100644 --- a/tests/test_streams.py +++ b/tests/test_streams.py @@ -127,6 +127,57 @@ async def main() -> float: assert elapsed == pytest.approx(0.1) # syn there + refusal back +def test_a_host_can_connect_to_its_own_listener() -> None: + # Both ends of this connection live on one machine, so the stream + # registry must tell them apart by more than the host name — the + # client's ephemeral port against the listener's port is what does it. + loop = SimLoop(seed=0) + loop.net.host("solo") + + async def handle( + reader: asyncio.StreamReader, writer: asyncio.StreamWriter + ) -> None: + await reader.readline() + writer.write(b"pong\n") + await writer.drain() + writer.close() + + async def main() -> bytes: + server = await asyncio.start_server(handle, "0.0.0.0", 9000) + reader, writer = await asyncio.open_connection("solo", 9000) + writer.write(b"ping\n") + await writer.drain() + reply = await reader.readline() + writer.close() + await writer.wait_closed() + server.close() + return reply + + try: + reply = loop.run_until_complete(loop.net.host("solo").create_task(main())) + finally: + loop.close() + assert reply == b"pong\n" + assert not loop.net._streams # both ends existed, and both were torn down + + +def test_a_loopback_connect_to_a_closed_port_is_refused() -> None: + # The refusal answers to the connector's own port. Addressed to the + # listener's port instead, it would land on the queue the syn already + # advanced and wait there forever — a deadlock instead of an error. + loop = SimLoop(seed=0) + loop.net.host("solo") + + async def main() -> None: + with pytest.raises(ConnectionRefusedError): + await asyncio.open_connection("solo", 9999) + + try: + loop.run_until_complete(loop.net.host("solo").create_task(main())) + finally: + loop.close() + + def test_bytes_arrive_complete_and_in_order_under_latency_chaos() -> None: loop = _network(seed=5) loop.net.set_defaults(latency=(0.001, 0.2))