Repository: joblib/loky

Issue:
DeprecationWarning: This process (pid=...) is multi-threaded, use of fork() may lead to deadlocks in the child.
Here is a pure-loky reproducer for the issue reported on the joblib repo under https://github.com/joblib/joblib/issues/1650:

```python
from _testcapi import _spawn_pthread_waiter, _end_spawned_pthread
from math import sqrt
import warnings
import loky


_spawn_pthread_waiter()

e = loky.get_reusable_executor()

try:
    warnings.simplefilter("always", category=DeprecationWarning)
    for each in e.map(sqrt, range(10)):
        print(each)

finally:
    _end_spawned_pthread()
```

Loky does use `os.fork` followed by `os.exec`:

https://github.com/joblib/loky/blob/2bfa386d336540ffe495afbfe433ddd4129db34c/loky/backend/fork_exec.py#L33-L43

While the posix `SpawnContext`'s `_Popen` implementation uses `spawnv_passfds`:

https://github.com/python/cpython/blob/9e474a98af4184615540467dea16da05f4d284d8/Lib/multiprocessing/util.py#L439-L450

Discussion:
ogrisel: Since we dropped Python 3.7 support, there is probably a lot of compat code that can be simplified in the resource tracker and the posix popen modules (and maybe the win32 module as well).
ogrisel: Some of the last commits above can be reverted to use #435 instead.
ogrisel: This should be ready for review @jeremiedbb @tomMoral @lesteve @ryandvmartin @sam-s.
ogrisel: The `test_parallel_call_cached_function_defined_in_jupyter` failure (part of the joblib test suite) is unrelated. I will open a PR in joblib to report more info in case of failure.
ogrisel: @tomMoral this is ready for review/merge.

Patch:
diff --git a/CHANGES.md b/CHANGES.md
index 6c46399..6e46295 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -1,5 +1,11 @@
 ### 3.5.0 - in development
 
+- Avoid raising `DeprecationWarning` related to `os.fork` when running in a
+  natively multi-threaded process. (#429).
+
+- Fix a crash when calling commands that access `stdin` via `subprocess.run` in
+  worker processes on POSIX systems. (#429).
+
 - Automatically call `faulthandler.enable()` when starting loky worker
   processes to report more informative information (post-mortem Python
   tracebacks in particular) on worker crashs. (#419).
diff --git a/continuous_integration/install_coverage_subprocess_pth.py b/continuous_integration/install_coverage_subprocess_pth.py
index 578286e..cf84652 100644
--- a/continuous_integration/install_coverage_subprocess_pth.py
+++ b/continuous_integration/install_coverage_subprocess_pth.py
@@ -3,13 +3,15 @@
 # http://coverage.readthedocs.io/en/latest/subprocess.html
 
 import os.path as op
-from distutils.sysconfig import get_python_lib
+from sysconfig import get_path
 
 FILE_CONTENT = """\
 import coverage; coverage.process_startup()
 """
 
-filename = op.join(get_python_lib(), "coverage_subprocess.pth")
+# get_path("platlib") returns the location of the `site-packages` folder:
+# https://docs.python.org/3/library/sysconfig.html
+filename = op.join(get_path("platlib"), "coverage_subprocess.pth")
 with open(filename, mode="w") as f:
     f.write(FILE_CONTENT)
 
diff --git a/loky/backend/fork_exec.py b/loky/backend/fork_exec.py
index 2353c42..e43a3a4 100644
--- a/loky/backend/fork_exec.py
+++ b/loky/backend/fork_exec.py
@@ -4,40 +4,65 @@
 #
 # author: Thomas Moreau and Olivier Grisel
 #
-import os
 import sys
+import os
+import subprocess
 
 
-def close_fds(keep_fds):  # pragma: no cover
-    """Close all the file descriptors except those in keep_fds."""
-
-    # Make sure to keep stdout and stderr open for logging purpose
-    keep_fds = {*keep_fds, 1, 2}
-
-    # We try to retrieve all the open fds
-    try:
-        open_fds = {int(fd) for fd in os.listdir("/proc/self/fd")}
-    except FileNotFoundError:
-        import resource
-
-        max_nfds = resource.getrlimit(resource.RLIMIT_NOFILE)[0]
-        open_fds = {*range(max_nfds)}
-
-    for i in open_fds - keep_fds:
-        try:
-            os.close(i)
-        except OSError:
-            pass
+def fork_exec(cmd, keep_fds, env=None):
+    import _posixsubprocess
 
+    # Encoded command args as bytes:
+    cmd = [os.fsencode(arg) for arg in cmd]
 
-def fork_exec(cmd, keep_fds, env=None):
-    # copy the environment variables to set in the child process
+    # Copy the environment variables to set in the child process (also encoded
+    # as bytes).
     env = env or {}
-    child_env = {**os.environ, **env}
+    env = {**os.environ, **env}
+    encoded_env = []
+    for key, value in env.items():
+        encoded_env.append(os.fsencode(f"{key}={value}"))
 
-    pid = os.fork()
-    if pid == 0:  # pragma: no cover
-        close_fds(keep_fds)
-        os.execve(sys.executable, cmd, child_env)
+    # Fds with fileno larger than 3 (stdin=0, stdout=1, stderr=2) are be closed
+    # in the child process, except for those passed in keep_fds.
+    keep_fds = tuple(sorted(map(int, keep_fds)))
+    errpipe_read, errpipe_write = os.pipe()
+
+    # VFORK is not supported on older Python versions.
+    if hasattr(subprocess, "_USE_VFORK"):
+        # Python 3.11 and later
+        pgid_to_set = [-1]
+        allow_vfork = [subprocess._USE_VFORK]
     else:
-        return pid
+        pgid_to_set = []
+        allow_vfork = []
+
+    try:
+        return _posixsubprocess.fork_exec(
+            cmd,  # args
+            cmd[0:1],  # executable_list
+            True,  # close_fds
+            keep_fds,  # pass_fds
+            None,  # cwd
+            encoded_env,  # env
+            -1,  # p2cread
+            -1,  # p2cwrite
+            -1,  # c2pread
+            -1,  # c2pwrite
+            -1,  # errread
+            -1,  # errwrite
+            errpipe_read,  # errpipe_read
+            errpipe_write,  # errpipe_write
+            False,  # restore_signal
+            False,  # call_setsid
+            *pgid_to_set,  # pgid_to_set
+            None,  # gid
+            None,  # extra_groups
+            None,  # uid
+            -1,  # child_umask
+            None,  # preexec_fn
+            *allow_vfork,  # extra flag if vfork is available
+        )
+    finally:
+        os.close(errpipe_read)
+        os.close(errpipe_write)
diff --git a/loky/backend/resource_tracker.py b/loky/backend/resource_tracker.py
index 25204a7..6af66de 100644
--- a/loky/backend/resource_tracker.py
+++ b/loky/backend/resource_tracker.py
@@ -4,16 +4,11 @@
 #
 # author: Thomas Moreau
 #
-# adapted from multiprocessing/semaphore_tracker.py  (17/02/2017)
-#  * include custom spawnv_passfds to start the process
-#  * add some VERBOSE logging
-#
-# TODO: multiprocessing.resource_tracker was contributed to Python 3.8 so
-# once loky drops support for Python 3.7 it might be possible to stop
-# maintaining this loky-specific fork. As a consequence, it might also be
-# possible to stop maintaining the loky.backend.synchronize fork of
-# multiprocessing.synchronize.
-
+# Adapted from multiprocessing/resource_tracker.py
+#  * add some VERBOSE logging,
+#  * add support to track folders,
+#  * add Windows support,
+#  * refcounting scheme to avoid unlinking resources still in use.
 #
 # On Unix we run a server process which keeps track of unlinked
 # resources. The server ignores SIGINT and SIGTERM and reads from a
@@ -40,7 +35,7 @@
 # Note that this behavior differs from CPython's resource_tracker, which only
 # implements list of shared resources, and not a proper refcounting scheme.
 # Also, CPython's resource tracker will only attempt to cleanup those shared
-# resources once all procsses connected to the resouce tracker have exited.
+# resources once all processes connected to the resource tracker have exited.
 
 
 import os
@@ -48,9 +43,11 @@ import shutil
 import sys
 import signal
 import warnings
-import threading
 from _multiprocessing import sem_unlink
 from multiprocessing import util
+from multiprocessing.resource_tracker import (
+    ResourceTracker as _ResourceTracker,
+)
 
 from . import spawn
 
@@ -74,15 +71,25 @@ if os.name == "posix":
 VERBOSE = False
 
 
-class ResourceTracker:
-    def __init__(self):
-        self._lock = threading.Lock()
-        self._fd = None
-        self._pid = None
+class ResourceTracker(_ResourceTracker):
+    """Resource tracker with refcounting scheme.
+
+    This class is an extension of the multiprocessing ResourceTracker class
+    which implements a reference counting scheme to avoid unlinking shared
+    resources still in use in other processes.
 
-    def getfd(self):
+    This feature is notably used by `joblib.Parallel` to share temporary
+    folders and memory mapped files between the main process and the worker
+    processes.
+
+    The actual implementation of the refcounting scheme is in the main
+    function, which is run in a dedicated process.
+    """
+
+    def maybe_unlink(self, name, rtype):
+        """Decrement the refcount of a resource, and delete it if it hits 0"""
         self.ensure_running()
-        return self._fd
+        self._send("MAYBE_UNLINK", name, rtype)
 
     def ensure_running(self):
         """Make sure that resource tracker process is running.
@@ -164,39 +171,6 @@ class ResourceTracker:
                 else:
                     os.close(r)
 
-    def _check_alive(self):
-        """Check for the existence of the resource tracker process."""
-        try:
-            self._send("PROBE", "", "")
-        except BrokenPipeError:
-            return False
-        else:
-            return True
-
-    def register(self, name, rtype):
-        """Register a named resource, and increment its refcount."""
-        self.ensure_running()
-        self._send("REGISTER", name, rtype)
-
-    def unregister(self, name, rtype):
-        """Unregister a named resource with resource tracker."""
-        self.ensure_running()
-        self._send("UNREGISTER", name, rtype)
-
-    def maybe_unlink(self, name, rtype):
-        """Decrement the refcount of a resource, and delete it if it hits 0"""
-        self.ensure_running()
-        self._send("MAYBE_UNLINK", name, rtype)
-
-    def _send(self, cmd, name, rtype):
-        if len(name) > 512:
-            # posix guarantees that writes to a pipe of less than PIPE_BUF
-            # bytes are atomic, and that PIPE_BUF >= 512
-            raise ValueError("name too long")
-        msg = f"{cmd}:{name}:{rtype}\n".encode("ascii")
-        nbytes = os.write(self._fd, msg)
-        assert nbytes == len(msg)
-
 
 _resource_tracker = ResourceTracker()
 ensure_running = _resource_tracker.ensure_running
@@ -348,25 +322,13 @@ def main(fd, verbose=0):
         util.debug("resource tracker shut down")
 
 
-#
-# Start a program with only specified fds kept open
-#
-
-
 def spawnv_passfds(path, args, passfds):
-    passfds = sorted(passfds)
     if sys.platform != "win32":
-        errpipe_read, errpipe_write = os.pipe()
-        try:
-            from .reduction import _mk_inheritable
-            from .fork_exec import fork_exec
-
-            _pass = [_mk_inheritable(fd) for fd in passfds]
-            return fork_exec(args, _pass)
-        finally:
-            os.close(errpipe_read)
-            os.close(errpipe_write)
+        args = [arg.encode("utf-8") for arg in args]
+        path = path.encode("utf-8")
+        return util.spawnv_passfds(path, args, passfds)
     else:
+        passfds = sorted(passfds)
         cmd = " ".join(f'"{x}"' for x in args)
         try:
             _, ht, pid, _ = _winapi.CreateProcess(
diff --git a/setup.py b/setup.py
index a4028bb..2acc403 100644
--- a/setup.py
+++ b/setup.py
@@ -78,6 +78,8 @@ setup(
         "Programming Language :: Python :: 3.9",
         "Programming Language :: Python :: 3.10",
         "Programming Language :: Python :: 3.11",
+        "Programming Language :: Python :: 3.12",
+        "Programming Language :: Python :: 3.13",
         "Topic :: Scientific/Engineering",
         "Topic :: Utilities",
         "Topic :: Software Development :: Libraries",


Test patch:
diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
index 04a38ef..4c9c217 100755
--- a/.github/workflows/test.yml
+++ b/.github/workflows/test.yml
@@ -46,24 +46,24 @@ jobs:
       matrix:
         include:
 
-          - name: windows-py311
+          - name: windows-py313
             os: windows-latest
-            PYTHON_VERSION: "3.11"
+            PYTHON_VERSION: "3.13"
           - name: windows-py39
             os: windows-latest
             PYTHON_VERSION: "3.9"
             LOKY_TEST_NO_CIM: "true"
 
-          - name: macos-py311
+          - name: macos-py313
             os: macos-latest
-            PYTHON_VERSION: "3.11"
+            PYTHON_VERSION: "3.13"
           - name: macos-py39
             os: macos-latest
             PYTHON_VERSION: "3.9"
 
-          - name: linux-py311
+          - name: linux-py313
             os: ubuntu-latest
-            PYTHON_VERSION: "3.11"
+            PYTHON_VERSION: "3.13"
             LOKY_TEST_NO_LSCPU: "true"
           - name: linux-py39-joblib-tests
             os: ubuntu-latest
@@ -112,7 +112,7 @@ jobs:
 
       - name: Upload to Codecov
         # always upload coverage even if tests fail
-        if: ${{ matrix.SKLEARN_TESTS != 'true' && (success() || failure()) }}
+        if: ${{ matrix.JOBLIB_TESTS != 'true' && (success() || failure()) }}
         uses: codecov/codecov-action@v5
         with:
           files: coverage.xml
diff --git a/tests/_test_process_executor.py b/tests/_test_process_executor.py
index 6588965..87031de 100644
--- a/tests/_test_process_executor.py
+++ b/tests/_test_process_executor.py
@@ -9,7 +9,6 @@ import platform
 import pytest
 import weakref
 import tempfile
-import traceback
 import threading
 import faulthandler
 import warnings
@@ -1195,6 +1194,32 @@ class ExecutorTest:
 
         executor.shutdown(wait=True)
 
+    def test_no_deprecation_warning_is_raised_on_fork(self):
+        # On POSIX, recent versions of Python detect if the process has native
+        # threads running when calling `os.fork` and similar. All supported
+        # process executors should not trigger this warning by not calling
+        # `os.fork` (and `os.execve`) manually but instead using the compound
+        # _posixsubprocess.fork_exec.
+        try:
+            from _testcapi import _spawn_pthread_waiter, _end_spawned_pthread
+        except ImportError:
+            pytest.skip("Cannot test without _testcapi module")
+
+        _spawn_pthread_waiter()
+        try:
+            with warnings.catch_warnings(
+                category=DeprecationWarning, record=True
+            ) as w:
+                warnings.simplefilter("always", category=DeprecationWarning)
+                executor = self.executor_type(max_workers=2)
+                try:
+                    list(executor.map(sqrt, range(10)))
+                finally:
+                    executor.shutdown(wait=True)
+            assert len(w) == 0, [w.message for w in w]
+        finally:
+            _end_spawned_pthread()
+
 
 def _custom_initializer():
     """_custom_initializer is module function to be picklable
diff --git a/tests/test_loky_backend.py b/tests/test_loky_backend.py
index 7f5b626..51d677d 100644
--- a/tests/test_loky_backend.py
+++ b/tests/test_loky_backend.py
@@ -71,7 +71,6 @@ class TestLokyBackend:
             kill_process_tree(child_process)
 
     def test_current(self):
-
         current = self.current_process()
         authkey = current.authkey
 
@@ -83,7 +82,6 @@ class TestLokyBackend:
         assert current.exitcode is None
 
     def test_daemon_argument(self):
-
         # By default uses the current process's daemon flag.
         proc0 = self.Process(target=self._test_process)
         assert proc0.daemon == self.current_process().daemon
@@ -296,7 +294,6 @@ class TestLokyBackend:
         time.sleep(100)
 
     def test_terminate(self):
-
         manager = self.Manager()
         event = manager.Event()
 
@@ -480,38 +477,42 @@ class TestLokyBackend:
 
         n_pipe = 0
         named_sem = []
+        unexpected = []
         for fd, t, name in zip(lines[::3], lines[1::3], lines[2::3]):
+            # Check if fd is a standard IO file.
+            is_std = fd in ["f0", "f1", "f2"]
 
-            # Check if fd is a standard IO file. For python 3.x stdin
-            # should be closed.
-            is_std = fd in ["f1", "f2"]
-
-            # Check if fd is a pipe
+            # Check if fd is a pipe.
             is_pipe = t in ["tPIPE", "tFIFO"]
             n_pipe += is_pipe
 
             # Check if fd is open for the rng. This can happen on different
-            # plateform and depending of the python version.
+            # platform and depending of the python version.
             is_rng = name == "n/dev/urandom"
 
+            # Not sure if this really expected to be open or not...
+            is_dev_null = name == "n/dev/null"
+
             # Check if fd is a semaphore or an open library. Store all the
-            # named semaphore
+            # named semaphore.
             is_mem = fd in ["fmem", "fDEL"]
             if sys.platform == "darwin":
                 is_mem |= "n/loky-" in name
             if is_mem and "n/dev/shm/sem." in name:
                 named_sem += [name[1:]]
 
-            # no other files should be opened at this stage in the process
-            assert is_pipe or is_std or is_rng or is_mem
+            # No other files should be opened at this stage in the process.
+            if not (is_pipe or is_std or is_rng or is_mem or is_dev_null):
+                unexpected.append((fd, t, name))
+
+        assert len(unexpected) == 0
 
         # there should be:
         # - one pipe for communication with main process
         # - loky's resource_tracker pipe
         # - the Connection pipe
-        # - additionally, on posix + Python 3.8: multiprocessing's
-        #   resource_tracker pipe
-        if sys.version_info >= (3, 8) and os.name == "posix":
+        # - multiprocessing's resource_tracker pipe (POSIX only)
+        if os.name == "posix":
             n_expected_pipes = 4
         else:
             n_expected_pipes = 3
@@ -552,7 +553,6 @@ class TestLokyBackend:
             )
             named_sem = []
             try:
-
                 p.start()
                 assert started.wait(5), "The process took too long to start"
                 r.close()
diff --git a/tests/test_reusable_executor.py b/tests/test_reusable_executor.py
index 2225f25..5043557 100644
--- a/tests/test_reusable_executor.py
+++ b/tests/test_reusable_executor.py
@@ -949,6 +949,17 @@ class TestGetReusableExecutor(ReusableExecutorMixin):
         executor4 = get_reusable_executor()
         assert executor4 is executor3
 
+    def test_no_crash_on_stdin_access_in_child_subprocess_run(self):
+        # non-regression test for #420.
+        def call_subprocess():
+            sleep(0.01)
+            subprocess.run(
+                [sys.executable, "-c", "import sys; sys.stdin.read(0)"]
+            ).check_returncode()
+
+        executor = get_reusable_executor(max_workers=4, timeout=2)
+        executor.submit(call_subprocess).result()
+
     def test_faulthandler_enabled(self):
         cmd = """if 1:
             from loky import get_reusable_executor