Handling workers with expiring allocation requests

[wgustafson]

I am trying to figure out how to handle the case of dask workers getting bumped from a cluster due to their requested allocation time expiring. From the intro YouTube video at https://www.youtube.com/watch?v=FXsgmwpRExM, it sounds like dask-jobqueue should detect when a worker expires and automatically start a replacement, which is what I want. However, my testing on DOE's edison computer at NERSC is not getting that behavior. If it matters, edison uses SLURM.

I have tried setting up my cluster two ways and both behave the same. I start a worker that uses dask.delayed to do a bunch of embarrassingly parallel tasks, the server spawns one worker, that worker does the first task or two, the worker expires, the server seems to hang, and nothing else happens.

The first approach I used to setup the cluster was with "scale":

    cluster = SLURMCluster(cores=1, processes=1)  # need all the memory for one task
    cluster.scale(1)  # testing with as simple as I can get, cycling 1 worker
    client = Client(cluster, timeout='45s')

@josephhardinee suggested a 2nd approach using "adapt" instead:

    cluster = SLURMCluster(cores=1, processes=1)  # need all the memory for one task
    cluster.adapt(minimum=1, maximum=1)  # trying adapt instead of scale
    client = Client(cluster, timeout='45s')

The dask-worker.err log concludes with:

slurmstepd: error: *** JOB 10234215 ON nid01242 CANCELLED AT 2018-08-10T13:25:30 DUE TO TIME LIMIT ***
distributed.dask_worker - INFO - Exiting on signal 15
distributed.nanny - INFO - Closing Nanny at 'tcp://10.128.4.227:35634'
distributed.dask_worker - INFO - Exiting on signal 15
distributed.dask_worker - INFO - End worker
distributed.process - WARNING - reaping stray process <ForkServerProcess(ForkServerProcess-1, started daemon)>

Am I expecting more from dask-jobqueue than I should? Or, is this a bug in my implementation or in dask.distributed of dask-jobqueue?

Thanks,
Bill

[guillaumeeb]

Thanks for this really interesting test! I haven't encounter this yet, as I am currently only using dask for interactive analysis, and usually computations I launch have enough workers to end in a few minutes maximum.

However, I believe that, at least with adaptive cluster, for long running processes, workers should be restarted when allocation time expires. I will certainly need this in the near future. Currently, this is clearly not the case for non adaptive cluster. I cannot see anything in the code that would restart workers when a job ends. For adaptive, I don't know enough dask distributed code base yet, but the answer is probably there. Hopefully @mrocklin can answer this more easily.

I will try to reproduce your use case in my system newt week.

In any case, if this is not a bug of adaptive or our use of it, I think we should propose a solution here for having workers automatically restarted. Any advice @jhamman @mrocklin @lesteve ?

[mrocklin]

Am I expecting more from dask-jobqueue than I should?

What you're describing seems like it's in scope for dask-jobqueue, but it's not currently implemented. This could be resolved by recording a set desired number of workers and re-scaling whenever a worker goes away, which could probably be handled by the remove_worker method of the JobQueuePlugin(SchedulerPlugin) class.

[guillaumeeb]

but it's not currently implemented

So I take it adaptive clusters in distributed doesn't handle this either? Shouldn't it?

which could probably be handled by the remove_worker method of the JobQueuePlugin(SchedulerPlugin)

I had that in mind too, I'd be happy to work on it.

[mrocklin]

The adaptive class checks the scheduler state frequently, so I would expect it to handle this case decently well.

…

On Mon, Aug 13, 2018 at 1:24 AM, Guillaume EB ***@***.***> wrote: but it's not currently implemented So I take it adaptive clusters in distributed doesn't handle this either? Shouldn't it? which could probably be handled by the remove_worker method of the JobQueuePlugin(SchedulerPlugin) I had that in mind too, I'd be happy to work on it. — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#122 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AASszNCD4cEK_JyqPxvWtgnFaEaAmNO8ks5uQQ2fgaJpZM4V44B5> .

[mrocklin]

I had that in mind too, I'd be happy to work on it.

Hooray!

…

On Mon, Aug 13, 2018 at 1:24 AM, Guillaume EB ***@***.***> wrote: but it's not currently implemented So I take it adaptive clusters in distributed doesn't handle this either? Shouldn't it? which could probably be handled by the remove_worker method of the JobQueuePlugin(SchedulerPlugin) I had that in mind too, I'd be happy to work on it. — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#122 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AASszNCD4cEK_JyqPxvWtgnFaEaAmNO8ks5uQQ2fgaJpZM4V44B5> .

[guillaumeeb]

So i've tried that with and without adaptive. As expected, without adaptive, once the worker is killed, nothing happens anymore.

With adaptive scaling, in my setup, a worker is restarted every time the previous one dies. Cool, that's what I want.

However, I'm still running into a difficult issue. Here is the code I use for some background:

cluster.adapt(minimum=1, maximum=1)
client = Client(cluster)
import time
def my_slow_task(i):
    time.sleep(10)
    return i**2
results = client.gather(client.map(my_slow_task, range(10)))

The problem is, the worker is keeping all results into its memory for sending them all at once. But it gets killed half way every time. So we end up after several tries with the following exception:

---------------------------------------------------------------------------
KilledWorker                              Traceback (most recent call last)
<ipython-input-6-69e1e4e9d452> in <module>()
      1 futures = client.map(my_slow_task, range(10))
----> 2 results = client.gather(futures)
      3 results

/work/ADM/hpc/eynardbg/Outils/miniconda3/envs/dask_dev/lib/python3.5/site-packages/distributed/client.py in gather(self, futures, errors, maxsize, direct, asynchronous)
   1560             return self.sync(self._gather, futures, errors=errors,
   1561                              direct=direct, local_worker=local_worker,
-> 1562                              asynchronous=asynchronous)
   1563 
   1564     @gen.coroutine

/work/ADM/hpc/eynardbg/Outils/miniconda3/envs/dask_dev/lib/python3.5/site-packages/distributed/client.py in sync(self, func, *args, **kwargs)
    650             return future
    651         else:
--> 652             return sync(self.loop, func, *args, **kwargs)
    653 
    654     def __repr__(self):

/work/ADM/hpc/eynardbg/Outils/miniconda3/envs/dask_dev/lib/python3.5/site-packages/distributed/utils.py in sync(loop, func, *args, **kwargs)
    273             e.wait(10)
    274     if error[0]:
--> 275         six.reraise(*error[0])
    276     else:
    277         return result[0]

/work/ADM/hpc/eynardbg/Outils/miniconda3/envs/dask_dev/lib/python3.5/site-packages/six.py in reraise(tp, value, tb)
    691             if value.__traceback__ is not tb:
    692                 raise value.with_traceback(tb)
--> 693             raise value
    694         finally:
    695             value = None

/work/ADM/hpc/eynardbg/Outils/miniconda3/envs/dask_dev/lib/python3.5/site-packages/distributed/utils.py in f()
    258             yield gen.moment
    259             thread_state.asynchronous = True
--> 260             result[0] = yield make_coro()
    261         except Exception as exc:
    262             error[0] = sys.exc_info()

/work/ADM/hpc/eynardbg/Outils/miniconda3/envs/dask_dev/lib/python3.5/site-packages/tornado/gen.py in run(self)
   1131 
   1132                     try:
-> 1133                         value = future.result()
   1134                     except Exception:
   1135                         self.had_exception = True

/work/ADM/hpc/eynardbg/Outils/miniconda3/envs/dask_dev/lib/python3.5/asyncio/futures.py in result(self)
    292             self._tb_logger = None
    293         if self._exception is not None:
--> 294             raise self._exception
    295         return self._result
    296 

/work/ADM/hpc/eynardbg/Outils/miniconda3/envs/dask_dev/lib/python3.5/site-packages/tornado/gen.py in run(self)
   1139                     if exc_info is not None:
   1140                         try:
-> 1141                             yielded = self.gen.throw(*exc_info)
   1142                         finally:
   1143                             # Break up a reference to itself

/work/ADM/hpc/eynardbg/Outils/miniconda3/envs/dask_dev/lib/python3.5/site-packages/distributed/client.py in _gather(self, futures, errors, direct, local_worker)
   1437                             six.reraise(type(exception),
   1438                                         exception,
-> 1439                                         traceback)
   1440                     if errors == 'skip':
   1441                         bad_keys.add(key)

/work/ADM/hpc/eynardbg/Outils/miniconda3/envs/dask_dev/lib/python3.5/site-packages/six.py in reraise(tp, value, tb)
    691             if value.__traceback__ is not tb:
    692                 raise value.with_traceback(tb)
--> 693             raise value
    694         finally:
    695             value = None

KilledWorker: ('my_slow_task-f7d3b870b0218db300f66ed70fe3fba2', 'tcp://10.135.36.29:44953')

[mrocklin]

Ideally you would have a few workers that died at different times. In this way a "herd" of workers could keep the data alive, rather than having them all die at the same time. You might also consider gathering results as they complete with as_completed

…

On Mon, Aug 13, 2018 at 8:31 AM, Guillaume EB ***@***.***> wrote: So i've tried that with and without adaptive. As expected, without adaptive, once the worker is killed, nothing happens anymore. With adaptive scaling, in my setup, a worker is restarted every time the previous one dies. Cool, that's what I want. However, I'm still running into a difficult issue. Here is the code I use for some background: cluster.adapt(minimum=1, maximum=1) client = Client(cluster)import timedef my_slow_task(i): time.sleep(10) return i**2 results = client.gather(client.map(my_slow_task, range(10))) The problem is, the worker is keeping all results into its memory for sending them all at once. But it gets killed half way every time. So we end up after several tries with the following exception: --------------------------------------------------------------------------- KilledWorker Traceback (most recent call last)<ipython-input-6-69e1e4e9d452> in <module>() 1 futures = client.map(my_slow_task, range(10))----> 2 results = client.gather(futures) 3 results /work/ADM/hpc/eynardbg/Outils/miniconda3/envs/dask_dev/lib/python3.5/site-packages/distributed/client.py in gather(self, futures, errors, maxsize, direct, asynchronous) 1560 return self.sync(self._gather, futures, errors=errors, 1561 direct=direct, local_worker=local_worker,-> 1562 asynchronous=asynchronous) 1563 1564 @gen.coroutine /work/ADM/hpc/eynardbg/Outils/miniconda3/envs/dask_dev/lib/python3.5/site-packages/distributed/client.py in sync(self, func, *args, **kwargs) 650 return future 651 else:--> 652 return sync(self.loop, func, *args, **kwargs) 653 654 def __repr__(self): /work/ADM/hpc/eynardbg/Outils/miniconda3/envs/dask_dev/lib/python3.5/site-packages/distributed/utils.py in sync(loop, func, *args, **kwargs) 273 e.wait(10) 274 if error[0]:--> 275 six.reraise(*error[0]) 276 else: 277 return result[0] /work/ADM/hpc/eynardbg/Outils/miniconda3/envs/dask_dev/lib/python3.5/site-packages/six.py in reraise(tp, value, tb) 691 if value.__traceback__ is not tb: 692 raise value.with_traceback(tb)--> 693 raise value 694 finally: 695 value = None /work/ADM/hpc/eynardbg/Outils/miniconda3/envs/dask_dev/lib/python3.5/site-packages/distributed/utils.py in f() 258 yield gen.moment 259 thread_state.asynchronous = True--> 260 result[0] = yield make_coro() 261 except Exception as exc: 262 error[0] = sys.exc_info() /work/ADM/hpc/eynardbg/Outils/miniconda3/envs/dask_dev/lib/python3.5/site-packages/tornado/gen.py in run(self) 1131 1132 try:-> 1133 value = future.result() 1134 except Exception: 1135 self.had_exception = True /work/ADM/hpc/eynardbg/Outils/miniconda3/envs/dask_dev/lib/python3.5/asyncio/futures.py in result(self) 292 self._tb_logger = None 293 if self._exception is not None:--> 294 raise self._exception 295 return self._result 296 /work/ADM/hpc/eynardbg/Outils/miniconda3/envs/dask_dev/lib/python3.5/site-packages/tornado/gen.py in run(self) 1139 if exc_info is not None: 1140 try:-> 1141 yielded = self.gen.throw(*exc_info) 1142 finally: 1143 # Break up a reference to itself /work/ADM/hpc/eynardbg/Outils/miniconda3/envs/dask_dev/lib/python3.5/site-packages/distributed/client.py in _gather(self, futures, errors, direct, local_worker) 1437 six.reraise(type(exception), 1438 exception,-> 1439 traceback) 1440 if errors == 'skip': 1441 bad_keys.add(key) /work/ADM/hpc/eynardbg/Outils/miniconda3/envs/dask_dev/lib/python3.5/site-packages/six.py in reraise(tp, value, tb) 691 if value.__traceback__ is not tb: 692 raise value.with_traceback(tb)--> 693 raise value 694 finally: 695 value = None KilledWorker: ('my_slow_task-f7d3b870b0218db300f66ed70fe3fba2', 'tcp://10.135.36.29:44953') — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#122 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AASszLkYEVwI0ORIAnuFnto5twZBKvWBks5uQXGegaJpZM4V44B5> .

[josephhardinee]

Out of curiosity, how does that work? I assume when a new worker needs to be started, a new jobscript has to be submitted. I assume that would need to be handled by jobqueue and not distributed?

[guillaumeeb]

I don't exactly know how adaptive cluster is implemented in distributed, but the idea is that something is watching for dask cluster load, and calling scale method accordingly, for example if there are a lot more pending tasks than available workers.

Considering the issue here:

• adaptive is already doing the first identified need: insuring there is always a minimum number of workers alive. Is it still worth implementing that as a dask-jonqueue default?
• Second thing is that users should not submit graphs that cannot finish within the job allocated time and the requested number of workers. Submitting several graphs should be fine though (so doing batches if needed). Is this correct? Should we add this in the docs somehow?

[mrocklin]

FWIW the adaptive implementation is only 300 lines of code. It might be a good evening read if anyone is interested:

https://github.com/dask/distributed/blob/master/distributed/deploy/adaptive.py

adaptive is already doing the first identified need: insuring there is always a minimum number of workers alive. Is it still worth implementing that as a dask-jonqueue default?

I don't know. @jhamman might have thoughts. Alternatively maybe there is some way to change Adaptive to maintain a set target rather than move that target around.

Second thing is that users should not submit graphs that cannot finish within the job allocated time and the requested number of workers. Submitting several graphs should be fine though (so doing batches if needed). Is this correct? Should we add this in the docs somehow?

It should be possible to run a single large graph with an evolving herd of workers, even if the workers don't live very long.

[jhamman]

Alternatively maybe there is some way to change Adaptive to maintain a set target rather than move that target around.

I think this is necessary to do this correctly. Ideally we could have a Cluster.target_workers property. This property would be updated by both Cluster.scale() and Adaptive. My recent attempts to improve the adaptive scaling in jobqueue in #97 have highlighted the need for this.

Another thing that would be worth discussing is the ability for Adaptive to query the Cluster for pending workers. This obviously comes up in jobqueue when there are jobs in a queue but I image this will also be useful to other dask deploy managers (dask-kubernetes, etc.).

[wgustafson]

Sounds like you have a pretty good idea of my workflow scenario and some initial ideas on what might be needed to enable it to work predictably. Thanks for looking into this.

From @guillaumeeb it looks like there might be something going on with NERSC's edison computer compared to his cluster since I was not seeing any restarts, whereas he was getting restarts but with data passing issues. If you would like me to run any tests on edison, let me know.

[guillaumeeb]

@wgustafson, it could still be useful if you could give us some simplified version of the code your trying to run, just to be sure.

For information, I've performed my tests with a dask-jobqueue installed from one of the latest commit in master branch, were you using this to, or did you just pip (or conda) installed the latest released package?

@jhamman just to be sure, appart from fixing some edge case issues detected with #97, you believe it will be a good idea to have automatic restart of finished jobs by default?

It should be possible to run a single large graph with an evolving herd of workers, even if the workers don't live very long.

@mrocklin from my recent experiment here, I cannot see how it could be possible for now. Are in memory data backed up or transmitted in some way between workers? The simple workflow of submitting a group of independent tasks seems to show that you need to have at least workers that last enough time to process nb_tasks / nb_worker tasks. But I did not try with several workers, so this is just a guess.

[jhamman]

@jhamman just to be sure, appart from fixing some edge case issues detected with #97, you believe it will be a good idea to have automatic restart of finished jobs by default?

I was thinking this would only work when using Adaptive. I think a non-adaptive job that dies due to a wall-clock expiration should stay dead.

[guillaumeeb]

Yep, I agree with you @jhamman, no reason to implement a more complex default behavior when most of the time this is not needed (and probably not what we want), and also when there is already a complex mechanism that answers the need.

So this works right now with Adaptive, and you've also opened #112 to fix the edge cases.

Another thing that would be worth discussing is the ability for Adaptive to query the Cluster for pending workers.

I let you open another issue for that part, and if you can provide some advantages it would have.

Now @wgustafson, lets try to solve your adaptive problem, and also to see if your workflow is compatible with dying and restarting workers.