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.

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

if there's an exception in the Client async context manager body then close fast #6920

Merged
merged 5 commits into from
Nov 2, 2022
Merged
Show file tree
Hide file tree
Changes from 4 commits
Commits
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
17 changes: 15 additions & 2 deletions distributed/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -1383,7 +1383,12 @@ async def __aenter__(self):
return self

async def __aexit__(self, exc_type, exc_value, traceback):
await self._close()
await self._close(
# if we're handling an exception, we assume that it's more
# important to deliver that exception than shutdown gracefully.
fast=exc_type
is not None
)

def __exit__(self, exc_type, exc_value, traceback):
self.close()
Expand Down Expand Up @@ -1513,7 +1518,15 @@ def _handle_error(self, exception=None):
logger.exception(exception)

async def _close(self, fast=False):
"""Send close signal and wait until scheduler completes"""
"""
Send close signal and wait until scheduler completes

If fast is True, the client will close forcefully, by cancelling tasks
the background _handle_report_task.
"""
# TODO: aclose more forcefully by aborting the RPC and cancelling all
# background tasks.
# see https://trio.readthedocs.io/en/stable/reference-io.html#trio.aclose_forcefully
if self.status == "closed":
return

Expand Down
19 changes: 19 additions & 0 deletions distributed/tests/test_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@
from threading import Semaphore
from time import sleep
from typing import Any
from unittest import mock

import psutil
import pytest
Expand Down Expand Up @@ -7538,3 +7539,21 @@ def __init__(self, *args, **kwargs):
(DeprecationWarning, "The io_loop property is deprecated"),
(DeprecationWarning, "setting the loop property is deprecated"),
]


@gen_cluster(client=False, nthreads=[])
async def test_fast_close_on_aexit_failure(s):
class MyException(Exception):
pass

c = Client(s.address, asynchronous=True)
with mock.patch.object(c, "_close", wraps=c._close) as _close_proxy:
with pytest.raises(MyException):
async with c:
start = time()
raise MyException
stop = time()

assert _close_proxy.mock_calls == [mock.call(fast=True)]
assert c.status == "closed"
assert (stop - start) < 2
Copy link
Member Author

@graingert graingert Aug 19, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I couldn't find any tests that tested what _close(fast=True) does:

it just reduces this 2 second timeout to 0 (I think fast=True should be quite a bit more aggressive, eg force close rpc etc)

await asyncio.wait_for(handle_report_task, 0 if fast else 2)

so that's why I used a spy here

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: Would you mind incorporating a description of what closing fast means into the docstring for _close? I was wondering the same thing when I started reading this PR.

Copy link
Member Author

@graingert graingert Aug 23, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

introduced here https://github.com/dask/distributed/pull/27/files#diff-ce261e4d4b0185c8defbfcd46ebe8311a3ecb7ab81d653a2c6664c840340ab2cR250-R251 where it was used as a flag to leak tasks - however we now don't leak tasks but instead cancel them and wait for them

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@hendrikmakait thanks for the review, it helped clarify my thinking on this!

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: Would it make sense to rename fast to force?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah I think "forcefully=True" is probably the best here