gh-146181: Spawn asyncio Unix subprocesses in a worker thread

Lib/asyncio/base_subprocess.py

@@ -14,7 +14,7 @@ class BaseSubprocessTransport(transports.SubprocessTransport):
 
     def __init__(self, loop, protocol, args, shell,
                  stdin, stdout, stderr, bufsize,
-                 waiter=None, extra=None, **kwargs):
+                 waiter=None, extra=None, _proc=None, **kwargs):
         super().__init__(extra)
         self._closed = False
         self._protocol = protocol
@@ -34,13 +34,19 @@ def __init__(self, loop, protocol, args, shell,
         if stderr == subprocess.PIPE:
             self._pipes[2] = None
 
-        # Create the child process: set the _proc attribute
-        try:
-            self._start(args=args, shell=shell, stdin=stdin, stdout=stdout,
-                        stderr=stderr, bufsize=bufsize, **kwargs)
-        except:
-            self.close()
-            raise
+        # Create the child process: set the _proc attribute.
+        # _proc lets a caller hand in an already-spawned Popen (e.g. one
+        # created off the event loop thread) instead of spawning it here.
+        if _proc is not None:
+            self._proc = _proc
+        else:
+            try:
+                self._start(args=args, shell=shell, stdin=stdin,
+                            stdout=stdout, stderr=stderr, bufsize=bufsize,
+                            **kwargs)
+            except:
+                self.close()
+                raise
 
         self._pid = self._proc.pid
         self._extra['subprocess'] = self._proc

Lib/asyncio/unix_events.py

@@ -1,6 +1,7 @@
 """Selector event loop for Unix with signal handling."""
 
 import errno
+import functools
 import io
 import itertools
 import os
@@ -201,10 +202,35 @@ async def _make_subprocess_transport(self, protocol, args, shell,
                                          extra=None, **kwargs):
         watcher = self._watcher
         waiter = self.create_future()
+
+        # subprocess.Popen() blocks the calling thread until the child
+        # has called execve(); spawn it in the default executor so a
+        # slow-to-start child (large binary, contended disk, etc.)
+        # doesn't stall the event loop. See gh-146181.
+        spawn_fut = self.run_in_executor(
+            None, functools.partial(_spawn_subprocess, args, shell, stdin,
+                                    stdout, stderr, bufsize, **kwargs))
+        try:
+            proc = await tasks.shield(spawn_fut)
+        except exceptions.CancelledError:
+            # The worker thread can't be interrupted -- Popen() keeps
+            # running to completion regardless of our cancellation.
+            # Wait for it, then kill and reap whatever it started, and
+            # close the pipe ends Popen opened for us, so a cancelled
+            # spawn never leaves an orphaned process or leaked fds
+            # behind (no transport exists yet to do this for us).
+            proc = await spawn_fut
+            proc.kill()
+            proc.wait()
+            for pipe in (proc.stdin, proc.stdout, proc.stderr):
+                if pipe is not None:
+                    pipe.close()
+            raise
+
         transp = _UnixSubprocessTransport(self, protocol, args, shell,
                                         stdin, stdout, stderr, bufsize,
                                         waiter=waiter, extra=extra,
-                                        **kwargs)
+                                        _proc=proc, **kwargs)
         watcher.add_child_handler(transp.get_pid(),
                                 self._child_watcher_callback, transp)
         try:
@@ -832,29 +858,41 @@ def _call_connection_lost(self, exc):
             self._loop = None
 
 
+def _spawn_subprocess(args, shell, stdin, stdout, stderr, bufsize, **kwargs):
+    # Split out of _UnixSubprocessTransport._start so it can also be
+    # called from a worker thread (see _UnixSelectorEventLoop.
+    # _make_subprocess_transport): subprocess.Popen() blocks until the
+    # child has called execve(), which can take a long time (page-table
+    # setup, image loading), and doing that on the event loop thread
+    # stalls every other task. See gh-146181.
+    stdin_w = None
+    if (stdin == subprocess.PIPE
+        and (sys.platform.startswith('aix') or sys.platform == 'cygwin')):
+        # Use a socket pair for stdin on AIX, since it does not
+        # support selecting read events on the write end of a
+        # socket (which we use in order to detect closing of the
+        # other end).
+        stdin, stdin_w = socket.socketpair()
+    try:
+        proc = subprocess.Popen(
+            args, shell=shell, stdin=stdin, stdout=stdout, stderr=stderr,
+            universal_newlines=False, bufsize=bufsize, **kwargs)
+        if stdin_w is not None:
+            stdin.close()
+            proc.stdin = open(stdin_w.detach(), 'wb', buffering=bufsize)
+            stdin_w = None
+    finally:
+        if stdin_w is not None:
+            stdin.close()
+            stdin_w.close()
+    return proc
+
+
 class _UnixSubprocessTransport(base_subprocess.BaseSubprocessTransport):
 
     def _start(self, args, shell, stdin, stdout, stderr, bufsize, **kwargs):
-        stdin_w = None
-        if (stdin == subprocess.PIPE
-            and (sys.platform.startswith('aix') or sys.platform == 'cygwin')):
-            # Use a socket pair for stdin on AIX, since it does not
-            # support selecting read events on the write end of a
-            # socket (which we use in order to detect closing of the
-            # other end).
-            stdin, stdin_w = socket.socketpair()
-        try:
-            self._proc = subprocess.Popen(
-                args, shell=shell, stdin=stdin, stdout=stdout, stderr=stderr,
-                universal_newlines=False, bufsize=bufsize, **kwargs)
-            if stdin_w is not None:
-                stdin.close()
-                self._proc.stdin = open(stdin_w.detach(), 'wb', buffering=bufsize)
-                stdin_w = None
-        finally:
-            if stdin_w is not None:
-                stdin.close()
-                stdin_w.close()
+        self._proc = _spawn_subprocess(
+            args, shell, stdin, stdout, stderr, bufsize, **kwargs)
 
 
 class _PidfdChildWatcher:

Lib/test/test_asyncio/test_subprocess.py

@@ -3,6 +3,7 @@
 import signal
 import sys
 import textwrap
+import threading
 import unittest
 import warnings
 from unittest import mock
@@ -1123,6 +1124,89 @@ async def run():
 
             self.loop.run_until_complete(run())
 
+        def test_subprocess_exec_does_not_block_loop(self):
+            # gh-146181: subprocess.Popen() is spawned in a worker thread
+            # so that a slow-to-start child (large binary, contended
+            # disk, etc.) doesn't stall every other task on the loop.
+            real_spawn = unix_events._spawn_subprocess
+            entered = threading.Event()
+            release = threading.Event()
+
+            def blocking_spawn(*args, **kwargs):
+                entered.set()
+                release.wait(10)
+                return real_spawn(*args, **kwargs)
+
+            async def main():
+                ticks = 0
+
+                async def ticker():
+                    nonlocal ticks
+                    while not release.is_set():
+                        ticks += 1
+                        await asyncio.sleep(0)
+
+                ticker_task = asyncio.ensure_future(ticker())
+                with mock.patch.object(unix_events, '_spawn_subprocess',
+                                       blocking_spawn):
+                    create_task = asyncio.ensure_future(
+                        asyncio.create_subprocess_exec(*PROGRAM_BLOCKED))
+                    # Wait until Popen() is actually running (and
+                    # blocked) in the worker thread.
+                    await asyncio.get_running_loop().run_in_executor(
+                        None, entered.wait, 10)
+                    # While the worker thread is stuck in Popen(), the
+                    # loop must still be able to run other coroutines.
+                    for _ in range(5):
+                        await asyncio.sleep(0)
+                    ticks_while_spawning = ticks
+                    release.set()
+                    proc = await create_task
+                await ticker_task
+                proc.kill()
+                await proc.wait()
+                return ticks_while_spawning
+
+            ticks_while_spawning = self.loop.run_until_complete(main())
+            self.assertGreater(ticks_while_spawning, 0)
+
+        def test_cancel_make_subprocess_transport_kills_process(self):
+            # gh-146181: cancelling subprocess creation while Popen() is
+            # still spawning in the worker thread (i.e. before a
+            # transport exists to own cleanup) must not leave an
+            # orphaned child process or leaked pipe fds behind.
+            real_spawn = unix_events._spawn_subprocess
+            procs = []
+
+            def capturing_spawn(*args, **kwargs):
+                proc = real_spawn(*args, **kwargs)
+                procs.append(proc)
+                return proc
+
+            async def main():
+                with mock.patch.object(unix_events, '_spawn_subprocess',
+                                       capturing_spawn):
+                    coro = self.loop.subprocess_exec(
+                        asyncio.SubprocessProtocol, *PROGRAM_BLOCKED)
+                    task = self.loop.create_task(coro)
+                    self.loop.call_soon(task.cancel)
+                    with self.assertRaises(asyncio.CancelledError):
+                        await task
+
+            with test_utils.disable_logger():
+                self.loop.run_until_complete(main())
+                test_utils.run_briefly(self.loop)
+
+            self.assertEqual(len(procs), 1)
+            proc = procs[0]
+            # The process must have been killed and reaped, not left
+            # running in the background.
+            self.assertIsNotNone(proc.poll())
+            # Its pipe ends must have been closed, not leaked.
+            for pipe in (proc.stdin, proc.stdout, proc.stderr):
+                if pipe is not None:
+                    self.assertTrue(pipe.closed)
+
 
     class SubprocessThreadedWatcherTests(SubprocessWatcherMixin,
                                          test_utils.TestCase):

Misc/NEWS.d/next/Library/2026-07-19-07-44-02.gh-issue-146181.k5Kjf9.rst

@@ -0,0 +1,5 @@
+On Unix, :mod:`asyncio` now spawns child processes (:func:`asyncio.create_subprocess_exec`,
+:func:`asyncio.create_subprocess_shell`, and :meth:`~asyncio.loop.subprocess_exec`) via
+:meth:`~asyncio.loop.run_in_executor` instead of calling :class:`subprocess.Popen`
+directly on the event loop thread, so a slow-to-start child process no longer
+blocks the event loop.