Repository: joblib/joblib
Pull request #1457 (2023-06-22 08:48:16)

DOC reference parallel_config instead of parallel_backend
Follow up for #1451 to use `parallel_config` as much as possible in examples and in the tests.

In doing so, I realized that it actually has a different behavior than `parallel_backend`.
This PR does:
- make it possible to specify the `default_n_jobs` per backend (in particular to have default `n_jobs=-1` for dask).
- Fix the nested `parallel_config` usage. Calling the two following examples is equivalent while the first one would give `LokyBackend` with `2` jobs with the current version on `master`.
```python
with parallel_config(backend='threading'):
    with parallel_config(n_jobs=2):
        p = Parallel()

with parallel_config(backend='loky', n_jobs=2):
        p = Parallel()
```


Patch:
diff --git a/.coveragerc b/.coveragerc
deleted file mode 100644
index 136b8b9..0000000
--- a/.coveragerc
+++ /dev/null
@@ -1,8 +0,0 @@
-[run]
-omit =
-  joblib/test/data/*
-  joblib/test/_openmp_test_helper/setup.py
-  */joblib/externals/*
-
-[report]
-show_missing = True
diff --git a/CHANGES.rst b/CHANGES.rst
index 25d3f27..35bc65d 100644
--- a/CHANGES.rst
+++ b/CHANGES.rst
@@ -163,7 +163,7 @@ Release 0.17.0
 Release 0.16.0
 --------------
 
-- Fix a problem in the constructors of of Parallel backends classes that
+- Fix a problem in the constructors of Parallel backends classes that
   inherit from the `AutoBatchingMixin` that prevented the dask backend to
   properly batch short tasks.
   https://github.com/joblib/joblib/pull/1062
@@ -744,7 +744,7 @@ Olivier Grisel
     Make joblib use the 'forkserver' start method by default under Python 3.4+
     to avoid causing crash with 3rd party libraries (such as Apple vecLib /
     Accelerate or the GCC OpenMP runtime) that use an internal thread pool that
-    is not not reinitialized when a ``fork`` system call happens.
+    is not reinitialized when a ``fork`` system call happens.
 
 Olivier Grisel
 
diff --git a/doc/conf.py b/doc/conf.py
index 429a996..4b99e8d 100644
--- a/doc/conf.py
+++ b/doc/conf.py
@@ -39,9 +39,9 @@ autosummary_generate = True
 intersphinx_mapping = {
     'python': ('https://docs.python.org/{.major}'.format(
         sys.version_info), None),
-    'numpy': ('https://docs.scipy.org/doc/numpy/', None),
-    'scipy': ('https://docs.scipy.org/doc/scipy/reference', None),
-    'distributed': ('https://distributed.readthedocs.io/en/latest/', None),
+    'numpy': ('https://numpy.org/doc/stable/', None),
+    'scipy': ('https://docs.scipy.org/doc/scipy', None),
+    'distributed': ('https://distributed.dask.org/en/latest/', None),
 }
 
 # sphinx-gallery configuration
diff --git a/doc/parallel.rst b/doc/parallel.rst
index be70fa3..7da9fce 100644
--- a/doc/parallel.rst
+++ b/doc/parallel.rst
@@ -72,11 +72,11 @@ instead of the default ``"loky"`` backend:
     ...     delayed(sqrt)(i ** 2) for i in range(10))
     [0.0, 1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0]
 
-The :class:`parallel_config` context manager helps selecting
+The :func:`~joblib.parallel_config` context manager helps selecting
 a specific backend implementation or setting the default number of jobs:
 
     >>> from joblib import parallel_config
-    >>> with parallel_config('threading', n_jobs=2):
+    >>> with parallel_config(backend='threading', n_jobs=2):
     ...    Parallel()(delayed(sqrt)(i ** 2) for i in range(10))
     [0.0, 1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0]
 
@@ -89,7 +89,7 @@ In prior versions, the same effect could be achieved by hardcoding a
 specific backend implementation such as ``backend="threading"`` in the
 call to :class:`joblib.Parallel` but this is now considered a bad pattern
 (when done in a library) as it does not make it possible to override that
-choice with the :func:`~joblib.parallel_backend` context manager.
+choice with the :func:`~joblib.parallel_config` context manager.
 
 
 .. topic:: The loky backend may not always be available
@@ -224,27 +224,24 @@ libraries:
 
 Since joblib 0.14, it is also possible to programmatically override the default
 number of threads using the ``inner_max_num_threads`` argument of the
-:func:`~joblib.parallel_backend` function as follows:
+:func:`~joblib.parallel_config` function as follows:
 
 .. code-block:: python
 
-    from joblib import Parallel, delayed, parallel_backend
+    from joblib import Parallel, delayed, parallel_config
 
-    with parallel_backend("loky", inner_max_num_threads=2):
+    with parallel_config(backend="loky", inner_max_num_threads=2):
         results = Parallel(n_jobs=4)(delayed(func)(x, y) for x, y in data)
 
 In this example, 4 Python worker processes will be allowed to use 2 threads
 each, meaning that this program will be able to use up to 8 CPUs concurrently.
 
 
-Custom backend API (experimental)
-=================================
+Custom backend API
+==================
 
 .. versionadded:: 0.10
 
-.. warning:: The custom backend API is experimental and subject to change
-    without going through a deprecation cycle.
-
 User can provide their own implementation of a parallel processing
 backend in addition to the ``'loky'``, ``'threading'``,
 ``'multiprocessing'`` backends provided by default. A backend is
@@ -274,9 +271,10 @@ for a remote cluster computing service::
     register_parallel_backend('custom', MyCustomBackend)
 
 The connection parameters can then be passed to the
-:func:`joblib.parallel_backend` context manager::
+:func:`~joblib.parallel_config` context manager::
 
-    with parallel_backend('custom', endpoint='http://compute', api_key='42'):
+    with parallel_config(backend='custom', endpoint='http://compute',
+                         api_key='42'):
         Parallel()(delayed(some_function)(i) for i in range(10))
 
 Using the context manager can be helpful when using a third-party library that
@@ -288,13 +286,13 @@ A problem exists that external packages that register new parallel backends
 must now be imported explicitly for their backends to be identified by joblib::
 
    >>> import joblib
-   >>> with joblib.parallel_backend('custom'):  # doctest: +SKIP
+   >>> with joblib.parallel_config(backend='custom'):  # doctest: +SKIP
    ...     ...  # this fails
    KeyError: 'custom'
 
    # Import library to register external backend
    >>> import my_custom_backend_library  # doctest: +SKIP
-   >>> with joblib.parallel_backend('custom'):  # doctest: +SKIP
+   >>> with joblib.parallel_config(backend='custom'):  # doctest: +SKIP
    ...     ... # this works
 
 This can be confusing for users.  To resolve this, external packages can
@@ -403,15 +401,12 @@ does not exist (but multiprocessing has more overhead).
 
 .. autofunction:: joblib.delayed
 
-.. autofunction:: joblib.register_parallel_backend
-
-.. autofunction:: joblib.parallel_backend
+.. autofunction:: joblib.parallel_config
+   :noindex:
 
 .. autofunction:: joblib.wrap_non_picklable_objects
 
-.. autofunction:: joblib.parallel.register_parallel_backend
-
-.. autoclass:: joblib.parallel.parallel_config
+.. autofunction:: joblib.register_parallel_backend
 
 .. autoclass:: joblib.parallel.ParallelBackendBase
 
diff --git a/doc/parallel_numpy.rst b/doc/parallel_numpy.rst
index 32f2acf..4c44990 100644
--- a/doc/parallel_numpy.rst
+++ b/doc/parallel_numpy.rst
@@ -28,7 +28,7 @@ worker processes.
 
   The following only applies with the ``"loky"` and
   ``'multiprocessing'`` process-backends. If your code can release the
-  GIL, then using a thread-based backend backend by passing
+  GIL, then using a thread-based backend by passing
   ``prefer='threads'`` is even more efficient because it makes it
   possible to avoid the communication overhead of process-based
   parallelism.
diff --git a/examples/parallel/distributed_backend_simple.py b/examples/parallel/distributed_backend_simple.py
index 91a1b61..6ccd06d 100644
--- a/examples/parallel/distributed_backend_simple.py
+++ b/examples/parallel/distributed_backend_simple.py
@@ -49,7 +49,7 @@ def long_running_function(i):
 ###############################################################################
 # The verbose messages below show that the backend is indeed the
 # dask.distributed one
-with joblib.parallel_backend("dask"):
+with joblib.parallel_config(backend="dask"):
     joblib.Parallel(verbose=100)(
         joblib.delayed(long_running_function)(i) for i in range(10)
     )
diff --git a/examples/serialization_and_wrappers.py b/examples/serialization_and_wrappers.py
index 0b21564..0d8bbf8 100644
--- a/examples/serialization_and_wrappers.py
+++ b/examples/serialization_and_wrappers.py
@@ -15,7 +15,7 @@ import sys
 import time
 import traceback
 from joblib.externals.loky import set_loky_pickler
-from joblib import parallel_backend
+from joblib import parallel_config
 from joblib import Parallel, delayed
 from joblib import wrap_non_picklable_objects
 
@@ -65,7 +65,7 @@ if sys.platform != 'win32':
     def func_async(i, *args):
         return 2 * i
 
-    with parallel_backend('multiprocessing'):
+    with parallel_config('multiprocessing'):
         t_start = time.time()
         Parallel(n_jobs=2)(
             delayed(func_async)(21, large_list) for _ in range(1))
@@ -117,10 +117,10 @@ except Exception:
 
 ###############################################################################
 # To have both fast pickling, safe process creation and serialization of
-# interactive functions, ``loky`` provides a wrapper function
-# ``wrap_non_picklable_objects`` to wrap the non-picklable function and
-# indicate to the serialization process that this specific function should be
-# serialized using ``cloudpickle``. This changes the serialization behavior
+# interactive functions, ``joblib`` provides a wrapper function
+# :func:`~joblib.wrap_non_picklable_objects` to wrap the non-picklable function
+# and indicate to the serialization process that this specific function should
+# be serialized using ``cloudpickle``. This changes the serialization behavior
 # only for this function and keeps using ``pickle`` for all other objects. The
 # drawback of this solution is that it modifies the object. This should not
 # cause many issues with functions but can have side effects with object
diff --git a/joblib/_dask.py b/joblib/_dask.py
index c96c7c7..b7dc1a2 100644
--- a/joblib/_dask.py
+++ b/joblib/_dask.py
@@ -8,7 +8,7 @@ import time
 from uuid import uuid4
 import weakref
 
-from .parallel import parallel_backend
+from .parallel import parallel_config
 from .parallel import AutoBatchingMixin, ParallelBackendBase
 
 try:
@@ -121,7 +121,7 @@ class Batch:
 
     def __call__(self, tasks=None):
         results = []
-        with parallel_backend('dask'):
+        with parallel_config(backend='dask'):
             for func, args, kwargs in tasks:
                 results.append(func(*args, **kwargs))
             return results
@@ -142,6 +142,7 @@ class DaskDistributedBackend(AutoBatchingMixin, ParallelBackendBase):
     MIN_IDEAL_BATCH_DURATION = 0.2
     MAX_IDEAL_BATCH_DURATION = 1.0
     supports_timeout = True
+    default_n_jobs = -1
 
     def __init__(self, scheduler_host=None, scatter=None,
                  client=None, loop=None, wait_for_workers_timeout=10,
@@ -243,14 +244,15 @@ class DaskDistributedBackend(AutoBatchingMixin, ParallelBackendBase):
         # task might cause the cluster to provision some workers.
         try:
             self.client.submit(_joblib_probe_task).result(
-                timeout=self.wait_for_workers_timeout)
+                timeout=self.wait_for_workers_timeout
+            )
         except _TimeoutError as e:
             error_msg = (
                 "DaskDistributedBackend has no worker after {} seconds. "
                 "Make sure that workers are started and can properly connect "
                 "to the scheduler and increase the joblib/dask connection "
                 "timeout with:\n\n"
-                "parallel_backend('dask', wait_for_workers_timeout={})"
+                "parallel_config(backend='dask', wait_for_workers_timeout={})"
             ).format(self.wait_for_workers_timeout,
                      max(10, 2 * self.wait_for_workers_timeout))
             raise TimeoutError(error_msg) from e
diff --git a/joblib/_memmapping_reducer.py b/joblib/_memmapping_reducer.py
index 37f05e2..283399c 100644
--- a/joblib/_memmapping_reducer.py
+++ b/joblib/_memmapping_reducer.py
@@ -530,7 +530,7 @@ class TemporaryResourcesManager(object):
             # It would be safer to not assign a default context id (less silent
             # bugs), but doing this while maintaining backward compatibility
             # with the previous, context-unaware version get_memmaping_executor
-            # exposes exposes too many low-level details.
+            # exposes too many low-level details.
             context_id = uuid4().hex
         self.set_current_context(context_id)
 
diff --git a/joblib/_parallel_backends.py b/joblib/_parallel_backends.py
index 53a7157..22f1345 100644
--- a/joblib/_parallel_backends.py
+++ b/joblib/_parallel_backends.py
@@ -28,6 +28,7 @@ class ParallelBackendBase(metaclass=ABCMeta):
 
     supports_inner_max_num_threads = False
     supports_retrieve_callback = False
+    default_n_jobs = 1
 
     @property
     def supports_return_generator(self):
diff --git a/joblib/executor.py b/joblib/executor.py
index c227748..6837a7d 100644
--- a/joblib/executor.py
+++ b/joblib/executor.py
@@ -78,7 +78,7 @@ class MemmappingExecutor(_ReusablePoolExecutor):
         # finalizer of their shared memmaps. The refcount of those memmaps may
         # be off by an unknown number, so instead of decref'ing them, we force
         # delete the whole temporary folder, and unregister them. There is no
-        # risk of PermissionError at folder deletion because because at this
+        # risk of PermissionError at folder deletion because at this
         # point, all child processes are dead, so all references to temporary
         # memmaps are closed. Otherwise, just try to delete as much as possible
         # with allow_non_empty=True but if we can't, it will be clean up later
diff --git a/joblib/externals/loky/reusable_executor.py b/joblib/externals/loky/reusable_executor.py
index d879da9..ad016fd 100644
--- a/joblib/externals/loky/reusable_executor.py
+++ b/joblib/externals/loky/reusable_executor.py
@@ -83,7 +83,7 @@ def get_reusable_executor(
     the main process. One can provide a dict ``{ENV: VAL}`` where ``ENV`` and
     ``VAL`` are string literals to overwrite the environment variable ``ENV``
     in the child processes to value ``VAL``. The environment variables are set
-    in the children before any module is loaded. This only works with with the
+    in the children before any module is loaded. This only works with the
     ``loky`` context.
     """
     _executor, _ = _ReusablePoolExecutor.get_reusable_executor(
diff --git a/joblib/numpy_pickle.py b/joblib/numpy_pickle.py
index 3c6974a..bf83bb0 100644
--- a/joblib/numpy_pickle.py
+++ b/joblib/numpy_pickle.py
@@ -219,7 +219,7 @@ class NumpyArrayWrapper(object):
                 current_pos % NUMPY_ARRAY_ALIGNMENT_BYTES != 0):
             message = (
                 f'The memmapped array {marray} loaded from the file '
-                f'{unpickler.file_handle.name} is not not bytes aligned. '
+                f'{unpickler.file_handle.name} is not byte aligned. '
                 'This may cause segmentation faults if this memmapped array '
                 'is used in some libraries like BLAS or PyTorch. '
                 'To get rid of this warning, regenerate your pickle file '
@@ -464,7 +464,7 @@ def dump(value, filename, compress=0, protocol=None, cache_size=None):
     -----
     Memmapping on load cannot be used for compressed files. Thus
     using compression can significantly slow down loading. In
-    addition, compressed files take extra extra memory during
+    addition, compressed files take up extra memory during
     dump and load.
 
     """
diff --git a/joblib/parallel.py b/joblib/parallel.py
index cf528dc..4149f02 100644
--- a/joblib/parallel.py
+++ b/joblib/parallel.py
@@ -47,13 +47,11 @@ BACKENDS = {
     'sequential': SequentialBackend,
 }
 # name of the backend used by default by Parallel outside of any context
-# managed by ``parallel_backend``.
+# managed by ``parallel_config`` or ``parallel_backend``.
 
 # threading is the only backend that is always everywhere
 DEFAULT_BACKEND = 'threading'
 
-DEFAULT_N_JOBS = 1
-
 MAYBE_AVAILABLE_BACKENDS = {'multiprocessing', 'loky'}
 
 # if multiprocessing is available, so is loky, we set it as the default
@@ -68,13 +66,13 @@ if mp is not None:
 DEFAULT_THREAD_BACKEND = 'threading'
 
 
-# Thread local value that can be overridden by the ``parallel_backend`` context
+# Thread local value that can be overridden by the ``parallel_config`` context
 # manager
 _backend = threading.local()
 
 
 def _register_dask():
-    """ Register Dask Backend if called with parallel_backend("dask") """
+    """Register Dask Backend if called with parallel_config(backend="dask")"""
     try:
         from ._dask import DaskDistributedBackend
         register_parallel_backend('dask', DaskDistributedBackend)
@@ -94,6 +92,7 @@ EXTERNAL_BACKENDS = {
 # Sentinels for the default values of the Parallel constructor and
 # the parallel_config and parallel_backend context managers
 default_parallel_config = {
+    "backend": _Sentinel(default_value=None),
     "n_jobs": _Sentinel(default_value=None),
     "verbose": _Sentinel(default_value=0),
     "temp_folder": _Sentinel(default_value=None),
@@ -118,15 +117,12 @@ def _get_config_param(param, context_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
+    # Otherwise, we are in the default_parallel_config,
+    # return the default value
     return param.default_value
 
 
@@ -137,7 +133,10 @@ def get_active_backend(
 ):
     """Return the active default backend"""
     backend, config = _get_active_backend(prefer, require, verbose)
-    return backend, config["n_jobs"]
+    n_jobs = _get_config_param(
+        default_parallel_config['n_jobs'], config, "n_jobs"
+    )
+    return backend, n_jobs
 
 
 def _get_active_backend(
@@ -146,16 +145,15 @@ def _get_active_backend(
     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")
+    backend_config = getattr(_backend, "config", default_parallel_config)
+
+    backend = _get_config_param(
+        default_parallel_config['backend'], backend_config, "backend"
+    )
+    prefer = _get_config_param(prefer, backend_config, "prefer")
+    require = _get_config_param(require, backend_config, "require")
+    verbose = _get_config_param(verbose, backend_config, "verbose")
 
     if prefer not in VALID_BACKEND_HINTS:
         raise ValueError(
@@ -173,40 +171,47 @@ def _get_active_backend(
             " are inconsistent settings"
         )
 
-    if backend is not None:
-        # Try to use the backend set by the user with the context manager.
-
-        nesting_level = backend.nesting_level
-        supports_sharedmem = getattr(backend, 'supports_sharedmem', False)
-        if require == 'sharedmem' and not supports_sharedmem:
-            # This backend does not match the shared memory constraint:
-            # fallback to the default thead-based backend.
-            sharedmem_backend = BACKENDS[DEFAULT_THREAD_BACKEND](
-                nesting_level=nesting_level)
-            if verbose >= 10:
-                print("Using %s as joblib.Parallel backend instead of %s "
-                      "as the latter does not provide shared memory semantics."
-                      % (sharedmem_backend.__class__.__name__,
-                         backend.__class__.__name__))
-            return sharedmem_backend, {"n_jobs": DEFAULT_N_JOBS}
-        else:
-            return backend, config
+    explicit_backend = True
+    if backend is None:
 
-    # 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)
+        # 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)
+        explicit_backend = False
+
+    # Try to use the backend set by the user with the context manager.
+
+    nesting_level = backend.nesting_level
     uses_threads = getattr(backend, 'uses_threads', False)
-    if ((require == 'sharedmem' and not supports_sharedmem) or
-            (prefer == 'threads' and not uses_threads)):
-        # Make sure the selected default backend match the soft hints and
-        # hard constraints:
-        backend = BACKENDS[DEFAULT_THREAD_BACKEND](nesting_level=0)
+    supports_sharedmem = getattr(backend, 'supports_sharedmem', False)
+    # Force to use thread-based backend if the provided backend does not
+    # match the shared memory constraint or if the backend is not explicitely
+    # given and threads are prefered.
+    force_threads = (require == 'sharedmem' and not supports_sharedmem)
+    force_threads |= (
+        not explicit_backend and prefer == 'threads' and not uses_threads
+    )
+    if force_threads:
+        # This backend does not match the shared memory constraint:
+        # fallback to the default thead-based backend.
+        sharedmem_backend = BACKENDS[DEFAULT_THREAD_BACKEND](
+            nesting_level=nesting_level
+        )
+        # Warn the user if we forced the backend to thread-based, while the
+        # user explicitely specified a non-thread-based backend.
+        if verbose >= 10 and explicit_backend:
+            print(
+                f"Using {sharedmem_backend.__class__.__name__} as "
+                f"joblib backend instead of {backend.__class__.__name__} "
+                "as the latter does not provide shared memory semantics."
+            )
+        # Force to n_jobs=1 by default
+        thread_config = backend_config.copy()
+        thread_config['n_jobs'] = 1
+        return sharedmem_backend, thread_config
 
-    if config is not None:
-        return backend, config
-    return backend, {"n_jobs": DEFAULT_N_JOBS}
+    return backend, backend_config
 
 
 class parallel_config:
@@ -334,14 +339,14 @@ class parallel_config:
 
     >>> from ray.util.joblib import register_ray  # doctest: +SKIP
     >>> register_ray()  # doctest: +SKIP
-    >>> with parallel_backend("ray"):  # doctest: +SKIP
+    >>> with parallel_config(backend="ray"):  # doctest: +SKIP
     ...     print(Parallel()(delayed(neg)(i + 1) for i in range(5)))
     [-1, -2, -3, -4, -5]
 
     """
     def __init__(
         self,
-        backend=None,
+        backend=default_parallel_config["backend"],
         *,
         n_jobs=default_parallel_config["n_jobs"],
         verbose=default_parallel_config["verbose"],
@@ -354,14 +359,15 @@ class parallel_config:
         **backend_params
     ):
         # Save the parallel info and set the active parallel config
-        self.old_parallel_config = getattr(_backend, "config", None)
+        self.old_parallel_config = getattr(
+            _backend, "config", default_parallel_config
+        )
 
         backend = self._check_backend(
             backend, inner_max_num_threads, **backend_params
         )
 
-        self.new_parallel_config = {
-            "backend": backend,
+        new_config = {
             "n_jobs": n_jobs,
             "verbose": verbose,
             "temp_folder": temp_folder,
@@ -369,18 +375,25 @@ class parallel_config:
             "mmap_mode": mmap_mode,
             "prefer": prefer,
             "require": require,
+            "backend": backend
         }
-        setattr(_backend, "config", self.new_parallel_config)
+        self.parallel_config = self.old_parallel_config.copy()
+        self.parallel_config.update({
+            k: v for k, v in new_config.items()
+            if not isinstance(v, _Sentinel)
+        })
+
+        setattr(_backend, "config", self.parallel_config)
 
     def _check_backend(self, backend, inner_max_num_threads, **backend_params):
-        if backend is None:
+        if backend is default_parallel_config['backend']:
             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
+            return backend
 
         if isinstance(backend, str):
             # Handle non-registered or missing backends
@@ -415,35 +428,30 @@ class parallel_config:
         # 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:
+            parent_backend = self.old_parallel_config['backend']
+            if parent_backend is default_parallel_config['backend']:
                 nesting_level = 0
             else:
-                nesting_level = (
-                    self.old_parallel_config["backend"].nesting_level
-                )
+                nesting_level = parent_backend.nesting_level
             backend.nesting_level = nesting_level
 
         return backend
 
     def __enter__(self):
-        return self.new_parallel_config
+        return self.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)
+        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 :class:`~joblib.parallel.parallel_config` context
-    manager instead, which allows more fine-grained control over the backend
+    It is advised to use the :class:`~joblib.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
@@ -516,8 +524,8 @@ class parallel_backend(parallel_config):
 
     See Also
     --------
-    joblib.parallel.parallel_config : context manager to change the backend
-                                      configuration
+    joblib.parallel_config : context manager to change the backend
+        configuration.
     """
     def __init__(self, backend, n_jobs=-1, inner_max_num_threads=None,
                  **backend_params):
@@ -537,8 +545,8 @@ class parallel_backend(parallel_config):
                 self.old_parallel_config["n_jobs"],
             )
         self.new_backend_and_jobs = (
-            self.new_parallel_config["backend"],
-            self.new_parallel_config["n_jobs"],
+            self.parallel_config["backend"],
+            self.parallel_config["n_jobs"],
         )
 
     def __enter__(self):
@@ -576,7 +584,7 @@ class BatchedCalls(object):
     def __call__(self):
         # Set the default nested backend to self._backend but do not set the
         # change the default number of processes to -1
-        with parallel_backend(self._backend, n_jobs=self._n_jobs):
+        with parallel_config(self._backend, n_jobs=self._n_jobs):
             return [func(*args, **kwargs)
                     for func, args, kwargs in self.items]
 
@@ -938,7 +946,7 @@ class Parallel(Logger):
             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
-            unless the call is performed under a :func:`~parallel_backend`
+            unless the call is performed under a :func:`~parallel_config`
             context manager that sets another value for ``n_jobs``.
         backend: str, ParallelBackendBase instance or None, default: 'loky'
             Specify the parallelization backend implementation.
@@ -966,7 +974,7 @@ class Parallel(Logger):
             :class:`~Parallel` in a library. Instead it is recommended to set
             soft hints (prefer) or hard constraints (require) so as to make it
             possible for library users to change the backend from the outside
-            using the :func:`~parallel_backend` context manager.
+            using the :func:`~parallel_config` context manager.
         return_generator: bool
             If True, calls to this instance will return a generator, yielding
             the results as soon as they are available, in the original order.
@@ -974,7 +982,7 @@ class Parallel(Logger):
             calls to the same Parallel object will result in a ``RuntimeError``
         prefer: str in {'processes', 'threads'} or None, default: None
             Soft hint to choose the default backend if no specific backend
-            was selected with the :func:`~parallel_backend` context manager.
+            was selected with the :func:`~parallel_config` context manager.
             The default process-based backend is 'loky' and the default
             thread-based backend is 'threading'. Ignored if the ``backend``
             parameter is specified.
@@ -982,7 +990,7 @@ class Parallel(Logger):
             Hard constraint to select the backend. If set to 'sharedmem',
             the selected backend will be single-host and thread-based even
             if the user asked for a non-thread based backend with
-            parallel_backend.
+            :func:`~joblib.parallel_config`.
         verbose: int, optional
             The verbosity level: if non zero, progress messages are
             printed. Above 50, the output is sent to stdout.
@@ -1160,7 +1168,7 @@ class Parallel(Logger):
     def __init__(
         self,
         n_jobs=default_parallel_config["n_jobs"],
-        backend=None,
+        backend=default_parallel_config['backend'],
         return_generator=False,
         verbose=default_parallel_config["verbose"],
         timeout=None,
@@ -1178,16 +1186,6 @@ class Parallel(Logger):
 
         nesting_level = active_backend.nesting_level
 
-        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_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 = _get_config_param(verbose, context_config, "verbose")
         self.timeout = timeout
         self.pre_dispatch = pre_dispatch
@@ -1220,7 +1218,7 @@ class Parallel(Logger):
         elif hasattr(mp, "get_context"):
             self._backend_args['context'] = mp.get_context()
 
-        if backend is None:
+        if backend is default_parallel_config['backend'] or backend is None:
             backend = active_backend
 
         elif isinstance(backend, ParallelBackendBase):
@@ -1253,6 +1251,13 @@ class Parallel(Logger):
                                  % (backend, sorted(BACKENDS.keys()))) from e
             backend = backend_factory(nesting_level=nesting_level)
 
+        n_jobs = _get_config_param(n_jobs, context_config, "n_jobs")
+        if n_jobs is None:
+            # No specific context override and no specific value request:
+            # default to the default of the backend.
+            n_jobs = backend.default_n_jobs
+        self.n_jobs = n_jobs
+
         if (require == 'sharedmem' and
                 not getattr(backend, 'supports_sharedmem', False)):
             raise ValueError("Backend %s does not support shared memory"
diff --git a/pyproject.toml b/pyproject.toml
index c6723bf..37ab20c 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -64,3 +64,35 @@ include-package-data = false
 [tool.setuptools.dynamic]
 version = {attr = "joblib.__version__"}
 
+
+[aliases]
+release = "egg_info -RDb ''"
+# Make sure the docs are uploaded when we do an upload
+upload = "upload upload_docs --upload-dir doc/_build/html"
+
+[bdist_rpm]
+doc_files = "doc"
+
+
+[tool.pytest.ini_options]
+doctest_optionflags = [
+    "NORMALIZE_WHITESPACE",
+    "ELLIPSIS"
+]
+addopts = [
+    "--doctest-glob='doc/*.rst'",
+    "--doctest-modules",
+    "--color=yes",
+]
+testpaths = "joblib"
+norecursedirs = "joblib/externals/*"
+
+[tool.coverage.run]
+omit = [
+    "joblib/test/data/*",
+    "joblib/test/_openmp_test_helper/setup.py",
+    "*/joblib/externals/*",
+]
+
+[tool.coverage.report]
+show_missing = true
diff --git a/setup.cfg b/setup.cfg
index f3bc114..56e8f46 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -1,18 +1,5 @@
-[aliases]
-release = egg_info -RDb ''
-# Make sure the docs are uploaded when we do an upload
-upload = upload upload_docs --upload-dir doc/_build/html
-
-[bdist_rpm]
-doc_files = doc
-
-[tool:pytest]
-doctest_optionflags = NORMALIZE_WHITESPACE ELLIPSIS
-addopts =
-    --doctest-glob="doc/*.rst"
-    --doctest-modules
-    --ignore joblib/externals
-testpaths = joblib
+# The prefered config file is pyproject.toml. The use of setup.cfg is
+# mostly for compatibility with flake8 so it should not be used if possible.
 
 [flake8]
 exclude=


Test patch:
diff --git a/joblib/test/test_config.py b/joblib/test/test_config.py
new file mode 100644
index 0000000..e7f6fb6
--- /dev/null
+++ b/joblib/test/test_config.py
@@ -0,0 +1,125 @@
+import os
+
+from joblib.parallel import parallel_config
+from joblib.parallel import parallel_backend
+from joblib.parallel import Parallel, delayed
+
+from joblib.parallel import BACKENDS
+from joblib.parallel import DEFAULT_BACKEND
+from joblib.parallel import EXTERNAL_BACKENDS
+
+from joblib._parallel_backends import LokyBackend
+from joblib._parallel_backends import ThreadingBackend
+from joblib._parallel_backends import MultiprocessingBackend
+
+from joblib.testing import parametrize, raises
+from joblib.test.common import np, with_numpy
+from joblib.test.common import with_multiprocessing
+from joblib.test.test_parallel import check_memmap
+
+
+@parametrize("context", [parallel_config, parallel_backend])
+def test_global_parallel_backend(context):
+    default = Parallel()._backend
+
+    pb = context('threading')
+    try:
+        assert isinstance(Parallel()._backend, ThreadingBackend)
+    finally:
+        pb.unregister()
+    assert type(Parallel()._backend) is type(default)
+
+
+@parametrize("context", [parallel_config, parallel_backend])
+def test_external_backends(context):
+    def register_foo():
+        BACKENDS['foo'] = ThreadingBackend
+
+    EXTERNAL_BACKENDS['foo'] = register_foo
+    try:
+        with context('foo'):
+            assert isinstance(Parallel()._backend, ThreadingBackend)
+    finally:
+        del EXTERNAL_BACKENDS['foo']
+
+
+@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
+
+    with raises(ValueError, match="only supported when backend is not None"):
+        with parallel_config(backend_param=1):
+            pass
+
+
+def test_parallel_config_nested():
+    # Check that nested configuration retrieves the info from the
+    # parent config and do not reset them.
+
+    with parallel_config(n_jobs=2):
+        p = Parallel()
+        assert isinstance(p._backend, BACKENDS[DEFAULT_BACKEND])
+        assert p.n_jobs == 2
+
+    with parallel_config(backend='threading'):
+        with parallel_config(n_jobs=2):
+            p = Parallel()
+            assert isinstance(p._backend, ThreadingBackend)
+            assert p.n_jobs == 2
+
+    with parallel_config(verbose=100):
+        with parallel_config(n_jobs=2):
+            p = Parallel()
+            assert p.verbose == 100
+            assert p.n_jobs == 2
+
+
+@with_numpy
+@with_multiprocessing
+@parametrize('backend', ['multiprocessing', 'threading',
+                         MultiprocessingBackend(), ThreadingBackend()])
+@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"):
+        context(backend, inner_max_num_threads=1)
diff --git a/joblib/test/test_dask.py b/joblib/test/test_dask.py
index 8a02190..1859b37 100644
--- a/joblib/test/test_dask.py
+++ b/joblib/test/test_dask.py
@@ -7,7 +7,7 @@ from random import random
 from uuid import uuid4
 from time import sleep
 
-from .. import Parallel, delayed, parallel_backend
+from .. import Parallel, delayed, parallel_config
 from ..parallel import ThreadingBackend, AutoBatchingMixin
 from .._dask import DaskDistributedBackend
 
@@ -42,7 +42,7 @@ def count_events(event_name, client):
 def test_simple(loop):
     with cluster() as (s, [a, b]):
         with Client(s['address'], loop=loop) as client:  # noqa: F841
-            with parallel_backend('dask') as (ba, _):
+            with parallel_config(backend='dask'):
                 seq = Parallel()(delayed(inc)(i) for i in range(10))
                 assert seq == [inc(i) for i in range(10)]
 
@@ -60,7 +60,7 @@ def test_dask_backend_uses_autobatching(loop):
 
     with cluster() as (s, [a, b]):
         with Client(s['address'], loop=loop) as client:  # noqa: F841
-            with parallel_backend('dask') as (ba, _):
+            with parallel_config(backend='dask'):
                 with Parallel() as parallel:
                     # The backend should be initialized with a default
                     # batch size of 1:
@@ -85,7 +85,7 @@ def random2():
 def test_dont_assume_function_purity(loop):
     with cluster() as (s, [a, b]):
         with Client(s['address'], loop=loop) as client:  # noqa: F841
-            with parallel_backend('dask') as (ba, _):
+            with parallel_config(backend='dask'):
                 x, y = Parallel()(delayed(random2)() for i in range(2))
                 assert x != y
 
@@ -106,7 +106,7 @@ def test_dask_funcname(loop, mixed):
 
     with cluster() as (s, [a, b]):
         with Client(s['address'], loop=loop) as client:
-            with parallel_backend('dask') as (ba, _):
+            with parallel_config(backend='dask'):
                 _ = Parallel(batch_size=2, pre_dispatch='all')(tasks)
 
             def f(dask_scheduler):
@@ -140,7 +140,7 @@ def test_no_undesired_distributed_cache_hit(loop):
     cluster = LocalCluster(n_workers=1, threads_per_worker=2)
     client = Client(cluster)
     try:
-        with parallel_backend('dask') as (ba, _):
+        with parallel_config(backend='dask'):
             # dispatches joblib.parallel.BatchedCalls
             res = Parallel()(
                 delayed(isolated_operation)(list_) for list_ in lists
@@ -157,7 +157,7 @@ def test_no_undesired_distributed_cache_hit(loop):
         assert sum(counts.values()) == 0
         assert all([len(r) == 1 for r in res])
 
-        with parallel_backend('dask') as (ba, _):
+        with parallel_config(backend='dask'):
             # Append a large array which will be scattered by dask, and
             # dispatch joblib._dask.Batch
             res = Parallel()(
@@ -199,7 +199,7 @@ def test_manual_scatter(loop):
 
     with cluster() as (s, [a, b]):
         with Client(s['address'], loop=loop) as client:  # noqa: F841
-            with parallel_backend('dask', scatter=[x, y]) as (ba, _):
+            with parallel_config(backend='dask', scatter=[x, y]):
                 f = delayed(add5)
                 tasks = [f(x, y, z, d=4, e=5),
                          f(x, z, y, d=5, e=4),
@@ -211,7 +211,7 @@ def test_manual_scatter(loop):
 
             # Scatter must take a list/tuple
             with pytest.raises(TypeError):
-                with parallel_backend('dask', loop=loop, scatter=1):
+                with parallel_config(backend='dask', loop=loop, scatter=1):
                     pass
 
     assert results == expected
@@ -237,7 +237,7 @@ def test_auto_scatter(loop_in_thread):
 
     with cluster() as (s, [a, b]):
         with Client(s['address'], loop=loop_in_thread) as client:
-            with parallel_backend('dask') as (ba, _):
+            with parallel_config(backend='dask'):
                 # Passing the same data as arg and kwarg triggers a single
                 # scatter operation whose result is reused.
                 Parallel()(delayed(noop)(data, data, i, opt=data)
@@ -250,7 +250,7 @@ def test_auto_scatter(loop_in_thread):
 
     with cluster() as (s, [a, b]):
         with Client(s['address'], loop=loop_in_thread) as client:
-            with parallel_backend('dask') as (ba, _):
+            with parallel_config(backend='dask'):
                 Parallel()(delayed(noop)(data1[:3], i) for i in range(5))
             # Small arrays are passed within the task definition without going
             # through a scatter operation.
@@ -272,7 +272,7 @@ def test_nested_scatter(loop, retry_no):
 
     def outer_function_joblib(array, i):
         client = get_client()  # noqa
-        with parallel_backend("dask"):
+        with parallel_config(backend="dask"):
             results = Parallel()(
                 delayed(my_sum)(array[j:], i, j) for j in range(
                     NUM_INNER_TASKS)
@@ -281,7 +281,7 @@ def test_nested_scatter(loop, retry_no):
 
     with cluster() as (s, [a, b]):
         with Client(s['address'], loop=loop) as _:
-            with parallel_backend("dask"):
+            with parallel_config(backend="dask"):
                 my_array = np.ones(10000)
                 _ = Parallel()(
                     delayed(outer_function_joblib)(
@@ -297,7 +297,7 @@ def test_nested_backend_context_manager(loop_in_thread):
 
     with cluster() as (s, [a, b]):
         with Client(s['address'], loop=loop_in_thread) as client:
-            with parallel_backend('dask') as (ba, _):
+            with parallel_config(backend='dask'):
                 pid_groups = Parallel(n_jobs=2)(
                     delayed(get_nested_pids)()
                     for _ in range(10)
@@ -307,7 +307,7 @@ def test_nested_backend_context_manager(loop_in_thread):
 
         # No deadlocks
         with Client(s['address'], loop=loop_in_thread) as client:  # noqa: F841
-            with parallel_backend('dask') as (ba, _):
+            with parallel_config(backend='dask'):
                 pid_groups = Parallel(n_jobs=2)(
                     delayed(get_nested_pids)()
                     for _ in range(10)
@@ -329,7 +329,7 @@ def test_nested_backend_context_manager_implicit_n_jobs(loop):
 
     with cluster() as (s, [a, b]):
         with Client(s['address'], loop=loop) as client:  # noqa: F841
-            with parallel_backend('dask') as (ba, _):
+            with parallel_config(backend='dask'):
                 with Parallel() as p:
                     assert _backend_type(p) == "DaskDistributedBackend"
                     assert p.n_jobs == -1
@@ -344,7 +344,7 @@ def test_nested_backend_context_manager_implicit_n_jobs(loop):
 
 def test_errors(loop):
     with pytest.raises(ValueError) as info:
-        with parallel_backend('dask'):
+        with parallel_config(backend='dask'):
             pass
 
     assert "create a dask client" in str(info.value).lower()
@@ -354,13 +354,13 @@ def test_correct_nested_backend(loop):
     with cluster() as (s, [a, b]):
         with Client(s['address'], loop=loop) as client:  # noqa: F841
             # No requirement, should be us
-            with parallel_backend('dask') as (ba, _):
+            with parallel_config(backend='dask'):
                 result = Parallel(n_jobs=2)(
                     delayed(outer)(nested_require=None) for _ in range(1))
                 assert isinstance(result[0][0][0], DaskDistributedBackend)
 
             # Require threads, should be threading
-            with parallel_backend('dask') as (ba, _):
+            with parallel_config(backend='dask'):
                 result = Parallel(n_jobs=2)(
                     delayed(outer)(nested_require='sharedmem')
                     for _ in range(1))
@@ -386,7 +386,7 @@ def inner():
 def test_secede_with_no_processes(loop):
     # https://github.com/dask/distributed/issues/1775
     with Client(loop=loop, processes=False, set_as_default=True):
-        with parallel_backend('dask'):
+        with parallel_config(backend='dask'):
             Parallel(n_jobs=4)(delayed(id)(i) for i in range(2))
 
 
@@ -398,12 +398,12 @@ def _worker_address(_):
 def test_dask_backend_keywords(loop):
     with cluster() as (s, [a, b]):
         with Client(s['address'], loop=loop) as client:  # noqa: F841
-            with parallel_backend('dask', workers=a['address']) as (ba, _):
+            with parallel_config(backend='dask', workers=a['address']):
                 seq = Parallel()(
                     delayed(_worker_address)(i) for i in range(10))
                 assert seq == [a['address']] * 10
 
-            with parallel_backend('dask', workers=b['address']) as (ba, _):
+            with parallel_config(backend='dask', workers=b['address']):
                 seq = Parallel()(
                     delayed(_worker_address)(i) for i in range(10))
                 assert seq == [b['address']] * 10
@@ -411,7 +411,7 @@ def test_dask_backend_keywords(loop):
 
 def test_cleanup(loop):
     with Client(processes=False, loop=loop) as client:
-        with parallel_backend('dask'):
+        with parallel_config(backend='dask'):
             Parallel()(delayed(inc)(i) for i in range(10))
 
         start = time()
@@ -438,7 +438,7 @@ def test_wait_for_workers(cluster_strategy):
         # to schedule work.
         cluster.scale(2)
     try:
-        with parallel_backend('dask'):
+        with parallel_config(backend='dask'):
             # The following should wait a bit for at least one worker to
             # become available.
             Parallel()(delayed(inc)(i) for i in range(10))
@@ -452,13 +452,13 @@ def test_wait_for_workers_timeout():
     cluster = LocalCluster(n_workers=0, processes=False, threads_per_worker=2)
     client = Client(cluster)
     try:
-        with parallel_backend('dask', wait_for_workers_timeout=0.1):
+        with parallel_config(backend='dask', wait_for_workers_timeout=0.1):
             # Short timeout: DaskDistributedBackend
             msg = "DaskDistributedBackend has no worker after 0.1 seconds."
             with pytest.raises(TimeoutError, match=msg):
                 Parallel()(delayed(inc)(i) for i in range(10))
 
-        with parallel_backend('dask', wait_for_workers_timeout=0):
+        with parallel_config(backend='dask', wait_for_workers_timeout=0):
             # No timeout: fallback to generic joblib failure:
             msg = "DaskDistributedBackend has no active worker"
             with pytest.raises(RuntimeError, match=msg):
diff --git a/joblib/test/test_parallel.py b/joblib/test/test_parallel.py
index 38a8d9b..c4770d2 100644
--- a/joblib/test/test_parallel.py
+++ b/joblib/test/test_parallel.py
@@ -61,11 +61,12 @@ from joblib._parallel_backends import ParallelBackendBase
 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 parallel_backend
+from joblib.parallel import register_parallel_backend
 from joblib.parallel import effective_n_jobs, cpu_count
 
-from joblib.parallel import mp, BACKENDS, DEFAULT_BACKEND, EXTERNAL_BACKENDS
+from joblib.parallel import mp, BACKENDS, DEFAULT_BACKEND
 
 
 RETURN_GENERATOR_BACKENDS = BACKENDS.copy()
@@ -593,11 +594,11 @@ def test_invalid_backend():
         Parallel(backend='unit-testing')
 
     with raises(ValueError, match="Invalid backend:"):
-        with parallel_backend('unit-testing'):
+        with parallel_config(backend='unit-testing'):
             pass
 
     with raises(ValueError, match="Invalid backend:"):
-        with parallel_config('unit-testing'):
+        with parallel_config(backend='unit-testing'):
             pass
 
 
@@ -636,7 +637,7 @@ def test_backend_no_multiprocessing():
         Parallel(backend='loky')(delayed(square)(i) for i in range(3))
 
     # The below should now work without problems
-    with parallel_backend('loky'):
+    with parallel_config(backend='loky'):
         Parallel()(delayed(square)(i) for i in range(3))
 
 
@@ -1549,7 +1550,7 @@ def test_backend_hinting_and_constraints_with_custom_backends(
         assert type(p._backend) == ThreadingBackend
 
         out, err = capsys.readouterr()
-        expected = ("Using ThreadingBackend as joblib.Parallel backend "
+        expected = ("Using ThreadingBackend as joblib backend "
                     "instead of MyCustomProcessingBackend as the latter "
                     "does not provide shared memory semantics.")
         assert out.strip() == expected
@@ -1580,28 +1581,6 @@ def test_invalid_backend_hinting_and_constraints():
             Parallel(backend='multiprocessing', require='sharedmem')
 
 
-@parametrize("context", [parallel_config, parallel_backend])
-def test_global_parallel_backend(context):
-    default = Parallel()._backend
-
-    pb = context('threading')
-    assert isinstance(Parallel()._backend, ThreadingBackend)
-
-    pb.unregister()
-    assert type(Parallel()._backend) is type(default)
-
-
-@parametrize("context", [parallel_config, parallel_backend])
-def test_external_backends(context):
-    def register_foo():
-        BACKENDS['foo'] = ThreadingBackend
-
-    EXTERNAL_BACKENDS['foo'] = register_foo
-
-    with context('foo'):
-        assert isinstance(Parallel()._backend, ThreadingBackend)
-
-
 def _recursive_backend_info(limit=3, **kwargs):
     """Perform nested parallel calls and introspect the backend on the way"""
 
@@ -1688,7 +1667,7 @@ def test_thread_bomb_mitigation(context, backend):
             # pickling it to be send back to the parent process. In this
             # case the worker crashes but the original traceback is still
             # printed on stderr. This could be improved but does not seem
-            # simple to do and this is is not critical for users (as long
+            # simple to do and this is not critical for users (as long
             # as there is no process or thread bomb happening).
             pytest.xfail("Loky worker crash when serializing RecursionError")
 
@@ -1902,17 +1881,6 @@ def test_threadpool_limitation_in_child_override(context, n_jobs, var_name):
             os.environ[var_name] = original_var_value
 
 
-@with_numpy
-@with_multiprocessing
-@parametrize('backend', ['multiprocessing', 'threading',
-                         MultiprocessingBackend(), ThreadingBackend()])
-@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"):
-        context(backend, inner_max_num_threads=1)
-
-
 @with_multiprocessing
 @parametrize('n_jobs', [2, 4, -1])
 def test_loky_reuse_workers(n_jobs):
@@ -1933,48 +1901,3 @@ 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