From 06c61c749310f6ef403786db8c1a40508d75caa6 Mon Sep 17 00:00:00 2001 From: Vansh-Sharma27 Date: Tue, 30 Jun 2026 23:27:27 +0000 Subject: [PATCH] Add spawn_in_current_thread option to run_process/open_process run_process and open_process spawn the child by calling Popen() inside trio.to_thread.run_sync, which actually runs on a cached worker thread, not the thread that called run_process. Python-level state like context vars gets copied over fine, but OS-level per-thread state (network namespace from setns, capabilities, CPU affinity) does not, so the child can end up inheriting whatever the worker thread's state happens to be instead of the caller's. That's the bug in #3360. Added an opt-in spawn_in_current_thread=True kwarg that calls Popen directly on the calling thread instead of dispatching to a worker thread, so the child actually gets the calling thread's OS state. This briefly blocks the event loop while Popen runs, so it's off by default and existing behavior is unchanged. Added a regression test that uses sched_getaffinity/sched_setaffinity to actually reproduce the bug on Linux (skipped elsewhere since those calls aren't available on macOS/Windows), plus an equivalence test checking run_process/open_process behave the same with the flag on or off other than which thread spawns the child. --- newsfragments/3360.feature.rst | 8 ++ src/trio/_subprocess.py | 101 +++++++++++++++++++-- src/trio/_tests/test_subprocess.py | 71 +++++++++++++++ src/trio/_tests/type_tests/subprocesses.py | 4 + 4 files changed, 174 insertions(+), 10 deletions(-) create mode 100644 newsfragments/3360.feature.rst diff --git a/newsfragments/3360.feature.rst b/newsfragments/3360.feature.rst new file mode 100644 index 0000000000..ce635a235b --- /dev/null +++ b/newsfragments/3360.feature.rst @@ -0,0 +1,8 @@ +`trio.run_process` and `trio.lowlevel.open_process` now accept a +``spawn_in_current_thread`` keyword argument. By default the child process is +spawned from a worker thread in trio's internal thread pool, so OS-level +per-thread state on the calling thread (network namespace, capabilities, CPU +affinity, etc.) is not necessarily reflected in the child. Passing +``spawn_in_current_thread=True`` spawns the child directly from the calling +thread instead, at the cost of briefly blocking the event loop while +``subprocess.Popen`` runs. diff --git a/src/trio/_subprocess.py b/src/trio/_subprocess.py index d73ba3dc23..5330232df0 100644 --- a/src/trio/_subprocess.py +++ b/src/trio/_subprocess.py @@ -309,6 +309,7 @@ async def _open_process( stdin: int | HasFileno | None = None, stdout: int | HasFileno | None = None, stderr: int | HasFileno | None = None, + spawn_in_current_thread: bool = False, **options: object, ) -> Process: r"""Execute a child program in a new process. @@ -328,6 +329,16 @@ async def _open_process( management of the child process. It's up to you to implement whatever semantics you want. + .. note:: By default, the child process is spawned from a worker thread + drawn from Trio's internal thread pool, not from the thread that calls + `open_process`. Most Python-level state (e.g. context variables) is + copied across, but OS-level per-thread state is not: things like the + Linux network namespace (:manpage:`setns(2)`), capabilities, and CPU + affinity (:manpage:`pthreads(7)`) on the worker thread can differ from + what you set up on the calling thread, and the child inherits the + worker thread's state instead. Pass ``spawn_in_current_thread=True`` + if your child process needs to inherit this kind of state faithfully. + Args: command: The command to run. Typically this is a sequence of strings or bytes such as ``['ls', '-l', 'directory with spaces']``, where the @@ -351,6 +362,13 @@ async def _open_process( which causes the child's standard output and standard error messages to be intermixed on a single standard output stream, attached to whatever the ``stdout`` option says to attach it to. + spawn_in_current_thread: If true, spawn the child process directly on + the thread that calls `open_process`, instead of on a worker + thread. This makes the child inherit OS-level thread state (e.g. + a network namespace set with :manpage:`setns(2)`) from the calling + thread, at the cost of briefly blocking the calling thread (and, + if called from Trio's main thread, the whole event loop) while + `subprocess.Popen` does its work. **options: Other :ref:`general subprocess options ` are also accepted. @@ -414,16 +432,21 @@ async def _open_process( always_cleanup.callback(os.close, stderr) cleanup_on_fail.callback(trio_stderr.close) - popen = await trio.to_thread.run_sync( - partial( - subprocess.Popen, - command, - stdin=stdin, - stdout=stdout, - stderr=stderr, - **options, - ), + spawn_subprocess = partial( + subprocess.Popen, + command, + stdin=stdin, + stdout=stdout, + stderr=stderr, + **options, ) + if spawn_in_current_thread: + # Caller accepts brief event-loop blocking in exchange for + # spawning in *this* thread's OS context (namespaces, caps, + # affinity) instead of a worker thread's. + popen = spawn_subprocess() + else: + popen = await trio.to_thread.run_sync(spawn_subprocess) # We did not fail, so dismiss the stack for the trio ends cleanup_on_fail.pop_all() @@ -472,6 +495,7 @@ async def _run_process( check: bool = True, deliver_cancel: Callable[[Process], Awaitable[object]] | None = None, task_status: TaskStatus[Process] = trio.TASK_STATUS_IGNORED, + spawn_in_current_thread: bool = False, **options: object, ) -> subprocess.CompletedProcess[bytes]: """Run ``command`` in a subprocess and wait for it to complete. @@ -567,6 +591,16 @@ async def _run_process( To get the `subprocess.run` semantics, use ``check=False, stdin=None``. + .. note:: By default, the child process is spawned from a worker thread + drawn from Trio's internal thread pool, not from the thread that calls + `run_process`. Most Python-level state (e.g. context variables) is + copied across, but OS-level per-thread state is not: things like the + Linux network namespace (:manpage:`setns(2)`), capabilities, and CPU + affinity (:manpage:`pthreads(7)`) on the worker thread can differ from + what you set up on the calling thread, and the child inherits the + worker thread's state instead. Pass ``spawn_in_current_thread=True`` + if your child process needs to inherit this kind of state faithfully. + Args: command (list or str): The command to run. Typically this is a sequence of strings such as ``['ls', '-l', 'directory with spaces']``, @@ -633,6 +667,14 @@ async def my_deliver_cancel(process): In any case, `run_process` will always wait for the child process to exit before raising `Cancelled`. + spawn_in_current_thread: If true, spawn the child process directly on + the thread that calls `run_process`, instead of on a worker + thread. This makes the child inherit OS-level thread state (e.g. + a network namespace set with :manpage:`setns(2)`) from the calling + thread, at the cost of briefly blocking the calling thread (and, + if called from Trio's main thread, the whole event loop) while + `subprocess.Popen` does its work. + **options: :func:`run_process` also accepts any :ref:`general subprocess options ` and passes them on to the :class:`~trio.Process` constructor. This includes the @@ -736,7 +778,11 @@ async def read_output( # Opening the process does not need to be inside the nursery, so we put it outside # so any exceptions get directly seen by users. - proc = await _open_process(command, **options) # type: ignore[arg-type] + proc = await _open_process( + command, + spawn_in_current_thread=spawn_in_current_thread, + **options, # type: ignore[arg-type] + ) async with trio.open_nursery() as nursery: try: if input_ is not None: @@ -825,6 +871,7 @@ async def open_process( command: StrOrBytesPath | Sequence[StrOrBytesPath], *, stdin: int | HasFileno | None = None, + spawn_in_current_thread: bool = False, **kwargs: Unpack[WindowsProcessArgs], ) -> trio.Process: r"""Execute a child program in a new process. @@ -844,6 +891,16 @@ async def open_process( management of the child process. It's up to you to implement whatever semantics you want. + .. note:: By default, the child process is spawned from a worker thread + drawn from Trio's internal thread pool, not from the thread that calls + `open_process`. Most Python-level state (e.g. context variables) is + copied across, but OS-level per-thread state is not: things like + capabilities and CPU affinity (:manpage:`pthreads(7)`) on the worker + thread can differ from what you set up on the calling thread, and the + child inherits the worker thread's state instead. Pass + ``spawn_in_current_thread=True`` if your child process needs to + inherit this kind of state faithfully. + Args: command (list or str): The command to run. Typically this is a sequence of strings such as ``['ls', '-l', 'directory with spaces']``, @@ -866,6 +923,10 @@ async def open_process( which causes the child's standard output and standard error messages to be intermixed on a single standard output stream, attached to whatever the ``stdout`` option says to attach it to. + spawn_in_current_thread: If true, spawn the child process directly on + the thread that calls `open_process`, instead of on a worker + thread, at the cost of briefly blocking the calling thread while + `subprocess.Popen` does its work. **options: Other :ref:`general subprocess options ` are also accepted. @@ -888,6 +949,7 @@ async def run_process( capture_stderr: bool = False, check: bool = True, deliver_cancel: Callable[[Process], Awaitable[object]] | None = None, + spawn_in_current_thread: bool = False, **kwargs: Unpack[WindowsProcessArgs], ) -> subprocess.CompletedProcess[bytes]: """Run ``command`` in a subprocess and wait for it to complete. @@ -983,6 +1045,16 @@ async def run_process( To get the `subprocess.run` semantics, use ``check=False, stdin=None``. + .. note:: By default, the child process is spawned from a worker thread + drawn from Trio's internal thread pool, not from the thread that calls + `run_process`. Most Python-level state (e.g. context variables) is + copied across, but OS-level per-thread state is not: things like + capabilities and CPU affinity (:manpage:`pthreads(7)`) on the worker + thread can differ from what you set up on the calling thread, and the + child inherits the worker thread's state instead. Pass + ``spawn_in_current_thread=True`` if your child process needs to + inherit this kind of state faithfully. + Args: command (list or str): The command to run. Typically this is a sequence of strings such as ``['ls', '-l', 'directory with spaces']``, @@ -1049,6 +1121,11 @@ async def my_deliver_cancel(process): In any case, `run_process` will always wait for the child process to exit before raising `Cancelled`. + spawn_in_current_thread: If true, spawn the child process directly on + the thread that calls `run_process`, instead of on a worker + thread, at the cost of briefly blocking the calling thread while + `subprocess.Popen` does its work. + **options: :func:`run_process` also accepts any :ref:`general subprocess options ` and passes them on to the :class:`~trio.Process` constructor. This includes the @@ -1139,6 +1216,7 @@ async def open_process( *, stdin: int | HasFileno | None = None, shell: Literal[True], + spawn_in_current_thread: bool = False, **kwargs: Unpack[UnixProcessArgs], ) -> trio.Process: ... @@ -1148,6 +1226,7 @@ async def open_process( *, stdin: int | HasFileno | None = None, shell: bool = False, + spawn_in_current_thread: bool = False, **kwargs: Unpack[UnixProcessArgs], ) -> trio.Process: ... @@ -1157,6 +1236,7 @@ async def run_process( *, stdin: bytes | bytearray | memoryview | int | HasFileno | None = b"", shell: Literal[True], + spawn_in_current_thread: bool = False, **kwargs: Unpack[UnixRunProcessArgs], ) -> subprocess.CompletedProcess[bytes]: ... @@ -1166,6 +1246,7 @@ async def run_process( *, stdin: bytes | bytearray | memoryview | int | HasFileno | None = b"", shell: bool = False, + spawn_in_current_thread: bool = False, **kwargs: Unpack[UnixRunProcessArgs], ) -> subprocess.CompletedProcess[bytes]: ... diff --git a/src/trio/_tests/test_subprocess.py b/src/trio/_tests/test_subprocess.py index 890dd0d4e9..06706795c6 100644 --- a/src/trio/_tests/test_subprocess.py +++ b/src/trio/_tests/test_subprocess.py @@ -1,6 +1,7 @@ from __future__ import annotations import gc +import json import os import random import signal @@ -766,3 +767,73 @@ async def wait_and_tell(proc: Process) -> None: # for everything to notice await noticed_exit.wait() assert noticed_exit.is_set(), "child task wasn't woken after poll, DEADLOCK" + + +async def test_spawn_in_current_thread_equivalence() -> None: + # spawn_in_current_thread only changes which thread calls Popen(); + # it shouldn't change anything about the resulting subprocess. + for in_current_thread in (False, True): + result = await run_process( + CAT, + stdin=b"hello", + capture_stdout=True, + spawn_in_current_thread=in_current_thread, + ) + assert result.returncode == 0 + assert result.stdout == b"hello" + + proc = await open_process( + EXIT_TRUE, + spawn_in_current_thread=in_current_thread, + ) + try: + assert await proc.wait() == 0 + finally: + proc.kill() + await proc.wait() + + +# regression test for #3360 +@pytest.mark.skipif( + not hasattr(os, "sched_getaffinity"), + reason="sched_getaffinity/sched_setaffinity are Linux-only", +) +async def test_spawn_in_current_thread_affinity() -> None: + # By default, run_process spawns the child from a worker thread out of + # trio's thread cache, so a child can inherit stale OS-level thread state + # (here, CPU affinity) from whatever thread happens to be reused, instead + # of from the thread that actually called run_process. + # spawn_in_current_thread=True should avoid that by spawning directly on + # the calling thread. + original_mask = os.sched_getaffinity(0) # type: ignore[attr-defined,unused-ignore] + if len(original_mask) < 2: + pytest.skip("need at least 2 available CPUs to exercise this") + + # warm up trio's worker thread cache while affinity is unrestricted, so + # there's a cached worker thread whose affinity doesn't match what we're + # about to set below + await run_process(EXIT_TRUE) + + restricted_mask = {next(iter(original_mask))} + os.sched_setaffinity(0, restricted_mask) # type: ignore[attr-defined,unused-ignore] + try: + check_affinity = python( + "import json, os; print(json.dumps(sorted(os.sched_getaffinity(0))))", + ) + + default_result = await run_process(check_affinity, capture_stdout=True) + default_mask = set(json.loads(default_result.stdout)) + + current_thread_result = await run_process( + check_affinity, + capture_stdout=True, + spawn_in_current_thread=True, + ) + current_thread_mask = set(json.loads(current_thread_result.stdout)) + finally: + os.sched_setaffinity(0, original_mask) # type: ignore[attr-defined,unused-ignore] + + # the cached worker thread still has the old, unrestricted affinity + assert default_mask != restricted_mask + # but spawning from the calling thread picks up the restriction + assert current_thread_mask == restricted_mask diff --git a/src/trio/_tests/type_tests/subprocesses.py b/src/trio/_tests/type_tests/subprocesses.py index de3b5e8906..cbc8680f45 100644 --- a/src/trio/_tests/type_tests/subprocesses.py +++ b/src/trio/_tests/type_tests/subprocesses.py @@ -21,3 +21,7 @@ async def test() -> None: # 3.11+: await trio.run_process("python", process_group=5) # type: ignore await trio.lowlevel.open_process("python", process_group=5) # type: ignore + + # spawn_in_current_thread is accepted on every platform/overload shape + await trio.run_process("python", spawn_in_current_thread=True) + await trio.lowlevel.open_process("python", spawn_in_current_thread=True)