Repository navigation
Resilience #1072
Description
Are there any plans for a highly available distributed cluster?
How much effort would it take (or is it possible at all) to:
- periodically persist the state of the scheduler for failover?
- or use a persistent backend instead of memory?
- have standby schedulers (via leader election etc.)?
Reacted by Huzefa Siyamwala, Guilherme Pires, Justin Poehnelt, Clark Zinzow and Daniel Davis
Activity
mrocklin commentedon May 10, 2017
I don't have any such plans. Of course if you feel up to this you're welcome to do so.
I suspect that all of these would take a non-trivial amount of effort.
Edit
Responding to your options:
Periodically persist the state of the scheduler for failover
Dumping data to disk doesn't sound terribly difficult.
Use a persistent backend instead of memory
My intuition says that this would be exceedingly slow. We operate in memory for a reason
Standby schedulers
You would need to keep scheduler state updated, this is feasible but probably challenging
My preference
If I were to do this myself (which I have no short-term plans of doing) I would instead have all of the workers and clients update a new scheduler with their state when the new scheduler comes online. This would require the clients to keep track of the graphs that they have sent so far. It would also require some non-trivial manipulations of the scheduler state.
Avoid the question entirely
Many people have asked about this question, but it's not clear that they need full resilience of the scheduler. Many times their needs are entirely satisfied by coupling Dask with a system like Kubernetes or Marathon that keeps a scheduler up, but doesn't guarantee persistence of fine-grained state. In the event of a failure people have to resubmit their computations, which, in some cases, they may be ok with.
mrocklin commentedon May 11, 2017
To be more clear (I have slightly more time now)
Your guess is as good as mine. Don't forget that you'd also have to ensure that workers respected this state at least somewhat on recovery.
I don't know of any persistent system that operates fast enough for this
Having a standby scheduler isn't hard. Maintaining updates might be hard.
My previous thought on how to solve this problem was to have everyone re-submit what they know about the state of things if the scheduler goes down and comes up again. Clients resubmit computations, workers check in with what they have. I don't know if this is best though.
Would love to see work here. I'm personally unlikely to work on this in the near future without serious enticement.
kszucs commentedon May 11, 2017
This is a good read: http://mesos.apache.org/documentation/latest/high-availability-framework-guide/
If I understand correctly the mesos agents (workers in our case) recovers their state to the newly elected master.
Do workers maintain such state?
mrocklin commentedon May 11, 2017
[-][RFC] Resilience improvements[/-][+]Resilience[/+]pitrou commentedon Nov 2, 2017
A possible way to approach this would be to have the scheduler dump its state to workers periodically (perhaps choosing a small number of different workers each time, to reduce the overall load). Then when a new scheduler instance goes up, reconnecting workers can announce they have some saved state of an old scheduler (e.g. tagged with a timestamp and a unique id). If the scheduler recognizes itself, it asks the worker with the most recent timestamp and asks to receive the saved state.
pitrou commentedon Nov 2, 2017
@mrocklin do you know how many tasks can the scheduler of a large cluster typically keep track of at any given time?
mrocklin commentedon Nov 3, 2017
How would we handle state created/destroyed in the meantime?
If clients maintain their submitted task graphs then I think that we have a full copy of the scheduler state already, and with full accuracy and timeliness.
mrocklin commentedon Nov 3, 2017
This varies highly, ranging from hundreds to millions of tasks.
pitrou commentedon Nov 3, 2017
I was going to say "we don't"... but I was only thinking about created state, not destroyed. If the scheduler doesn't know that some keys were released in the meantime, there's a memory leak :-/
The clients and workers, I assume? But is it important to keep, say, dependencies or restrictions information for finished tasks (in the "memory" or "erred" state...)? Perhaps an explicit retry functionality would need it, but otherwise?
pitrou commentedon Nov 3, 2017
I think in the millions case, the idea of dumping the whole scheduler state somewhere becomes impractical. It sounds plausible that the serialized dump will have from 100 to 1000 bytes per key, so that would mean hundreds of megabytes to ship to a worker (or write to disk on a shared location). Perhaps in common cases it would be smaller (because pickle is quite good at compressing when a given object, e.g. string key, appears a lot of times), but still.
mrocklin commentedon Nov 3, 2017
FWIW I don't have much confidence in the periodic-backup approach. We would also need to keep some sort of short term real-time log, which probably isn't feasible for us for performance reasons (and it'd be a hassle to move it around in case of failure).
martindurant commentedon Nov 3, 2017
How about the concept of user-initiated checkpointing? It can then take a while, saving everything to disc in a thread, but happen only rarely.
mrocklin commentedon Nov 3, 2017
There are many sensible approaches. I think that maintaining state on the client is simpler, automatic, and fits nicely with the already existing infrastructure.
Dracio commentedon Sep 25, 2018
Is there any plans to implement resilience if a scheduler reboots? I was hoping to use dask, but my tasks can range from 1 min to several hours.
I like this idea:
If I were to do this myself (which I have no short-term plans of doing) I would instead have all of the workers and clients update a new scheduler with their state when the new scheduler comes online. This would require the clients to keep track of the graphs that they have sent so far. It would also require some non-trivial manipulations of the scheduler state.
mrocklin commentedon Sep 25, 2018
The plan is the same as what is above. No one has implemented this, nor do I think that anyone has any plans to implement this. This could be accelerated if someone volunteers to do this work, or if an institution chooses to fund this work.
Do you expect your scheduler to go down in this period? Schedulers can live indefinitely if nothing else kills them.
Dracio commentedon Sep 25, 2018
I was planning to run it indefinitely, but when if the scheduler goes down (machine dies), it would be nice for the scheduler to see if the workers are already executing a task, so it does try to send another task.
mrocklin commentedon Sep 25, 2018
Rev 01 for Issue dask#1072
PatrikZhong commentedon Mar 2, 2023
I’ve been looking to prototype a replicated/highly-available scheduler with a primary scheduler and a few backups. The idea is that the primary continuously replicate its state to the backups so that in case the primary crashes, a backup scheduler can take over without having to restart the computations. My main question is what data structures from the scheduler-side would be needed to be replicated in order for backups to know which executions have happened in the task graph?
Looking into the source code for the scheduler and the API docs, the scheduler contains the states of all tasks in the “tasks dict”, and a log of all transitions that have happened can be found in the transition log. Are there any other data structures that would be necessary to keep a track of to keep the other schedulers consistent?
fjetter commentedon Mar 3, 2023
FWIW I doubt that simply replicating the scheduler state somewhere will be sufficient to ensure you can continue without recomputation unless you have a perfect copy , e.g. without any delays. You'll need some kind of state reconciliation that includes the state of all still-alive workers to account for the state drift and any lost messages while the first scheduler was down.
Note: Right now, workers are hard wired to shut down in case the scheduler dies. We actually removed the possibilities for workers to reconnect a while ago because we were struggling to correct this state drift properly (even, or particularly, when reconnecting to the same instance). See #6361 Reintroducing the reconnect is very easy iff state drift can be corrected properly.
Concretely, you'll likely need everything in the class
SchedulerStateand additionally even a couple of "extensions" depending what kind of features should survive this (e.g. WorkStealing, P2PShuffling)A word of caution: These low level internals are subject to frequent change and stability varies.
PatrikZhong commentedon Mar 8, 2023
Assuming that we have managed to replicate a consistent state of the scheduler, how can we use it to restart a new scheduler with that state (to take over if the primary scheduler fails)?
As I understand it, extracting the state from the scheduler can be done by simply interacting with the Scheduler object and extracting its attributes, but giving the scheduler an existing state seems trickier, as there's no obvious entry point in the API or in the source code.