Add Windows CI for parallel checking (python#20803)

I am only running a subset of tests, just to make things faster.

Author: ilevkivskyi

.github/workflows/test.yml

@@ -112,6 +112,11 @@ jobs:
           toxenv: py
           tox_extra_args: "-n 4 --mypy-num-workers=4 mypy/test/testcheck.py"
           test_mypyc: true
+        - name: Parallel tests with py314-windows-64, interpreted
+          python: '3.14'
+          os: windows-latest
+          toxenv: py
+          tox_extra_args: "-n 2 --mypy-num-workers=2 mypy/test/testcheck.py -k 'incremental or modules or classes'"
 
         - name: Type check our own code (py310-ubuntu)
           python: '3.10'

mypy/build.py

@@ -100,7 +100,16 @@
 )
 from mypy.graph_utils import prepare_sccs, strongly_connected_components, topsort
 from mypy.indirection import TypeIndirectionVisitor
-from mypy.ipc import BadStatus, IPCClient, IPCMessage, read_status, ready_to_read, receive, send
+from mypy.ipc import (
+    BadStatus,
+    IPCClient,
+    IPCException,
+    IPCMessage,
+    read_status,
+    ready_to_read,
+    receive,
+    send,
+)
 from mypy.messages import MessageBuilder
 from mypy.nodes import (
     ClassDef,
@@ -280,9 +289,8 @@ def connect(self) -> None:
                 assert isinstance(pid, int), f"Bad PID: {pid}"
                 assert isinstance(connection_name, str), f"Bad connection name: {connection_name}"
                 if sys.platform != "win32":
-                    # TODO(emmatyping): for some reason this does not work on Windows. Probably
-                    # because we don't fork? We should check this
-                    # Double-check this status file is created by us.
+                    # Windows uses "wrapper processes" to run Python, so we cannot
+                    # verify PIDs reliably.
                     assert pid == self.proc.pid, f"PID mismatch: {pid} vs {self.proc.pid}"
                 self.conn = IPCClient(connection_name, WORKER_CONNECTION_TIMEOUT)
                 return
@@ -402,7 +410,7 @@ def default_flush_errors(
         for worker in workers:
             try:
                 send(worker.conn, SccRequestMessage(scc_id=None))
-            except OSError:
+            except (OSError, IPCException):
                 pass
         for worker in workers:
             worker.close()

mypy/build_worker/worker.py

@@ -41,7 +41,7 @@
 from mypy.defaults import RECURSION_LIMIT, WORKER_CONNECTION_TIMEOUT
 from mypy.errors import CompileError, Errors, report_internal_error
 from mypy.fscache import FileSystemCache
-from mypy.ipc import IPCServer, receive, send
+from mypy.ipc import IPCException, IPCServer, receive, send
 from mypy.modulefinder import BuildSource, BuildSourceSet, compute_search_paths
 from mypy.options import Options
 from mypy.util import read_py_file
@@ -82,9 +82,13 @@ def main(argv: list[str]) -> None:
     status_file = args.status_file
     server = IPCServer(CONNECTION_NAME, WORKER_CONNECTION_TIMEOUT)
 
-    with open(status_file, "w") as f:
-        json.dump({"pid": os.getpid(), "connection_name": server.connection_name}, f)
-        f.write("\n")
+    try:
+        with open(status_file, "w") as f:
+            json.dump({"pid": os.getpid(), "connection_name": server.connection_name}, f)
+            f.write("\n")
+    except Exception as exc:
+        print(f"Error writing status file {status_file}:", exc)
+        raise
 
     fscache = FileSystemCache()
     cached_read = fscache.read
@@ -94,7 +98,7 @@ def main(argv: list[str]) -> None:
     try:
         with server:
             serve(server, ctx)
-    except OSError as exc:
+    except (OSError, IPCException) as exc:
         if options.verbosity >= 1:
             print("Error communicating with coordinator:", exc)
     except Exception as exc:

mypy/ipc.py

@@ -41,6 +41,8 @@
 HEADER_SIZE = 4
 
 
+# TODO: we should make sure consistent exceptions are raised on different platforms.
+# Currently we raise either IPCException or OSError for equivalent conditions.
 class IPCException(Exception):
     """Exception for IPC issues."""
 
@@ -354,7 +356,7 @@ def read_status(status_file: str) -> dict[str, object]:
         try:
             data = json.load(f)
         except Exception as e:
-            raise BadStatus("Malformed status file (not JSON)") from e
+            raise BadStatus(f"Malformed status file: {str(e)}") from e
     if not isinstance(data, dict):
         raise BadStatus(f"Invalid status file (not a dict): {data}")
     return data
@@ -380,7 +382,9 @@ def ready_to_read(conns: list[IPCClient], timeout: float | None = None) -> list[
             try:
                 ov, err = _winapi.ReadFile(conn.connection, 1, overlapped=True)
             except OSError:
-                # Broken/closed pipe
+                # Broken/closed pipe. Mimic Linux behavior here, caller will get
+                # the exception when trying to read from this socket.
+                ready.append(i)
                 continue
             if err == _winapi.ERROR_IO_PENDING:
                 events.append(ov.event)