Avoid client connecting to scheduler twice
Resolves intermittent failure test_channels.py::test_worker_client
@@ -386,6 +386,7 @@ class Client(object):
self.extensions = {}
self.scheduler_file = scheduler_file
self._startup_kwargs = kwargs
+ self._connecting_to_scheduler = False
if hasattr(address, "scheduler_address"):
# It's a LocalCluster or LocalCluster-compatible object
@@ -528,16 +529,23 @@ class Client(object):
@gen.coroutine
def _ensure_connected(self, timeout=5):
- if self.scheduler_comm and not self.scheduler_comm.closed():
+ if (self.scheduler_comm and not self.scheduler_comm.closed() or
+ self._connecting_to_scheduler):
return
+ self._connecting_to_scheduler = True
+
+ try:
comm = yield connect(self.scheduler.address,
timeout=timeout)
yield self.scheduler.identity()
yield comm.write({'op': 'register-client',
- 'client': self.id, 'reply': False})
+ 'client': self.id,
+ 'reply': False})
+ finally:
+ self._connecting_to_scheduler = False
msg = yield comm.read()
assert len(msg) == 1
assert msg[0]['op'] == 'stream-start'
