Resilience

[kszucs]

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.)?

[mrocklin]

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:

1. Periodically persist the state of the scheduler for failover

   Dumping data to disk doesn't sound terribly difficult.

2. Use a persistent backend instead of memory

   My intuition says that this would be exceedingly slow. We operate in memory for a reason

3. 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]

To be more clear (I have slightly more time now)

periodically persist the state of the scheduler for failover?

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.

or use a persistent backend instead of memory?

I don't know of any persistent system that operates fast enough for this

have standby schedulers (via leader election etc.)?

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]

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]

See http://distributed.readthedocs.io/en/latest/worker.html#api-documentation for worker state

…

On Thu, May 11, 2017 at 4:42 AM, Krisztián Szűcs ***@***.***> wrote: 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? — You are receiving this because you commented. Reply to this email directly, view it on GitHub <#1072 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AASszPCMCXTXBEXsWIRxCTylC2JYQAmaks5r4snigaJpZM4NW5-_> .

[pitrou]

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]

@mrocklin do you know how many tasks can the scheduler of a large cluster typically keep track of at any given time?

[mrocklin]

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.

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]

@mrocklin do you know how many tasks can the scheduler of a large cluster typically keep track of at any given time?

This varies highly, ranging from hundreds to millions of tasks.

[pitrou]

How would we handle state created/destroyed in the meantime?

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 :-/

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.

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]

This varies highly, ranging from hundreds to millions of tasks.

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]

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 :-/

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]

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]

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]

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.