Exception kills all workers/scheduler

[jhamman]

I'm running dask-distributed with xarray using dask-mpi to launch the scheduler/workers on a PBS system (cheyenne). Under certain conditions, when an Exception is raised, it appears dask is killing the workers and all my PBS jobs are killed. Ideally, distributed would handle these failures a bit more gracefully and at a minimum keep the PBS jobs running.

The exit code given to PBS was:

265: SIGKILL (e.g. kill -9 command), possible out-of-memory error

cc @mrocklin

Traceback:

Details

distributed.utils - ERROR - ("('open_dataset-ec23f9cbbe2cf1b92d4d771bb7537415t_mean-dc61bc3a3dff54591f7eee3c1e60aa57', 23, 0, 0)", 'tcp://10.148.13.182:54442')
Traceback (most recent call last):
  File "/glade/u/home/jhamman/anaconda/envs/pangeo/lib/python3.6/site-packages/distributed/utils.py", line 229, in f
    result[0] = yield make_coro()
  File "/glade/u/home/jhamman/anaconda/envs/pangeo/lib/python3.6/site-packages/tornado/gen.py", line 1055, in run
    value = future.result()
  File "/glade/u/home/jhamman/anaconda/envs/pangeo/lib/python3.6/site-packages/tornado/concurrent.py", line 238, in result
    raise_exc_info(self._exc_info)
  File "<string>", line 4, in raise_exc_info
  File "/glade/u/home/jhamman/anaconda/envs/pangeo/lib/python3.6/site-packages/tornado/gen.py", line 1063, in run
    yielded = self.gen.throw(*exc_info)
  File "/glade/u/home/jhamman/anaconda/envs/pangeo/lib/python3.6/site-packages/distributed/client.py", line 1246, in _gather
    traceback)
  File "/glade/u/home/jhamman/anaconda/envs/pangeo/lib/python3.6/site-packages/six.py", line 686, in reraise
    raise value
distributed.scheduler.KilledWorker: ("('open_dataset-ec23f9cbbe2cf1b92d4d771bb7537415t_mean-dc61bc3a3dff54591f7eee3c1e60aa57', 23, 0, 0)", 'tcp://10.148.13.182:54442')
---------------------------------------------------------------------------
KilledWorker                              Traceback (most recent call last)
<timed exec> in <module>()

~/anaconda/envs/pangeo/lib/python3.6/site-packages/xarray/core/dataarray.py in compute(self)
    590         """
    591         new = self.copy(deep=False)
--> 592         return new.load()
    593 
    594     def persist(self):

~/anaconda/envs/pangeo/lib/python3.6/site-packages/xarray/core/dataarray.py in load(self)
    573         working with many file objects on disk.
    574         """
--> 575         ds = self._to_temp_dataset().load()
    576         new = self._from_temp_dataset(ds)
    577         self._variable = new._variable

~/anaconda/envs/pangeo/lib/python3.6/site-packages/xarray/core/dataset.py in load(self)
    461 
    462             # evaluate all the dask arrays simultaneously
--> 463             evaluated_data = da.compute(*lazy_data.values())
    464 
    465             for k, data in zip(lazy_data, evaluated_data):

~/anaconda/envs/pangeo/lib/python3.6/site-packages/dask/base.py in compute(*args, **kwargs)
    204     dsk = collections_to_dsk(variables, optimize_graph, **kwargs)
    205     keys = [var._keys() for var in variables]
--> 206     results = get(dsk, keys, **kwargs)
    207 
    208     results_iter = iter(results)

~/anaconda/envs/pangeo/lib/python3.6/site-packages/distributed/client.py in get(self, dsk, keys, restrictions, loose_restrictions, resources, sync, asynchronous, **kwargs)
   1921         if sync:
   1922             try:
-> 1923                 results = self.gather(packed, asynchronous=asynchronous)
   1924             finally:
   1925                 for f in futures.values():

~/anaconda/envs/pangeo/lib/python3.6/site-packages/distributed/client.py in gather(self, futures, errors, maxsize, direct, asynchronous)
   1366             return self.sync(self._gather, futures, errors=errors,
   1367                              direct=direct, local_worker=local_worker,
-> 1368                              asynchronous=asynchronous)
   1369 
   1370     @gen.coroutine

~/anaconda/envs/pangeo/lib/python3.6/site-packages/distributed/client.py in sync(self, func, *args, **kwargs)
    538             return future
    539         else:
--> 540             return sync(self.loop, func, *args, **kwargs)
    541 
    542     def __str__(self):

~/anaconda/envs/pangeo/lib/python3.6/site-packages/distributed/utils.py in sync(loop, func, *args, **kwargs)
    239         e.wait(1000000)
    240     if error[0]:
--> 241         six.reraise(*error[0])
    242     else:
    243         return result[0]

~/anaconda/envs/pangeo/lib/python3.6/site-packages/six.py in reraise(tp, value, tb)
    684         if value.__traceback__ is not tb:
    685             raise value.with_traceback(tb)
--> 686         raise value
    687 
    688 else:

~/anaconda/envs/pangeo/lib/python3.6/site-packages/distributed/utils.py in f()
    227             yield gen.moment
    228             thread_state.asynchronous = True
--> 229             result[0] = yield make_coro()
    230         except Exception as exc:
    231             logger.exception(exc)

~/anaconda/envs/pangeo/lib/python3.6/site-packages/tornado/gen.py in run(self)
   1053 
   1054                     try:
-> 1055                         value = future.result()
   1056                     except Exception:
   1057                         self.had_exception = True

~/anaconda/envs/pangeo/lib/python3.6/site-packages/tornado/concurrent.py in result(self, timeout)
    236         if self._exc_info is not None:
    237             try:
--> 238                 raise_exc_info(self._exc_info)
    239             finally:
    240                 self = None

~/anaconda/envs/pangeo/lib/python3.6/site-packages/tornado/util.py in raise_exc_info(exc_info)

~/anaconda/envs/pangeo/lib/python3.6/site-packages/tornado/gen.py in run(self)
   1061                     if exc_info is not None:
   1062                         try:
-> 1063                             yielded = self.gen.throw(*exc_info)
   1064                         finally:
   1065                             # Break up a reference to itself

~/anaconda/envs/pangeo/lib/python3.6/site-packages/distributed/client.py in _gather(self, futures, errors, direct, local_worker)
   1244                             six.reraise(type(exception),
   1245                                         exception,
-> 1246                                         traceback)
   1247                     if errors == 'skip':
   1248                         bad_keys.add(key)

~/anaconda/envs/pangeo/lib/python3.6/site-packages/six.py in reraise(tp, value, tb)
    684         if value.__traceback__ is not tb:
    685             raise value.with_traceback(tb)
--> 686         raise value
    687 
    688 else:

KilledWorker: ("('open_dataset-ec23f9cbbe2cf1b92d4d771bb7537415t_mean-dc61bc3a3dff54591f7eee3c1e60aa57', 23, 0, 0)", 'tcp://10.148.13.182:54442')```
</details>

[mrocklin]

Is it possible that you are trying a computation that doesn't fit into memory? Once one worker goes down the others try to pick up the slack and fail themselves.

Normally this problem would be handled by dumping excess data to disk and gracefully degrading performance. Unfortunately we don't have local storage on Cheyenne, so this typically just manifests as a full collapse.

My next step would be to verify that memory is in fact the issue. I would look at the worker logs if they are available. I would also look at the top-left plot of the status page of the diagnostic dashboard.

[jhamman]

@mrocklin - yes. Its does seem that I was persisting a fairly large dataset and ran out of memory. I'm wondering why we need to kill the workers entirely? I'd like to see a MemoryError raised but not a hard exit, like we got here/

[mrocklin]

NumPy only raises a MemoryError if you allocate a huge array. If you repeatedly allocate many small arrays then it will depend on the OS to write excess data in RAM to disk. In our case we don't have local storage, so we are going to fail one way or another. There is no clear place for us to raise a MemoryError.

[mrocklin]

@jhamman would these options suffice to resolve your problem? http://distributed.readthedocs.io/en/latest/worker.html#memory-management

You might consider tweaking those values in your ~/.dask/config.yaml file.

[jhamman]

Question after reading these docs:

At 95% of memory load, terminate and restart the worker

How hard would it be to raise a MemoryError from dask instead of killing the worker?

[mrocklin]

Hard. There is no way to send that error to the user.

…

On Tue, Oct 31, 2017 at 6:59 PM, Joe Hamman ***@***.***> wrote: Question after reading these docs: At 95% of memory load, terminate and restart the worker How hard would it be to raise a MemoryError from dask instead of killing the worker? — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#1469 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AASszEl3MNg98NR-VgytkHZSqo-6oki4ks5sx6ZGgaJpZM4P2Dk3> .

[mrocklin]

This would just raise within a periodic callback on the worker. On Tue, Oct 31, 2017 at 7:02 PM, Matthew Rocklin <mrocklin@anaconda.com> wrote:

…

Hard. There is no way to send that error to the user. On Tue, Oct 31, 2017 at 6:59 PM, Joe Hamman ***@***.***> wrote: > Question after reading these docs: > > At 95% of memory load, terminate and restart the worker > > How hard would it be to raise a MemoryError from dask instead of killing > the worker? > > — > You are receiving this because you were mentioned. > Reply to this email directly, view it on GitHub > <#1469 (comment)>, > or mute the thread > <https://github.com/notifications/unsubscribe-auth/AASszEl3MNg98NR-VgytkHZSqo-6oki4ks5sx6ZGgaJpZM4P2Dk3> > . >

[jhamman]

okay, thanks. Forgive my ignorance here. One more question. Is there a way to kill the workers without sending PBS an exit code that aborts the job? I'm fine with a hard reset of the scheduler/workers, but if we can avoid having to resubmit all the workers, that would be ideal.

[mrocklin]

You should take a look at Scheduler.retire_workers. One can activate this from a client using

client.sync(client.scheduler.retire_workers, ...)

What is your objective?

[mrocklin]

@jhamman I just realized that because you're using dask-mpi we actually don't have nannies, and so we're not watching our workers and shutting them down if they run over 95%. I think that we avoided nannies originally because of some MPI + forking issue. I'm not sure what the current status of this is; it may have resolved itself. If you're interested the thing to do here is to replace Worker with Nanny in the distributed/cli/dask_mpi.py script, try it out on a distributed system like Cheyenne, and see how it does.

[jhamman]

@mrocklin - I think this can be closed by #1520.

[mrocklin]

For the record @jhamman and I also discovered that our (my) understanding of how much RAM our machines had was much higher than was accurate. Machines were running out of memory quickly and being OOM-killed by the job scheduler.

[bw4sz]

I am running into the same error, but i'm just commenting here until I figure out if this is a dask distributed problem, a dask-jobqueue problem, or something else. My error does not have to do with large memory. I have many image files that are read in parallel. After an hour, completing atleast 40 tasks, it ran into a corrupt image and killed all 10 workers. Because the dashboard died along with the workers (can I revive a previous run's dashboard?) its hard to tell exactly what happened. SLURM reports the job as Failed.

Here is my submission script on a SLURM cluster

def run_HPC(data_paths):
        
    #################
    # Setup dask cluster
    #################
    
    from dask_jobqueue import SLURMCluster
    from dask.distributed import Client
    from dask import delayed
    
    DeepForest_config = config.load_config("train")
    
    num_workers=DeepForest_config["num_hipergator_workers"]
    
    #job args
    extra_args=[
        "--error=/home/b.weinstein/logs/dask-worker-%j.err",
        "--account=ewhite",
        "--output=/home/b.weinstein/logs/dask-worker-%j.out"
    ]
    
    cluster = SLURMCluster(
        processes=1,
        queue='hpg2-compute',
        cores=1, 
        memory='20GB', 
        walltime='48:00:00',
        job_extra=extra_args,
        local_directory="/home/b.weinstein/logs/",death_timeout=300)
    
    print(cluster.job_script())
    cluster.scale(num_workers)
    
    dask_client = Client(cluster)
        
    #Start dask dashboard
    dask_client.run_on_scheduler(start_tunnel)    
    
    #Set scheduler.
    values = [delayed(Generate.run)(x) for x in data_paths]
    
    #Compute tiles    
    results = compute(*values,scheduler=dask_client)    
    return results

I'm trying to condense this down, since the traceback is quite long.

distributed.utils - ERROR - 'start'
Traceback (most recent call last):
  File "/home/b.weinstein/miniconda/envs/DeepLidar/lib/python3.6/site-packages/distributed/utils.py", line 648, in log_errors
    yield
  File "/home/b.weinstein/miniconda/envs/DeepLidar/lib/python3.6/site-packages/distributed/bokeh/components.py", line 545, in update
    ts = metadata['keys'][self.key]
KeyError: 'start'
tornado.application - ERROR - Exception in callback functools.partial(<function wrap.<locals>.null_wrapper at 0x2b14b02378c8>, <Future finished exception=KeyError('start',)>)
Traceback (most recent call last):
  File "/home/b.weinstein/miniconda/envs/DeepLidar/lib/python3.6/site-packages/tornado/ioloop.py", line 758, in _run_callback
    ret = callback()
  File "/home/b.weinstein/miniconda/envs/DeepLidar/lib/python3.6/site-packages/tornado/stack_context.py", line 300, in null_wrapper
    return fn(*args, **kwargs)
  File "/home/b.weinstein/miniconda/envs/DeepLidar/lib/python3.6/site-packages/tornado/ioloop.py", line 779, in _discard_future_result
    future.result()
  File "/home/b.weinstein/miniconda/envs/DeepLidar/lib/python3.6/site-packages/tornado/gen.py", line 1147, in run
    yielded = self.gen.send(value)
  File "/home/b.weinstein/miniconda/envs/DeepLidar/lib/python3.6/site-packages/bokeh/server/session.py", line 51, in _needs_document_lock_wrapper
    result = yield yield_for_all_futures(func(self, *args, **kwargs))
  File "/home/b.weinstein/miniconda/envs/DeepLidar/lib/python3.6/site-packages/bokeh/server/session.py", line 157, in with_document_locked
    return func(*args, **kwargs)
  File "/home/b.weinstein/miniconda/envs/DeepLidar/lib/python3.6/site-packages/bokeh/document/document.py", line 1127, in wrapper
    return doc._with_self_as_curdoc(invoke)
  File "/home/b.weinstein/miniconda/envs/DeepLidar/lib/python3.6/site-packages/bokeh/document/document.py", line 1113, in _with_self_as_curdoc
    return f()
  File "/home/b.weinstein/miniconda/envs/DeepLidar/lib/python3.6/site-packages/bokeh/document/document.py", line 1126, in invoke
    return f(*args, **kwargs)
  File "/home/b.weinstein/miniconda/envs/DeepLidar/lib/python3.6/site-packages/bokeh/document/document.py", line 916, in remove_then_invoke
    return callback(*args, **kwargs)
  File "/home/b.weinstein/miniconda/envs/DeepLidar/lib/python3.6/site-packages/distributed/bokeh/components.py", line 564, in <lambda>
    self.doc().add_next_tick_callback(lambda: self.update(prof, metadata))
  File "/home/b.weinstein/miniconda/envs/DeepLidar/lib/python3.6/site-packages/distributed/bokeh/components.py", line 545, in update
    ts = metadata['keys'][self.key]
KeyError: 'start'
Traceback (most recent call last):
  File "/home/b.weinstein/DeepLidar/dask_generate.py", line 105, in <module>
    run_HPC(data_paths)
  File "/home/b.weinstein/DeepLidar/dask_generate.py", line 92, in run_HPC
    results = compute(*values,scheduler=dask_client)
  File "/home/b.weinstein/miniconda/envs/DeepLidar/lib/python3.6/site-packages/dask/base.py", line 397, in compute
    results = schedule(dsk, keys, **kwargs)
  File "/home/b.weinstein/miniconda/envs/DeepLidar/lib/python3.6/site-packages/distributed/client.py", line 2318, in get
    direct=direct)
  File "/home/b.weinstein/miniconda/envs/DeepLidar/lib/python3.6/site-packages/distributed/client.py", line 1652, in gather
    asynchronous=asynchronous)
  File "/home/b.weinstein/miniconda/envs/DeepLidar/lib/python3.6/site-packages/distributed/client.py", line 670, in sync
    return sync(self.loop, func, *args, **kwargs)
  File "/home/b.weinstein/miniconda/envs/DeepLidar/lib/python3.6/site-packages/distributed/utils.py", line 277, in sync
    six.reraise(*error[0])
  File "/apps/geos/3.6.2/lib/python3.6/site-packages/six.py", line 693, in reraise
    raise value
  File "/home/b.weinstein/miniconda/envs/DeepLidar/lib/python3.6/site-packages/distributed/utils.py", line 262, in f
    result[0] = yield future
  File "/home/b.weinstein/miniconda/envs/DeepLidar/lib/python3.6/site-packages/tornado/gen.py", line 1133, in run
    value = future.result()
  File "/home/b.weinstein/miniconda/envs/DeepLidar/lib/python3.6/site-packages/tornado/gen.py", line 1141, in run
    yielded = self.gen.throw(*exc_info)
  File "/home/b.weinstein/miniconda/envs/DeepLidar/lib/python3.6/site-packages/distributed/client.py", line 1497, in _gather
    traceback)
  File "/apps/geos/3.6.2/lib/python3.6/site-packages/six.py", line 692, in reraise
    raise value.with_traceback(tb)
  File "/home/b.weinstein/DeepLidar/DeepForest/Generate.py", line 28, in run
    windows = preprocess.create_windows(data,DeepForest_config)
  File "/home/b.weinstein/DeepLidar/DeepForest/preprocess.py", line 306, in create_windows
    windows=compute_windows(image=image_path, pixels=DeepForest_config["patch_size"], overlap=DeepForest_config["patch_overlap"])
  File "/home/b.weinstein/DeepLidar/DeepForest/preprocess.py", line 151, in compute_windows
    im = Image.open(image)
  File "/home/b.weinstein/miniconda/envs/DeepLidar/lib/python3.6/site-packages/PIL/Image.py", line 2657, in open
    % (filename if filename else fp))
OSError: cannot identify image file '/ufrc/ewhite/b.weinstein/NeonData/SJER/DP3.30010.001/2018/FullSite/D17/2018_SJER_3/L3/Camera/Mosaic/V01/2018_SJER_3_252000_4114000_image.tif'
tornado.application - ERROR - Exception in Future <Future cancelled> after timeout
Traceback (most recent call last):
  File "/home/b.weinstein/miniconda/envs/DeepLidar/lib/python3.6/site-packages/tornado/gen.py", line 970, in error_callback
    future.result()
concurrent.futures._base.CancelledError

All workers are then killed, even though there are many tasks remaining. I can confirm that there is indeed an error here that is raised correctly, so maybe nothing is wrong with dask itself and this is a SLURM problem?

(DeepLidar) [b.weinstein@login3 ~]$ python
Python 3.6.7 | packaged by conda-forge | (default, Nov 21 2018, 02:32:25)
[GCC 4.8.2 20140120 (Red Hat 4.8.2-15)] on linux
Type "help", "copyright", "credits" or "license" for more information.
>>> from PIL import Image
>>> Image.open("/ufrc/ewhite/b.weinstein/NeonData/SJER/DP3.30010.001/2018/FullSite/D17/2018_SJER_3/L3/Camera/Mosaic/V01/2018_SJER_3_252000_4114000_image.tif")
/home/b.weinstein/miniconda/envs/DeepLidar/lib/python3.6/site-packages/PIL/TiffImagePlugin.py:771: UserWarning: Corrupt EXIF data.  Expecting to read 2 bytes but only got 0.
  warnings.warn(str(msg))
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/home/b.weinstein/miniconda/envs/DeepLidar/lib/python3.6/site-packages/PIL/Image.py", line 2657, in open
    % (filename if filename else fp))
OSError: cannot identify image file '/ufrc/ewhite/b.weinstein/NeonData/SJER/DP3.30010.001/2018/FullSite/D17/2018_SJER_3/L3/Camera/Mosaic/V01/2018_SJER_3_252000_4114000_image.tif'

I would have expected for a dask task to be caught, or at worst the worker to die, but not kill all other workers, following (http://distributed.dask.org/en/latest/resilience.html#user-code-failures).

System info: Large HPC cluster at University of Florida

Python 3.6.7 | packaged by conda-forge | (default, Nov 21 2018, 02:32:25)
[GCC 4.8.2 20140120 (Red Hat 4.8.2-15)] on linux
Type "help", "copyright", "credits" or "license" for more information.
>>> import dask
>>> dask.__version__
'0.20.2'
>>> import dask_jobqueue
>>> dask_jobqueue.__version__
'0.4.1'

As an aside, amazing progress on dask_jobqueue - i just back to it after 6 months and it was really seamless!

I will update here unless directed to a new issue or SO. I am currently wrapping the Image.open() into an explicit try-except IOError to see if that helps. It shouldn't matter that I have the debugging logger on, right?

import logging
logging.basicConfig(format='%(levelname)s:%(message)s', level=logging.DEBUG)