Add spawn_in_current_thread option to run_process/open_process

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.

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 <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 <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 <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 <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]: ...
 

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

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)