From 792ab1fe20325d69c9600fa52d513ece647c95e3 Mon Sep 17 00:00:00 2001 From: Jacob Tomlinson Date: Mon, 30 Sep 2019 15:32:52 +0100 Subject: [PATCH 1/4] Retry scheduler connect multiple times --- distributed/comm/core.py | 23 +++++++++++++---------- 1 file changed, 13 insertions(+), 10 deletions(-) diff --git a/distributed/comm/core.py b/distributed/comm/core.py index 1bbc043f52d..5fa0c12df47 100644 --- a/distributed/comm/core.py +++ b/distributed/comm/core.py @@ -1,6 +1,7 @@ from abc import ABC, abstractmethod, abstractproperty from datetime import timedelta import logging +from warnings import ignoring import weakref import dask @@ -188,6 +189,7 @@ async def connect(addr, timeout=None, deserialize=True, connection_args=None): scheme, loc = parse_address(addr) backend = registry.get_backend(scheme) connector = backend.get_connector() + comm = None start = time() deadline = start + timeout @@ -205,14 +207,17 @@ def _raise(error): # This starts a thread while True: try: - future = connector.connect( - loc, deserialize=deserialize, **(connection_args or {}) - ) - comm = await gen.with_timeout( - timedelta(seconds=deadline - time()), - future, - quiet_exceptions=EnvironmentError, - ) + while deadline - time(): + future = connector.connect( + loc, deserialize=deserialize, **(connection_args or {}) + ) + with ignoring(gen.TimeoutError): + comm = await gen.with_timeout( + timedelta(seconds=1), future, quiet_exceptions=EnvironmentError + ) + break + if not comm: + _raise(error) except FatalCommClosedError: raise except EnvironmentError as e: @@ -222,8 +227,6 @@ def _raise(error): logger.debug("sleeping on connect") else: _raise(error) - except gen.TimeoutError: - _raise(error) else: break From caabe2c7b0981d7fc1360cab1ebbaf18bf376286 Mon Sep 17 00:00:00 2001 From: Jacob Tomlinson Date: Mon, 30 Sep 2019 15:36:39 +0100 Subject: [PATCH 2/4] Fix while --- distributed/comm/core.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/distributed/comm/core.py b/distributed/comm/core.py index 5fa0c12df47..1d45199d9cf 100644 --- a/distributed/comm/core.py +++ b/distributed/comm/core.py @@ -207,7 +207,7 @@ def _raise(error): # This starts a thread while True: try: - while deadline - time(): + while deadline - time() > 0: future = connector.connect( loc, deserialize=deserialize, **(connection_args or {}) ) From 26f438ad981c9814bcfabe27a1f1262ace53d380 Mon Sep 17 00:00:00 2001 From: Jacob Tomlinson Date: Mon, 30 Sep 2019 15:44:49 +0100 Subject: [PATCH 3/4] Import ignoring from the right place --- distributed/comm/core.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/distributed/comm/core.py b/distributed/comm/core.py index 1d45199d9cf..1586cf46276 100644 --- a/distributed/comm/core.py +++ b/distributed/comm/core.py @@ -1,14 +1,13 @@ from abc import ABC, abstractmethod, abstractproperty from datetime import timedelta import logging -from warnings import ignoring import weakref import dask from tornado import gen from ..metrics import time -from ..utils import parse_timedelta +from ..utils import parse_timedelta, ignoring from . import registry from .addressing import parse_address From 4cca2d194bd296e2d1255b271ecc9d5c698fc581 Mon Sep 17 00:00:00 2001 From: Jacob Tomlinson Date: Mon, 30 Sep 2019 15:57:45 +0100 Subject: [PATCH 4/4] Use timeout if less than 1 second. --- distributed/comm/core.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/distributed/comm/core.py b/distributed/comm/core.py index 1586cf46276..256a17de3a5 100644 --- a/distributed/comm/core.py +++ b/distributed/comm/core.py @@ -212,7 +212,9 @@ def _raise(error): ) with ignoring(gen.TimeoutError): comm = await gen.with_timeout( - timedelta(seconds=1), future, quiet_exceptions=EnvironmentError + timedelta(seconds=min(deadline - time(), 1)), + future, + quiet_exceptions=EnvironmentError, ) break if not comm: