fix: avoid importing asyncio on Windows subprocess platform (#3551)

newsfragments/3551.bugfix.rst

@@ -0,0 +1 @@
+Avoid importing ``asyncio`` on Windows by implementing ``windows_pipe`` directly with ``_winapi``, reducing startup latency.

src/trio/_subprocess_platform/__init__.py

@@ -78,6 +78,7 @@ def create_pipe_from_child_output() -> tuple[ClosableReceiveStream, int]:
 except ImportError as ex:  # pragma: no cover
     _wait_child_exiting_error = ex
 
+
 try:
     if TYPE_CHECKING:
         # Not worth type checking these definitions
@@ -96,15 +97,8 @@ def create_pipe_from_child_output() -> tuple[trio.lowlevel.FdStream, int]:
     elif os.name == "nt":
         import msvcrt
 
-        # This isn't exported or documented, but it's also not
-        # underscore-prefixed, and seems kosher to use. The asyncio docs
-        # for 3.5 included an example that imported socketpair from
-        # windows_utils (before socket.socketpair existed on Windows), and
-        # when asyncio.windows_utils.socketpair was removed in 3.7, the
-        # removal was mentioned in the release notes.
-        from asyncio.windows_utils import pipe as windows_pipe
-
         from .._windows_pipes import PipeReceiveStream, PipeSendStream
+        from .windows import windows_pipe
 
         def create_pipe_to_child_stdin() -> tuple[PipeSendStream, int]:
             # for stdin, we want the write end (our end) to use overlapped I/O

src/trio/_subprocess_platform/windows.py

@@ -1,11 +1,91 @@
+from __future__ import annotations
+
+import _winapi
+import itertools
+import os
+import tempfile
 from typing import TYPE_CHECKING
 
 from .._wait_for_object import WaitForSingleObject
 
 if TYPE_CHECKING:
     from .. import _subprocess
 
+_pipe_counter = itertools.count()
+
+
+def windows_pipe(
+    *,
+    duplex: bool = False,
+    overlapped: tuple[bool, bool] = (True, True),
+    bufsize: int = 8192,
+) -> tuple[int, int]:
+    """Like os.pipe() but with overlapped support and using handles not fds.
+
+    Adapted from CPython's asyncio.windows_utils.pipe().
+    Copyright (c) 2001-2026 Python Software Foundation; All Rights Reserved.
+    Licensed under the Python Software Foundation License Version 2.
+
+    Changes from CPython:
+    - Renamed pipe prefix from r'\\.\\pipe\asyncio-pipe-*' to r'\\.\\pipe\trio-pipe-*'.
+    - Modernized type hints for Trio's static typing and mypy checks.
+    - Explicit exception guard ensures early handle is closed if connection fails.
+    """
+    address = tempfile.mktemp(
+        prefix=rf"\\.\pipe\trio-pipe-{os.getpid():d}-{next(_pipe_counter):d}-"
+    )
+
+    if duplex:
+        openmode = _winapi.PIPE_ACCESS_DUPLEX
+        access = _winapi.GENERIC_READ | _winapi.GENERIC_WRITE
+        obsize, ibsize = bufsize, bufsize
+    else:
+        openmode = _winapi.PIPE_ACCESS_INBOUND
+        access = _winapi.GENERIC_WRITE
+        obsize, ibsize = 0, bufsize
+
+    openmode |= _winapi.FILE_FLAG_FIRST_PIPE_INSTANCE
+
+    if overlapped[0]:
+        openmode |= _winapi.FILE_FLAG_OVERLAPPED
+
+    flags_and_attribs = _winapi.FILE_FLAG_OVERLAPPED if overlapped[1] else 0
+
+    h1: int | None = None
+    h2: int | None = None
+    try:
+        h1 = _winapi.CreateNamedPipe(
+            address,
+            openmode,
+            _winapi.PIPE_WAIT,
+            1,
+            obsize,
+            ibsize,
+            _winapi.NMPWAIT_WAIT_FOREVER,
+            _winapi.NULL,
+        )
+
+        h2 = _winapi.CreateFile(
+            address,
+            access,
+            0,
+            _winapi.NULL,
+            _winapi.OPEN_EXISTING,
+            flags_and_attribs,
+            _winapi.NULL,
+        )
+
+        ov = _winapi.ConnectNamedPipe(h1, overlapped=True)
+        ov.GetOverlappedResult(True)
+        return h1, h2
+    except:
+        if h1 is not None:
+            _winapi.CloseHandle(h1)
+        if h2 is not None:
+            _winapi.CloseHandle(h2)
+        raise
+
 
-async def wait_child_exiting(process: "_subprocess.Process") -> None:
+async def wait_child_exiting(process: _subprocess.Process) -> None:
     # _handle is not in Popen stubs, though it is present on Windows.
     await WaitForSingleObject(int(process._proc._handle))  # type: ignore[attr-defined]

src/trio/_tests/test_trio.py

@@ -1,8 +1,25 @@
-def test_trio_import() -> None:
-    import sys
-
-    for module in list(sys.modules.keys()):
-        if module.startswith("trio"):
-            del sys.modules[module]
-
-    import trio  # noqa: F401
+from __future__ import annotations
+
+import subprocess
+import sys
+
+
+def test_trio_import() -> None:
+    for module in list(sys.modules.keys()):
+        if module.startswith("trio"):
+            del sys.modules[module]
+
+    import trio  # noqa: F401
+
+
+def test_trio_import_does_not_import_asyncio() -> None:
+    res = subprocess.run(
+        [
+            sys.executable,
+            "-c",
+            "import sys, trio; assert 'asyncio' not in sys.modules, f'unexpected asyncio modules: {[m for m in sys.modules if \"asyncio\" in m]}'",
+        ],
+        capture_output=True,
+        text=True,
+    )
+    assert res.returncode == 0, f"import trio loaded asyncio:\n{res.stderr}"