Refactor occupancy #7075

hendrikmakait · 2022-09-27T13:16:28Z

This is an implementation of the suggestion in #7027

Supersedes

Refactor occupancy #7030

Tests added / passed
Passes pre-commit run --all-files

github-actions · 2022-09-27T13:58:14Z

Unit Test Results

See test report for an extended history of previous test failures. This is useful for diagnosing flaky tests.

      15 files ±  0       15 suites ±0 6h 34m 22s ⏱️ + 11m 6s
  3 135 tests - 11   3 048 ✔️ -   11   85 💤 ±0 2 ❌ ±0
23 191 runs - 95 22 271 ✔️ - 104 918 💤 +9 2 ❌ ±0

For more details on these failures, see this check.

Results for commit f8f11fe. ± Comparison against base commit 68e5a6a.

♻️ This comment has been updated with latest results.

distributed/scheduler.py

jrbourbeau

@crusaderky @gjoseph92 would either of you have time to review this? @hendrikmakait mentioned he'd like to see this included in the release tomorrow

gjoseph92

Overall, very happy to see this change. This metric feels pretty sensible to me, and I very much like having it always be correct.

Mostly nits, but a couple more significant comments too.

distributed/scheduler.py

gjoseph92 · 2022-09-30T00:26:41Z

distributed/scheduler.py

+    @property
+    def scheduler(self):
+        assert self.scheduler_ref
+        s = self.scheduler_ref()


I get the reason for this pattern of the weakref to the scheduler, but I don't love it. It just feels a little odd, and also means every access to self.scheduler has to resolve a few references.

How awkward would it be if all the TaskState methods that needed to do something to the scheduler state took the scheduler as an argument? That would also make it explicit that they mutate the scheduler.

The more I look at this, the more I consider moving these functions (or at least the parts that require access to a SchedulerState back up the hierarchy into the scheduler state. Since we need access to the SchedulerState as a whole, it seems that this would be the better correct root object to handle those operations. This would also avoid problems like acquiring resources when using Scheduler._add_to_processing but not doing so in WorkerState.add_to_processing. It's not clear which of these methods should be used by others such as WorkStealing and I'm increasingly convinced that it should not be the WorkerState-based ones.

This is a bit of an elaborate change, so I think it might make sense to extract it into an individual PR to avoid blocking this one and litter some technically unrelated refactoring changes into this.

distributed/scheduler.py

distributed/tests/test_scheduler.py

gjoseph92 · 2022-09-30T01:12:46Z

distributed/tests/test_steal.py

@@ -1361,6 +1356,7 @@ async def test_reschedule_concurrent_requests_deadlock(c, s, *workers):
    assert msgs in (expect1, expect2, expect3)


+@pytest.mark.skip("executing heartbeats not considered yet")


This also seems important to fix before a release. I have a feeling that main use cases where people actually depend on stealing right now is submitting a bunch of very, very slow tasks, then scaling up the cluster.

See #7030 (comment) for @fjetter's thoughts on this.

gjoseph92 · 2022-09-30T01:15:37Z

distributed/tests/test_steal.py

-            s._reevaluate_occupancy_worker(ws)
+    # Re-evaluate idle/saturated classification to avoid outdated classifications due to
+    # the initialization order of workers. On a real cluster, this would get constantly
+    # updated by tasks completing (except for stragglers).


Hm, as I mentioned in the comment above, there are definitely use-cases right now where people submit very, very slow tasks (~hours) and expect them to rebalance to new workers, even before any tasks have completed.

Adding this re-evaluation in here to every test case might hide things that could cause actual problems in this case?

I've adjusted the comment. The idle/saturated classification happens whenever a task completes or is added.

In your example, if we assume tasks being scheduled roughly round-robin, all workers should end up as saturated in the beginning. Since new workers would be idle, stealing should happen.

The problem in this particular test implementation is that we do not schedule tasks round-robin but for one worker after another. Thus, if we first schedule all specified tasks on a worker that's supposed to be idle in the grand scheme of the test, it would be classified as saturated for the lack of other even more saturated workers. These other workers only get filled up after this worker, which requires us to reclassify once after we completed the test setup.

Co-authored-by: Gabe Joseph <[email protected]>

hendrikmakait · 2022-09-30T13:20:26Z

distributed/scheduler.py

        assert ws in state.running, state.running
        assert (o := state.workers.get(ws.address)) is ws, (ws, o)

-    state._set_duration_estimate(ts, ws)
+    ws.add_to_processing(ts)
    ts.processing_on = ws
    ts.state = "processing"
    state.acquire_resources(ts, ws)


IIUC, we never acquire(d) resources in stealing. Is that on purpose or an oversight?

See #5937 (comment); I think this is a bug

Let's address this in another PR

hendrikmakait · 2022-09-30T17:51:00Z

A/B test: https://github.com/coiled/coiled-runtime/actions/runs/3159631751

Results
With n==7, there is no visible effect of the refactoring that could not be attributed to noise:

hendrikmakait · 2022-10-04T12:22:47Z

distributed/tests/test_steal.py

+
+
+# Reproducer from https://github.com/dask/distributed/issues/6573
+@gen_cluster(


Added reproducer from #6573 for regression testing. This has been solved in #7036.

hendrikmakait · 2022-10-04T14:37:15Z

distributed/stealing.py

@@ -487,7 +481,7 @@ def balance(self) -> None:
                    )

            if log:
-                self.log(log)
+                self.log(("request", log))


Driveby: Add an identifier to the logged bulk event.

hendrikmakait · 2022-10-05T15:01:15Z

CI flakes

CI failing with test_get_task_stream_save #6312

Co-authored-by: fjetter <[email protected]> Co-authored-by: Gabe Joseph <[email protected]>

fjetter and others added 16 commits September 12, 2022 19:16

Refactor occupancy

1c3fcb4

fix test_balance

0190c46

Fix long running

588ab76

more fixes

ccfd1f1

Skip a couple of tests

8ab749d

normalize network occ

fca0e09

skip one more test

7d00c9e

Ensure long running tasks do not generate negative counts

c8743b5

Ensure heartbeats are considered

b20fd63

review

47770a7

Delete test

007e1cb

Ensure leaving worker resets total occupancy

35a86b8

Merge remote-tracking branch 'origin/main' into occupancy_refactor

d4a595b

Review comments

a98049c

Improve typing

1d9b3ca

Fix WorkerState.clean

32feebe

hendrikmakait changed the title ~~Occupancy refactor~~ Refactor occupancy Sep 27, 2022

hendrikmakait commented Sep 27, 2022

View reviewed changes

distributed/scheduler.py Outdated Show resolved Hide resolved

hendrikmakait added 6 commits September 29, 2022 14:21

Ensure long-running is a subset of processing

9f6643d

Merge branch 'main' into occupancy_refactor

fed7c03

Merge branch 'main' into occupancy_refactor

1f5e085

Update docstring

c837ed8

Fix tests

ab4c4c9

fix test_secede_cancelled_or_resumed_scheduler

a5ec27e

hendrikmakait mentioned this pull request Sep 29, 2022

Release 2022.9.2 dask/community#278

Closed

5 tasks

jrbourbeau reviewed Sep 29, 2022

View reviewed changes

gjoseph92 reviewed Sep 30, 2022

View reviewed changes

hendrikmakait and others added 2 commits September 30, 2022 08:31

unskip

03ecf02

Apply suggestions from code review

036b4ac

Co-authored-by: Gabe Joseph <[email protected]>

hendrikmakait added 5 commits September 30, 2022 09:05

Review comment

e14e509

Improve comment

bc71ac3

Drop test

79de0a4

Do not remove from processing in remove_worker

1b2d4f3

Improve test_include_communication_in_occupancy

d31b215

crusaderky assigned hendrikmakait Sep 30, 2022

hendrikmakait added 5 commits September 30, 2022 12:31

Reorder operations to fix remove_worker

1eac929

Minor

94ff8dd

Add validate sections for replicas

eb525db

Add more validation

ea34ba2

Minor

af302ee

hendrikmakait commented Sep 30, 2022

View reviewed changes

Merge branch 'main' into occupancy_refactor

e31cdc6

Add reproducer from 6573

b4b2978

hendrikmakait commented Oct 4, 2022

View reviewed changes

hendrikmakait added 2 commits October 4, 2022 14:27

Add event descriptor to log

36f4ce4

Adjust stealing-event-based functionality

a6e501e

hendrikmakait commented Oct 4, 2022

View reviewed changes

Turn off AMM for work-stealing tests

f8f11fe

fjetter approved these changes Oct 7, 2022

View reviewed changes

fjetter merged commit 07e2259 into dask:main Oct 7, 2022

fjetter mentioned this pull request Oct 10, 2022

Do not include psutil memory_info in GC duration measurement #7127

Open

wence- mentioned this pull request Oct 13, 2022

flaky test: test_balance_multiple_to_replica #7137

Open

hayesgb mentioned this pull request Oct 17, 2022

Timeboxed push for simplifying work stealing #6993

Closed

4 tasks

fjetter mentioned this pull request Oct 18, 2022

Root-ish tasks all schedule onto one worker #6573

Closed

gjoseph92 added a commit to gjoseph92/distributed that referenced this pull request Oct 31, 2022

Refactor occupancy (dask#7075)

947f995

Co-authored-by: fjetter <[email protected]> Co-authored-by: Gabe Joseph <[email protected]>

hendrikmakait mentioned this pull request Nov 3, 2022

Replace test_(do_not_)steal_communication_heavy_tasks tests with more robust versions #7243

Merged

2 tasks

gjoseph92 mentioned this pull request Nov 10, 2022

Scheduler.total_occupancy is significant runtime cost #7256

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Refactor occupancy #7075

Refactor occupancy #7075

hendrikmakait commented Sep 27, 2022 •

edited

Loading

github-actions bot commented Sep 27, 2022 •

edited

Loading

jrbourbeau left a comment

gjoseph92 left a comment

gjoseph92 Sep 30, 2022

hendrikmakait Sep 30, 2022 •

edited

Loading

hendrikmakait Sep 30, 2022

gjoseph92 Sep 30, 2022

hendrikmakait Sep 30, 2022

gjoseph92 Sep 30, 2022

hendrikmakait Sep 30, 2022

hendrikmakait Sep 30, 2022

gjoseph92 Sep 30, 2022

fjetter Oct 7, 2022

hendrikmakait commented Sep 30, 2022 •

edited

Loading

hendrikmakait Oct 4, 2022

hendrikmakait Oct 4, 2022 •

edited

Loading

hendrikmakait commented Oct 5, 2022

		@@ -1361,6 +1356,7 @@ async def test_reschedule_concurrent_requests_deadlock(c, s, *workers):
		assert msgs in (expect1, expect2, expect3)


		@pytest.mark.skip("executing heartbeats not considered yet")



		# Reproducer from https://github.com/dask/distributed/issues/6573
		@gen_cluster(

Refactor occupancy #7075

Refactor occupancy #7075

Conversation

hendrikmakait commented Sep 27, 2022 • edited Loading

github-actions bot commented Sep 27, 2022 • edited Loading

Unit Test Results

jrbourbeau left a comment

Choose a reason for hiding this comment

gjoseph92 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hendrikmakait Sep 30, 2022 • edited Loading

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

hendrikmakait commented Sep 30, 2022 • edited Loading

Choose a reason for hiding this comment

hendrikmakait Oct 4, 2022 • edited Loading

Choose a reason for hiding this comment

hendrikmakait commented Oct 5, 2022

hendrikmakait commented Sep 27, 2022 •

edited

Loading

github-actions bot commented Sep 27, 2022 •

edited

Loading

hendrikmakait Sep 30, 2022 •

edited

Loading

hendrikmakait commented Sep 30, 2022 •

edited

Loading

hendrikmakait Oct 4, 2022 •

edited

Loading