[Feature] add support for retrying failed tasks

[amosonn]

When running on a heterogeneous cluster, sometimes a task fails not because of a fundamental error in the function, but because of node-specific problems, such as low memory.
For such tasks, it would be useful to be able to pass the executor a "retry_count=n" kwarg, which will mean that if the task fails, the scheduler will attempt to rerun it (pref. on a different worker). Possibly, with a list of exceptions which are "expected" in such a case.

[mrocklin]

This makes sense and can be done. It involves adding more state to the scheduler to track which exceptions can fail. I'm always a bit hesitant to add more things to track (the current list is quite long) but this seems reasonable and nice to have.

@amosonn can you give more information about your use case? Mostly I'm trying to verify that this is something that people actually need to solve real problems vs something that people think might be a good idea.

[amosonn]

This is a very concrete problem we have, and we were considering the alternatives of resubmitting failed jobs at our code vs. adding this to the scheduler.

First of all it should be noted that the cluster we are using is an in-house cluster at a university. It has many nodes of different cpu count and memory sizes. Therefore, it is reasonable to assume that even if a weaker node dropped a job, a stronger one will be able to pull it through.
We have two different types of job failures:

1. We use an external, compiled commandline tool to do the heavy-lifting of some of our computations, wrapped around in python logic. This tool does not have very pleasant API, so on some jobs it allocates too much memory and dies. We use plumbum to run it, so we get a decent exception, but it is not so easy to estimate in advance whether the node will have enough memory.
2. We let jobs interact directly with our database, via Django ORM. This requires taking some care when starting the workers, but we managed to get a setup which works well. Some jobs are very long, populating thousands of objects, and the database connection eventually times-out. Again, it can be that a stronger node will be able to complete the job before the timeout.

Now, the second use-case is a bit more esoteric, but the first one seems to me to be relevant to more people.

[mrocklin]

Do you know which jobs are likely to require more resources? Do you know which computers have more resources? If the answer to both questions is "yes" then have you considered targeting resource heavy jobs to heavy workers using the workers= keyword argument to submit or map? This seems like a cleaner solution to me than retrying a job until it finds a good worker.

http://distributed.readthedocs.io/en/latest/locality.html#user-control

[amosonn]

Yes, I saw this feature. Estimating which jobs are heavier is not fool-proof but reasonable, the problem is we don't have a good API for getting the node strength. Perhaps for RAM we could use the distributed API to ask how much RAM a worker has. As for CPU, that's probably more complicated.

However, I think that even if you take precautions, sometimes a job might fail, and rerunning it might work. Perhaps a better option than the "retry" kwarg is an API for rerunning a future, something like fut.retry() or fut.clear(), which clears the result (or exception) of this future from the who_has diagram, returning it to the the pool of available jobs.

[amosonn]

I've looked a bit through the code, and it seems that the scheduler keeps the tasks while they're being run, so that adding the retry will be mostly in mark_task_erred, where if the task still has retries it is marked as free instead of as failed.
Conversely, with the fut.clear() API, we would probably need a new method such as mark_task_clean, with an outward-facing handler.

[mrocklin]

Estimating which jobs are heavier is not fool-proof but reasonable, the problem is we don't have a good API for getting the node strength. Perhaps for RAM we could use the distributed API to ask how much RAM a worker has. As for CPU, that's probably more complicated

Dask.distributed does track this information internally for all workers. We gather information using the psutil module. We could expose the information we gather pretty easily, either through a TCP route (how the Executor tends to interact) or through the bundled HTTP server. Alternatively one could also do this today just by using e.run and psutil together. Something like the following:

In [1]: from distributed import Executor

In [2]: e = Executor('localhost:8786')

In [3]: e
Out[3]: <Executor: scheduler="localhost:8786" processes=2 cores=16>

In [4]: import psutil

In [5]: e.run(psutil.virtual_memory)
Out[5]: 
{'127.0.0.1:33402': svmem(total=16701911040, available=14316855296, percent=14.3, used=4636807168, free=12065103872, active=2451337216, inactive=1854357504, buffers=144670720, cached=2107080704),
 '127.0.0.1:36003': svmem(total=16701911040, available=14316855296, percent=14.3, used=4636807168, free=12065103872, active=2451337216, inactive=1854357504, buffers=144670720, cached=2107080704)}

[mrocklin]

However, both explicitly rerunning failed jobs and having an auto-retry do sound generally valuable. I'm not against this idea, just exploring ways around it.

It sounds like you've been digging through the scheduler code a bit. Is this something that you're interested in trying to implement? It'd be good to get some more people active within the scheduler's internals.

[amosonn]

Yeah, I could give it a shot. It seems to me that though used for the same purpose, they are quite different and independent features. I'll deliberate a bit and choose which one to start with; my tendency is to choose rerunning failed jobs, which is the more flexible tool of the two.

Do you have any words of wisdom on a preferable approach, or some coding guidelines?

[amosonn]

It seems this is off the chart for our priorities here at work, so I have to let this one go for now. Sorry. Maybe I'll get to it sometime in the future; or someone else might get interested and pick it up.

[robinsonkwame]

@mrocklin I'm considering the same issue here and you may find tenacity as an appropriate library for retrying from your code base vs. a dask PR

[mrocklin]

This is done with the retries= keyword

[mrocklin]

Whoops, wrong issue

[mrocklin]

Closing this. We allow providing a retries= keyword. I'll open another issue for retrying after-the-fact.