FIX support for frozen executable on all platforms

.azure_pipeline.yml

@@ -109,3 +109,51 @@ jobs:
         curl -s https://codecov.io/bash | bash
       displayName: 'Upload to codecov'
       condition: and(succeeded(), ne(variables['joblib.tests'], 'true'))
+
+
+- job: 'test_frozen_loky'
+  strategy:
+    matrix:
+
+      windows-py310:
+        imageName: windows-latest
+        python.version: "3.10"
+      macos-py310:
+        imageName: "macos-latest"
+        python.version: "3.10"
+      linux-py310:
+        imageName: "ubuntu-latest"
+        python.version: "3.10"
+  pool:
+    vmImage: $(imageName)
+  variables:
+    JUNITXML: 'test-data.xml'
+    PYINSTALLER_TESTS: "true"
+  steps:
+    - task: UsePythonVersion@0
+      inputs:
+        versionSpec: '$(python.version)'
+      displayName: 'Use Python $(python.version)'
+
+    # azure-pipelines unpredictably switches between Git\bin\bash and
+    # Git\usr\bin\bash when running a bash script inside Windows environments.
+    # The latter may use wrong bash commands, resulting in errors when codecov
+    # tries to upload the coverage results.
+    - bash: echo "##vso[task.prependpath]C:/Program Files/Git/bin"
+      displayName: 'Override Git bash shell for Windows'
+      condition: eq(variables['Agent.OS'], 'Windows_NT')
+
+    - script: |
+        bash continuous_integration/runtests.sh
+      displayName: 'Test loky with PyInstaller'
+
+    - task: PublishTestResults@2
+      inputs:
+        testResultsFiles: '$(JUNITXML)'
+      displayName: 'Publish Test Results'
+      condition: succeededOrFailed()
+
+    - bash: |
+        curl -s https://codecov.io/bash | bash
+      displayName: 'Upload to codecov'
+      condition: and(succeeded(), ne(variables['joblib.tests'], 'true'))

continuous_integration/runtests.sh

@@ -23,6 +23,24 @@ if [ "$JOBLIB_TESTS" = "true" ]; then
     cp "$BUILD_SOURCESDIRECTORY"/continuous_integration/copy_loky.sh $JOBLIB/externals
     (cd $JOBLIB/externals && bash copy_loky.sh "$BUILD_SOURCESDIRECTORY")
     pytest -vl --ignore $JOBLIB/externals --pyargs joblib
+elif [ "$PYINSTALLER_TESTS" = "true" ]; then
+    python -m venv venv/
+    if [ -d "./venv/Scripts" ]; then
+        source ./venv/Scripts/activate
+    else
+        source ./venv/bin/activate
+    fi
+    which python
+    pip install pytest pytest-timeout psutil coverage pyinstaller
+    pip install .
+    python -c "import loky; print('loky.cpu_count():', loky.cpu_count())"
+    python -c "import os; print('os.cpu_count():', os.cpu_count())"
+    export COVERAGE_PROCESS_START=`pwd`/.coveragerc
+    python continuous_integration/install_coverage_subprocess_pth.py
+    pytest -vl --maxfail=5 --timeout=60 -k pyinstaller --junitxml="${JUNITXML}"
+    coverage combine --quiet --append
+    coverage xml -i  # language agnostic report for the codecov upload script
+    coverage report  # display the report as text on stdout
 else
     # Make sure that we have the python docker image cached locally to avoid
     # a timeout in a test that needs it.

loky/__init__.py

@@ -16,28 +16,34 @@
 
 from ._base import Future
 from .backend.context import cpu_count
+from .backend.spawn import freeze_support
 from .backend.reduction import set_loky_pickler
 from .reusable_executor import get_reusable_executor
 from .cloudpickle_wrapper import wrap_non_picklable_objects
 from .process_executor import BrokenProcessPool, ProcessPoolExecutor
 
 
 __all__ = [
-    "get_reusable_executor",
-    "cpu_count",
-    "wait",
-    "as_completed",
-    "Future",
+    # Constants
+    "ALL_COMPLETED",
+    "FIRST_COMPLETED",
+    "FIRST_EXCEPTION",
+    # Classes
     "Executor",
+    "Future",
     "ProcessPoolExecutor",
+    # Functions
+    "as_completed",
+    "cpu_count",
+    "freeze_support",
+    "get_reusable_executor",
+    "set_loky_pickler",
+    "wait",
+    "wrap_non_picklable_objects",
+    # Errors
     "BrokenProcessPool",
     "CancelledError",
     "TimeoutError",
-    "FIRST_COMPLETED",
-    "FIRST_EXCEPTION",
-    "ALL_COMPLETED",
-    "wrap_non_picklable_objects",
-    "set_loky_pickler",
 ]
 
 

loky/backend/fork_exec.py

@@ -35,9 +35,12 @@ def fork_exec(cmd, keep_fds, env=None):
     env = env or {}
     child_env = {**os.environ, **env}
 
+    # make sure fds are inheritable
+    [os.set_inheritable(fd, True) for fd in keep_fds]
+
     pid = os.fork()
     if pid == 0:  # pragma: no cover
         close_fds(keep_fds)
-        os.execve(sys.executable, cmd, child_env)
+        os.execve(cmd[0], cmd, child_env)
     else:
         return pid

loky/backend/popen_loky_posix.py

@@ -6,9 +6,8 @@
 import os
 import sys
 import signal
-import pickle
 from io import BytesIO
-from multiprocessing import util, process
+from multiprocessing import util
 from multiprocessing.connection import wait
 from multiprocessing.context import set_spawning_popen
 
@@ -31,6 +30,19 @@ def detach(self):
         return self.fd
 
 
+#
+# Backward compat for pypy and python<=3.7
+# XXX: to remove once 3.7 is not supported anymore.
+#
+if not hasattr(util, "close_fds"):
+
+    def _close_fds(*fds):
+        for fd in fds:
+            os.close(fd)
+
+    util.close_fds = _close_fds
+
+
 #
 # Start child process using subprocess.Popen
 #
@@ -49,7 +61,7 @@ def __init__(self, process_obj):
 
     def duplicate_for_child(self, fd):
         self._fds.append(fd)
-        return reduction._mk_inheritable(fd)
+        return fd
 
     def poll(self, flag=os.WNOHANG):
         if self.returncode is None:
@@ -90,7 +102,6 @@ def terminate(self):
                     raise
 
     def _launch(self, process_obj):
-
         tracker_fd = resource_tracker._resource_tracker.getfd()
 
         fp = BytesIO()
@@ -109,15 +120,12 @@ def _launch(self, process_obj):
         try:
             parent_r, child_w = os.pipe()
             child_r, parent_w = os.pipe()
-            # for fd in self._fds:
-            #     _mk_inheritable(fd)
-
-            cmd_python = [sys.executable]
-            cmd_python += ["-m", self.__module__]
-            cmd_python += ["--process-name", str(process_obj.name)]
-            cmd_python += ["--pipe", str(reduction._mk_inheritable(child_r))]
-            reduction._mk_inheritable(child_w)
-            reduction._mk_inheritable(tracker_fd)
+
+            cmd_python = spawn.get_command_line(
+                pipe_handle=child_r,
+                parent_pid=os.getpid(),
+                process_name=process_obj.name,
+            )
             self._fds += [child_r, child_w, tracker_fd]
             if sys.version_info >= (3, 8) and os.name == "posix":
                 mp_tracker_fd = prep_data["mp_tracker_args"]["fd"]
@@ -129,65 +137,30 @@ def _launch(self, process_obj):
             util.debug(
                 f"launched python with pid {pid} and cmd:\n{cmd_python}"
             )
-            self.sentinel = parent_r
 
+            # Write the preparation data in the queue in a backward compatible
+            # way.
+            # XXX: can this be simplify now that we only support python3.7+
             method = "getbuffer"
             if not hasattr(fp, method):
                 method = "getvalue"
-            with os.fdopen(parent_w, "wb") as f:
+            with os.fdopen(parent_w, "wb", closefd=False) as f:
                 f.write(getattr(fp, method)())
+
+            # Store the process's information
             self.pid = pid
+            self.sentinel = parent_r
         finally:
-            if parent_r is not None:
-                util.Finalize(self, os.close, (parent_r,))
+            fds_to_close = []
+            for fd in (parent_r, parent_w):
+                if fd is not None:
+                    fds_to_close.append(fd)
+            self.finalizer = util.Finalize(self, util.close_fds, fds_to_close)
+
             for fd in (child_r, child_w):
                 if fd is not None:
                     os.close(fd)
 
     @staticmethod
     def thread_is_spawning():
         return True
-
-
-if __name__ == "__main__":
-    import argparse
-
-    parser = argparse.ArgumentParser("Command line parser")
-    parser.add_argument(
-        "--pipe", type=int, required=True, help="File handle for the pipe"
-    )
-    parser.add_argument(
-        "--process-name",
-        type=str,
-        default=None,
-        help="Identifier for debugging purpose",
-    )
-
-    args = parser.parse_args()
-
-    info = {}
-    exitcode = 1
-    try:
-        with os.fdopen(args.pipe, "rb") as from_parent:
-            process.current_process()._inheriting = True
-            try:
-                prep_data = pickle.load(from_parent)
-                spawn.prepare(prep_data)
-                process_obj = pickle.load(from_parent)
-            finally:
-                del process.current_process()._inheriting
-
-        exitcode = process_obj._bootstrap()
-    except Exception:
-        print("\n\n" + "-" * 80)
-        print(f"{args.process_name} failed with traceback: ")
-        print("-" * 80)
-        import traceback
-
-        print(traceback.format_exc())
-        print("\n" + "-" * 80)
-    finally:
-        if from_parent is not None:
-            from_parent.close()
-
-        sys.exit(exitcode)

loky/backend/popen_loky_win32.py

@@ -2,16 +2,22 @@
 import sys
 import msvcrt
 import _winapi
-from pickle import load
-from multiprocessing import process, util
+
+from multiprocessing import util
 from multiprocessing.context import set_spawning_popen
+from multiprocessing.popen_spawn_win32 import _close_handles
 from multiprocessing.popen_spawn_win32 import Popen as _Popen
 
 from . import reduction, spawn
 
 
 __all__ = ["Popen"]
 
+POPEN_FLAG = 0
+if spawn.OPEN_CONSOLE_FOR_SUBPROCESSES:
+    POPEN_FLAG = _winapi.CREATE_NEW_CONSOLE
+
+
 #
 #
 #
@@ -65,9 +71,11 @@ def __init__(self, process_obj):
         # terminated before it could steal the handle from the parent process.
         rhandle, whandle = _winapi.CreatePipe(None, 0)
         wfd = msvcrt.open_osfhandle(whandle, 0)
-        cmd = get_command_line(parent_pid=os.getpid(), pipe_handle=rhandle)
 
-        python_exe = spawn.get_executable()
+        cmd = spawn.get_command_line(
+            pipe_handle=rhandle, parent_pid=os.getpid()
+        )
+        python_exe = cmd[0]
 
         # copy the environment variables to set in the child process
         child_env = {**os.environ, **process_obj.env}
@@ -79,7 +87,6 @@ def __init__(self, process_obj):
             child_env["__PYVENV_LAUNCHER__"] = sys.executable
 
         cmd = " ".join(f'"{x}"' for x in cmd)
-
         with open(wfd, "wb") as to_child:
             # start process
             try:
@@ -115,59 +122,3 @@ def __init__(self, process_obj):
                 reduction.dump(process_obj, to_child)
             finally:
                 set_spawning_popen(None)
-
-
-def get_command_line(pipe_handle, parent_pid, **kwds):
-    """Returns prefix of command line used for spawning a child process."""
-    if getattr(sys, "frozen", False):
-        return [sys.executable, "--multiprocessing-fork", pipe_handle]
-    else:
-        prog = (
-            "from loky.backend.popen_loky_win32 import main; "
-            f"main(pipe_handle={pipe_handle}, parent_pid={parent_pid})"
-        )
-        opts = util._args_from_interpreter_flags()
-        return [
-            spawn.get_executable(),
-            *opts,
-            "-c",
-            prog,
-            "--multiprocessing-fork",
-        ]
-
-
-def is_forking(argv):
-    """Return whether commandline indicates we are forking."""
-    if len(argv) >= 2 and argv[1] == "--multiprocessing-fork":
-        return True
-    else:
-        return False
-
-
-def main(pipe_handle, parent_pid=None):
-    """Run code specified by data received over pipe."""
-    assert is_forking(sys.argv), "Not forking"
-
-    if parent_pid is not None:
-        source_process = _winapi.OpenProcess(
-            _winapi.SYNCHRONIZE | _winapi.PROCESS_DUP_HANDLE, False, parent_pid
-        )
-    else:
-        source_process = None
-    new_handle = reduction.duplicate(
-        pipe_handle, source_process=source_process
-    )
-    fd = msvcrt.open_osfhandle(new_handle, os.O_RDONLY)
-    parent_sentinel = source_process
-
-    with os.fdopen(fd, "rb", closefd=True) as from_parent:
-        process.current_process()._inheriting = True
-        try:
-            preparation_data = load(from_parent)
-            spawn.prepare(preparation_data, parent_sentinel)
-            self = load(from_parent)
-        finally:
-            del process.current_process()._inheriting
-
-    exitcode = self._bootstrap(parent_sentinel)
-    sys.exit(exitcode)