Repository: joblib/joblib
Pull request #1704 (2025-04-03 15:26:17)

FIX Fix issue in memmap reducing when base array is 1d
Fix https://github.com/joblib/joblib/issues/1703

It looks like in #1703 case the base array is 1d (hence both Fortran-contiguous and C-contiguous) and the memmap reducing incorrectly reconstruct the array as Fortran order.

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/joblib/joblib/pull/1704?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 `0%` with `1 line` in your changes missing coverage. Please review.
> Project coverage is 70.08%. Comparing base [(`71c2ce7`)](https://app.codecov.io/gh/joblib/joblib/commit/71c2ce7f1b8bebc804ff8e9eb77571bb486ed34d?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=joblib) to head [(`cabe3fe`)](https://app.codecov.io/gh/joblib/joblib/commit/cabe3fe121337f4a01afccea7ad3aeffa2038c68?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/1704?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/\_memmapping\_reducer.py](https://app.codecov.io/gh/joblib/joblib/pull/1704?src=pr&el=tree&filepath=joblib%2F_memmapping_reducer.py&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=joblib#diff-am9ibGliL19tZW1tYXBwaW5nX3JlZHVjZXIucHk=) | 0.00% | [1 Missing :warning: ](https://app.codecov.io/gh/joblib/joblib/pull/1704?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 (71c2ce7) and HEAD (cabe3fe). Click for more details.
> 
> <details><summary>HEAD has 22 uploads less than BASE</summary>
>
>| Flag | BASE (71c2ce7) | HEAD (cabe3fe) |
>|------|------|------|
>||23|1|
></details>

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


```diff
@@             Coverage Diff             @@
##             main    #1704       +/-   ##
===========================================
- Coverage   95.51%   70.08%   -25.43%     
===========================================
  Files          46       46               
  Lines        7775     7773        -2     
===========================================
- Hits         7426     5448     -1978     
- Misses        349     2325     +1976     
```

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/joblib/joblib/pull/1704?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).

<details><summary> :rocket: New features to boost your workflow: </summary>

- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
</details>
lesteve: I pushed a test closer to the original issue #1703, marking this as ready for review.

Patch:
diff --git a/CHANGES.rst b/CHANGES.rst
index a06cbbb..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
 
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/joblib/_memmapping_reducer.py b/joblib/_memmapping_reducer.py
index 3396248..d11ec58 100644
--- a/joblib/_memmapping_reducer.py
+++ b/joblib/_memmapping_reducer.py
@@ -305,7 +305,9 @@ def _reduce_memmap_backed(a, m):
     # offset from the backing memmap
     offset += m.offset
 
-    if m.flags["F_CONTIGUOUS"]:
+    # 1D arrays are both F and C contiguous, so only set the flag in
+    # higher dimensions. See https://github.com/joblib/joblib/pull/1704.
+    if m.ndim > 1 and m.flags["F_CONTIGUOUS"]:
         order = "F"
     else:
         # The backing memmap buffer is necessarily contiguous hence C if not
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/loky/__init__.py b/joblib/externals/loky/__init__.py
index 1517bf8..9a1a1fa 100644
--- a/joblib/externals/loky/__init__.py
+++ b/joblib/externals/loky/__init__.py
@@ -42,4 +42,4 @@ __all__ = [
 ]
 
 
-__version__ = "3.5.1"
+__version__ = "3.5.2"
diff --git a/joblib/externals/loky/backend/popen_loky_posix.py b/joblib/externals/loky/backend/popen_loky_posix.py
index 74395be..74fc3f2 100644
--- a/joblib/externals/loky/backend/popen_loky_posix.py
+++ b/joblib/externals/loky/backend/popen_loky_posix.py
@@ -120,7 +120,7 @@ class Popen:
             reduction._mk_inheritable(tracker_fd)
             self._fds += [child_r, child_w, tracker_fd]
             if sys.version_info >= (3, 8) and os.name == "posix":
-                mp_tracker_fd = prep_data["mp_tracker_args"]["fd"]
+                mp_tracker_fd = prep_data["mp_tracker_fd"]
                 self.duplicate_for_child(mp_tracker_fd)
 
             from .fork_exec import fork_exec
diff --git a/joblib/externals/loky/backend/resource_tracker.py b/joblib/externals/loky/backend/resource_tracker.py
index 6af66de..c0176da 100644
--- a/joblib/externals/loky/backend/resource_tracker.py
+++ b/joblib/externals/loky/backend/resource_tracker.py
@@ -171,6 +171,17 @@ class ResourceTracker(_ResourceTracker):
                 else:
                     os.close(r)
 
+    def __del__(self):
+        # ignore error due to trying to clean up child process which has already been
+        # shutdown on windows See https://github.com/joblib/loky/pull/450
+        # This is only required if __del__ is defined
+        if not hasattr(ResourceTracker, "__del__"):
+            return
+        try:
+            super().__del__()
+        except ChildProcessError:
+            pass
+
 
 _resource_tracker = ResourceTracker()
 ensure_running = _resource_tracker.ensure_running
diff --git a/joblib/externals/loky/backend/spawn.py b/joblib/externals/loky/backend/spawn.py
index d011c39..f1963e9 100644
--- a/joblib/externals/loky/backend/spawn.py
+++ b/joblib/externals/loky/backend/spawn.py
@@ -82,11 +82,10 @@ def get_preparation_data(name, init_main_module=True):
     from .resource_tracker import _resource_tracker
 
     _resource_tracker.ensure_running()
-    d["tracker_args"] = {"pid": _resource_tracker._pid}
     if sys.platform == "win32":
-        d["tracker_args"]["fh"] = msvcrt.get_osfhandle(_resource_tracker._fd)
+        d["tracker_fd"] = msvcrt.get_osfhandle(_resource_tracker._fd)
     else:
-        d["tracker_args"]["fd"] = _resource_tracker._fd
+        d["tracker_fd"] = _resource_tracker._fd
 
     if sys.version_info >= (3, 8) and os.name == "posix":
         # joblib/loky#242: allow loky processes to retrieve the resource
@@ -105,10 +104,7 @@ def get_preparation_data(name, init_main_module=True):
         # 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"] = {
-            "fd": mp_resource_tracker._fd,
-            "pid": mp_resource_tracker._pid,
-        }
+        d["mp_tracker_fd"] = mp_resource_tracker._fd
 
     # Figure out whether to initialise main in the subprocess as a module
     # or through direct execution (or to leave it alone entirely)
@@ -172,23 +168,21 @@ def prepare(data, parent_sentinel=None):
     if "orig_dir" in data:
         process.ORIGINAL_DIR = data["orig_dir"]
 
-    if "mp_tracker_args" in data:
+    if "mp_tracker_fd" in data:
         from multiprocessing.resource_tracker import (
             _resource_tracker as mp_resource_tracker,
         )
 
-        mp_resource_tracker._fd = data["mp_tracker_args"]["fd"]
-        mp_resource_tracker._pid = data["mp_tracker_args"]["pid"]
-    if "tracker_args" in data:
+        mp_resource_tracker._fd = data["mp_tracker_fd"]
+    if "tracker_fd" in data:
         from .resource_tracker import _resource_tracker
 
-        _resource_tracker._pid = data["tracker_args"]["pid"]
         if sys.platform == "win32":
-            handle = data["tracker_args"]["fh"]
+            handle = data["tracker_fd"]
             handle = duplicate(handle, source_process=parent_sentinel)
             _resource_tracker._fd = msvcrt.open_osfhandle(handle, os.O_RDONLY)
         else:
-            _resource_tracker._fd = data["tracker_args"]["fd"]
+            _resource_tracker._fd = data["tracker_fd"]
 
     if "init_main_from_name" in data:
         _fixup_main_from_name(data["init_main_from_name"])
diff --git a/joblib/externals/loky/cloudpickle_wrapper.py b/joblib/externals/loky/cloudpickle_wrapper.py
index 099debc..90e5d10 100644
--- a/joblib/externals/loky/cloudpickle_wrapper.py
+++ b/joblib/externals/loky/cloudpickle_wrapper.py
@@ -54,7 +54,7 @@ def _wrap_objects_when_needed(obj):
             **{
                 k: _wrap_objects_when_needed(v)
                 for k, v in obj.keywords.items()
-            }
+            },
         )
     if callable(obj):
         # Need wrap if the object is a function defined in a local scope of
diff --git a/joblib/numpy_pickle.py b/joblib/numpy_pickle.py
index 46052c7..7869fd0 100644
--- a/joblib/numpy_pickle.py
+++ b/joblib/numpy_pickle.py
@@ -38,6 +38,7 @@ from .numpy_pickle_utils import (
     Unpickler,
     _ensure_native_byte_order,
     _read_bytes,
+    _reconstruct,
     _validate_fileobject_and_memmap,
     _write_fileobject,
 )
@@ -288,9 +289,7 @@ class NumpyArrayWrapper(object):
             unpickler.np.memmap,
         ):
             # We need to reconstruct another subclass
-            new_array = unpickler.np.core.multiarray._reconstruct(
-                self.subclass, (0,), "b"
-            )
+            new_array = _reconstruct(self.subclass, (0,), "b")
             return new_array.__array_prepare__(array)
         else:
             return array
diff --git a/joblib/numpy_pickle_compat.py b/joblib/numpy_pickle_compat.py
index 38289df..5e26c13 100644
--- a/joblib/numpy_pickle_compat.py
+++ b/joblib/numpy_pickle_compat.py
@@ -6,7 +6,12 @@ import pickle
 import zlib
 from io import BytesIO
 
-from .numpy_pickle_utils import _ZFILE_PREFIX, Unpickler, _ensure_native_byte_order
+from .numpy_pickle_utils import (
+    _ZFILE_PREFIX,
+    Unpickler,
+    _ensure_native_byte_order,
+    _reconstruct,
+)
 
 
 def hex_str(an_int):
@@ -115,9 +120,7 @@ class NDArrayWrapper(object):
             unpickler.np.memmap,
         ):
             # We need to reconstruct another subclass
-            new_array = unpickler.np.core.multiarray._reconstruct(
-                self.subclass, (0,), "b"
-            )
+            new_array = _reconstruct(self.subclass, (0,), "b")
             return new_array.__array_prepare__(array)
         else:
             return array
@@ -149,7 +152,7 @@ class ZNDArrayWrapper(NDArrayWrapper):
         # Here we a simply reproducing the unpickling mechanism for numpy
         # arrays
         filename = os.path.join(unpickler._dirname, self.filename)
-        array = unpickler.np.core.multiarray._reconstruct(*self.init_args)
+        array = _reconstruct(*self.init_args)
         with open(filename, "rb") as f:
             data = read_zfile(f)
         state = self.state + (data,)
diff --git a/joblib/numpy_pickle_utils.py b/joblib/numpy_pickle_utils.py
index c241715..4f7840c 100644
--- a/joblib/numpy_pickle_utils.py
+++ b/joblib/numpy_pickle_utils.py
@@ -274,3 +274,18 @@ def _read_bytes(fp, size, error_template="ran out of data"):
         raise ValueError(msg % (error_template, size, len(data)))
     else:
         return data
+
+
+def _reconstruct(*args, **kwargs):
+    # Wrapper for numpy._core.multiarray._reconstruct with backward compat
+    # for numpy 1.X
+    #
+    # XXX: Remove this function when numpy 1.X is not supported anymore
+
+    np_major_version = np.__version__[:2]
+    if np_major_version == "1.":
+        from numpy.core.multiarray import _reconstruct as np_reconstruct
+    elif np_major_version == "2.":
+        from numpy._core.multiarray import _reconstruct as np_reconstruct
+
+    return np_reconstruct(*args, **kwargs)
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(


Test patch:
diff --git a/joblib/test/test_memmapping.py b/joblib/test/test_memmapping.py
index 17de521..7797536 100644
--- a/joblib/test/test_memmapping.py
+++ b/joblib/test/test_memmapping.py
@@ -101,6 +101,10 @@ def test_memmap_based_array_reducing(tmpdir):
     # b is an memmap sliced view on an memmap instance
     b = a[1:-1, 2:-1, 2:4]
 
+    # b2 is a memmap 2d with memmap 1d as base
+    # non-regression test for https://github.com/joblib/joblib/issues/1703
+    b2 = buffer.reshape(10, 50)
+
     # c and d are array views
     c = np.asarray(b)
     d = c.T
@@ -123,6 +127,11 @@ def test_memmap_based_array_reducing(tmpdir):
     assert has_shareable_memory(b_reconstructed)
     assert_array_equal(b_reconstructed, b)
 
+    # Reconstruct memmap 2d with memmap 1d as base
+    b2_reconstructed = reconstruct_array_or_memmap(b2)
+    assert has_shareable_memory(b2_reconstructed)
+    assert_array_equal(b2_reconstructed, b2)
+
     # Reconstruct arrays views on memmap base
     c_reconstructed = reconstruct_array_or_memmap(c)
     assert not isinstance(c_reconstructed, np.memmap)
@@ -1228,16 +1237,44 @@ def test_direct_mmap(tmpdir):
 
     arr = _read_array()
 
-    # this is expected to work and gives the reference
+    # this gives the reference result of the function with an array
     ref = Parallel(n_jobs=2)(delayed(func)(x) for x in [a])
 
-    # now test that it work with the mmap array
+    # now test that it works with the mmap array
     results = Parallel(n_jobs=2)(delayed(func)(x) for x in [arr])
     np.testing.assert_array_equal(results, ref)
 
-    # also test with a mmap array read in the subprocess
-    def worker():
-        return _read_array()
-
-    results = Parallel(n_jobs=2)(delayed(worker)() for _ in range(1))
+    # also test that a mmap array read in the subprocess is correctly returned
+    results = Parallel(n_jobs=2)(delayed(_read_array)() for _ in range(1))
     np.testing.assert_array_equal(results[0], arr)
+
+
+@with_numpy
+@with_multiprocessing
+def test_parallel_memmap2d_as_memmap_1d_base(tmpdir):
+    # non-regression test for https://github.com/joblib/joblib/issues/1703,
+    # where 2D arrays backed by 1D memmap had un-wanted order changes.
+    testfile = str(tmpdir.join("arr2.dat"))
+    a = np.arange(10, dtype="uint8").reshape(5, 2)
+    a.tofile(testfile)
+
+    def _read_array():
+        mm = np.memmap(testfile)
+        return mm.reshape(5, 2)
+
+    def func(x):
+        return x**2
+
+    arr = _read_array()
+
+    # this gives the reference result of the function with an array
+    ref = Parallel(n_jobs=2)(delayed(func)(x) for x in [a])
+
+    # now test that it works with a view on a 1D mmap array
+    results = Parallel(n_jobs=2)(delayed(func)(x) for x in [arr])
+    assert not results[0].flags["F_CONTIGUOUS"]
+    np.testing.assert_array_equal(results, ref)
+
+    # also test that returned memmap arrays are correctly ordered
+    results = Parallel(n_jobs=2)(delayed(_read_array)() for _ in range(1))
+    np.testing.assert_array_equal(results[0], a)
diff --git a/joblib/test/test_numpy_pickle.py b/joblib/test/test_numpy_pickle.py
index 9bbd927..ed32049 100644
--- a/joblib/test/test_numpy_pickle.py
+++ b/joblib/test/test_numpy_pickle.py
@@ -419,7 +419,11 @@ def _check_pickle(filename, expected_list, mmap_mode=None):
             expected_nb_warnings = (
                 expected_nb_deprecation_warnings + expected_nb_user_warnings
             )
-            assert len(warninfo) == expected_nb_warnings
+            assert len(warninfo) == expected_nb_warnings, (
+                "Did not get the expected number of warnings. Expected "
+                f"{expected_nb_warnings} but got warnings: "
+                f"{[w.message for w in warninfo]}"
+            )
 
             deprecation_warnings = [
                 w for w in warninfo if issubclass(w.category, DeprecationWarning)
diff --git a/joblib/test/test_parallel.py b/joblib/test/test_parallel.py
index 0f8a15a..6e4cda7 100644
--- a/joblib/test/test_parallel.py
+++ b/joblib/test/test_parallel.py
@@ -1088,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
 
@@ -1323,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))
@@ -1529,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
@@ -1641,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):
@@ -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"
+    )