Handle errant workers when another worker has data

distributed/scheduler.py

@@ -2038,11 +2038,12 @@ def handle_missing_data(self, key=None, errant_worker=None, **kwargs):
         ts = self.tasks.get(key)
         if ts is None or not ts.who_has:
             return
-        ws = self.workers[errant_worker]
-        if ws in ts.who_has:
-            ts.who_has.remove(ws)
-            ws.has_what.remove(ts)
-            ws.nbytes -= ts.get_nbytes()
+        if errant_worker in self.workers:
+            ws = self.workers[errant_worker]
+            if ws in ts.who_has:
+                ts.who_has.remove(ws)
+                ws.has_what.remove(ts)
+                ws.nbytes -= ts.get_nbytes()
         if not ts.who_has:
             if ts.run_spec:
                 self.transitions({key: 'released'})
@@ -3320,9 +3321,9 @@ def transition_processing_memory(self, key, nbytes=None, type=None,
                 return {key: 'released'}
 
             if ws is not ts.processing_on:  # someone else has this task
-                logger.warning("Unexpected worker completed task, likely due to"
-                               " work stealing.  Expected: %s, Got: %s, Key: %s",
-                               ts.processing_on, ws, key)
+                logger.info("Unexpected worker completed task, likely due to"
+                            " work stealing.  Expected: %s, Got: %s, Key: %s",
+                            ts.processing_on, ws, key)
                 return {}
 
             if startstops:

distributed/tests/test_scheduler.py

@@ -1200,3 +1200,20 @@ def test_retries(c, s, a, b):
     with pytest.raises(ZeroDivisionError) as exc_info:
         res = yield future
     exc_info.match("one")
+
+
+@pytest.mark.xfail(reason="second worker also errant for some reason")
+@gen_cluster(client=True, ncores=[('127.0.0.1', 1)] * 3, timeout=5)
+def test_mising_data_errant_worker(c, s, w1, w2, w3):
+    with set_config({'connect-timeout': '1s'}):
+        np = pytest.importorskip('numpy')
+
+        x = c.submit(np.random.random, 10000000, workers=w1.address)
+        yield wait(x)
+        yield c.replicate(x, workers=[w1.address, w2.address])
+
+        y = c.submit(len, x, workers=w3.address)
+        while not w3.tasks:
+            yield gen.sleep(0.001)
+        w1._close()
+        yield wait(y)

distributed/worker.py

@@ -534,7 +534,8 @@ def get_data(self, comm, keys=None, who=None):
         try:
             compressed = yield comm.write(msg)
         except EnvironmentError:
-            logger.exception('failed during get data', exc_info=True)
+            logger.exception('failed during get data with %s -> %s',
+                             self.address, who, exc_info=True)
             comm.abort()
             raise
         stop = time()