Skip to content
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

Replace @gen.coroutine with async in Server.close #6365

Closed
wants to merge 3 commits into from
Closed

Replace @gen.coroutine with async in Server.close #6365

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
d51943f
Replace @gen.coroutine with async
hendrikmakait May 18, 2022
b20105d
Fix gather
hendrikmakait May 18, 2022
2564924
Minor
hendrikmakait May 18, 2022
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
15 changes: 8 additions & 7 deletions distributed/core.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -666,8 +666,7 @@ async def handle_stream(self, comm, extra=None):
await comm.close()
assert comm.closed()

@gen.coroutine
def close(self):
async def close(self):
for pc in self.periodic_callbacks.values():
pc.stop()

Expand All @@ -676,23 +675,25 @@ def close(self):
for listener in self.listeners:
future = listener.stop()
if inspect.isawaitable(future):
yield future
await future
for i in range(20):
# If there are still handlers running at this point, give them a
# second to finish gracefully themselves, otherwise...
if any(self._comms.values()):
yield asyncio.sleep(0.05)
await asyncio.sleep(0.05)
else:
break
yield self.rpc.close()
yield [comm.close() for comm in list(self._comms)] # then forcefully close
await self.rpc.close()
await asyncio.gather(
*(comm.close() for comm in self._comms)
) # then forcefully close
for cb in self._ongoing_coroutines:
cb.cancel()
for i in range(10):
if all(c.cancelled() for c in self._ongoing_coroutines):
break
else:
yield asyncio.sleep(0.01)
await asyncio.sleep(0.01)

self._event_finished.set()

Expand Down