Refactor all event handlers #6410

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

                "who_has": who_has,
                "stimulus_id": stimulus_id,
            },
        )

-    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

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Refactor all event handlers #6410

Refactor all event handlers #6410

crusaderky commented May 22, 2022 •

edited

Loading

github-actions bot commented May 22, 2022 •

edited

Loading

crusaderky May 22, 2022

pentschev commented May 26, 2022

gjoseph92 left a comment

gjoseph92 Jun 6, 2022

crusaderky Jun 6, 2022

gjoseph92 Jun 6, 2022

crusaderky Jun 6, 2022

gjoseph92 Jun 6, 2022

crusaderky Jun 6, 2022

gjoseph92 Jun 6, 2022

crusaderky Jun 6, 2022

gjoseph92 Jun 6, 2022

crusaderky Jun 6, 2022

gjoseph92 Jun 6, 2022

crusaderky Jun 6, 2022

	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

Refactor all event handlers #6410

Refactor all event handlers #6410

Conversation

crusaderky commented May 22, 2022 • edited Loading

github-actions bot commented May 22, 2022 • edited Loading

Unit Test Results

Choose a reason for hiding this comment

pentschev commented May 26, 2022

gjoseph92 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

crusaderky commented May 22, 2022 •

edited

Loading

github-actions bot commented May 22, 2022 •

edited

Loading