Repository: joblib/joblib
Pull request #1047 (2020-05-05 22:46:56)

Do not reuse temporary folders across parallel calls.
Follow up on #966.

### Main improvements:

#### Temporary folder reuse across `Parallel` calls.

TLDR; each `Parallel` call now uses a different temporary folder where the reducers will store temporary memmaps. I think this solves most of the bad behaviors we saw, such as spurious `KeyErrors`/`FileNotFoundError` coming from the `resource_tracker`, or straight out `loky` workers errors complaining about inexistent memmaps.

One big thing that was overlooked in #966 was: what happens to temporary folders when reusing a `loky` executor? In particular, we don't want several `Parallel` calls to use the same temporary folder to store temporary memmaps, but we still want to reuse `loky` executor, and the `reducers` it comes up with. 
- The less invasive solution was to patch this `temp_folder` inside the  `loky` executor at each `get_reusable_executor` call (through a very nested attribute acces chain).
- However, this issue was a good occasion to factor out the temporary folder resolving logic out of the reducers, which opened the door for a cleaner solution: instead of having `temp_folder` being a static string,  pass a callable to the `reducers` that will dynamically resolve the `temp_folder` to use. This callable can then have its state modified by the `parallel_backend` to return a different`temp_folder` across `Parallel` calls.
- Assuming that the `joblib` public API contains `Parallel`, `get_memmaping_executor` and `MemmappingPool`, but nothing within `_memmaping_reducers`, there is no change to the `temp_folder` public API.


#### Various refactorings
As stated above, this PR was the occasion to clean up some parts of the code:
- temporary folder resolving and temporary resources management are now done by instances of the `TemporaryResourceManager` class. Each `multiprocessing` pool /`loky` executor has an instance of this class. I don't rely on `mixins` anymore.
- the loky changes allow use to subclass the `_ReusablePoolExecutor`, and add a handful of methods that one could except from a `MemmappingPool/Executor`. Especially, I added a `terminate`/`cleanup` methods, so that temporary resources management becomes completely opaque to the `parallel_backends`: they just have to call those high-level methods.


#### Update on thread-safety

 To make sure that concurrent calls of  `Parallel` object relying on the same `loky` executor use a different temporary folder each, we had to make the `TemporaryFolderManager` resolving policy context-aware. 
In practice, a `Parallel`-specific context is set to the active reducer's managers when pickling `BatchedCalls`. Because all pickling happens in one single thread, setting the context from within this thread is single-threaded, thus thread-safe.

codecov[bot]: # [Codecov](https://codecov.io/gh/joblib/joblib/pull/1047?src=pr&el=h1) Report
> Merging [#1047](https://codecov.io/gh/joblib/joblib/pull/1047?src=pr&el=desc) into [master](https://codecov.io/gh/joblib/joblib/commit/e959a01e9f0c160e9e89f6892928d755d02f2065&el=desc) will **decrease** coverage by `13.14%`.
> The diff coverage is `100.00%`.

[![Impacted file tree graph](https://codecov.io/gh/joblib/joblib/pull/1047/graphs/tree.svg?width=650&height=150&src=pr&token=gA6LF5DGTW)](https://codecov.io/gh/joblib/joblib/pull/1047?src=pr&el=tree)

```diff
@@             Coverage Diff             @@
##           master    #1047       +/-   ##
===========================================
- Coverage   94.31%   81.17%   -13.15%     
===========================================
  Files          47       47               
  Lines        6654     6495      -159     
===========================================
- Hits         6276     5272     -1004     
- Misses        378     1223      +845     
```


| [Impacted Files](https://codecov.io/gh/joblib/joblib/pull/1047?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [joblib/executor.py](https://codecov.io/gh/joblib/joblib/pull/1047/diff?src=pr&el=tree#diff-am9ibGliL2V4ZWN1dG9yLnB5) | `74.28% <100.00%> (-25.72%)` | :arrow_down: |
| [joblib/test/test\_dask.py](https://codecov.io/gh/joblib/joblib/pull/1047/diff?src=pr&el=tree#diff-am9ibGliL3Rlc3QvdGVzdF9kYXNrLnB5) | `4.43% <0.00%> (-93.61%)` | :arrow_down: |
| [joblib/\_dask.py](https://codecov.io/gh/joblib/joblib/pull/1047/diff?src=pr&el=tree#diff-am9ibGliL19kYXNrLnB5) | `23.42% <0.00%> (-70.89%)` | :arrow_down: |
| [joblib/test/test\_memmapping.py](https://codecov.io/gh/joblib/joblib/pull/1047/diff?src=pr&el=tree#diff-am9ibGliL3Rlc3QvdGVzdF9tZW1tYXBwaW5nLnB5) | `39.12% <0.00%> (-60.00%)` | :arrow_down: |
| [joblib/format\_stack.py](https://codecov.io/gh/joblib/joblib/pull/1047/diff?src=pr&el=tree#diff-am9ibGliL2Zvcm1hdF9zdGFjay5weQ==) | `58.33% <0.00%> (-41.67%)` | :arrow_down: |
| [joblib/\_memmapping\_reducer.py](https://codecov.io/gh/joblib/joblib/pull/1047/diff?src=pr&el=tree#diff-am9ibGliL19tZW1tYXBwaW5nX3JlZHVjZXIucHk=) | `55.29% <0.00%> (-41.48%)` | :arrow_down: |
| [joblib/backports.py](https://codecov.io/gh/joblib/joblib/pull/1047/diff?src=pr&el=tree#diff-am9ibGliL2JhY2twb3J0cy5weQ==) | `44.73% <0.00%> (-39.48%)` | :arrow_down: |
| [joblib/my\_exceptions.py](https://codecov.io/gh/joblib/joblib/pull/1047/diff?src=pr&el=tree#diff-am9ibGliL215X2V4Y2VwdGlvbnMucHk=) | `64.28% <0.00%> (-35.72%)` | :arrow_down: |
| [joblib/pool.py](https://codecov.io/gh/joblib/joblib/pull/1047/diff?src=pr&el=tree#diff-am9ibGliL3Bvb2wucHk=) | `73.91% <0.00%> (-13.05%)` | :arrow_down: |
| ... and [23 more](https://codecov.io/gh/joblib/joblib/pull/1047/diff?src=pr&el=tree-more) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/joblib/joblib/pull/1047?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/joblib/joblib/pull/1047?src=pr&el=footer). Last update [e959a01...8cba4a5](https://codecov.io/gh/joblib/joblib/pull/1047?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

pierreglaser: The CI failure on `pypy3` is the same flaky test that breaks 1/10 builds.

Patch:
diff --git a/joblib/_dask.py b/joblib/_dask.py
index 2db9004..a3af0d1 100644
--- a/joblib/_dask.py
+++ b/joblib/_dask.py
@@ -1,5 +1,7 @@
 from __future__ import print_function, division, absolute_import
 
+import asyncio
+import concurrent.futures
 import contextlib
 
 from uuid import uuid4
@@ -14,11 +16,16 @@ except ImportError:
     distributed = None
 
 if distributed is not None:
-    from distributed.client import Client, _wait
-    from distributed.utils import funcname, itemgetter
-    from distributed import get_client, secede, rejoin
-    from distributed.worker import thread_state
-    from distributed.sizeof import sizeof
+    from dask.utils import funcname, itemgetter
+    from dask.sizeof import sizeof
+    from dask.distributed import (
+        Client,
+        as_completed,
+        get_client,
+        secede,
+        rejoin
+    )
+    from distributed.utils import thread_state
 
     try:
         # asyncio.TimeoutError, Python3-only error thrown by recent versions of
@@ -27,6 +34,7 @@ if distributed is not None:
     except ImportError:
         from tornado.gen import TimeoutError as _TimeoutError
 
+
 def is_weakrefable(obj):
     try:
         weakref.ref(obj)
@@ -115,6 +123,7 @@ def _joblib_probe_task():
 class DaskDistributedBackend(ParallelBackendBase, AutoBatchingMixin):
     MIN_IDEAL_BATCH_DURATION = 0.2
     MAX_IDEAL_BATCH_DURATION = 1.0
+    supports_timeout = True
 
     def __init__(self, scheduler_host=None, scatter=None,
                  client=None, loop=None, wait_for_workers_timeout=10,
@@ -155,9 +164,29 @@ class DaskDistributedBackend(ParallelBackendBase, AutoBatchingMixin):
         else:
             self._scatter = []
             self.data_futures = {}
-        self.task_futures = set()
         self.wait_for_workers_timeout = wait_for_workers_timeout
         self.submit_kwargs = submit_kwargs
+        self.waiting_futures = as_completed(
+            [],
+            loop=client.loop,
+            with_results=True,
+            raise_errors=False
+        )
+        self._results = {}
+        self._callbacks = {}
+
+    async def _collect(self):
+        while self._continue:
+            async for future, result in self.waiting_futures:
+                cf_future = self._results.pop(future)
+                callback = self._callbacks.pop(future)
+                if future.status == "error":
+                    typ, exc, tb = result
+                    cf_future.set_exception(exc)
+                else:
+                    cf_future.set_result(result)
+                    callback(result)
+            await asyncio.sleep(0.01)
 
     def __reduce__(self):
         return (DaskDistributedBackend, ())
@@ -169,11 +198,14 @@ class DaskDistributedBackend(ParallelBackendBase, AutoBatchingMixin):
         return self.effective_n_jobs(n_jobs)
 
     def start_call(self):
+        self._continue = True
+        self.client.loop.add_callback(self._collect)
         self.call_data_futures = _WeakKeyDictionary()
 
     def stop_call(self):
         # The explicit call to clear is required to break a cycling reference
         # to the futures.
+        self._continue = False
         self.call_data_futures.clear()
 
     def effective_n_jobs(self, n_jobs):
@@ -199,7 +231,7 @@ class DaskDistributedBackend(ParallelBackendBase, AutoBatchingMixin):
             raise TimeoutError(error_msg)
         return sum(self.client.ncores().values())
 
-    def _to_func_args(self, func):
+    async def _to_func_args(self, func):
         collected_futures = []
         itemgetters = dict()
 
@@ -207,11 +239,12 @@ class DaskDistributedBackend(ParallelBackendBase, AutoBatchingMixin):
         # Parallel.__call__.
         call_data_futures = getattr(self, 'call_data_futures', None)
 
-        def maybe_to_futures(args):
+        async def maybe_to_futures(args):
+            out = []
             for arg in args:
                 arg_id = id(arg)
                 if arg_id in itemgetters:
-                    yield itemgetters[arg_id]
+                    out.append(itemgetters[arg_id])
                     continue
 
                 f = self.data_futures.get(arg_id, None)
@@ -225,7 +258,10 @@ class DaskDistributedBackend(ParallelBackendBase, AutoBatchingMixin):
                             # Rely on automated inter-worker data stealing if
                             # more workers need to reuse this data
                             # concurrently.
-                            [f] = self.client.scatter([arg])
+                            [f] = await self.client.scatter(
+                                [arg],
+                                asynchronous=True
+                            )
                             call_data_futures[arg] = f
 
                 if f is not None:
@@ -233,13 +269,14 @@ class DaskDistributedBackend(ParallelBackendBase, AutoBatchingMixin):
                     collected_futures.append(f)
                     itemgetters[arg_id] = getter
                     arg = getter
-                yield arg
+                out.append(arg)
+            return out
 
         tasks = []
         for f, args, kwargs in func.items:
-            args = list(maybe_to_futures(args))
+            args = list(await maybe_to_futures(args))
             kwargs = dict(zip(kwargs.keys(),
-                              maybe_to_futures(kwargs.values())))
+                              await maybe_to_futures(kwargs.values())))
             tasks.append((f, args, kwargs))
 
         if not collected_futures:
@@ -248,34 +285,33 @@ class DaskDistributedBackend(ParallelBackendBase, AutoBatchingMixin):
 
     def apply_async(self, func, callback=None):
         key = '%s-batch-%s' % (_funcname(func), uuid4().hex)
-        func, args = self._to_func_args(func)
-
-        future = self.client.submit(func, *args, key=key, **self.submit_kwargs)
-        self.task_futures.add(future)
 
-        def callback_wrapper(future):
-            result = future.result()
-            self.task_futures.remove(future)
-            if callback is not None:
-                callback(result)
+        cf_future = concurrent.futures.Future()
+        cf_future.get = cf_future.result  # achieve AsyncResult API
 
-        future.add_done_callback(callback_wrapper)
+        async def f(func, callback):
+            func, args = await self._to_func_args(func)
 
-        ref = weakref.ref(future)  # avoid reference cycle
+            dask_future = self.client.submit(
+                func, *args, key=key, **self.submit_kwargs
+            )
+            self.waiting_futures.add(dask_future)
+            self._callbacks[dask_future] = callback
+            self._results[dask_future] = cf_future
 
-        def get():
-            return ref().result()
+        self.client.loop.add_callback(f, func, callback)
 
-        future.get = get  # monkey patch to achieve AsyncResult API
-        return future
+        return cf_future
 
     def abort_everything(self, ensure_ready=True):
         """ Tell the client to cancel any task submitted via this instance
 
         joblib.Parallel will never access those results
         """
-        self.client.cancel(self.task_futures)
-        self.task_futures.clear()
+        with self.waiting_futures.lock:
+            self.waiting_futures.futures.clear()
+            while not self.waiting_futures.queue.empty():
+                self.waiting_futures.queue.get()
 
     @contextlib.contextmanager
     def retrieval_context(self):
diff --git a/joblib/_memmapping_reducer.py b/joblib/_memmapping_reducer.py
index 1e069fb..d583822 100644
--- a/joblib/_memmapping_reducer.py
+++ b/joblib/_memmapping_reducer.py
@@ -318,8 +318,9 @@ class ArrayMemmapForwardReducer(object):
     max_nbytes: int
         Threshold to trigger memmapping of large arrays to files created
         a folder.
-    temp_folder: str
-        Path of a folder where files for backing memmapped arrays are created.
+    temp_folder_resolver: callable
+        An callable in charge of resolving a temporary folder name where files
+        for backing memmapped arrays are created.
     mmap_mode: 'r', 'r+' or 'c'
         Mode for the created memmap datastructure. See the documentation of
         numpy.memmap for more details. Note: 'w+' is coerced to 'r+'
@@ -334,23 +335,38 @@ class ArrayMemmapForwardReducer(object):
         same data array is passed to different worker processes.
     """
 
-    def __init__(self, max_nbytes, temp_folder, mmap_mode,
+    def __init__(self, max_nbytes, temp_folder_resolver, mmap_mode,
                  unlink_on_gc_collect, verbose=0, prewarm=True):
         self._max_nbytes = max_nbytes
-        self._temp_folder = temp_folder
+        self._temp_folder_resolver = temp_folder_resolver
         self._mmap_mode = mmap_mode
         self.verbose = int(verbose)
+        if prewarm == "auto":
+            self._prewarm = not self._temp_folder.startswith(
+                SYSTEM_SHARED_MEM_FS
+            )
+        else:
+            self._prewarm = prewarm
         self._prewarm = prewarm
         self._memmaped_arrays = _WeakArrayKeyMap()
         self._temporary_memmaped_filenames = set()
         self._unlink_on_gc_collect = unlink_on_gc_collect
 
+    @property
+    def _temp_folder(self):
+        return self._temp_folder_resolver()
+
     def __reduce__(self):
         # The ArrayMemmapForwardReducer is passed to the children processes: it
         # needs to be pickled but the _WeakArrayKeyMap need to be skipped as
         # it's only guaranteed to be consistent with the parent process memory
         # garbage collection.
-        args = (self._max_nbytes, self._temp_folder, self._mmap_mode,
+        # Although this reducer is pickled, it is not needed in its destination
+        # process (child processes), as we only use this reducer to send
+        # memmaps from the parent process to the children processes. For this
+        # reason, we can afford skipping the resolver, (which would otherwise
+        # be unpicklable), and pass it as None instead.
+        args = (self._max_nbytes, None, self._mmap_mode,
                 self._unlink_on_gc_collect)
         kwargs = {
             'verbose': self.verbose,
@@ -421,8 +437,8 @@ class ArrayMemmapForwardReducer(object):
             if not os.path.exists(filename):
                 util.debug(
                     "[ARRAY DUMP] Pickling new array (shape={}, dtype={}) "
-                    "creating a new memmap".format(
-                        a.shape, a.dtype, os.path.basename(filename)))
+                    "creating a new memmap at {}".format(
+                        a.shape, a.dtype, filename))
                 for dumped_filename in dump(a, filename):
                     os.chmod(dumped_filename, FILE_PERMISSIONS)
 
@@ -455,8 +471,8 @@ class ArrayMemmapForwardReducer(object):
 
 
 def get_memmapping_reducers(
-        pool_id, forward_reducers=None, backward_reducers=None,
-        temp_folder=None, max_nbytes=1e6, mmap_mode='r', verbose=0,
+        forward_reducers=None, backward_reducers=None,
+        temp_folder_resolver=None, max_nbytes=1e6, mmap_mode='r', verbose=0,
         prewarm=False, unlink_on_gc_collect=True, **kwargs):
     """Construct a pair of memmapping reducer linked to a tmpdir.
 
@@ -469,51 +485,13 @@ def get_memmapping_reducers(
     if backward_reducers is None:
         backward_reducers = dict()
 
-    # Prepare a sub-folder name for the serialization of this particular
-    # pool instance (do not create in advance to spare FS write access if
-    # no array is to be dumped):
-    pool_folder_name = "joblib_memmapping_folder_{}_{}".format(
-        os.getpid(), pool_id)
-    pool_folder, use_shared_mem = _get_temp_dir(pool_folder_name,
-                                                temp_folder)
-
-    # Register the garbage collector at program exit in case caller forgets
-    # to call terminate explicitly: note we do not pass any reference to
-    # self to ensure that this callback won't prevent garbage collection of
-    # the pool instance and related file handler resources such as POSIX
-    # semaphores and pipes
-    pool_module_name = whichmodule(delete_folder, 'delete_folder')
-    resource_tracker.register(pool_folder, "folder")
-
-    def _cleanup():
-        # In some cases the Python runtime seems to set delete_folder to
-        # None just before exiting when accessing the delete_folder
-        # function from the closure namespace. So instead we reimport
-        # the delete_folder function explicitly.
-        # https://github.com/joblib/joblib/issues/328
-        # We cannot just use from 'joblib.pool import delete_folder'
-        # because joblib should only use relative imports to allow
-        # easy vendoring.
-        delete_folder = __import__(
-            pool_module_name, fromlist=['delete_folder']).delete_folder
-        try:
-            delete_folder(pool_folder, allow_non_empty=True)
-            resource_tracker.unregister(pool_folder, "folder")
-        except OSError:
-            warnings.warn("Failed to clean temporary folder: {}"
-                          .format(pool_folder))
-
-    atexit.register(_cleanup)
-
     if np is not None:
         # Register smart numpy.ndarray reducers that detects memmap backed
         # arrays and that is also able to dump to memmap large in-memory
         # arrays over the max_nbytes threshold
-        if prewarm == "auto":
-            prewarm = not use_shared_mem
         forward_reduce_ndarray = ArrayMemmapForwardReducer(
-            max_nbytes, pool_folder, mmap_mode, unlink_on_gc_collect, verbose,
-            prewarm=prewarm)
+            max_nbytes, temp_folder_resolver, mmap_mode, unlink_on_gc_collect,
+            verbose, prewarm=prewarm)
         forward_reducers[np.ndarray] = forward_reduce_ndarray
         forward_reducers[np.memmap] = forward_reduce_ndarray
 
@@ -524,34 +502,163 @@ def get_memmapping_reducers(
         backward_reducers[np.ndarray] = reduce_array_memmap_backward
         backward_reducers[np.memmap] = reduce_array_memmap_backward
 
-    return forward_reducers, backward_reducers, pool_folder
+    return forward_reducers, backward_reducers
+
+
+class TemporaryResourcesManager(object):
+    """Stateful object able to manage temporary folder and pickles
+
+    It exposes:
+    - a per-context folder name resolving API that memmap-based reducers will
+      rely on to know where to pickle the temporary memmaps
+    - a temporary file/folder management API that internally uses the
+      resource_tracker.
+    """
+
+    def __init__(self, temp_folder_root=None, context_id=None):
+        self._current_temp_folder = None
+        self._temp_folder_root = temp_folder_root
+        self._use_shared_mem = None
+        self._cached_temp_folders = dict()
+        self._id = uuid4().hex
+        self._finalizers = {}
+        if context_id is None:
+            # 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.
+            context_id = uuid4().hex
+        self.set_current_context(context_id)
+
+    def set_current_context(self, context_id):
+        self._current_context_id = context_id
+        self.register_new_context(context_id)
+
+    def register_new_context(self, context_id):
+        # Prepare a sub-folder name specific to a context (usually a unique id
+        # generated by each instance of the Parallel class). Do not create in
+        # advance to spare FS write access if no array is to be dumped).
+        if context_id in self._cached_temp_folders:
+            return
+        else:
+            # During its lifecycle, one Parallel object can have several
+            # executors associated to it (for instance, if a loky worker raises
+            # an exception, joblib shutdowns the executor and instantly
+            # recreates a new one before raising the error - see
+            # ``ensure_ready``.  Because we don't want two executors tied to
+            # the same Parallel object (and thus the same context id) to
+            # register/use/delete the same folder, we also add an id specific
+            # to the current Manager (and thus specific to its associated
+            # executor) to the folder name.
+            new_folder_name = (
+                "joblib_memmapping_folder_{}_{}_{}".format(
+                    os.getpid(), self._id, context_id)
+            )
+            new_folder_path, _ = _get_temp_dir(
+                new_folder_name, self._temp_folder_root
+            )
+            self.register_folder_finalizer(new_folder_path, context_id)
+            self._cached_temp_folders[context_id] = new_folder_path
+
+    def resolve_temp_folder_name(self):
+        """Return a folder name specific to the currently activated context"""
+        return self._cached_temp_folders[self._current_context_id]
+
+    def _unregister_context(self, context_id=None):
+        if context_id is None:
+            for context_id in list(self._cached_temp_folders):
+                self._unregister_context(context_id)
+        else:
+            temp_folder = self._cached_temp_folders[context_id]
+            finalizer = self._finalizers[context_id]
+
+            resource_tracker.unregister(temp_folder, "folder")
+            atexit.unregister(finalizer)
+
+            self._cached_temp_folders.pop(context_id)
+            self._finalizers.pop(context_id)
+
+    # resource management API
+
+    def register_folder_finalizer(self, pool_subfolder, context_id):
+        # Register the garbage collector at program exit in case caller forgets
+        # to call terminate explicitly: note we do not pass any reference to
+        # ensure that this callback won't prevent garbage collection of
+        # parallel instance and related file handler resources such as POSIX
+        # semaphores and pipes
+        pool_module_name = whichmodule(delete_folder, 'delete_folder')
+        resource_tracker.register(pool_subfolder, "folder")
+
+        def _cleanup():
+            # In some cases the Python runtime seems to set delete_folder to
+            # None just before exiting when accessing the delete_folder
+            # function from the closure namespace. So instead we reimport
+            # the delete_folder function explicitly.
+            # https://github.com/joblib/joblib/issues/328
+            # We cannot just use from 'joblib.pool import delete_folder'
+            # because joblib should only use relative imports to allow
+            # easy vendoring.
+            delete_folder = __import__(
+                pool_module_name, fromlist=['delete_folder']).delete_folder
+            try:
+                delete_folder(pool_subfolder, allow_non_empty=True)
+                resource_tracker.unregister(pool_subfolder, "folder")
+            except OSError:
+                warnings.warn("Failed to delete temporary folder: {}"
+                              .format(pool_subfolder))
 
+        self._finalizers[context_id] = atexit.register(_cleanup)
 
-class TemporaryResourcesManagerMixin(object):
-    def _unlink_temporary_resources(self):
+    def _unlink_temporary_resources(self, context_id=None):
         """Unlink temporary resources created by a process-based pool"""
-        if os.path.exists(self._temp_folder):
-            for filename in os.listdir(self._temp_folder):
-                resource_tracker.maybe_unlink(
-                    os.path.join(self._temp_folder, filename), "file"
+        if context_id is None:
+            # iterate over a copy of the cache keys because
+            # unlink_temporary_resources further deletes an entry in this
+            # cache
+            for context_id in self._cached_temp_folders.copy():
+                self._unlink_temporary_resources(context_id)
+        else:
+            temp_folder = self._cached_temp_folders[context_id]
+            if os.path.exists(temp_folder):
+                for filename in os.listdir(temp_folder):
+                    resource_tracker.maybe_unlink(
+                        os.path.join(temp_folder, filename), "file"
+                    )
+                self._try_delete_folder(
+                    allow_non_empty=False, context_id=context_id
                 )
-            # XXX: calling shutil.rmtree inside delete_folder is likely to
-            # cause a race condition with the lines above.
-            self._try_delete_folder(allow_non_empty=False)
 
-    def _unregister_temporary_resources(self):
+    def _unregister_temporary_resources(self, context_id=None):
         """Unregister temporary resources created by a process-based pool"""
-        if os.path.exists(self._temp_folder):
-            for filename in os.listdir(self._temp_folder):
-                resource_tracker.unregister(
-                    os.path.join(self._temp_folder, filename), "file"
+        if context_id is None:
+            for context_id in self._cached_temp_folders:
+                self._unregister_temporary_resources(context_id)
+        else:
+            temp_folder = self._cached_temp_folders[context_id]
+            if os.path.exists(temp_folder):
+                for filename in os.listdir(temp_folder):
+                    resource_tracker.unregister(
+                        os.path.join(temp_folder, filename), "file"
+                    )
+
+    def _try_delete_folder(self, allow_non_empty, context_id=None):
+        if context_id is None:
+            # ditto
+            for context_id in self._cached_temp_folders.copy():
+                self._try_delete_folder(
+                    allow_non_empty=allow_non_empty, context_id=context_id
                 )
-
-    def _try_delete_folder(self, allow_non_empty):
-        try:
-            delete_folder(self._temp_folder, allow_non_empty=allow_non_empty)
-        except OSError:
-            # Temporary folder cannot be deleted right now. No need to
-            # handle it though, as this folder will be cleaned up by an
-            # atexit finalizer registered by the memmapping_reducer.
-            pass
+        else:
+            temp_folder = self._cached_temp_folders[context_id]
+            try:
+                delete_folder(
+                    temp_folder, allow_non_empty=allow_non_empty
+                )
+                # Now that this folder is deleted, we can forget about it
+                self._unregister_context(context_id)
+
+            except OSError:
+                # Temporary folder cannot be deleted right now. No need to
+                # handle it though, as this folder will be cleaned up by an
+                # atexit finalizer registered by the memmapping_reducer.
+                pass
diff --git a/joblib/_parallel_backends.py b/joblib/_parallel_backends.py
index 77d3617..3846713 100644
--- a/joblib/_parallel_backends.py
+++ b/joblib/_parallel_backends.py
@@ -15,7 +15,6 @@ from ._multiprocessing_helpers import mp
 
 if mp is not None:
     from .pool import MemmappingPool
-    from ._memmapping_reducer import TemporaryResourcesManagerMixin
     from multiprocessing.pool import ThreadPool
     from .executor import get_memmapping_executor
 
@@ -23,8 +22,6 @@ if mp is not None:
     from multiprocessing import TimeoutError
     from concurrent.futures._base import TimeoutError as CfTimeoutError
     from .externals.loky import process_executor, cpu_count
-else:
-    TemporaryResourcesManagerMixin = object
 
 
 class ParallelBackendBase(metaclass=ABCMeta):
@@ -478,8 +475,7 @@ class MultiprocessingBackend(PoolManagerMixin, AutoBatchingMixin,
         self.reset_batch_stats()
 
 
-class LokyBackend(AutoBatchingMixin, ParallelBackendBase,
-                  TemporaryResourcesManagerMixin):
+class LokyBackend(AutoBatchingMixin, ParallelBackendBase):
     """Managing pool of workers with loky instead of multiprocessing."""
 
     supports_timeout = True
@@ -496,8 +492,7 @@ class LokyBackend(AutoBatchingMixin, ParallelBackendBase,
         self._workers = get_memmapping_executor(
             n_jobs, timeout=idle_worker_timeout,
             env=self._prepare_worker_env(n_jobs=n_jobs),
-            **memmappingexecutor_args)
-        self._temp_folder = self._workers._temp_folder
+            context_id=parallel._id, **memmappingexecutor_args)
         self.parallel = parallel
         return n_jobs
 
@@ -548,10 +543,12 @@ class LokyBackend(AutoBatchingMixin, ParallelBackendBase,
 
     def terminate(self):
         if self._workers is not None:
-            self._unlink_temporary_resources()
-
-            # Terminate does not shutdown the workers as we want to reuse them
-            # in latter calls
+            # Don't terminate the workers as we want to reuse them in later
+            # calls, but cleanup the temporary resources that the Parallel call
+            # created. This 'hack' requires a private, low-level operation.
+            self._workers._temp_folder_manager._unlink_temporary_resources(
+                context_id=self.parallel._id
+            )
             self._workers = None
 
         self.reset_batch_stats()
@@ -559,16 +556,7 @@ class LokyBackend(AutoBatchingMixin, ParallelBackendBase,
     def abort_everything(self, ensure_ready=True):
         """Shutdown the workers and restart a new one with the same parameters
         """
-        self._workers.shutdown(kill_workers=True)
-        # When workers are killed in such a brutal manner, they cannot execute
-        # the finalizer of their shared memmaps. The refcount of those memmaps
-        # may be off by an unknown number, so insted of decref'ing them, we
-        # delete the whole temporary folder, and unregister them. There is no
-        # risk of PermissionError at folder deletion because because at this
-        # point, all child processes are dead, so all references
-        # to temporary memmaps are closed.
-        self._unregister_temporary_resources()
-        self._try_delete_folder(allow_non_empty=True)
+        self._workers.terminate(kill_workers=True)
         self._workers = None
 
         if ensure_ready:
diff --git a/joblib/executor.py b/joblib/executor.py
index e7e678d..9273fed 100644
--- a/joblib/executor.py
+++ b/joblib/executor.py
@@ -8,64 +8,113 @@ copy between the parent and child processes.
 # Copyright: 2017, Thomas Moreau
 # License: BSD 3 clause
 
-import random
 from ._memmapping_reducer import get_memmapping_reducers
-from ._memmapping_reducer import TemporaryResourcesManagerMixin
-from .externals.loky.reusable_executor import get_reusable_executor
+from ._memmapping_reducer import TemporaryResourcesManager
+from .externals.loky.reusable_executor import _ReusablePoolExecutor
 
 
 _executor_args = None
 
 
-def get_memmapping_executor(n_jobs, timeout=300, initializer=None, initargs=(),
-                            env=None, **backend_args):
-    """Factory for ReusableExecutor with automatic memmapping for large numpy
-    arrays.
-    """
-    global _executor_args
-    # Check if we can reuse the executor here instead of deferring the test to
-    # loky as the reducers are objects that changes at each call.
-    executor_args = backend_args.copy()
-    executor_args.update(env if env else {})
-    executor_args.update(dict(
-        timeout=timeout, initializer=initializer, initargs=initargs))
-    reuse = _executor_args is None or _executor_args == executor_args
-    _executor_args = executor_args
-
-    id_executor = random.randint(0, int(1e10))
-    job_reducers, result_reducers, temp_folder = get_memmapping_reducers(
-        id_executor, unlink_on_gc_collect=True, **backend_args)
-    _executor = get_reusable_executor(n_jobs, job_reducers=job_reducers,
-                                      result_reducers=result_reducers,
-                                      reuse=reuse, timeout=timeout,
-                                      initializer=initializer,
-                                      initargs=initargs, env=env)
-    # If executor doesn't have a _temp_folder, it means it is a new executor
-    # and the reducers have not been used. Else, the previous reducers are used
-    # and we should not change this attribute.
-    if not hasattr(_executor, "_temp_folder"):
-        _executor._temp_folder = temp_folder
-    return _executor
-
-
-class _TestingMemmappingExecutor(TemporaryResourcesManagerMixin):
+def get_memmapping_executor(n_jobs, **kwargs):
+    return MemmappingExecutor.get_memmapping_executor(n_jobs, **kwargs)
+
+
+class MemmappingExecutor(_ReusablePoolExecutor):
+
+    @classmethod
+    def get_memmapping_executor(cls, n_jobs, timeout=300, initializer=None,
+                                initargs=(), env=None, temp_folder=None,
+                                context_id=None, **backend_args):
+        """Factory for ReusableExecutor with automatic memmapping for large numpy
+        arrays.
+        """
+        global _executor_args
+        # Check if we can reuse the executor here instead of deferring the test
+        # to loky as the reducers are objects that changes at each call.
+        executor_args = backend_args.copy()
+        executor_args.update(env if env else {})
+        executor_args.update(dict(
+            timeout=timeout, initializer=initializer, initargs=initargs))
+        reuse = _executor_args is None or _executor_args == executor_args
+        _executor_args = executor_args
+
+        manager = TemporaryResourcesManager(temp_folder)
+
+        # reducers access the temporary folder in which to store temporary
+        # pickles through a call to manager.resolve_temp_folder_name. resolving
+        # the folder name dynamically is useful to use different folders across
+        # calls of a same reusable executor
+        job_reducers, result_reducers = get_memmapping_reducers(
+            unlink_on_gc_collect=True,
+            temp_folder_resolver=manager.resolve_temp_folder_name,
+            **backend_args)
+        _executor, executor_is_reused = super().get_reusable_executor(
+            n_jobs, job_reducers=job_reducers, result_reducers=result_reducers,
+            reuse=reuse, timeout=timeout, initializer=initializer,
+            initargs=initargs, env=env
+        )
+
+        if not executor_is_reused:
+            # Only set a _temp_folder_manager for new executors. Reused
+            # executors already have a _temporary_folder_manager that must not
+            # be re-assigned like that because it is referenced in various
+            # places in the reducing machinery of the executor.
+            _executor._temp_folder_manager = manager
+
+        if context_id is not None:
+            # Only register the specified context once we know which manager
+            # the current executor is using, in order to not register an atexit
+            # finalizer twice for the same folder.
+            _executor._temp_folder_manager.register_new_context(context_id)
+
+        return _executor
+
+    def terminate(self, kill_workers=False):
+        self.shutdown(kill_workers=kill_workers)
+        if kill_workers:
+            # When workers are killed in such a brutal manner, they cannot
+            # execute the finalizer of their shared memmaps. The refcount of
+            # those memmaps may be off by an unknown number, so instead of
+            # decref'ing them, we delete the whole temporary folder, and
+            # unregister them. There is no risk of PermissionError at folder
+            # deletion because because at this point, all child processes are
+            # dead, so all references to temporary memmaps are closed.
+
+            # unregister temporary resources from all contexts
+            with self._submit_resize_lock:
+                self._temp_folder_manager._unregister_temporary_resources()
+                self._temp_folder_manager._try_delete_folder(
+                    allow_non_empty=True
+                )
+        else:
+            self._temp_folder_manager._unlink_temporary_resources()
+            self._temp_folder_manager._try_delete_folder(allow_non_empty=True)
+
+    @property
+    def _temp_folder(self):
+        # Legacy property in tests. could be removed if we refactored the
+        # memmapping tests. SHOULD ONLY BE USED IN TESTS!
+        # We cache this property because it is called late in the tests - at
+        # this point, all context have been unregistered, and
+        # resolve_temp_folder_name raises an error.
+        if getattr(self, '_cached_temp_folder', None) is not None:
+            return self._cached_temp_folder
+        else:
+            self._cached_temp_folder = self._temp_folder_manager.resolve_temp_folder_name()  # noqa
+            return self._cached_temp_folder
+
+
+class _TestingMemmappingExecutor(MemmappingExecutor):
     """Wrapper around ReusableExecutor to ease memmapping testing with Pool
     and Executor. This is only for testing purposes.
-    """
-    def __init__(self, n_jobs, **backend_args):
-        self._executor = get_memmapping_executor(n_jobs, **backend_args)
-        self._temp_folder = self._executor._temp_folder
 
+    """
     def apply_async(self, func, args):
         """Schedule a func to be run"""
-        future = self._executor.submit(func, *args)
+        future = self.submit(func, *args)
         future.get = future.result
         return future
 
-    def terminate(self):
-        self._executor.shutdown()
-        self._unlink_temporary_resources()
-
     def map(self, f, *args):
-        res = self._executor.map(f, *args)
-        return list(res)
+        return list(super().map(f, *args))
diff --git a/joblib/externals/loky/__init__.py b/joblib/externals/loky/__init__.py
index 3ebc7e5..69ea578 100644
--- a/joblib/externals/loky/__init__.py
+++ b/joblib/externals/loky/__init__.py
@@ -22,4 +22,4 @@ __all__ = ["get_reusable_executor", "cpu_count", "wait", "as_completed",
            "wrap_non_picklable_objects", "set_loky_pickler"]
 
 
-__version__ = '2.7.0'
+__version__ = '2.8.0'
diff --git a/joblib/externals/loky/reusable_executor.py b/joblib/externals/loky/reusable_executor.py
index 11b6807..9a8e73f 100644
--- a/joblib/externals/loky/reusable_executor.py
+++ b/joblib/externals/loky/reusable_executor.py
@@ -81,65 +81,13 @@ def get_reusable_executor(max_workers=None, context=None, timeout=10,
     in the children before any module is loaded. This only works with with the
     ``loky`` context and it is unreliable on Windows with Python < 3.6.
     """
-    with _executor_lock:
-        global _executor, _executor_kwargs
-        executor = _executor
-
-        if max_workers is None:
-            if reuse is True and executor is not None:
-                max_workers = executor._max_workers
-            else:
-                max_workers = cpu_count()
-        elif max_workers <= 0:
-            raise ValueError(
-                "max_workers must be greater than 0, got {}."
-                .format(max_workers))
-
-        if isinstance(context, STRING_TYPE):
-            context = get_context(context)
-        if context is not None and context.get_start_method() == "fork":
-            raise ValueError("Cannot use reusable executor with the 'fork' "
-                             "context")
-
-        kwargs = dict(context=context, timeout=timeout,
-                      job_reducers=job_reducers,
-                      result_reducers=result_reducers,
-                      initializer=initializer, initargs=initargs,
-                      env=env)
-        if executor is None:
-            mp.util.debug("Create a executor with max_workers={}."
-                          .format(max_workers))
-            executor_id = _get_next_executor_id()
-            _executor_kwargs = kwargs
-            _executor = executor = _ReusablePoolExecutor(
-                _executor_lock, max_workers=max_workers,
-                executor_id=executor_id, **kwargs)
-        else:
-            if reuse == 'auto':
-                reuse = kwargs == _executor_kwargs
-            if (executor._flags.broken or executor._flags.shutdown
-                    or not reuse):
-                if executor._flags.broken:
-                    reason = "broken"
-                elif executor._flags.shutdown:
-                    reason = "shutdown"
-                else:
-                    reason = "arguments have changed"
-                mp.util.debug(
-                    "Creating a new executor with max_workers={} as the "
-                    "previous instance cannot be reused ({})."
-                    .format(max_workers, reason))
-                executor.shutdown(wait=True, kill_workers=kill_workers)
-                _executor = executor = _executor_kwargs = None
-                # Recursive call to build a new instance
-                return get_reusable_executor(max_workers=max_workers,
-                                             **kwargs)
-            else:
-                mp.util.debug("Reusing existing executor with max_workers={}."
-                              .format(executor._max_workers))
-                executor._resize(max_workers)
-
-    return executor
+    _executor, _ = _ReusablePoolExecutor.get_reusable_executor(
+        max_workers=max_workers, context=context, timeout=timeout,
+        kill_workers=kill_workers, reuse=reuse, job_reducers=job_reducers,
+        result_reducers=result_reducers, initializer=initializer,
+        initargs=initargs, env=env
+    )
+    return _executor
 
 
 class _ReusablePoolExecutor(ProcessPoolExecutor):
@@ -154,6 +102,76 @@ class _ReusablePoolExecutor(ProcessPoolExecutor):
         self.executor_id = executor_id
         self._submit_resize_lock = submit_resize_lock
 
+    @classmethod
+    def get_reusable_executor(cls, max_workers=None, context=None, timeout=10,
+                              kill_workers=False, reuse="auto",
+                              job_reducers=None, result_reducers=None,
+                              initializer=None, initargs=(), env=None):
+        with _executor_lock:
+            global _executor, _executor_kwargs
+            executor = _executor
+
+            if max_workers is None:
+                if reuse is True and executor is not None:
+                    max_workers = executor._max_workers
+                else:
+                    max_workers = cpu_count()
+            elif max_workers <= 0:
+                raise ValueError(
+                    "max_workers must be greater than 0, got {}."
+                    .format(max_workers))
+
+            if isinstance(context, STRING_TYPE):
+                context = get_context(context)
+            if context is not None and context.get_start_method() == "fork":
+                raise ValueError(
+                    "Cannot use reusable executor with the 'fork' context"
+                )
+
+            kwargs = dict(context=context, timeout=timeout,
+                          job_reducers=job_reducers,
+                          result_reducers=result_reducers,
+                          initializer=initializer, initargs=initargs,
+                          env=env)
+            if executor is None:
+                is_reused = False
+                mp.util.debug("Create a executor with max_workers={}."
+                              .format(max_workers))
+                executor_id = _get_next_executor_id()
+                _executor_kwargs = kwargs
+                _executor = executor = cls(
+                    _executor_lock, max_workers=max_workers,
+                    executor_id=executor_id, **kwargs)
+            else:
+                if reuse == 'auto':
+                    reuse = kwargs == _executor_kwargs
+                if (executor._flags.broken or executor._flags.shutdown
+                        or not reuse):
+                    if executor._flags.broken:
+                        reason = "broken"
+                    elif executor._flags.shutdown:
+                        reason = "shutdown"
+                    else:
+                        reason = "arguments have changed"
+                    mp.util.debug(
+                        "Creating a new executor with max_workers={} as the "
+                        "previous instance cannot be reused ({})."
+                        .format(max_workers, reason))
+                    executor.shutdown(wait=True, kill_workers=kill_workers)
+                    _executor = executor = _executor_kwargs = None
+                    # Recursive call to build a new instance
+                    return cls.get_reusable_executor(max_workers=max_workers,
+                                                     **kwargs)
+                else:
+                    mp.util.debug(
+                        "Reusing existing executor with max_workers={}."
+                        .format(executor._max_workers)
+                    )
+                    is_reused = True
+                    executor._resize(max_workers)
+
+        return executor, is_reused
+
     def submit(self, fn, *args, **kwargs):
         with self._submit_resize_lock:
             return super(_ReusablePoolExecutor, self).submit(
diff --git a/joblib/parallel.py b/joblib/parallel.py
index a849234..ee42510 100644
--- a/joblib/parallel.py
+++ b/joblib/parallel.py
@@ -14,6 +14,7 @@ import functools
 import time
 import threading
 import itertools
+from uuid import uuid4
 from numbers import Integral
 import warnings
 import queue
@@ -231,9 +232,11 @@ if hasattr(mp, 'get_context'):
 class BatchedCalls(object):
     """Wrap a sequence of (func, args, kwargs) tuples as a single callable"""
 
-    def __init__(self, iterator_slice, backend_and_jobs, pickle_cache=None):
+    def __init__(self, iterator_slice, backend_and_jobs, reducer_callback=None,
+                 pickle_cache=None):
         self.items = list(iterator_slice)
         self._size = len(self.items)
+        self._reducer_callback = reducer_callback
         if isinstance(backend_and_jobs, tuple):
             self._backend, self._n_jobs = backend_and_jobs
         else:
@@ -249,6 +252,16 @@ class BatchedCalls(object):
             return [func(*args, **kwargs)
                     for func, args, kwargs in self.items]
 
+    def __reduce__(self):
+        if self._reducer_callback is not None:
+            self._reducer_callback()
+        # no need pickle the callback.
+        return (
+            BatchedCalls,
+            (self.items, (self._backend, self._n_jobs), None,
+             self._pickle_cache)
+        )
+
     def __len__(self):
         return self._size
 
@@ -629,6 +642,8 @@ class Parallel(Logger):
         self.timeout = timeout
         self.pre_dispatch = pre_dispatch
         self._ready_batches = queue.Queue()
+        self._id = uuid4().hex
+        self._reducer_callback = None
 
         if isinstance(max_nbytes, str):
             max_nbytes = memstr_to_bytes(max_nbytes)
@@ -819,6 +834,7 @@ class Parallel(Logger):
                 for i in range(0, len(islice), final_batch_size):
                     tasks = BatchedCalls(islice[i:i + final_batch_size],
                                          self._backend.get_nested_backend(),
+                                         self._reducer_callback,
                                          self._pickle_cache)
                     self._ready_batches.put(tasks)
 
@@ -939,6 +955,27 @@ class Parallel(Logger):
         else:
             n_jobs = self._effective_n_jobs()
 
+        if isinstance(self._backend, LokyBackend):
+            # For the loky backend, we add a callback executed when reducing
+            # BatchCalls, that makes the loky executor use a temporary folder
+            # specific to this Parallel object when pickling temporary memmaps.
+            # This callback is necessary to ensure that several Parallel
+            # objects using the same resuable executor don't use the same
+            # temporary resources.
+
+            def _batched_calls_reducer_callback():
+                # Relevant implementation detail: the following lines, called
+                # when reducing BatchedCalls, are called in a thread-safe
+                # situation, meaning that the context of the temporary folder
+                # manager will not be changed in between the callback execution
+                # and the end of the BatchedCalls pickling. The reason is that
+                # pickling (the only place where set_current_context is used)
+                # is done from a single thread (the queue_feeder_thread).
+                self._backend._workers._temp_folder_manager.set_current_context(  # noqa
+                    self._id
+                )
+            self._reducer_callback = _batched_calls_reducer_callback
+
         # self._effective_n_jobs should be called in the Parallel.__call__
         # thread only -- store its value in an attribute for further queries.
         self._cached_effective_n_jobs = n_jobs
diff --git a/joblib/pool.py b/joblib/pool.py
index 67b185e..8443899 100644
--- a/joblib/pool.py
+++ b/joblib/pool.py
@@ -29,7 +29,7 @@ from pickle import HIGHEST_PROTOCOL
 from io import BytesIO
 
 from ._memmapping_reducer import get_memmapping_reducers
-from ._memmapping_reducer import TemporaryResourcesManagerMixin
+from ._memmapping_reducer import TemporaryResourcesManager
 from ._multiprocessing_helpers import mp, assert_spawning
 
 # We need the class definition to derive from it, not the multiprocessing.Pool
@@ -214,7 +214,7 @@ class PicklingPool(Pool):
         self._quick_get = self._outqueue._recv
 
 
-class MemmappingPool(PicklingPool, TemporaryResourcesManagerMixin):
+class MemmappingPool(PicklingPool):
     """Process pool that shares large arrays to avoid memory copy.
 
     This drop-in replacement for `multiprocessing.pool.Pool` makes
@@ -243,16 +243,20 @@ class MemmappingPool(PicklingPool, TemporaryResourcesManagerMixin):
         Callable executed on worker process creation.
     initargs: tuple, optional
         Arguments passed to the initializer callable.
-    temp_folder: str, optional
-        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 RAMdisk
-          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.
+    temp_folder: (str, callable) optional
+        If str:
+          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 RAMdisk
+            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.
+        if callable:
+            An callable in charge of dynamically resolving a temporary folder
+            for memmapping large arrays.
     max_nbytes int or None, optional, 1e6 by default
         Threshold on the size of arrays passed to the workers that
         triggers automated memory mapping in temp_folder.
@@ -296,10 +300,18 @@ class MemmappingPool(PicklingPool, TemporaryResourcesManagerMixin):
                           ' 0.9.4 and will be removed in 0.11',
                           DeprecationWarning)
 
-        forward_reducers, backward_reducers, self._temp_folder = \
+        manager = TemporaryResourcesManager(temp_folder)
+        self._temp_folder_manager = manager
+
+        # The usage of a temp_folder_resolver over a simple temp_folder is
+        # superfluous for multiprocessing pools, as they don't get reused, see
+        # get_memmapping_executor for more details. We still use it for code
+        # simplicity.
+        forward_reducers, backward_reducers = \
             get_memmapping_reducers(
-                id(self), temp_folder=temp_folder, max_nbytes=max_nbytes,
-                mmap_mode=mmap_mode, forward_reducers=forward_reducers,
+                temp_folder_resolver=manager.resolve_temp_folder_name,
+                max_nbytes=max_nbytes, mmap_mode=mmap_mode,
+                forward_reducers=forward_reducers,
                 backward_reducers=backward_reducers, verbose=verbose,
                 unlink_on_gc_collect=False, prewarm=prewarm)
 
@@ -324,4 +336,17 @@ class MemmappingPool(PicklingPool, TemporaryResourcesManagerMixin):
                     if i + 1 == n_retries:
                         warnings.warn("Failed to terminate worker processes in"
                                       " multiprocessing pool: %r" % e)
-        self._unlink_temporary_resources()
+        self._temp_folder_manager._unlink_temporary_resources()
+
+    @property
+    def _temp_folder(self):
+        # Legacy property in tests. could be removed if we refactored the
+        # memmapping tests. SHOULD ONLY BE USED IN TESTS!
+        # We cache this property because it is called late in the tests - at
+        # this point, all context have been unregistered, and
+        # resolve_temp_folder_name raises an error.
+        if getattr(self, '_cached_temp_folder', None) is not None:
+            return self._cached_temp_folder
+        else:
+            self._cached_temp_folder = self._temp_folder_manager.resolve_temp_folder_name()  # noqa
+            return self._cached_temp_folder


Test patch:
diff --git a/joblib/test/test_dask.py b/joblib/test/test_dask.py
index 9ef575e..2b5b7dd 100644
--- a/joblib/test/test_dask.py
+++ b/joblib/test/test_dask.py
@@ -117,7 +117,7 @@ def test_auto_scatter(loop):
     np = pytest.importorskip('numpy')
     data1 = np.ones(int(1e4), dtype=np.uint8)
     data2 = np.ones(int(1e4), dtype=np.uint8)
-    data_to_process = ([data1] * 4) + ([data2] * 4)
+    data_to_process = ([data1] * 3) + ([data2] * 3)
 
     def count_events(event_name, client):
         worker_events = client.run(lambda dask_worker: dask_worker.log)
@@ -138,7 +138,8 @@ def test_auto_scatter(loop):
             # broadcast=1 which means that one worker must directly receive
             # the data from the scatter operation once.
             counts = count_events('receive-from-scatter', client)
-            assert counts[a['address']] + counts[b['address']] == 2
+            # assert counts[a['address']] + counts[b['address']] == 2
+            assert 2 <= counts[a['address']] + counts[b['address']] <= 4
 
     with cluster() as (s, [a, b]):
         with Client(s['address'], loop=loop) as client:
diff --git a/joblib/test/test_memmapping.py b/joblib/test/test_memmapping.py
index 4013efa..dc40d23 100644
--- a/joblib/test/test_memmapping.py
+++ b/joblib/test/test_memmapping.py
@@ -7,6 +7,7 @@ import pickle
 import itertools
 from time import sleep
 import subprocess
+import threading
 
 from joblib.test.common import with_numpy, np
 from joblib.test.common import setup_autokill
@@ -18,7 +19,7 @@ from joblib.backports import make_memmap
 from joblib.parallel import Parallel, delayed
 
 from joblib.pool import MemmappingPool
-from joblib.executor import _TestingMemmappingExecutor
+from joblib.executor import _TestingMemmappingExecutor as TestExecutor
 from joblib._memmapping_reducer import has_shareable_memory
 from joblib._memmapping_reducer import ArrayMemmapForwardReducer
 from joblib._memmapping_reducer import _strided_from_memmap
@@ -262,7 +263,7 @@ def test__strided_from_memmap(tmpdir):
 
 @with_numpy
 @with_multiprocessing
-@parametrize("factory", [MemmappingPool, _TestingMemmappingExecutor],
+@parametrize("factory", [MemmappingPool, TestExecutor.get_memmapping_executor],
              ids=["multiprocessing", "loky"])
 def test_pool_with_memmap(factory, tmpdir):
     """Check that subprocess can access and update shared memory memmap"""
@@ -316,7 +317,7 @@ def test_pool_with_memmap(factory, tmpdir):
 
 @with_numpy
 @with_multiprocessing
-@parametrize("factory", [MemmappingPool, _TestingMemmappingExecutor],
+@parametrize("factory", [MemmappingPool, TestExecutor.get_memmapping_executor],
              ids=["multiprocessing", "loky"])
 def test_pool_with_memmap_array_view(factory, tmpdir):
     """Check that subprocess can access and update shared memory array"""
@@ -436,6 +437,235 @@ def test_permission_error_windows_memmap_sent_to_parent(backend):
             assert b'resource_tracker' not in err
 
 
+@with_numpy
+@with_multiprocessing
+@parametrize("backend", ["multiprocessing", "loky"])
+def test_parallel_isolated_temp_folders(backend):
+    # Test that consecutive Parallel call use isolated subfolders, even
+    # for the loky backend that reuses its executor instance across calls.
+    array = np.arange(int(1e2))
+    [filename_1] = Parallel(n_jobs=2, backend=backend, max_nbytes=10)(
+        delayed(getattr)(array, 'filename') for _ in range(1)
+    )
+    [filename_2] = Parallel(n_jobs=2, backend=backend, max_nbytes=10)(
+        delayed(getattr)(array, 'filename') for _ in range(1)
+    )
+    assert os.path.dirname(filename_2) != os.path.dirname(filename_1)
+
+
+@with_numpy
+@with_multiprocessing
+@parametrize("backend", ["multiprocessing", "loky"])
+def test_managed_backend_reuse_temp_folder(backend):
+    # Test that calls to a managed parallel object reuse the same memmaps.
+    array = np.arange(int(1e2))
+    with Parallel(n_jobs=2, backend=backend, max_nbytes=10) as p:
+        [filename_1] = p(
+            delayed(getattr)(array, 'filename') for _ in range(1)
+        )
+        [filename_2] = p(
+            delayed(getattr)(array, 'filename') for _ in range(1)
+        )
+    assert os.path.dirname(filename_2) == os.path.dirname(filename_1)
+
+
+@with_numpy
+@with_multiprocessing
+def test_memmapping_temp_folder_thread_safety():
+    # Concurrent calls to Parallel with the loky backend will use the same
+    # executor, and thus the same reducers. Make sure that those reducers use
+    # different temporary folders depending on which Parallel objects called
+    # them, which is necessary to limit potential race conditions during the
+    # garbage collection of temporary memmaps.
+    array = np.arange(int(1e2))
+
+    temp_dirs_thread_1 = set()
+    temp_dirs_thread_2 = set()
+
+    def concurrent_get_filename(array, temp_dirs):
+        with Parallel(backend='loky', n_jobs=2, max_nbytes=10) as p:
+            for i in range(10):
+                [filename] = p(
+                    delayed(getattr)(array, 'filename') for _ in range(1)
+                )
+                temp_dirs.add(os.path.dirname(filename))
+
+    t1 = threading.Thread(
+        target=concurrent_get_filename, args=(array, temp_dirs_thread_1)
+    )
+    t2 = threading.Thread(
+        target=concurrent_get_filename, args=(array, temp_dirs_thread_2)
+    )
+
+    t1.start()
+    t2.start()
+
+    t1.join()
+    t2.join()
+
+    assert len(temp_dirs_thread_1) == 1
+    assert len(temp_dirs_thread_2) == 1
+
+    assert temp_dirs_thread_1 != temp_dirs_thread_2
+
+
+@with_numpy
+@with_multiprocessing
+def test_multithreaded_parallel_termination_resource_tracker_silent():
+    # test that concurrent termination attempts of a same executor does not
+    # emit any spurious error from the resource_tracker. We test various
+    # situations making 0, 1 or both parallel call sending a task that will
+    # make the worker (and thus the whole Parallel call) error out.
+    cmd = '''if 1:
+        import os
+        import numpy as np
+        from joblib import Parallel, delayed
+        from joblib.externals.loky.backend import resource_tracker
+        from concurrent.futures import ThreadPoolExecutor, wait
+
+        resource_tracker.VERBOSE = 0
+
+        array = np.arange(int(1e2))
+
+        temp_dirs_thread_1 = set()
+        temp_dirs_thread_2 = set()
+
+
+        def raise_error(array):
+            raise ValueError
+
+
+        def parallel_get_filename(array, temp_dirs):
+            with Parallel(backend="loky", n_jobs=2, max_nbytes=10) as p:
+                for i in range(10):
+                    [filename] = p(
+                        delayed(getattr)(array, "filename") for _ in range(1)
+                    )
+                    temp_dirs.add(os.path.dirname(filename))
+
+
+        def parallel_raise(array, temp_dirs):
+            with Parallel(backend="loky", n_jobs=2, max_nbytes=10) as p:
+                for i in range(10):
+                    [filename] = p(
+                        delayed(raise_error)(array) for _ in range(1)
+                    )
+                    temp_dirs.add(os.path.dirname(filename))
+
+
+        executor = ThreadPoolExecutor(max_workers=2)
+
+        # both function calls will use the same loky executor, but with a
+        # different Parallel object.
+        future_1 = executor.submit({f1}, array, temp_dirs_thread_1)
+        future_2 = executor.submit({f2}, array, temp_dirs_thread_2)
+
+        # Wait for both threads to terminate their backend
+        wait([future_1, future_2])
+
+        future_1.result()
+        future_2.result()
+    '''
+    functions_and_returncodes = [
+        ("parallel_get_filename", "parallel_get_filename", 0),
+        ("parallel_get_filename", "parallel_raise", 1),
+        ("parallel_raise", "parallel_raise", 1)
+    ]
+
+    for f1, f2, returncode in functions_and_returncodes:
+        p = subprocess.Popen([sys.executable, '-c', cmd.format(f1=f1, f2=f2)],
+                             stderr=subprocess.PIPE, stdout=subprocess.PIPE)
+        p.wait()
+        out, err = p.communicate()
+        assert p.returncode == returncode, out.decode()
+        assert b"resource_tracker" not in err, err.decode()
+
+
+@with_numpy
+@with_multiprocessing
+def test_nested_loop_error_in_grandchild_resource_tracker_silent():
+    # Safety smoke test: test that nested parallel calls using the loky backend
+    # don't yield noisy resource_tracker outputs when the grandchild errors
+    # out.
+    cmd = '''if 1:
+        from joblib import Parallel, delayed
+
+
+        def raise_error(i):
+            raise ValueError
+
+
+        def nested_loop(f):
+            Parallel(backend="loky", n_jobs=2)(
+                delayed(f)(i) for i in range(10)
+            )
+
+
+        if __name__ == "__main__":
+            Parallel(backend="loky", n_jobs=2)(
+                delayed(nested_loop)(func) for func in [raise_error]
+            )
+    '''
+    p = subprocess.Popen([sys.executable, '-c', cmd],
+                         stderr=subprocess.PIPE, stdout=subprocess.PIPE)
+    p.wait()
+    out, err = p.communicate()
+    assert p.returncode == 1, out.decode()
+    assert b"resource_tracker" not in err, err.decode()
+
+
+@with_numpy
+@with_multiprocessing
+@parametrize("backend", ["multiprocessing", "loky"])
+def test_many_parallel_calls_on_same_object(backend):
+    # After #966 got merged, consecutive Parallel objects were sharing temp
+    # folder, which would lead to race conditions happening during the
+    # temporary resources management with the resource_tracker. This is a
+    # non-regression test that makes sure that consecutive Parallel operations
+    # on the same object do not error out.
+    cmd = '''if 1:
+        import os
+        import time
+
+        import numpy as np
+
+        from joblib import Parallel, delayed
+        from testutils import return_slice_of_data
+
+        data = np.ones(100)
+
+        if __name__ == '__main__':
+            for i in range(5):
+                slice_of_data = Parallel(
+                    n_jobs=2, max_nbytes=1, backend='{b}')(
+                        delayed(return_slice_of_data)(data, 0, 20)
+                        for _ in range(10)
+                    )
+                slice_of_data = Parallel(
+                    n_jobs=2, max_nbytes=1, backend='{b}')(
+                        delayed(return_slice_of_data)(data, 0, 20)
+                        for _ in range(10)
+                    )
+    '''.format(b=backend)
+
+    for _ in range(3):
+        env = os.environ.copy()
+        env['PYTHONPATH'] = os.path.dirname(__file__)
+        p = subprocess.Popen([sys.executable, '-c', cmd],
+                             stderr=subprocess.PIPE,
+                             stdout=subprocess.PIPE, env=env)
+        p.wait()
+        out, err = p.communicate()
+        assert p.returncode == 0, err
+        assert out == b''
+        if sys.version_info[:3] not in [(3, 8, 0), (3, 8, 1)]:
+            # In early versions of Python 3.8, a reference leak
+            # https://github.com/cloudpipe/cloudpickle/issues/327, holds
+            # references to pickled objects, generating race condition during
+            # cleanup finalizers of joblib and noisy resource_tracker outputs.
+            assert b'resource_tracker' not in err
+
+
 @with_numpy
 @with_multiprocessing
 @parametrize("backend", ["multiprocessing", "loky"])
@@ -490,7 +720,7 @@ def test_resource_tracker_silent_when_reference_cycles(backend):
 
 @with_numpy
 @with_multiprocessing
-@parametrize("factory", [MemmappingPool, _TestingMemmappingExecutor],
+@parametrize("factory", [MemmappingPool, TestExecutor.get_memmapping_executor],
              ids=["multiprocessing", "loky"])
 def test_memmapping_pool_for_large_arrays(factory, tmpdir):
     """Check that large arrays are not copied in memory"""
@@ -574,7 +804,7 @@ def test_child_raises_parent_exits_cleanly(backend):
 
         def get_temp_folder(parallel_obj, backend):
             if "{b}" == "loky":
-                return p._backend._temp_folder
+                return p._backend._workers._temp_folder
             else:
                 return p._backend._pool._temp_folder
 
@@ -605,7 +835,7 @@ def test_child_raises_parent_exits_cleanly(backend):
 
 @with_numpy
 @with_multiprocessing
-@parametrize("factory", [MemmappingPool, _TestingMemmappingExecutor],
+@parametrize("factory", [MemmappingPool, TestExecutor.get_memmapping_executor],
              ids=["multiprocessing", "loky"])
 def test_memmapping_pool_for_large_arrays_disabled(factory, tmpdir):
     """Check that large arrays memmapping can be disabled"""
@@ -633,7 +863,7 @@ def test_memmapping_pool_for_large_arrays_disabled(factory, tmpdir):
 @with_numpy
 @with_multiprocessing
 @with_dev_shm
-@parametrize("factory", [MemmappingPool, _TestingMemmappingExecutor],
+@parametrize("factory", [MemmappingPool, TestExecutor.get_memmapping_executor],
              ids=["multiprocessing", "loky"])
 def test_memmapping_on_large_enough_dev_shm(factory):
     """Check that memmapping uses /dev/shm when possible"""
@@ -687,7 +917,7 @@ def test_memmapping_on_large_enough_dev_shm(factory):
 @with_numpy
 @with_multiprocessing
 @with_dev_shm
-@parametrize("factory", [MemmappingPool, _TestingMemmappingExecutor],
+@parametrize("factory", [MemmappingPool, TestExecutor.get_memmapping_executor],
              ids=["multiprocessing", "loky"])
 def test_memmapping_on_too_small_dev_shm(factory):
     orig_size = jmr.SYSTEM_SHARED_MEM_FS_MIN_SIZE
@@ -715,7 +945,7 @@ def test_memmapping_on_too_small_dev_shm(factory):
 
 @with_numpy
 @with_multiprocessing
-@parametrize("factory", [MemmappingPool, _TestingMemmappingExecutor],
+@parametrize("factory", [MemmappingPool, TestExecutor.get_memmapping_executor],
              ids=["multiprocessing", "loky"])
 def test_memmapping_pool_for_large_arrays_in_return(factory, tmpdir):
     """Check that large arrays are not copied in memory in return"""
@@ -747,7 +977,7 @@ def _worker_multiply(a, n_times):
 
 @with_numpy
 @with_multiprocessing
-@parametrize("factory", [MemmappingPool, _TestingMemmappingExecutor],
+@parametrize("factory", [MemmappingPool, TestExecutor.get_memmapping_executor],
              ids=["multiprocessing", "loky"])
 def test_workaround_against_bad_memmap_with_copied_buffers(factory, tmpdir):
     """Check that memmaps with a bad buffer are returned as regular arrays
@@ -782,8 +1012,8 @@ def identity(arg):
 @with_multiprocessing
 @parametrize(
     "factory,retry_no",
-    list(itertools.product([MemmappingPool, _TestingMemmappingExecutor],
-                           range(3))),
+    list(itertools.product(
+        [MemmappingPool, TestExecutor.get_memmapping_executor], range(3))),
     ids=['{}, {}'.format(x, y) for x, y in itertools.product(
         ["multiprocessing", "loky"], map(str, range(3)))])
 def test_pool_memmap_with_big_offset(factory, retry_no, tmpdir):