For loop never ending

[eXpensia]

Hello,

I'm using dask to analyse some images, my code looks like this :

somefunc(n):
    try:
        long calculation
        return result
    except:
         return to be filtered result

def main():
    num_image = 268828
    X = np.empty((num_image,7), dtype=np.float32)
    cluster = SLURMCluster(cores=32,
                           processes=32,
                           interface ="ib0",
                           walltime="04:10:00",
                           memory = "80GB")
    cluster.scale(600)
    client = Client(cluster)

    with tqdm (total = num_image) as pbar:
        future = client.map(somefunc, range(num_image))
        for future,result in as_completed(future,with_results = True):
            X[result[0],:] = result
            pbar.update(1)
            future.release()

    return X


if __name__ == '__main__':

    x = main()
    io.savemat('/gpfs/home/savedata.mat', {'data': x})

The scripts very well excepts for some iterations and it stays stuck without doing some iterations

100%|█████████▉| 268711/268828 [3:40:11<00:31, 3.68it/s]

Looking at the log files I found that it first looped on a first error :

Traceback (most recent call last):
  File "/gpfs/home/maxlavaud/miniconda3/envs/env_dask/lib/python3.6/site-packages/tornado/ioloop.py", line 743, in _run_callback
    ret = callback()
  File "/gpfs/home/maxlavaud/miniconda3/envs/env_dask/lib/python3.6/site-packages/tornado/ioloop.py", line 767, in _discard_future_result
    future.result()
  File "/gpfs/home/maxlavaud/miniconda3/envs/env_dask/lib/python3.6/site-packages/distributed/worker.py", line 875, in heartbeat
    metrics=await self.get_metrics(),
  File "/gpfs/home/maxlavaud/miniconda3/envs/env_dask/lib/python3.6/site-packages/distributed/core.py", line 747, in send_recv_from_rpc
    comm = await self.pool.connect(self.addr)
  File "/gpfs/home/maxlavaud/miniconda3/envs/env_dask/lib/python3.6/site-packages/distributed/core.py", line 874, in connect
    connection_args=self.connection_args,
  File "/gpfs/home/maxlavaud/miniconda3/envs/env_dask/lib/python3.6/site-packages/distributed/comm/core.py", line 227, in connect
    _raise(error)
  File "/gpfs/home/maxlavaud/miniconda3/envs/env_dask/lib/python3.6/site-packages/distributed/comm/core.py", line 204, in _raise
    raise IOError(msg)
OSError: Timed out trying to connect to 'tcp://192.168.11.3:16892' after 10 s: in <distributed.comm.tcp.TCPConnector object at 0x2b1a8a2bc630>: ConnectionRefusedError: [Errno 111] Connection refused
tornado.application - ERROR - Exception in callback functools.partial(<bound method IOLoop._discard_future_result of <tornado.platform.asyncio.AsyncIOLoop object at 0x2b1a32811c88>>, <Task finished coro=<Worker.heartbeat() done, defined at /gpfs/home/maxlavaud/miniconda3/envs/env_dask/lib/python3.6/site-packages/distributed/worker.py:866> exception=OSError("Timed out trying to connect to 'tcp://192.168.11.3:16892' after 10 s: in <distributed.comm.tcp.TCPConnector object at 0x2b1a8a346c50>: ConnectionRefusedError: [Errno 111] Connection refused",)>)
Traceback (most recent call last):
  File "/gpfs/home/maxlavaud/miniconda3/envs/env_dask/lib/python3.6/site-packages/distributed/comm/core.py", line 215, in connect
    quiet_exceptions=EnvironmentError,
tornado.util.TimeoutError: Timeout

And then I think it's when the wall time as been reached it loops on another one :

Traceback (most recent call last):
  File "/gpfs/home/maxlavaud/miniconda3/envs/env_dask/lib/python3.6/site-packages/tornado/ioloop.py", line 907, in _run
    return self.callback()
  File "/gpfs/home/maxlavaud/miniconda3/envs/env_dask/lib/python3.6/site-packages/distributed/worker.py", line 857, in <lambda>
    lambda: self.batched_stream.send({"op": "keep-alive"}),
  File "/gpfs/home/maxlavaud/miniconda3/envs/env_dask/lib/python3.6/site-packages/distributed/batched.py", line 117, in send
    raise CommClosedError
distributed.comm.core.CommClosedError

Also I've been warned that there is some process emited by dask that are using a lot of memory (10-100MB) on the login node, I don't quit understand where do they come from.

[lesteve]

I don't have any suggestions of the top of my head. Can you post the dask, distributed and dask-jobqueue versions for future references?

If you are able to reproduce with LocalCluster (or with SLURMCluster but a ) that might help other people debug the problem further. Also something I would try (just in case) would be to make sure to use the latest releeased versions of dask, distributed and dask-jobqueue.

Also I've been warned that there is some process emited by dask that are using a lot of memory (10-100MB) on the login node, I don't quit understand where do they come from.

If you are running your main script on the login node, that may be expected. Your X is already 7MB big, plus your dask scheduler lives on the same node as you are launching your main script. 10-100MB does not seem a lot but if it is too much for your cluster policy, an easy work-around is to launch your main script from an interactive job.

[eXpensia]

I have the 2.3.0 version of dask (installed from conda) and 0.6.3 version of dask-jobqueue.
This problem seems to arise when i do a lot of iterations so it might not be possible with LocalCluster

an easy work-around is to launch your main script from an interactive job.

I don't find any documentations on that, when i try to lunch it from an interactive node i get the error : sbatch: error: Batch job submission failed: Access/permission denied

So i understand that we can't use the sbatch command inside a computing node, do we have to replace this command by another one ? Or is this that we have to change :shebang: "#!/usr/bin/env bash" ?

[willirath]

I don't find any documentations on that, when i try to lunch it from an interactive node i get the error : sbatch: error: Batch job submission failed: Access/permission denied

This indeed sounds like submissions from compute nodes are not possible. Are you in touch with any of your cluster admins about this?

So i understand that we can't use the sbatch command inside a computing node, do we have to replace this command by another one ? Or is this that we have to change :shebang: "#!/usr/bin/env bash" ?

No, the shebang doesn't have any impact here. As the job script that contains it isn't even submitted.

[eXpensia]

If you are running your main script on the login node, that may be expected. Your X is already 7MB big, plus your dask scheduler lives on the same node as you are launching your main script. 10-100MB does not seem a lot but if it is too much for your cluster policy, an easy work-around is to launch your main script from an interactive job.

I miss spelled it's using 10-100**GB** of memory (even when jobs are still pending).

I don't find any documentations on that, when i try to lunch it from an interactive node i get the error : sbatch: error: Batch job submission failed: Access/permission denied

This indeed sounds like submissions from compute nodes are not possible. Are you in touch with any of your cluster admins about this?

Indeed i talk with an admin today, they've got some problems with some code which generated infinite loops of sbatch so we don't have the permission to use it from the compute nodes.
He told me that he will try to create a work around with a sheduler on a compute node and we send the jobs to it from the login node.

[lesteve]

Indeed i talk with an admin today, they've got some problems with some code which generated infinite loops of sbatch so we don't have the permission to use it from the compute nodes.

Wow yet another possible quirk in the phase-space of cluster configurations ... sorry to hear that ... A work-around you could do (this comes from OAR where the submission command is not installed on computing nodes):

cluster = SLURMCluster(...)
cluster.submit_command = 'ssh your_login_node sbatch'
# Hopefully you can use qdel from a computing node otherwise uncomment the following line
# cluster.cancel_command = 'ssh your_login_node qdel'

This is a bit hacky but this could work ...

I miss spelled it's using 10-100GB of memory (even when jobs are still pending).

This is unexpected, probably not related to dask but to the objects you are creating in your script. If I was to debug this problem I would remove the dask-related part and see how much memory the script is using. If you still think that this is related to dask, a stand-alone snippet reproducing the problem is needed.

[guillaumeeb]

@eXpensia were you able to solve this problem?

[guillaumeeb]

@eXpensia did you find a solution to your Original Problem where some iterations were never ending?

[eXpensia]

Hello, Actually no, I think it was due to a memory leak wich maybe also explain the memory problem. I used an another to compute what I needed and I don't need anymore parallelization.

[guillaumeeb]

Out of curiosity, could you tell us what other thing did you use for your computation?

[eXpensia]

Sure, basically what i'm doing is fitting 2D images to track particles. And a group which is using the same technique implemented it with cython and cupy (or numba if no GPU available) which permits to do it as fast as 500 cores but with only one machine :)

You can find the code here :
https://github.com/davidgrier/pylorenzmie

[guillaumeeb]

Nice example of optimization !