Refactor all event handlers by crusaderky · Pull Request #6410 · dask/distributed

crusaderky · 2022-05-22T12:53:21Z

Part of #5736

github-actions · 2022-05-22T14:26:42Z

Unit Test Results

      15 files +      15       15 suites +15 6h 35m 10s ⏱️ + 6h 35m 10s
  2 852 tests +  2 852   2 736 ✔️ +  2 736   82 💤 +  82 31 ❌ +31 3 🔥 +3
21 129 runs +21 129 20 146 ✔️ +20 146 946 💤 +946 34 ❌ +34 3 🔥 +3

For more details on these failures and errors, see this check.

Results for commit 42d399f. ± Comparison against base commit 7d280fd.

♻️ This comment has been updated with latest results.

crusaderky · 2022-05-22T22:13:35Z

distributed/worker.py

+    @functools.singledispatchmethod
+    def handle_event(self, ev: StateMachineEvent) -> RecsInstrs:
+        raise TypeError(ev)  # pragma: nocover
+


Moved so that I didn't have to move the various refactored functions, thus minimising the diff.
This is temporary - this and all the registered methods will be moved to worker_state_machine.py.

distributed/worker.py

pentschev · 2022-05-26T15:57:24Z

rerun tests

Note: rerunning gpuCI tests since those errors should be fixed by #6434

gjoseph92

LGTM overall. All comments are nits and could be ignored, besides the list->set question and using descriptive function names.

gjoseph92 · 2022-06-06T16:08:42Z

distributed/scheduler.py

-            ts = self.tasks[key]
-            who_has[key] = {ws.address for ws in ts.who_has}
-
+        who_has = {key: [ws.address for ws in self.tasks[key].who_has] for key in keys}


Suggested change

who_has = {key: [ws.address for ws in self.tasks[key].who_has] for key in keys}

who_has = {key: {ws.address for ws in self.tasks[key].who_has} for key in keys}

This was changed from a list to a set, was that intentional?

from a set to a list. yes, as the set doesn't offer any useful feature here and a list is fractionally faster. This also makes it coherent with Scheduler.get_who_has.

gjoseph92 · 2022-06-06T16:09:24Z

distributed/scheduler.py


-    def request_remove_replicas(self, addr: str, keys: list, *, stimulus_id: str):
+    def request_remove_replicas(
+        self, addr: str, keys: list[str], *, stimulus_id: str


Suggested change

self, addr: str, keys: list[str], *, stimulus_id: str

self, addr: str, keys: Iterable[str], *, stimulus_id: str

No, because the variable is sent as-is through the msgpack RPC, so only a list, set, or tuple are accepted.

gjoseph92 · 2022-06-06T16:16:11Z

distributed/worker.py

+        return {"nbytes": {k: sizeof(v) for k, v in data.items()}, "status": "OK"}
+
+    @handle_event.register
+    def _(self, ev: UpdateDataEvent) -> RecsInstrs:


Suggested change

def _(self, ev: UpdateDataEvent) -> RecsInstrs:

def handle_update_data(self, ev: UpdateDataEvent) -> RecsInstrs:

Could we please name these functions recognizably, even though they're singledispatch handlers? It makes it much easier to navigate the code. I usually fuzzy-find search for terms like update data to find the handler.

gjoseph92 · 2022-06-06T20:14:10Z

distributed/worker_state_machine.py

+        if isinstance(self.run_spec, dict):
+            self.run_spec = SerializedTask(**self.run_spec)
+        elif not isinstance(self.run_spec, SerializedTask):
+            self.run_spec = SerializedTask(task=self.run_spec)


A bit nitpicky, feel free to ignore. But it might be nice to use init-only variables for function, args, kwargs, and task, make run_spec a non-init field, and go back to this when constructing the message:

if isinstance(ts.run_spec, dict): msg.update(ts.run_spec) else: msg["task"] = ts.run_spec

Just a little more explicitness and type checking around what has to be passed.

The reason why I packed it this way was to avoid replicating all the (fairly convoluted) annotations in the SerializedTask class. The SerializedTask class is messy to begin with - given more time, I'd redesign it thoroughly.

gjoseph92 · 2022-06-06T20:23:40Z

distributed/tests/test_worker_state_machine.py

+    ev2 = ev.to_loggable(handled=11.22)
+    assert ev2.handled == 11.22
+    assert ev2.run_spec == SerializedTask(task=None)
+    assert ev.run_spec == SerializedTask(function=b"blob", args=b"blob")


Suggested change

assert ev.run_spec == SerializedTask(function=b"blob", args=b"blob")

assert ev.run_spec == SerializedTask(function=b"blob", args=b"blob")

assert not ev.handled

This isn't a feature we want to test, just an implementation detail. As a matter of fact, the base method StateMachineEvent.to_loggable overwrites the original object for the sake of performance.

gjoseph92 · 2022-06-06T20:25:19Z

distributed/tests/test_worker_state_machine.py

+    )
+    ev2 = ev.to_loggable(handled=11.22)
+    assert ev2.handled == 11.22
+    assert ev2.data == {"x": None, "y": None}


Suggested change

assert ev2.data == {"x": None, "y": None}

assert ev2.data == {"x": None, "y": None}

assert not ev.handled

assert ev.data == {"x": "foo", "y": "bar"}

Hotfix for #6624 by reverting the compute-task message format almost to the original state before #6410

crusaderky self-assigned this May 22, 2022

crusaderky force-pushed the WSMR/refactor_handlers branch from 933127a to d16cf26 Compare May 22, 2022 13:37

crusaderky force-pushed the WSMR/refactor_handlers branch from d16cf26 to 64d4b4a Compare May 22, 2022 20:55

crusaderky commented May 22, 2022

View reviewed changes

crusaderky marked this pull request as ready for review May 22, 2022 22:16

crusaderky commented May 22, 2022

View reviewed changes

distributed/worker.py Outdated Show resolved Hide resolved

crusaderky commented May 22, 2022

View reviewed changes

distributed/worker.py Outdated Show resolved Hide resolved

crusaderky added a commit to crusaderky/distributed that referenced this pull request May 25, 2022

Move unrelated change to dask#6410

933c9cf

crusaderky mentioned this pull request May 30, 2022

Yank state machine out of Worker class #6476

Closed

crusaderky linked an issue May 30, 2022 that may be closed by this pull request

Yank state machine out of Worker class #6476

Closed

Refactor all Worker event handlers

052828c

crusaderky force-pushed the WSMR/refactor_handlers branch from cd37b28 to 052828c Compare June 1, 2022 12:07

Merge branch 'main' into WSMR/refactor_handlers

6a0bb98

fjetter mentioned this pull request Jun 1, 2022

Cosmetic review of story() #6442

Merged

crusaderky added 5 commits June 1, 2022 22:54

Merge branch 'main' into WSMR/refactor_handlers

8ba2605

Remove public transition method

b65ad99

Merge branch 'main' into WSMR/refactor_handlers

cf58b5f

Merge branch 'main' into WSMR/refactor_handlers

72b11ed

polish

a9f6439

hayesgb requested a review from gjoseph92 June 6, 2022 14:55

crusaderky added 2 commits June 6, 2022 16:20

Fix failing tests

249e819

Merge branch 'main' into WSMR/refactor_handlers

42d399f

gjoseph92 reviewed Jun 6, 2022

View reviewed changes

crusaderky added 3 commits June 7, 2022 00:06

use generic Collection[str] in events

089c545

name all handle_event dispatcher

89b6d36

Merge branch 'main' into WSMR/refactor_handlers

b9cb485

gjoseph92 approved these changes Jun 6, 2022

View reviewed changes

crusaderky merged commit bd98e66 into dask:main Jun 7, 2022

crusaderky deleted the WSMR/refactor_handlers branch June 7, 2022 00:15

This was referenced Jun 24, 2022

map_overlap tasks fail to deserialize on workers - keywords must be strings #6624

Closed

Partial revert of compute-task message format #6626

Merged

jsignell pushed a commit that referenced this pull request Jun 24, 2022

Partial revert of compute-task message format (#6626)

a156c35

Hotfix for #6624 by reverting the compute-task message format almost to the original state before #6410

	who_has = {key: [ws.address for ws in self.tasks[key].who_has] for key in keys}
	who_has = {key: {ws.address for ws in self.tasks[key].who_has} for key in keys}

	self, addr: str, keys: list[str], *, stimulus_id: str
	self, addr: str, keys: Iterable[str], *, stimulus_id: str

	def _(self, ev: UpdateDataEvent) -> RecsInstrs:
	def handle_update_data(self, ev: UpdateDataEvent) -> RecsInstrs:

	assert ev.run_spec == SerializedTask(function=b"blob", args=b"blob")
	assert ev.run_spec == SerializedTask(function=b"blob", args=b"blob")
	assert not ev.handled

Uh oh!

Conversation

crusaderky commented May 22, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

github-actions bot commented May 22, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Unit Test Results

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

pentschev commented May 26, 2022

Uh oh!

gjoseph92 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

crusaderky commented May 22, 2022 •

edited

Loading

github-actions bot commented May 22, 2022 •

edited

Loading