Skip to content

Add Cluster.called_from_running_loop and fix Cluster.asynchronous #7941

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
jrbourbeau merged 3 commits into from
Jun 26, 2023
Merged

Add Cluster.called_from_running_loop and fix Cluster.asynchronous #7941

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
0adc10b
Add Cluster.called_from_running_loop and fix Cluster.asynchronous
jacobtomlinson Jun 22, 2023
2a15f00
Store asynchronous kwarg in private attribute
jacobtomlinson Jun 22, 2023
e9601a9
Linting
jacobtomlinson Jun 22, 2023
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
10 changes: 10 additions & 0 deletions distributed/deploy/cluster.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -71,6 +71,7 @@ def __init__(
scheduler_sync_interval=1,
):
self._loop_runner = LoopRunner(loop=loop, asynchronous=asynchronous)
self.__asynchronous = asynchronous

@fjetter fjetter Jun 26, 2023

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

I'm slightly confused. The SyncMethodMixin is checking for _asynchronous but you chose to use an attribute with name mangling (double underscore) so this attribute will never be detected by the mixin. is this intentional?

@jacobtomlinson jacobtomlinson Jun 26, 2023

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

As far as I can see _asynchronous is never set anywhere apart from Client and Actor. So the mixin always falls back to the default for cluster objects. I went with the double underscore because I only wanted to use it for this one purpose and didn't want to collide with whatever _asynchronous is being used for, given that it was unclear to me what _asynchronous is even doing.

Happy to do something else here, do you have a suggestion?

@fjetter fjetter Jun 26, 2023

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

All good. Just wanted to make sure this is intentional. For this usecase the mangling is a good choice.


self.scheduler_info = {"workers": {}}
self.periodic_callbacks = {}
Expand Down Expand Up @@ -114,6 +115,15 @@ def loop(self, value: IOLoop) -> None:
raise ValueError("expected an IOLoop, got None")
self.__loop = value

@property
def called_from_running_loop(self):
try:
return (
getattr(self.loop, "asyncio_loop", None) is asyncio.get_running_loop()
)
except RuntimeError:
return self.__asynchronous

@property
def name(self):
return self._cluster_info["name"]
Expand Down
9 changes: 1 addition & 8 deletions distributed/deploy/spec.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -279,14 +279,7 @@ def __init__(
scheduler_sync_interval=scheduler_sync_interval,
)

try:
called_from_running_loop = (
getattr(loop, "asyncio_loop", None) is asyncio.get_running_loop()
)
except RuntimeError:
called_from_running_loop = asynchronous

if not called_from_running_loop:
if not self.called_from_running_loop:
self._loop_runner.start()
self.sync(self._start)
try:
Expand Down
7 changes: 6 additions & 1 deletion distributed/utils.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -334,7 +334,12 @@ class SyncMethodMixin:
@property
def asynchronous(self):
"""Are we running in the event loop?"""
return in_async_call(self.loop, default=getattr(self, "_asynchronous", False))
try:
return in_async_call(
self.loop, default=getattr(self, "_asynchronous", False)
)
except RuntimeError:
return False

def sync(self, func, *args, asynchronous=None, callback_timeout=None, **kwargs):
"""Call `func` with `args` synchronously or asynchronously depending on
Expand Down