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)