Fix errors with multi-process workers on Windows

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.

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.