We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Distributed.joblib code fails to affect sklearn
The comments in [this stackoverflow answer](http://stackoverflow.com/questions/38601026/easy-way-to-use-parallel-options-of-scikit-learn-functions-on-hpc/38814491#38814491) raise concerns about the effectiveness of `distributed.joblib` to parallelize vanilla sklearn code.  It appears that sklearn ships with its own version of Joblib, which the plugin registration in `distributed.joblib` does not affect.

It would be good to test sklearn functionality and, if necessary, add plugin registration to `sklearn.externals.joblib` in the same way we do to normal `joblib`.


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `distributed/joblib.py`
Content:
```
1 from __future__ import print_function, division, absolute_import
2 
3 from joblib._parallel_backends import ParallelBackendBase, AutoBatchingMixin
4 from joblib.parallel import register_parallel_backend
5 from tornado import gen
6 
7 from .executor import Executor, _wait
8 
9 
10 class DistributedBackend(ParallelBackendBase, AutoBatchingMixin):
11     MIN_IDEAL_BATCH_DURATION = 0.2
12     MAX_IDEAL_BATCH_DURATION = 1.0
13 
14     def __init__(self, scheduler_host='127.0.0.1:8786', loop=None):
15         self.executor = Executor(scheduler_host, loop=loop)
16         self.futures = set()
17 
18     def configure(self, n_jobs=1, parallel=None, **backend_args):
19         return self.effective_n_jobs(n_jobs)
20 
21     def effective_n_jobs(self, n_jobs=1):
22         return sum(self.executor.ncores().values())
23 
24     def apply_async(self, func, *args, **kwargs):
25         callback = kwargs.pop('callback', None)
26         kwargs['pure'] = False
27         future = self.executor.submit(func, *args, **kwargs)
28         self.futures.add(future)
29 
30         @gen.coroutine
31         def callback_wrapper():
32             result = yield _wait([future])
33             self.futures.remove(future)
34             callback(result)  # gets called in separate thread
35 
36         self.executor.loop.add_callback(callback_wrapper)
37 
38         future.get = future.result  # monkey patch to achieve AsyncResult API
39         return future
40 
41     def abort_everything(self, ensure_ready=True):
42         # Tell the executor to cancel any task submitted via this instance
43         # as joblib.Parallel will never access those results.
44         self.executor.cancel(self.futures)
45         self.futures.clear()
46 
47 
48 register_parallel_backend('distributed', DistributedBackend)
49
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/distributed/joblib.py b/distributed/joblib.py
--- a/distributed/joblib.py
+++ b/distributed/joblib.py
@@ -1,10 +1,36 @@
 from __future__ import print_function, division, absolute_import
 
-from joblib._parallel_backends import ParallelBackendBase, AutoBatchingMixin
-from joblib.parallel import register_parallel_backend
+from distutils.version import LooseVersion
+
 from tornado import gen
 
 from .executor import Executor, _wait
+from .utils import ignoring
+
+
+# A user could have installed joblib, sklearn, both, or neither. Further, only
+# joblib >= 0.10.0 supports backends, so we also need to check for that. This
+# bit of logic is to ensure that we create and register the backend for all
+# viable installations of joblib.
+joblib = sk_joblib = None
+with ignoring(ImportError):
+    import joblib
+    if LooseVersion(joblib.__version__) < '0.10.0':
+        joblib = None
+with ignoring(ImportError):
+    import sklearn.externals.joblib as sk_joblib
+    if LooseVersion(sk_joblib.__version__) < '0.10.0':
+        sk_joblib = None
+
+if joblib:
+    from joblib._parallel_backends import (ParallelBackendBase,
+        AutoBatchingMixin)
+elif sk_joblib:
+    from sklearn.externals.joblib._parallel_backends import (
+        ParallelBackendBase, AutoBatchingMixin)
+else:
+    raise RuntimeError("Joblib backend requires either `joblib` >= '0.10.0' "
+                       " or `sklearn` > '0.17.1'. Please install or upgrade")
 
 
 class DistributedBackend(ParallelBackendBase, AutoBatchingMixin):
@@ -45,4 +71,8 @@
         self.futures.clear()
 
 
-register_parallel_backend('distributed', DistributedBackend)
+# Register the backend with any available versions of joblib
+if joblib:
+    joblib.register_parallel_backend('distributed', DistributedBackend)
+if sk_joblib:
+    sk_joblib.register_parallel_backend('distributed', DistributedBackend)

{"golden_diff": "diff --git a/distributed/joblib.py b/distributed/joblib.py\n--- a/distributed/joblib.py\n+++ b/distributed/joblib.py\n@@ -1,10 +1,36 @@\n from __future__ import print_function, division, absolute_import\n \n-from joblib._parallel_backends import ParallelBackendBase, AutoBatchingMixin\n-from joblib.parallel import register_parallel_backend\n+from distutils.version import LooseVersion\n+\n from tornado import gen\n \n from .executor import Executor, _wait\n+from .utils import ignoring\n+\n+\n+# A user could have installed joblib, sklearn, both, or neither. Further, only\n+# joblib >= 0.10.0 supports backends, so we also need to check for that. This\n+# bit of logic is to ensure that we create and register the backend for all\n+# viable installations of joblib.\n+joblib = sk_joblib = None\n+with ignoring(ImportError):\n+    import joblib\n+    if LooseVersion(joblib.__version__) < '0.10.0':\n+        joblib = None\n+with ignoring(ImportError):\n+    import sklearn.externals.joblib as sk_joblib\n+    if LooseVersion(sk_joblib.__version__) < '0.10.0':\n+        sk_joblib = None\n+\n+if joblib:\n+    from joblib._parallel_backends import (ParallelBackendBase,\n+        AutoBatchingMixin)\n+elif sk_joblib:\n+    from sklearn.externals.joblib._parallel_backends import (\n+        ParallelBackendBase, AutoBatchingMixin)\n+else:\n+    raise RuntimeError(\"Joblib backend requires either `joblib` >= '0.10.0' \"\n+                       \" or `sklearn` > '0.17.1'. Please install or upgrade\")\n \n \n class DistributedBackend(ParallelBackendBase, AutoBatchingMixin):\n@@ -45,4 +71,8 @@\n         self.futures.clear()\n \n \n-register_parallel_backend('distributed', DistributedBackend)\n+# Register the backend with any available versions of joblib\n+if joblib:\n+    joblib.register_parallel_backend('distributed', DistributedBackend)\n+if sk_joblib:\n+    sk_joblib.register_parallel_backend('distributed', DistributedBackend)\n", "issue": "Distributed.joblib code fails to affect sklearn\nThe comments in [this stackoverflow answer](http://stackoverflow.com/questions/38601026/easy-way-to-use-parallel-options-of-scikit-learn-functions-on-hpc/38814491#38814491) raise concerns about the effectiveness of `distributed.joblib` to parallelize vanilla sklearn code.  It appears that sklearn ships with its own version of Joblib, which the plugin registration in `distributed.joblib` does not affect.\n\nIt would be good to test sklearn functionality and, if necessary, add plugin registration to `sklearn.externals.joblib` in the same way we do to normal `joblib`.\n\n", "before_files": [{"content": "from __future__ import print_function, division, absolute_import\n\nfrom joblib._parallel_backends import ParallelBackendBase, AutoBatchingMixin\nfrom joblib.parallel import register_parallel_backend\nfrom tornado import gen\n\nfrom .executor import Executor, _wait\n\n\nclass DistributedBackend(ParallelBackendBase, AutoBatchingMixin):\n    MIN_IDEAL_BATCH_DURATION = 0.2\n    MAX_IDEAL_BATCH_DURATION = 1.0\n\n    def __init__(self, scheduler_host='127.0.0.1:8786', loop=None):\n        self.executor = Executor(scheduler_host, loop=loop)\n        self.futures = set()\n\n    def configure(self, n_jobs=1, parallel=None, **backend_args):\n        return self.effective_n_jobs(n_jobs)\n\n    def effective_n_jobs(self, n_jobs=1):\n        return sum(self.executor.ncores().values())\n\n    def apply_async(self, func, *args, **kwargs):\n        callback = kwargs.pop('callback', None)\n        kwargs['pure'] = False\n        future = self.executor.submit(func, *args, **kwargs)\n        self.futures.add(future)\n\n        @gen.coroutine\n        def callback_wrapper():\n            result = yield _wait([future])\n            self.futures.remove(future)\n            callback(result)  # gets called in separate thread\n\n        self.executor.loop.add_callback(callback_wrapper)\n\n        future.get = future.result  # monkey patch to achieve AsyncResult API\n        return future\n\n    def abort_everything(self, ensure_ready=True):\n        # Tell the executor to cancel any task submitted via this instance\n        # as joblib.Parallel will never access those results.\n        self.executor.cancel(self.futures)\n        self.futures.clear()\n\n\nregister_parallel_backend('distributed', DistributedBackend)\n", "path": "distributed/joblib.py"}], "after_files": [{"content": "from __future__ import print_function, division, absolute_import\n\nfrom distutils.version import LooseVersion\n\nfrom tornado import gen\n\nfrom .executor import Executor, _wait\nfrom .utils import ignoring\n\n\n# A user could have installed joblib, sklearn, both, or neither. Further, only\n# joblib >= 0.10.0 supports backends, so we also need to check for that. This\n# bit of logic is to ensure that we create and register the backend for all\n# viable installations of joblib.\njoblib = sk_joblib = None\nwith ignoring(ImportError):\n    import joblib\n    if LooseVersion(joblib.__version__) < '0.10.0':\n        joblib = None\nwith ignoring(ImportError):\n    import sklearn.externals.joblib as sk_joblib\n    if LooseVersion(sk_joblib.__version__) < '0.10.0':\n        sk_joblib = None\n\nif joblib:\n    from joblib._parallel_backends import (ParallelBackendBase,\n        AutoBatchingMixin)\nelif sk_joblib:\n    from sklearn.externals.joblib._parallel_backends import (\n        ParallelBackendBase, AutoBatchingMixin)\nelse:\n    raise RuntimeError(\"Joblib backend requires either `joblib` >= '0.10.0' \"\n                       \" or `sklearn` > '0.17.1'. Please install or upgrade\")\n\n\nclass DistributedBackend(ParallelBackendBase, AutoBatchingMixin):\n    MIN_IDEAL_BATCH_DURATION = 0.2\n    MAX_IDEAL_BATCH_DURATION = 1.0\n\n    def __init__(self, scheduler_host='127.0.0.1:8786', loop=None):\n        self.executor = Executor(scheduler_host, loop=loop)\n        self.futures = set()\n\n    def configure(self, n_jobs=1, parallel=None, **backend_args):\n        return self.effective_n_jobs(n_jobs)\n\n    def effective_n_jobs(self, n_jobs=1):\n        return sum(self.executor.ncores().values())\n\n    def apply_async(self, func, *args, **kwargs):\n        callback = kwargs.pop('callback', None)\n        kwargs['pure'] = False\n        future = self.executor.submit(func, *args, **kwargs)\n        self.futures.add(future)\n\n        @gen.coroutine\n        def callback_wrapper():\n            result = yield _wait([future])\n            self.futures.remove(future)\n            callback(result)  # gets called in separate thread\n\n        self.executor.loop.add_callback(callback_wrapper)\n\n        future.get = future.result  # monkey patch to achieve AsyncResult API\n        return future\n\n    def abort_everything(self, ensure_ready=True):\n        # Tell the executor to cancel any task submitted via this instance\n        # as joblib.Parallel will never access those results.\n        self.executor.cancel(self.futures)\n        self.futures.clear()\n\n\n# Register the backend with any available versions of joblib\nif joblib:\n    joblib.register_parallel_backend('distributed', DistributedBackend)\nif sk_joblib:\n    sk_joblib.register_parallel_backend('distributed', DistributedBackend)\n", "path": "distributed/joblib.py"}]}