Replace loop.call_later and loop.add_callback with background tasks added to Server.

distributed/cli/dask_worker.py

@@ -14,7 +14,7 @@
 
 import click
 from tlz import valmap
-from tornado.ioloop import IOLoop, TimeoutError
+from tornado.ioloop import TimeoutError
 
 import dask
 from dask.system import CPU_COUNT
@@ -448,14 +448,11 @@ def del_pid_file():
     signal_fired = False
 
     async def run():
-        loop = IOLoop.current()
-
         nannies = [
             t(
                 scheduler,
                 scheduler_file=scheduler_file,
                 nthreads=nthreads,
-                loop=loop,
                 resources=resources,
                 security=sec,
                 contact_address=contact_address,

distributed/comm/inproc.py

@@ -88,6 +88,13 @@ class QueueEmpty(Exception):
     pass
 
 
+def _set_result_unless_cancelled(fut, result):
+    """Helper setting the result only if the future was not cancelled."""
+    if fut.cancelled():
+        return
+    fut.set_result(result)
+
+
 class Queue:
     """
     A single-reader, single-writer, non-threadsafe, peekable queue.
@@ -119,7 +126,7 @@ def put_nowait(self, value):
         if fut is not None:
             assert len(q) == 0
             self._read_future = None
-            fut.set_result(value)
+            _set_result_unless_cancelled(fut, value)
         else:
             q.append(value)