diff --git a/mypy/ipc.py b/mypy/ipc.py index 08ca0caf75f12..7bf130cd70dd9 100644 --- a/mypy/ipc.py +++ b/mypy/ipc.py @@ -65,7 +65,8 @@ class IPCBase: def __init__(self, name: str, timeout: float | None) -> None: self.name = name - self.timeout = timeout + self.timeout = timeout # Connections + self.io_timeout = timeout # Reads and writes self.message_size: int | None = None self.buffer = bytearray() @@ -101,7 +102,9 @@ def read_bytes(self, size: int = MAX_READ) -> bytes: ov, err = _winapi.ReadFile(self.connection, size, overlapped=True) try: if err == _winapi.ERROR_IO_PENDING: - timeout = int(self.timeout * 1000) if self.timeout else _winapi.INFINITE + timeout = ( + int(self.io_timeout * 1000) if self.io_timeout else _winapi.INFINITE + ) res = _winapi.WaitForSingleObject(ov.event, timeout) if res != _winapi.WAIT_OBJECT_0: raise IPCException(f"Bad result from I/O wait: {res}") @@ -158,7 +161,9 @@ def write_bytes(self, data: bytes) -> None: ov, err = _winapi.WriteFile(self.connection, encoded_data, overlapped=True) try: if err == _winapi.ERROR_IO_PENDING: - timeout = int(self.timeout * 1000) if self.timeout else _winapi.INFINITE + timeout = ( + int(self.io_timeout * 1000) if self.io_timeout else _winapi.INFINITE + ) res = _winapi.WaitForSingleObject(ov.event, timeout) if res != _winapi.WAIT_OBJECT_0: raise IPCException(f"Bad result from I/O wait: {res}") @@ -247,6 +252,11 @@ def __init__(self, name: str, timeout: float | None = None) -> None: else: name = f"{name}.sock" super().__init__(name, timeout) + # Unlike the client, a server applies its timeout only to accepting a + # connection, never to the traffic that follows: see __enter__ below. Once a + # peer is connected it may legitimately stay silent for a long time while it + # computes. + self.io_timeout = None if sys.platform == "win32": self.connection = _winapi.CreateNamedPipe( self.name, @@ -298,6 +308,10 @@ def __enter__(self) -> IPCServer: assert err == 0 else: try: + # Note self.timeout is set on the listening socket in __init__, but this + # applies to accept() only: the socket returned below is always blocking + # (see socket.accept()). This is why self.io_timeout is None -- it ensures + # equivalent behavior between Windows and POSIX. self.connection, _ = self.sock.accept() # This is already default on Linux, we set same buffer size # for macOS vs Linux consistency to simplify reasoning. diff --git a/mypy/test/testipc.py b/mypy/test/testipc.py index 0224035a7b61a..1f23c96bb2263 100644 --- a/mypy/test/testipc.py +++ b/mypy/test/testipc.py @@ -34,6 +34,20 @@ def server_multi_message_echo(q: Queue[str]) -> None: server.cleanup() +def server_with_short_timeout(msg: str, q: Queue[str]) -> None: + """Serve a client that takes longer to speak than the connection timeout. + + The timeout bounds accepting a connection, not the traffic after it, so the + read below must wait however long the client needs. + """ + server = IPCServer(CONNECTION_NAME, timeout=1) + q.put(server.connection_name) + with server: + data = server.read() + server.write(data + msg) + server.cleanup() + + class IPCTests(TestCase): def setUp(self) -> None: if sys.platform == "linux": @@ -98,6 +112,27 @@ def test_multiple_messages(self) -> None: p.join() assert p.exitcode == 0 + def test_server_timeout_does_not_apply_after_connecting(self) -> None: + # A server's timeout bounds accepting a connection only. A client that is + # merely slow to speak must not be mistaken for one that has hung up: that + # is what killed build workers waiting on the coordinator to load the graph + # (see #21484). POSIX gets this for free because accept() hands back a + # blocking socket, so this mainly guards the Windows path. + queue: Queue[str] = self.ctx.Queue() + msg = " -- echoed" + p = self.ctx.Process(target=server_with_short_timeout, args=(msg, queue), daemon=True) + p.start() + connection_name = queue.get() + with IPCClient(connection_name, timeout=1) as client: + # Stay quiet for well past the server's 1s connection timeout. + time.sleep(2.5) + client.write("hello") + assert client.read() == "hello" + msg + queue.close() + queue.join_thread() + p.join() + assert p.exitcode == 0 + # Run test_connect_twice a lot, in the hopes of finding issues. # This is really slow, so it is skipped, but can be enabled if # needed to debug IPC issues.