Repository: agronholm/anyio

Issue:
Sub-processes get SIGINT before cancellation when signal receiver is used
Maybe I'm doing something very silly here.  What I'm trying to do is catch SIGINT and schedule cancellation of tasks (some running sub-processes) later.

My example script below runs `bash -c "trap 'echo SIGINT CAUGHT' INT; sleep 100"` so that it will print `SIGINT CAUGHT` to stodut when it receives the signal and exits.  `open_process_callback_output()` is used to immediately print stdout from this process for debug sake.

```python
from contextlib import asynccontextmanager
import signal
import subprocess
import time

import anyio
from anyio.streams.text import TextReceiveStream

@asynccontextmanager
async def open_process_callback_output(cmd, on_stdout = None, on_stderr = None):
    """
    Open a process and if defined, call `on_stdout` for each line of stdout, and `on_stderr` for each line of stderr.
    """
    async def process_output(stream, func) -> None:
        async for text in TextReceiveStream(stream):
            for line in text.splitlines():
                func(line)

    proc = await anyio.open_process(cmd, stdout=PIPE, stderr=PIPE)
    async with anyio.create_task_group() as task_group:
        if on_stdout is not None:
            task_group.start_soon(process_output, proc.stdout, on_stdout)
        if on_stderr is not None:
            task_group.start_soon(process_output, proc.stderr, on_stderr)

        yield proc


async def task():
    async with open_process_callback_output('''bash -c "trap 'echo SIGINT CAUGHT' INT; sleep 100"''', on_stdout=lambda text: print(time.time(), f"\tstdout: {text}")) as proc:
        try:
            print(time.time(), "\ttask: waiting")
            await proc.wait()
        except anyio.get_cancelled_exc_class():
            print(time.time(), "\ttask: cancelled")
            raise
        finally:
            print(time.time(), "\ttask: done")


async def main():
    async def signal_handler():
        with anyio.open_signal_receiver(signal.SIGINT) as signals:
            async for signum in signals:
                print(time.time(), "\tsignal_handler: Got signal, cancel in 1 sec")
                await anyio.sleep(1)
                print(time.time(), "\tsignal_handler: cancelling")
                task_group.cancel_scope.cancel()

    async with anyio.create_task_group() as task_group:
        task_group.start_soon(task)
        task_group.start_soon(signal_handler)

print("asyncio")
anyio.run(main, backend='asyncio')
print("\ntrio")
anyio.run(main, backend='trio')
```

When I run this script and press CTRL-C after `task: waiting` is printed, I get:
```
asyncio
1639779714.8087683      task: waiting
^C1639779716.5815845    signal_handler: Got signal, cancel in 1 sec
1639779716.5836995      stdout: SIGINT CAUGHT
1639779716.5879638      task: done
1639779717.584897       signal_handler: cancelling

trio
1639779717.951388       task: waiting
^C1639779718.873392     stdout: SIGINT CAUGHT
1639779718.8739002      signal_handler: Got signal, cancel in 1 sec
1639779718.8772044      task: done
1639779719.8766265      signal_handler: cancelling
```

This shows that for both asyncio and trio, the process receives a signal before the `signal_handler` task and thus the bash script exits.  Soon after, the `task` coroutine exits (via `finally:`) without going through cancellation.

What I'd expect to happen is that SIGINT/CTRL-C would be caught by the signal handler and not immediately passed to the process.  Then, a second later the task would be cancelled, the process terminated, and so on.  For example:

```
asyncio
1639779714.8087683      task: waiting
^C1639779716.5815845    signal_handler: Got signal, cancel in 1 sec
1639779717.584897       signal_handler: cancelling
1639779717.5856995      stdout: SIGINT CAUGHT
1639779716.5879638      task: cancelled
1639779716.5879638      task: done

trio
1639779714.8087683      task: waiting
^C1639779716.5815845    signal_handler: Got signal, cancel in 1 sec
1639779717.584897       signal_handler: cancelling
1639779717.5856995      stdout: SIGINT CAUGHT
1639779716.5879638      task: cancelled
1639779716.5879638      task: done
```

Discussion:
uSpike: Looks like mypy released 0.920 and broke a number of things
agronholm: > Looks like mypy released 0.920 and broke a number of things

Yup, I'm aware and am looking into fixing these issues.
uSpike: OK, I'll remove my changes to fix those errors.
agronholm: The new test fails on Windows due to lack of `os.setsid()`. You need to skip the test on Windows.
uSpike: Whoops, fixed.

Patch:
diff --git a/docs/versionhistory.rst b/docs/versionhistory.rst
index df01fcf..afc4b92 100644
--- a/docs/versionhistory.rst
+++ b/docs/versionhistory.rst
@@ -5,6 +5,7 @@ This library adheres to `Semantic Versioning 2.0 <http://semver.org/>`_.
 
 **UNRELEASED**
 
+- Added ``start_new_session`` keyword argument to ``run_process()`` and ``open_process()``.
 - Fixed deadlock in synchronization primitives on asyncio which can happen if a task acquiring a
   primitive is hit with a native (not AnyIO) cancellation with just the right timing, leaving the
   next acquiring task waiting forever (`#398 <https://github.com/agronholm/anyio/issues/398>`_)
diff --git a/src/anyio/_backends/_asyncio.py b/src/anyio/_backends/_asyncio.py
index 4cb5f3b..f2929b4 100644
--- a/src/anyio/_backends/_asyncio.py
+++ b/src/anyio/_backends/_asyncio.py
@@ -946,16 +946,18 @@ class Process(abc.Process):
 async def open_process(command: Union[str, Sequence[str]], *, shell: bool,
                        stdin: int, stdout: int, stderr: int,
                        cwd: Union[str, bytes, PathLike, None] = None,
-                       env: Optional[Mapping[str, str]] = None) -> Process:
+                       env: Optional[Mapping[str, str]] = None,
+                       start_new_session: bool = False) -> Process:
     await checkpoint()
     if shell:
         process = await asyncio.create_subprocess_shell(
             command, stdin=stdin, stdout=stdout,  # type: ignore[arg-type]
-            stderr=stderr, cwd=cwd, env=env,
+            stderr=stderr, cwd=cwd, env=env, start_new_session=start_new_session,
         )
     else:
         process = await asyncio.create_subprocess_exec(*command, stdin=stdin, stdout=stdout,
-                                                       stderr=stderr, cwd=cwd, env=env)
+                                                       stderr=stderr, cwd=cwd, env=env,
+                                                       start_new_session=start_new_session)
 
     stdin_stream = StreamWriterWrapper(process.stdin) if process.stdin else None
     stdout_stream = StreamReaderWrapper(process.stdout) if process.stdout else None
diff --git a/src/anyio/_backends/_trio.py b/src/anyio/_backends/_trio.py
index 34c65a5..440e54f 100644
--- a/src/anyio/_backends/_trio.py
+++ b/src/anyio/_backends/_trio.py
@@ -318,9 +318,11 @@ class Process(abc.Process):
 async def open_process(command: Union[str, Sequence[str]], *, shell: bool,
                        stdin: int, stdout: int, stderr: int,
                        cwd: Union[str, bytes, PathLike, None] = None,
-                       env: Optional[Mapping[str, str]] = None) -> Process:
+                       env: Optional[Mapping[str, str]] = None,
+                       start_new_session: bool = False) -> Process:
     process = await trio_open_process(command, stdin=stdin, stdout=stdout, stderr=stderr,
-                                      shell=shell, cwd=cwd, env=env)
+                                      shell=shell, cwd=cwd, env=env,
+                                      start_new_session=start_new_session)
     stdin_stream = SendStreamWrapper(process.stdin) if process.stdin else None
     stdout_stream = ReceiveStreamWrapper(process.stdout) if process.stdout else None
     stderr_stream = ReceiveStreamWrapper(process.stderr) if process.stderr else None
diff --git a/src/anyio/_core/_subprocesses.py b/src/anyio/_core/_subprocesses.py
index 832184f..f4ae139 100644
--- a/src/anyio/_core/_subprocesses.py
+++ b/src/anyio/_core/_subprocesses.py
@@ -11,7 +11,8 @@ from ._tasks import create_task_group
 async def run_process(command: Union[str, Sequence[str]], *, input: Optional[bytes] = None,
                       stdout: int = PIPE, stderr: int = PIPE, check: bool = True,
                       cwd: Union[str, bytes, 'PathLike[str]', None] = None,
-                      env: Optional[Mapping[str, str]] = None) -> 'CompletedProcess[bytes]':
+                      env: Optional[Mapping[str, str]] = None, start_new_session: bool = False,
+                      ) -> 'CompletedProcess[bytes]':
     """
     Run an external command in a subprocess and wait until it completes.
 
@@ -28,6 +29,8 @@ async def run_process(command: Union[str, Sequence[str]], *, input: Optional[byt
     :param cwd: If not ``None``, change the working directory to this before running the command
     :param env: if not ``None``, this mapping replaces the inherited environment variables from the
         parent process
+    :param start_new_session: if ``true`` the setsid() system call will be made in the child
+        process prior to the execution of the subprocess. (POSIX only)
     :return: an object representing the completed process
     :raises ~subprocess.CalledProcessError: if ``check`` is ``True`` and the process exits with a
         nonzero return code
@@ -41,7 +44,8 @@ async def run_process(command: Union[str, Sequence[str]], *, input: Optional[byt
         stream_contents[index] = buffer.getvalue()
 
     async with await open_process(command, stdin=PIPE if input else DEVNULL, stdout=stdout,
-                                  stderr=stderr, cwd=cwd, env=env) as process:
+                                  stderr=stderr, cwd=cwd, env=env,
+                                  start_new_session=start_new_session) as process:
         stream_contents: List[Optional[bytes]] = [None, None]
         try:
             async with create_task_group() as tg:
@@ -68,7 +72,8 @@ async def run_process(command: Union[str, Sequence[str]], *, input: Optional[byt
 async def open_process(command: Union[str, Sequence[str]], *, stdin: int = PIPE,
                        stdout: int = PIPE, stderr: int = PIPE,
                        cwd: Union[str, bytes, 'PathLike[str]', None] = None,
-                       env: Optional[Mapping[str, str]] = None) -> Process:
+                       env: Optional[Mapping[str, str]] = None,
+                       start_new_session: bool = False) -> Process:
     """
     Start an external command in a subprocess.
 
@@ -83,9 +88,12 @@ async def open_process(command: Union[str, Sequence[str]], *, stdin: int = PIPE,
     :param cwd: If not ``None``, the working directory is changed before executing
     :param env: If env is not ``None``, it must be a mapping that defines the environment
         variables for the new process
+    :param start_new_session: if ``true`` the setsid() system call will be made in the child
+        process prior to the execution of the subprocess. (POSIX only)
     :return: an asynchronous process object
 
     """
     shell = isinstance(command, str)
     return await get_asynclib().open_process(command, shell=shell, stdin=stdin, stdout=stdout,
-                                             stderr=stderr, cwd=cwd, env=env)
+                                             stderr=stderr, cwd=cwd, env=env,
+                                             start_new_session=start_new_session)


Test patch:
diff --git a/tests/test_subprocesses.py b/tests/test_subprocesses.py
index 7402cbf..13b3936 100644
--- a/tests/test_subprocesses.py
+++ b/tests/test_subprocesses.py
@@ -85,3 +85,17 @@ async def test_process_env() -> None:
     cmd = [sys.executable, "-c", "import os; print(os.environ['foo'])"]
     result = await run_process(cmd, env=env)
     assert result.stdout.decode().strip() == env["foo"]
+
+
+@pytest.mark.skipif(platform.system() == 'Windows',
+                    reason='Windows does not have os.getsid()')
+async def test_process_new_session_sid() -> None:
+    """Test that start_new_session is successfully passed to the subprocess implementation"""
+    sid = os.getsid(os.getpid())
+    cmd = [sys.executable, "-c", "import os; print(os.getsid(os.getpid()))"]
+
+    result = await run_process(cmd)
+    assert result.stdout.decode().strip() == str(sid)
+
+    result = await run_process(cmd, start_new_session=True)
+    assert result.stdout.decode().strip() != str(sid)