Repository: joblib/joblib

Issue:
readthedocs still documents joblib 0.11
Hello,

https://pythonhosted.org/joblib/index.html still documents joblib 0.11

This is disturbing, especially because there are some "big" changes on 0.12 (backend, etc.), and the 0.12 switch already happened (at least with anoconda).

Thanks.

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

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

```diff
@@            Coverage Diff             @@
##           master     #725      +/-   ##
==========================================
- Coverage   95.23%   95.09%   -0.14%     
==========================================
  Files          40       40              
  Lines        5769     5769              
==========================================
- Hits         5494     5486       -8     
- Misses        275      283       +8
```


| [Impacted Files](https://codecov.io/gh/joblib/joblib/pull/725?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [joblib/disk.py](https://codecov.io/gh/joblib/joblib/pull/725/diff?src=pr&el=tree#diff-am9ibGliL2Rpc2sucHk=) | `81.66% <0%> (-6.67%)` | :arrow_down: |
| [joblib/\_parallel\_backends.py](https://codecov.io/gh/joblib/joblib/pull/725/diff?src=pr&el=tree#diff-am9ibGliL19wYXJhbGxlbF9iYWNrZW5kcy5weQ==) | `96% <0%> (-0.8%)` | :arrow_down: |
| [joblib/test/test\_memory.py](https://codecov.io/gh/joblib/joblib/pull/725/diff?src=pr&el=tree#diff-am9ibGliL3Rlc3QvdGVzdF9tZW1vcnkucHk=) | `97.49% <0%> (-0.36%)` | :arrow_down: |

------

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


Patch:
diff --git a/.travis.yml b/.travis.yml
index 2c999f8..594615f 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -11,13 +11,13 @@ env:
     - PYTHON_VERSION="2.7" NUMPY_VERSION="1.8" COVERAGE="true"
     - PYTHON_VERSION="3.4" NUMPY_VERSION="1.10"
     # NUMPY_VERSION not set means numpy is not installed
-    - PYTHON_VERSION="3.4" COVERAGE="true"
+    - PYTHON_VERSION="3.6" COVERAGE="true" DISTRIBUTED_VERSION="1.22"
     - PYTHON_VERSION="3.5" NUMPY_VERSION="1.12"
     - PYTHON_VERSION="3.6" NUMPY_VERSION="1.14" COVERAGE="true" CYTHON="true"
     # multiprocesssing disabled via the JOBLIB_MULTIPROCESSING environment variable
     - PYTHON_VERSION="3.6" NUMPY_VERSION="1.14" JOBLIB_MULTIPROCESSING=0 COVERAGE="true"
     # Make sure we do not depend on lz4 to run joblib.
-    - PYTHON_VERSION="3.6" NUMPY_VERSION="1.12" COVERAGE="true" NO_LZ4="true" CYTHON="true"
+    - PYTHON_VERSION="3.6" NUMPY_VERSION="1.14" COVERAGE="true" NO_LZ4="true" CYTHON="true" DISTRIBUTED_VERSION="1.22"
     # flake8 linting on diff wrt common ancestor with upstream/master
     - SKIP_TESTS="true" FLAKE8_VERSION="3.5" PYTHON_VERSION="3.6"
 
diff --git a/continuous_integration/appveyor/requirements.txt b/continuous_integration/appveyor/requirements.txt
index 14384f7..b581b00 100644
--- a/continuous_integration/appveyor/requirements.txt
+++ b/continuous_integration/appveyor/requirements.txt
@@ -4,4 +4,5 @@ pytest
 pytest-cov
 pytest-timeout
 codecov
+distributed
 lz4
diff --git a/continuous_integration/travis/install.sh b/continuous_integration/travis/install.sh
index 98ec96b..ae24c71 100755
--- a/continuous_integration/travis/install.sh
+++ b/continuous_integration/travis/install.sh
@@ -20,7 +20,7 @@ print_conda_requirements() {
     #   - for scikit-learn, SCIKIT_LEARN_VERSION is used
     TO_INSTALL_ALWAYS="pip pytest"
     REQUIREMENTS="$TO_INSTALL_ALWAYS"
-    TO_INSTALL_MAYBE="python numpy flake8"
+    TO_INSTALL_MAYBE="python numpy distributed flake8"
     for PACKAGE in $TO_INSTALL_MAYBE; do
         # Capitalize package name and add _VERSION
         PACKAGE_VERSION_VARNAME="${PACKAGE^^}_VERSION"
diff --git a/examples/compressors_comparison.py b/examples/compressors_comparison.py
index 5f0f288..7a09a73 100644
--- a/examples/compressors_comparison.py
+++ b/examples/compressors_comparison.py
@@ -34,6 +34,8 @@ names = ("duration, protocol_type, service, flag, src_bytes, "
          "num_file_creations, ").split(', ')
 
 data = pd.read_csv(url, names=names)
+# Take the first 1e6 data points, for a faster example
+data = data.iloc[:int(1e6)]
 
 ###############################################################################
 # Dump and load the dataset without compression
@@ -190,13 +192,13 @@ file_sizes = (raw_file_size, lz4_file_size, zlib_file_size, lzma_file_size)
 ind = np.arange(N)
 width = 0.5
 
-plt.figure(1)
+plt.figure(1, figsize=(5, 4))
 p1 = plt.bar(ind, dump_durations, width)
 p2 = plt.bar(ind, load_durations, width, bottom=dump_durations)
 plt.ylabel('Time in seconds')
 plt.title('Dump and load durations')
 plt.xticks(ind, ('Raw', 'LZ4', 'Zlib', 'LZMA'))
-plt.yticks(np.arange(0, 30, 5))
+plt.yticks(np.arange(0, 5))
 plt.legend((p1[0], p2[0]), ('Dump duration', 'Load duration'))
 
 ###############################################################################
@@ -211,7 +213,7 @@ plt.legend((p1[0], p2[0]), ('Dump duration', 'Load duration'))
 # LZMA and Zlib, even if always slower for dumping data, are quite fast when
 # re-loading compressed data from disk.
 
-plt.figure(2)
+plt.figure(2, figsize=(5, 4))
 plt.bar(ind, file_sizes, width, log=True)
 plt.ylabel('File size in MB')
 plt.xticks(ind, ('Raw', 'LZ4', 'Zlib', 'LZMA'))
diff --git a/joblib/_dask.py b/joblib/_dask.py
new file mode 100644
index 0000000..92b9627
--- /dev/null
+++ b/joblib/_dask.py
@@ -0,0 +1,259 @@
+from __future__ import print_function, division, absolute_import
+
+import contextlib
+
+from uuid import uuid4
+import weakref
+
+from .parallel import AutoBatchingMixin, ParallelBackendBase, BatchedCalls
+from .parallel import parallel_backend
+
+try:
+    import distributed
+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 tornado import gen
+
+
+def is_weakrefable(obj):
+    try:
+        weakref.ref(obj)
+        return True
+    except TypeError:
+        return False
+
+
+class _WeakKeyDictionary:
+    """A variant of weakref.WeakKeyDictionary for unhashable objects.
+
+    This datastructure is used to store futures for broadcasted data objects
+    such as large numpy arrays or pandas dataframes that are not hashable and
+    therefore cannot be used as keys of traditional python dicts.
+
+    Futhermore using a dict with id(array) as key is not safe because the
+    Python is likely to reuse id of recently collected arrays.
+    """
+
+    def __init__(self):
+        self._data = {}
+
+    def __getitem__(self, obj):
+        ref, val = self._data[id(obj)]
+        if ref() is not obj:
+            # In case of a race condition with on_destroy.
+            raise KeyError(obj)
+        return val
+
+    def __setitem__(self, obj, value):
+        key = id(obj)
+        try:
+            ref, _ = self._data[key]
+            if ref() is not obj:
+                # In case of race condition with on_destroy.
+                raise KeyError(obj)
+        except KeyError:
+            # Insert the new entry in the mapping along with a weakref
+            # callback to automatically delete the entry from the mapping
+            # as soon as the object used as key is garbage collected.
+            def on_destroy(_):
+                del self._data[key]
+            ref = weakref.ref(obj, on_destroy)
+        self._data[key] = ref, value
+
+    def __len__(self):
+        return len(self._data)
+
+    def clear(self):
+        self._data.clear()
+
+
+def _funcname(x):
+    try:
+        if isinstance(x, BatchedCalls):
+            x = x.items[0][0]
+    except Exception:
+        pass
+    return funcname(x)
+
+
+class Batch(object):
+    def __init__(self, tasks):
+        self.tasks = tasks
+
+    def __call__(self, *data):
+        results = []
+        with parallel_backend('dask'):
+            for func, args, kwargs in self.tasks:
+                args = [a(data) if isinstance(a, itemgetter) else a
+                        for a in args]
+                kwargs = {k: v(data) if isinstance(v, itemgetter) else v
+                          for (k, v) in kwargs.items()}
+                results.append(func(*args, **kwargs))
+        return results
+
+    def __reduce__(self):
+        return Batch, (self.tasks,)
+
+
+class DaskDistributedBackend(ParallelBackendBase, AutoBatchingMixin):
+    MIN_IDEAL_BATCH_DURATION = 0.2
+    MAX_IDEAL_BATCH_DURATION = 1.0
+
+    def __init__(self, scheduler_host=None, scatter=None,
+                 client=None, loop=None, **submit_kwargs):
+        if client is None:
+            if scheduler_host:
+                client = Client(scheduler_host, loop=loop,
+                                set_as_default=False)
+            else:
+                try:
+                    client = get_client()
+                except ValueError:
+                    msg = ("To use Joblib with Dask first create a Dask Client"
+                           "\n\n"
+                           "    from dask.distributed import Client\n"
+                           "    client = Client()\n"
+                           "or\n"
+                           "    client = Client('scheduler-address:8786')")
+                    raise ValueError(msg)
+
+        self.client = client
+
+        if scatter is not None and not isinstance(scatter, (list, tuple)):
+            raise TypeError("scatter must be a list/tuple, got "
+                            "`%s`" % type(scatter).__name__)
+
+        if scatter is not None and len(scatter) > 0:
+            # Keep a reference to the scattered data to keep the ids the same
+            self._scatter = list(scatter)
+            scattered = self.client.scatter(scatter, broadcast=True)
+            self.data_futures = {id(x): f for x, f in zip(scatter, scattered)}
+        else:
+            self._scatter = []
+            self.data_futures = {}
+        self.task_futures = set()
+        self.submit_kwargs = submit_kwargs
+
+    def __reduce__(self):
+        return (DaskDistributedBackend, ())
+
+    def get_nested_backend(self):
+        return DaskDistributedBackend(client=self.client)
+
+    def configure(self, n_jobs=1, parallel=None, **backend_args):
+        return self.effective_n_jobs(n_jobs)
+
+    def start_call(self):
+        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.call_data_futures.clear()
+
+    def effective_n_jobs(self, n_jobs):
+        return sum(self.client.ncores().values())
+
+    def _to_func_args(self, func):
+        collected_futures = []
+        itemgetters = dict()
+
+        # Futures that are dynamically generated during a single call to
+        # Parallel.__call__.
+        call_data_futures = getattr(self, 'call_data_futures', None)
+
+        def maybe_to_futures(args):
+            for arg in args:
+                arg_id = id(arg)
+                if arg_id in itemgetters:
+                    yield itemgetters[arg_id]
+                    continue
+
+                f = self.data_futures.get(arg_id, None)
+                if f is None and call_data_futures is not None:
+                    try:
+                        f = call_data_futures[arg]
+                    except KeyError:
+                        if is_weakrefable(arg) and sizeof(arg) > 1e3:
+                            # Automatically scatter large objects to some of
+                            # the workers to avoid duplicated data transfers.
+                            # Rely on automated inter-worker data stealing if
+                            # more workers need to reuse this data
+                            # concurrently.
+                            [f] = self.client.scatter([arg])
+                            call_data_futures[arg] = f
+
+                if f is not None:
+                    getter = itemgetter(len(collected_futures))
+                    collected_futures.append(f)
+                    itemgetters[arg_id] = getter
+                    arg = getter
+                yield arg
+
+        tasks = []
+        for f, args, kwargs in func.items:
+            args = list(maybe_to_futures(args))
+            kwargs = dict(zip(kwargs.keys(),
+                              maybe_to_futures(kwargs.values())))
+            tasks.append((f, args, kwargs))
+
+        if not collected_futures:
+            return func, ()
+        return (Batch(tasks), collected_futures)
+
+    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)
+
+        @gen.coroutine
+        def callback_wrapper():
+            result = yield _wait([future])
+            self.task_futures.remove(future)
+            if callback is not None:
+                callback(result)  # gets called in separate thread
+
+        self.client.loop.add_callback(callback_wrapper)
+
+        ref = weakref.ref(future)  # avoid reference cycle
+
+        def get():
+            return ref().result()
+
+        future.get = get  # monkey patch to achieve AsyncResult API
+        return 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()
+
+    @contextlib.contextmanager
+    def retrieval_context(self):
+        """Override ParallelBackendBase.retrieval_context to avoid deadlocks.
+
+        This removes thread from the worker's thread pool (using 'secede').
+        Seceding avoids deadlock in nested parallelism settings.
+        """
+        # See 'joblib.Parallel.__call__' and 'joblib.Parallel.retrieve' for how
+        # this is used.
+        if hasattr(thread_state, 'execution_state'):
+            # we are in a worker. Secede to avoid deadlock.
+            secede()
+
+        yield
+
+        if hasattr(thread_state, 'execution_state'):
+            rejoin()
diff --git a/joblib/parallel.py b/joblib/parallel.py
index 299f79f..266b2be 100644
--- a/joblib/parallel.py
+++ b/joblib/parallel.py
@@ -59,7 +59,8 @@ VALID_BACKEND_CONSTRAINTS = ('sharedmem', None)
 def _register_dask():
     """ Register Dask Backend if called with parallel_backend("dask") """
     try:
-        import distributed.joblib  # noqa: #F401
+        from ._dask import DaskDistributedBackend
+        register_parallel_backend('dask', DaskDistributedBackend)
     except ImportError:
         msg = ("To use the dask.distributed backend you must install both "
                "the `dask` and distributed modules.\n\n"
diff --git a/setup.cfg b/setup.cfg
index 40a2be3..3551800 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -13,8 +13,8 @@ universal=1
 addopts =
     --doctest-glob="doc/*.rst"
     --doctest-modules
-     -p no:warnings
-     --ignore joblib/externals
+    -p no:warnings
+    --ignore joblib/externals
 testpaths = joblib
 
 [flake8]


Test patch:
diff --git a/conftest.py b/conftest.py
index 3bf7460..bec21db 100644
--- a/conftest.py
+++ b/conftest.py
@@ -5,6 +5,10 @@ from _pytest.doctest import DoctestItem
 
 import logging
 from joblib.parallel import mp
+try:
+    import lz4
+except ImportError:
+    lz4 = None
 
 
 def pytest_collection_modifyitems(config, items):
@@ -30,6 +34,11 @@ def pytest_collection_modifyitems(config, items):
             if isinstance(item, DoctestItem):
                 item.add_marker(skip_marker)
 
+    if lz4 is None:
+        for item in items:
+            if item.name == 'persistence.rst':
+                item.add_marker(pytest.mark.skip(reason='lz4 is missing'))
+
 
 def pytest_configure(config):
     """Setup multiprocessing logging for the tests"""
diff --git a/joblib/test/test_dask.py b/joblib/test/test_dask.py
new file mode 100644
index 0000000..13aa8ee
--- /dev/null
+++ b/joblib/test/test_dask.py
@@ -0,0 +1,257 @@
+from __future__ import print_function, division, absolute_import
+import os
+
+import pytest
+from random import random
+from time import sleep
+
+from .. import Parallel, delayed, parallel_backend
+from ..parallel import ThreadingBackend
+from .._dask import DaskDistributedBackend
+
+distributed = pytest.importorskip('distributed')
+from distributed import Client
+from distributed.metrics import time
+from distributed.utils_test import cluster, inc
+from distributed.utils_test import loop # noqa F401
+
+
+def noop(*args, **kwargs):
+    pass
+
+
+def slow_raise_value_error(condition, duration=0.05):
+    sleep(duration)
+    if condition:
+        raise ValueError("condition evaluated to True")
+
+
+def test_simple(loop):  # noqa: F811
+    with cluster() as (s, [a, b]):
+        with Client(s['address'], loop=loop) as client:  # noqa: F841
+            with parallel_backend('dask') as (ba, _):
+                seq = Parallel()(delayed(inc)(i) for i in range(10))
+                assert seq == [inc(i) for i in range(10)]
+
+                with pytest.raises(ValueError):
+                    Parallel()(delayed(slow_raise_value_error)(i == 3)
+                               for i in range(10))
+
+                seq = Parallel()(delayed(inc)(i) for i in range(10))
+                assert seq == [inc(i) for i in range(10)]
+
+
+def random2():
+    return random()
+
+
+def test_dont_assume_function_purity(loop):  # noqa: F811
+    with cluster() as (s, [a, b]):
+        with Client(s['address'], loop=loop) as client:  # noqa: F841
+            with parallel_backend('dask') as (ba, _):
+                x, y = Parallel()(delayed(random2)() for i in range(2))
+                assert x != y
+
+
+def test_dask_funcname(loop):  # noqa: F811
+    with cluster() as (s, [a, b]):
+        with Client(s['address'], loop=loop) as client:
+            with parallel_backend('dask') as (ba, _):
+                x, y = Parallel()(delayed(inc)(i) for i in range(2))
+
+            def f(dask_scheduler):
+                return list(dask_scheduler.transition_log)
+            log = client.run_on_scheduler(f)
+            assert all(tup[0].startswith('inc-batch') for tup in log)
+
+
+def add5(a, b, c, d=0, e=0):
+    return a + b + c + d + e
+
+
+class CountSerialized(object):
+    def __init__(self, x):
+        self.x = x
+        self.count = 0
+
+    def __add__(self, other):
+        return self.x + getattr(other, 'x', other)
+
+    __radd__ = __add__
+
+    def __reduce__(self):
+        self.count += 1
+        return (CountSerialized, (self.x,))
+
+
+def test_manual_scatter(loop):  # noqa: F811
+    x = CountSerialized(1)
+    y = CountSerialized(2)
+    z = CountSerialized(3)
+
+    with cluster() as (s, [a, b]):
+        with Client(s['address'], loop=loop) as client:  # noqa: F841
+            with parallel_backend('dask', scatter=[x, y]) as (ba, _):
+                f = delayed(add5)
+                tasks = [f(x, y, z, d=4, e=5),
+                         f(x, z, y, d=5, e=4),
+                         f(y, x, z, d=x, e=5),
+                         f(z, z, x, d=z, e=y)]
+                expected = [func(*args, **kwargs)
+                            for func, args, kwargs in tasks]
+                results = Parallel()(tasks)
+
+            # Scatter must take a list/tuple
+            with pytest.raises(TypeError):
+                with parallel_backend('dask', loop=loop, scatter=1):
+                    pass
+
+    assert results == expected
+
+    # Scattered variables only serialized once
+    assert x.count == 1
+    assert y.count == 1
+    assert z.count == 4
+
+
+def test_auto_scatter(loop):  # noqa: F811
+    np = pytest.importorskip('numpy')
+    data = np.ones(int(1e7), dtype=np.uint8)
+
+    def count_events(event_name, client):
+        worker_events = client.run(lambda dask_worker: dask_worker.log)
+        event_counts = {}
+        for w, events in worker_events.items():
+            event_counts[w] = len([event for event in list(events)
+                                   if event[1] == event_name])
+        return event_counts
+
+    with cluster() as (s, [a, b]):
+        with Client(s['address'], loop=loop) as client:
+            with parallel_backend('dask') as (ba, _):
+                # Passing the same data as arg and kwarg triggers a single
+                # scatter operation whose result is reused.
+                Parallel()(delayed(noop)(data, data, i, opt=data)
+                           for i in range(5))
+            # By default large array are automatically scattered with
+            # 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']] == 1
+
+    with cluster() as (s, [a, b]):
+        with Client(s['address'], loop=loop) as client:
+            with parallel_backend('dask') as (ba, _):
+                Parallel()(delayed(noop)(data[:3], i) for i in range(5))
+            # Small arrays are passed within the task definition without going
+            # through a scatter operation.
+            counts = count_events('receive-from-scatter', client)
+            assert counts[a['address']] == 0
+            assert counts[b['address']] == 0
+
+
+def test_nested_backend_context_manager(loop):  # noqa: F811
+    def get_nested_pids():
+        pids = set(Parallel(n_jobs=2)(delayed(os.getpid)() for _ in range(2)))
+        pids |= set(Parallel(n_jobs=2)(delayed(os.getpid)() for _ in range(2)))
+        return pids
+
+    with cluster() as (s, [a, b]):
+        with Client(s['address'], loop=loop) as client:
+            with parallel_backend('dask') as (ba, _):
+                pid_groups = Parallel(n_jobs=2)(
+                    delayed(get_nested_pids, check_pickle=False)()
+                    for _ in range(10)
+                )
+                for pid_group in pid_groups:
+                    assert len(set(pid_group)) <= 2
+
+        # No deadlocks
+        with Client(s['address'], loop=loop) as client:  # noqa: F841
+            with parallel_backend('dask') as (ba, _):
+                pid_groups = Parallel(n_jobs=2)(
+                    delayed(get_nested_pids, check_pickle=False)()
+                    for _ in range(10)
+                )
+                for pid_group in pid_groups:
+                    assert len(set(pid_group)) <= 2
+
+
+def test_errors(loop):  # noqa: F811
+    with pytest.raises(ValueError) as info:
+        with parallel_backend('dask'):
+            pass
+
+    assert "create a dask client" in str(info.value).lower()
+
+
+def test_correct_nested_backend(loop):  # noqa: F811
+    with cluster() as (s, [a, b]):
+        with Client(s['address'], loop=loop) as client:  # noqa: F841
+            # No requirement, should be us
+            with parallel_backend('dask') as (ba, _):
+                result = Parallel(n_jobs=2)(
+                    delayed(outer)(nested_require=None) for _ in range(1))
+                assert isinstance(result[0][0][0], DaskDistributedBackend)
+
+            # Require threads, should be threading
+            with parallel_backend('dask') as (ba, _):
+                result = Parallel(n_jobs=2)(
+                    delayed(outer)(nested_require='sharedmem')
+                    for _ in range(1))
+                assert isinstance(result[0][0][0], ThreadingBackend)
+
+
+def outer(nested_require):
+    return Parallel(n_jobs=2, prefer='threads')(
+        delayed(middle)(nested_require) for _ in range(1)
+    )
+
+
+def middle(require):
+    return Parallel(n_jobs=2, require=require)(
+        delayed(inner)() for _ in range(1)
+    )
+
+
+def inner():
+    return Parallel()._backend
+
+
+def test_secede_with_no_processes(loop):  # noqa: F811
+    # https://github.com/dask/distributed/issues/1775
+    with Client(loop=loop, processes=False, set_as_default=True):
+        with parallel_backend('dask'):
+            Parallel(n_jobs=4)(delayed(id)(i) for i in range(2))
+
+
+def _worker_address(_):
+    from distributed import get_worker
+    return get_worker().address
+
+
+def test_dask_backend_keywords(loop):  # noqa: F811
+    with cluster() as (s, [a, b]):
+        with Client(s['address'], loop=loop) as client:  # noqa: F841
+            with parallel_backend('dask', workers=a['address']) as (ba, _):
+                seq = Parallel()(
+                    delayed(_worker_address)(i) for i in range(10))
+                assert seq == [a['address']] * 10
+
+            with parallel_backend('dask', workers=b['address']) as (ba, _):
+                seq = Parallel()(
+                    delayed(_worker_address)(i) for i in range(10))
+                assert seq == [b['address']] * 10
+
+
+def test_cleanup(loop):  # noqa: F811
+    with Client(processes=False, loop=loop) as client:
+        with parallel_backend('dask'):
+            Parallel()(delayed(inc)(i) for i in range(10))
+
+        start = time()
+        while client.cluster.scheduler.tasks:
+            sleep(0.01)
+            assert time() < start + 5
+
+        assert not client.futures
diff --git a/joblib/test/test_parallel.py b/joblib/test/test_parallel.py
index 789a1f4..ca5321a 100644
--- a/joblib/test/test_parallel.py
+++ b/joblib/test/test_parallel.py
@@ -16,6 +16,7 @@ from math import sqrt
 from time import sleep
 from pickle import PicklingError
 from multiprocessing import TimeoutError
+import pytest
 
 import joblib
 from joblib import dump, load
@@ -1093,6 +1094,8 @@ def test_lambda_expression():
 
 
 def test_delayed_check_pickle_deprecated():
+    if sys.version_info < (3, 4):
+        pytest.skip("Warning check unstable under Python 2, life is too short")
 
     class UnpicklableCallable(object):
 
@@ -1288,21 +1291,21 @@ def test_external_backends():
         assert isinstance(Parallel()._backend, ThreadingBackend)
 
 
-def _recursive_backend_info(limit=3):
+def _recursive_backend_info(limit=3, **kwargs):
     """Perform nested parallel calls and introspect the backend on the way"""
 
     with Parallel() as p:
         this_level = [(type(p._backend).__name__, p._backend.nesting_level)]
         if limit == 0:
             return this_level
-        results = p(delayed(_recursive_backend_info)(limit=limit - 1)
+        results = p(delayed(_recursive_backend_info)(limit=limit - 1, **kwargs)
                     for i in range(1))
         return this_level + results[0]
 
 
 @with_multiprocessing
 @parametrize('backend', ['loky', 'threading'])
-def test_nested_parallel_limit(backend):
+def test_nested_parallelism_limit(backend):
     with parallel_backend(backend, n_jobs=2):
         backend_types_and_levels = _recursive_backend_info()
 
@@ -1316,6 +1319,28 @@ def test_nested_parallel_limit(backend):
     assert backend_types_and_levels == expected_types_and_levels
 
 
+@with_numpy
+def test_nested_parallelism_with_dask():
+    distributed = pytest.importorskip('distributed')
+    client = distributed.Client()  # noqa
+
+    # 10 MB of data as argument to trigger implicit scattering
+    data = np.ones(int(1e7), dtype=np.uint8)
+    for i in range(2):
+        with parallel_backend('dask'):
+            backend_types_and_levels = _recursive_backend_info(data=data)
+        assert len(backend_types_and_levels) == 4
+        assert all(name == 'DaskDistributedBackend'
+                   for name, _ in backend_types_and_levels)
+
+    # No argument
+    with parallel_backend('dask'):
+        backend_types_and_levels = _recursive_backend_info()
+    assert len(backend_types_and_levels) == 4
+    assert all(name == 'DaskDistributedBackend'
+               for name, _ in backend_types_and_levels)
+
+
 def _recursive_parallel(nesting_limit=None):
     """A horrible function that does recursive parallel calls"""
     return Parallel()(delayed(_recursive_parallel)() for i in range(2))