Repository: joblib/joblib
Pull request #1025 (2020-03-21 23:46:10)

Collect Dask results as they complete
Previously we would wait until joblib called lazy_result.get() before
collecting results.  This would trigger a transfer in the main thread,
which would block things a bit.

Now we collect results as soon as they are available using
dask.distributed.as_completed.  This helps to reduce overhead in joblib
a bit and improve overall bandwidth.

Somewhat related to (but doesn't entirely fix) https://github.com/joblib/joblib/issues/1020 and https://github.com/dask/dask/issues/5993

There are still performance issues.  We have a lot of downtime in this process, but it's not happening in Dask (all of our profilers show that we're not spending a ton of time in Dask code).  I suspect that something on the joblib end is blocking things, but profiling so far hasn't shown anything.

mrocklin: This currently fails on `test_auto_scatter`
codecov[bot]: # [Codecov](https://codecov.io/gh/joblib/joblib/pull/1025?src=pr&el=h1) Report
> Merging [#1025](https://codecov.io/gh/joblib/joblib/pull/1025?src=pr&el=desc) into [master](https://codecov.io/gh/joblib/joblib/commit/e426cb731a530ebaf8415e3992795e2eae88f317&el=desc) will **decrease** coverage by `8.43%`.
> The diff coverage is `14.28%`.

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

```diff
@@            Coverage Diff             @@
##           master    #1025      +/-   ##
==========================================
- Coverage   94.81%   86.37%   -8.44%     
==========================================
  Files          45       45              
  Lines        6803     6820      +17     
==========================================
- Hits         6450     5891     -559     
- Misses        353      929     +576     
```


| [Impacted Files](https://codecov.io/gh/joblib/joblib/pull/1025?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [joblib/\_dask.py](https://codecov.io/gh/joblib/joblib/pull/1025/diff?src=pr&el=tree#diff-am9ibGliL19kYXNrLnB5) | `23.73% <14.28%> (-70.74%)` | :arrow_down: |
| [joblib/test/test\_dask.py](https://codecov.io/gh/joblib/joblib/pull/1025/diff?src=pr&el=tree#diff-am9ibGliL3Rlc3QvdGVzdF9kYXNrLnB5) | `4.41% <0.00%> (-90.69%)` | :arrow_down: |
| [joblib/\_memory\_helpers.py](https://codecov.io/gh/joblib/joblib/pull/1025/diff?src=pr&el=tree#diff-am9ibGliL19tZW1vcnlfaGVscGVycy5weQ==) | `3.07% <0.00%> (-70.77%)` | :arrow_down: |
| [joblib/backports.py](https://codecov.io/gh/joblib/joblib/pull/1025/diff?src=pr&el=tree#diff-am9ibGliL2JhY2twb3J0cy5weQ==) | `33.33% <0.00%> (-56.26%)` | :arrow_down: |
| [joblib/\_compat.py](https://codecov.io/gh/joblib/joblib/pull/1025/diff?src=pr&el=tree#diff-am9ibGliL19jb21wYXQucHk=) | `80.00% <0.00%> (-20.00%)` | :arrow_down: |
| [joblib/\_multiprocessing\_helpers.py](https://codecov.io/gh/joblib/joblib/pull/1025/diff?src=pr&el=tree#diff-am9ibGliL19tdWx0aXByb2Nlc3NpbmdfaGVscGVycy5weQ==) | `70.58% <0.00%> (-14.71%)` | :arrow_down: |
| [joblib/format\_stack.py](https://codecov.io/gh/joblib/joblib/pull/1025/diff?src=pr&el=tree#diff-am9ibGliL2Zvcm1hdF9zdGFjay5weQ==) | `71.77% <0.00%> (-12.92%)` | :arrow_down: |
| [joblib/pool.py](https://codecov.io/gh/joblib/joblib/pull/1025/diff?src=pr&el=tree#diff-am9ibGliL3Bvb2wucHk=) | `85.34% <0.00%> (-9.49%)` | :arrow_down: |
| [joblib/my\_exceptions.py](https://codecov.io/gh/joblib/joblib/pull/1025/diff?src=pr&el=tree#diff-am9ibGliL215X2V4Y2VwdGlvbnMucHk=) | `88.67% <0.00%> (-9.44%)` | :arrow_down: |
| ... and [17 more](https://codecov.io/gh/joblib/joblib/pull/1025/diff?src=pr&el=tree-more) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/joblib/joblib/pull/1025?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/1025?src=pr&el=footer). Last update [e426cb7...bfba768](https://codecov.io/gh/joblib/joblib/pull/1025?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

lesteve: I looked a bit at the failure on Python 3 (Python 2 failures can be ignored since there is a non-Python 2 compatible change in this PR)  and I can reproduce it locally, it manifests itself as a "hang" (sorry for the vague term) i.e. the `Parallel` call never completes:
https://github.com/joblib/joblib/blob/e426cb731a530ebaf8415e3992795e2eae88f317/joblib/test/test_dask.py#L135-L136

What I observed is that changing the input data slightly, the "hang" goes away and the test pass. I just pushed a commit to see whether that was the case on the CI. I don't have any clue why this is the case at the moment.


lesteve: > What I observed is that changing the input data slightly, the "hang" goes away and the test pass. I just pushed a commit to see whether that was the case on the CI. I don't have any clue why this is the case at the moment.

So this quirk is confirmed on the CI:
- timeout after 1 hour: https://dev.azure.com/joblib/joblib/_build/results?buildId=470&view=logs&j=9ab96c21-2bfa-5360-3f04-0ac557a9e9b9
- pass in ~3 minutes with my last commit: https://dev.azure.com/joblib/joblib/_build/results?buildId=471&view=logs&j=9ab96c21-2bfa-5360-3f04-0ac557a9e9b9


mrocklin: > Side-comment: the "hang" was fully deterministic locally i.e. it does not seem due to some race condition.

Right, I wonder if the imbalance of data before was intentional some way.  Maybe the current implementation is not as robust to some bad user inputs.
mrocklin: > Python 2 failures can be ignored since there is a non-Python 2 compatible change in this PR

Is Joblib still supporting Python 2?  If so, I can try to switch to older API.
lesteve: > Is Joblib still supporting Python 2? If so, I can try to switch to older API.

I have not been following joblib very closely, but there is an ongoing PR for dropping Python 2: #1018. I would not worry too much about Python 2 support.
pierreglaser: Thank you very much for this fix @mrocklin, I can confirm this PR fixes at least #957
pierreglaser: @mrocklin we merged #1018 into master, so you might want to rebase.
mrocklin: Force pushed
mrocklin: I'm not sure that I understand the test failures.  Are they unrelated perhaps?
pierreglaser: Probably. Feel free to push an empty commit to trigger the CI again.

Patch:
diff --git a/azure-pipelines.yml b/azure-pipelines.yml
index aba849c..375a1b6 100644
--- a/azure-pipelines.yml
+++ b/azure-pipelines.yml
@@ -70,9 +70,6 @@ jobs:
         imageName: "vs2017-win2016"
         PYTHON_VERSION: "3.8"
         EXTRA_CONDA_PACKAGES: "numpy=1.18"
-      windows_py36_no_numpy:
-        imageName: "vs2017-win2016"
-        PYTHON_VERSION: "3.6"
 
       macos_py38:
         imageName: "macos-10.14"
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 39681ac..1e069fb 100644
--- a/joblib/_memmapping_reducer.py
+++ b/joblib/_memmapping_reducer.py
@@ -12,9 +12,11 @@ import stat
 import threading
 import atexit
 import tempfile
+import time
 import warnings
 import weakref
 from uuid import uuid4
+from multiprocessing import util
 
 from pickle import whichmodule, loads, dumps, HIGHEST_PROTOCOL, PicklingError
 
@@ -29,10 +31,10 @@ try:
 except ImportError:
     np = None
 
-from .numpy_pickle import load
-from .numpy_pickle import dump
+from .numpy_pickle import dump, load, load_temporary_memmap
 from .backports import make_memmap
 from .disk import delete_folder
+from .externals.loky.backend import resource_tracker
 
 # Some system have a ramdisk mounted by default, we can use it instead of /tmp
 # as the default folder to dump big arrays to share with subprocesses.
@@ -48,6 +50,59 @@ SYSTEM_SHARED_MEM_FS_MIN_SIZE = int(2e9)
 FOLDER_PERMISSIONS = stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR
 FILE_PERMISSIONS = stat.S_IRUSR | stat.S_IWUSR
 
+# Set used in joblib workers, referencing the filenames of temporary memmaps
+# created by joblib to speed up data communication. In child processes, we add
+# a finalizer to these memmaps that sends a maybe_unlink call to the
+# resource_tracker, in order to free main memory as fast as possible.
+JOBLIB_MMAPS = set()
+
+
+def _log_and_unlink(filename):
+    from .externals.loky.backend.resource_tracker import _resource_tracker
+    util.debug(
+        "[FINALIZER CALL] object mapping to {} about to be deleted,"
+        " decrementing the refcount of the file (pid: {})".format(
+            os.path.basename(filename), os.getpid()))
+    _resource_tracker.maybe_unlink(filename, "file")
+
+
+def add_maybe_unlink_finalizer(memmap):
+    util.debug(
+        "[FINALIZER ADD] adding finalizer to {} (id {}, filename {}, pid  {})"
+        "".format(type(memmap), id(memmap), os.path.basename(memmap.filename),
+                  os.getpid()))
+    weakref.finalize(memmap, _log_and_unlink, memmap.filename)
+
+
+def unlink_file(filename):
+    """Wrapper around os.unlink with a retry mechanism.
+
+    The retry mechanism has been implemented primarily to overcome a race
+    condition happening during the finalizer of a np.memmap: when a process
+    holding the last reference to a mmap-backed np.memmap/np.array is about to
+    delete this array (and close the reference), it sends a maybe_unlink
+    request to the resource_tracker. This request can be processed faster than
+    it takes for the last reference of the memmap to be closed, yielding (on
+    Windows) a PermissionError in the resource_tracker loop.
+    """
+    NUM_RETRIES = 10
+    for retry_no in range(1, NUM_RETRIES + 1):
+        try:
+            os.unlink(filename)
+            break
+        except PermissionError:
+            util.debug(
+                '[ResourceTracker] tried to unlink {}, got '
+                'PermissionError'.format(filename)
+            )
+            if retry_no == NUM_RETRIES:
+                raise
+            else:
+                time.sleep(.2)
+
+
+resource_tracker._CLEANUP_FUNCS['file'] = unlink_file
+
 
 class _WeakArrayKeyMap:
     """A variant of weakref.WeakKeyDictionary for unhashable numpy arrays.
@@ -175,7 +230,7 @@ def has_shareable_memory(a):
 
 
 def _strided_from_memmap(filename, dtype, mode, offset, order, shape, strides,
-                         total_buffer_len):
+                         total_buffer_len, unlink_on_gc_collect):
     """Reconstruct an array view on a memory mapped file."""
     if mode == 'w+':
         # Do not zero the original data when unpickling
@@ -183,13 +238,17 @@ def _strided_from_memmap(filename, dtype, mode, offset, order, shape, strides,
 
     if strides is None:
         # Simple, contiguous memmap
-        return make_memmap(filename, dtype=dtype, shape=shape, mode=mode,
-                           offset=offset, order=order)
+        return make_memmap(
+            filename, dtype=dtype, shape=shape, mode=mode, offset=offset,
+            order=order, unlink_on_gc_collect=unlink_on_gc_collect
+        )
     else:
         # For non-contiguous data, memmap the total enclosing buffer and then
         # extract the non-contiguous view with the stride-tricks API
-        base = make_memmap(filename, dtype=dtype, shape=total_buffer_len,
-                           mode=mode, offset=offset, order=order)
+        base = make_memmap(
+            filename, dtype=dtype, shape=total_buffer_len, offset=offset,
+            mode=mode, order=order, unlink_on_gc_collect=unlink_on_gc_collect
+        )
         return as_strided(base, shape=shape, strides=strides)
 
 
@@ -201,6 +260,8 @@ def _reduce_memmap_backed(a, m):
     attribute ancestry of a. ``m.base`` should be the real python mmap object.
     """
     # offset that comes from the striding differences between a and m
+    util.debug('[MEMMAP REDUCE] reducing a memmap-backed array '
+               '(shape, {}, pid: {})'.format(a.shape, os.getpid()))
     a_start, a_end = np.byte_bounds(a)
     m_start = np.byte_bounds(m)[0]
     offset = a_start - m_start
@@ -224,26 +285,32 @@ def _reduce_memmap_backed(a, m):
         # view will be extracted.
         strides = a.strides
         total_buffer_len = (a_end - a_start) // a.itemsize
+
     return (_strided_from_memmap,
             (m.filename, a.dtype, m.mode, offset, order, a.shape, strides,
-             total_buffer_len))
+             total_buffer_len, False))
 
 
-def reduce_memmap(a):
-    """Pickle the descriptors of a memmap instance to reopen on same file."""
+def reduce_array_memmap_backward(a):
+    """reduce a np.array or a np.memmap from a child process"""
     m = _get_backing_memmap(a)
-    if m is not None:
-        # m is a real mmap backed memmap instance, reduce a preserving striding
-        # information
+    if isinstance(m, np.memmap) and m.filename not in JOBLIB_MMAPS:
+        # if a is backed by a memmaped file, reconstruct a using the
+        # memmaped file.
         return _reduce_memmap_backed(a, m)
     else:
-        # This memmap instance is actually backed by a regular in-memory
-        # buffer: this can happen when using binary operators on numpy.memmap
-        # instances
-        return (loads, (dumps(np.asarray(a), protocol=HIGHEST_PROTOCOL),))
-
-
-class ArrayMemmapReducer(object):
+        # a is either a regular (not memmap-backed) numpy array, or an array
+        # backed by a shared temporary file created by joblib. In the latter
+        # case, in order to limit the lifespan of these temporary files, we
+        # serialize the memmap as a regular numpy array, and decref the
+        # file backing the memmap (done implicitly in a previously registered
+        # finalizer, see ``unlink_on_gc_collect`` for more details)
+        return (
+            loads, (dumps(np.asarray(a), protocol=HIGHEST_PROTOCOL), )
+        )
+
+
+class ArrayMemmapForwardReducer(object):
     """Reducer callable to dump large arrays to memmap files.
 
     Parameters
@@ -267,26 +334,29 @@ class ArrayMemmapReducer(object):
         same data array is passed to different worker processes.
     """
 
-    def __init__(self, max_nbytes, temp_folder, mmap_mode, verbose=0,
-                 prewarm=True):
+    def __init__(self, max_nbytes, temp_folder, mmap_mode,
+                 unlink_on_gc_collect, verbose=0, prewarm=True):
         self._max_nbytes = max_nbytes
         self._temp_folder = temp_folder
         self._mmap_mode = mmap_mode
         self.verbose = int(verbose)
         self._prewarm = prewarm
         self._memmaped_arrays = _WeakArrayKeyMap()
+        self._temporary_memmaped_filenames = set()
+        self._unlink_on_gc_collect = unlink_on_gc_collect
 
     def __reduce__(self):
-        # The ArrayMemmapReducer 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
+        # 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)
+        args = (self._max_nbytes, self._temp_folder, self._mmap_mode,
+                self._unlink_on_gc_collect)
         kwargs = {
             'verbose': self.verbose,
             'prewarm': self._prewarm,
         }
-        return ArrayMemmapReducer, args, kwargs
+        return ArrayMemmapForwardReducer, args, kwargs
 
     def __call__(self, a):
         m = _get_backing_memmap(a)
@@ -320,13 +390,39 @@ class ArrayMemmapReducer(object):
             # In case the same array with the same content is passed several
             # times to the pool subprocess children, serialize it only once
 
-            # XXX: implement an explicit reference counting scheme to make it
-            # possible to delete temporary files as soon as the workers are
-            # done processing this data.
+            is_new_memmap = filename not in self._temporary_memmaped_filenames
+
+            # add the memmap to the list of temporary memmaps created by joblib
+            self._temporary_memmaped_filenames.add(filename)
+
+            if self._unlink_on_gc_collect:
+                # Bump reference count of the memmap by 1 to account for
+                # shared usage of the memmap by a child process. The
+                # corresponding decref call will be executed upon calling
+                # resource_tracker.maybe_unlink, registered as a finalizer in
+                # the child.
+                # the incref/decref calls here are only possible when the child
+                # and the parent share the same resource_tracker. It is not the
+                # case for the multiprocessing backend, but it does not matter
+                # because unlinking a memmap from a child process is only
+                # useful to control the memory usage of long-lasting child
+                # processes, while the multiprocessing-based pools terminate
+                # their workers at the end of a map() call.
+                resource_tracker.register(filename, "file")
+
+            if is_new_memmap:
+                # Incref each temporary memmap created by joblib one extra
+                # time.  This means that these memmaps will only be deleted
+                # once an extra maybe_unlink() is called, which is done once
+                # all the jobs have completed (or been canceled) in the
+                # Parallel._terminate_backend() method.
+                resource_tracker.register(filename, "file")
+
             if not os.path.exists(filename):
-                if self.verbose > 0:
-                    print("Memmapping (shape={}, dtype={}) to new file {}"
-                          .format(a.shape, a.dtype, filename))
+                util.debug(
+                    "[ARRAY DUMP] Pickling new array (shape={}, dtype={}) "
+                    "creating a new memmap".format(
+                        a.shape, a.dtype, os.path.basename(filename)))
                 for dumped_filename in dump(a, filename):
                     os.chmod(dumped_filename, FILE_PERMISSIONS)
 
@@ -337,25 +433,31 @@ class ArrayMemmapReducer(object):
                     # concurrent memmap creation in multiple children
                     # processes.
                     load(filename, mmap_mode=self._mmap_mode).max()
-            elif self.verbose > 1:
-                print("Memmapping (shape={}, dtype={}) to old file {}"
-                      .format(a.shape, a.dtype, filename))
+
+            else:
+                util.debug(
+                    "[ARRAY DUMP] Pickling known array (shape={}, dtype={}) "
+                    "reusing memmap file: {}".format(
+                        a.shape, a.dtype, os.path.basename(filename)))
 
             # The worker process will use joblib.load to memmap the data
-            return (load, (filename, self._mmap_mode))
+            return (
+                (load_temporary_memmap, (filename, self._mmap_mode,
+                                         self._unlink_on_gc_collect))
+            )
         else:
             # do not convert a into memmap, let pickler do its usual copy with
             # the default system pickler
-            if self.verbose > 1:
-                print("Pickling array (shape={}, dtype={})."
-                      .format(a.shape, a.dtype))
+            util.debug(
+                '[ARRAY DUMP] Pickling array (NO MEMMAPPING) (shape={}, '
+                ' dtype={}).'.format(a.shape, a.dtype))
             return (loads, (dumps(a, protocol=HIGHEST_PROTOCOL),))
 
 
 def get_memmapping_reducers(
         pool_id, forward_reducers=None, backward_reducers=None,
         temp_folder=None, max_nbytes=1e6, mmap_mode='r', verbose=0,
-        prewarm=False, **kwargs):
+        prewarm=False, unlink_on_gc_collect=True, **kwargs):
     """Construct a pair of memmapping reducer linked to a tmpdir.
 
     This function manage the creation and the clean up of the temporary folders
@@ -381,6 +483,7 @@ def get_memmapping_reducers(
     # 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
@@ -394,8 +497,9 @@ def get_memmapping_reducers(
         delete_folder = __import__(
             pool_module_name, fromlist=['delete_folder']).delete_folder
         try:
-            delete_folder(pool_folder)
-        except WindowsError:
+            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))
 
@@ -407,19 +511,47 @@ def get_memmapping_reducers(
         # arrays over the max_nbytes threshold
         if prewarm == "auto":
             prewarm = not use_shared_mem
-        forward_reduce_ndarray = ArrayMemmapReducer(
-            max_nbytes, pool_folder, mmap_mode, verbose,
+        forward_reduce_ndarray = ArrayMemmapForwardReducer(
+            max_nbytes, pool_folder, mmap_mode, unlink_on_gc_collect, verbose,
             prewarm=prewarm)
         forward_reducers[np.ndarray] = forward_reduce_ndarray
-        forward_reducers[np.memmap] = reduce_memmap
+        forward_reducers[np.memmap] = forward_reduce_ndarray
 
         # Communication from child process to the parent process always
         # pickles in-memory numpy.ndarray without dumping them as memmap
         # to avoid confusing the caller and make it tricky to collect the
         # temporary folder
-        backward_reduce_ndarray = ArrayMemmapReducer(
-            None, pool_folder, mmap_mode, verbose)
-        backward_reducers[np.ndarray] = backward_reduce_ndarray
-        backward_reducers[np.memmap] = reduce_memmap
+        backward_reducers[np.ndarray] = reduce_array_memmap_backward
+        backward_reducers[np.memmap] = reduce_array_memmap_backward
 
     return forward_reducers, backward_reducers, pool_folder
+
+
+class TemporaryResourcesManagerMixin(object):
+    def _unlink_temporary_resources(self):
+        """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"
+                )
+            # 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):
+        """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"
+                )
+
+    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
diff --git a/joblib/_parallel_backends.py b/joblib/_parallel_backends.py
index c82fe1b..77d3617 100644
--- a/joblib/_parallel_backends.py
+++ b/joblib/_parallel_backends.py
@@ -12,16 +12,19 @@ from abc import ABCMeta, abstractmethod
 
 from .my_exceptions import WorkerInterrupt
 from ._multiprocessing_helpers import mp
+
 if mp is not None:
-    from .disk import delete_folder
     from .pool import MemmappingPool
+    from ._memmapping_reducer import TemporaryResourcesManagerMixin
     from multiprocessing.pool import ThreadPool
     from .executor import get_memmapping_executor
 
     # Compat between concurrent.futures and multiprocessing TimeoutError
     from multiprocessing import TimeoutError
-    from .externals.loky._base import TimeoutError as LokyTimeoutError
+    from concurrent.futures._base import TimeoutError as CfTimeoutError
     from .externals.loky import process_executor, cpu_count
+else:
+    TemporaryResourcesManagerMixin = object
 
 
 class ParallelBackendBase(metaclass=ABCMeta):
@@ -415,9 +418,6 @@ class MultiprocessingBackend(PoolManagerMixin, AutoBatchingMixin,
     However, does not suffer from the Python Global Interpreter Lock.
     """
 
-    # Environment variables to protect against bad situations when nesting
-    JOBLIB_SPAWNED_PROCESS = "__JOBLIB_SPAWNED_PARALLEL__"
-
     supports_timeout = True
 
     def effective_n_jobs(self, n_jobs):
@@ -466,19 +466,6 @@ class MultiprocessingBackend(PoolManagerMixin, AutoBatchingMixin,
             raise FallbackToBackend(
                 SequentialBackend(nesting_level=self.nesting_level))
 
-        already_forked = int(os.environ.get(self.JOBLIB_SPAWNED_PROCESS, 0))
-        if already_forked:
-            raise ImportError(
-                '[joblib] Attempting to do parallel computing '
-                'without protecting your import on a system that does '
-                'not support forking. To use parallel-computing in a '
-                'script, you must protect your main loop using "if '
-                "__name__ == '__main__'"
-                '". Please see the joblib documentation on Parallel '
-                'for more information')
-        # Set an environment variable to avoid infinite loops
-        os.environ[self.JOBLIB_SPAWNED_PROCESS] = '1'
-
         # Make sure to free as much memory as possible before forking
         gc.collect()
         self._pool = MemmappingPool(n_jobs, **memmappingpool_args)
@@ -488,13 +475,11 @@ class MultiprocessingBackend(PoolManagerMixin, AutoBatchingMixin,
     def terminate(self):
         """Shutdown the process or thread pool"""
         super(MultiprocessingBackend, self).terminate()
-        if self.JOBLIB_SPAWNED_PROCESS in os.environ:
-            del os.environ[self.JOBLIB_SPAWNED_PROCESS]
-
         self.reset_batch_stats()
 
 
-class LokyBackend(AutoBatchingMixin, ParallelBackendBase):
+class LokyBackend(AutoBatchingMixin, ParallelBackendBase,
+                  TemporaryResourcesManagerMixin):
     """Managing pool of workers with loky instead of multiprocessing."""
 
     supports_timeout = True
@@ -512,6 +497,7 @@ class LokyBackend(AutoBatchingMixin, ParallelBackendBase):
             n_jobs, timeout=idle_worker_timeout,
             env=self._prepare_worker_env(n_jobs=n_jobs),
             **memmappingexecutor_args)
+        self._temp_folder = self._workers._temp_folder
         self.parallel = parallel
         return n_jobs
 
@@ -557,15 +543,15 @@ class LokyBackend(AutoBatchingMixin, ParallelBackendBase):
         AsyncResults.get from multiprocessing."""
         try:
             return future.result(timeout=timeout)
-        except LokyTimeoutError:
+        except CfTimeoutError:
             raise TimeoutError()
 
     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 but we free as much memory as we can by deleting
-            # the shared memory
-            delete_folder(self._workers._temp_folder)
+            # in latter calls
             self._workers = None
 
         self.reset_batch_stats()
@@ -574,8 +560,17 @@ class LokyBackend(AutoBatchingMixin, ParallelBackendBase):
         """Shutdown the workers and restart a new one with the same parameters
         """
         self._workers.shutdown(kill_workers=True)
-        delete_folder(self._workers._temp_folder)
+        # 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 = None
+
         if ensure_ready:
             self.configure(n_jobs=self.parallel.n_jobs, parallel=self.parallel)
 
diff --git a/joblib/backports.py b/joblib/backports.py
index 77858d8..cb2f723 100644
--- a/joblib/backports.py
+++ b/joblib/backports.py
@@ -5,26 +5,43 @@ import os
 import time
 
 from distutils.version import LooseVersion
+from os.path import basename
+from multiprocessing import util
+
 
 try:
     import numpy as np
 
     def make_memmap(filename, dtype='uint8', mode='r+', offset=0,
-                    shape=None, order='C'):
-        """Backport of numpy memmap offset fix.
-
-        See https://github.com/numpy/numpy/pull/8443 for more details.
+                    shape=None, order='C', unlink_on_gc_collect=False):
+        """Custom memmap constructor compatible with numpy.memmap.
 
-        The numpy fix will be available in numpy 1.13.
+        This function:
+        - is a backport the numpy memmap offset fix (See
+          https://github.com/numpy/numpy/pull/8443 for more details.
+          The numpy fix is available starting numpy 1.13)
+        - adds ``unlink_on_gc_collect``, which specifies  explicitly whether
+          the process re-constructing the memmap owns a reference to the
+          underlying file. If set to True, it adds a finalizer to the
+          newly-created memmap that sends a maybe_unlink request for the
+          memmaped file to resource_tracker.
         """
+        util.debug(
+            "[MEMMAP READ] creating a memmap (shape {}, filename {}, "
+            "pid {})".format(shape, basename(filename), os.getpid())
+        )
+
         mm = np.memmap(filename, dtype=dtype, mode=mode, offset=offset,
                        shape=shape, order=order)
         if LooseVersion(np.__version__) < '1.13':
             mm.offset = offset
+        if unlink_on_gc_collect:
+            from ._memmapping_reducer import add_maybe_unlink_finalizer
+            add_maybe_unlink_finalizer(mm)
         return mm
 except ImportError:
     def make_memmap(filename, dtype='uint8', mode='r+', offset=0,
-                    shape=None, order='C'):
+                    shape=None, order='C', unlink_on_gc_collect=False):
         raise NotImplementedError(
             "'joblib.backports.make_memmap' should not be used "
             'if numpy is not installed.')
diff --git a/joblib/disk.py b/joblib/disk.py
index c90c3df..3ce02fa 100644
--- a/joblib/disk.py
+++ b/joblib/disk.py
@@ -13,7 +13,8 @@ import sys
 import time
 import errno
 import shutil
-import warnings
+
+from multiprocessing import util
 
 
 try:
@@ -101,7 +102,7 @@ def rm_subdirs(path, onerror=None):
         delete_folder(fullname, onerror=onerror)
 
 
-def delete_folder(folder_path, onerror=None):
+def delete_folder(folder_path, onerror=None, allow_non_empty=True):
     """Utility function to cleanup a temporary folder if it still exists."""
     if os.path.isdir(folder_path):
         if onerror is not None:
@@ -111,14 +112,25 @@ def delete_folder(folder_path, onerror=None):
             # if the error is raised again, fail
             err_count = 0
             while True:
+                files = os.listdir(folder_path)
                 try:
-                    shutil.rmtree(folder_path, False, None)
-                    break
+                    if len(files) == 0 or allow_non_empty:
+                        shutil.rmtree(
+                            folder_path, ignore_errors=False, onerror=None
+                        )
+                        util.debug(
+                            "Sucessfully deleted {}".format(folder_path))
+                        break
+                    else:
+                        raise OSError(
+                            "Expected empty folder {} but got {} "
+                            "files.".format(folder_path, len(files))
+                        )
                 except (OSError, WindowsError):
                     err_count += 1
                     if err_count > RM_SUBDIRS_N_RETRY:
-                        warnings.warn(
-                            "Unable to delete folder {} after {} tentatives."
-                            .format(folder_path, RM_SUBDIRS_N_RETRY))
+                        # the folder cannot be deleted right now. It maybe
+                        # because some temporary files have not been deleted
+                        # yet.
                         raise
-                    time.sleep(RM_SUBDIRS_RETRY_TIME)
+                time.sleep(RM_SUBDIRS_RETRY_TIME)
diff --git a/joblib/executor.py b/joblib/executor.py
index a6c0e81..e7e678d 100644
--- a/joblib/executor.py
+++ b/joblib/executor.py
@@ -9,8 +9,8 @@ copy between the parent and child processes.
 # License: BSD 3 clause
 
 import random
-from .disk import delete_folder
 from ._memmapping_reducer import get_memmapping_reducers
+from ._memmapping_reducer import TemporaryResourcesManagerMixin
 from .externals.loky.reusable_executor import get_reusable_executor
 
 
@@ -34,23 +34,21 @@ def get_memmapping_executor(n_jobs, timeout=300, initializer=None, initargs=(),
 
     id_executor = random.randint(0, int(1e10))
     job_reducers, result_reducers, temp_folder = get_memmapping_reducers(
-        id_executor, **backend_args)
+        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 been used. Else, the previous reducers are used
-    # and we should not change this attibute.
+    # 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
-    else:
-        delete_folder(temp_folder)
     return _executor
 
 
-class _TestingMemmappingExecutor():
+class _TestingMemmappingExecutor(TemporaryResourcesManagerMixin):
     """Wrapper around ReusableExecutor to ease memmapping testing with Pool
     and Executor. This is only for testing purposes.
     """
@@ -66,7 +64,7 @@ class _TestingMemmappingExecutor():
 
     def terminate(self):
         self._executor.shutdown()
-        delete_folder(self._temp_folder)
+        self._unlink_temporary_resources()
 
     def map(self, f, *args):
         res = self._executor.map(f, *args)
diff --git a/joblib/externals/cloudpickle/__init__.py b/joblib/externals/cloudpickle/__init__.py
index 4f41d8e..695be77 100644
--- a/joblib/externals/cloudpickle/__init__.py
+++ b/joblib/externals/cloudpickle/__init__.py
@@ -8,4 +8,4 @@ from .cloudpickle import *
 if sys.version_info[:2] >= (3, 8):
     from .cloudpickle_fast import CloudPickler, dumps, dump
 
-__version__ = '1.2.2'
+__version__ = '1.4.1'
diff --git a/joblib/externals/cloudpickle/cloudpickle.py b/joblib/externals/cloudpickle/cloudpickle.py
index 1d4c85c..c639daa 100644
--- a/joblib/externals/cloudpickle/cloudpickle.py
+++ b/joblib/externals/cloudpickle/cloudpickle.py
@@ -42,8 +42,9 @@ SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 """
 from __future__ import print_function
 
+import abc
+import builtins
 import dis
-from functools import partial
 import io
 import itertools
 import logging
@@ -53,17 +54,30 @@ import pickle
 import platform
 import struct
 import sys
-import traceback
 import types
 import weakref
 import uuid
 import threading
+import typing
+from enum import Enum
+
+from typing import Generic, Union, Tuple, Callable
+from pickle import _Pickler as Pickler
+from pickle import _getattribute
+from io import BytesIO
+from importlib._bootstrap import _find_spec
+
+try:  # pragma: no branch
+    import typing_extensions as _typing_extensions
+    from typing_extensions import Literal, Final
+except ImportError:
+    _typing_extensions = Literal = Final = None
 
+if sys.version_info >= (3, 5, 3):
+    from typing import ClassVar
+else:  # pragma: no cover
+    ClassVar = None
 
-try:
-    from enum import Enum
-except ImportError:
-    Enum = None
 
 # cloudpickle is meant for inter process communication: we expect all
 # communicating processes to run the same Python version hence we favor
@@ -84,28 +98,10 @@ if PYPY:
     # builtin-code objects only exist in pypy
     builtin_code_type = type(float.__new__.__code__)
 
-if sys.version_info[0] < 3:  # pragma: no branch
-    from pickle import Pickler
-    try:
-        from cStringIO import StringIO
-    except ImportError:
-        from StringIO import StringIO
-    string_types = (basestring,)  # noqa
-    PY3 = False
-    PY2 = True
-else:
-    types.ClassType = type
-    from pickle import _Pickler as Pickler
-    from io import BytesIO as StringIO
-    string_types = (str,)
-    PY3 = True
-    PY2 = False
-    from importlib._bootstrap import _find_spec
-
 _extract_code_globals_cache = weakref.WeakKeyDictionary()
 
 
-def _ensure_tracking(class_def):
+def _get_or_create_tracker_id(class_def):
     with _DYNAMIC_CLASS_TRACKER_LOCK:
         class_tracker_id = _DYNAMIC_CLASS_TRACKER_BY_CLASS.get(class_def)
         if class_tracker_id is None:
@@ -123,21 +119,6 @@ def _lookup_class_or_track(class_tracker_id, class_def):
             _DYNAMIC_CLASS_TRACKER_BY_CLASS[class_def] = class_tracker_id
     return class_def
 
-if sys.version_info[:2] >= (3, 5):
-    from pickle import _getattribute
-elif sys.version_info[:2] >= (3, 4):
-    from pickle import _getattribute as _py34_getattribute
-    #  pickle._getattribute does not return the parent under Python 3.4
-    def _getattribute(obj, name):
-        return _py34_getattribute(obj, name), None
-else:
-    # pickle._getattribute is a python3 addition and enchancement of getattr,
-    # that can handle dotted attribute names. In cloudpickle for python2,
-    # handling dotted names is not needed, so we simply define _getattribute as
-    # a wrapper around getattr.
-    def _getattribute(obj, name):
-        return getattr(obj, name, None), None
-
 
 def _whichmodule(obj, name):
     """Find the module an object belongs to.
@@ -148,13 +129,31 @@ def _whichmodule(obj, name):
     - Errors arising during module introspection are ignored, as those errors
       are considered unwanted side effects.
     """
-    module_name = getattr(obj, '__module__', None)
+    if sys.version_info[:2] < (3, 7) and isinstance(obj, typing.TypeVar):  # pragma: no branch  # noqa
+        # Workaround bug in old Python versions: prior to Python 3.7,
+        # T.__module__ would always be set to "typing" even when the TypeVar T
+        # would be defined in a different module.
+        #
+        # For such older Python versions, we ignore the __module__ attribute of
+        # TypeVar instances and instead exhaustively lookup those instances in
+        # all currently imported modules.
+        module_name = None
+    else:
+        module_name = getattr(obj, '__module__', None)
+
     if module_name is not None:
         return module_name
-    # Protect the iteration by using a list copy of sys.modules against dynamic
-    # modules that trigger imports of other modules upon calls to getattr.
-    for module_name, module in list(sys.modules.items()):
-        if module_name == '__main__' or module is None:
+    # Protect the iteration by using a copy of sys.modules against dynamic
+    # modules that trigger imports of other modules upon calls to getattr or
+    # other threads importing at the same time.
+    for module_name, module in sys.modules.copy().items():
+        # Some modules such as coverage can inject non-module objects inside
+        # sys.modules
+        if (
+                module_name == '__main__' or
+                module is None or
+                not isinstance(module, types.ModuleType)
+        ):
             continue
         try:
             if _getattribute(module, name)[0] is obj:
@@ -164,11 +163,18 @@ def _whichmodule(obj, name):
     return None
 
 
-def _is_global(obj, name=None):
+def _is_importable_by_name(obj, name=None):
     """Determine if obj can be pickled as attribute of a file-backed module"""
+    return _lookup_module_and_qualname(obj, name=name) is not None
+
+
+def _lookup_module_and_qualname(obj, name=None):
     if name is None:
         name = getattr(obj, '__qualname__', None)
-    if name is None:
+    if name is None:  # pragma: no cover
+        # This used to be needed for Python 2.7 support but is probably not
+        # needed anymore. However we keep the __name__ introspection in case
+        # users of cloudpickle rely on this old behavior for unknown reasons.
         name = getattr(obj, '__name__', None)
 
     module_name = _whichmodule(obj, name)
@@ -176,10 +182,10 @@ def _is_global(obj, name=None):
     if module_name is None:
         # In this case, obj.__module__ is None AND obj was not found in any
         # imported module. obj is thus treated as dynamic.
-        return False
+        return None
 
     if module_name == "__main__":
-        return False
+        return None
 
     module = sys.modules.get(module_name, None)
     if module is None:
@@ -188,18 +194,20 @@ def _is_global(obj, name=None):
         # types.ModuleType. The other possibility is that module was removed
         # from sys.modules after obj was created/imported. But this case is not
         # supported, as the standard pickle does not support it either.
-        return False
+        return None
 
     # module has been added to sys.modules, but it can still be dynamic.
     if _is_dynamic(module):
-        return False
+        return None
 
     try:
         obj2, parent = _getattribute(module, name)
     except AttributeError:
         # obj was not found inside the module it points to
-        return False
-    return obj2 is obj
+        return None
+    if obj2 is not obj:
+        return None
+    return module, name
 
 
 def _extract_code_globals(co):
@@ -346,41 +354,23 @@ def _make_cell_set_template_code():
 
     co = _cell_set_factory.__code__
 
-    if PY2:  # pragma: no branch
-        _cell_set_template_code = types.CodeType(
-            co.co_argcount,
-            co.co_nlocals,
-            co.co_stacksize,
-            co.co_flags,
-            co.co_code,
-            co.co_consts,
-            co.co_names,
-            co.co_varnames,
-            co.co_filename,
-            co.co_name,
-            co.co_firstlineno,
-            co.co_lnotab,
-            co.co_cellvars,  # co_freevars is initialized with co_cellvars
-            (),  # co_cellvars is made empty
-        )
-    else:
-        _cell_set_template_code = types.CodeType(
-            co.co_argcount,
-            co.co_kwonlyargcount,   # Python 3 only argument
-            co.co_nlocals,
-            co.co_stacksize,
-            co.co_flags,
-            co.co_code,
-            co.co_consts,
-            co.co_names,
-            co.co_varnames,
-            co.co_filename,
-            co.co_name,
-            co.co_firstlineno,
-            co.co_lnotab,
-            co.co_cellvars,  # co_freevars is initialized with co_cellvars
-            (),  # co_cellvars is made empty
-        )
+    _cell_set_template_code = types.CodeType(
+        co.co_argcount,
+        co.co_kwonlyargcount,   # Python 3 only argument
+        co.co_nlocals,
+        co.co_stacksize,
+        co.co_flags,
+        co.co_code,
+        co.co_consts,
+        co.co_names,
+        co.co_varnames,
+        co.co_filename,
+        co.co_name,
+        co.co_firstlineno,
+        co.co_lnotab,
+        co.co_cellvars,  # co_freevars is initialized with co_cellvars
+        (),  # co_cellvars is made empty
+    )
     return _cell_set_template_code
 
 
@@ -403,44 +393,23 @@ for k, v in types.__dict__.items():
 
 
 def _builtin_type(name):
+    if name == "ClassType":  # pragma: no cover
+        # Backward compat to load pickle files generated with cloudpickle
+        # < 1.3 even if loading pickle files from older versions is not
+        # officially supported.
+        return type
     return getattr(types, name)
 
 
-if sys.version_info < (3, 4):  # pragma: no branch
-    def _walk_global_ops(code):
-        """
-        Yield (opcode, argument number) tuples for all
-        global-referencing instructions in *code*.
-        """
-        code = getattr(code, 'co_code', b'')
-        if PY2:  # pragma: no branch
-            code = map(ord, code)
-
-        n = len(code)
-        i = 0
-        extended_arg = 0
-        while i < n:
-            op = code[i]
-            i += 1
-            if op >= HAVE_ARGUMENT:
-                oparg = code[i] + code[i + 1] * 256 + extended_arg
-                extended_arg = 0
-                i += 2
-                if op == EXTENDED_ARG:
-                    extended_arg = oparg * 65536
-                if op in GLOBAL_OPS:
-                    yield op, oparg
-
-else:
-    def _walk_global_ops(code):
-        """
-        Yield (opcode, argument number) tuples for all
-        global-referencing instructions in *code*.
-        """
-        for instr in dis.get_instructions(code):
-            op = instr.opcode
-            if op in GLOBAL_OPS:
-                yield op, instr.arg
+def _walk_global_ops(code):
+    """
+    Yield (opcode, argument number) tuples for all
+    global-referencing instructions in *code*.
+    """
+    for instr in dis.get_instructions(code):
+        op = instr.opcode
+        if op in GLOBAL_OPS:
+            yield op, instr.arg
 
 
 def _extract_class_dict(cls):
@@ -465,6 +434,32 @@ def _extract_class_dict(cls):
     return clsdict
 
 
+if sys.version_info[:2] < (3, 7):  # pragma: no branch
+    def _is_parametrized_type_hint(obj):
+        # This is very cheap but might generate false positives.
+        # general typing Constructs
+        is_typing = getattr(obj, '__origin__', None) is not None
+
+        # typing_extensions.Literal
+        is_litteral = getattr(obj, '__values__', None) is not None
+
+        # typing_extensions.Final
+        is_final = getattr(obj, '__type__', None) is not None
+
+        # typing.Union/Tuple for old Python 3.5
+        is_union = getattr(obj, '__union_params__', None) is not None
+        is_tuple = getattr(obj, '__tuple_params__', None) is not None
+        is_callable = (
+            getattr(obj, '__result__', None) is not None and
+            getattr(obj, '__args__', None) is not None
+        )
+        return any((is_typing, is_litteral, is_final, is_union, is_tuple,
+                    is_callable))
+
+    def _create_parametrized_type_hint(origin, args):
+        return origin[args]
+
+
 class CloudPickler(Pickler):
 
     dispatch = Pickler.dispatch.copy()
@@ -487,22 +482,22 @@ class CloudPickler(Pickler):
             else:
                 raise
 
+    def save_typevar(self, obj):
+        self.save_reduce(*_typevar_reduce(obj), obj=obj)
+
+    dispatch[typing.TypeVar] = save_typevar
+
     def save_memoryview(self, obj):
         self.save(obj.tobytes())
 
     dispatch[memoryview] = save_memoryview
 
-    if PY2:  # pragma: no branch
-        def save_buffer(self, obj):
-            self.save(str(obj))
-
-        dispatch[buffer] = save_buffer  # noqa: F821 'buffer' was removed in Python 3
-
     def save_module(self, obj):
         """
         Save a module as an import
         """
         if _is_dynamic(obj):
+            obj.__dict__.pop('__builtins__', None)
             self.save_reduce(dynamic_subimport, (obj.__name__, vars(obj)),
                              obj=obj)
         else:
@@ -514,29 +509,22 @@ class CloudPickler(Pickler):
         """
         Save a code object
         """
-        if PY3:  # pragma: no branch
-            if hasattr(obj, "co_posonlyargcount"):  # pragma: no branch
-                args = (
-                    obj.co_argcount, obj.co_posonlyargcount,
-                    obj.co_kwonlyargcount, obj.co_nlocals, obj.co_stacksize,
-                    obj.co_flags, obj.co_code, obj.co_consts, obj.co_names,
-                    obj.co_varnames, obj.co_filename, obj.co_name,
-                    obj.co_firstlineno, obj.co_lnotab, obj.co_freevars,
-                    obj.co_cellvars
-                )
-            else:
-                args = (
-                    obj.co_argcount, obj.co_kwonlyargcount, obj.co_nlocals,
-                    obj.co_stacksize, obj.co_flags, obj.co_code, obj.co_consts,
-                    obj.co_names, obj.co_varnames, obj.co_filename,
-                    obj.co_name, obj.co_firstlineno, obj.co_lnotab,
-                    obj.co_freevars, obj.co_cellvars
-                )
+        if hasattr(obj, "co_posonlyargcount"):  # pragma: no branch
+            args = (
+                obj.co_argcount, obj.co_posonlyargcount,
+                obj.co_kwonlyargcount, obj.co_nlocals, obj.co_stacksize,
+                obj.co_flags, obj.co_code, obj.co_consts, obj.co_names,
+                obj.co_varnames, obj.co_filename, obj.co_name,
+                obj.co_firstlineno, obj.co_lnotab, obj.co_freevars,
+                obj.co_cellvars
+            )
         else:
             args = (
-                obj.co_argcount, obj.co_nlocals, obj.co_stacksize, obj.co_flags, obj.co_code,
-                obj.co_consts, obj.co_names, obj.co_varnames, obj.co_filename, obj.co_name,
-                obj.co_firstlineno, obj.co_lnotab, obj.co_freevars, obj.co_cellvars
+                obj.co_argcount, obj.co_kwonlyargcount, obj.co_nlocals,
+                obj.co_stacksize, obj.co_flags, obj.co_code, obj.co_consts,
+                obj.co_names, obj.co_varnames, obj.co_filename,
+                obj.co_name, obj.co_firstlineno, obj.co_lnotab,
+                obj.co_freevars, obj.co_cellvars
             )
         self.save_reduce(types.CodeType, args, obj=obj)
 
@@ -548,7 +536,7 @@ class CloudPickler(Pickler):
         Determines what kind of function obj is (e.g. lambda, defined at
         interactive prompt, etc) and handles the pickling appropriately.
         """
-        if _is_global(obj, name=name):
+        if _is_importable_by_name(obj, name=name):
             return Pickler.save_global(self, obj, name=name)
         elif PYPY and isinstance(obj.__code__, builtin_code_type):
             return self.save_pypy_builtin_func(obj)
@@ -590,13 +578,12 @@ class CloudPickler(Pickler):
         """
         members = dict((e.name, e.value) for e in obj)
 
-        # Python 2.7 with enum34 can have no qualname:
-        qualname = getattr(obj, "__qualname__", None)
-
-        self.save_reduce(_make_skeleton_enum,
-                         (obj.__bases__, obj.__name__, qualname, members,
-                          obj.__module__, _ensure_tracking(obj), None),
-                         obj=obj)
+        self.save_reduce(
+                _make_skeleton_enum,
+                (obj.__bases__, obj.__name__, obj.__qualname__,
+                 members, obj.__module__, _get_or_create_tracker_id(obj), None),
+                obj=obj
+         )
 
         # Cleanup the clsdict that will be passed to _rehydrate_skeleton_class:
         # Those attributes are already handled by the metaclass.
@@ -617,26 +604,38 @@ class CloudPickler(Pickler):
         clsdict = _extract_class_dict(obj)
         clsdict.pop('__weakref__', None)
 
-        # For ABCMeta in python3.7+, remove _abc_impl as it is not picklable.
-        # This is a fix which breaks the cache but this only makes the first
-        # calls to issubclass slower.
-        if "_abc_impl" in clsdict:
-            import abc
-            (registry, _, _, _) = abc._get_dump(obj)
-            clsdict["_abc_impl"] = [subclass_weakref()
-                                    for subclass_weakref in registry]
+        if issubclass(type(obj), abc.ABCMeta):
+            # If obj is an instance of an ABCMeta subclass, dont pickle the
+            # cache/negative caches populated during isinstance/issubclass
+            # checks, but pickle the list of registered subclasses of obj.
+            clsdict.pop('_abc_cache', None)
+            clsdict.pop('_abc_negative_cache', None)
+            clsdict.pop('_abc_negative_cache_version', None)
+            registry = clsdict.pop('_abc_registry', None)
+            if registry is None:
+                # in Python3.7+, the abc caches and registered subclasses of a
+                # class are bundled into the single _abc_impl attribute
+                clsdict.pop('_abc_impl', None)
+                (registry, _, _, _) = abc._get_dump(obj)
+
+                clsdict["_abc_impl"] = [subclass_weakref()
+                                        for subclass_weakref in registry]
+            else:
+                # In the above if clause, registry is a set of weakrefs -- in
+                # this case, registry is a WeakSet
+                clsdict["_abc_impl"] = [type_ for type_ in registry]
 
         # On PyPy, __doc__ is a readonly attribute, so we need to include it in
         # the initial skeleton class.  This is safe because we know that the
         # doc can't participate in a cycle with the original class.
         type_kwargs = {'__doc__': clsdict.pop('__doc__', None)}
 
-        if hasattr(obj, "__slots__"):
+        if "__slots__" in clsdict:
             type_kwargs['__slots__'] = obj.__slots__
             # pickle string length optimization: member descriptors of obj are
             # created automatically from obj's __slots__ attribute, no need to
             # save them in obj's state
-            if isinstance(obj.__slots__, string_types):
+            if isinstance(obj.__slots__, str):
                 clsdict.pop(obj.__slots__)
             else:
                 for k in obj.__slots__:
@@ -644,6 +643,7 @@ class CloudPickler(Pickler):
 
         # If type overrides __dict__ as a property, include it in the type
         # kwargs. In Python 2, we can't set this attribute after construction.
+        # XXX: can this ever happen in Python 3? If so add a test.
         __dict__ = clsdict.pop('__dict__', None)
         if isinstance(__dict__, property):
             type_kwargs['__dict__'] = __dict__
@@ -677,8 +677,8 @@ class CloudPickler(Pickler):
             # "Regular" class definition:
             tp = type(obj)
             self.save_reduce(_make_skeleton_class,
-                             (tp, obj.__name__, obj.__bases__, type_kwargs,
-                              _ensure_tracking(obj), None),
+                             (tp, obj.__name__, _get_bases(obj), type_kwargs,
+                              _get_or_create_tracker_id(obj), None),
                              obj=obj)
 
         # Now save the rest of obj's __dict__. Any references to obj
@@ -747,9 +747,7 @@ class CloudPickler(Pickler):
             'doc': func.__doc__,
             '_cloudpickle_submodules': submodules
         }
-        if hasattr(func, '__annotations__') and sys.version_info >= (3, 7):
-            # Although annotations were added in Python3.4, It is not possible
-            # to properly pickle them until Python3.7. (See #193)
+        if hasattr(func, '__annotations__'):
             state['annotations'] = func.__annotations__
         if hasattr(func, '__qualname__'):
             state['qualname'] = func.__qualname__
@@ -811,45 +809,6 @@ class CloudPickler(Pickler):
 
         return (code, f_globals, defaults, closure, dct, base_globals)
 
-    if not PY3:  # pragma: no branch
-        # Python3 comes with native reducers that allow builtin functions and
-        # methods pickling as module/class attributes.  The following method
-        # extends this for python2.
-        # Please note that currently, neither pickle nor cloudpickle support
-        # dynamically created builtin functions/method pickling.
-        def save_builtin_function_or_method(self, obj):
-            is_bound = getattr(obj, '__self__', None) is not None
-            if is_bound:
-                # obj is a bound builtin method.
-                rv = (getattr, (obj.__self__, obj.__name__))
-                return self.save_reduce(obj=obj, *rv)
-
-            is_unbound = hasattr(obj, '__objclass__')
-            if is_unbound:
-                # obj is an unbound builtin method (accessed from its class)
-                rv = (getattr, (obj.__objclass__, obj.__name__))
-                return self.save_reduce(obj=obj, *rv)
-
-            # Otherwise, obj is not a method, but a function. Fallback to
-            # default pickling by attribute.
-            return Pickler.save_global(self, obj)
-
-        dispatch[types.BuiltinFunctionType] = save_builtin_function_or_method
-
-        # A comprehensive summary of the various kinds of builtin methods can
-        # be found in PEP 579: https://www.python.org/dev/peps/pep-0579/
-        classmethod_descriptor_type = type(float.__dict__['fromhex'])
-        wrapper_descriptor_type = type(float.__repr__)
-        method_wrapper_type = type(1.5.__repr__)
-
-        dispatch[classmethod_descriptor_type] = save_builtin_function_or_method
-        dispatch[wrapper_descriptor_type] = save_builtin_function_or_method
-        dispatch[method_wrapper_type] = save_builtin_function_or_method
-
-    if sys.version_info[:2] < (3, 4):
-        method_descriptor = type(str.upper)
-        dispatch[method_descriptor] = save_builtin_function_or_method
-
     def save_getset_descriptor(self, obj):
         return self.save_reduce(getattr, (obj.__objclass__, obj.__name__))
 
@@ -871,81 +830,36 @@ class CloudPickler(Pickler):
         elif obj in _BUILTIN_TYPE_NAMES:
             return self.save_reduce(
                 _builtin_type, (_BUILTIN_TYPE_NAMES[obj],), obj=obj)
+
+        if sys.version_info[:2] < (3, 7) and _is_parametrized_type_hint(obj):  # noqa  # pragma: no branch
+            # Parametrized typing constructs in Python < 3.7 are not compatible
+            # with type checks and ``isinstance`` semantics. For this reason,
+            # it is easier to detect them using a duck-typing-based check
+            # (``_is_parametrized_type_hint``) than to populate the Pickler's
+            # dispatch with type-specific savers.
+            self._save_parametrized_type_hint(obj)
         elif name is not None:
             Pickler.save_global(self, obj, name=name)
-        elif not _is_global(obj, name=name):
+        elif not _is_importable_by_name(obj, name=name):
             self.save_dynamic_class(obj)
         else:
             Pickler.save_global(self, obj, name=name)
 
     dispatch[type] = save_global
-    dispatch[types.ClassType] = save_global
 
     def save_instancemethod(self, obj):
         # Memoization rarely is ever useful due to python bounding
         if obj.__self__ is None:
             self.save_reduce(getattr, (obj.im_class, obj.__name__))
         else:
-            if PY3:  # pragma: no branch
-                self.save_reduce(types.MethodType, (obj.__func__, obj.__self__), obj=obj)
-            else:
-                self.save_reduce(
-                    types.MethodType,
-                    (obj.__func__, obj.__self__, type(obj.__self__)), obj=obj)
+            self.save_reduce(types.MethodType, (obj.__func__, obj.__self__), obj=obj)
 
     dispatch[types.MethodType] = save_instancemethod
 
-    def save_inst(self, obj):
-        """Inner logic to save instance. Based off pickle.save_inst"""
-        cls = obj.__class__
-
-        # Try the dispatch table (pickle module doesn't do it)
-        f = self.dispatch.get(cls)
-        if f:
-            f(self, obj)  # Call unbound method with explicit self
-            return
-
-        memo = self.memo
-        write = self.write
-        save = self.save
-
-        if hasattr(obj, '__getinitargs__'):
-            args = obj.__getinitargs__()
-            len(args)  # XXX Assert it's a sequence
-            pickle._keep_alive(args, memo)
-        else:
-            args = ()
-
-        write(pickle.MARK)
-
-        if self.bin:
-            save(cls)
-            for arg in args:
-                save(arg)
-            write(pickle.OBJ)
-        else:
-            for arg in args:
-                save(arg)
-            write(pickle.INST + cls.__module__ + '\n' + cls.__name__ + '\n')
-
-        self.memoize(obj)
-
-        try:
-            getstate = obj.__getstate__
-        except AttributeError:
-            stuff = obj.__dict__
-        else:
-            stuff = getstate()
-            pickle._keep_alive(stuff, memo)
-        save(stuff)
-        write(pickle.BUILD)
-
-    if PY2:  # pragma: no branch
-        dispatch[types.InstanceType] = save_inst
-
     def save_property(self, obj):
         # properties not correctly saved in python
-        self.save_reduce(property, (obj.fget, obj.fset, obj.fdel, obj.__doc__), obj=obj)
+        self.save_reduce(property, (obj.fget, obj.fset, obj.fdel, obj.__doc__),
+                         obj=obj)
 
     dispatch[property] = save_property
 
@@ -993,10 +907,6 @@ class CloudPickler(Pickler):
 
     def save_file(self, obj):
         """Save a file"""
-        try:
-            import StringIO as pystringIO  # we can't use cStringIO as it lacks the name attribute
-        except ImportError:
-            import io as pystringIO
 
         if not hasattr(obj, 'name') or not hasattr(obj, 'mode'):
             raise pickle.PicklingError("Cannot pickle files that do not map to an actual file")
@@ -1015,7 +925,8 @@ class CloudPickler(Pickler):
 
         name = obj.name
 
-        retval = pystringIO.StringIO()
+        # TODO: also support binary mode files with io.BytesIO
+        retval = io.StringIO()
 
         try:
             # Read the whole file
@@ -1038,11 +949,7 @@ class CloudPickler(Pickler):
     def save_not_implemented(self, obj):
         self.save_reduce(_gen_not_implemented, ())
 
-    try:               # Python 2
-        dispatch[file] = save_file
-    except NameError:  # Python 3  # pragma: no branch
-        dispatch[io.TextIOWrapper] = save_file
-
+    dispatch[io.TextIOWrapper] = save_file
     dispatch[type(Ellipsis)] = save_ellipsis
     dispatch[type(NotImplemented)] = save_not_implemented
 
@@ -1072,6 +979,57 @@ class CloudPickler(Pickler):
         """Plug in system. Register additional pickling functions if modules already loaded"""
         pass
 
+    if sys.version_info < (3, 7):  # pragma: no branch
+        def _save_parametrized_type_hint(self, obj):
+            # The distorted type check sematic for typing construct becomes:
+            # ``type(obj) is type(TypeHint)``, which means "obj is a
+            # parametrized TypeHint"
+            if type(obj) is type(Literal):  # pragma: no branch
+                initargs = (Literal, obj.__values__)
+            elif type(obj) is type(Final):  # pragma: no branch
+                initargs = (Final, obj.__type__)
+            elif type(obj) is type(ClassVar):
+                initargs = (ClassVar, obj.__type__)
+            elif type(obj) is type(Generic):
+                parameters = obj.__parameters__
+                if len(obj.__parameters__) > 0:
+                    # in early Python 3.5, __parameters__ was sometimes
+                    # preferred to __args__
+                    initargs = (obj.__origin__, parameters)
+                else:
+                    initargs = (obj.__origin__, obj.__args__)
+            elif type(obj) is type(Union):
+                if sys.version_info < (3, 5, 3):  # pragma: no cover
+                    initargs = (Union, obj.__union_params__)
+                else:
+                    initargs = (Union, obj.__args__)
+            elif type(obj) is type(Tuple):
+                if sys.version_info < (3, 5, 3):  # pragma: no cover
+                    initargs = (Tuple, obj.__tuple_params__)
+                else:
+                    initargs = (Tuple, obj.__args__)
+            elif type(obj) is type(Callable):
+                if sys.version_info < (3, 5, 3):  # pragma: no cover
+                    args = obj.__args__
+                    result = obj.__result__
+                    if args != Ellipsis:
+                        if isinstance(args, tuple):
+                            args = list(args)
+                        else:
+                            args = [args]
+                else:
+                    (*args, result) = obj.__args__
+                    if len(args) == 1 and args[0] is Ellipsis:
+                        args = Ellipsis
+                    else:
+                        args = list(args)
+                initargs = (Callable, (args, result))
+            else:  # pragma: no cover
+                raise pickle.PicklingError(
+                    "Cloudpickle Error: Unknown type {}".format(type(obj))
+                )
+            self.save_reduce(_create_parametrized_type_hint, initargs, obj=obj)
+
 
 # Tornado support
 
@@ -1119,7 +1077,7 @@ def dumps(obj, protocol=None):
     Set protocol=pickle.DEFAULT_PROTOCOL instead if you need to ensure
     compatibility with older versions of Python.
     """
-    file = StringIO()
+    file = BytesIO()
     try:
         cp = CloudPickler(file, protocol=protocol)
         cp.dump(obj)
@@ -1142,6 +1100,7 @@ def subimport(name):
 def dynamic_subimport(name, vars):
     mod = types.ModuleType(name)
     mod.__dict__.update(vars)
+    mod.__dict__['__builtins__'] = builtins.__dict__
     return mod
 
 
@@ -1293,7 +1252,10 @@ def _make_skeleton_class(type_constructor, name, bases, type_kwargs,
     The "extra" variable is meant to be a dict (or None) that can be used for
     forward compatibility shall the need arise.
     """
-    skeleton_class = type_constructor(name, bases, type_kwargs)
+    skeleton_class = types.new_class(
+        name, bases, {'metaclass': type_constructor},
+        lambda ns: ns.update(type_kwargs)
+    )
     return _lookup_class_or_track(class_tracker_id, skeleton_class)
 
 
@@ -1340,10 +1302,7 @@ def _make_skeleton_enum(bases, name, qualname, members, module,
         classdict[member_name] = member_value
     enum_class = metacls.__new__(metacls, name, bases, classdict)
     enum_class.__module__ = module
-
-    # Python 2.7 compat
-    if qualname is not None:
-        enum_class.__qualname__ = qualname
+    enum_class.__qualname__ = qualname
 
     return _lookup_class_or_track(class_tracker_id, enum_class)
 
@@ -1357,41 +1316,69 @@ def _is_dynamic(module):
     if hasattr(module, '__file__'):
         return False
 
-    if hasattr(module, '__spec__'):
-        if module.__spec__ is not None:
-            return False
-
-        # In PyPy, Some built-in modules such as _codecs can have their
-        # __spec__ attribute set to None despite being imported.  For such
-        # modules, the ``_find_spec`` utility of the standard library is used.
-        parent_name = module.__name__.rpartition('.')[0]
-        if parent_name:  # pragma: no cover
-            # This code handles the case where an imported package (and not
-            # module) remains with __spec__ set to None. It is however untested
-            # as no package in the PyPy stdlib has __spec__ set to None after
-            # it is imported.
-            try:
-                parent = sys.modules[parent_name]
-            except KeyError:
-                msg = "parent {!r} not in sys.modules"
-                raise ImportError(msg.format(parent_name))
-            else:
-                pkgpath = parent.__path__
+    if module.__spec__ is not None:
+        return False
+
+    # In PyPy, Some built-in modules such as _codecs can have their
+    # __spec__ attribute set to None despite being imported.  For such
+    # modules, the ``_find_spec`` utility of the standard library is used.
+    parent_name = module.__name__.rpartition('.')[0]
+    if parent_name:  # pragma: no cover
+        # This code handles the case where an imported package (and not
+        # module) remains with __spec__ set to None. It is however untested
+        # as no package in the PyPy stdlib has __spec__ set to None after
+        # it is imported.
+        try:
+            parent = sys.modules[parent_name]
+        except KeyError:
+            msg = "parent {!r} not in sys.modules"
+            raise ImportError(msg.format(parent_name))
         else:
-            pkgpath = None
-        return _find_spec(module.__name__, pkgpath, module) is None
+            pkgpath = parent.__path__
+    else:
+        pkgpath = None
+    return _find_spec(module.__name__, pkgpath, module) is None
+
 
+def _make_typevar(name, bound, constraints, covariant, contravariant,
+                  class_tracker_id):
+    tv = typing.TypeVar(
+        name, *constraints, bound=bound,
+        covariant=covariant, contravariant=contravariant
+    )
+    if class_tracker_id is not None:
+        return _lookup_class_or_track(class_tracker_id, tv)
+    else:  # pragma: nocover
+        # Only for Python 3.5.3 compat.
+        return tv
+
+
+def _decompose_typevar(obj):
+    try:
+        class_tracker_id = _get_or_create_tracker_id(obj)
+    except TypeError:  # pragma: nocover
+        # TypeVar instances are not weakref-able in Python 3.5.3
+        class_tracker_id = None
+    return (
+        obj.__name__, obj.__bound__, obj.__constraints__,
+        obj.__covariant__, obj.__contravariant__,
+        class_tracker_id,
+    )
+
+
+def _typevar_reduce(obj):
+    # TypeVar instances have no __qualname__ hence we pass the name explicitly.
+    module_and_name = _lookup_module_and_qualname(obj, name=obj.__name__)
+    if module_and_name is None:
+        return (_make_typevar, _decompose_typevar(obj))
+    return (getattr, module_and_name)
+
+
+def _get_bases(typ):
+    if hasattr(typ, '__orig_bases__'):
+        # For generic types (see PEP 560)
+        bases_attr = '__orig_bases__'
     else:
-        # Backward compat for Python 2
-        import imp
-        try:
-            path = None
-            for part in module.__name__.split('.'):
-                if path is not None:
-                    path = [path]
-                f, path, description = imp.find_module(part, path)
-                if f is not None:
-                    f.close()
-        except ImportError:
-            return True
-        return False
+        # For regular class objects
+        bases_attr = '__bases__'
+    return getattr(typ, bases_attr)
diff --git a/joblib/externals/cloudpickle/cloudpickle_fast.py b/joblib/externals/cloudpickle/cloudpickle_fast.py
index dfe554d..7df95f8 100644
--- a/joblib/externals/cloudpickle/cloudpickle_fast.py
+++ b/joblib/externals/cloudpickle/cloudpickle_fast.py
@@ -20,21 +20,22 @@ import pickle
 import sys
 import types
 import weakref
+import typing
 
 from _pickle import Pickler
 
 from .cloudpickle import (
     _is_dynamic, _extract_code_globals, _BUILTIN_TYPE_NAMES, DEFAULT_PROTOCOL,
-    _find_imported_submodules, _get_cell_contents, _is_global, _builtin_type,
-    Enum, _ensure_tracking,  _make_skeleton_class, _make_skeleton_enum,
-    _extract_class_dict, string_types, dynamic_subimport, subimport
+    _find_imported_submodules, _get_cell_contents, _is_importable_by_name, _builtin_type,
+    Enum, _get_or_create_tracker_id,  _make_skeleton_class, _make_skeleton_enum,
+    _extract_class_dict, dynamic_subimport, subimport, _typevar_reduce, _get_bases,
 )
 
 load, loads = _pickle.load, _pickle.loads
 
 
 # Shorthands similar to pickle.dump/pickle.dumps
-def dump(obj, file, protocol=None):
+def dump(obj, file, protocol=None, buffer_callback=None):
     """Serialize obj as bytes streamed into file
 
     protocol defaults to cloudpickle.DEFAULT_PROTOCOL which is an alias to
@@ -44,10 +45,10 @@ def dump(obj, file, protocol=None):
     Set protocol=pickle.DEFAULT_PROTOCOL instead if you need to ensure
     compatibility with older versions of Python.
     """
-    CloudPickler(file, protocol=protocol).dump(obj)
+    CloudPickler(file, protocol=protocol, buffer_callback=buffer_callback).dump(obj)
 
 
-def dumps(obj, protocol=None):
+def dumps(obj, protocol=None, buffer_callback=None):
     """Serialize obj as a string of bytes allocated in memory
 
     protocol defaults to cloudpickle.DEFAULT_PROTOCOL which is an alias to
@@ -58,7 +59,7 @@ def dumps(obj, protocol=None):
     compatibility with older versions of Python.
     """
     with io.BytesIO() as file:
-        cp = CloudPickler(file, protocol=protocol)
+        cp = CloudPickler(file, protocol=protocol, buffer_callback=buffer_callback)
         cp.dump(obj)
         return file.getvalue()
 
@@ -68,21 +69,21 @@ def dumps(obj, protocol=None):
 
 def _class_getnewargs(obj):
     type_kwargs = {}
-    if hasattr(obj, "__slots__"):
+    if "__slots__" in obj.__dict__:
         type_kwargs["__slots__"] = obj.__slots__
 
     __dict__ = obj.__dict__.get('__dict__', None)
     if isinstance(__dict__, property):
         type_kwargs['__dict__'] = __dict__
 
-    return (type(obj), obj.__name__, obj.__bases__, type_kwargs,
-            _ensure_tracking(obj), None)
+    return (type(obj), obj.__name__, _get_bases(obj), type_kwargs,
+            _get_or_create_tracker_id(obj), None)
 
 
 def _enum_getnewargs(obj):
     members = dict((e.name, e.value) for e in obj)
     return (obj.__bases__, obj.__name__, obj.__qualname__, members,
-            obj.__module__, _ensure_tracking(obj), None)
+            obj.__module__, _get_or_create_tracker_id(obj), None)
 
 
 # COLLECTION OF OBJECTS RECONSTRUCTORS
@@ -136,18 +137,20 @@ def _class_getstate(obj):
     clsdict = _extract_class_dict(obj)
     clsdict.pop('__weakref__', None)
 
-    # For ABCMeta in python3.7+, remove _abc_impl as it is not picklable.
-    # This is a fix which breaks the cache but this only makes the first
-    # calls to issubclass slower.
-    if "_abc_impl" in clsdict:
+    if issubclass(type(obj), abc.ABCMeta):
+        # If obj is an instance of an ABCMeta subclass, dont pickle the
+        # cache/negative caches populated during isinstance/issubclass
+        # checks, but pickle the list of registered subclasses of obj.
+        clsdict.pop('_abc_impl', None)
         (registry, _, _, _) = abc._get_dump(obj)
         clsdict["_abc_impl"] = [subclass_weakref()
                                 for subclass_weakref in registry]
-    if hasattr(obj, "__slots__"):
+
+    if "__slots__" in clsdict:
         # pickle string length optimization: member descriptors of obj are
         # created automatically from obj's __slots__ attribute, no need to
         # save them in obj's state
-        if isinstance(obj.__slots__, string_types):
+        if isinstance(obj.__slots__, str):
             clsdict.pop(obj.__slots__)
         else:
             for k in obj.__slots__:
@@ -274,6 +277,7 @@ def _memoryview_reduce(obj):
 
 def _module_reduce(obj):
     if _is_dynamic(obj):
+        obj.__dict__.pop('__builtins__', None)
         return dynamic_subimport, (obj.__name__, vars(obj))
     else:
         return subimport, (obj.__name__,)
@@ -291,6 +295,10 @@ def _root_logger_reduce(obj):
     return logging.getLogger, ()
 
 
+def _property_reduce(obj):
+    return property, (obj.fget, obj.fset, obj.fdel, obj.__doc__)
+
+
 def _weakset_reduce(obj):
     return weakref.WeakSet, (list(obj),)
 
@@ -325,7 +333,7 @@ def _class_reduce(obj):
         return type, (NotImplemented,)
     elif obj in _BUILTIN_TYPE_NAMES:
         return _builtin_type, (_BUILTIN_TYPE_NAMES[obj],)
-    elif not _is_global(obj):
+    elif not _is_importable_by_name(obj):
         return _dynamic_class_reduce(obj)
     return NotImplemented
 
@@ -406,6 +414,7 @@ class CloudPickler(Pickler):
     dispatch[logging.Logger] = _logger_reduce
     dispatch[logging.RootLogger] = _root_logger_reduce
     dispatch[memoryview] = _memoryview_reduce
+    dispatch[property] = _property_reduce
     dispatch[staticmethod] = _classmethod_reduce
     dispatch[types.CellType] = _cell_reduce
     dispatch[types.CodeType] = _code_reduce
@@ -414,11 +423,12 @@ class CloudPickler(Pickler):
     dispatch[types.MethodType] = _method_reduce
     dispatch[types.MappingProxyType] = _mappingproxy_reduce
     dispatch[weakref.WeakSet] = _weakset_reduce
+    dispatch[typing.TypeVar] = _typevar_reduce
 
-    def __init__(self, file, protocol=None):
+    def __init__(self, file, protocol=None, buffer_callback=None):
         if protocol is None:
             protocol = DEFAULT_PROTOCOL
-        Pickler.__init__(self, file, protocol=protocol)
+        Pickler.__init__(self, file, protocol=protocol, buffer_callback=buffer_callback)
         # map functions __globals__ attribute ids, to ensure that functions
         # sharing the same global namespace at pickling time also share their
         # global namespace at unpickling time.
@@ -495,7 +505,7 @@ class CloudPickler(Pickler):
         As opposed to cloudpickle.py, There no special handling for builtin
         pypy functions because cloudpickle_fast is CPython-specific.
         """
-        if _is_global(obj):
+        if _is_importable_by_name(obj):
             return NotImplemented
         else:
             return self._dynamic_function_reduce(obj)
diff --git a/joblib/externals/loky/__init__.py b/joblib/externals/loky/__init__.py
index 374ee26..3ebc7e5 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.6.0'
+__version__ = '2.7.0'
diff --git a/joblib/externals/loky/backend/popen_loky_posix.py b/joblib/externals/loky/backend/popen_loky_posix.py
index cfe4f72..970dead 100644
--- a/joblib/externals/loky/backend/popen_loky_posix.py
+++ b/joblib/externals/loky/backend/popen_loky_posix.py
@@ -149,6 +149,10 @@ if sys.platform != "win32":
                 reduction._mk_inheritable(child_w)
                 reduction._mk_inheritable(tracker_fd)
                 self._fds.extend([child_r, child_w, tracker_fd])
+                if sys.version_info >= (3, 8) and os.name == 'posix':
+                    mp_tracker_fd = prep_data['mp_tracker_args']['fd']
+                    self.duplicate_for_child(mp_tracker_fd)
+
                 from .fork_exec import fork_exec
                 pid = fork_exec(cmd_python, self._fds, env=process_obj.env)
                 util.debug("launched python with pid {} and cmd:\n{}"
diff --git a/joblib/externals/loky/backend/queues.py b/joblib/externals/loky/backend/queues.py
index 0f9dfea..f2cbf9b 100644
--- a/joblib/externals/loky/backend/queues.py
+++ b/joblib/externals/loky/backend/queues.py
@@ -209,6 +209,10 @@ class SimpleQueue(mp_SimpleQueue):
         # Add possiblity to use custom reducers
         self._reducers = reducers
 
+    def close(self):
+        self._reader.close()
+        self._writer.close()
+
     # Use custom queue set/get state to be able to reduce the custom reducers
     def __getstate__(self):
         assert_spawning(self)
diff --git a/joblib/externals/loky/backend/resource_tracker.py b/joblib/externals/loky/backend/resource_tracker.py
index ce519fe..95dff35 100644
--- a/joblib/externals/loky/backend/resource_tracker.py
+++ b/joblib/externals/loky/backend/resource_tracker.py
@@ -13,15 +13,32 @@
 #
 # On Unix we run a server process which keeps track of unlinked
 # resources. The server ignores SIGINT and SIGTERM and reads from a
-# pipe.  Every other process of the program has a copy of the writable
-# end of the pipe, so we get EOF when all other processes have exited.
-# Then the server process unlinks any remaining resources.
-#
+# pipe. The resource_tracker implements a reference counting scheme: each time
+# a Python process anticipates the shared usage of a resource by another
+# process, it signals the resource_tracker of this shared usage, and in return,
+# the resource_tracker increments the resource's reference count by 1.
+# Similarly, when access to a resource is closed by a Python process, the
+# process notifies the resource_tracker by asking it to decrement the
+# resource's reference count by 1.  When the reference count drops to 0, the
+# resource_tracker attempts to clean up the underlying resource.
+
+# Finally, every other process connected to the resource tracker has a copy of
+# the writable end of the pipe used to communicate with it, so the resource
+# tracker gets EOF when all other processes have exited. Then the
+# resource_tracker process unlinks any remaining leaked resources (with
+# reference count above 0)
+
 # For semaphores, this is important because the system only supports a limited
 # number of named semaphores, and they will not be automatically removed till
 # the next reboot.  Without this resource tracker process, "killall python"
 # would probably leave unlinked semaphores.
 
+# Note that this behavior differs from CPython's resource_tracker, which only
+# implements list of shared resources, and not a proper refcounting scheme.
+# Also, CPython's resource tracker will only attempt to cleanup those shared
+# resources once all procsses connected to the resouce tracker have exited.
+
+
 import os
 import shutil
 import sys
@@ -52,12 +69,14 @@ _HAVE_SIGMASK = hasattr(signal, 'pthread_sigmask')
 _IGNORED_SIGNALS = (signal.SIGINT, signal.SIGTERM)
 
 _CLEANUP_FUNCS = {
-    'folder': shutil.rmtree
+    'folder': shutil.rmtree,
+    'file': os.unlink
 }
 
 if os.name == "posix":
     _CLEANUP_FUNCS['semlock'] = sem_unlink
 
+
 VERBOSE = False
 
 
@@ -167,7 +186,7 @@ class ResourceTracker(object):
             return True
 
     def register(self, name, rtype):
-        '''Register a named resource with resource tracker.'''
+        '''Register a named resource, and increment its refcount.'''
         self.ensure_running()
         self._send('REGISTER', name, rtype)
 
@@ -176,6 +195,11 @@ class ResourceTracker(object):
         self.ensure_running()
         self._send('UNREGISTER', name, rtype)
 
+    def maybe_unlink(self, name, rtype):
+        '''Decrement the refcount of a resource, and delete it if it hits 0'''
+        self.ensure_running()
+        self._send("MAYBE_UNLINK", name, rtype)
+
     def _send(self, cmd, name, rtype):
         msg = '{0}:{1}:{2}\n'.format(cmd, name, rtype).encode('ascii')
         if len(name) > 512:
@@ -189,6 +213,7 @@ class ResourceTracker(object):
 _resource_tracker = ResourceTracker()
 ensure_running = _resource_tracker.ensure_running
 register = _resource_tracker.register
+maybe_unlink = _resource_tracker.maybe_unlink
 unregister = _resource_tracker.unregister
 getfd = _resource_tracker.getfd
 
@@ -196,6 +221,9 @@ getfd = _resource_tracker.getfd
 def main(fd, verbose=0):
     '''Run resource tracker.'''
     # protect the process from ^C and "killall python" etc
+    if verbose:
+        util.log_to_stderr(level=util.DEBUG)
+
     signal.signal(signal.SIGINT, signal.SIG_IGN)
     signal.signal(signal.SIGTERM, signal.SIG_IGN)
 
@@ -208,17 +236,19 @@ def main(fd, verbose=0):
         except Exception:
             pass
 
-    if verbose:  # pragma: no cover
-        sys.stderr.write("Main resource tracker is running\n")
-        sys.stderr.flush()
+    if verbose:
+        util.debug("Main resource tracker is running")
 
-    cache = {rtype: set() for rtype in _CLEANUP_FUNCS.keys()}
+    registry = {rtype: dict() for rtype in _CLEANUP_FUNCS.keys()}
     try:
         # keep track of registered/unregistered resources
         if sys.platform == "win32":
             fd = msvcrt.open_osfhandle(fd, os.O_RDONLY)
         with open(fd, 'rb') as f:
-            for line in f:
+            while True:
+                line = f.readline()
+                if line == b'':  # EOF
+                    break
                 try:
                     splitted = line.strip().decode('ascii').split(':')
                     # name can potentially contain separator symbols (for
@@ -237,18 +267,42 @@ def main(fd, verbose=0):
                                 name, rtype, list(_CLEANUP_FUNCS.keys())))
 
                     if cmd == 'REGISTER':
-                        cache[rtype].add(name)
-                        if verbose:  # pragma: no cover
-                            sys.stderr.write("[ResourceTracker] register {}"
-                                             " {}\n" .format(rtype, name))
-                            sys.stderr.flush()
+                        if name not in registry[rtype]:
+                            registry[rtype][name] = 1
+                        else:
+                            registry[rtype][name] += 1
+
+                        if verbose:
+                            util.debug(
+                                "[ResourceTracker] incremented refcount of {} "
+                                "{} (current {})".format(
+                                    rtype, name, registry[rtype][name]))
                     elif cmd == 'UNREGISTER':
-                        cache[rtype].remove(name)
-                        if verbose:  # pragma: no cover
-                            sys.stderr.write("[ResourceTracker] unregister {}"
-                                             " {}: cache({})\n"
-                                             .format(name, rtype, len(cache)))
-                            sys.stderr.flush()
+                        del registry[rtype][name]
+                        if verbose:
+                            util.debug(
+                                "[ResourceTracker] unregister {} {}: "
+                                "registry({})".format(name, rtype, len(registry)))
+                    elif cmd == 'MAYBE_UNLINK':
+                        registry[rtype][name] -= 1
+                        if verbose:
+                            util.debug(
+                                "[ResourceTracker] decremented refcount of {} "
+                                "{} (current {})".format(
+                                    rtype, name, registry[rtype][name]))
+
+                        if registry[rtype][name] == 0:
+                            del registry[rtype][name]
+                            try:
+                                if verbose:
+                                    util.debug(
+                                            "[ResourceTracker] unlink {}"
+                                            .format(name))
+                                _CLEANUP_FUNCS[rtype](name)
+                            except Exception as e:
+                                warnings.warn(
+                                    'resource_tracker: %s: %r' % (name, e))
+
                     else:
                         raise RuntimeError('unrecognized command %r' % cmd)
                 except BaseException:
@@ -258,30 +312,43 @@ def main(fd, verbose=0):
                         pass
     finally:
         # all processes have terminated; cleanup any remaining resources
-        for rtype, rtype_cache in cache.items():
-            if rtype_cache:
+        def _unlink_resources(rtype_registry, rtype):
+            if rtype_registry:
                 try:
                     warnings.warn('resource_tracker: There appear to be %d '
                                   'leaked %s objects to clean up at shutdown' %
-                                  (len(rtype_cache), rtype))
+                                  (len(rtype_registry), rtype))
                 except Exception:
                     pass
-            for name in rtype_cache:
+            for name in rtype_registry:
                 # For some reason the process which created and registered this
                 # resource has failed to unregister it. Presumably it has
                 # died.  We therefore clean it up.
                 try:
                     _CLEANUP_FUNCS[rtype](name)
-                    if verbose:  # pragma: no cover
-                        sys.stderr.write("[ResourceTracker] unlink {}\n"
+                    if verbose:
+                        util.debug("[ResourceTracker] unlink {}"
                                          .format(name))
-                        sys.stderr.flush()
                 except Exception as e:
                     warnings.warn('resource_tracker: %s: %r' % (name, e))
 
-    if verbose:  # pragma: no cover
-        sys.stderr.write("resource tracker shut down\n")
-        sys.stderr.flush()
+        for rtype, rtype_registry in registry.items():
+            if rtype == "folder":
+                continue
+            else:
+                _unlink_resources(rtype_registry, rtype)
+
+        # The default cleanup routine for folders deletes everything inside
+        # those folders recursively, which can include other resources tracked
+        # by the resource tracker). To limit the risk of the resource tracker
+        # attempting to delete twice a resource (once as part of a tracked
+        # folder, and once as a resource), we delete the folders after all
+        # other resource types.
+        if "folder" in registry:
+            _unlink_resources(registry["folder"], "folder")
+
+    if verbose:
+        util.debug("resource tracker shut down")
 
 
 #
diff --git a/joblib/externals/loky/backend/spawn.py b/joblib/externals/loky/backend/spawn.py
index 0877f33..2a16c84 100644
--- a/joblib/externals/loky/backend/spawn.py
+++ b/joblib/externals/loky/backend/spawn.py
@@ -92,6 +92,26 @@ def get_preparation_data(name, init_main_module=True):
     else:
         d["tracker_args"]["fd"] = _resource_tracker._fd
 
+    if sys.version_info >= (3, 8) and os.name == 'posix':
+        # joblib/loky#242: allow loky processes to retrieve the resource
+        # tracker of their parent in case the child processes depickles
+        # shared_memory objects, that are still tracked by multiprocessing's
+        # resource_tracker by default.
+        # XXX: this is a workaround that may be error prone: in the future, it
+        # would be better to have loky subclass multiprocessing's shared_memory
+        # to force registration of shared_memory segments via loky's
+        # resource_tracker.
+        from multiprocessing.resource_tracker import (
+            _resource_tracker as mp_resource_tracker
+        )
+        # multiprocessing's resource_tracker must be running before loky
+        # process is created (othewise the child won't be able to use it if it
+        # is created later on)
+        mp_resource_tracker.ensure_running()
+        d["mp_tracker_args"] = {
+            'fd': mp_resource_tracker._fd, 'pid': mp_resource_tracker._pid
+        }
+
     # Figure out whether to initialise main in the subprocess as a module
     # or through direct execution (or to leave it alone entirely)
     if init_main_module:
@@ -155,6 +175,12 @@ def prepare(data):
     if 'orig_dir' in data:
         process.ORIGINAL_DIR = data['orig_dir']
 
+    if 'mp_tracker_args' in data:
+        from multiprocessing.resource_tracker import (
+            _resource_tracker as mp_resource_tracker
+        )
+        mp_resource_tracker._fd = data['mp_tracker_args']['fd']
+        mp_resource_tracker._pid = data['mp_tracker_args']['pid']
     if 'tracker_args' in data:
         from .resource_tracker import _resource_tracker
         _resource_tracker._pid = data["tracker_args"]['pid']
diff --git a/joblib/externals/loky/process_executor.py b/joblib/externals/loky/process_executor.py
index e9ae29e..06483cb 100644
--- a/joblib/externals/loky/process_executor.py
+++ b/joblib/externals/loky/process_executor.py
@@ -5,7 +5,7 @@
 #
 # adapted from concurrent/futures/process_pool_executor.py (17/02/2017)
 #  * Backport for python2.7/3.3,
-#  * Add an extra management thread to detect queue_management_thread failures,
+#  * Add an extra management thread to detect executor_manager_thread failures,
 #  * Improve the shutdown process to avoid deadlocks,
 #  * Add timeout for workers,
 #  * More robust pickling process.
@@ -79,7 +79,7 @@ from .backend.compat import queue
 from .backend.compat import wait
 from .backend.compat import set_cause
 from .backend.context import cpu_count
-from .backend.queues import Queue, SimpleQueue, Full
+from .backend.queues import Queue, SimpleQueue
 from .backend.reduction import set_loky_pickler, get_loky_pickler_name
 from .backend.utils import recursive_terminate, get_exitcodes_terminated_worker
 
@@ -94,23 +94,6 @@ if sys.version_info[0] == 2:
     ProcessLookupError = OSError
 
 
-# Workers are created as daemon threads and processes. This is done to allow
-# the interpreter to exit when there are still idle processes in a
-# ProcessPoolExecutor's process pool (i.e. shutdown() was not called). However,
-# allowing workers to die with the interpreter has two undesirable properties:
-#   - The workers would still be running during interpreter shutdown,
-#     meaning that they would fail in unpredictable ways.
-#   - The workers could be killed while evaluating a work item, which could
-#     be bad if the callable being evaluated has external side-effects e.g.
-#     writing to a file.
-#
-# To work around this problem, an exit handler is installed which tells the
-# workers to exit when their work queues are empty and then waits until the
-# threads/processes finish.
-
-_threads_wakeups = weakref.WeakKeyDictionary()
-_global_shutdown = False
-
 # Mechanism to prevent infinite process spawning. When a worker of a
 # ProcessPoolExecutor nested in MAX_DEPTH Executor tries to create a new
 # Executor, a LokyRecursionError is raised
@@ -121,7 +104,7 @@ _CURRENT_DEPTH = 0
 _MEMORY_LEAK_CHECK_DELAY = 1.
 
 # Number of bytes of memory usage allowed over the reference process size.
-_MAX_MEMORY_LEAK_SIZE = int(1e8)
+_MAX_MEMORY_LEAK_SIZE = int(3e8)
 
 
 try:
@@ -140,43 +123,49 @@ except ImportError:
 
 class _ThreadWakeup:
     def __init__(self):
+        self._closed = False
         self._reader, self._writer = mp.Pipe(duplex=False)
 
     def close(self):
-        self._writer.close()
-        self._reader.close()
+        if not self._closed:
+            self._closed = True
+            self._writer.close()
+            self._reader.close()
 
     def wakeup(self):
-        if sys.platform == "win32" and sys.version_info[:2] < (3, 4):
-            # Compat for python2.7 on windows, where poll return false for
-            # b"" messages. Use the slightly larger message b"0".
-            self._writer.send_bytes(b"0")
-        else:
-            self._writer.send_bytes(b"")
+        if not self._closed:
+            if sys.platform == "win32" and sys.version_info[:2] < (3, 4):
+                # Compat for python2.7 on windows, where poll return false for
+                # b"" messages. Use the slightly larger message b"0".
+                self._writer.send_bytes(b"0")
+            else:
+                self._writer.send_bytes(b"")
 
     def clear(self):
-        while self._reader.poll():
-            self._reader.recv_bytes()
+        if not self._closed:
+            while self._reader.poll():
+                self._reader.recv_bytes()
 
 
 class _ExecutorFlags(object):
     """necessary references to maintain executor states without preventing gc
 
-    It permits to keep the information needed by queue_management_thread
+    It permits to keep the information needed by executor_manager_thread
     and crash_detection_thread to maintain the pool without preventing the
     garbage collection of unreferenced executors.
     """
-    def __init__(self):
+    def __init__(self, shutdown_lock):
 
         self.shutdown = False
         self.broken = None
         self.kill_workers = False
-        self.shutdown_lock = threading.Lock()
+        self.shutdown_lock = shutdown_lock
 
-    def flag_as_shutting_down(self, kill_workers=False):
+    def flag_as_shutting_down(self, kill_workers=None):
         with self.shutdown_lock:
             self.shutdown = True
-            self.kill_workers = kill_workers
+            if kill_workers is not None:
+                self.kill_workers = kill_workers
 
     def flag_as_broken(self, broken):
         with self.shutdown_lock:
@@ -184,19 +173,43 @@ class _ExecutorFlags(object):
             self.broken = broken
 
 
+# Prior to 3.9, executor_manager_thread is created as daemon thread. This means
+# that it is not joined automatically when the interpreter is shutting down.
+# To work around this problem, an exit handler is installed to tell the
+# thread to exit when the interpreter is shutting down and then waits until
+# it finishes. The thread needs to be daemonized because the atexit hooks are
+# called after all non daemonized threads are joined.
+#
+# Starting 3.9, there exists a specific atexit hook to be called before joining
+# the threads so the executor_manager_thread does not need to be daemonized
+# anymore.
+#
+# The atexit hooks are registered when starting the first ProcessPoolExecutor
+# to avoid import having an effect on the interpreter.
+
+_threads_wakeups = weakref.WeakKeyDictionary()
+_global_shutdown = False
+
+
 def _python_exit():
     global _global_shutdown
     _global_shutdown = True
     items = list(_threads_wakeups.items())
-    mp.util.debug("Interpreter shutting down. Waking up queue_manager_threads "
-                  "{}".format(items))
-    for thread, thread_wakeup in items:
-        if thread.is_alive():
+    mp.util.debug("Interpreter shutting down. Waking up "
+                  "executor_manager_thread {}".format(items))
+    for _, (shutdown_lock, thread_wakeup) in items:
+        with shutdown_lock:
             thread_wakeup.wakeup()
     for thread, _ in items:
         thread.join()
 
 
+# With the fork context, _thread_wakeups is propagated to children.
+# Clear it after fork to avoid some situation that can cause some
+# freeze when joining the workers.
+mp.util.register_after_fork(_threads_wakeups, lambda obj: obj.clear())
+
+
 # Module variable to register the at_exit call
 process_pool_executor_at_exit = None
 
@@ -211,7 +224,7 @@ class _RemoteTraceback(Exception):
     """Embed stringification of remote traceback in local traceback
     """
     def __init__(self, tb=None):
-        self.tb = tb
+        self.tb = '\n"""\n{}"""'.format(tb)
 
     def __str__(self):
         return self.tb
@@ -226,7 +239,7 @@ class _ExceptionWithTraceback(BaseException):
         tb = traceback.format_exception(type(exc), exc, tb)
         tb = ''.join(tb)
         self.exc = exc
-        self.tb = '\n"""\n%s"""' % tb
+        self.tb = tb
 
     def __reduce__(self):
         return _rebuild_exc, (self.exc, self.tb)
@@ -297,12 +310,12 @@ class _SafeQueue(Queue):
                     "Could not pickle the task to send it to the workers.")
             tb = traceback.format_exception(
                 type(e), e, getattr(e, "__traceback__", None))
-            raised_error = set_cause(raised_error, _RemoteTraceback(
-                                     '\n"""\n{}"""'.format(''.join(tb))))
+            raised_error = set_cause(raised_error,
+                                     _RemoteTraceback(''.join(tb)))
             work_item = self.pending_work_items.pop(obj.work_id, None)
             self.running_work_items.remove(obj.work_id)
             # work_item can be None if another process terminated. In this
-            # case, the queue_manager_thread fails all work_items with
+            # case, the executor_manager_thread fails all work_items with
             # BrokenProcessPool
             if work_item is not None:
                 work_item.future.set_exception(raised_error)
@@ -465,224 +478,196 @@ def _process_worker(call_queue, result_queue, initializer, initargs,
                 _last_memory_leak_check = time()
 
 
-def _add_call_item_to_queue(pending_work_items,
-                            running_work_items,
-                            work_ids,
-                            call_queue):
-    """Fills call_queue with _WorkItems from pending_work_items.
+class _ExecutorManagerThread(threading.Thread):
+    """Manages the communication between this process and the worker processes.
 
-    This function never blocks.
+    The manager is run in a local thread.
 
     Args:
-        pending_work_items: A dict mapping work ids to _WorkItems e.g.
-            {5: <_WorkItem...>, 6: <_WorkItem...>, ...}
-        work_ids: A queue.Queue of work ids e.g. Queue([5, 6, ...]). Work ids
-            are consumed and the corresponding _WorkItems from
-            pending_work_items are transformed into _CallItems and put in
-            call_queue.
-        call_queue: A ctx.Queue that will be filled with _CallItems
-            derived from _WorkItems.
+        executor: A reference to the ProcessPoolExecutor that owns
+            this thread. A weakref will be own by the manager as well as
+            references to internal objects used to introspect the state of
+            the executor.
     """
-    while True:
-        if call_queue.full():
-            return
-        try:
-            work_id = work_ids.get(block=False)
-        except queue.Empty:
-            return
-        else:
-            work_item = pending_work_items[work_id]
-
-            if work_item.future.set_running_or_notify_cancel():
-                running_work_items += [work_id]
-                call_queue.put(_CallItem(work_id,
-                                         work_item.fn,
-                                         work_item.args,
-                                         work_item.kwargs),
-                               block=True)
-            else:
-                del pending_work_items[work_id]
-                continue
 
+    def __init__(self, executor):
+        # Store references to necessary internals of the executor.
+
+        # A _ThreadWakeup to allow waking up the executor_manager_thread from
+        # the main Thread and avoid deadlocks caused by permanently
+        # locked queues.
+        self.thread_wakeup = executor._executor_manager_thread_wakeup
+        self.shutdown_lock = executor._shutdown_lock
+
+        # A weakref.ref to the ProcessPoolExecutor that owns this thread. Used
+        # to determine if the ProcessPoolExecutor has been garbage collected
+        # and that the manager can exit.
+        # When the executor gets garbage collected, the weakref callback
+        # will wake up the queue management thread so that it can terminate
+        # if there is no pending work item.
+        def weakref_cb(_,
+                       thread_wakeup=self.thread_wakeup,
+                       shutdown_lock=self.shutdown_lock):
+            mp.util.debug('Executor collected: triggering callback for'
+                          ' QueueManager wakeup')
+            with shutdown_lock:
+                thread_wakeup.wakeup()
 
-def _queue_management_worker(executor_reference,
-                             executor_flags,
-                             processes,
-                             pending_work_items,
-                             running_work_items,
-                             work_ids_queue,
-                             call_queue,
-                             result_queue,
-                             thread_wakeup,
-                             processes_management_lock):
-    """Manages the communication between this process and the worker processes.
+        self.executor_reference = weakref.ref(executor, weakref_cb)
 
-    This function is run in a local thread.
+        # The flags of the executor
+        self.executor_flags = executor._flags
 
-    Args:
-        executor_reference: A weakref.ref to the ProcessPoolExecutor that owns
-            this thread. Used to determine if the ProcessPoolExecutor has been
-            garbage collected and that this function can exit.
-        executor_flags: A ExecutorFlags holding internal states of the
-            ProcessPoolExecutor. It permits to know if the executor is broken
-            even the object has been gc.
-        process: A list of the ctx.Process instances used as
-            workers.
-        pending_work_items: A dict mapping work ids to _WorkItems e.g.
-            {5: <_WorkItem...>, 6: <_WorkItem...>, ...}
-        work_ids_queue: A queue.Queue of work ids e.g. Queue([5, 6, ...]).
-        call_queue: A ctx.Queue that will be filled with _CallItems
-            derived from _WorkItems for processing by the process workers.
-        result_queue: A ctx.SimpleQueue of _ResultItems generated by the
-            process workers.
-        thread_wakeup: A _ThreadWakeup to allow waking up the
-            queue_manager_thread from the main Thread and avoid deadlocks
-            caused by permanently locked queues.
-    """
-    executor = None
+        # A list of the ctx.Process instances used as workers.
+        self.processes = executor._processes
 
-    def is_shutting_down():
-        # No more work items can be added if:
-        #   - The interpreter is shutting down OR
-        #   - The executor that own this worker is not broken AND
-        #        * The executor that owns this worker has been collected OR
-        #        * The executor that owns this worker has been shutdown.
-        # If the executor is broken, it should be detected in the next loop.
-        return (_global_shutdown or
-                ((executor is None or executor_flags.shutdown)
-                 and not executor_flags.broken))
+        # A ctx.Queue that will be filled with _CallItems derived from
+        # _WorkItems for processing by the process workers.
+        self.call_queue = executor._call_queue
 
-    def shutdown_all_workers():
-        mp.util.debug("queue management thread shutting down")
-        executor_flags.flag_as_shutting_down()
-        # Create a list to avoid RuntimeError due to concurrent modification of
-        # processes. nb_children_alive is thus an upper bound. Also release the
-        # processes' _worker_exit_lock to accelerate the shutdown procedure, as
-        # there is no need for hand-shake here.
-        with processes_management_lock:
-            n_children_alive = 0
-            for p in list(processes.values()):
-                p._worker_exit_lock.release()
-                n_children_alive += 1
-        n_children_to_stop = n_children_alive
-        n_sentinels_sent = 0
-        # Send the right number of sentinels, to make sure all children are
-        # properly terminated.
-        while n_sentinels_sent < n_children_to_stop and n_children_alive > 0:
-            for i in range(n_children_to_stop - n_sentinels_sent):
-                try:
-                    call_queue.put_nowait(None)
-                    n_sentinels_sent += 1
-                except Full:
-                    break
-            with processes_management_lock:
-                n_children_alive = sum(
-                    p.is_alive() for p in list(processes.values())
-                )
+        # A ctx.SimpleQueue of _ResultItems generated by the process workers.
+        self.result_queue = executor._result_queue
 
-        # Release the queue's resources as soon as possible. Flag the feeder
-        # thread for clean exit to avoid having the crash detection thread flag
-        # the Executor as broken during the shutdown. This is safe as either:
-        #  * We don't need to communicate with the workers anymore
-        #  * There is nothing left in the Queue buffer except None sentinels
-        mp.util.debug("closing call_queue")
-        call_queue.close()
+        # A queue.Queue of work ids e.g. Queue([5, 6, ...]).
+        self.work_ids_queue = executor._work_ids
 
-        mp.util.debug("joining processes")
-        # If .join() is not called on the created processes then
-        # some ctx.Queue methods may deadlock on Mac OS X.
-        while processes:
-            _, p = processes.popitem()
-            p.join()
-        mp.util.debug("queue management thread clean shutdown of worker "
-                      "processes: {}".format(list(processes)))
+        # A dict mapping work ids to _WorkItems e.g.
+        #     {5: <_WorkItem...>, 6: <_WorkItem...>, ...}
+        self.pending_work_items = executor._pending_work_items
 
-    result_reader = result_queue._reader
-    wakeup_reader = thread_wakeup._reader
-    readers = [result_reader, wakeup_reader]
+        # A list of the work_ids that are currently running
+        self.running_work_items = executor._running_work_items
 
-    while True:
-        _add_call_item_to_queue(pending_work_items,
-                                running_work_items,
-                                work_ids_queue,
-                                call_queue)
+        # A lock to avoid concurrent shutdown of workers on timeout and spawn
+        # of new processes or shut down
+        self.processes_management_lock = executor._processes_management_lock
+
+        super(_ExecutorManagerThread, self).__init__()
+        if sys.version_info < (3, 9):
+            self.daemon = True
+
+    def run(self):
+        # Main loop for the executor manager thread.
+
+        while True:
+            self.add_call_item_to_queue()
+
+            result_item, is_broken, bpe = self.wait_result_broken_or_wakeup()
+
+            if is_broken:
+                self.terminate_broken(bpe)
+                return
+            if result_item is not None:
+                self.process_result_item(result_item)
+                # Delete reference to result_item to avoid keeping references
+                # while waiting on new results.
+                del result_item
+
+            if self.is_shutting_down():
+                self.flag_executor_shutting_down()
+
+                # Since no new work items can be added, it is safe to shutdown
+                # this thread if there are no pending work items.
+                if not self.pending_work_items:
+                    self.join_executor_internals()
+                    return
+
+    def add_call_item_to_queue(self):
+        # Fills call_queue with _WorkItems from pending_work_items.
+        # This function never blocks.
+        while True:
+            if self.call_queue.full():
+                return
+            try:
+                work_id = self.work_ids_queue.get(block=False)
+            except queue.Empty:
+                return
+            else:
+                work_item = self.pending_work_items[work_id]
+
+                if work_item.future.set_running_or_notify_cancel():
+                    self.running_work_items += [work_id]
+                    self.call_queue.put(_CallItem(work_id,
+                                                  work_item.fn,
+                                                  work_item.args,
+                                                  work_item.kwargs),
+                                        block=True)
+                else:
+                    del self.pending_work_items[work_id]
+                    continue
+
+    def wait_result_broken_or_wakeup(self):
         # Wait for a result to be ready in the result_queue while checking
         # that all worker processes are still running, or for a wake up
         # signal send. The wake up signals come either from new tasks being
         # submitted, from the executor being shutdown/gc-ed, or from the
         # shutdown of the python interpreter.
-        worker_sentinels = [p.sentinel for p in list(processes.values())]
+        result_reader = self.result_queue._reader
+        wakeup_reader = self.thread_wakeup._reader
+        readers = [result_reader, wakeup_reader]
+        worker_sentinels = [p.sentinel for p in self.processes.values()]
         ready = wait(readers + worker_sentinels)
 
-        broken = ("A worker process managed by the executor was unexpectedly "
-                  "terminated. This could be caused by a segmentation fault "
-                  "while calling the function or by an excessive memory usage "
-                  "causing the Operating System to kill the worker.", None,
-                  TerminatedWorkerError)
+        bpe = None
+        is_broken = True
+        result_item = None
         if result_reader in ready:
             try:
                 result_item = result_reader.recv()
-                broken = None
                 if isinstance(result_item, _RemoteTraceback):
-                    broken = ("A task has failed to un-serialize. Please "
-                              "ensure that the arguments of the function are "
-                              "all picklable.", result_item.tb,
-                              BrokenProcessPool)
+                    bpe = BrokenProcessPool(
+                        "A task has failed to un-serialize. Please ensure that"
+                        " the arguments of the function are all picklable."
+                    )
+                    set_cause(bpe, result_item)
+                else:
+                    is_broken = False
             except BaseException as e:
-                tb = getattr(e, "__traceback__", None)
-                if tb is None:
-                    _, _, tb = sys.exc_info()
-                broken = ("A result has failed to un-serialize. Please "
-                          "ensure that the objects returned by the function "
-                          "are always picklable.",
-                          traceback.format_exception(type(e), e, tb),
-                          BrokenProcessPool)
+                bpe = BrokenProcessPool(
+                    "A result has failed to un-serialize. Please ensure that "
+                    "the objects returned by the function are always "
+                    "picklable."
+                )
+                tb = traceback.format_exception(
+                    type(e), e, getattr(e, "__traceback__", None))
+                set_cause(bpe,  _RemoteTraceback(''.join(tb)))
+
         elif wakeup_reader in ready:
-            broken = None
-            result_item = None
-        thread_wakeup.clear()
-        if broken is not None:
-            msg, cause_tb, exc_type = broken
-            if (issubclass(exc_type, TerminatedWorkerError) and
-                    (sys.platform != "win32")):
+            # This is simply a wake-up event that might either trigger putting
+            # more tasks in the queue or trigger the clean up of resources.
+            is_broken = False
+        else:
+            # A worker has terminated and we don't know why, set the state of
+            # the executor as broken
+            exit_codes = ''
+            if sys.platform != "win32":
                 # In Windows, introspecting terminated workers exitcodes seems
                 # unstable, therefore they are not appended in the exception
                 # message.
-                msg += " The exit codes of the workers are {}".format(
-                    get_exitcodes_terminated_worker(processes))
-
-            bpe = exc_type(msg)
-            if cause_tb is not None:
-                bpe = set_cause(bpe, _RemoteTraceback(
-                          "\n'''\n{}'''".format(''.join(cause_tb))))
-            # Mark the process pool broken so that submits fail right now.
-            executor_flags.flag_as_broken(bpe)
-
-            # All futures in flight must be marked failed
-            for work_id, work_item in pending_work_items.items():
-                work_item.future.set_exception(bpe)
-                # Delete references to object. See issue16284
-                del work_item
-            pending_work_items.clear()
+                exit_codes = "\nThe exit codes of the workers are {}".format(
+                    get_exitcodes_terminated_worker(self.processes))
+            bpe = TerminatedWorkerError(
+                "A worker process managed by the executor was unexpectedly "
+                "terminated. This could be caused by a segmentation fault "
+                "while calling the function or by an excessive memory usage "
+                "causing the Operating System to kill the worker.\n"
+                "{}".format(exit_codes)
+            )
 
-            # Terminate remaining workers forcibly: the queues or their
-            # locks may be in a dirty state and block forever.
-            while processes:
-                _, p = processes.popitem()
-                mp.util.debug('terminate process {}'.format(p.name))
-                try:
-                    recursive_terminate(p)
-                except ProcessLookupError:  # pragma: no cover
-                    pass
+        self.thread_wakeup.clear()
+
+        return result_item, is_broken, bpe
+
+    def process_result_item(self, result_item):
+        # Process the received a result_item. This can be either the PID of a
+        # worker that exited gracefully or a _ResultItem
 
-            shutdown_all_workers()
-            return
         if isinstance(result_item, int):
             # Clean shutdown of a worker using its PID, either on request
             # by the executor.shutdown method or by the timeout of the worker
             # itself: we should not mark the executor as broken.
-            with processes_management_lock:
-                p = processes.pop(result_item, None)
+            with self.processes_management_lock:
+                p = self.processes.pop(result_item, None)
 
             # p can be None is the executor is concurrently shutting down.
             if p is not None:
@@ -694,12 +679,12 @@ def _queue_management_worker(executor_reference,
             # worker timeout while some jobs were submitted. If some work is
             # pending or there is less processes than running items, we need to
             # start a new Process and raise a warning.
-            n_pending = len(pending_work_items)
-            n_running = len(running_work_items)
-            if (n_pending - n_running > 0 or n_running > len(processes)):
-                executor = executor_reference()
+            n_pending = len(self.pending_work_items)
+            n_running = len(self.running_work_items)
+            if (n_pending - n_running > 0 or n_running > len(self.processes)):
+                executor = self.executor_reference()
                 if (executor is not None
-                        and len(processes) < executor._max_workers):
+                        and len(self.processes) < executor._max_workers):
                     warnings.warn(
                         "A worker stopped while some jobs were given to the "
                         "executor. This can be caused by a too short worker "
@@ -707,54 +692,140 @@ def _queue_management_worker(executor_reference,
                     )
                     executor._adjust_process_count()
                     executor = None
-
-        elif result_item is not None:
-            work_item = pending_work_items.pop(result_item.work_id, None)
-            # work_item can be None if another process terminated
+        else:
+            # Received a _ResultItem so mark the future as completed.
+            work_item = self.pending_work_items.pop(result_item.work_id, None)
+            # work_item can be None if another process terminated (see above)
             if work_item is not None:
                 if result_item.exception:
                     work_item.future.set_exception(result_item.exception)
                 else:
                     work_item.future.set_result(result_item.result)
-                # Delete references to object. See issue16284
-                del work_item
-                running_work_items.remove(result_item.work_id)
-            # Delete reference to result_item
-            del result_item
+                self.running_work_items.remove(result_item.work_id)
 
-        # Check whether we should start shutting down.
-        executor = executor_reference()
+    def is_shutting_down(self):
+        # Check whether we should start shutting down the executor.
+        executor = self.executor_reference()
         # No more work items can be added if:
         #   - The interpreter is shutting down OR
-        #   - The executor that owns this worker has been collected OR
-        #   - The executor that owns this worker has been shutdown.
-        if is_shutting_down():
-            # bpo-33097: Make sure that the executor is flagged as shutting
-            # down even if it is shutdown by the interpreter exiting.
-            with executor_flags.shutdown_lock:
-                executor_flags.shutdown = True
-            if executor_flags.kill_workers:
-                while pending_work_items:
-                    _, work_item = pending_work_items.popitem()
-                    work_item.future.set_exception(ShutdownExecutorError(
-                        "The Executor was shutdown before this job could "
-                        "complete."))
-                    del work_item
-                # Terminate remaining workers forcibly: the queues or their
-                # locks may be in a dirty state and block forever.
-                while processes:
-                    _, p = processes.popitem()
-                    recursive_terminate(p)
-                shutdown_all_workers()
-                return
-            # Since no new work items can be added, it is safe to shutdown
-            # this thread if there are no pending work items.
-            if not pending_work_items:
-                shutdown_all_workers()
-                return
-        elif executor_flags.broken:
-            return
-        executor = None
+        #   - The executor that owns this thread is not broken AND
+        #        * The executor that owns this worker has been collected OR
+        #        * The executor that owns this worker has been shutdown.
+        # If the executor is broken, it should be detected in the next loop.
+        return (_global_shutdown or
+                ((executor is None or self.executor_flags.shutdown)
+                 and not self.executor_flags.broken))
+
+    def terminate_broken(self, bpe):
+        # Terminate the executor because it is in a broken state. The bpe
+        # argument can be used to display more information on the error that
+        # lead the executor into becoming broken.
+
+        # Mark the process pool broken so that submits fail right now.
+        self.executor_flags.flag_as_broken(bpe)
+
+        # Mark pending tasks as failed.
+        for work_id, work_item in self.pending_work_items.items():
+            work_item.future.set_exception(bpe)
+            # Delete references to object. See issue16284
+            del work_item
+        self.pending_work_items.clear()
+
+        # Terminate remaining workers forcibly: the queues or their
+        # locks may be in a dirty state and block forever.
+        self.kill_workers()
+
+        # clean up resources
+        self.join_executor_internals()
+
+    def flag_executor_shutting_down(self):
+        # Flag the executor as shutting down and cancel remaining tasks if
+        # requested as early as possible if it is not gc-ed yet.
+        self.executor_flags.flag_as_shutting_down()
+
+        # Cancel pending work items if requested.
+        if self.executor_flags.kill_workers:
+            while self.pending_work_items:
+                _, work_item = self.pending_work_items.popitem()
+                work_item.future.set_exception(ShutdownExecutorError(
+                    "The Executor was shutdown with `kill_workers=True` "
+                    "before this job could complete."))
+                del work_item
+
+            # Kill the remaining worker forcibly to no waste time joining them
+            self.kill_workers()
+
+    def kill_workers(self):
+        # Terminate the remaining workers using SIGKILL. This function also
+        # terminates descendant workers of the children in case there is some
+        # nested parallelism.
+        while self.processes:
+            _, p = self.processes.popitem()
+            mp.util.debug('terminate process {}'.format(p.name))
+            try:
+                recursive_terminate(p)
+            except ProcessLookupError:  # pragma: no cover
+                pass
+
+    def shutdown_workers(self):
+        # shutdown all workers in self.processes
+
+        # Create a list to avoid RuntimeError due to concurrent modification of
+        # processes. nb_children_alive is thus an upper bound. Also release the
+        # processes' _worker_exit_lock to accelerate the shutdown procedure, as
+        # there is no need for hand-shake here.
+        with self.processes_management_lock:
+            n_children_to_stop = 0
+            for p in list(self.processes.values()):
+                p._worker_exit_lock.release()
+                n_children_to_stop += 1
+
+        # Send the right number of sentinels, to make sure all children are
+        # properly terminated. Do it with a mechanism that avoid hanging on
+        # Full queue when all workers have already been shutdown.
+        n_sentinels_sent = 0
+        while (n_sentinels_sent < n_children_to_stop
+                and self.get_n_children_alive() > 0):
+            for i in range(n_children_to_stop - n_sentinels_sent):
+                try:
+                    self.call_queue.put_nowait(None)
+                    n_sentinels_sent += 1
+                except queue.Full:
+                    break
+
+    def join_executor_internals(self):
+        self.shutdown_workers()
+
+        # Release the queue's resources as soon as possible. Flag the feeder
+        # thread for clean exit to avoid having the crash detection thread flag
+        # the Executor as broken during the shutdown. This is safe as either:
+        #  * We don't need to communicate with the workers anymore
+        #  * There is nothing left in the Queue buffer except None sentinels
+        mp.util.debug("closing call_queue")
+        self.call_queue.close()
+        self.call_queue.join_thread()
+
+        # Closing result_queue
+        mp.util.debug("closing result_queue")
+        self.result_queue.close()
+
+        mp.util.debug("closing thread_wakeup")
+        with self.shutdown_lock:
+            self.thread_wakeup.close()
+
+        # If .join() is not called on the created processes then
+        # some ctx.Queue methods may deadlock on Mac OS X.
+        mp.util.debug("joining processes")
+        for p in self.processes.values():
+            p.join()
+
+        mp.util.debug("executor management thread clean shutdown of worker "
+                      "processes: {}".format(list(self.processes)))
+
+    def get_n_children_alive(self):
+        # This is an upper bound on the number of children alive.
+        with self.processes_management_lock:
+            return sum(p.is_alive() for p in list(self.processes.values()))
 
 
 _system_limits_checked = False
@@ -906,6 +977,12 @@ class ProcessPoolExecutor(_base.Executor):
         # Timeout
         self._timeout = timeout
 
+        # Management thread
+        self._executor_manager_thread = None
+
+        # Map of pids to processes
+        self._processes = {}
+
         # Internal variables of the ProcessPoolExecutor
         self._processes = {}
         self._queue_count = 0
@@ -913,19 +990,22 @@ class ProcessPoolExecutor(_base.Executor):
         self._running_work_items = []
         self._work_ids = queue.Queue()
         self._processes_management_lock = self._context.Lock()
-        self._queue_management_thread = None
+        self._executor_manager_thread = None
+        self._shutdown_lock = threading.Lock()
 
         # _ThreadWakeup is a communication channel used to interrupt the wait
-        # of the main loop of queue_manager_thread from another thread (e.g.
+        # of the main loop of executor_manager_thread from another thread (e.g.
         # when calling executor.submit or executor.shutdown). We do not use the
-        # _result_queue to send the wakeup signal to the queue_manager_thread
+        # _result_queue to send wakeup signals to the executor_manager_thread
         # as it could result in a deadlock if a worker process dies with the
         # _result_queue write lock still acquired.
-        self._queue_management_thread_wakeup = _ThreadWakeup()
+        #
+        # _shutdown_lock must be locked to access _ThreadWakeup.wakeup.
+        self._executor_manager_thread_wakeup = _ThreadWakeup()
 
         # Flag to hold the state of the Executor. This permits to introspect
         # the Executor state even once it has been garbage collected.
-        self._flags = _ExecutorFlags()
+        self._flags = _ExecutorFlags(self._shutdown_lock)
 
         # Finally setup the queues for interprocess communication
         self._setup_queues(job_reducers, result_reducers)
@@ -941,7 +1021,7 @@ class ProcessPoolExecutor(_base.Executor):
         self._call_queue = _SafeQueue(
             max_size=queue_size, pending_work_items=self._pending_work_items,
             running_work_items=self._running_work_items,
-            thread_wakeup=self._queue_management_thread_wakeup,
+            thread_wakeup=self._executor_manager_thread_wakeup,
             reducers=job_reducers, ctx=self._context)
         # Killed worker processes can produce spurious "broken pipe"
         # tracebacks in the queue's own worker thread. But we detect killed
@@ -951,48 +1031,43 @@ class ProcessPoolExecutor(_base.Executor):
         self._result_queue = SimpleQueue(reducers=result_reducers,
                                          ctx=self._context)
 
-    def _start_queue_management_thread(self):
-        if self._queue_management_thread is None:
-            mp.util.debug('_start_queue_management_thread called')
+    def _start_executor_manager_thread(self):
+        if self._executor_manager_thread is None:
+            mp.util.debug('_start_executor_manager_thread called')
 
             # When the executor gets garbarge collected, the weakref callback
             # will wake up the queue management thread so that it can terminate
             # if there is no pending work item.
-            def weakref_cb(_,
-                           thread_wakeup=self._queue_management_thread_wakeup):
+            def weakref_cb(
+                    _, thread_wakeup=self._executor_manager_thread_wakeup,
+                    shutdown_lock=self._shutdown_lock):
                 mp.util.debug('Executor collected: triggering callback for'
                               ' QueueManager wakeup')
-                thread_wakeup.wakeup()
+                with self._shutdown_lock:
+                    thread_wakeup.wakeup()
 
             # Start the processes so that their sentinels are known.
-            self._queue_management_thread = threading.Thread(
-                target=_queue_management_worker,
-                args=(weakref.ref(self, weakref_cb),
-                      self._flags,
-                      self._processes,
-                      self._pending_work_items,
-                      self._running_work_items,
-                      self._work_ids,
-                      self._call_queue,
-                      self._result_queue,
-                      self._queue_management_thread_wakeup,
-                      self._processes_management_lock),
-                name="QueueManagerThread")
-            self._queue_management_thread.daemon = True
-            self._queue_management_thread.start()
+            self._executor_manager_thread = _ExecutorManagerThread(self)
+            self._executor_manager_thread.start()
 
             # register this executor in a mechanism that ensures it will wakeup
             # when the interpreter is exiting.
-            _threads_wakeups[self._queue_management_thread] = \
-                self._queue_management_thread_wakeup
+            _threads_wakeups[self._executor_manager_thread] = \
+                (self._shutdown_lock,
+                 self._executor_manager_thread_wakeup)
 
             global process_pool_executor_at_exit
             if process_pool_executor_at_exit is None:
                 # Ensure that the _python_exit function will be called before
                 # the multiprocessing.Queue._close finalizers which have an
                 # exitpriority of 10.
-                process_pool_executor_at_exit = mp.util.Finalize(
-                    None, _python_exit, exitpriority=20)
+
+                if sys.version_info < (3, 9):
+                    process_pool_executor_at_exit = mp.util.Finalize(
+                        None, _python_exit, exitpriority=20)
+                else:
+                    process_pool_executor_at_exit = threading._register_atexit(
+                        _python_exit)
 
     def _adjust_process_count(self):
         for _ in range(len(self._processes), self._max_workers):
@@ -1019,7 +1094,7 @@ class ProcessPoolExecutor(_base.Executor):
         with self._processes_management_lock:
             if len(self._processes) != self._max_workers:
                 self._adjust_process_count()
-            self._start_queue_management_thread()
+            self._start_executor_manager_thread()
 
     def submit(self, fn, *args, **kwargs):
         with self._flags.shutdown_lock:
@@ -1042,7 +1117,7 @@ class ProcessPoolExecutor(_base.Executor):
             self._work_ids.put(self._queue_count)
             self._queue_count += 1
             # Wake up queue management thread
-            self._queue_management_thread_wakeup.wakeup()
+            self._executor_manager_thread_wakeup.wakeup()
 
             self._ensure_executor_running()
             return f
@@ -1084,35 +1159,23 @@ class ProcessPoolExecutor(_base.Executor):
         mp.util.debug('shutting down executor %s' % self)
 
         self._flags.flag_as_shutting_down(kill_workers)
-        qmt = self._queue_management_thread
-        qmtw = self._queue_management_thread_wakeup
-        if qmt:
-            self._queue_management_thread = None
-            if qmtw:
-                self._queue_management_thread_wakeup = None
+        executor_manager_thread = self._executor_manager_thread
+        executor_manager_thread_wakeup = self._executor_manager_thread_wakeup
+
+        if executor_manager_thread_wakeup is not None:
             # Wake up queue management thread
-            if qmtw is not None:
-                try:
-                    qmtw.wakeup()
-                except OSError:
-                    # Can happen in case of concurrent calls to shutdown.
-                    pass
-            if wait:
-                qmt.join()
-
-        cq = self._call_queue
-        if cq:
-            self._call_queue = None
-            cq.close()
-            if wait:
-                cq.join_thread()
+            with self._shutdown_lock:
+                self._executor_manager_thread_wakeup.wakeup()
+
+        if executor_manager_thread is not None and wait:
+            executor_manager_thread.join()
+
+        # To reduce the risk of opening too many files, remove references to
+        # objects that use file descriptors.
+        self._executor_manager_thread = None
+        self._executor_manager_thread_wakeup = None
+        self._call_queue = None
         self._result_queue = None
         self._processes_management_lock = None
 
-        if qmtw:
-            try:
-                qmtw.close()
-            except OSError:
-                # Can happen in case of concurrent calls to shutdown.
-                pass
     shutdown.__doc__ = _base.Executor.shutdown.__doc__
diff --git a/joblib/externals/loky/reusable_executor.py b/joblib/externals/loky/reusable_executor.py
index b2af6f4..11b6807 100644
--- a/joblib/externals/loky/reusable_executor.py
+++ b/joblib/externals/loky/reusable_executor.py
@@ -166,8 +166,8 @@ class _ReusablePoolExecutor(ProcessPoolExecutor):
             elif max_workers == self._max_workers:
                 return
 
-            if self._queue_management_thread is None:
-                # If the queue_management_thread has not been started
+            if self._executor_manager_thread is None:
+                # If the executor_manager_thread has not been started
                 # then no processes have been spawned and we can just
                 # update _max_workers and return
                 self._max_workers = max_workers
diff --git a/joblib/externals/vendor_loky.sh b/joblib/externals/vendor_loky.sh
old mode 100644
new mode 100755
index 21631eb..da57b52
--- a/joblib/externals/vendor_loky.sh
+++ b/joblib/externals/vendor_loky.sh
@@ -19,12 +19,13 @@ rm -rf $INSTALL_FOLDER
 # Note: BSD sed -i needs an argument unders OSX
 # so first renaming to .bak and then deleting backup files
 find loky -name "*.py" | xargs sed -i.bak "s/from loky/from joblib.externals.loky/"
-find loky -name "*.bak" | xargs rm
 
 for f in $(git grep -l "cloudpickle" loky); do
     echo $f;
-    sed -i 's/import cloudpickle/from joblib.externals import cloudpickle/' $f
-    sed -i 's/from cloudpickle import/from joblib.externals.cloudpickle import/' $f
+    sed -i.bak 's/import cloudpickle/from joblib.externals import cloudpickle/' $f
+    sed -i.bak 's/from cloudpickle import/from joblib.externals.cloudpickle import/' $f
 done
 
-sed -i "s/loky.backend.popen_loky/joblib.externals.loky.backend.popen_loky/" loky/backend/popen_loky_posix.py
+sed -i.bak "s/loky.backend.popen_loky/joblib.externals.loky.backend.popen_loky/" loky/backend/popen_loky_posix.py
+
+find loky -name "*.bak" | xargs rm
diff --git a/joblib/numpy_pickle.py b/joblib/numpy_pickle.py
index 83afb81..ad716fe 100644
--- a/joblib/numpy_pickle.py
+++ b/joblib/numpy_pickle.py
@@ -37,6 +37,7 @@ register_compressor('lzma', LZMACompressorWrapper())
 register_compressor('xz', XZCompressorWrapper())
 register_compressor('lz4', LZ4CompressorWrapper())
 
+
 ###############################################################################
 # Utility objects for persistence.
 
@@ -518,6 +519,15 @@ def _unpickle(fobj, filename="", mmap_mode=None):
     return obj
 
 
+def load_temporary_memmap(filename, mmap_mode, unlink_on_gc_collect):
+    from ._memmapping_reducer import JOBLIB_MMAPS, add_maybe_unlink_finalizer
+    obj = load(filename, mmap_mode)
+    JOBLIB_MMAPS.add(obj.filename)
+    if unlink_on_gc_collect:
+        add_maybe_unlink_finalizer(obj)
+    return obj
+
+
 def load(filename, mmap_mode=None):
     """Reconstruct a Python object from a file persisted with joblib.dump.
 
@@ -573,5 +583,4 @@ def load(filename, mmap_mode=None):
                     return load_compatibility(fobj)
 
                 obj = _unpickle(fobj, filename, mmap_mode)
-
     return obj
diff --git a/joblib/pool.py b/joblib/pool.py
index f3fe30a..67b185e 100644
--- a/joblib/pool.py
+++ b/joblib/pool.py
@@ -28,8 +28,8 @@ from pickle import Pickler
 from pickle import HIGHEST_PROTOCOL
 from io import BytesIO
 
-from .disk import delete_folder
 from ._memmapping_reducer import get_memmapping_reducers
+from ._memmapping_reducer import TemporaryResourcesManagerMixin
 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):
+class MemmappingPool(PicklingPool, TemporaryResourcesManagerMixin):
     """Process pool that shares large arrays to avoid memory copy.
 
     This drop-in replacement for `multiprocessing.pool.Pool` makes
@@ -301,7 +301,7 @@ class MemmappingPool(PicklingPool):
                 id(self), temp_folder=temp_folder, max_nbytes=max_nbytes,
                 mmap_mode=mmap_mode, forward_reducers=forward_reducers,
                 backward_reducers=backward_reducers, verbose=verbose,
-                prewarm=prewarm)
+                unlink_on_gc_collect=False, prewarm=prewarm)
 
         poolargs = dict(
             processes=processes,
@@ -324,4 +324,4 @@ class MemmappingPool(PicklingPool):
                     if i + 1 == n_retries:
                         warnings.warn("Failed to terminate worker processes in"
                                       " multiprocessing pool: %r" % e)
-        delete_folder(self._temp_folder)
+        self._unlink_temporary_resources()


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 a753bed..4013efa 100644
--- a/joblib/test/test_memmapping.py
+++ b/joblib/test/test_memmapping.py
@@ -4,26 +4,27 @@ import sys
 import platform
 import gc
 import pickle
+import itertools
 from time import sleep
+import subprocess
 
 from joblib.test.common import with_numpy, np
 from joblib.test.common import setup_autokill
 from joblib.test.common import teardown_autokill
 from joblib.test.common import with_multiprocessing
 from joblib.test.common import with_dev_shm
-from joblib.testing import raises, parametrize, skipif
+from joblib.testing import raises, parametrize, skipif, xfail, param
 from joblib.backports import make_memmap
 from joblib.parallel import Parallel, delayed
 
 from joblib.pool import MemmappingPool
 from joblib.executor import _TestingMemmappingExecutor
 from joblib._memmapping_reducer import has_shareable_memory
-from joblib._memmapping_reducer import ArrayMemmapReducer
-from joblib._memmapping_reducer import reduce_memmap
+from joblib._memmapping_reducer import ArrayMemmapForwardReducer
 from joblib._memmapping_reducer import _strided_from_memmap
-from joblib._memmapping_reducer import _get_backing_memmap
 from joblib._memmapping_reducer import _get_temp_dir
 from joblib._memmapping_reducer import _WeakArrayKeyMap
+from joblib._memmapping_reducer import _get_backing_memmap
 import joblib._memmapping_reducer as jmr
 
 
@@ -35,6 +36,11 @@ def teardown_module():
     teardown_autokill(__name__)
 
 
+def check_memmap_and_send_back(array):
+    assert _get_backing_memmap(array) is not None
+    return array
+
+
 def check_array(args):
     """Dummy helper function to be executed in subprocesses
 
@@ -92,34 +98,30 @@ def test_memmap_based_array_reducing(tmpdir):
     d = c.T
 
     # Array reducer with auto dumping disabled
-    reducer = ArrayMemmapReducer(None, tmpdir.strpath, 'c')
+    reducer = ArrayMemmapForwardReducer(None, tmpdir.strpath, 'c', True)
 
-    def reconstruct_array(x):
+    def reconstruct_array_or_memmap(x):
         cons, args = reducer(x)
         return cons(*args)
 
-    def reconstruct_memmap(x):
-        cons, args = reduce_memmap(x)
-        return cons(*args)
-
     # Reconstruct original memmap
-    a_reconstructed = reconstruct_memmap(a)
+    a_reconstructed = reconstruct_array_or_memmap(a)
     assert has_shareable_memory(a_reconstructed)
     assert isinstance(a_reconstructed, np.memmap)
     assert_array_equal(a_reconstructed, a)
 
     # Reconstruct strided memmap view
-    b_reconstructed = reconstruct_memmap(b)
+    b_reconstructed = reconstruct_array_or_memmap(b)
     assert has_shareable_memory(b_reconstructed)
     assert_array_equal(b_reconstructed, b)
 
     # Reconstruct arrays views on memmap base
-    c_reconstructed = reconstruct_array(c)
+    c_reconstructed = reconstruct_array_or_memmap(c)
     assert not isinstance(c_reconstructed, np.memmap)
     assert has_shareable_memory(c_reconstructed)
     assert_array_equal(c_reconstructed, c)
 
-    d_reconstructed = reconstruct_array(d)
+    d_reconstructed = reconstruct_array_or_memmap(d)
     assert not isinstance(d_reconstructed, np.memmap)
     assert has_shareable_memory(d_reconstructed)
     assert_array_equal(d_reconstructed, d)
@@ -128,7 +130,7 @@ def test_memmap_based_array_reducing(tmpdir):
     # buffers
     a3 = a * 3
     assert not has_shareable_memory(a3)
-    a3_reconstructed = reconstruct_memmap(a3)
+    a3_reconstructed = reconstruct_array_or_memmap(a3)
     assert not has_shareable_memory(a3_reconstructed)
     assert not isinstance(a3_reconstructed, np.memmap)
     assert_array_equal(a3_reconstructed, a * 3)
@@ -137,12 +139,58 @@ def test_memmap_based_array_reducing(tmpdir):
     b3 = np.asarray(a3)
     assert not has_shareable_memory(b3)
 
-    b3_reconstructed = reconstruct_array(b3)
+    b3_reconstructed = reconstruct_array_or_memmap(b3)
     assert isinstance(b3_reconstructed, np.ndarray)
     assert not has_shareable_memory(b3_reconstructed)
     assert_array_equal(b3_reconstructed, b3)
 
 
+@skipif(sys.platform != "win32",
+        reason="PermissionError only easily triggerable on Windows")
+def test_resource_tracker_retries_when_permissionerror(tmpdir):
+    # Test resource_tracker retry mechanism when unlinking memmaps.  See more
+    # thorough information in the ``unlink_file`` documentation of joblib.
+    filename = tmpdir.join('test.mmap').strpath
+    cmd = """if 1:
+    import os
+    import numpy as np
+    import time
+    from joblib.externals.loky.backend import resource_tracker
+    resource_tracker.VERBOSE = 1
+
+    # Start the resource tracker
+    resource_tracker.ensure_running()
+    time.sleep(1)
+
+    # Create a file containing numpy data
+    memmap = np.memmap(r"{filename}", dtype=np.float64, shape=10, mode='w+')
+    memmap[:] = np.arange(10).astype(np.int8).data
+    memmap.flush()
+    assert os.path.exists(r"{filename}")
+    del memmap
+
+    # Create a np.memmap backed by this file
+    memmap = np.memmap(r"{filename}", dtype=np.float64, shape=10, mode='w+')
+    resource_tracker.register(r"{filename}", "file")
+
+    # Ask the resource_tracker to delete the file backing the np.memmap , this
+    # should raise PermissionError that the resource_tracker will log.
+    resource_tracker.maybe_unlink(r"{filename}", "file")
+
+    # Wait for the resource_tracker to process the maybe_unlink before cleaning
+    # up the memmap
+    time.sleep(2)
+    """.format(filename=filename)
+    p = subprocess.Popen([sys.executable, '-c', cmd], stderr=subprocess.PIPE,
+                         stdout=subprocess.PIPE)
+    p.wait()
+    out, err = p.communicate()
+    assert p.returncode == 0
+    assert out == b''
+    msg = 'tried to unlink {}, got PermissionError'.format(filename)
+    assert msg in err.decode()
+
+
 @with_numpy
 @with_multiprocessing
 def test_high_dimension_memmap_array_reducing(tmpdir):
@@ -161,29 +209,31 @@ def test_high_dimension_memmap_array_reducing(tmpdir):
     d = a[:, :, :, 0]
     e = a[1:3:4]
 
-    def reconstruct_memmap(x):
-        cons, args = reduce_memmap(x)
-        res = cons(*args)
-        return res
+    # Array reducer with auto dumping disabled
+    reducer = ArrayMemmapForwardReducer(None, tmpdir.strpath, 'c', True)
+
+    def reconstruct_array_or_memmap(x):
+        cons, args = reducer(x)
+        return cons(*args)
 
-    a_reconstructed = reconstruct_memmap(a)
+    a_reconstructed = reconstruct_array_or_memmap(a)
     assert has_shareable_memory(a_reconstructed)
     assert isinstance(a_reconstructed, np.memmap)
     assert_array_equal(a_reconstructed, a)
 
-    b_reconstructed = reconstruct_memmap(b)
+    b_reconstructed = reconstruct_array_or_memmap(b)
     assert has_shareable_memory(b_reconstructed)
     assert_array_equal(b_reconstructed, b)
 
-    c_reconstructed = reconstruct_memmap(c)
+    c_reconstructed = reconstruct_array_or_memmap(c)
     assert has_shareable_memory(c_reconstructed)
     assert_array_equal(c_reconstructed, c)
 
-    d_reconstructed = reconstruct_memmap(d)
+    d_reconstructed = reconstruct_array_or_memmap(d)
     assert has_shareable_memory(d_reconstructed)
     assert_array_equal(d_reconstructed, d)
 
-    e_reconstructed = reconstruct_memmap(e)
+    e_reconstructed = reconstruct_array_or_memmap(e)
     assert has_shareable_memory(e_reconstructed)
     assert_array_equal(e_reconstructed, e)
 
@@ -198,13 +248,15 @@ def test__strided_from_memmap(tmpdir):
     # filename, dtype, mode, offset, order, shape, strides, total_buffer_len
     memmap_obj = _strided_from_memmap(fname, dtype='uint8', mode='r',
                                       offset=offset, order='C', shape=size,
-                                      strides=None, total_buffer_len=None)
+                                      strides=None, total_buffer_len=None,
+                                      unlink_on_gc_collect=False)
     assert isinstance(memmap_obj, np.memmap)
     assert memmap_obj.offset == offset
-    memmap_backed_obj = _strided_from_memmap(fname, dtype='uint8', mode='r',
-                                             offset=offset, order='C',
-                                             shape=(size // 2,), strides=(2,),
-                                             total_buffer_len=size)
+    memmap_backed_obj = _strided_from_memmap(
+        fname, dtype='uint8', mode='r', offset=offset, order='C',
+        shape=(size // 2,), strides=(2,), total_buffer_len=size,
+        unlink_on_gc_collect=False
+    )
     assert _get_backing_memmap(memmap_backed_obj).offset == offset
 
 
@@ -301,6 +353,141 @@ def test_pool_with_memmap_array_view(factory, tmpdir):
         del p
 
 
+@with_numpy
+@parametrize("backend", ["multiprocessing", "loky"])
+def test_permission_error_windows_reference_cycle(backend):
+    # Non regression test for:
+    # https://github.com/joblib/joblib/issues/806
+    #
+    # The issue happens when trying to delete a memory mapped file that has
+    # not yet been closed by one of the worker processes.
+    cmd = """if 1:
+        import numpy as np
+        from joblib import Parallel, delayed
+
+
+        data = np.random.rand(int(2e6)).reshape((int(1e6), 2))
+
+        # Build a complex cyclic reference that is likely to delay garbage
+        # collection of the memmapped array in the worker processes.
+        first_list = current_list = [data]
+        for i in range(10):
+            current_list = [current_list]
+        first_list.append(current_list)
+
+        if __name__ == "__main__":
+            results = Parallel(n_jobs=2, backend="{b}")(
+                delayed(len)(current_list) for i in range(10))
+            assert results == [1] * 10
+    """.format(b=backend)
+    p = subprocess.Popen([sys.executable, '-c', cmd], stderr=subprocess.PIPE,
+                         stdout=subprocess.PIPE)
+    p.wait()
+    out, err = p.communicate()
+    assert p.returncode == 0, out.decode() + "\n\n" + err.decode()
+
+
+@with_numpy
+@parametrize("backend", ["multiprocessing", "loky"])
+def test_permission_error_windows_memmap_sent_to_parent(backend):
+    # Second non-regression test for:
+    # https://github.com/joblib/joblib/issues/806
+    # previously, child process would not convert temporary memmaps to numpy
+    # arrays when sending the data back to the parent process. This would lead
+    # to permission errors on windows when deleting joblib's temporary folder,
+    # as the memmaped files handles would still opened in the parent process.
+    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(int(2e6))
+
+        if __name__ == '__main__':
+            # warm-up call to launch the workers and start the resource_tracker
+            _ = Parallel(n_jobs=2, verbose=5, backend='{b}')(
+                delayed(id)(i) for i in range(20))
+
+            time.sleep(0.5)
+
+            slice_of_data = Parallel(n_jobs=2, verbose=5, 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"])
+def test_memmap_returned_as_regular_array(backend):
+    data = np.ones(int(1e3))
+    # Check that child processes send temporary memmaps back as numpy arrays.
+    [result] = Parallel(n_jobs=2, backend=backend, max_nbytes=100)(
+        delayed(check_memmap_and_send_back)(data) for _ in range(1))
+    assert _get_backing_memmap(result) is None
+
+
+@with_numpy
+@with_multiprocessing
+@parametrize("backend", ["multiprocessing", param("loky", marks=xfail)])
+def test_resource_tracker_silent_when_reference_cycles(backend):
+    # There is a variety of reasons that can make joblib with loky backend
+    # output noisy warnings when a reference cycle is preventing a memmap from
+    # being garbage collected. Especially, joblib's main process finalizer
+    # deletes the temporary folder if it was not done before, which can
+    # interact badly with the resource_tracker. We don't risk leaking any
+    # resources, but this will likely make joblib output a lot of low-level
+    # confusing messages. This test is marked as xfail for now: but a next PR
+    # should fix this behavior.
+    # Note that the script in ``cmd`` is the exact same script as in
+    # test_permission_error_windows_reference_cycle.
+    cmd = """if 1:
+        import numpy as np
+        from joblib import Parallel, delayed
+
+
+        data = np.random.rand(int(2e6)).reshape((int(1e6), 2))
+
+        # Build a complex cyclic reference that is likely to delay garbage
+        # collection of the memmapped array in the worker processes.
+        first_list = current_list = [data]
+        for i in range(10):
+            current_list = [current_list]
+        first_list.append(current_list)
+
+        if __name__ == "__main__":
+            results = Parallel(n_jobs=2, backend="{b}")(
+                delayed(len)(current_list) for i in range(10))
+            assert results == [1] * 10
+    """.format(b=backend)
+    p = subprocess.Popen([sys.executable, '-c', cmd], stderr=subprocess.PIPE,
+                         stdout=subprocess.PIPE)
+    p.wait()
+    out, err = p.communicate()
+    assert p.returncode == 0, out.decode()
+    assert b"resource_tracker" not in err, err.decode()
+
+
 @with_numpy
 @with_multiprocessing
 @parametrize("factory", [MemmappingPool, _TestingMemmappingExecutor],
@@ -346,10 +533,76 @@ def test_memmapping_pool_for_large_arrays(factory, tmpdir):
     finally:
         # check FS garbage upon pool termination
         p.terminate()
-        assert not os.path.exists(p._temp_folder)
+        for i in range(10):
+            sleep(.1)
+            if not os.path.exists(p._temp_folder):
+                break
+        else:  # pragma: no cover
+            raise AssertionError(
+                'temporary folder of {} was not deleted'.format(p)
+            )
         del p
 
 
+@with_numpy
+@with_multiprocessing
+@parametrize("backend", ["multiprocessing", "loky"])
+def test_child_raises_parent_exits_cleanly(backend):
+    # When a task executed by a child process raises an error, the parent
+    # process's backend is notified, and calls abort_everything.
+    # In loky, abort_everything itself calls shutdown(kill_workers=True) which
+    # sends SIGKILL to the worker, preventing it from running the finalizers
+    # supposed to signal the resource_tracker when the worker is done using
+    # objects relying on a shared resource (e.g np.memmaps). Because this
+    # behavior is prone to :
+    # - cause a resource leak
+    # - make the resource tracker emit noisy resource warnings
+    # we explicitly test that, when the said situation occurs:
+    # - no resources are actually leaked
+    # - the temporary resources are deleted as soon as possible (typically, at
+    #   the end of the failing Parallel call)
+    # - the resource_tracker does not emit any warnings.
+    cmd = """if 1:
+        import os
+
+        import numpy as np
+        from joblib import Parallel, delayed
+        from testutils import print_filename_and_raise
+
+        data = np.random.rand(1000)
+
+
+        def get_temp_folder(parallel_obj, backend):
+            if "{b}" == "loky":
+                return p._backend._temp_folder
+            else:
+                return p._backend._pool._temp_folder
+
+
+        if __name__ == "__main__":
+            try:
+                with Parallel(n_jobs=2, backend="{b}", max_nbytes=100) as p:
+                    temp_folder = get_temp_folder(p, "{b}")
+                    p(delayed(print_filename_and_raise)(data)
+                              for i in range(1))
+            except ValueError:
+                # the temporary folder should be deleted by the end of this
+                # call
+                assert not os.path.exists(temp_folder)
+    """.format(b=backend)
+    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()
+    out, err = out.decode(), err.decode()
+    filename = out.split('\n')[0]
+    assert p.returncode == 0, out
+    assert err == ''  # no resource_tracker warnings.
+    assert not os.path.exists(filename)
+
+
 @with_numpy
 @with_multiprocessing
 @parametrize("factory", [MemmappingPool, _TestingMemmappingExecutor],
@@ -419,8 +672,14 @@ def test_memmapping_on_large_enough_dev_shm(factory):
             # Cleanup open file descriptors
             p.terminate()
             del p
-        # The temp folder is cleaned up upon pool termination
-        assert not os.path.exists(pool_temp_folder)
+
+        for i in range(100):
+            # The temp folder is cleaned up upon pool termination
+            if not os.path.exists(pool_temp_folder):
+                break
+            sleep(.1)
+        else:  # pragma: no cover
+            raise AssertionError('temporary folder of pool was not deleted')
     finally:
         jmr.SYSTEM_SHARED_MEM_FS_MIN_SIZE = orig_size
 
@@ -521,9 +780,13 @@ def identity(arg):
 
 @with_numpy
 @with_multiprocessing
-@parametrize("factory", [MemmappingPool, _TestingMemmappingExecutor],
-             ids=["multiprocessing", "loky"])
-def test_pool_memmap_with_big_offset(factory, tmpdir):
+@parametrize(
+    "factory,retry_no",
+    list(itertools.product([MemmappingPool, _TestingMemmappingExecutor],
+                           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):
     # Test that numpy memmap offset is set correctly if greater than
     # mmap.ALLOCATIONGRANULARITY, see
     # https://github.com/joblib/joblib/issues/451 and
@@ -539,6 +802,7 @@ def test_pool_memmap_with_big_offset(factory, tmpdir):
     assert isinstance(result, np.memmap)
     assert result.offset == offset
     np.testing.assert_array_equal(obj, result)
+    p.terminate()
 
 
 def test_pool_get_temp_dir(tmpdir):
diff --git a/joblib/test/test_parallel.py b/joblib/test/test_parallel.py
index ffb0834..26ac40d 100644
--- a/joblib/test/test_parallel.py
+++ b/joblib/test/test_parallel.py
@@ -551,28 +551,6 @@ def test_nested_exception_dispatch(backend):
     assert type(excinfo.value) is ValueError
 
 
-def _reload_joblib():
-    # Retrieve the path of the parallel module in a robust way
-    joblib_path = Parallel.__module__.split(os.sep)
-    joblib_path = joblib_path[:1]
-    joblib_path.append('parallel.py')
-    joblib_path = '/'.join(joblib_path)
-    module = __import__(joblib_path)
-    # Reload the module. This should trigger a fail
-    reload(module)
-
-
-def test_multiple_spawning():
-    # Test that attempting to launch a new Python after spawned
-    # subprocesses will raise an error, to avoid infinite loops on
-    # systems that do not support fork
-    if not int(os.environ.get('JOBLIB_MULTIPROCESSING', 1)):
-        raise SkipTest()
-    with raises(ImportError):
-        Parallel(n_jobs=2, pre_dispatch='all')(
-            [delayed(_reload_joblib)() for i in range(10)])
-
-
 class FakeParallelBackend(SequentialBackend):
     """Pretends to run concurrently while running sequentially."""
 
@@ -1208,13 +1186,23 @@ def test_memmapping_leaks(backend, tmpdir):
 
     # Make sure that the shared memory is cleaned at the end when we exit
     # the context
-    assert not os.listdir(tmpdir)
+    for _ in range(100):
+        if not os.listdir(tmpdir):
+            break
+        sleep(.1)
+    else:
+        raise AssertionError('temporary directory of Parallel was not removed')
 
     # Make sure that the shared memory is cleaned at the end of a call
     p = Parallel(n_jobs=2, max_nbytes=1, backend=backend)
     p(delayed(check_memmap)(a) for a in [np.random.random(10)] * 2)
 
-    assert not os.listdir(tmpdir)
+    for _ in range(100):
+        if not os.listdir(tmpdir):
+            break
+        sleep(.1)
+    else:
+        raise AssertionError('temporary directory of Parallel was not removed')
 
 
 @parametrize('backend', [None, 'loky', 'threading'])
diff --git a/joblib/test/testutils.py b/joblib/test/testutils.py
new file mode 100644
index 0000000..20ec8c1
--- /dev/null
+++ b/joblib/test/testutils.py
@@ -0,0 +1,8 @@
+def return_slice_of_data(arr, start_idx, end_idx):
+    return arr[start_idx:end_idx]
+
+
+def print_filename_and_raise(arr):
+    from joblib._memmapping_reducer import _get_backing_memmap
+    print(_get_backing_memmap(arr).filename)
+    raise ValueError
diff --git a/joblib/testing.py b/joblib/testing.py
index 5b9c45f..28f7931 100644
--- a/joblib/testing.py
+++ b/joblib/testing.py
@@ -20,6 +20,8 @@ skipif = pytest.mark.skipif
 fixture = pytest.fixture
 parametrize = pytest.mark.parametrize
 timeout = pytest.mark.timeout
+xfail = pytest.mark.xfail
+param = pytest.param
 
 
 def warnings_to_stdout():