Repository: joblib/joblib

Issue:
No simple way to pass initializer for process
multiprocessing.Pool has a handy initializer parameter to pass a callable for setting up per-process resources (database connections, loggers, etc) but joblib doesn't expose a way to pass this.

I see in 0.10 I can pass a custom multiprocessing context, which I hope I can use to achieve this, but per-process setup is likely something that many users will likely need, so would be good if there was an easier way.

(I'd suggest an `initializer` parameter to `Parallel` that's picked up by `MultiprocessingBackend`)

Discussion:
@cjw296: ...and sadly, `context` is only available on Python 3.4+, the project I'm working on is Python 2.7 :-(
@cjw296: Hmm, actually, not even sure passing a context will let you do per-process initialisation...
@astromancer: Custom initializers for processes is a must have feature! @gdlmx: For who is searching for a dirty workaround, I wrote a [simple function](https://stackoverflow.com/a/55566003/6238076) to inject additional initializer to `Parallel` instance with a Loky backend. 

```python
def with_initializer(self, f_init):
    hasattr(self._backend, '_workers') or self.__enter__()
    origin_init = self._backend._workers._initializer
    def new_init():
        origin_init()
        f_init()
    self._backend._workers._initializer = new_init if callable(origin_init) else f_init
    return self
```

Example usage:

```python
import matlab
from joblib import Parallel, delayed

x = matlab.double([[0.0]]) # this object can only be loaded after importing matlab

def f(i):
    print(i, x)

def _init_matlab():
    import matlab

with Parallel(4) as para:
    for _ in with_initializer(para, _init_matlab)(delayed(f)(i) for i in range(10)):
        pass
```

Data objects of some complex libraries such as `matlab` can only be loaded after importing the python module. An initializer seems to be the only way to guarantee to load a third party module before the child processes try to unpickle those global data objects. @cwindolf: (See https://stackoverflow.com/questions/55424095/error-pickling-a-matlab-object-in-joblib-parallel-context for context on the above)@shwina: Hi -- is this is possible today? If not, would a PR implementing the following suggestion be welcome?

> (I'd suggest an initializer parameter to Parallel that's picked up by MultiprocessingBackend)

Alternatively, I imagine it's possible to expose the parameter in `joblib.parallel_config`?@jakirkham: What do you think @ogrisel ? 🙂@astromancer: I've been using [this recipe](https://github.com/astromancer/recipes/blob/main/src/recipes/parallel/joblib.py), adapted from the [comment](https://github.com/joblib/joblib/issues/381#issuecomment-480910348) above:

### Code
```python
from joblib._parallel_backends import MultiprocessingBackend, SequentialBackend


def noop(*_, **__):
    """Do nothing."""


def initialized(self, initializer=noop, args=(), **kws):
    """
    Custom per-process process initialization for `joblib.Parallel`.

    Parameters
    ----------
    initializer : callable, optional
        Your process initializer, by default noop, which does nothing.
    args : tuple, optional
        Parameters for your initializer, by default ()

    Returns
    -------
    joblib.Parallel
    """
    if isinstance(self._backend, SequentialBackend):
        return self

    if isinstance(self._backend, MultiprocessingBackend):
        self._backend_args.update(initializer=initializer, initargs=args)
        return self

    if not hasattr(self._backend, '_workers'):
        self.__enter__()

    workers = self._backend._workers
    original_init = workers._initializer

    def new_init():
        if callable(original_init):
            original_init()

        initializer(*args, **kws)

    workers._initializer = new_init

    return self

```
### Usage
```python
import contextlib as ctx
import multiprocessing as mp
from joblib.parallel import Parallel, delayed


class ContextStack(ctx.ExitStack):
    """Manage nested contexts."""

    def __init__(self, contexts=()):
        super().__init__()
        self.contexts = list(contexts)

    def __enter__(self):
        return next(filter(None, map(self.enter_context, self.contexts)), None)

    def add(self, context):
        # assert isinstance(context, ctx.AbstractContextManager)
        self.contexts.append(context)


# ---------------------------------------------------------------------------- #
# main
memory_lock = mp.Lock()


def set_lock(lock):
    # Initialize each process with a global variable lock.
    print('process setup')
    global memory_lock
    memory_lock = lock


def work(*args, **kws):
    print('doing work:', args, kws)


def get_workload():
    yield from range(10)


njobs = 10
worker = delayed(work)
context = ContextStack()
executor = Parallel(njobs, verbose=10)
context.add(initialized(executor, set_lock, (memory_lock, )))
with context as compute:
    compute(worker(data) for data in get_workload())
```
```shell
[Parallel(n_jobs=10)]: Using backend LokyBackend with 10 concurrent workers.
process setup
doing work: (0,) {}
doing work: (1,) {}
doing work: (2,) {}
[Parallel(n_jobs=10)]: Done   3 out of  10 | elapsed:    2.6s remaining:    6.0s
doing work: (3,) {}
doing work: (4,) {}
[Parallel(n_jobs=10)]: Done   5 out of  10 | elapsed:    2.6s remaining:    2.6s
doing work: (5,) {}
doing work: (6,) {}
[Parallel(n_jobs=10)]: Done   7 out of  10 | elapsed:    2.6s remaining:    1.1s
doing work: (7,) {}
process setup
doing work: (8,) {}
doing work: (9,) {}
[Parallel(n_jobs=10)]: Done  10 out of  10 | elapsed:    2.6s finished
process setup
process setup
process setup
process setup
process setup
process setup
process setup
process setup
```
@shwina: Thank you @astromancer , that's slick!

I'm hoping to find  a solution for when I don't have access to the `Parallel()` instance. This is the case when, say you're not using joblib directly, but rather using a library that itself uses `joblib`. 

I suppose monkeypatching `Parallel` is a possibility but as others have said, this would be great to have as part of the "official" joblib API (`parallel_config`).@ogrisel: If someone want to work on a PR with at least support for the most common backends (e.g. loky, multiprocessing and maybe dask), I might find the time to review it.@shwina: Thanks! I'll work on that.@ogrisel: Note however, that contrary to the multiprocessing backend, the loky and dask backends can reuse workers across consecutive calls to different `Parallel` instances. So the worker initialization semantics and the documentation should be aware of that.@Phlos: Hello, is there any news on this PR? I'd be very interested in using the feature <3@ancestor-mithril: Any progress on this issue?@fcharras: I had this issue once. This is how I solved it, maybe you can find this snippet useful:

```
from typing import Any, Callable, override

import joblib  # type: ignore
from joblib.parallel import LokyBackend  # type: ignore


class _LokyBackendWithInitializer(LokyBackend):  # type: ignore
    """
    A specialization of the LokyBackend with an initializer.

    It enables initializing worker processes with custom function calls before
    actually starting the parallel compute.
    """

    @override
    def __init__(self, *, initializer: Callable[[], None], **kwargs: Any):
        super().__init__(**kwargs)
        self.initializer = initializer

    @override
    def configure(self, *args: Any, **kwargs: Any) -> int:  # type: ignore
        return super().configure(*args, initializer=self.initializer, **kwargs)  # type: ignore


result = joblib.Parallel(
    n_jobs,
    backend=_LokyBackendWithInitializer(
        initializer=some_initializer_func
    )
)(...some_task_iterator...)
```

provided you can use the Loky backend.@tomMoral: There is also this PR #1525 implementing this feature directly for both backends.
It would help if someone tests it and make sure it fits the need.

Solution patch:
diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
index d77456034..ba966891c 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/CHANGES.rst b/CHANGES.rst
index 497b6e78d..b7a66c54a 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 fa8d97468..b4b9e8603 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/continuous_integration/run_tests.sh b/continuous_integration/run_tests.sh
index 95dab0ec1..a0c0ba09d 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/doc/memory.rst b/doc/memory.rst
index 9a4760c17..bda47677a 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 cf4a5ce50..309103b3a 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 1bfe04b75..94f353600 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 20ec44a20..53114a851 100644
--- a/joblib/_parallel_backends.py
+++ b/joblib/_parallel_backends.py
@@ -48,10 +48,13 @@ def supports_timeout(self):
 
     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 @@ def retrieve_result(self, out, timeout=None):
             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 @@ def abort_everything(self, ensure_ready=True):
             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 @@ def effective_n_jobs(self, n_jobs):
         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 @@ def configure(
         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 58a8d086f..3768a936e 100644
--- a/joblib/externals/cloudpickle/__init__.py
+++ b/joblib/externals/cloudpickle/__init__.py
@@ -3,7 +3,7 @@
 
 __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 eb43a9676..4d532e5de 100644
--- a/joblib/externals/cloudpickle/cloudpickle.py
+++ b/joblib/externals/cloudpickle/cloudpickle.py
@@ -63,7 +63,7 @@
 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 @@ class id will also reuse this class definition.
     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 52d6732e4..20280f0ca 100644
--- a/joblib/externals/cloudpickle/cloudpickle_fast.py
+++ b/joblib/externals/cloudpickle/cloudpickle_fast.py
@@ -6,6 +6,7 @@
 
 See: tests/test_backward_compat.py
 """
+
 from . import cloudpickle
 
 
diff --git a/joblib/externals/loky/__init__.py b/joblib/externals/loky/__init__.py
index ceb97dc79..1517bf868 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 @@
 ]
 
 
-__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 ea4b8bea2..efd98bf31 100644
--- a/joblib/externals/loky/backend/context.py
+++ b/joblib/externals/loky/backend/context.py
@@ -18,20 +18,15 @@
 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 2353c42f5..f5b7ca691 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 e92ad8220..5afd99b42 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 bed32ba9e..c6f9297c5 100644
--- a/joblib/externals/loky/backend/reduction.py
+++ b/joblib/externals/loky/backend/reduction.py
@@ -53,13 +53,12 @@ def h(cls):
 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 36454cbed..6af66de8b 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 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 @@
 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 @@ def ensure_running(self):
 
                 warnings.warn(
                     "resource_tracker: process died unexpectedly, "
-                    "relaunching.  Some folders/semaphores might "
+                    "relaunching.  Some folders/sempahores might "
                     "leak."
                 )
 
@@ -164,39 +171,6 @@ def ensure_running(self):
                 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 _unlink_resources(rtype_registry, rtype):
         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 59fc89029..d011c3980 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 387b07dd1..099debcb7 100644
--- a/joblib/externals/loky/cloudpickle_wrapper.py
+++ b/joblib/externals/loky/cloudpickle_wrapper.py
@@ -19,7 +19,7 @@ def __reduce__(self):
         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 2576bf4b2..3b20ad312 100644
--- a/joblib/externals/loky/process_executor.py
+++ b/joblib/externals/loky/process_executor.py
@@ -58,6 +58,7 @@
 __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 @@ def __init__(
         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 @@ def _on_queue_feeder_error(self, e, obj):
             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 @@ def wait_result_broken_or_wakeup(self):
                 "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 @@ class TerminatedWorkerError(BrokenProcessPool):
 
 
 class ShutdownExecutorError(RuntimeError):
-
     """
     Raised when a ProcessPoolExecutor is shutdown while a future was in the
     running or pending state.
@@ -1141,6 +1170,7 @@ def _setup_queues(self, job_reducers, result_reducers, queue_size=None):
             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 c8c17bccf..2055acf85 100644
--- a/joblib/hashing.py
+++ b/joblib/hashing.py
@@ -131,18 +131,22 @@ def save_global(self, obj, name=None, pack=struct.pack):
     # 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 8dc6b7a7a..46052c7d5 100644
--- a/joblib/numpy_pickle.py
+++ b/joblib/numpy_pickle.py
@@ -155,7 +155,7 @@ def write_array(self, array, pickler):
             ):
                 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 @@ def read_array(self, unpickler):
             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 @@ def read_mmap(self, unpickler):
             )
             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 @@ def read(self, unpickler):
         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 @@ def read(self, unpickler):
         """
         # 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 @@ def __init__(self, filename, file_handle, mmap_mode=None):
         # 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 @@ def load_build(self):
             # 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 cd4d16c4f..c3ea1c783 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 @@ def __init__(
         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 @@ def __init__(
         # 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 @@ def __init__(
             # 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 @@ def _initialize_backend(self):
         """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 ac5b0e224..332f59be1 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/joblib/test/test_memory.py b/joblib/test/test_memory.py
index 985593bf0..6bc8dcf8f 100644
--- a/joblib/test/test_memory.py
+++ b/joblib/test/test_memory.py
@@ -217,7 +217,11 @@ def f(x):
                 # 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 6c6171040..9bbd92744 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 48c995b3b..6e4cda73f 100644
--- a/joblib/test/test_parallel.py
+++ b/joblib/test/test_parallel.py
@@ -421,7 +421,8 @@ def __reduce__(self):
 @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 inspect_byteorder(x):
 
     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 parallel_call(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"
+    )