Avoid multiple data connections to the same worker

distributed/bokeh/worker.py

@@ -53,7 +53,7 @@ def update(self):
                  'Executing': ['%d / %d' % (len(w.executing), w.ncores)],
                  'Ready': [len(w.ready)],
                  'Waiting': [len(w.waiting_for_data)],
-                 'Connections': [len(w.connections)],
+                 'Connections': [len(w.in_flight_workers)],
                  'Serving': [len(w._listen_streams)]}
             self.source.data.update(d)
 
@@ -169,7 +169,7 @@ def update(self):
         with log_errors():
             self.source.stream({'x': [time() * 1000],
                                 'out': [len(self.worker._listen_streams)],
-                                'in': [len(self.worker.connections)]},
+                                'in': [len(self.worker.in_flight_workers)]},
                                 10000)
 
 

distributed/scheduler.py

@@ -1100,11 +1100,6 @@ def send_task_to_worker(self, worker, key):
                    'key': key,
                    'priority': self.priority[key],
                    'duration': self.task_duration.get(key_split(key), 0.5)}
-            if key not in self.loose_restrictions:
-                if key in self.host_restrictions:
-                    msg['host_restrictions'] = list(self.host_restrictions[key])
-                if key in self.worker_restrictions:
-                    msg['worker_restrictions'] = list(self.worker_restrictions[key])
             if key in self.resource_restrictions:
                 msg['resource_restrictions'] = self.resource_restrictions[key]
 

distributed/tests/test_worker.py

@@ -507,43 +507,12 @@ def test_system_monitor(s, a, b):
 
 
 
-@pytest.mark.skipif(not sys.platform.startswith('linux'),
-                    reason="Need 127.0.0.2 to mean localhost")
 @gen_cluster(client=True, ncores=[('127.0.0.1', 2, {'resources': {'A': 1}}),
-                                  ('127.0.0.2', 1)])
+                                  ('127.0.0.1', 1)])
 def test_restrictions(c, s, a, b):
-    # Worker restrictions
-    x = c.submit(inc, 1, workers=a.address)
-    yield x._result()
-    assert a.host_restrictions == {}
-    assert a.worker_restrictions == {x.key: {a.address}}
-    assert a.resource_restrictions == {}
-
-    yield c._cancel(x)
-
-    while x.key in a.task_state:
-        yield gen.sleep(0.01)
-
-    assert a.worker_restrictions == {}
-
-    # Host restrictions
-    x = c.submit(inc, 1, workers=['127.0.0.1'])
-    yield x._result()
-    assert a.host_restrictions == {x.key: {'127.0.0.1'}}
-    assert a.worker_restrictions == {}
-    assert a.resource_restrictions == {}
-    yield c._cancel(x)
-
-    while x.key in a.task_state:
-        yield gen.sleep(0.01)
-
-    assert a.host_restrictions == {}
-
     # Resource restrictions
     x = c.submit(inc, 1, resources={'A': 1})
     yield x._result()
-    assert a.host_restrictions == {}
-    assert a.worker_restrictions == {}
     assert a.resource_restrictions == {x.key: {'A': 1}}
     yield c._cancel(x)
 
@@ -600,3 +569,29 @@ def test_multiple_transfers(c, s, w1, w2, w3):
     r = w3.startstops[z.key]
     transfers = [t for t in r if t[0] == 'transfer']
     assert len(transfers) == 2
+
+
+@gen_cluster(client=True, ncores=[('127.0.0.1', 1)] * 3)
+def test_share_communication(c, s, w1, w2, w3):
+    x = c.submit(mul, b'1', int(w3.target_message_size + 1), workers=w1.address)
+    y = c.submit(mul, b'2', int(w3.target_message_size + 1), workers=w2.address)
+    yield _wait([x, y])
+    yield c._replicate([x, y], workers=[w1.address, w2.address])
+    z = c.submit(add, x, y, workers=w3.address)
+    yield _wait(z)
+    assert len(w3.incoming_transfer_log) == 2
+    assert w1.outgoing_transfer_log
+    assert w2.outgoing_transfer_log
+
+
+@gen_cluster(client=True)
+def test_dont_overlap_communications_to_same_worker(c, s, a, b):
+    x = c.submit(mul, b'1', int(b.target_message_size + 1), workers=a.address)
+    y = c.submit(mul, b'2', int(b.target_message_size + 1), workers=a.address)
+    yield _wait([x, y])
+    z = c.submit(add, x, y, workers=b.address)
+    yield _wait(z)
+    assert len(b.incoming_transfer_log) == 2
+    l1, l2 = b.incoming_transfer_log
+
+    assert l1['stop'] < l2['start']