Repository navigation
Tasks are not stolen if worker runs out of memory #3761
Description
Executive summary
The original intent around the design of worker pause was to deal with a spill disk drive that is much slower than the tasks that produce managed memory.
However, if a worker reaches the paused state either because of unmanaged memory alone or because it runs out of spill disk space, it will never get out of it.
Any tasks that are queued on the worker will remain there waiting forever, causing whole computations to hang.
This issue requires the pause on the worker to trigger all queued tasks on the scheduler (as opposed to currently running) to be immediately sent back to the scheduler so that they can be computed somewhere else.
Original ticket
distributed git tip
Python 3.8
Linux x64
16 cores, 32 threads CPU
I'm using a dask distributed cluster running on localhost to perform a leaf-to-root aggregation of a large hierarchical tree with ~420k nodes, each node being a dask.delayed function that partially blocks the GIL for 1-100ms. The client retrieves the very small output of the root node.
My dask workers are running with 1 thread per worker.
My computation is getting randomly stuck. Some things I noticed:
- the issue gets substantially less frequent if I optimize the graph with fuse with an extremely aggressive ave_width=50, thus reducing the number of nodes to ~27k
- it gets even less frequent, but doesn't disappear completely, if I pass fuse_subgraphs=True to fuse. This leads me to strongly suspect that the time it takes to pickle, unpickle, or traverse each node makes a difference.
- it gets substantially worse if I feed multiple copies of the problem (with different, randomized keys) to the scheduler in parallel. E.g. if I feed 10 copies of the tree in parallel I'm pretty much guaranteed I'll experience a deadlock.
- if I send SIGTERM to one worker at a time, it eventually gets unstuck.
- using another client to push a task to explicitly singled out workers gets stuck forever.
I have a high degree of confidence that there are no potential race conditions in my application code.
I have spent a considerable amount of effort trying to reproduce the issue in a POC and failed.
I have no idea how to debug it and I'm open to suggestions...
Activity
crusaderky commentedon May 1, 2020
It's stuck again right now. The progress bar says 11 pending tasks out of 27186 (this is downstream of fuse).
Running
Client.has_whatshows me 3-4 keys per worker, all with status: completed, for a total of 123 keys or 61 unique keys.Client.who_hasanalogously returns 27186 keys, of which only 61 are held by 1+ workers.Client.gatheron all of these 61 keys successfully retrieves all values.who_has also shows me that my graph is stuck in a specific branch:
My keys are indicative of the hierarchy; e.g.
mtm.compute_portfolio-PMI-5eabf03e0000000000000003is the parent ofmtm.compute_portfolio-PMI/PMI2013-5eabf03e0000000000000003. You can see how how the fault is underPMI/PMI2013/PMI TRD/PMI TRD MEX/REFINADOS/RESIDUALES/FUEL OIL, which does not appear in the above list.Further:
which highlights that the exact issue is in
mtm.compute_portfolio-PMI/PMI2013/PMI TRD/PMI TRD MEX/REFINADOS/RESIDUALES/FUEL OIL/FUEL OIL 03 2013-5eabf03e0000000000000003.Trying to retrieve its key hangs.
Future.retry() does nothing.
I tried rerunning the specific branch of hierarchy 100 times in parallel and went through without any glitches.
mrocklin commentedon May 1, 2020
Who_has and has_what will always return completed tasks that are in-memory. It doesn't list processing tasks. In your situation I would probably try to ask the workers what they're working on, maybe with something like
If that doesn't return then you know that the worker itself has seized up, which may tell you something. If it returns then maybe the function itself is just not making any progress. Answering this question might help to better isolate the issue.
crusaderky commentedon May 1, 2020
Hi @mrocklin , many thanks for the help.
It looks like the scheduler thinks that a task is running while the worker has no idea about it (different run from before, so the key has changed):
All my functions emit a log line at the beginning and one at the end (also in case of crash). I've identified the function that matches the key above and there's no trace of it in the logs - it never started.
mrocklin commentedon May 1, 2020
crusaderky commentedon May 1, 2020
@mrocklin 3 transitions returned:
DelayedFactory.state is a global state node that is identically fed in input to all nodes.
mrocklin commentedon May 1, 2020
OK, so nothing odd happened on the scheduler (like the key transitioning from processing back to released and then to memory or something).
I would do the same thing on the worker to see if anything is odd there. If that also doesn't show much then I would start looking at logs, and maybe even increase the logging verbosity to debug.
Welcome to debugging scheduling issues. In the early years of Dask these skillsets were really important (things would deadlock a lot). Nowadays very few people have to look at this stuff :)
crusaderky commentedon May 1, 2020
@mrocklin same command on the workers.
mrocklin commentedon May 1, 2020
Interesting. It looks like this task has been moved around to a bunch of different workers. I wonder why. It looks like one of them has the task in a ready-to-compute state. I wonder what else is going on on that worker. Is it busy with something else perhaps? What does
dask_worker.executinglook like? (or was this empty from before). If so, I wonder if giving the worker a kick withdask_worker.ensure_computing()would make things run to completion.crusaderky commentedon May 1, 2020
@mrocklin, the worker is completely idle;
dask_worker.executingreturns an empty set.dask_worker.ensure_computing()has no visible effect. It does not cause any further transitions indask_worker.story().Maybe because my tasks (after fusing) take a very variable amount of time between 100ms and 10s, so the scheduler struggles to balance them up front?
mrocklin commentedon May 1, 2020
That might explain it. To be clear, this should be fine and shouldn't result in a deadlock of any sort. It just was something strange that we saw.
That's odd. The next thing I would do would be to look through the
ensure_computingfunction and see which paths we're going down and why that task isn't transitioning to theexecutingstate.crusaderky commentedon May 1, 2020
distributed/distributed/worker.py
Lines 2430 to 2432 in e7ba316
mrocklin commentedon May 1, 2020
Ah, that'll do it. So the machine is close to out of memory, so it's not doing any other work. However oddly enough none of the other machines are stealing the task. I suspect that this is because stealing tries not to steal from workers that have fewer tasks than threads (presumably they should be running) but it probably isn't taking paused workers into account.
crusaderky commentedon May 1, 2020
Yep, in the logs I found:
Running this twice unblocked the batch:
I was already aware of the memory leak but I don't know yet which library is responsible.
However I expected that workers would just be restarted when they exceed the threshold?
[-]Very large graphs cause worker deadlock[/-][+]Tasks are not stolen if worker runs out of memory[/+]mrocklin commentedon May 1, 2020
There are a variety of thresholds. https://distributed.dask.org/en/latest/worker.html?highlight=pause#memory-management
There are, I think, two solid solutions here:
Short term, you could just turn off the pause behavior for workers
11 remaining items
mrocklin commentedon May 9, 2020
Certainly. But people giving back is the far exception. You two are exceptional in that you have stuck around over time, and are comfortable actually solving the problems that you raise. I'd say you're something like 1-in-50 . For every fifty people that I help guide through something, only one of them ends up being as good of a time investment as either of you. However, to find someone who fixes not only the problems that affect them, but also reviews the work of others is very rare, I'd say something like 1-in-200. These people almost always get hired eventually.
You two are great, and I really appreciate the work that you put in, but even given that work I personally am unable to review things forever, and unfortunately it's hard to convince other people to review in this repository because it's a bit more complicated. I like your idea that "this time investment will pay off" but over the last five years the evidence is entirely to the contrary. Almost no one reviews other peoples' work in this PR, except a couple of folks whose employers I've convinced to make it part of their job. But hey, proving me wrong here is easy, you only need to go and review other peoples' work.
The challenge here is that "contributing code" is also a "tax on reviewers". So people in your position justifiably feel "hey, I'm volunteering time here to fix a bug for everyone, why aren't people appreciative and contributing back by reviewing my solution" which is a reasonable response. Unfortunately it also sets an expectation on reviewers that they work for you indefinitely. The only way I've really found to make this work is to pay people.
Anyway, I spent a few minutes taking a longer glance through the example above. I don't currently understand what it does unfortunately after this look. Given that it says that it's mostly copied from the existing restart method my guess is that this is mostly my fault.
mrocklin commentedon May 9, 2020
Hopefully one of the other maintainers comes around to help?
mrocklin commentedon May 9, 2020
My guess is that this is still the solution.
mrocklin commentedon May 9, 2020
I'm also a bit frazzled right now because I'm trying to make a company that will get money from other companies to help solve this problem long term. If you know anyone that can help with this effort let me know. Until then I'm likely to be pretty burnt out.
crusaderky commentedon May 9, 2020
To be honest, sounds like a typical open source project to me. And most people using dask are literate in coding - which is more than a lot of projects can say...
This said, I am fully appreciative of your lack of capacity to follow issues and PRs.
I am more than happy to contribute, but alas I don't think I can guarantee any kind of consistency in my time investment. My contributions are typically in spikes on whatever library is giving me pain on my daily work. I used to have to invest large amounts of time on xarray, then it became pint, now it's distributed - take a look at the changelogs of those two projects and you'll see the pattern.
mrocklin commentedon May 10, 2020
dhirschfeld commentedon May 10, 2020
I agree, the "conversion rate" is probably very low - especially for a complex project like dask/distributed. I'm not sure what the solution is. Open source sustainability is a hard problem and I don't think anyone has solved it yet. I'm hopeful that things are improving though - and that companies like Coiled are part of the solution 👍
crusaderky commentedon May 11, 2020
Considering how dask is used by (1) universities and (2) large corporations, IMHO you should offer the latter a way to easily throw money at you. I think the current "donate" button is a too open-ended for the corp mindset; a typical manager will want to spend money for consulting hours to fix very specific issues - and expect that, once he does so, the capacity on the consultant's side somehow appears in short time. Not sure how you people are organized on this regard?
mrocklin commentedon May 11, 2020
mrocklin commentedon Jan 19, 2021
A PR that tries to gracefully kill a worker when it is paused without the opportunity to move forward is here: #4221
crusaderky commentedon Jun 14, 2022
Related issues
Impact
This ticket wants to change the cluster behaviour regarding tasks in
waitingorconstrainedstate (i.e. notexecuting) that are sitting on a worker inpausedorclosing_gracefullystatus.Current state of the art
The scheduler does not assign any new tasks to a worker in
pausedorclosing_graceullystatus.The Active Memory Manager (AMM) does not send any
{op: acquire-replicas}events to such workers.Waiting tasks on a closing worker sit there until all in-memory tasks have been replicated somewhere else. If the worker holds many GiBs worth of spilled data, this could potentially take several minutes. When that happens, the scheduler reschedules them somewhere else.
Waiting tasks on a paused worker sit there until the worker emerges from pause. In case of severe memory leak, this may never happen, thus causing a cluster-wide deadlock. #5999 mitigates this issue by shutting down the worker after ~30s.
High level design
When a worker enters
pausedorclosing_gracefullystatus, allwaitingandconstrainedtasks are (almost) immediately sent back to the scheduler so that they can be computed somewhere else.executingtasks remain on the worker.Low level design
There are two ways to implement this.
The first one is through stealing. A
pausedorclosing_gracefullyworker should be treated as a worker which is expected to be busy for >30s.I personally like another option more, as IMHO it's definitely cleaner:
When the worker enters
pausedorclosing_gracefullystate, it sends a{op: reschedule}message for all tasks inwaitingandconstrainedstate to the scheduler and transitions all tasks inwaiting,constrained, andfetchstate tocancelled. This last bit aborts both fetching the dependencies of the pending tasks as well as any pending AMM{op: acquire-replicas}events.This causes the following chain of events:
{op: compute-task}message to other workers for all tasks that were inrunningstate and have now been rescheduledremove-replicasrequest to the paused/closing worker.{op: acquire-replicas}messages to another worker for all tasks that were previously being fetched on request of the AMM (e.g. they were in fetch state on the worker but not a depenendency of another task).