Repository: joblib/joblib
Pull request #1525 (2023-11-20 14:16:21)

Add option to specify an initialization function for `'loky'` and `'multiprocessing'` backends
Closes https://github.com/joblib/joblib/issues/381

This PR adds the ability to specify an initialization function that is run once per worker process when using the `'loky'` and `'multiprocessing'` backends.

Usage:

```python
Parallel(initializer=my_func, initargs=(x, y, z))
# or
with parallel_config(initializer=my_func, initargs=(x, y, z)):
    ...
```

shwina: Hi @ogrisel - wondering if you have any thoughts on how to test the implementation here correctly? 
bdice: Hi @ogrisel, this feature request has come up a couple times recently, including in a StackOverflow question: https://stackoverflow.com/questions/78642680/using-load-ext-cudf-pandas-throws-attributeerror

Would it be possible to work with a joblib developer to identify how to move forward?
SJ-Innovation: Hi, just wondering is theres been any behind-the-scenes motion on this, as this would be a REALLY handy piece of functionality to have for pre-seeding workers with large, constant, but procedurally generated/loaded datasets at the start of a large number of jobs.  
codecov[bot]: ## [Codecov](https://app.codecov.io/gh/joblib/joblib/pull/1525?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=joblib) Report
Attention: Patch coverage is `40.90909%` with `65 lines` in your changes missing coverage. Please review.
> Project coverage is 69.67%. Comparing base [(`2a28aa5`)](https://app.codecov.io/gh/joblib/joblib/commit/2a28aa52f0f86f185df5ea8a1b67871fea607f6a?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=joblib) to head [(`943110a`)](https://app.codecov.io/gh/joblib/joblib/commit/943110a488887527d257c82e0d1689e9d276570b?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=joblib).

| [Files with missing lines](https://app.codecov.io/gh/joblib/joblib/pull/1525?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=joblib) | Patch % | Lines |
|---|---|---|
| [joblib/test/test\_parallel.py](https://app.codecov.io/gh/joblib/joblib/pull/1525?src=pr&el=tree&filepath=joblib%2Ftest%2Ftest_parallel.py&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=joblib#diff-am9ibGliL3Rlc3QvdGVzdF9wYXJhbGxlbC5weQ==) | 39.24% | [48 Missing :warning: ](https://app.codecov.io/gh/joblib/joblib/pull/1525?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=joblib) |
| [joblib/parallel.py](https://app.codecov.io/gh/joblib/joblib/pull/1525?src=pr&el=tree&filepath=joblib%2Fparallel.py&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=joblib#diff-am9ibGliL3BhcmFsbGVsLnB5) | 44.00% | [14 Missing :warning: ](https://app.codecov.io/gh/joblib/joblib/pull/1525?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=joblib) |
| [joblib/\_parallel\_backends.py](https://app.codecov.io/gh/joblib/joblib/pull/1525?src=pr&el=tree&filepath=joblib%2F_parallel_backends.py&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=joblib#diff-am9ibGliL19wYXJhbGxlbF9iYWNrZW5kcy5weQ==) | 50.00% | [3 Missing :warning: ](https://app.codecov.io/gh/joblib/joblib/pull/1525?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=joblib) |

> :exclamation:  There is a different number of reports uploaded between BASE (2a28aa5) and HEAD (943110a). Click for more details.
> 
> <details><summary>HEAD has 22 uploads less than BASE</summary>
>
>| Flag | BASE (2a28aa5) | HEAD (943110a) |
>|------|------|------|
>||23|1|
></details>

<details><summary>Additional details and impacted files</summary>


```diff
@@             Coverage Diff             @@
##             main    #1525       +/-   ##
===========================================
- Coverage   95.48%   69.67%   -25.81%     
===========================================
  Files          46       46               
  Lines        7706     7664       -42     
===========================================
- Hits         7358     5340     -2018     
- Misses        348     2324     +1976     
```

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/joblib/joblib/pull/1525?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=joblib).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=joblib).


Patch:
diff --git a/CHANGES.rst b/CHANGES.rst
index 497b6e7..b7a66c5 100644
--- a/CHANGES.rst
+++ b/CHANGES.rst
@@ -34,6 +34,10 @@ Parallel:
 - Pretty printing of ``Parallel`` execution progress when the number of tasks is
   known. https://github.com/joblib/joblib/pull/1608
 
+- Make it possible to pass extra arguments to the ``LokyBackend`` and
+  ``MultiprocessingBackend``, enabling the use of ``initializer``.
+  https://github.com/joblib/joblib/pull/1525
+
 - Refactor and document the custom parallel backend API.
   https://github.com/joblib/joblib/pull/1667
 
@@ -56,6 +60,29 @@ Maintenance:
   the warnings that are emitted in this case.
   https://github.com/joblib/joblib/pull/1681
 
+- Fix a regression in 1.3 and 1.4 that caused large big endian arrays to trigger
+  a serialization error. https://github.com/joblib/joblib/issues/1545
+
+- Added a ``ensure_native_byte_order`` parameter to ``joblib.load``. When
+  ``True`` and ``mmap_mode`` is ``None``, loaded arrays are automatically coerced
+  to a byte order that matches the endianness of the host system. This behavior
+  has been the default since ``joblib==1.3``, and can now be disabled if the
+  parameter is set to ``False`` instead. Note that setting it to ``True`` will
+  raise an error if ``mmap_mode`` is not null. The default value ``'auto'`` is
+  equivalent to always setting ``True`` if ``mmap_mode`` is ``None``, else always
+  ``False``.  https://github.com/joblib/joblib/pull/1561
+
+- Fix support for python 3.14 in ``hashing``, with the addition of
+  an extra argument in ``Pickler._batch_setitems``.
+  https://github.com/joblib/joblib/pull/1688
+
+- Bump vendored cloudpickle to 3.1.1 to support Python 3.14 (dev) and
+  various other fixes.
+
+- Bump vendored loky to 3.5.1 to support recent Python versions without raising
+  the warning on calls to `os.fork` and fix various sources of crashes and
+  deadlocks.
+
 
 Release 1.4.2 -- 2024/05/02
 ---------------------------
diff --git a/continuous_integration/install.sh b/continuous_integration/install.sh
index fa8d974..b4b9e86 100755
--- a/continuous_integration/install.sh
+++ b/continuous_integration/install.sh
@@ -11,28 +11,16 @@ set -xe
 
 create_new_conda_env() {
     conda config --set solver libmamba
+    if [[ "$PYTHON_VERSION" == free-threaded* ]]; then
+        PYTHON_VERSION=${PYTHON_VERSION/free-threaded-/}
+        EXTRA_CONDA_PACKAGES="python-freethreading $EXTRA_CONDA_PACKAGES"
+    fi
     to_install="python=$PYTHON_VERSION pip pytest $EXTRA_CONDA_PACKAGES"
     conda create -n testenv --yes -c conda-forge $to_install
     conda activate testenv
 }
 
-create_new_free_threaded_env() {
-    sudo apt-get -yq update
-    sudo apt-get install -yq ccache
-    sudo apt-get install -yq software-properties-common
-    sudo add-apt-repository --yes ppa:deadsnakes/nightly
-    sudo apt-get update -yq
-    sudo apt-get install -yq --no-install-recommends python3.13-dev python3.13-venv python3.13-nogil
-
-    python3.13t -m venv testenv
-    source testenv/bin/activate
-}
-
-if [[ "$PYTHON_VERSION" == free-threaded* ]]; then
-    create_new_free_threaded_env
-else
-    create_new_conda_env
-fi
+create_new_conda_env
 
 # Install pytest timeout to fasten failure in deadlocking tests
 PIP_INSTALL_PACKAGES="pytest-timeout pytest-asyncio==0.21.1 threadpoolctl"
diff --git a/doc/memory.rst b/doc/memory.rst
index 9a4760c..bda4767 100644
--- a/doc/memory.rst
+++ b/doc/memory.rst
@@ -462,8 +462,8 @@ keys:
   - ``input_args``: a dictionary of keywords arguments for the cached function call.
 
 Note a validity duration for cached results can be defined via
-:func:`joblib.expires_after` by providing similar with arguments similar to the
-ones of a ``datetime.timedelta``:
+:func:`joblib.expires_after` by providing arguments similar to the ones of a
+``datetime.timedelta``:
 
     >>> from joblib import expires_after
     >>> @memory.cache(cache_validation_callback=expires_after(seconds=0.5))
diff --git a/doc/parallel.rst b/doc/parallel.rst
index cf4a5ce..309103b 100644
--- a/doc/parallel.rst
+++ b/doc/parallel.rst
@@ -129,8 +129,20 @@ The context manager allow to set various backend parameters:
   of the backend. See :ref:`auto_memmapping_doc` for more details.
 
 Extra arguments passed in the :func:`~joblib.parallel_config` context are
-passed to the backend constructor, giving the possibility to set up the
-backend with additional parameters.
+passed to the backend constructor, allowing additional parameters to be set up:
+
+- ``LokyBackend``
+    - ``initializer``, ``initargs``: setup function and its arguments to call
+      in each worker process.
+    - ``idle_worker_timeout``: timeout in seconds for a worker to wait
+      for a new task before being clean up. The default is `300 s`.
+- ``MultprocessingBackend``
+    - ``initializer``, ``initargs``: setup function and its arguments to call
+      in each worker process.
+    - ``maxtasksperchild``: maximum number of tasks a worker can
+      execute before being replace with a fresh one.
+    - ``context``: specify the start method to use for creating new worker
+      processes.
 
 In addition to the builtin joblib backends, there are several cluster-specific
 backends you can use:
diff --git a/doc/parallel_numpy.rst b/doc/parallel_numpy.rst
index 1bfe04b..94f3536 100644
--- a/doc/parallel_numpy.rst
+++ b/doc/parallel_numpy.rst
@@ -23,7 +23,7 @@ worker processes.
 
 .. note::
 
-  The following only applies with the ``"loky"` and
+  The following only applies with the ``"loky"`` and
   ``'multiprocessing'`` process-backends. If your code can release the
   GIL, then using a thread-based backend by passing
   ``prefer='threads'`` is even more efficient because it makes it
diff --git a/joblib/_parallel_backends.py b/joblib/_parallel_backends.py
index 20ec44a..53114a8 100644
--- a/joblib/_parallel_backends.py
+++ b/joblib/_parallel_backends.py
@@ -48,10 +48,13 @@ class ParallelBackendBase(metaclass=ABCMeta):
 
     nesting_level = None
 
-    def __init__(self, nesting_level=None, inner_max_num_threads=None, **kwargs):
-        super().__init__(**kwargs)
+    def __init__(
+        self, nesting_level=None, inner_max_num_threads=None, **backend_kwargs
+    ):
+        super().__init__()
         self.nesting_level = nesting_level
         self.inner_max_num_threads = inner_max_num_threads
+        self.backend_kwargs = backend_kwargs
 
     MAX_NUM_THREADS_VARS = [
         "OMP_NUM_THREADS",
@@ -152,12 +155,7 @@ class ParallelBackendBase(metaclass=ABCMeta):
             return out.get()
 
     def configure(
-        self,
-        n_jobs=1,
-        parallel=None,
-        prefer=None,
-        require=None,
-        **backend_args,
+        self, n_jobs=1, parallel=None, prefer=None, require=None, **backend_kwargs
     ):
         """Reconfigure the backend and return the number of workers.
 
@@ -360,7 +358,7 @@ class PoolManagerMixin(object):
             self.configure(
                 n_jobs=self.parallel.n_jobs,
                 parallel=self.parallel,
-                **self.parallel._backend_args,
+                **self.parallel._backend_kwargs,
             )
 
 
@@ -489,7 +487,7 @@ class ThreadingBackend(PoolManagerMixin, ParallelBackendBase):
     uses_threads = True
     supports_sharedmem = True
 
-    def configure(self, n_jobs=1, parallel=None, **backend_args):
+    def configure(self, n_jobs=1, parallel=None, **backend_kwargs):
         """Build a process or thread pool and return the number of workers"""
         n_jobs = self.effective_n_jobs(n_jobs)
         if n_jobs == 1:
@@ -574,16 +572,26 @@ class MultiprocessingBackend(PoolManagerMixin, AutoBatchingMixin, ParallelBacken
         return super(MultiprocessingBackend, self).effective_n_jobs(n_jobs)
 
     def configure(
-        self, n_jobs=1, parallel=None, prefer=None, require=None, **memmappingpool_args
+        self,
+        n_jobs=1,
+        parallel=None,
+        prefer=None,
+        require=None,
+        **memmapping_pool_kwargs,
     ):
         """Build a process or thread pool and return the number of workers"""
         n_jobs = self.effective_n_jobs(n_jobs)
         if n_jobs == 1:
             raise FallbackToBackend(SequentialBackend(nesting_level=self.nesting_level))
 
+        memmapping_pool_kwargs = {
+            **self.backend_kwargs,
+            **memmapping_pool_kwargs,
+        }
+
         # Make sure to free as much memory as possible before forking
         gc.collect()
-        self._pool = MemmappingPool(n_jobs, **memmappingpool_args)
+        self._pool = MemmappingPool(n_jobs, **memmapping_pool_kwargs)
         self.parallel = parallel
         return n_jobs
 
@@ -605,20 +613,35 @@ class LokyBackend(AutoBatchingMixin, ParallelBackendBase):
         parallel=None,
         prefer=None,
         require=None,
-        idle_worker_timeout=300,
-        **memmappingexecutor_args,
+        idle_worker_timeout=None,
+        **memmapping_executor_kwargs,
     ):
         """Build a process executor and return the number of workers"""
         n_jobs = self.effective_n_jobs(n_jobs)
         if n_jobs == 1:
             raise FallbackToBackend(SequentialBackend(nesting_level=self.nesting_level))
 
+        memmapping_executor_kwargs = {
+            **self.backend_kwargs,
+            **memmapping_executor_kwargs,
+        }
+
+        # Prohibit the use of 'timeout' in the LokyBackend, as 'idle_worker_timeout'
+        # better describes the backend's behavior.
+        if "timeout" in memmapping_executor_kwargs:
+            raise ValueError(
+                "The 'timeout' parameter is not supported by the LokyBackend. "
+                "Please use the `idle_worker_timeout` parameter instead."
+            )
+        if idle_worker_timeout is None:
+            idle_worker_timeout = self.backend_kwargs.get("idle_worker_timeout", 300)
+
         self._workers = get_memmapping_executor(
             n_jobs,
             timeout=idle_worker_timeout,
             env=self._prepare_worker_env(n_jobs=n_jobs),
             context_id=parallel._id,
-            **memmappingexecutor_args,
+            **memmapping_executor_kwargs,
         )
         self.parallel = parallel
         return n_jobs
diff --git a/joblib/externals/cloudpickle/__init__.py b/joblib/externals/cloudpickle/__init__.py
index 58a8d08..3768a93 100644
--- a/joblib/externals/cloudpickle/__init__.py
+++ b/joblib/externals/cloudpickle/__init__.py
@@ -3,7 +3,7 @@ from .cloudpickle import *  # noqa
 
 __doc__ = cloudpickle.__doc__
 
-__version__ = "3.0.0"
+__version__ = "3.1.1"
 
 __all__ = [  # noqa
     "__version__",
diff --git a/joblib/externals/cloudpickle/cloudpickle.py b/joblib/externals/cloudpickle/cloudpickle.py
index eb43a96..4d532e5 100644
--- a/joblib/externals/cloudpickle/cloudpickle.py
+++ b/joblib/externals/cloudpickle/cloudpickle.py
@@ -63,7 +63,7 @@ import itertools
 import logging
 import opcode
 import pickle
-from pickle import _getattribute
+from pickle import _getattribute as _pickle_getattribute
 import platform
 import struct
 import sys
@@ -126,7 +126,7 @@ def _lookup_class_or_track(class_tracker_id, class_def):
 
 
 def register_pickle_by_value(module):
-    """Register a module to make it functions and classes picklable by value.
+    """Register a module to make its functions and classes picklable by value.
 
     By default, functions and classes that are attributes of an importable
     module are to be pickled by reference, that is relying on re-importing
@@ -192,6 +192,14 @@ def _is_registered_pickle_by_value(module):
     return False
 
 
+if sys.version_info >= (3, 14):
+    def _getattribute(obj, name):
+        return _pickle_getattribute(obj, name.split('.'))
+else:
+    def _getattribute(obj, name):
+        return _pickle_getattribute(obj, name)[0]
+
+
 def _whichmodule(obj, name):
     """Find the module an object belongs to.
 
@@ -213,12 +221,13 @@ def _whichmodule(obj, name):
         # sys.modules
         if (
             module_name == "__main__"
+            or module_name == "__mp_main__"
             or module is None
             or not isinstance(module, types.ModuleType)
         ):
             continue
         try:
-            if _getattribute(module, name)[0] is obj:
+            if _getattribute(module, name) is obj:
                 return module_name
         except Exception:
             pass
@@ -292,7 +301,7 @@ def _lookup_module_and_qualname(obj, name=None):
         return None
 
     try:
-        obj2, parent = _getattribute(module, name)
+        obj2 = _getattribute(module, name)
     except AttributeError:
         # obj was not found inside the module it points to
         return None
@@ -409,7 +418,10 @@ def _walk_global_ops(code):
 
 def _extract_class_dict(cls):
     """Retrieve a copy of the dict of a class without the inherited method."""
-    clsdict = dict(cls.__dict__)  # copy dict proxy to a dict
+    # Hack to circumvent non-predictable memoization caused by string interning.
+    # See the inline comment in _class_setstate for details.
+    clsdict = {"".join(k): cls.__dict__[k] for k in sorted(cls.__dict__)}
+
     if len(cls.__bases__) == 1:
         inherited_dict = cls.__bases__[0].__dict__
     else:
@@ -533,9 +545,15 @@ def _make_skeleton_class(
     The "extra" variable is meant to be a dict (or None) that can be used for
     forward compatibility shall the need arise.
     """
+    # We need to intern the keys of the type_kwargs dict to avoid having
+    # different pickles for the same dynamic class depending on whether it was
+    # dynamically created or reconstructed from a pickled stream.
+    type_kwargs = {sys.intern(k): v for k, v in type_kwargs.items()}
+
     skeleton_class = types.new_class(
         name, bases, {"metaclass": type_constructor}, lambda ns: ns.update(type_kwargs)
     )
+
     return _lookup_class_or_track(class_tracker_id, skeleton_class)
 
 
@@ -694,8 +712,10 @@ def _function_getstate(func):
     #   unpickling time by iterating over slotstate and calling setattr(func,
     #   slotname, slotvalue)
     slotstate = {
-        "__name__": func.__name__,
-        "__qualname__": func.__qualname__,
+        # Hack to circumvent non-predictable memoization caused by string interning.
+        # See the inline comment in _class_setstate for details.
+        "__name__": "".join(func.__name__),
+        "__qualname__": "".join(func.__qualname__),
         "__annotations__": func.__annotations__,
         "__kwdefaults__": func.__kwdefaults__,
         "__defaults__": func.__defaults__,
@@ -721,7 +741,9 @@ def _function_getstate(func):
     )
     slotstate["__globals__"] = f_globals
 
-    state = func.__dict__
+    # Hack to circumvent non-predictable memoization caused by string interning.
+    # See the inline comment in _class_setstate for details.
+    state = {"".join(k): v for k, v in func.__dict__.items()}
     return state, slotstate
 
 
@@ -802,6 +824,19 @@ def _code_reduce(obj):
     # of the specific type from types, for example:
     # >>> from types import CodeType
     # >>> help(CodeType)
+
+    # Hack to circumvent non-predictable memoization caused by string interning.
+    # See the inline comment in _class_setstate for details.
+    co_name = "".join(obj.co_name)
+
+    # Create shallow copies of these tuple to make cloudpickle payload deterministic.
+    # When creating a code object during load, copies of these four tuples are
+    # created, while in the main process, these tuples can be shared.
+    # By always creating copies, we make sure the resulting payload is deterministic.
+    co_names = tuple(name for name in obj.co_names)
+    co_varnames = tuple(name for name in obj.co_varnames)
+    co_freevars = tuple(name for name in obj.co_freevars)
+    co_cellvars = tuple(name for name in obj.co_cellvars)
     if hasattr(obj, "co_exceptiontable"):
         # Python 3.11 and later: there are some new attributes
         # related to the enhanced exceptions.
@@ -814,16 +849,16 @@ def _code_reduce(obj):
             obj.co_flags,
             obj.co_code,
             obj.co_consts,
-            obj.co_names,
-            obj.co_varnames,
+            co_names,
+            co_varnames,
             obj.co_filename,
-            obj.co_name,
+            co_name,
             obj.co_qualname,
             obj.co_firstlineno,
             obj.co_linetable,
             obj.co_exceptiontable,
-            obj.co_freevars,
-            obj.co_cellvars,
+            co_freevars,
+            co_cellvars,
         )
     elif hasattr(obj, "co_linetable"):
         # Python 3.10 and later: obj.co_lnotab is deprecated and constructor
@@ -837,14 +872,14 @@ def _code_reduce(obj):
             obj.co_flags,
             obj.co_code,
             obj.co_consts,
-            obj.co_names,
-            obj.co_varnames,
+            co_names,
+            co_varnames,
             obj.co_filename,
-            obj.co_name,
+            co_name,
             obj.co_firstlineno,
             obj.co_linetable,
-            obj.co_freevars,
-            obj.co_cellvars,
+            co_freevars,
+            co_cellvars,
         )
     elif hasattr(obj, "co_nmeta"):  # pragma: no cover
         # "nogil" Python: modified attributes from 3.9
@@ -859,15 +894,15 @@ def _code_reduce(obj):
             obj.co_flags,
             obj.co_code,
             obj.co_consts,
-            obj.co_varnames,
+            co_varnames,
             obj.co_filename,
-            obj.co_name,
+            co_name,
             obj.co_firstlineno,
             obj.co_lnotab,
             obj.co_exc_handlers,
             obj.co_jump_table,
-            obj.co_freevars,
-            obj.co_cellvars,
+            co_freevars,
+            co_cellvars,
             obj.co_free2reg,
             obj.co_cell2reg,
         )
@@ -882,14 +917,14 @@ def _code_reduce(obj):
             obj.co_flags,
             obj.co_code,
             obj.co_consts,
-            obj.co_names,
-            obj.co_varnames,
+            co_names,
+            co_varnames,
             obj.co_filename,
-            obj.co_name,
+            co_name,
             obj.co_firstlineno,
             obj.co_lnotab,
-            obj.co_freevars,
-            obj.co_cellvars,
+            co_freevars,
+            co_cellvars,
         )
     return types.CodeType, args
 
@@ -1127,7 +1162,30 @@ def _class_setstate(obj, state):
         if attrname == "_abc_impl":
             registry = attr
         else:
+            # Note: setting attribute names on a class automatically triggers their
+            # interning in CPython:
+            # https://github.com/python/cpython/blob/v3.12.0/Objects/object.c#L957
+            #
+            # This means that to get deterministic pickling for a dynamic class that
+            # was initially defined in a different Python process, the pickler
+            # needs to ensure that dynamic class and function attribute names are
+            # systematically copied into a non-interned version to avoid
+            # unpredictable pickle payloads.
+            #
+            # Indeed the Pickler's memoizer relies on physical object identity to break
+            # cycles in the reference graph of the object being serialized.
             setattr(obj, attrname, attr)
+
+    if sys.version_info >= (3, 13) and "__firstlineno__" in state:
+        # Set the Python 3.13+ only __firstlineno__  attribute one more time, as it
+        # will be automatically deleted by the `setattr(obj, attrname, attr)` call
+        # above when `attrname` is "__firstlineno__". We assume that preserving this
+        # information might be important for some users and that it not stale in the
+        # context of cloudpickle usage, hence legitimate to propagate. Furthermore it
+        # is necessary to do so to keep deterministic chained pickling as tested in
+        # test_deterministic_str_interning_for_chained_dynamic_class_pickling.
+        obj.__firstlineno__ = state["__firstlineno__"]
+
     if registry is not None:
         for subclass in registry:
             obj.register(subclass)
diff --git a/joblib/externals/cloudpickle/cloudpickle_fast.py b/joblib/externals/cloudpickle/cloudpickle_fast.py
index 52d6732..20280f0 100644
--- a/joblib/externals/cloudpickle/cloudpickle_fast.py
+++ b/joblib/externals/cloudpickle/cloudpickle_fast.py
@@ -6,6 +6,7 @@ namespace.
 
 See: tests/test_backward_compat.py
 """
+
 from . import cloudpickle
 
 
diff --git a/joblib/externals/loky/__init__.py b/joblib/externals/loky/__init__.py
index ceb97dc..1517bf8 100644
--- a/joblib/externals/loky/__init__.py
+++ b/joblib/externals/loky/__init__.py
@@ -1,8 +1,9 @@
-r"""The :mod:`loky` module manages a pool of worker that can be reused across time.
+r"""The :mod:`loky` module manages a pool of worker that can be re-used across time.
 It provides a robust and dynamic implementation os the
 :class:`ProcessPoolExecutor` and a function :func:`get_reusable_executor` which
 hide the pool management under the hood.
 """
+
 from concurrent.futures import (
     ALL_COMPLETED,
     FIRST_COMPLETED,
@@ -41,4 +42,4 @@ __all__ = [
 ]
 
 
-__version__ = "3.4.1"
+__version__ = "3.5.1"
diff --git a/joblib/externals/loky/backend/context.py b/joblib/externals/loky/backend/context.py
index ea4b8be..efd98bf 100644
--- a/joblib/externals/loky/backend/context.py
+++ b/joblib/externals/loky/backend/context.py
@@ -18,20 +18,15 @@ import warnings
 import multiprocessing as mp
 from multiprocessing import get_context as mp_get_context
 from multiprocessing.context import BaseContext
+from concurrent.futures.process import _MAX_WINDOWS_WORKERS
 
 
 from .process import LokyProcess, LokyInitMainProcess
 
 # Apparently, on older Python versions, loky cannot work 61 workers on Windows
 # but instead 60: ¯\_(ツ)_/¯
-if sys.version_info >= (3, 8):
-    from concurrent.futures.process import _MAX_WINDOWS_WORKERS
-
-    if sys.version_info < (3, 10):
-        _MAX_WINDOWS_WORKERS = _MAX_WINDOWS_WORKERS - 1
-else:
-    # compat for versions before 3.8 which do not define this.
-    _MAX_WINDOWS_WORKERS = 60
+if sys.version_info < (3, 10):
+    _MAX_WINDOWS_WORKERS = _MAX_WINDOWS_WORKERS - 1
 
 START_METHODS = ["loky", "loky_init_main", "spawn"]
 if sys.platform != "win32":
@@ -177,7 +172,7 @@ def _cpu_count_cgroup(os_cpu_count):
             return math.ceil(cpu_quota_us / cpu_period_us)
         else:  # pragma: no cover
             # Setting a negative cpu_quota_us value is a valid way to disable
-            # cgroup CPU bandwidth limits
+            # cgroup CPU bandwith limits
             return os_cpu_count
 
 
@@ -189,8 +184,8 @@ def _cpu_count_affinity(os_cpu_count):
         except NotImplementedError:
             pass
 
-    # On PyPy and possibly other platforms, os.sched_getaffinity does not exist
-    # or raises NotImplementedError, let's try with the psutil if installed.
+    # On some platforms, os.sched_getaffinity does not exist or raises
+    # NotImplementedError, let's try with the psutil if installed.
     try:
         import psutil
 
@@ -203,13 +198,13 @@ def _cpu_count_affinity(os_cpu_count):
             sys.platform == "linux"
             and os.environ.get("LOKY_MAX_CPU_COUNT") is None
         ):
-            # PyPy does not implement os.sched_getaffinity on Linux which
+            # Some platforms don't implement os.sched_getaffinity on Linux which
             # can cause severe oversubscription problems. Better warn the
             # user in this particularly pathological case which can wreck
             # havoc, typically on CI workers.
             warnings.warn(
                 "Failed to inspect CPU affinity constraints on this system. "
-                "Please install psutil or explicitly set LOKY_MAX_CPU_COUNT."
+                "Please install psutil or explictly set LOKY_MAX_CPU_COUNT."
             )
 
     # This can happen for platforms that do not implement any kind of CPU
@@ -247,33 +242,11 @@ def _count_physical_cores():
     # Not cached yet, find it
     try:
         if sys.platform == "linux":
-            cpu_info = subprocess.run(
-                "lscpu --parse=core".split(), capture_output=True, text=True
-            )
-            cpu_info = cpu_info.stdout.splitlines()
-            cpu_info = {line for line in cpu_info if not line.startswith("#")}
-            cpu_count_physical = len(cpu_info)
+            cpu_count_physical = _count_physical_cores_linux()
         elif sys.platform == "win32":
-            cpu_info = subprocess.run(
-                "wmic CPU Get NumberOfCores /Format:csv".split(),
-                capture_output=True,
-                text=True,
-            )
-            cpu_info = cpu_info.stdout.splitlines()
-            cpu_info = [
-                l.split(",")[1]
-                for l in cpu_info
-                if (l and l != "Node,NumberOfCores")
-            ]
-            cpu_count_physical = sum(map(int, cpu_info))
+            cpu_count_physical = _count_physical_cores_win32()
         elif sys.platform == "darwin":
-            cpu_info = subprocess.run(
-                "sysctl -n hw.physicalcpu".split(),
-                capture_output=True,
-                text=True,
-            )
-            cpu_info = cpu_info.stdout
-            cpu_count_physical = int(cpu_info)
+            cpu_count_physical = _count_physical_cores_darwin()
         else:
             raise NotImplementedError(f"unsupported platform: {sys.platform}")
 
@@ -291,6 +264,60 @@ def _count_physical_cores():
     return cpu_count_physical, exception
 
 
+def _count_physical_cores_linux():
+    try:
+        cpu_info = subprocess.run(
+            "lscpu --parse=core".split(), capture_output=True, text=True
+        )
+        cpu_info = cpu_info.stdout.splitlines()
+        cpu_info = {line for line in cpu_info if not line.startswith("#")}
+        return len(cpu_info)
+    except:
+        pass  # fallback to /proc/cpuinfo
+
+    cpu_info = subprocess.run(
+        "cat /proc/cpuinfo".split(), capture_output=True, text=True
+    )
+    cpu_info = cpu_info.stdout.splitlines()
+    cpu_info = {line for line in cpu_info if line.startswith("core id")}
+    return len(cpu_info)
+
+
+def _count_physical_cores_win32():
+    try:
+        cmd = "-Command (Get-CimInstance -ClassName Win32_Processor).NumberOfCores"
+        cpu_info = subprocess.run(
+            f"powershell.exe {cmd}".split(),
+            capture_output=True,
+            text=True,
+        )
+        cpu_info = cpu_info.stdout.splitlines()
+        return int(cpu_info[0])
+    except:
+        pass  # fallback to wmic (older Windows versions; deprecated now)
+
+    cpu_info = subprocess.run(
+        "wmic CPU Get NumberOfCores /Format:csv".split(),
+        capture_output=True,
+        text=True,
+    )
+    cpu_info = cpu_info.stdout.splitlines()
+    cpu_info = [
+        l.split(",")[1] for l in cpu_info if (l and l != "Node,NumberOfCores")
+    ]
+    return sum(map(int, cpu_info))
+
+
+def _count_physical_cores_darwin():
+    cpu_info = subprocess.run(
+        "sysctl -n hw.physicalcpu".split(),
+        capture_output=True,
+        text=True,
+    )
+    cpu_info = cpu_info.stdout
+    return int(cpu_info)
+
+
 class LokyContext(BaseContext):
     """Context relying on the LokyProcess."""
 
diff --git a/joblib/externals/loky/backend/fork_exec.py b/joblib/externals/loky/backend/fork_exec.py
index 2353c42..f5b7ca6 100644
--- a/joblib/externals/loky/backend/fork_exec.py
+++ b/joblib/externals/loky/backend/fork_exec.py
@@ -4,40 +4,70 @@
 #
 # 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()
+
+    if sys.version_info >= (3, 14):
+        # Python >= 3.14 removed allow_vfork from _posixsubprocess.fork_exec,
+        # see https://github.com/python/cpython/pull/121383
+        pgid_to_set = [-1]
+        allow_vfork = []
+    elif sys.version_info >= (3, 11):
+        # Python 3.11 - 3.13 has allow_vfork in _posixsubprocess.fork_exec
+        pgid_to_set = [-1]
+        allow_vfork = [subprocess._USE_VFORK]
     else:
-        return pid
+        # Python < 3.11
+        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/joblib/externals/loky/backend/queues.py b/joblib/externals/loky/backend/queues.py
index e92ad82..5afd99b 100644
--- a/joblib/externals/loky/backend/queues.py
+++ b/joblib/externals/loky/backend/queues.py
@@ -197,7 +197,7 @@ class SimpleQueue(mp_SimpleQueue):
     def __init__(self, reducers=None, ctx=None):
         super().__init__(ctx=ctx)
 
-        # Add possibility to use custom reducers
+        # Add possiblity to use custom reducers
         self._reducers = reducers
 
     def close(self):
diff --git a/joblib/externals/loky/backend/reduction.py b/joblib/externals/loky/backend/reduction.py
index bed32ba..c6f9297 100644
--- a/joblib/externals/loky/backend/reduction.py
+++ b/joblib/externals/loky/backend/reduction.py
@@ -53,13 +53,12 @@ register(type(_C().f), _reduce_method)
 register(type(_C.h), _reduce_method)
 
 
-if not hasattr(sys, "pypy_version_info"):
-    # PyPy uses functions instead of method_descriptors and wrapper_descriptors
-    def _reduce_method_descriptor(m):
-        return getattr, (m.__objclass__, m.__name__)
+def _reduce_method_descriptor(m):
+    return getattr, (m.__objclass__, m.__name__)
 
-    register(type(list.append), _reduce_method_descriptor)
-    register(type(int.__add__), _reduce_method_descriptor)
+
+register(type(list.append), _reduce_method_descriptor)
+register(type(int.__add__), _reduce_method_descriptor)
 
 
 # Make partial func pickable
diff --git a/joblib/externals/loky/backend/resource_tracker.py b/joblib/externals/loky/backend/resource_tracker.py
index 36454cb..6af66de 100644
--- a/joblib/externals/loky/backend/resource_tracker.py
+++ b/joblib/externals/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 resource 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.
@@ -112,7 +119,7 @@ class ResourceTracker:
 
                 warnings.warn(
                     "resource_tracker: process died unexpectedly, "
-                    "relaunching.  Some folders/semaphores might "
+                    "relaunching.  Some folders/sempahores might "
                     "leak."
                 )
 
@@ -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
@@ -238,13 +212,13 @@ def main(fd, verbose=0):
                 if line == b"":  # EOF
                     break
                 try:
-                    split = line.strip().decode("ascii").split(":")
+                    splitted = line.strip().decode("ascii").split(":")
                     # name can potentially contain separator symbols (for
                     # instance folders on Windows)
                     cmd, name, rtype = (
-                        split[0],
-                        ":".join(split[1:-1]),
-                        split[-1],
+                        splitted[0],
+                        ":".join(splitted[1:-1]),
+                        splitted[-1],
                     )
 
                     if cmd == "PROBE":
@@ -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/joblib/externals/loky/backend/spawn.py b/joblib/externals/loky/backend/spawn.py
index 59fc890..d011c39 100644
--- a/joblib/externals/loky/backend/spawn.py
+++ b/joblib/externals/loky/backend/spawn.py
@@ -102,7 +102,7 @@ def get_preparation_data(name, init_main_module=True):
         )
 
         # multiprocessing's resource_tracker must be running before loky
-        # process is created (otherwise the child won't be able to use it if it
+        # process is created (othewise the child won't be able to use it if it
         # is created later on)
         mp_resource_tracker.ensure_running()
         d["mp_tracker_args"] = {
diff --git a/joblib/externals/loky/cloudpickle_wrapper.py b/joblib/externals/loky/cloudpickle_wrapper.py
index 387b07d..099debc 100644
--- a/joblib/externals/loky/cloudpickle_wrapper.py
+++ b/joblib/externals/loky/cloudpickle_wrapper.py
@@ -19,7 +19,7 @@ class CloudpickledObjectWrapper:
         return _reconstruct_wrapper, (_pickled_object, self._keep_wrapper)
 
     def __getattr__(self, attr):
-        # Ensure that the wrapped object can be used seamlessly as the
+        # Ensure that the wrapped object can be used seemlessly as the
         # previous object.
         if attr not in ["_obj", "_keep_wrapper"]:
             return getattr(self._obj, attr)
diff --git a/joblib/externals/loky/process_executor.py b/joblib/externals/loky/process_executor.py
index 2576bf4..3b20ad3 100644
--- a/joblib/externals/loky/process_executor.py
+++ b/joblib/externals/loky/process_executor.py
@@ -58,6 +58,7 @@ Process #1..n:
 __author__ = "Thomas Moreau (thomas.moreau.2010@gmail.com)"
 
 
+import faulthandler
 import os
 import gc
 import sys
@@ -192,7 +193,7 @@ def _python_exit():
     items = list(_threads_wakeups.items())
     if len(items) > 0:
         mp.util.debug(
-            "Interpreter shutting down. Waking up {len(items)}"
+            f"Interpreter shutting down. Waking up {len(items)}"
             f"executor_manager_thread:\n{items}"
         )
 
@@ -205,8 +206,7 @@ def _python_exit():
     # Collect the executor_manager_thread's to make sure we exit cleanly.
     for thread, _ in items:
         # This locks is to prevent situations where an executor is gc'ed in one
-        # thread while the atexit finalizer is running in another thread. This
-        # can happen when joblib is used in pypy for instance.
+        # thread while the atexit finalizer is running in another thread.
         with _global_shutdown_lock:
             thread.join()
 
@@ -306,9 +306,11 @@ class _SafeQueue(Queue):
         pending_work_items=None,
         running_work_items=None,
         thread_wakeup=None,
+        shutdown_lock=None,
         reducers=None,
     ):
         self.thread_wakeup = thread_wakeup
+        self.shutdown_lock = shutdown_lock
         self.pending_work_items = pending_work_items
         self.running_work_items = running_work_items
         super().__init__(max_size, reducers=reducers, ctx=ctx)
@@ -337,7 +339,8 @@ class _SafeQueue(Queue):
             if work_item is not None:
                 work_item.future.set_exception(raised_error)
                 del work_item
-            self.thread_wakeup.wakeup()
+            with self.shutdown_lock:
+                self.thread_wakeup.wakeup()
         else:
             super()._on_queue_feeder_error(e, obj)
 
@@ -375,6 +378,28 @@ def _sendback_result(result_queue, work_id, result=None, exception=None):
         result_queue.put(_ResultItem(work_id, exception=exc))
 
 
+def _enable_faulthandler_if_needed():
+    if "PYTHONFAULTHANDLER" in os.environ:
+        # Respect the environment variable to configure faulthandler. This
+        # makes it possible to never enable faulthandler in the loky workers by
+        # setting PYTHONFAULTHANDLER=0 explicitly in the environment.
+        mp.util.debug(
+            f"faulthandler explicitly configured by environment variable: "
+            f"PYTHONFAULTHANDLER={os.environ['PYTHONFAULTHANDLER']}."
+        )
+    else:
+        if faulthandler.is_enabled():
+            # Fault handler is already enabled, possibly via a custom
+            # initializer to customize the behavior.
+            mp.util.debug("faulthandler already enabled.")
+        else:
+            # Enable faulthandler by default with default paramaters otherwise.
+            mp.util.debug(
+                "Enabling faulthandler to report tracebacks on worker crashes."
+            )
+            faulthandler.enable()
+
+
 def _process_worker(
     call_queue,
     result_queue,
@@ -421,6 +446,8 @@ def _process_worker(
     pid = os.getpid()
 
     mp.util.debug(f"Worker started with timeout={timeout}")
+    _enable_faulthandler_if_needed()
+
     while True:
         try:
             call_item = call_queue.get(block=True, timeout=timeout)
@@ -494,7 +521,7 @@ def _process_worker(
                     # The GC managed to free the memory: everything is fine.
                     continue
 
-                # The process is leaking memory: let the main process
+                # The process is leaking memory: let the master process
                 # know that we need to start a new worker.
                 mp.util.info("Memory leak detected: shutting down worker")
                 result_queue.put(pid)
@@ -710,7 +737,10 @@ class _ExecutorManagerThread(threading.Thread):
                 "terminated. This could be caused by a segmentation fault "
                 "while calling the function or by an excessive memory usage "
                 "causing the Operating System to kill the worker.\n"
-                f"{exit_codes}"
+                f"{exit_codes}\n"
+                "Detailed tracebacks of the workers should have been printed "
+                "to stderr in the executor process if faulthandler was not "
+                "disabled."
             )
 
         self.thread_wakeup.clear()
@@ -983,7 +1013,7 @@ def _check_max_depth(context):
     if 0 < MAX_DEPTH and _CURRENT_DEPTH + 1 > MAX_DEPTH:
         raise LokyRecursionError(
             "Could not spawn extra nested processes at depth superior to "
-            f"MAX_DEPTH={MAX_DEPTH}. If this is intended, you can change "
+            f"MAX_DEPTH={MAX_DEPTH}. If this is intendend, you can change "
             "this limit with the LOKY_MAX_DEPTH environment variable."
         )
 
@@ -1014,7 +1044,6 @@ BrokenExecutor = BrokenProcessPool
 
 
 class ShutdownExecutorError(RuntimeError):
-
     """
     Raised when a ProcessPoolExecutor is shutdown while a future was in the
     running or pending state.
@@ -1141,6 +1170,7 @@ class ProcessPoolExecutor(Executor):
             pending_work_items=self._pending_work_items,
             running_work_items=self._running_work_items,
             thread_wakeup=self._executor_manager_thread_wakeup,
+            shutdown_lock=self._shutdown_lock,
             reducers=job_reducers,
             ctx=self._context,
         )
diff --git a/joblib/hashing.py b/joblib/hashing.py
index c8c17bc..2055acf 100644
--- a/joblib/hashing.py
+++ b/joblib/hashing.py
@@ -131,18 +131,22 @@ class Hasher(Pickler):
     # function
     dispatch[type(pickle.dump)] = save_global
 
-    def _batch_setitems(self, items):
+    # We use *args in _batch_setitems signature because _batch_setitems has an
+    # additional 'obj' argument in Python 3.14
+    def _batch_setitems(self, items, *args):
         # forces order of keys in dict to ensure consistent hash.
         try:
             # Trying first to compare dict assuming the type of keys is
             # consistent and orderable.
             # This fails on python 3 when keys are unorderable
             # but we keep it in a try as it's faster.
-            Pickler._batch_setitems(self, iter(sorted(items)))
+            Pickler._batch_setitems(self, iter(sorted(items)), *args)
         except TypeError:
             # If keys are unorderable, sorting them using their hash. This is
             # slower but works in any case.
-            Pickler._batch_setitems(self, iter(sorted((hash(k), v) for k, v in items)))
+            Pickler._batch_setitems(
+                self, iter(sorted((hash(k), v) for k, v in items)), *args
+            )
 
     def save_set(self, set_items):
         # forces order of items in Set to ensure consistent hash
diff --git a/joblib/numpy_pickle.py b/joblib/numpy_pickle.py
index 8dc6b7a..46052c7 100644
--- a/joblib/numpy_pickle.py
+++ b/joblib/numpy_pickle.py
@@ -155,7 +155,7 @@ class NumpyArrayWrapper(object):
             ):
                 pickler.file_handle.write(chunk.tobytes("C"))
 
-    def read_array(self, unpickler):
+    def read_array(self, unpickler, ensure_native_byte_order):
         """Read array from unpickler file handle.
 
         This function is an adaptation of the numpy read_array function
@@ -205,8 +205,11 @@ class NumpyArrayWrapper(object):
             else:
                 array.shape = self.shape
 
-        # Detect byte order mismatch and swap as needed.
-        return _ensure_native_byte_order(array)
+        if ensure_native_byte_order:
+            # Detect byte order mismatch and swap as needed.
+            array = _ensure_native_byte_order(array)
+
+        return array
 
     def read_mmap(self, unpickler):
         """Read an array using numpy memmap."""
@@ -250,9 +253,9 @@ class NumpyArrayWrapper(object):
             )
             warnings.warn(message)
 
-        return _ensure_native_byte_order(marray)
+        return marray
 
-    def read(self, unpickler):
+    def read(self, unpickler, ensure_native_byte_order):
         """Read the array corresponding to this wrapper.
 
         Use the unpickler to get all information to correctly read the array.
@@ -260,6 +263,9 @@ class NumpyArrayWrapper(object):
         Parameters
         ----------
         unpickler: NumpyUnpickler
+        ensure_native_byte_order: bool
+            If true, coerce the array to use the native endianness of the
+            host system.
 
         Returns
         -------
@@ -268,9 +274,13 @@ class NumpyArrayWrapper(object):
         """
         # When requested, only use memmap mode if allowed.
         if unpickler.mmap_mode is not None and self.allow_mmap:
+            assert not ensure_native_byte_order, (
+                "Memmaps cannot be coerced to a given byte order, "
+                "this code path is impossible."
+            )
             array = self.read_mmap(unpickler)
         else:
-            array = self.read_array(unpickler)
+            array = self.read_array(unpickler, ensure_native_byte_order)
 
         # Manage array subclass case
         if hasattr(array, "__array_prepare__") and self.subclass not in (
@@ -395,6 +405,9 @@ class NumpyUnpickler(Unpickler):
         The memorymap mode to use for reading numpy arrays.
     file_handle: file_like
         File object to unpickle from.
+    ensure_native_byte_order: bool
+        If True, coerce the array to use the native endianness of the
+        host system.
     filename: str
         Name of the file to unpickle from. It should correspond to file_handle.
         This parameter is required when using mmap_mode.
@@ -405,7 +418,7 @@ class NumpyUnpickler(Unpickler):
 
     dispatch = Unpickler.dispatch.copy()
 
-    def __init__(self, filename, file_handle, mmap_mode=None):
+    def __init__(self, filename, file_handle, ensure_native_byte_order, mmap_mode=None):
         # The next line is for backward compatibility with pickle generated
         # with joblib versions less than 0.10.
         self._dirname = os.path.dirname(filename)
@@ -415,6 +428,7 @@ class NumpyUnpickler(Unpickler):
         # filename is required for numpy mmap mode.
         self.filename = filename
         self.compat_mode = False
+        self.ensure_native_byte_order = ensure_native_byte_order
         Unpickler.__init__(self, self.file_handle)
         try:
             import numpy as np
@@ -444,7 +458,11 @@ class NumpyUnpickler(Unpickler):
             # the end of the unpickling.
             if isinstance(array_wrapper, NDArrayWrapper):
                 self.compat_mode = True
-            self.stack.append(array_wrapper.read(self))
+                _array_payload = array_wrapper.read(self)
+            else:
+                _array_payload = array_wrapper.read(self, self.ensure_native_byte_order)
+
+            self.stack.append(_array_payload)
 
     # Be careful to register our new method.
     dispatch[pickle.BUILD[0]] = load_build
@@ -593,7 +611,7 @@ def dump(value, filename, compress=0, protocol=None):
     return [filename]
 
 
-def _unpickle(fobj, filename="", mmap_mode=None):
+def _unpickle(fobj, ensure_native_byte_order, filename="", mmap_mode=None):
     """Internal unpickling function."""
     # We are careful to open the file handle early and keep it open to
     # avoid race-conditions on renames.
@@ -601,7 +619,9 @@ def _unpickle(fobj, filename="", mmap_mode=None):
     # the case with the old persistence format, moving the directory
     # will create a race when joblib tries to access the companion
     # files.
-    unpickler = NumpyUnpickler(filename, fobj, mmap_mode=mmap_mode)
+    unpickler = NumpyUnpickler(
+        filename, fobj, ensure_native_byte_order, mmap_mode=mmap_mode
+    )
     obj = None
     try:
         obj = unpickler.load()
@@ -628,14 +648,31 @@ def _unpickle(fobj, filename="", mmap_mode=None):
 def load_temporary_memmap(filename, mmap_mode, unlink_on_gc_collect):
     from ._memmapping_reducer import JOBLIB_MMAPS, add_maybe_unlink_finalizer
 
-    obj = load(filename, mmap_mode)
+    with open(filename, "rb") as f:
+        with _validate_fileobject_and_memmap(f, filename, mmap_mode) as (
+            fobj,
+            validated_mmap_mode,
+        ):
+            # Memmap are used for interprocess communication, which should
+            # keep the objects untouched. We pass `ensure_native_byte_order=False`
+            # to remain consistent with the loading behavior of non-memmaped arrays
+            # in workers, where the byte order is preserved.
+            # Note that we do not implement endianness change for memmaps, as this
+            # would result in inconsistent behavior.
+            obj = _unpickle(
+                fobj,
+                ensure_native_byte_order=False,
+                filename=filename,
+                mmap_mode=validated_mmap_mode,
+            )
+
     JOBLIB_MMAPS.add(obj.filename)
     if unlink_on_gc_collect:
         add_maybe_unlink_finalizer(obj)
     return obj
 
 
-def load(filename, mmap_mode=None):
+def load(filename, mmap_mode=None, ensure_native_byte_order="auto"):
     """Reconstruct a Python object from a file persisted with joblib.dump.
 
     Read more in the :ref:`User Guide <persistence>`.
@@ -653,6 +690,12 @@ def load(filename, mmap_mode=None):
         mode has no effect for compressed files. Note that in this
         case the reconstructed object might no longer match exactly
         the originally pickled object.
+    ensure_native_byte_order: bool, or 'auto', default=='auto'
+        If True, ensures that the byte order of the loaded arrays matches the
+        native byte ordering (or _endianness_) of the host system. This is not
+        compatible with memory-mapped arrays and using non-null `mmap_mode`
+        parameter at the same time will raise an error. The default 'auto'
+        parameter is equivalent to True if `mmap_mode` is None, else False.
 
     Returns
     -------
@@ -672,6 +715,15 @@ def load(filename, mmap_mode=None):
     object might not match the original pickled object. Note that if the
     file was saved with compression, the arrays cannot be memmapped.
     """
+    if ensure_native_byte_order == "auto":
+        ensure_native_byte_order = mmap_mode is None
+
+    if ensure_native_byte_order and mmap_mode is not None:
+        raise ValueError(
+            "Native byte ordering can only be enforced if 'mmap_mode' parameter "
+            f"is set to None, but got 'mmap_mode={mmap_mode}' instead."
+        )
+
     if Path is not None and isinstance(filename, Path):
         filename = str(filename)
 
@@ -679,7 +731,7 @@ def load(filename, mmap_mode=None):
         fobj = filename
         filename = getattr(fobj, "name", "")
         with _validate_fileobject_and_memmap(fobj, filename, mmap_mode) as (fobj, _):
-            obj = _unpickle(fobj)
+            obj = _unpickle(fobj, ensure_native_byte_order=ensure_native_byte_order)
     else:
         with open(filename, "rb") as f:
             with _validate_fileobject_and_memmap(f, filename, mmap_mode) as (
@@ -692,5 +744,14 @@ def load(filename, mmap_mode=None):
                     # Joblib so we load it with joblib compatibility function.
                     return load_compatibility(fobj)
 
-                obj = _unpickle(fobj, filename, validated_mmap_mode)
+                # A memory-mapped array has to be mapped with the endianness
+                # it has been written with. Other arrays are coerced to the
+                # native endianness of the host system.
+                obj = _unpickle(
+                    fobj,
+                    ensure_native_byte_order=ensure_native_byte_order,
+                    filename=filename,
+                    mmap_mode=validated_mmap_mode,
+                )
+
     return obj
diff --git a/joblib/parallel.py b/joblib/parallel.py
index cd4d16c..c3ea1c7 100644
--- a/joblib/parallel.py
+++ b/joblib/parallel.py
@@ -175,7 +175,6 @@ def _get_active_backend(
         # context manager or the context manager did not set a backend.
         # create the default backend instance now.
         backend = BACKENDS[DEFAULT_BACKEND](nesting_level=0)
-
         explicit_backend = False
 
     # Try to use the backend set by the user with the context manager.
@@ -300,7 +299,7 @@ class parallel_config:
           overridden with ``TMP``, ``TMPDIR`` or ``TEMP`` environment
           variables, typically ``/tmp`` under Unix operating systems.
 
-    max_nbytes int, str, or None, optional, default='1M'
+    max_nbytes: int, str, or None, optional, default='1M'
         Threshold on the size of arrays passed to the workers that
         triggers automated memory mapping in temp_folder. Can be an int
         in Bytes, or a human-readable string, e.g., '1M' for 1 megabyte.
@@ -542,8 +541,7 @@ class parallel_backend(parallel_config):
 
     See Also
     --------
-    joblib.parallel_config: context manager to change the backend
-        configuration.
+    joblib.parallel_config: context manager to change the backend configuration.
     """
 
     def __init__(
@@ -1084,6 +1082,8 @@ class Parallel(Logger):
         disable memmapping, other modes defined in the numpy.memmap doc:
         https://numpy.org/doc/stable/reference/generated/numpy.memmap.html
         Also, see 'max_nbytes' parameter documentation for more details.
+    backend_kwargs: dict, optional
+        Additional parameters to pass to the backend `configure` method.
 
     Notes
     -----
@@ -1222,6 +1222,7 @@ class Parallel(Logger):
         mmap_mode=default_parallel_config["mmap_mode"],
         prefer=default_parallel_config["prefer"],
         require=default_parallel_config["require"],
+        **backend_kwargs,
     ):
         # Initiate parent Logger class state
         super().__init__()
@@ -1253,28 +1254,31 @@ class Parallel(Logger):
         # Check if we are under a parallel_config or parallel_backend
         # context manager and use the config from the context manager
         # for arguments that are not explicitly set.
-        self._backend_args = {
-            k: _get_config_param(param, context_config, k)
-            for param, k in [
-                (max_nbytes, "max_nbytes"),
-                (temp_folder, "temp_folder"),
-                (mmap_mode, "mmap_mode"),
-                (prefer, "prefer"),
-                (require, "require"),
-                (verbose, "verbose"),
-            ]
+        self._backend_kwargs = {
+            **backend_kwargs,
+            **{
+                k: _get_config_param(param, context_config, k)
+                for param, k in [
+                    (max_nbytes, "max_nbytes"),
+                    (temp_folder, "temp_folder"),
+                    (mmap_mode, "mmap_mode"),
+                    (prefer, "prefer"),
+                    (require, "require"),
+                    (verbose, "verbose"),
+                ]
+            },
         }
 
-        if isinstance(self._backend_args["max_nbytes"], str):
-            self._backend_args["max_nbytes"] = memstr_to_bytes(
-                self._backend_args["max_nbytes"]
+        if isinstance(self._backend_kwargs["max_nbytes"], str):
+            self._backend_kwargs["max_nbytes"] = memstr_to_bytes(
+                self._backend_kwargs["max_nbytes"]
             )
-        self._backend_args["verbose"] = max(0, self._backend_args["verbose"] - 50)
+        self._backend_kwargs["verbose"] = max(0, self._backend_kwargs["verbose"] - 50)
 
         if DEFAULT_MP_CONTEXT is not None:
-            self._backend_args["context"] = DEFAULT_MP_CONTEXT
+            self._backend_kwargs["context"] = DEFAULT_MP_CONTEXT
         elif hasattr(mp, "get_context"):
-            self._backend_args["context"] = mp.get_context()
+            self._backend_kwargs["context"] = mp.get_context()
 
         if backend is default_parallel_config["backend"] or backend is None:
             backend = active_backend
@@ -1289,7 +1293,7 @@ class Parallel(Logger):
             # Make it possible to pass a custom multiprocessing context as
             # backend to change the start method to forkserver or spawn or
             # preload modules on the forkserver helper process.
-            self._backend_args["context"] = backend
+            self._backend_kwargs["context"] = backend
             backend = MultiprocessingBackend(nesting_level=nesting_level)
 
         elif backend not in BACKENDS and backend in MAYBE_AVAILABLE_BACKENDS:
@@ -1372,7 +1376,7 @@ class Parallel(Logger):
         """Build a process or thread pool and return the number of workers"""
         try:
             n_jobs = self._backend.configure(
-                n_jobs=self.n_jobs, parallel=self, **self._backend_args
+                n_jobs=self.n_jobs, parallel=self, **self._backend_kwargs
             )
             if self.timeout is not None and not self._backend.supports_timeout:
                 warnings.warn(
diff --git a/pyproject.toml b/pyproject.toml
index ac5b0e2..332f59b 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -120,7 +120,7 @@ exclude = [
 
 [tool.codespell]
 # Ref: https://github.com/codespell-project/codespell#using-a-config-file
-skip = '.git*,*.svg,venv,*.css,test_func_inspect_special_encoding.py'
+skip = '.git*,*.svg,venv,*.css,test_func_inspect_special_encoding.py,./joblib/externals'
 check-hidden = true
 # ignore-regex = ''
 ignore-words-list = 'fo'


Test patch:
diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
index d774560..ba96689 100644
--- a/.github/workflows/test.yml
+++ b/.github/workflows/test.yml
@@ -85,7 +85,7 @@ jobs:
             os: ubuntu-latest
             PYTHON_VERSION: "3.9"
             JOBLIB_TESTS_DEFAULT_PARALLEL_BACKEND: "threading"
-          - name: python-nogil
+          - name: python-free-threaded
             os: 'ubuntu-latest'
             PYTHON_VERSION: "free-threaded-3.13"
 
diff --git a/continuous_integration/run_tests.sh b/continuous_integration/run_tests.sh
index 95dab0e..a0c0ba0 100755
--- a/continuous_integration/run_tests.sh
+++ b/continuous_integration/run_tests.sh
@@ -3,16 +3,17 @@
 set -xe
 
 echo "Activating test environment:"
+conda activate testenv
+
 if [[ "$PYTHON_VERSION" == free-threaded* ]]; then
-    source testenv/bin/activate
     # This is needed because for now some C extensions have not declared their
     # thread-safety with free-threaded Python, for example numpy and coverage.tracer
     export PYTHON_GIL=0
-else
-    conda activate testenv
 fi
+
 which python
-python -V
+# Show python version and build information (e.g. free-threaded or not)
+python -VV
 python -c "import multiprocessing as mp; print('multiprocessing.cpu_count():', mp.cpu_count())"
 python -c "import joblib; print('joblib.cpu_count():', joblib.cpu_count())"
 
diff --git a/joblib/test/test_memory.py b/joblib/test/test_memory.py
index 985593b..6bc8dcf 100644
--- a/joblib/test/test_memory.py
+++ b/joblib/test/test_memory.py
@@ -217,7 +217,11 @@ def test_parallel_call_cached_function_defined_in_jupyter(tmpdir, call_before_re
                 # MemorizedFunc.__reduce__.
                 Parallel(n_jobs=2)(delayed(cached_f)(i) for i in [1, 2])
                 # Ensure the child process has time to close the file.
-                time.sleep(0.2)
+                # Wait up to 5 seconds for slow CI runs
+                for _ in range(25):
+                    if len(os.listdir(f_cache_directory / "f")) == 3:
+                        break
+                    time.sleep(0.2)  # pragma: no cover
                 assert len(os.listdir(f_cache_directory / "f")) == 3
 
                 cached_f(3)
diff --git a/joblib/test/test_numpy_pickle.py b/joblib/test/test_numpy_pickle.py
index 6c61710..9bbd927 100644
--- a/joblib/test/test_numpy_pickle.py
+++ b/joblib/test/test_numpy_pickle.py
@@ -347,6 +347,46 @@ def test_compressed_pickle_dump_and_load(tmpdir):
             assert result == expected
 
 
+@with_numpy
+def test_memmap_load(tmpdir):
+    little_endian_dtype = np.dtype("<i8")
+    big_endian_dtype = np.dtype(">i8")
+    all_dtypes = (little_endian_dtype, big_endian_dtype)
+
+    le_array = np.arange(5, dtype=little_endian_dtype)
+    be_array = np.arange(5, dtype=big_endian_dtype)
+
+    fname = tmpdir.join("temp.pkl").strpath
+
+    numpy_pickle.dump([le_array, be_array], fname)
+
+    le_array_native_load, be_array_native_load = numpy_pickle.load(
+        fname, ensure_native_byte_order=True
+    )
+
+    assert le_array_native_load.dtype == be_array_native_load.dtype
+    assert le_array_native_load.dtype in all_dtypes
+
+    le_array_nonnative_load, be_array_nonnative_load = numpy_pickle.load(
+        fname, ensure_native_byte_order=False
+    )
+
+    assert le_array_nonnative_load.dtype == le_array.dtype
+    assert be_array_nonnative_load.dtype == be_array.dtype
+
+
+def test_invalid_parameters_raise():
+    expected_msg = (
+        "Native byte ordering can only be enforced if 'mmap_mode' parameter "
+        "is set to None, but got 'mmap_mode=r+' instead."
+    )
+
+    with raises(ValueError, match=re.escape(expected_msg)):
+        numpy_pickle.load(
+            "/path/to/some/dump.pkl", ensure_native_byte_order=True, mmap_mode="r+"
+        )
+
+
 def _check_pickle(filename, expected_list, mmap_mode=None):
     """Helper function to test joblib pickle content.
 
diff --git a/joblib/test/test_parallel.py b/joblib/test/test_parallel.py
index 48c995b..6e4cda7 100644
--- a/joblib/test/test_parallel.py
+++ b/joblib/test/test_parallel.py
@@ -421,7 +421,8 @@ def test_parallel_pickling():
 @with_numpy
 @with_multiprocessing
 @parametrize("byteorder", ["<", ">", "="])
-def test_parallel_byteorder_corruption(byteorder):
+@parametrize("max_nbytes", [1, "1M"])
+def test_parallel_byteorder_corruption(byteorder, max_nbytes):
     def inspect_byteorder(x):
         return x, x.dtype.byteorder
 
@@ -429,7 +430,7 @@ def test_parallel_byteorder_corruption(byteorder):
 
     initial_np_byteorder = x.dtype.byteorder
 
-    result = Parallel(n_jobs=2, backend="loky")(
+    result = Parallel(n_jobs=2, backend="loky", max_nbytes=max_nbytes)(
         delayed(inspect_byteorder)(x) for _ in range(3)
     )
 
@@ -1087,7 +1088,7 @@ def test_dispatch_race_condition(n_tasks, n_jobs, pre_dispatch, batch_size):
 def test_default_mp_context():
     mp_start_method = mp.get_start_method()
     p = Parallel(n_jobs=2, backend="multiprocessing")
-    context = p._backend_args.get("context")
+    context = p._backend_kwargs.get("context")
     start_method = context.get_start_method()
     assert start_method == mp_start_method
 
@@ -1322,11 +1323,6 @@ def test_parallel_with_exhausted_iterator():
     assert Parallel(n_jobs=2)(exhausted_iterator) == []
 
 
-def _cleanup_worker():
-    """Helper function to force gc in each worker."""
-    time.sleep(0.1)
-
-
 def check_memmap(a):
     if not isinstance(a, np.memmap):
         raise TypeError("Expected np.memmap instance, got %r", type(a))
@@ -1528,8 +1524,8 @@ def test_multiple_generator_call_managed(backend, return_as, n_jobs):
 
         # Make sure that the error is raised quickly
         assert time.time() - t_start < 2, (
-            "The error should be raised immediately when submitting a new task"
-            " but it took more than 2s."
+            "The error should be raised immediately when submitting a new task "
+            "but it took more than 2s."
         )
 
     del g
@@ -1640,7 +1636,6 @@ def test_memmapping_leaks(backend, tmpdir):
     # Make sure that the shared memory is cleaned at the end of a call
     p = Parallel(n_jobs=2, max_nbytes=1, backend=backend)
     p(delayed(check_memmap)(a) for a in [np.random.random(10)] * 2)
-    p(delayed(_cleanup_worker)() for _ in range(2))
 
     for _ in range(100):
         if not os.listdir(tmpdir):
@@ -1983,9 +1978,8 @@ def _check_numpy_threadpool_limits():
     # implementation is actually used so as to force its initialization.
     a = np.random.randn(100, 100)
     np.dot(a, a)
-    from threadpoolctl import threadpool_info
-
-    return threadpool_info()
+    threadpoolctl = pytest.importorskip("threadpoolctl")
+    return threadpoolctl.threadpool_info()
 
 
 def _parent_max_num_threads_for(child_module, parent_info):
@@ -2122,3 +2116,123 @@ def test_loky_reuse_workers(n_jobs):
         parallel_call(n_jobs)
         executor = get_reusable_executor(reuse=True)
         assert executor == first_executor
+
+
+def _set_initialized(status):
+    status[os.getpid()] = "initialized"
+
+
+def _check_status(status, n_jobs, wait_workers=False):
+    pid = os.getpid()
+    state = status.get(pid, None)
+    assert state in ("initialized", "started"), (
+        f"worker should have been in initialized state, got {state}"
+    )
+    if not wait_workers:
+        return
+
+    status[pid] = "started"
+    # wait up to 30 seconds for the workers to be initialized
+    deadline = time.time() + 30
+    n_started = len([pid for pid, v in status.items() if v == "started"])
+    while time.time() < deadline and n_started < n_jobs:
+        time.sleep(0.1)
+        n_started = len([pid for pid, v in status.items() if v == "started"])
+
+    if time.time() >= deadline:
+        raise TimeoutError("Waited more than 30s to start all the workers")
+
+    return pid
+
+
+@with_multiprocessing
+@parametrize("n_jobs", [2, 4])
+@parametrize("backend", PROCESS_BACKENDS)
+@parametrize("context", [parallel_config, parallel_backend])
+def test_initializer_context(n_jobs, backend, context):
+    n_jobs = effective_n_jobs(n_jobs)
+    manager = mp.Manager()
+    status = manager.dict()
+
+    # pass the initializer to the backend context
+    with context(
+        backend=backend,
+        n_jobs=n_jobs,
+        initializer=_set_initialized,
+        initargs=(status,),
+    ):
+        # check_status checks that the initializer is correctly call
+        Parallel()(delayed(_check_status)(status, n_jobs) for i in range(100))
+
+
+@with_multiprocessing
+@parametrize("n_jobs", [2, 4])
+@parametrize("backend", PROCESS_BACKENDS)
+def test_initializer_parallel(n_jobs, backend):
+    n_jobs = effective_n_jobs(n_jobs)
+    manager = mp.Manager()
+    status = manager.dict()
+
+    # pass the initializer directly to the Parallel call
+    # check_status checks that the initializer is called in all tasks
+    Parallel(
+        backend=backend,
+        n_jobs=n_jobs,
+        initializer=_set_initialized,
+        initargs=(status,),
+    )(delayed(_check_status)(status, n_jobs) for i in range(100))
+
+
+@with_multiprocessing
+@pytest.mark.parametrize("n_jobs", [2, 4])
+def test_initializer_reused(n_jobs):
+    # Check that it is possible to pass initializer config via the `Parallel`
+    # call directly and the worker are reused when the arguments are the same.
+    n_repetitions = 3
+    n_jobs = effective_n_jobs(n_jobs)
+    manager = mp.Manager()
+    status = manager.dict()
+
+    pids = set()
+    for i in range(n_repetitions):
+        results = Parallel(
+            backend="loky",
+            n_jobs=n_jobs,
+            initializer=_set_initialized,
+            initargs=(status,),
+        )(
+            delayed(_check_status)(status, n_jobs, wait_workers=True)
+            for i in range(n_jobs)
+        )
+        pids = pids.union(set(results))
+    assert len(pids) == n_jobs, (
+        "The workers should be reused when the initializer is the same"
+    )
+
+
+@with_multiprocessing
+@pytest.mark.parametrize("n_jobs", [2, 4])
+def test_initializer_not_reused(n_jobs):
+    # Check that when changing the initializer arguments, each parallel call uses its
+    # own initializer args, independently of the previous calls, hence the loky workers
+    # are not reused.
+    n_repetitions = 3
+    n_jobs = effective_n_jobs(n_jobs)
+    manager = mp.Manager()
+
+    pids = set()
+    for i in range(n_repetitions):
+        status = manager.dict()
+        results = Parallel(
+            backend="loky",
+            n_jobs=n_jobs,
+            initializer=_set_initialized,
+            initargs=(status,),
+        )(
+            delayed(_check_status)(status, n_jobs, wait_workers=True)
+            for i in range(n_jobs)
+        )
+        pids = pids.union(set(results))
+    assert len(pids) == n_repetitions * n_jobs, (
+        "The workers should not be reused when the initializer arguments change"
+    )