Repository: joblib/joblib
Pull request #1392 (2023-02-16 16:54:03)

FEA parallel_config context manager to allow more fine-grained control
Fixes https://github.com/joblib/joblib/issues/912
Fixes https://github.com/joblib/joblib/issues/979
related to https://github.com/joblib/joblib/issues/1376

adds a new context_manager `parallel_config` that should be used in place of `parallel_backend`. It exposes more parameters and does not enforce setting a backend.

It allows to do something like:
```py
from joblib.parallel import parallel_config
from joblib.memory import Memory
from joblib import Parallel, delayed
 
with parallel_config(prefer="threads", n_jobs=3):
    res = Parallel(verbose=100)(delayed(lambda :1)() for _ in range(10))
 
# [Parallel(n_jobs=3)]: Using backend ThreadingBackend with 3 concurrent workers.
# [Parallel(n_jobs=3)]: Done   1 tasks      | elapsed:    0.0s
# [Parallel(n_jobs=3)]: Done   2 tasks      | elapsed:    0.0s
# [Parallel(n_jobs=3)]: Done   3 tasks      | elapsed:    0.0s
# [Parallel(n_jobs=3)]: Done   4 tasks      | elapsed:    0.0s
# [Parallel(n_jobs=3)]: Done   5 tasks      | elapsed:    0.0s
# [Parallel(n_jobs=3)]: Done   6 out of  10 | elapsed:    0.0s remaining:    0.0s
# [Parallel(n_jobs=3)]: Done   7 out of  10 | elapsed:    0.0s remaining:    0.0s
# [Parallel(n_jobs=3)]: Done   8 out of  10 | elapsed:    0.0s remaining:    0.0s
# [Parallel(n_jobs=3)]: Done  10 out of  10 | elapsed:    0.0s remaining:    0.0s
# [Parallel(n_jobs=3)]: Done  10 out of  10 | elapsed:    0.0s finished
```
It's using the threading backend instead of the default loky, with 3 jobs instead of the default 1.

The config parameters of the context are used only if they are not explicitely set in `Parallel`. It makes us forced to use sentinels for the default values, adding a little bit of complexity.

TODO
- [x] raise an error if backend constructor params are set but backend is left to None
- [x] add tests

codecov[bot]: # [Codecov](https://codecov.io/gh/joblib/joblib/pull/1392?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=joblib) Report
Base: **93.94**% // Head: **71.05**% // Decreases project coverage by **`-22.89%`** :warning:
> Coverage data is based on head [(`ad384db`)](https://codecov.io/gh/joblib/joblib/pull/1392?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=joblib) compared to base [(`061d4ad`)](https://codecov.io/gh/joblib/joblib/commit/061d4ad3b5bd3bab597a2b22c0e67376fb1e25c8?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=joblib).
> Patch coverage: 89.47% of modified lines in pull request are covered.

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


```diff
@@             Coverage Diff             @@
##           master    #1392       +/-   ##
===========================================
- Coverage   93.94%   71.05%   -22.89%     
===========================================
  Files          52       52               
  Lines        7328     7187      -141     
===========================================
- Hits         6884     5107     -1777     
- Misses        444     2080     +1636     
```


| [Impacted Files](https://codecov.io/gh/joblib/joblib/pull/1392?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=joblib) | Coverage Δ | |
|---|---|---|
| [joblib/parallel.py](https://codecov.io/gh/joblib/joblib/pull/1392?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=joblib#diff-am9ibGliL3BhcmFsbGVsLnB5) | `82.81% <88.88%> (-13.22%)` | :arrow_down: |
| [joblib/\_utils.py](https://codecov.io/gh/joblib/joblib/pull/1392?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=joblib#diff-am9ibGliL191dGlscy5weQ==) | `100.00% <100.00%> (ø)` | |
| [joblib/test/test\_dask.py](https://codecov.io/gh/joblib/joblib/pull/1392?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=joblib#diff-am9ibGliL3Rlc3QvdGVzdF9kYXNrLnB5) | `3.92% <0.00%> (-87.27%)` | :arrow_down: |
| [joblib/executor.py](https://codecov.io/gh/joblib/joblib/pull/1392?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=joblib#diff-am9ibGliL2V4ZWN1dG9yLnB5) | `30.95% <0.00%> (-69.05%)` | :arrow_down: |
| [joblib/\_dask.py](https://codecov.io/gh/joblib/joblib/pull/1392?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=joblib#diff-am9ibGliL19kYXNrLnB5) | `24.01% <0.00%> (-68.67%)` | :arrow_down: |
| [joblib/test/test\_memmapping.py](https://codecov.io/gh/joblib/joblib/pull/1392?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=joblib#diff-am9ibGliL3Rlc3QvdGVzdF9tZW1tYXBwaW5nLnB5) | `35.30% <0.00%> (-63.95%)` | :arrow_down: |
| [joblib/\_multiprocessing\_helpers.py](https://codecov.io/gh/joblib/joblib/pull/1392?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=joblib#diff-am9ibGliL19tdWx0aXByb2Nlc3NpbmdfaGVscGVycy5weQ==) | `26.92% <0.00%> (-61.54%)` | :arrow_down: |
| [joblib/pool.py](https://codecov.io/gh/joblib/joblib/pull/1392?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=joblib#diff-am9ibGliL3Bvb2wucHk=) | `27.04% <0.00%> (-60.76%)` | :arrow_down: |
| [joblib/\_memmapping\_reducer.py](https://codecov.io/gh/joblib/joblib/pull/1392?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=joblib#diff-am9ibGliL19tZW1tYXBwaW5nX3JlZHVjZXIucHk=) | `38.72% <0.00%> (-57.16%)` | :arrow_down: |
| [joblib/test/test\_module.py](https://codecov.io/gh/joblib/joblib/pull/1392?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=joblib#diff-am9ibGliL3Rlc3QvdGVzdF9tb2R1bGUucHk=) | `52.94% <0.00%> (-47.06%)` | :arrow_down: |
| ... and [29 more](https://codecov.io/gh/joblib/joblib/pull/1392?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=joblib) | |

Help us with your feedback. Take ten seconds to tell us [how you rate us](https://about.codecov.io/nps?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=joblib). Have a feature suggestion? [Share it here.](https://app.codecov.io/gh/feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=joblib)

</details>

[:umbrella: View full report at Codecov](https://codecov.io/gh/joblib/joblib/pull/1392?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=joblib).   
:loudspeaker: Do you have feedback about the report comment? [Let us know in this issue](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=joblib).

ogrisel: @jeremiedbb There are a few conflicts to resolve after the merge of the `return_generator` branch but it does not look too complicated.

Patch:
diff --git a/CHANGES.rst b/CHANGES.rst
index 90f2b61..3decb56 100644
--- a/CHANGES.rst
+++ b/CHANGES.rst
@@ -40,6 +40,12 @@ In development
   tracebacks and more efficient running time.
   https://github.com/joblib/joblib/pull/1393
 
+- Add the `parallel_config` context manager to allow for more fine-grained
+  control over the backend configuration. It should be used in place of the
+  `parallel_backend` context manager. In particular, it has the advantage
+  of not requiring to set a specific backend in the context manager.
+  https://github.com/joblib/joblib/pull/1392
+
 Release 1.2.0
 -------------
 
diff --git a/joblib/__init__.py b/joblib/__init__.py
index 540511c..bc34d81 100644
--- a/joblib/__init__.py
+++ b/joblib/__init__.py
@@ -122,6 +122,7 @@ from .parallel import delayed
 from .parallel import cpu_count
 from .parallel import register_parallel_backend
 from .parallel import parallel_backend
+from .parallel import parallel_config
 from .parallel import effective_n_jobs
 from ._cloudpickle_wrapper import wrap_non_picklable_objects
 
@@ -130,7 +131,7 @@ __all__ = ['Memory', 'MemorizedResult', 'PrintTime', 'Logger', 'hash', 'dump',
            'load', 'Parallel', 'delayed', 'cpu_count', 'effective_n_jobs',
            'register_parallel_backend', 'parallel_backend',
            'register_store_backend', 'register_compressor',
-           'wrap_non_picklable_objects']
+           'wrap_non_picklable_objects', 'parallel_config']
 
 
 # Workaround issue discovered in intel-openmp 2019.5:
diff --git a/joblib/_utils.py b/joblib/_utils.py
index 2dbd4f6..7693310 100644
--- a/joblib/_utils.py
+++ b/joblib/_utils.py
@@ -1,6 +1,7 @@
 # Adapted from https://stackoverflow.com/a/9558001/2536294
 
 import ast
+from dataclasses import dataclass
 import operator as op
 
 # supported operators
@@ -42,3 +43,12 @@ def eval_(node):
         return operators[type(node.op)](eval_(node.operand))
     else:
         raise TypeError(node)
+
+
+@dataclass(frozen=True)
+class _Sentinel:
+    """A sentinel to mark a parameter as not explicitly set"""
+    default_value: object
+
+    def __repr__(self):
+        return f"default({self.default_value!r})"
diff --git a/joblib/parallel.py b/joblib/parallel.py
index 8d38b5a..a4700a7 100644
--- a/joblib/parallel.py
+++ b/joblib/parallel.py
@@ -31,7 +31,7 @@ from .disk import memstr_to_bytes
 from ._parallel_backends import (FallbackToBackend, MultiprocessingBackend,
                                  ThreadingBackend, SequentialBackend,
                                  LokyBackend)
-from ._utils import eval_expr
+from ._utils import eval_expr, _Sentinel
 
 # Make sure that those two classes are part of the public joblib.parallel API
 # so that 3rd party backend implementers can import them from here.
@@ -72,9 +72,6 @@ DEFAULT_THREAD_BACKEND = 'threading'
 # manager
 _backend = threading.local()
 
-VALID_BACKEND_HINTS = ('processes', 'threads', None)
-VALID_BACKEND_CONSTRAINTS = ('sharedmem', None)
-
 
 def _register_dask():
     """ Register Dask Backend if called with parallel_backend("dask") """
@@ -94,23 +91,91 @@ EXTERNAL_BACKENDS = {
 }
 
 
-def get_active_backend(prefer=None, require=None, verbose=0):
+# Sentinels for the default values of the Parallel constructor and
+# the parallel_config and parallel_backend context managers
+default_parallel_config = {
+    "n_jobs": _Sentinel(default_value=None),
+    "verbose": _Sentinel(default_value=0),
+    "temp_folder": _Sentinel(default_value=None),
+    "max_nbytes": _Sentinel(default_value="1M"),
+    "mmap_mode": _Sentinel(default_value="r"),
+    "prefer": _Sentinel(default_value=None),
+    "require": _Sentinel(default_value=None),
+}
+
+
+VALID_BACKEND_HINTS = ('processes', 'threads', None)
+VALID_BACKEND_CONSTRAINTS = ('sharedmem', None)
+
+
+def _get_config_param(param, context_config, key):
+    """Return the value of a parallel config parameter
+
+    Explicitly setting it in Parallel has priority over setting in a
+    parallel_(config/backend) context manager.
+    """
+    if param is not default_parallel_config[key]:
+        # param is explicitely set, return it
+        return param
+
+    if context_config is None or key not in context_config:
+        # no context manager just return the default value.
+        return param.default_value
+
+    if context_config[key] is not default_parallel_config[key]:
+        # there's a context manager and the key is set, return it
+        return context_config[key]
+
+    # there's a context manager but the key is not set, return the default
+    return param.default_value
+
+
+def get_active_backend(
+    prefer=default_parallel_config["prefer"],
+    require=default_parallel_config["require"],
+    verbose=default_parallel_config["verbose"],
+):
+    """Return the active default backend"""
+    backend, config = _get_active_backend(prefer, require, verbose)
+    return backend, config["n_jobs"]
+
+
+def _get_active_backend(
+    prefer=default_parallel_config["prefer"],
+    require=default_parallel_config["require"],
+    verbose=default_parallel_config["verbose"],
+):
     """Return the active default backend"""
+    backend_config = getattr(_backend, 'config', None)
+    if backend_config is not None:
+        backend = backend_config['backend']
+        config = {k: v for k, v in backend_config.items() if k != 'backend'}
+    else:
+        backend, config = None, None
+
+    prefer = _get_config_param(prefer, config, "prefer")
+    require = _get_config_param(require, config, "require")
+    verbose = _get_config_param(verbose, config, "verbose")
+
     if prefer not in VALID_BACKEND_HINTS:
-        raise ValueError("prefer=%r is not a valid backend hint, "
-                         "expected one of %r" % (prefer, VALID_BACKEND_HINTS))
+        raise ValueError(
+            f"prefer={prefer} is not a valid backend hint, "
+            f"expected one of {VALID_BACKEND_HINTS}"
+        )
     if require not in VALID_BACKEND_CONSTRAINTS:
-        raise ValueError("require=%r is not a valid backend constraint, "
-                         "expected one of %r"
-                         % (require, VALID_BACKEND_CONSTRAINTS))
-
+        raise ValueError(
+            f"require={require} is not a valid backend constraint, "
+            f"expected one of {VALID_BACKEND_CONSTRAINTS}"
+        )
     if prefer == 'processes' and require == 'sharedmem':
-        raise ValueError("prefer == 'processes' and require == 'sharedmem'"
-                         " are inconsistent settings")
-    backend_and_jobs = getattr(_backend, 'backend_and_jobs', None)
-    if backend_and_jobs is not None:
+        raise ValueError(
+            "prefer == 'processes' and require == 'sharedmem'"
+            " are inconsistent settings"
+        )
+
+    if backend is not None:
         # Try to use the backend set by the user with the context manager.
-        backend, n_jobs = backend_and_jobs
+
         nesting_level = backend.nesting_level
         supports_sharedmem = getattr(backend, 'supports_sharedmem', False)
         if require == 'sharedmem' and not supports_sharedmem:
@@ -123,11 +188,12 @@ def get_active_backend(prefer=None, require=None, verbose=0):
                       "as the latter does not provide shared memory semantics."
                       % (sharedmem_backend.__class__.__name__,
                          backend.__class__.__name__))
-            return sharedmem_backend, DEFAULT_N_JOBS
+            return sharedmem_backend, {"n_jobs": DEFAULT_N_JOBS}
         else:
-            return backend_and_jobs
+            return backend, config
 
-    # We are outside of the scope of any parallel_backend context manager,
+    # We are either outside of the scope of any parallel_(config/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)
     supports_sharedmem = getattr(backend, 'supports_sharedmem', False)
@@ -137,12 +203,247 @@ def get_active_backend(prefer=None, require=None, verbose=0):
         # Make sure the selected default backend match the soft hints and
         # hard constraints:
         backend = BACKENDS[DEFAULT_THREAD_BACKEND](nesting_level=0)
-    return backend, DEFAULT_N_JOBS
 
+    if config is not None:
+        return backend, config
+    return backend, {"n_jobs": DEFAULT_N_JOBS}
+
+
+class parallel_config:
+    """Change the default backend or configuration used by :class:`~Parallel`
 
-class parallel_backend(object):
+    This is an alternative to directly passing keyword arguments to the
+    :class:`~Parallel` class constructor. It is particularly useful when
+    calling into library code that uses joblib internally but does not expose
+    the various parallel configuration arguments in its own API.
+
+    Parameters
+    ----------
+    backend : str or ParallelBackendBase instance, default=None
+        If ``backend`` is a string it must match a previously registered
+        implementation using the :func:`~register_parallel_backend` function.
+
+        By default the following backends are available:
+
+        - 'loky': single-host, process-based parallelism (used by default),
+        - 'threading': single-host, thread-based parallelism,
+        - 'multiprocessing': legacy single-host, process-based parallelism.
+
+        'loky' is recommended to run functions that manipulate Python objects.
+        'threading' is a low-overhead alternative that is most efficient for
+        functions that release the Global Interpreter Lock: e.g. I/O-bound
+        code or CPU-bound code in a few calls to native code that explicitly
+        releases the GIL. Note that on some rare systems (such as pyodide),
+        multiprocessing and loky may not be available, in which case joblib
+        defaults to threading.
+
+        In addition, if the `dask` and `distributed` Python packages are
+        installed, it is possible to use the 'dask' backend for better
+        scheduling of nested parallel calls without over-subscription and
+        potentially distribute parallel calls over a networked cluster of
+        several hosts.
+
+        It is also possible to use the distributed 'ray' backend for
+        distributing the workload to a cluster of nodes. See more details
+        in the Examples section below.
+
+        Alternatively the backend can be passed directly as an instance.
+
+    n_jobs : int, default=None
+        The maximum number of concurrently running jobs, such as the number
+        of Python worker processes when `backend="loky"` or the size of the
+        thread-pool when `backend="threading"`.
+        If -1 all CPUs are used. If 1 is given, no parallel computing code
+        is used at all, which is useful for debugging. For `n_jobs` below -1,
+        (n_cpus + 1 + n_jobs) are used. Thus for `n_jobs=-2`, all
+        CPUs but one are used.
+        `None` is a marker for 'unset' that will be interpreted as `n_jobs=1`
+        in most backends.
+
+    verbose : int, default=0
+        The verbosity level: if non zero, progress messages are
+        printed. Above 50, the output is sent to stdout.
+        The frequency of the messages increases with the verbosity level.
+        If it more than 10, all iterations are reported.
+
+    temp_folder : str, default=None
+        Folder to be used by the pool for memmapping large arrays
+        for sharing memory with worker processes. If None, this will try in
+        order:
+
+        - a folder pointed by the `JOBLIB_TEMP_FOLDER` environment
+          variable,
+        - `/dev/shm` if the folder exists and is writable: this is a
+          RAM disk filesystem available by default on modern Linux
+          distributions,
+        - the default system temporary folder that can be
+          overridden with `TMP`, `TMPDIR` or `TEMP` environment
+          variables, typically `/tmp` under Unix operating systems.
+
+    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.
+        Use None to disable memmapping of large arrays.
+
+    mmap_mode: {None, 'r+', 'r', 'w+', 'c'}, default='r'
+        Memmapping mode for numpy arrays passed to workers. None will
+        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.
+
+    prefer: str in {'processes', 'threads'} or None, default=None
+        Soft hint to choose the default backend.
+        The default process-based backend is 'loky' and the default
+        thread-based backend is 'threading'. Ignored if the ``backend``
+        parameter is specified.
+
+    require: 'sharedmem' or None, default=None
+        Hard constraint to select the backend. If set to 'sharedmem',
+        the selected backend will be single-host and thread-based.
+
+    inner_max_num_threads : int, default=None
+        If not None, overwrites the limit set on the number of threads
+        usable in some third-party library threadpools like OpenBLAS,
+        MKL or OpenMP. This is only used with the `loky` backend.
+
+    backend_params : dict
+        Additional parameters to pass to the backend constructor when
+        backend is a string.
+
+    Notes
+    -----
+    Joblib tries to limit the oversubscription by limiting the number of
+    threads usable in some third-party library threadpools like OpenBLAS, MKL
+    or OpenMP. The default limit in each worker is set to
+    ``max(cpu_count() // effective_n_jobs, 1)`` but this limit can be
+    overwritten with the ``inner_max_num_threads`` argument which will be used
+    to set this limit in the child processes.
+
+    Examples
+    --------
+    >>> from operator import neg
+    >>> with parallel_config(backend='threading'):
+    ...     print(Parallel()(delayed(neg)(i + 1) for i in range(5)))
+    ...
+    [-1, -2, -3, -4, -5]
+
+    To use the 'ray' joblib backend add the following lines::
+
+    >>> from ray.util.joblib import register_ray  # doctest: +SKIP
+    >>> register_ray()  # doctest: +SKIP
+    >>> with parallel_backend("ray"):  # doctest: +SKIP
+    ...     print(Parallel()(delayed(neg)(i + 1) for i in range(5)))
+    [-1, -2, -3, -4, -5]
+
+    .. versionadded:: 1.3
+    """
+    def __init__(
+        self,
+        backend=None,
+        *,
+        n_jobs=default_parallel_config["n_jobs"],
+        verbose=default_parallel_config["verbose"],
+        temp_folder=default_parallel_config["temp_folder"],
+        max_nbytes=default_parallel_config["max_nbytes"],
+        mmap_mode=default_parallel_config["mmap_mode"],
+        prefer=default_parallel_config["prefer"],
+        require=default_parallel_config["require"],
+        inner_max_num_threads=None,
+        **backend_params
+    ):
+        # Save the parallel info and set the active parallel config
+        self.old_parallel_config = getattr(_backend, "config", None)
+
+        backend = self._check_backend(
+            backend, inner_max_num_threads, **backend_params
+        )
+
+        self.new_parallel_config = {
+            "backend": backend,
+            "n_jobs": n_jobs,
+            "verbose": verbose,
+            "temp_folder": temp_folder,
+            "max_nbytes": max_nbytes,
+            "mmap_mode": mmap_mode,
+            "prefer": prefer,
+            "require": require,
+        }
+        setattr(_backend, "config", self.new_parallel_config)
+
+    def _check_backend(self, backend, inner_max_num_threads, **backend_params):
+        if backend is None:
+            if inner_max_num_threads is not None or len(backend_params) > 0:
+                raise ValueError(
+                    "inner_max_num_threads and other constructor "
+                    "parameters backend_params are only supported "
+                    "when backend is not None."
+                )
+            return None
+
+        if isinstance(backend, str):
+            # Handle non-registered or missing backends
+            if backend not in BACKENDS:
+                if backend in EXTERNAL_BACKENDS:
+                    register = EXTERNAL_BACKENDS[backend]
+                    register()
+                elif backend in MAYBE_AVAILABLE_BACKENDS:
+                    warnings.warn(
+                        f"joblib backend '{backend}' is not available on "
+                        f"your system, falling back to {DEFAULT_BACKEND}.",
+                        UserWarning,
+                        stacklevel=2
+                    )
+                    BACKENDS[backend] = BACKENDS[DEFAULT_BACKEND]
+                else:
+                    raise ValueError(
+                        f"Invalid backend: {backend}, expected one of "
+                        f"{sorted(BACKENDS.keys())}"
+                    )
+
+            backend = BACKENDS[backend](**backend_params)
+
+        if inner_max_num_threads is not None:
+            msg = (
+                f"{backend.__class__.__name__} does not accept setting the "
+                "inner_max_num_threads argument."
+            )
+            assert backend.supports_inner_max_num_threads, msg
+            backend.inner_max_num_threads = inner_max_num_threads
+
+        # If the nesting_level of the backend is not set previously, use the
+        # nesting level from the previous active_backend to set it
+        if backend.nesting_level is None:
+            if self.old_parallel_config is None:
+                nesting_level = 0
+            else:
+                nesting_level = (
+                    self.old_parallel_config["backend"].nesting_level
+                )
+            backend.nesting_level = nesting_level
+
+        return backend
+
+    def __enter__(self):
+        return self.new_parallel_config
+
+    def __exit__(self, type, value, traceback):
+        self.unregister()
+
+    def unregister(self):
+        if self.old_parallel_config is None:
+            if getattr(_backend, "config", None) is not None:
+                delattr(_backend, "config")
+        else:
+            setattr(_backend, "config", self.old_parallel_config)
+
+
+class parallel_backend(parallel_config):
     """Change the default backend used by Parallel inside a with block.
 
+    It is advised to use the `parallel_config` context manager instead,
+    which allows more fine-grained control over the backend configuration.
+
     If ``backend`` is a string it must match a previously registered
     implementation using the :func:`~register_parallel_backend` function.
 
@@ -202,9 +503,6 @@ class parallel_backend(object):
     ...
     [-1, -2, -3, -4, -5]
 
-    Warning: this function is experimental and subject to change in a future
-    version of joblib.
-
     Joblib also tries to limit the oversubscription by limiting the number of
     threads usable in some third-party library threadpools like OpenBLAS, MKL
     or OpenMP. The default limit in each worker is set to
@@ -212,65 +510,38 @@ class parallel_backend(object):
     overwritten with the ``inner_max_num_threads`` argument which will be used
     to set this limit in the child processes.
 
+    See Also
+    --------
+    parallel_config : context manager to change the backend configuration
+
     .. versionadded:: 0.10
 
     """
     def __init__(self, backend, n_jobs=-1, inner_max_num_threads=None,
                  **backend_params):
-        if isinstance(backend, str):
-            if backend not in BACKENDS:
-                if backend in EXTERNAL_BACKENDS:
-                    register = EXTERNAL_BACKENDS[backend]
-                    register()
-                elif backend in MAYBE_AVAILABLE_BACKENDS:
-                    warnings.warn(
-                        f"joblib backend '{backend}' is not available on "
-                        f"your system, falling back to {DEFAULT_BACKEND}.",
-                        UserWarning,
-                        stacklevel=2)
-                    BACKENDS[backend] = BACKENDS[DEFAULT_BACKEND]
-                else:
-                    raise ValueError("Invalid backend: %s, expected one of %r"
-                                     % (backend, sorted(BACKENDS.keys())))
-
-            backend = BACKENDS[backend](**backend_params)
-
-        if inner_max_num_threads is not None:
-            msg = ("{} does not accept setting the inner_max_num_threads "
-                   "argument.".format(backend.__class__.__name__))
-            assert backend.supports_inner_max_num_threads, msg
-            backend.inner_max_num_threads = inner_max_num_threads
-
-        # If the nesting_level of the backend is not set previously, use the
-        # nesting level from the previous active_backend to set it
-        current_backend_and_jobs = getattr(_backend, 'backend_and_jobs', None)
-        if backend.nesting_level is None:
-            if current_backend_and_jobs is None:
-                nesting_level = 0
-            else:
-                nesting_level = current_backend_and_jobs[0].nesting_level
-
-            backend.nesting_level = nesting_level
 
-        # Save the backends info and set the active backend
-        self.old_backend_and_jobs = current_backend_and_jobs
-        self.new_backend_and_jobs = (backend, n_jobs)
+        super().__init__(
+            backend=backend,
+            n_jobs=n_jobs,
+            inner_max_num_threads=inner_max_num_threads,
+            **backend_params
+        )
 
-        _backend.backend_and_jobs = (backend, n_jobs)
+        if self.old_parallel_config is None:
+            self.old_backend_and_jobs = None
+        else:
+            self.old_backend_and_jobs = (
+                self.old_parallel_config["backend"],
+                self.old_parallel_config["n_jobs"],
+            )
+        self.new_backend_and_jobs = (
+            self.new_parallel_config["backend"],
+            self.new_parallel_config["n_jobs"],
+        )
 
     def __enter__(self):
         return self.new_backend_and_jobs
 
-    def __exit__(self, type, value, traceback):
-        self.unregister()
-
-    def unregister(self):
-        if self.old_backend_and_jobs is None:
-            if getattr(_backend, 'backend_and_jobs', None) is not None:
-                del _backend.backend_and_jobs
-        else:
-            _backend.backend_and_jobs = self.old_backend_and_jobs
-
 
 # Under Linux or OS X the default start method of multiprocessing
 # can cause third party libraries to crash. Under Python 3.4+ it is possible
@@ -885,39 +1156,65 @@ class Parallel(Logger):
         [Parallel(n_jobs=2)]: Done 6 out of 6 | elapsed:  0.0s remaining: 0.0s
         [Parallel(n_jobs=2)]: Done 6 out of 6 | elapsed:  0.0s finished
 
-    '''  # noqa: E501
-    def __init__(self, n_jobs=None, backend=None, return_generator=False,
-                 verbose=0, timeout=None, pre_dispatch='2 * n_jobs',
-                 batch_size='auto', temp_folder=None, max_nbytes='1M',
-                 mmap_mode='r', prefer=None, require=None):
-        active_backend, context_n_jobs = get_active_backend(
-            prefer=prefer, require=require, verbose=verbose)
+    '''
+    def __init__(
+        self,
+        n_jobs=default_parallel_config["n_jobs"],
+        backend=None,
+        return_generator=False,
+        verbose=default_parallel_config["verbose"],
+        timeout=None,
+        pre_dispatch='2 * n_jobs',
+        batch_size='auto',
+        temp_folder=default_parallel_config["temp_folder"],
+        max_nbytes=default_parallel_config["max_nbytes"],
+        mmap_mode=default_parallel_config["mmap_mode"],
+        prefer=default_parallel_config["prefer"],
+        require=default_parallel_config["require"],
+    ):
+        active_backend, context_config = _get_active_backend(
+            prefer=prefer, require=require, verbose=verbose
+        )
+
         nesting_level = active_backend.nesting_level
-        if backend is None and n_jobs is None:
+
+        if backend is None and n_jobs is default_parallel_config["n_jobs"]:
             # If we are under a parallel_backend context manager, look up
             # the default number of jobs and use that instead:
-            n_jobs = context_n_jobs
-        if n_jobs is None:
+            n_jobs = context_config["n_jobs"]
+        if n_jobs is None or n_jobs is default_parallel_config["n_jobs"]:
             # No specific context override and no specific value request:
             # default to 1.
             n_jobs = 1
         self.n_jobs = n_jobs
-        self.verbose = verbose
+
+        self.verbose = _get_config_param(verbose, context_config, "verbose")
         self.timeout = timeout
         self.pre_dispatch = pre_dispatch
         self.return_generator = return_generator
 
-        if isinstance(max_nbytes, str):
-            max_nbytes = memstr_to_bytes(max_nbytes)
-
-        self._backend_args = dict(
-            max_nbytes=max_nbytes,
-            mmap_mode=mmap_mode,
-            temp_folder=temp_folder,
-            prefer=prefer,
-            require=require,
-            verbose=max(0, self.verbose - 50),
+        # 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"),
+            ]
+        }
+
+        if isinstance(self._backend_args["max_nbytes"], str):
+            self._backend_args["max_nbytes"] = memstr_to_bytes(
+                self._backend_args["max_nbytes"]
+            )
+        self._backend_args["verbose"] = max(
+            0, self._backend_args["verbose"] - 50
         )
+
         if DEFAULT_MP_CONTEXT is not None:
             self._backend_args['context'] = DEFAULT_MP_CONTEXT
         elif hasattr(mp, "get_context"):


Test patch:
diff --git a/joblib/test/test_parallel.py b/joblib/test/test_parallel.py
index 57b554b..38a8d9b 100644
--- a/joblib/test/test_parallel.py
+++ b/joblib/test/test_parallel.py
@@ -19,7 +19,6 @@ from time import sleep
 from pickle import PicklingError
 from contextlib import nullcontext
 from multiprocessing import TimeoutError
-
 import pytest
 
 import joblib
@@ -63,6 +62,7 @@ from joblib._parallel_backends import LokyBackend
 
 from joblib.parallel import Parallel, delayed
 from joblib.parallel import register_parallel_backend, parallel_backend
+from joblib.parallel import parallel_config
 from joblib.parallel import effective_n_jobs, cpu_count
 
 from joblib.parallel import mp, BACKENDS, DEFAULT_BACKEND, EXTERNAL_BACKENDS
@@ -143,16 +143,17 @@ def test_effective_n_jobs():
     assert effective_n_jobs() > 0
 
 
+@parametrize("context", [parallel_config, parallel_backend])
 @pytest.mark.parametrize(
     "backend_n_jobs, expected_n_jobs",
     [(3, 3), (-1, effective_n_jobs(n_jobs=-1)), (None, 1)],
     ids=["positive-int", "negative-int", "None"]
 )
 @with_multiprocessing
-def test_effective_n_jobs_None(backend_n_jobs, expected_n_jobs):
+def test_effective_n_jobs_None(context, backend_n_jobs, expected_n_jobs):
     # check the number of effective jobs when `n_jobs=None`
     # non-regression test for https://github.com/joblib/joblib/issues/984
-    with parallel_backend("threading", n_jobs=backend_n_jobs):
+    with context("threading", n_jobs=backend_n_jobs):
         # when using a backend, the default of number jobs will be the one set
         # in the backend
         assert effective_n_jobs(n_jobs=None) == expected_n_jobs
@@ -588,14 +589,16 @@ class FakeParallelBackend(SequentialBackend):
 
 
 def test_invalid_backend():
-    with raises(ValueError) as excinfo:
+    with raises(ValueError, match="Invalid backend:"):
         Parallel(backend='unit-testing')
-    assert "Invalid backend:" in str(excinfo.value)
 
-    with raises(ValueError) as excinfo:
+    with raises(ValueError, match="Invalid backend:"):
         with parallel_backend('unit-testing'):
             pass
-    assert "Invalid backend:" in str(excinfo.value)
+
+    with raises(ValueError, match="Invalid backend:"):
+        with parallel_config('unit-testing'):
+            pass
 
 
 @parametrize('backend', ALL_VALID_BACKENDS)
@@ -637,8 +640,8 @@ def test_backend_no_multiprocessing():
         Parallel()(delayed(square)(i) for i in range(3))
 
 
-def check_backend_context_manager(backend_name):
-    with parallel_backend(backend_name, n_jobs=3):
+def check_backend_context_manager(context, backend_name):
+    with context(backend_name, n_jobs=3):
         active_backend, active_n_jobs = parallel.get_active_backend()
         assert active_n_jobs == 3
         assert effective_n_jobs(3) == 3
@@ -666,20 +669,21 @@ all_backends_for_context_manager.extend(
 
 @with_multiprocessing
 @parametrize('backend', all_backends_for_context_manager)
-def test_backend_context_manager(monkeypatch, backend):
+@parametrize('context', [parallel_backend, parallel_config])
+def test_backend_context_manager(monkeypatch, backend, context):
     if backend not in BACKENDS:
         monkeypatch.setitem(BACKENDS, backend, FakeParallelBackend)
 
     assert _active_backend_type() == DefaultBackend
     # check that this possible to switch parallel backends sequentially
-    check_backend_context_manager(backend)
+    check_backend_context_manager(context, backend)
 
     # The default backend is restored
     assert _active_backend_type() == DefaultBackend
 
     # Check that context manager switching is thread safe:
     Parallel(n_jobs=2, backend='threading')(
-        delayed(check_backend_context_manager)(b)
+        delayed(check_backend_context_manager)(context, b)
         for b in all_backends_for_context_manager if not b)
 
     # The default backend is again restored
@@ -695,12 +699,13 @@ class ParameterizedParallelBackend(SequentialBackend):
         self.param = param
 
 
-def test_parameterized_backend_context_manager(monkeypatch):
+@parametrize("context", [parallel_config, parallel_backend])
+def test_parameterized_backend_context_manager(monkeypatch, context):
     monkeypatch.setitem(BACKENDS, 'param_backend',
                         ParameterizedParallelBackend)
     assert _active_backend_type() == DefaultBackend
 
-    with parallel_backend('param_backend', param=42, n_jobs=3):
+    with context('param_backend', param=42, n_jobs=3):
         active_backend, active_n_jobs = parallel.get_active_backend()
         assert type(active_backend) == ParameterizedParallelBackend
         assert active_backend.param == 42
@@ -715,12 +720,13 @@ def test_parameterized_backend_context_manager(monkeypatch):
     assert _active_backend_type() == DefaultBackend
 
 
-def test_directly_parameterized_backend_context_manager():
+@parametrize("context", [parallel_config, parallel_backend])
+def test_directly_parameterized_backend_context_manager(context):
     assert _active_backend_type() == DefaultBackend
 
     # Check that it's possible to pass a backend instance directly,
     # without registration
-    with parallel_backend(ParameterizedParallelBackend(param=43), n_jobs=5):
+    with context(ParameterizedParallelBackend(param=43), n_jobs=5):
         active_backend, active_n_jobs = parallel.get_active_backend()
         assert type(active_backend) == ParameterizedParallelBackend
         assert active_backend.param == 43
@@ -764,11 +770,12 @@ register_parallel_backend('back_compat_backend', MyBackend)
 @with_multiprocessing
 @parametrize('backend', ['threading', 'loky', 'multiprocessing',
                          'back_compat_backend'])
-def test_nested_backend_context_manager(backend):
+@parametrize("context", [parallel_config, parallel_backend])
+def test_nested_backend_context_manager(context, backend):
     # Check that by default, nested parallel calls will always use the
     # ThreadingBackend
 
-    with parallel_backend(backend):
+    with context(backend):
         pid_groups = Parallel(n_jobs=2)(
             delayed(get_nested_pids)()
             for _ in range(10)
@@ -780,7 +787,8 @@ def test_nested_backend_context_manager(backend):
 @with_multiprocessing
 @parametrize('n_jobs', [2, -1, None])
 @parametrize('backend', PARALLEL_BACKENDS)
-def test_nested_backend_in_sequential(backend, n_jobs):
+@parametrize("context", [parallel_config, parallel_backend])
+def test_nested_backend_in_sequential(backend, n_jobs, context):
     # Check that by default, nested parallel calls will always use the
     # ThreadingBackend
 
@@ -799,38 +807,44 @@ def test_nested_backend_in_sequential(backend, n_jobs):
         for _ in range(10)
     )
 
-    with parallel_backend(backend, n_jobs=n_jobs):
+    with context(backend, n_jobs=n_jobs):
         Parallel(n_jobs=1)(
             delayed(check_nested_backend)(backend, n_jobs)
             for _ in range(10)
         )
 
 
-def check_nesting_level(inner_backend, expected_level):
-    with parallel_backend(inner_backend) as (backend, n_jobs):
+def check_nesting_level(context, inner_backend, expected_level):
+    with context(inner_backend) as ctx:
+        if context is parallel_config:
+            backend = ctx["backend"]
+        if context is parallel_backend:
+            backend = ctx[0]
         assert backend.nesting_level == expected_level
 
 
 @with_multiprocessing
 @parametrize('outer_backend', PARALLEL_BACKENDS)
 @parametrize('inner_backend', PARALLEL_BACKENDS)
-def test_backend_nesting_level(outer_backend, inner_backend):
+@parametrize("context", [parallel_config, parallel_backend])
+def test_backend_nesting_level(context, outer_backend, inner_backend):
     # Check that the nesting level for the backend is correctly set
-    check_nesting_level(outer_backend, 0)
+    check_nesting_level(context, outer_backend, 0)
 
     Parallel(n_jobs=2, backend=outer_backend)(
-        delayed(check_nesting_level)(inner_backend, 1)
+        delayed(check_nesting_level)(context, inner_backend, 1)
         for _ in range(10)
     )
 
-    with parallel_backend(inner_backend, n_jobs=2):
-        Parallel()(delayed(check_nesting_level)(inner_backend, 1)
+    with context(inner_backend, n_jobs=2):
+        Parallel()(delayed(check_nesting_level)(context, inner_backend, 1)
                    for _ in range(10))
 
 
 @with_multiprocessing
+@parametrize("context", [parallel_config, parallel_backend])
 @parametrize('with_retrieve_callback', [True, False])
-def test_retrieval_context(with_retrieve_callback):
+def test_retrieval_context(context, with_retrieve_callback):
     import contextlib
 
     class MyBackend(ThreadingBackend):
@@ -847,12 +861,15 @@ def test_retrieval_context(with_retrieve_callback):
     def nested_call(n):
         return Parallel(n_jobs=2)(delayed(id)(i) for i in range(n))
 
-    with parallel_backend("retrieval") as (ba, _):
+    with context("retrieval") as ctx:
         Parallel(n_jobs=2)(
             delayed(nested_call)(i)
             for i in range(5)
         )
-        assert ba.i == 1
+        if context is parallel_config:
+            assert ctx["backend"].i == 1
+        if context is parallel_backend:
+            assert ctx[0].i == 1
 
 
 ###############################################################################
@@ -1440,7 +1457,8 @@ def test_backend_batch_statistics_reset(backend):
 
 
 @with_multiprocessing
-def test_backend_hinting_and_constraints():
+@parametrize("context", [parallel_config, parallel_backend])
+def test_backend_hinting_and_constraints(context):
     for n_jobs in [1, 2, -1]:
         assert type(Parallel(n_jobs=n_jobs)._backend) == DefaultBackend
 
@@ -1458,20 +1476,20 @@ def test_backend_hinting_and_constraints():
     p = Parallel(n_jobs=2, backend='loky', prefer='threads')
     assert type(p._backend) == LokyBackend
 
-    with parallel_backend('loky', n_jobs=2):
+    with context('loky', n_jobs=2):
         # Explicit backend selection by the user with the context manager
         # should be respected when combined with backend hints only.
         p = Parallel(prefer='threads')
         assert type(p._backend) == LokyBackend
         assert p.n_jobs == 2
 
-    with parallel_backend('loky', n_jobs=2):
+    with context('loky', n_jobs=2):
         # Locally hard-coded n_jobs value is respected.
         p = Parallel(n_jobs=3, prefer='threads')
         assert type(p._backend) == LokyBackend
         assert p.n_jobs == 3
 
-    with parallel_backend('loky', n_jobs=2):
+    with context('loky', n_jobs=2):
         # Explicit backend selection by the user with the context manager
         # should be ignored when the Parallel call has hard constraints.
         # In this case, the default backend that supports shared mem is
@@ -1480,13 +1498,16 @@ def test_backend_hinting_and_constraints():
         assert type(p._backend) == ThreadingBackend
         assert p.n_jobs == 1
 
-    with parallel_backend('loky', n_jobs=2):
+    with context('loky', n_jobs=2):
         p = Parallel(n_jobs=3, require='sharedmem')
         assert type(p._backend) == ThreadingBackend
         assert p.n_jobs == 3
 
 
-def test_backend_hinting_and_constraints_with_custom_backends(capsys):
+@parametrize("context", [parallel_config, parallel_backend])
+def test_backend_hinting_and_constraints_with_custom_backends(
+    capsys, context
+):
     # Custom backends can declare that they use threads and have shared memory
     # semantics:
     class MyCustomThreadingBackend(ParallelBackendBase):
@@ -1499,7 +1520,7 @@ def test_backend_hinting_and_constraints_with_custom_backends(capsys):
         def effective_n_jobs(self, n_jobs):
             return n_jobs
 
-    with parallel_backend(MyCustomThreadingBackend()):
+    with context(MyCustomThreadingBackend()):
         p = Parallel(n_jobs=2, prefer='processes')  # ignored
         assert type(p._backend) == MyCustomThreadingBackend
 
@@ -1516,7 +1537,7 @@ def test_backend_hinting_and_constraints_with_custom_backends(capsys):
         def effective_n_jobs(self, n_jobs):
             return n_jobs
 
-    with parallel_backend(MyCustomProcessingBackend()):
+    with context(MyCustomProcessingBackend()):
         p = Parallel(n_jobs=2, prefer='processes')
         assert type(p._backend) == MyCustomProcessingBackend
 
@@ -1559,23 +1580,25 @@ def test_invalid_backend_hinting_and_constraints():
             Parallel(backend='multiprocessing', require='sharedmem')
 
 
-def test_global_parallel_backend():
+@parametrize("context", [parallel_config, parallel_backend])
+def test_global_parallel_backend(context):
     default = Parallel()._backend
 
-    pb = parallel_backend('threading')
+    pb = context('threading')
     assert isinstance(Parallel()._backend, ThreadingBackend)
 
     pb.unregister()
     assert type(Parallel()._backend) is type(default)
 
 
-def test_external_backends():
+@parametrize("context", [parallel_config, parallel_backend])
+def test_external_backends(context):
     def register_foo():
         BACKENDS['foo'] = ThreadingBackend
 
     EXTERNAL_BACKENDS['foo'] = register_foo
 
-    with parallel_backend('foo'):
+    with context('foo'):
         assert isinstance(Parallel()._backend, ThreadingBackend)
 
 
@@ -1593,8 +1616,9 @@ def _recursive_backend_info(limit=3, **kwargs):
 
 @with_multiprocessing
 @parametrize('backend', ['loky', 'threading'])
-def test_nested_parallelism_limit(backend):
-    with parallel_backend(backend, n_jobs=2):
+@parametrize("context", [parallel_config, parallel_backend])
+def test_nested_parallelism_limit(context, backend):
+    with context(backend, n_jobs=2):
         backend_types_and_levels = _recursive_backend_info()
 
     if cpu_count() == 1:
@@ -1616,20 +1640,21 @@ def test_nested_parallelism_limit(backend):
 
 @with_numpy
 @skipif(distributed is None, reason='This test requires dask')
-def test_nested_parallelism_with_dask():
+@parametrize("context", [parallel_config, parallel_backend])
+def test_nested_parallelism_with_dask(context):
     client = distributed.Client(n_workers=2, threads_per_worker=2)  # noqa
 
     # 10 MB of data as argument to trigger implicit scattering
     data = np.ones(int(1e7), dtype=np.uint8)
     for i in range(2):
-        with parallel_backend('dask'):
+        with context('dask'):
             backend_types_and_levels = _recursive_backend_info(data=data)
         assert len(backend_types_and_levels) == 4
         assert all(name == 'DaskDistributedBackend'
                    for name, _ in backend_types_and_levels)
 
     # No argument
-    with parallel_backend('dask'):
+    with context('dask'):
         backend_types_and_levels = _recursive_backend_info()
     assert len(backend_types_and_levels) == 4
     assert all(name == 'DaskDistributedBackend'
@@ -1641,15 +1666,16 @@ def _recursive_parallel(nesting_limit=None):
     return Parallel()(delayed(_recursive_parallel)() for i in range(2))
 
 
+@pytest.mark.no_cover
+@parametrize("context", [parallel_config, parallel_backend])
 @parametrize(
     'backend', (['threading'] if mp is None else ['loky', 'threading'])
 )
-@pytest.mark.no_cover
-def test_thread_bomb_mitigation(backend):
+def test_thread_bomb_mitigation(context, backend):
     # Test that recursive parallelism raises a recursion rather than
     # saturating the operating system resources by creating a unbounded number
     # of threads.
-    with parallel_backend(backend, n_jobs=2):
+    with context(backend, n_jobs=2):
         with raises(BaseException) as excinfo:
             _recursive_parallel()
     exc = excinfo.value
@@ -1697,12 +1723,14 @@ def test_parallel_thread_limit(backend):
 
 @skipif(distributed is not None,
         reason='This test requires dask NOT installed')
-def test_dask_backend_when_dask_not_installed():
+@parametrize("context", [parallel_config, parallel_backend])
+def test_dask_backend_when_dask_not_installed(context):
     with raises(ValueError, match='Please install dask'):
-        parallel_backend('dask')
+        context('dask')
 
 
-def test_zero_worker_backend():
+@parametrize("context", [parallel_config, parallel_backend])
+def test_zero_worker_backend(context):
     # joblib.Parallel should reject with an explicit error message parallel
     # backends that have no worker.
     class ZeroWorkerBackend(ThreadingBackend):
@@ -1716,7 +1744,7 @@ def test_zero_worker_backend():
             return 0
 
     expected_msg = "ZeroWorkerBackend has no active worker"
-    with parallel_backend(ZeroWorkerBackend()):
+    with context(ZeroWorkerBackend()):
         with pytest.raises(RuntimeError, match=expected_msg):
             Parallel(n_jobs=2)(delayed(id)(i) for i in range(2))
 
@@ -1811,7 +1839,10 @@ def test_threadpool_limitation_in_child_loky(n_jobs):
 @with_multiprocessing
 @parametrize('inner_max_num_threads', [1, 2, 4, None])
 @parametrize('n_jobs', [2, -1])
-def test_threadpool_limitation_in_child_context(n_jobs, inner_max_num_threads):
+@parametrize("context", [parallel_config, parallel_backend])
+def test_threadpool_limitation_in_child_context(
+    context, n_jobs, inner_max_num_threads
+):
     # Check that the protection against oversubscription in workers is working
     # using threadpoolctl functionalities.
 
@@ -1820,7 +1851,7 @@ def test_threadpool_limitation_in_child_context(n_jobs, inner_max_num_threads):
     if len(parent_info) == 0:
         pytest.skip(msg="Need a version of numpy linked to BLAS")
 
-    with parallel_backend('loky', inner_max_num_threads=inner_max_num_threads):
+    with context('loky', inner_max_num_threads=inner_max_num_threads):
         workers_threadpool_infos = Parallel(n_jobs=n_jobs)(
             delayed(_check_numpy_threadpool_limits)() for i in range(2))
 
@@ -1839,7 +1870,8 @@ def test_threadpool_limitation_in_child_context(n_jobs, inner_max_num_threads):
 @parametrize('var_name', ["OPENBLAS_NUM_THREADS",
                           "MKL_NUM_THREADS",
                           "OMP_NUM_THREADS"])
-def test_threadpool_limitation_in_child_override(n_jobs, var_name):
+@parametrize("context", [parallel_config, parallel_backend])
+def test_threadpool_limitation_in_child_override(context, n_jobs, var_name):
     # Check that environment variables set by the user on the main process
     # always have the priority.
 
@@ -1858,7 +1890,7 @@ def test_threadpool_limitation_in_child_override(n_jobs, var_name):
             delayed(_get_env)(var_name) for i in range(2))
         assert results == ["4", "4"]
 
-        with parallel_backend('loky', inner_max_num_threads=1):
+        with context('loky', inner_max_num_threads=1):
             results = Parallel(n_jobs=n_jobs)(
                 delayed(_get_env)(var_name) for i in range(2))
         assert results == ["1", "1"]
@@ -1874,10 +1906,11 @@ def test_threadpool_limitation_in_child_override(n_jobs, var_name):
 @with_multiprocessing
 @parametrize('backend', ['multiprocessing', 'threading',
                          MultiprocessingBackend(), ThreadingBackend()])
-def test_threadpool_limitation_in_child_context_error(backend):
+@parametrize("context", [parallel_config, parallel_backend])
+def test_threadpool_limitation_in_child_context_error(context, backend):
 
     with raises(AssertionError, match=r"does not acc.*inner_max_num_threads"):
-        parallel_backend(backend, inner_max_num_threads=1)
+        context(backend, inner_max_num_threads=1)
 
 
 @with_multiprocessing
@@ -1900,3 +1933,48 @@ def test_loky_reuse_workers(n_jobs):
         parallel_call(n_jobs)
         executor = get_reusable_executor(reuse=True)
         assert executor == first_executor
+
+
+@with_numpy
+@with_multiprocessing
+def test_parallel_config_no_backend(tmpdir):
+    # Check that parallel_config allows to change the config
+    # even if no backend is set.
+    with parallel_config(n_jobs=2, max_nbytes=1, temp_folder=tmpdir):
+        with Parallel(prefer="processes") as p:
+            assert isinstance(p._backend, LokyBackend)
+            assert p.n_jobs == 2
+
+            # Checks that memmapping is enabled
+            p(delayed(check_memmap)(a) for a in [np.random.random(10)] * 2)
+            assert len(os.listdir(tmpdir)) > 0
+
+
+@with_numpy
+@with_multiprocessing
+def test_parallel_config_params_explicit_set(tmpdir):
+    with parallel_config(n_jobs=3, max_nbytes=1, temp_folder=tmpdir):
+        with Parallel(n_jobs=2, prefer="processes", max_nbytes='1M') as p:
+            assert isinstance(p._backend, LokyBackend)
+            assert p.n_jobs == 2
+
+            # Checks that memmapping is disabled
+            with raises(TypeError, match="Expected np.memmap instance"):
+                p(delayed(check_memmap)(a) for a in [np.random.random(10)] * 2)
+
+
+@parametrize("param", ["prefer", "require"])
+def test_parallel_config_bad_params(param):
+    # Check that an error is raised when setting a wrong backend
+    # hint or constraint
+    with raises(ValueError, match=f"{param}=wrong is not a valid"):
+        with parallel_config(**{param: "wrong"}):
+            Parallel()
+
+
+def test_parallel_config_constructor_params():
+    # Check that an error is raised when backend is None
+    # but backend constructor params are given
+    with raises(ValueError, match="only supported when backend is not None"):
+        with parallel_config(inner_max_num_threads=1):
+            pass