Add SpecificationCluster

continuous_integration/setup_conda_environment.cmd

@@ -50,7 +50,7 @@ call activate %CONDA_ENV%
 %PIP_INSTALL% git+https://github.com/joblib/joblib.git --upgrade
 %PIP_INSTALL% git+https://github.com/dask/zict --upgrade
 
-%PIP_INSTALL% pytest-repeat pytest-timeout pytest-faulthandler sortedcollections
+%PIP_INSTALL% pytest-repeat pytest-timeout pytest-faulthandler sortedcollections pytest-asyncio
 
 @rem Display final environment (for reproducing)
 %CONDA% list

continuous_integration/travis/install.sh

@@ -59,7 +59,7 @@ conda install -q \
 conda install -c defaults -c conda-forge libunwind
 conda install --no-deps -c defaults -c numba -c conda-forge stacktrace
 
-pip install -q pytest-repeat pytest-faulthandler
+pip install -q pytest-repeat pytest-faulthandler pytest-asyncio
 
 pip install -q git+https://github.com/dask/dask.git --upgrade --no-deps
 pip install -q git+https://github.com/joblib/joblib.git --upgrade --no-deps

distributed/__init__.py

@@ -4,7 +4,7 @@
 from dask.config import config
 from .actor import Actor, ActorFuture
 from .core import connect, rpc
-from .deploy import LocalCluster, Adaptive
+from .deploy import LocalCluster, Adaptive, SpecCluster
 from .diagnostics import progress
 from .client import (
     Client,

distributed/cli/dask_worker.py

@@ -114,7 +114,7 @@
 @click.option(
     "--name",
     type=str,
-    default="",
+    default=None,
     help="A unique name for this worker like 'worker-1'. "
     "If used with --nprocs then the process number "
     "will be appended like name-0, name-1, name-2, ...",

distributed/client.py

@@ -2,8 +2,8 @@
 
 import atexit
 from collections import defaultdict
-from concurrent.futures import ThreadPoolExecutor
-from concurrent.futures._base import DoneAndNotDoneFutures, CancelledError
+from concurrent.futures import ThreadPoolExecutor, CancelledError
+from concurrent.futures._base import DoneAndNotDoneFutures
 from contextlib import contextmanager
 import copy
 from datetime import timedelta
@@ -44,6 +44,8 @@
 from tornado.ioloop import IOLoop
 from tornado.queues import Queue
 
+from asyncio import iscoroutine
+
 from .batched import BatchedSend
 from .utils_comm import (
     WrappedKey,
@@ -1301,7 +1303,13 @@ def close(self, timeout=no_default):
 
         if self._start_arg is None:
             with ignoring(AttributeError):
-                self.cluster.close()
+                f = self.cluster.close()
+                if iscoroutine(f):
+
+                    async def _():
+                        await f
+
+                    self.sync(_)
 
         sync(self.loop, self._close, fast=True)
 
@@ -1636,10 +1644,11 @@ def wait(k):
                             st = self.futures[key]
                             exception = st.exception
                             traceback = st.traceback
-                        except (AttributeError, KeyError):
-                            six.reraise(CancelledError, CancelledError(key), None)
+                        except (KeyError, AttributeError):
+                            exc = CancelledError(key)
                         else:
                             six.reraise(type(exception), exception, traceback)
+                        raise exc
                     if errors == "skip":
                         bad_keys.add(key)
                         bad_data[key] = None
@@ -4059,7 +4068,10 @@ def _track_future(self, future):
         except CancelledError:
             pass
         if self.with_results:
-            result = yield future._result(raiseit=False)
+            try:
+                result = yield future._result(raiseit=False)
+            except CancelledError as exc:
+                result = exc
         with self.lock:
             self.futures[future] -= 1
             if not self.futures[future]:

distributed/comm/tcp.py

@@ -431,6 +431,7 @@ def start(self):
                 break
         else:
             raise exc
+        self.get_host_port()  # trigger assignment to self.bound_address
 
     def stop(self):
         tcp_server, self.tcp_server = self.tcp_server, None

distributed/core.py

@@ -489,14 +489,16 @@ def handle_stream(self, comm, extra=None, every_cycle=[]):
 
     @gen.coroutine
     def close(self):
-        self.listener.stop()
+        for pc in self.periodic_callbacks.values():
+            pc.stop()
+        if self.listener:
+            self.listener.stop()
         for i in range(20):  # let comms close naturally for a second
             if not self._comms:
                 break
             else:
                 yield gen.sleep(0.05)
-        for comm in self._comms:
-            comm.close()
+        yield [comm.close() for comm in self._comms]
         for cb in self._ongoing_coroutines:
             cb.cancel()
         for i in range(10):

distributed/deploy/__init__.py

@@ -4,6 +4,7 @@
 
 from .cluster import Cluster
 from .local import LocalCluster
+from .spec import SpecCluster
 from .adaptive import Adaptive
 
 with ignoring(ImportError):

distributed/deploy/adaptive.py

@@ -272,7 +272,7 @@ def _retire_workers(self, workers=None):
 
             logger.info("Retiring workers %s", workers)
             f = self.cluster.scale_down(workers)
-            if gen.is_future(f):
+            if hasattr(f, "__await__"):
                 yield f
 
             raise gen.Return(workers)
@@ -354,7 +354,7 @@ def _adapt(self):
             if status == "up":
                 f = self.cluster.scale_up(**recommendations)
                 self.log.append((time(), "up", recommendations))
-                if gen.is_future(f):
+                if hasattr(f, "__await__"):
                     yield f
 
             elif status == "down":

distributed/deploy/cluster.py

@@ -1,12 +1,23 @@
+from datetime import timedelta
 import logging
 import os
 from weakref import ref
 
 import dask
+from tornado import gen
 
 from .adaptive import Adaptive
 
-from ..utils import format_bytes, PeriodicCallback, log_errors, ignoring
+from ..compatibility import get_thread_identity
+from ..utils import (
+    format_bytes,
+    PeriodicCallback,
+    log_errors,
+    ignoring,
+    sync,
+    thread_state,
+)
+
 
 logger = logging.getLogger(__name__)
 
@@ -215,3 +226,22 @@ def update():
 
     def _ipython_display_(self, **kwargs):
         return self._widget()._ipython_display_(**kwargs)
+
+    @property
+    def asynchronous(self):
+        return (
+            self._asynchronous
+            or getattr(thread_state, "asynchronous", False)
+            or hasattr(self.loop, "_thread_identity")
+            and self.loop._thread_identity == get_thread_identity()
+        )
+
+    def sync(self, func, *args, **kwargs):
+        if kwargs.pop("asynchronous", None) or self.asynchronous:
+            callback_timeout = kwargs.pop("callback_timeout", None)
+            future = func(*args, **kwargs)
+            if callback_timeout is not None:
+                future = gen.with_timeout(timedelta(seconds=callback_timeout), future)
+            return future
+        else:
+            return sync(self.loop, func, *args, **kwargs)