prompt: You will be provided with a partial code base and an issue statement explaining a problem to resolve.
  <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`.
  
  
  </issue>
  <code>
  [start of distributed/joblib.py]
  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 of distributed/joblib.py]
  </code>
  
  I need you to solve this issue by generating a single patch file that I can apply directly to this repository using git apply. Please respond with a single patch file in the following format.
  <patch>
  diff --git a/file.py b/file.py
  --- a/file.py
  +++ b/file.py
  @@ -1,27 +1,35 @@
   def euclidean(a, b):
  -    while b:
  -        a, b = b, a % b
  -    return a
  +    if b == 0:
  +        return a
  +    return euclidean(b, a % b)
   
   
   def bresenham(x0, y0, x1, y1):
       points = []
       dx = abs(x1 - x0)
       dy = abs(y1 - y0)
  -    sx = 1 if x0 < x1 else -1
  -    sy = 1 if y0 < y1 else -1
  -    err = dx - dy
  +    x, y = x0, y0
  +    sx = -1 if x0 > x1 else 1
  +    sy = -1 if y0 > y1 else 1
   
  -    while True:
  -        points.append((x0, y0))
  -        if x0 == x1 and y0 == y1:
  -            break
  -        e2 = 2 * err
  -        if e2 > -dy:
  -            err -= dy
  -            x0 += sx
  -        if e2 < dx:
  -            err += dx
  -            y0 += sy
  +    if dx > dy:
  +        err = dx / 2.0
  +        while x != x1:
  +            points.append((x, y))
  +            err -= dy
  +            if err < 0:
  +                y += sy
  +                err += dx
  +            x += sx
  +    else:
  +        err = dy / 2.0
  +        while y != y1:
  +            points.append((x, y))
  +            err -= dx
  +            if err < 0:
  +                x += sx
  +                err += dy
  +            y += sy
  +
  +    points.append((x, y))
       return points
  
  </patch>
golden_diff: 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)
  
verification_info: {"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"}]}