Forget erred tasks // Fix deadlocks on worker

[fjetter]

I need some feedback because the cluster behaves differently than I would expect and I would like to know for sure what we expect to happen.

The test case constructs a very simple graph consisting of three tasks, F, G and Res where F and G are simple dependencies executed on different workers while Res is task which will fail. I pinned the workers down explicitly to ensure determinism but otherwise this is not relevant.

What happens is that the task errs which is expected. We keep all tasks in memory, still expected. Afterwards, two things happen which surprised me and I would like to know if those are bugs or expected behaviour I misunderstood

1. I release refs to F and G and would expect them to be released, i.e. the memory is freed because Res is done and we no longer hold a reference to the dependencies F and G. Instead, both stay in memory. Is this expected?
2. After I move on, I release refs to Res and would expect the cluster to reset, i.e. forget about all tasks. This isn't happening but instead F and G are released and Res stays erred. The tasks are never forgotten.

cc @jrbourbeau

---

Edit:

While investigating the free error tasks, I found two deadlock situations which were more likely to be triggered in this branch than in others. This increased the scope of this PR

[fjetter]

This state is actually never left so I do consider this a bug. Eventually the tasks should be released (I commented out the wait below)

[fjetter]

I would expect g to be released on worker A since it was only on worker A as a dependency. However, it is kept after the task erred. Do we consider this sane behaviour? Do we intentionally keep this dependency in memory to allow for retries?

[fjetter]

For reference, the scheduler decides in this case that the keys should be kept, or rather it doesn't decide to release them. I am questioning whether this is the correct decision

[fjetter]

Even after explicit refs to f and g are released, we still keep holding on to them since there is the dependency res which erred. I'm wondering what the expected behaviour should be here

[fjetter]

For reference, the scheduler forgot all tasks at this point

[fjetter]

One of the reasons why the erred task state is never forgotten is that keys are only "remembered" in processing_on (processing) or who_has (memory), depending on the state of the task. This would be the equivalent for erred tasks to allow us to tell the worker to forget the task.

@jakirkham you have been involved in the scheduler state machine a lot recently. Just pinging in case you have thoughts about adding more state here or if you see other options

[fjetter]

I ran into some issues with the nbytes counting and wondered if this is correct. While the test states this explicitly, I'm a bit surprised and believe there are code areas where this is treated differently

[fjetter]

To not release too many tasks but release those which are free I ended up introducing additional state on the scheduler erred_on and there is one additional attribute on worker TaskState side, scheduler_holds_ref. This is a subtle problem I was trying to solve in #4772 slightly differently but the end result is the same. The worker needs to know whether the scheduler still requires a task or not. Unless we replicate the entire graph we otherwise do not have a chance to make the right decision. Even if we were to replicate the entire graph we probably would need to introduce something like this to give the scheduler ultimate authority. That problem will pop up again once we approach speculative task assignment.

This additional reference count is required because in the past, when tasks and dependencies were clearly separated, this reference count was implicit. Tasks were always referred to by the scheduler, dependencies were just a side product and could be forgotten asap. You will therefore notice that all "compute-task" TaskState objects have this attribute set and dependencies don't.

Tests look promising and while this is not the end of the story for the stability issues (I still have reproducing tests which deadlock) this is an important step.

Gentle ping to @jrbourbeau and @gforsyth for review

[gforsyth]

Hey @fjetter -- this looks good to me. I flagged one small typo in the scheduler attributes but that doesn't need to block this.

[fjetter]

Hitting an issue in the py3.8 ubu test which might be connected although I'm not sure, yet.

Traceback (most recent call last):
  File "/home/runner/work/distributed/distributed/distributed/worker.py", line 1720, in transition_fetch_flight
    assert ts.dependents
AssertionError
distributed.worker - ERROR - 
Traceback (most recent call last):
  File "/home/runner/work/distributed/distributed/distributed/worker.py", line 2057, in ensure_communicating
    self.transition(self.tasks[d], "flight", worker=worker)
  File "/home/runner/work/distributed/distributed/distributed/worker.py", line 1615, in transition
    state = func(ts, **kwargs)
  File "/home/runner/work/distributed/distributed/distributed/worker.py", line 1720, in transition_fetch_flight
    assert ts.dependents
AssertionError
distributed.core - ERROR - 
Traceback (most recent call last):
  File "/home/runner/work/distributed/distributed/distributed/core.py", line 582, in handle_stream
    func()
  File "/home/runner/work/distributed/distributed/distributed/worker.py", line 2057, in ensure_communicating
    self.transition(self.tasks[d], "flight", worker=worker)
  File "/home/runner/work/distributed/distributed/distributed/worker.py", line 1615, in transition
    state = func(ts, **kwargs)
  File "/home/runner/work/distributed/distributed/distributed/worker.py", line 1720, in transition_fetch_flight
    assert ts.dependents
  File "/home/runner/work/distributed/distributed/distributed/worker.py", line 1615, in transition
    state = func(ts, **kwargs)
  File "/home/runner/work/distributed/distributed/distributed/worker.py", line 1773, in transition_flight_memory
    self.batched_stream.send({"op": "add-keys", "keys": [ts.key]})
  File "/home/runner/work/distributed/distributed/distributed/batched.py", line 136, in send
    raise CommClosedError()
distributed.comm.core.CommClosedError
distributed.utils - ERROR - "('sum-partial-5b2c0f6d9d691a15dc449febe739d843', 0, 1)"
Traceback (most recent call last):
  File "/home/runner/work/distributed/distributed/distributed/utils.py", line 671, in log_errors
    yield
  File "/home/runner/work/distributed/distributed/distributed/scheduler.py", line 4042, in add_worker
    typename=types[key],
KeyError: "('sum-partial-5b2c0f6d9d691a15dc449febe739d843', 0, 1)"
distributed.core - ERROR - Exception while handling op register-worker
Traceback (most recent call last):
  File "/home/runner/work/distributed/distributed/distributed/core.py", line 501, in handle_comm
    result = await result
  File "/home/runner/work/distributed/distributed/distributed/scheduler.py", line 4042, in add_worker
    typename=types[key],

This also triggers a similar error as reported in #4800 with

distributed.utils - ERROR - "('sum-partial-5b2c0f6d9d691a15dc449febe739d843', 0, 1)"
Traceback (most recent call last):
  File "/home/runner/work/distributed/distributed/distributed/utils.py", line 671, in log_errors
    yield
  File "/home/runner/work/distributed/distributed/distributed/scheduler.py", line 4042, in add_worker
    typename=types[key],
KeyError: "('sum-partial-5b2c0f6d9d691a15dc449febe739d843', 0, 1)"

at least the very first assert could be related

[fjetter]

I assume the above is connected since all three failures (ubu 3.8 and the two osx failures) are the same failing test. However, it's one of these incredibly rare events which is hard to reproduce locally

[fjetter]

All tests green 🎉

[fjetter]

friendly ping to @gforsyth and @jrbourbeau for another review.

What changed since the last time I asked for reviews are the handlers which control task releases on the worker and how and when the scheduler invokes them. I changed the names for all of them since I consider "release" to be much too ambiguous but maybe the new names are equally confusing. Happy to give everything different names if smbd comes up with a suggestion

[gforsyth]

quick first pass -- I'll try to find time to dig in a bit deeper

[fjetter]

@jrbourbeau and @gforsyth can you please indicate whether I can expect a review? If you are too busy that's fine. I just would like to know if I am waiting for something to happen. Any further work on the worker is kind of blocked by this PR and I am confident that this improves stability.

[fjetter]

OSX test failure is unrelated and connected to unclosed RPC after startup/teardown of an adaptive cluster

[jrbourbeau]

I began reviewing the changes here yesterday, but didn't have much dedicated time to focus on this today. If you're confident in the changes here, then feel free to merge them in. Otherwise, I will have time to review tomorrow

[mrocklin]

A couple of minor nits, and one genuine question :)

[mrocklin]

I'm curious about why there should be multiple methods for this. In principle I think of our current approach as the scheduler saying

I don't need you to have this data

The worker though, as you say, may disagree and hold on to it. I don't imagine a situation where the scheduler would override the worker, and so I don't see a reason to have multiple methods for this signal.

Thoughts?

[fjetter]

First of all, I think this particular signal I called "superfluous data" should not exist and I believe this is what you're saying as well.
This signal is issued here (link to main branch)

distributed/distributed/scheduler.py

Lines 6276 to 6284 in ddec3c9

	if ts is not None and ts._state == "memory":
	if ts not in ws._has_what:
	ws._nbytes += ts.get_nbytes()
	ws._has_what[ts] = None
	ts._who_has.add(ws)
	else:
	self.worker_send(
	worker, {"op": "delete-data", "keys": [key], "report": False}
	)

I am actually a bit fuzzy myself how this particular race condition occurs but I think something like the following causes otherwise a race condition which results in a deadlock

Task A (Worker A) -->  Task C (Worker B)
                           - ^
                        /        
                      /
Task B (Worker B)

(pardon aweful ascii art. C depends on A and B)

• Worker B tries to execute C but needs to fetch Task A to do so
• Worker A submits data to B
• Data arrives at B and is being deserialized or is waiting for a loop, whatever
• Worker A dies and the scheduler notices
• Scheduler thinks all is lost and transitions Task A to released and tries to reschedule on Worker B
• ... In the meantime Worker B is done deserializing and puts key in memory, notifies scheduler usign add-keys
• Worker B receives the compute-task signal from scheduler and thinks he's lucky since he got the data already, cool. He tells the schdeuler -> task-finished signal
• Scheduler receives the add-keys response but thinks "This task is not in memory any more, please forget it" -> delete-data signal (what I call superfluous data)
• Worker B receives the delete-data signal and forgets everything (this is where my superfluous-data signal has a safeguard)
• Scheduler receives the task-finished signal from before and thinks everything is OK and proceeds as before, however, worker B does not have the data.

I assume this only is a problem due to another bug in how the worker corrects missing data (which I am also trying to fix) which would eventually heal the cluster but I'm not entirely certain

[mrocklin]

Ha, that's a fun comedy of errors.

Yeah, so I agree that this feels like a different bug. In principle it'd be nice to fix that bug rather than add this signal, but I can imagine that it might be best to do that in future work.

[fjetter]

This is a test case for the last deadlock we were made aware of. It requires a bit of patching since it is otherwise not possible to stimulate the same-host with multiple who-has edge case this is hitting

This happens if there are multiple who_has such that the scheduler doesn't retrigger a clean computation but the worker always pick the same erroneous worker. If the worker doesn't forget about that wrong worker after first try, we're running in a loop.

distributed/distributed/worker.py

Lines 2089 to 2093 in ddec3c9

	local = [w for w in workers if get_address_host(w) == host]
	if local:
	worker = random.choice(local)
	else:
	worker = random.choice(list(workers))

[fjetter]

This test is rather close to the implementation since we somehow need to wait for worker B to actually start deserializing and we do not have any sane hooks exposing this at the moment, therefore I'm watching the connection pool to make this work.
see #4784 (comment)

This test proves the race condition (replace handle-superfluous-data with handle-free-keys and it is stuck) but I couldn't prove that we require any handler like this at all. I kept this path to preserver "previous functionality" and added a really stupid test below which constructs the situation without any actual work, just to have the code path covered. I will try further to reconstruct a case for this but I don't want this to block the PR (see test_forget_data_not_supposed_to_have)

[fjetter]

Win-py38: #4859
osx-py37: #4905

[fjetter]

Since the test failures appear to be unrelated, I will go ahead and merge this now. Feel free to leave review comments afterwards, I'll address everything that comes up.