'Dask-SLURMCluster: [Errno 104] Connection reset by peer

I'm running into a problem using a Xarray together with SLURMcluster from Dask.

I'm using pandas_plink to load some data into a Xarray, then filtering it and making some calculations. Everything works fine as long I'm not using a SLURMCluster. In that case .compute() kills the kernel. I think that it might load the whole file even if I'm only using a part of it.

import os
import numpy as np
import pandas as pd
import math
import time
import warnings

from joblib import parallel_backend

from pandas_plink import read_plink
from pandas_plink import read_plink1_bin
from pandas_plink import get_data_folder
from pandas_plink import Chunk

import dask as da
import dask.array
from dask.dataframe import from_pandas
import dask_ml.model_selection as dcv
import dask.multiprocessing
from dask_jobqueue import SLURMCluster
from dask.distributed import Client, progress, LocalCluster
import xarray as xr

np.random.seed(42)
da.config.set({"array.slicing.split_large_chunks": True}) 
os.environ["TUNE_MAX_PENDING_TRIALS_PG"] = "8"

G = read_plink1_bin('xxx.bed',verbose=True,chunk=Chunk(nsamples=4000, nvariants=4000))

G = G.sel(sample=G.sample[0:1000])
G = G.where(G.chrom == 11, drop=False)
G = G.sel(variant=G.variant[0:100])
G = G.fillna(G.mean(dim='sample'))

print('%.1f MB' % (G.nbytes / 1e6)) #Output: 0.4 MB

print(G.compute())

print(G)

#Output:

<xarray.DataArray 'genotype' (sample: 100, variant: 1000)>
dask.array<where, shape=(100, 1000), dtype=float32, chunksize=(50, 1000), chunktype=numpy.ndarray>
Coordinates: (12/14)
  * sample   (sample) object '12' '13' ... '15' '16'
  * variant  (variant) <U13 'variant0' 'variant1' ... 'variant998' 'variant999'
    fid      (sample) object '12' '13' ... '15' '16'
    iid      (sample) object '12' '13' ... '15' '16'
    father   (sample) object '0' '0' '0' '0' '0' '0' ... '0' '0' '0' '0' '0' '0'
    mother   (sample) object '0' '0' '0' '0' '0' '0' ... '0' '0' '0' '0' '0' '0'
    ...       ...
    chrom    (variant) object '1' '1' '1' '1' '1' '1' ... '1' '1' '1' '1' '1'
    snp      (variant) object '123' '124' ... '999'
    cm       (variant) float64 0.0 0.0 0.0 0.0 0.0 0.0 ... 0.0 0.0 0.0 0.0 0.0
    pos      (variant) int32 123456 123457 123458 ... 987654 987655 987656
    a0       (variant) object 'A' 'C' 'A' 'A' 'A' 'C' ... 'C' 'A' 'T' 'A' 'T'
    a1       (variant) object 'G' 'T' 'G' 'C' 'G' 'T' ... 'T' 'G' 'C' 'G' 'C'

This works fine. But, if I'm starting a cluster before, it doesn't work anymore.

cluster = SLURMCluster(name='dask-cluster',
                      cores=8,
                      n_workers=8,
                      memory='20GB', 
                      walltime='12:00:00')
cluster.adapt(minimum=0, maximum=10)
client = Client(cluster)

In that case, print(G.sel(sample='123',variant='variant0').values) runs a while before the kernel dies and wants to restart. I guess for some reason it runs out of memory, even the array just has 0.4 MB.

Is there any explanation for that behavior?

UPDATE:

I'm getting the following error from the cluster:

...
2022-04-29 15:11:22,923 - distributed.worker - INFO - -------------------------------------------------
2022-04-29 15:11:22,923 - distributed.core - INFO - Starting established connection
2022-04-29 15:20:37,364 - distributed.worker - WARNING - Heartbeat to scheduler failed
Traceback (most recent call last):
  File "/mnt/nfs/clustersw/Debian/bullseye/JupyterPython/2.0/lib/python3.9/site-packages/tornado/iostream.py", line 867, in _read_to_buffer
    bytes_read = self.read_from_fd(buf)
  File "/mnt/nfs/clustersw/Debian/bullseye/JupyterPython/2.0/lib/python3.9/site-packages/tornado/iostream.py", line 1140, in read_from_fd
    return self.socket.recv_into(buf, len(buf))
ConnectionResetError: [Errno 104] Connection reset by peer

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/nfs/scistore13/robingrp/csyrowat/.local/lib/python3.9/site-packages/distributed/worker.py", line 1137, in heartbeat
    response = await retry_operation(
  File "/nfs/scistore13/robingrp/csyrowat/.local/lib/python3.9/site-packages/distributed/utils_comm.py", line 381, in retry_operation
    return await retry(
  File "/nfs/scistore13/robingrp/csyrowat/.local/lib/python3.9/site-packages/distributed/utils_comm.py", line 366, in retry
    return await coro()
  File "/nfs/scistore13/robingrp/csyrowat/.local/lib/python3.9/site-packages/distributed/core.py", line 940, in send_recv_from_rpc
    return await send_recv(comm=comm, op=key, **kwargs)
  File "/nfs/scistore13/robingrp/csyrowat/.local/lib/python3.9/site-packages/distributed/core.py", line 709, in send_recv
    response = await comm.read(deserializers=deserializers)
  File "/nfs/scistore13/robingrp/csyrowat/.local/lib/python3.9/site-packages/distributed/comm/tcp.py", line 242, in read
    convert_stream_closed_error(self, e)
  File "/nfs/scistore13/robingrp/csyrowat/.local/lib/python3.9/site-packages/distributed/comm/tcp.py", line 148, in convert_stream_closed_error
    raise CommClosedError(f"in {obj}: {exc.__class__.__name__}: {exc}") from exc
distributed.comm.core.CommClosedError: in <TCP (closed) ConnectionPool.heartbeat_worker local=tcp://10.36.97.77:32856 remote=tcp://10.36.97.65:41221>: ConnectionResetError: [Errno 104] Connection reset by peer
2022-04-29 15:20:37,364 - distributed.worker - WARNING - Heartbeat to scheduler failed
Traceback (most recent call last):
  File "/mnt/nfs/clustersw/Debian/bullseye/JupyterPython/2.0/lib/python3.9/site-packages/tornado/iostream.py", line 867, in _read_to_buffer
    bytes_read = self.read_from_fd(buf)
  File "/mnt/nfs/clustersw/Debian/bullseye/JupyterPython/2.0/lib/python3.9/site-packages/tornado/iostream.py", line 1140, in read_from_fd
    return self.socket.recv_into(buf, len(buf))
ConnectionResetError: [Errno 104] Connection reset by peer

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/nfs/scistore13/robingrp/csyrowat/.local/lib/python3.9/site-packages/distributed/worker.py", line 1137, in heartbeat
    response = await retry_operation(
  File "/nfs/scistore13/robingrp/csyrowat/.local/lib/python3.9/site-packages/distributed/utils_comm.py", line 381, in retry_operation
    return await retry(
  File "/nfs/scistore13/robingrp/csyrowat/.local/lib/python3.9/site-packages/distributed/utils_comm.py", line 366, in retry
    return await coro()
  File "/nfs/scistore13/robingrp/csyrowat/.local/lib/python3.9/site-packages/distributed/core.py", line 940, in send_recv_from_rpc
    return await send_recv(comm=comm, op=key, **kwargs)
  File "/nfs/scistore13/robingrp/csyrowat/.local/lib/python3.9/site-packages/distributed/core.py", line 709, in send_recv
    response = await comm.read(deserializers=deserializers)
  File "/nfs/scistore13/robingrp/csyrowat/.local/lib/python3.9/site-packages/distributed/comm/tcp.py", line 242, in read
    convert_stream_closed_error(self, e)
  File "/nfs/scistore13/robingrp/csyrowat/.local/lib/python3.9/site-packages/distributed/comm/tcp.py", line 148, in convert_stream_closed_error
    raise CommClosedError(f"in {obj}: {exc.__class__.__name__}: {exc}") from exc
distributed.comm.core.CommClosedError: in <TCP (closed) ConnectionPool.heartbeat_worker local=tcp://10.36.97.77:32854 remote=tcp://10.36.97.65:41221>: ConnectionResetError: [Errno 104] Connection reset by peer
2022-04-29 15:20:37,364 - distributed.worker - WARNING - Heartbeat to scheduler failed
Traceback (most recent call last):
  File "/mnt/nfs/clustersw/Debian/bullseye/JupyterPython/2.0/lib/python3.9/site-packages/tornado/iostream.py", line 867, in _read_to_buffer
    bytes_read = self.read_from_fd(buf)
  File "/mnt/nfs/clustersw/Debian/bullseye/JupyterPython/2.0/lib/python3.9/site-packages/tornado/iostream.py", line 1140, in read_from_fd
    return self.socket.recv_into(buf, len(buf))
ConnectionResetError: [Errno 104] Connection reset by peer

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/nfs/scistore13/robingrp/csyrowat/.local/lib/python3.9/site-packages/distributed/worker.py", line 1137, in heartbeat
    response = await retry_operation(
  File "/nfs/scistore13/robingrp/csyrowat/.local/lib/python3.9/site-packages/distributed/utils_comm.py", line 381, in retry_operation
    return await retry(
  File "/nfs/scistore13/robingrp/csyrowat/.local/lib/python3.9/site-packages/distributed/utils_comm.py", line 366, in retry
    return await coro()
  File "/nfs/scistore13/robingrp/csyrowat/.local/lib/python3.9/site-packages/distributed/core.py", line 940, in send_recv_from_rpc
    return await send_recv(comm=comm, op=key, **kwargs)
  File "/nfs/scistore13/robingrp/csyrowat/.local/lib/python3.9/site-packages/distributed/core.py", line 709, in send_recv
    response = await comm.read(deserializers=deserializers)
  File "/nfs/scistore13/robingrp/csyrowat/.local/lib/python3.9/site-packages/distributed/comm/tcp.py", line 242, in read
    convert_stream_closed_error(self, e)
  File "/nfs/scistore13/robingrp/csyrowat/.local/lib/python3.9/site-packages/distributed/comm/tcp.py", line 148, in convert_stream_closed_error
    raise CommClosedError(f"in {obj}: {exc.__class__.__name__}: {exc}") from exc
distributed.comm.core.CommClosedError: in <TCP (closed) ConnectionPool.heartbeat_worker local=tcp://10.36.97.77:32852 remote=tcp://10.36.97.65:41221>: ConnectionResetError: [Errno 104] Connection reset by peer
2022-04-29 15:20:37,364 - distributed.worker - WARNING - Heartbeat to scheduler failed
Traceback (most recent call last):
  File "/mnt/nfs/clustersw/Debian/bullseye/JupyterPython/2.0/lib/python3.9/site-packages/tornado/iostream.py", line 867, in _read_to_buffer
    bytes_read = self.read_from_fd(buf)
  File "/mnt/nfs/clustersw/Debian/bullseye/JupyterPython/2.0/lib/python3.9/site-packages/tornado/iostream.py", line 1140, in read_from_fd
    return self.socket.recv_into(buf, len(buf))
ConnectionResetError: [Errno 104] Connection reset by peer

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/nfs/scistore13/robingrp/csyrowat/.local/lib/python3.9/site-packages/distributed/worker.py", line 1137, in heartbeat
    response = await retry_operation(
  File "/nfs/scistore13/robingrp/csyrowat/.local/lib/python3.9/site-packages/distributed/utils_comm.py", line 381, in retry_operation
    return await retry(
  File "/nfs/scistore13/robingrp/csyrowat/.local/lib/python3.9/site-packages/distributed/utils_comm.py", line 366, in retry
    return await coro()
  File "/nfs/scistore13/robingrp/csyrowat/.local/lib/python3.9/site-packages/distributed/core.py", line 940, in send_recv_from_rpc
    return await send_recv(comm=comm, op=key, **kwargs)
  File "/nfs/scistore13/robingrp/csyrowat/.local/lib/python3.9/site-packages/distributed/core.py", line 709, in send_recv
    response = await comm.read(deserializers=deserializers)
  File "/nfs/scistore13/robingrp/csyrowat/.local/lib/python3.9/site-packages/distributed/comm/tcp.py", line 242, in read
    convert_stream_closed_error(self, e)
  File "/nfs/scistore13/robingrp/csyrowat/.local/lib/python3.9/site-packages/distributed/comm/tcp.py", line 148, in convert_stream_closed_error
    raise CommClosedError(f"in {obj}: {exc.__class__.__name__}: {exc}") from exc
distributed.comm.core.CommClosedError: in <TCP (closed) ConnectionPool.heartbeat_worker local=tcp://10.36.97.77:32858 remote=tcp://10.36.97.65:41221>: ConnectionResetError: [Errno 104] Connection reset by peer
2022-04-29 15:20:37,367 - distributed.worker - INFO - Connection to scheduler broken.  Reconnecting...
2022-04-29 15:20:37,367 - distributed.worker - INFO - Connection to scheduler broken.  Reconnecting...
2022-04-29 15:20:37,367 - distributed.worker - INFO - Connection to scheduler broken.  Reconnecting...
2022-04-29 15:20:37,367 - distributed.worker - INFO - Connection to scheduler broken.  Reconnecting...
2022-04-29 15:21:07,367 - distributed.worker - INFO - Stopping worker at tcp://10.36.97.77:33851
2022-04-29 15:21:07,368 - distributed.worker - INFO - Stopping worker at tcp://10.36.97.77:41869
2022-04-29 15:21:07,368 - distributed.worker - INFO - Stopping worker at tcp://10.36.97.77:41763
2022-04-29 15:21:07,368 - distributed.worker - INFO - Stopping worker at tcp://10.36.97.77:40759
2022-04-29 15:21:07,379 - distributed.nanny - INFO - Worker closed
2022-04-29 15:21:07,380 - distributed.nanny - INFO - Worker closed
2022-04-29 15:21:07,380 - distributed.nanny - INFO - Worker closed
2022-04-29 15:21:07,380 - distributed.nanny - INFO - Worker closed
2022-04-29 15:21:07,653 - distributed.nanny - INFO - Closing Nanny at 'tcp://10.36.97.77:46739'
2022-04-29 15:21:07,656 - distributed.nanny - INFO - Closing Nanny at 'tcp://10.36.97.77:37165'
2022-04-29 15:21:07,656 - distributed.nanny - INFO - Closing Nanny at 'tcp://10.36.97.77:38269'
2022-04-29 15:21:07,659 - distributed.nanny - INFO - Closing Nanny at 'tcp://10.36.97.77:39827'
2022-04-29 15:21:07,659 - distributed.dask_worker - INFO - End worker

UPDATE2:

In case, I'm subsampling the PLINK-file, does it drop the rest of the array?

G = G.isel(sample = slice(0,n_samples), variant = slice(0, n_features),drop=True)

I suspect that when calling .compute() that it loads the whole file, which of course is too large to fit into memory.

But, how do I run a model then and get e.g. R^2 scores?

e.g. if I'm running a linear regression like below the kernel crashes when calling mod.score(G, yS).compute().

from dask_ml.linear_model import LinearRegression

mod = da.delayed(LinearRegression()).fit(G,yS)
mod.score(G, yS).compute()

Sources

This article follows the attribution requirements of Stack Overflow and is licensed under CC BY-SA 3.0.

Source: Stack Overflow

Solution	Source

'Dask-SLURMCluster: [Errno 104] Connection reset by peer

Sources

Related Questions