Skip to content

Retry scheduler connect multiple times #3104

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
mrocklin merged 4 commits into from
Sep 30, 2019
Merged

Retry scheduler connect multiple times #3104

Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
792ab1f
Retry scheduler connect multiple times
jacobtomlinson Sep 30, 2019
caabe2c
Fix while
jacobtomlinson Sep 30, 2019
26f438a
Import ignoring from the right place
jacobtomlinson Sep 30, 2019
4cca2d1
Use timeout if less than 1 second.
jacobtomlinson Sep 30, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 15 additions & 11 deletions distributed/comm/core.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@
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

Expand Down Expand Up @@ -188,6 +188,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
Expand All @@ -205,14 +206,19 @@ 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() > 0:
future = connector.connect(
loc, deserialize=deserialize, **(connection_args or {})
)
with ignoring(gen.TimeoutError):
comm = await gen.with_timeout(
timedelta(seconds=min(deadline - time(), 1)),
future,
quiet_exceptions=EnvironmentError,
)
break
if not comm:
_raise(error)
except FatalCommClosedError:
raise
except EnvironmentError as e:
Expand All @@ -222,8 +228,6 @@ def _raise(error):
logger.debug("sleeping on connect")
else:
_raise(error)
except gen.TimeoutError:
_raise(error)
else:
break

Expand Down