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))