df.compute() on Mac OS LocalCluster with Client(processes=True) generates OSError

[chambersmatt]

What happened:

Attempting to process dask dataframe on LocalCluster on Mac OS 11.2.3 with 8 cores and 16gb of RAM.

df.compute() works when Client(processes=False) but fails if Client(processes=True).

Other operations on the dataframe complete successfully, but attempting to return the entirety of the dataframe fails, eg:

df[df.columns[:120]].compute() succeeds, while df.compute() fails.

See traceback below:

---------------------------------------------------------------------------
OSError                                   Traceback (most recent call last)
<ipython-input-38-9bb416d45ef6> in <module>
----> 1 df.compute()

~/.local/share/virtualenvs/stonks-ncnmWyPa/lib/python3.9/site-packages/dask/base.py in compute(self, **kwargs)
    282         dask.base.compute
    283         """
--> 284         (result,) = compute(self, traverse=False, **kwargs)
    285         return result
    286 

~/.local/share/virtualenvs/stonks-ncnmWyPa/lib/python3.9/site-packages/dask/base.py in compute(*args, **kwargs)
    564         postcomputes.append(x.__dask_postcompute__())
    565 
--> 566     results = schedule(dsk, keys, **kwargs)
    567     return repack([f(r, *a) for r, (f, a) in zip(results, postcomputes)])
    568 

~/.local/share/virtualenvs/stonks-ncnmWyPa/lib/python3.9/site-packages/distributed/client.py in get(self, dsk, keys, workers, allow_other_workers, resources, sync, asynchronous, direct, retries, priority, fifo_timeout, actors, **kwargs)
   2664                     should_rejoin = False
   2665             try:
-> 2666                 results = self.gather(packed, asynchronous=asynchronous, direct=direct)
   2667             finally:
   2668                 for f in futures.values():

~/.local/share/virtualenvs/stonks-ncnmWyPa/lib/python3.9/site-packages/distributed/client.py in gather(self, futures, errors, direct, asynchronous)
   1973             else:
   1974                 local_worker = None
-> 1975             return self.sync(
   1976                 self._gather,
   1977                 futures,

~/.local/share/virtualenvs/stonks-ncnmWyPa/lib/python3.9/site-packages/distributed/client.py in sync(self, func, asynchronous, callback_timeout, *args, **kwargs)
    841             return future
    842         else:
--> 843             return sync(
    844                 self.loop, func, *args, callback_timeout=callback_timeout, **kwargs
    845             )

~/.local/share/virtualenvs/stonks-ncnmWyPa/lib/python3.9/site-packages/distributed/utils.py in sync(loop, func, callback_timeout, *args, **kwargs)
    351     if error[0]:
    352         typ, exc, tb = error[0]
--> 353         raise exc.with_traceback(tb)
    354     else:
    355         return result[0]

~/.local/share/virtualenvs/stonks-ncnmWyPa/lib/python3.9/site-packages/distributed/utils.py in f()
    334             if callback_timeout is not None:
    335                 future = asyncio.wait_for(future, callback_timeout)
--> 336             result[0] = yield future
    337         except Exception as exc:
    338             error[0] = sys.exc_info()

~/.local/share/virtualenvs/stonks-ncnmWyPa/lib/python3.9/site-packages/tornado/gen.py in run(self)
    760 
    761                     try:
--> 762                         value = future.result()
    763                     except Exception:
    764                         exc_info = sys.exc_info()

~/.local/share/virtualenvs/stonks-ncnmWyPa/lib/python3.9/site-packages/distributed/client.py in _gather(self, futures, errors, direct, local_worker)
   1867                 else:
   1868                     self._gather_future = future
-> 1869                 response = await future
   1870 
   1871             if response["status"] == "error":

~/.local/share/virtualenvs/stonks-ncnmWyPa/lib/python3.9/site-packages/distributed/client.py in _gather_remote(self, direct, local_worker)
   1918 
   1919             else:  # ask scheduler to gather data for us
-> 1920                 response = await retry_operation(self.scheduler.gather, keys=keys)
   1921 
   1922         return response

~/.local/share/virtualenvs/stonks-ncnmWyPa/lib/python3.9/site-packages/distributed/utils_comm.py in retry_operation(coro, operation, *args, **kwargs)
    383         dask.config.get("distributed.comm.retry.delay.max"), default="s"
    384     )
--> 385     return await retry(
    386         partial(coro, *args, **kwargs),
    387         count=retry_count,

~/.local/share/virtualenvs/stonks-ncnmWyPa/lib/python3.9/site-packages/distributed/utils_comm.py in retry(coro, count, delay_min, delay_max, jitter_fraction, retry_on_exceptions, operation)
    368                 delay *= 1 + random.random() * jitter_fraction
    369             await asyncio.sleep(delay)
--> 370     return await coro()
    371 
    372 

~/.local/share/virtualenvs/stonks-ncnmWyPa/lib/python3.9/site-packages/distributed/core.py in send_recv_from_rpc(**kwargs)
    860             name, comm.name = comm.name, "ConnectionPool." + key
    861             try:
--> 862                 result = await send_recv(comm=comm, op=key, **kwargs)
    863             finally:
    864                 self.pool.reuse(self.addr, comm)

~/.local/share/virtualenvs/stonks-ncnmWyPa/lib/python3.9/site-packages/distributed/core.py in send_recv(comm, reply, serializers, deserializers, **kwargs)
    643         await comm.write(msg, serializers=serializers, on_error="raise")
    644         if reply:
--> 645             response = await comm.read(deserializers=deserializers)
    646         else:
    647             response = None

~/.local/share/virtualenvs/stonks-ncnmWyPa/lib/python3.9/site-packages/distributed/comm/tcp.py in read(self, deserializers)
    198 
    199             frames = bytearray(frames_nbytes)
--> 200             n = await stream.read_into(frames)
    201             assert n == frames_nbytes, (n, frames_nbytes)
    202         except StreamClosedError as e:

~/.local/share/virtualenvs/stonks-ncnmWyPa/lib/python3.9/site-packages/tornado/iostream.py in read_into(self, buf, partial)
    473 
    474         try:
--> 475             self._try_inline_read()
    476         except:
    477             future.add_done_callback(lambda f: f.exception())

~/.local/share/virtualenvs/stonks-ncnmWyPa/lib/python3.9/site-packages/tornado/iostream.py in _try_inline_read(self)
    840             return
    841         self._check_closed()
--> 842         pos = self._read_to_buffer_loop()
    843         if pos is not None:
    844             self._read_from_buffer(pos)

~/.local/share/virtualenvs/stonks-ncnmWyPa/lib/python3.9/site-packages/tornado/iostream.py in _read_to_buffer_loop(self)
    753             # can't see it; the only way to find out if it's there is to
    754             # try to read it.
--> 755             if self._read_to_buffer() == 0:
    756                 break
    757 

~/.local/share/virtualenvs/stonks-ncnmWyPa/lib/python3.9/site-packages/tornado/iostream.py in _read_to_buffer(self)
    865                     else:
    866                         buf = bytearray(self.read_chunk_size)
--> 867                     bytes_read = self.read_from_fd(buf)
    868                 except (socket.error, IOError, OSError) as e:
    869                     # ssl.SSLError is a subclass of socket.error

~/.local/share/virtualenvs/stonks-ncnmWyPa/lib/python3.9/site-packages/tornado/iostream.py in read_from_fd(***failed resolving arguments***)
   1138     def read_from_fd(self, buf: Union[bytearray, memoryview]) -> Optional[int]:
   1139         try:
-> 1140             return self.socket.recv_into(buf, len(buf))
   1141         except BlockingIOError:
   1142             return None

OSError: [Errno 22] Invalid argument

What you expected to happen:

Successful return of computed dataframe using multiple processes.

Minimal Complete Verifiable Example:

from dask.distributed import Client
from dask import dataframe as dd

# load 3.5gb CSV
basefile = '/path/to/basefile.csv'
dtypes = {'foo': 'int8', 'bar': 'float16', 'total': 'int64'}
df = dd.read_csv(basefile, sep='\t', dtype=dtypes, parse_dates=['Date'])

client = Client(processes=True)

# fails
df.compute()

# succeeds
df[df.columns[:120]].compute()

Environment:

• Dask version: dask-2021.4.0
• Python version: 3.9.2
• Operating System: Mac OS 11.2.3
• Install method (conda, pip, source): pip

[fjetter]

Hi @chambersmatt . Can you try reproducing the error using generated data instead of a CSV file? This would help with reproducing the issue.

For generating data, you could try https://github.com/dask/dask/blob/c5633c21e754df6b2c13195f1e2688e347c48963/dask/datasets.py#L6-L53

For this particular error message, the pip freeze output of your environment would also be helpful to narrow down the problem

[chambersmatt]

Thanks @fjetter - see below for generated data, triggering the same error:

from dask.distributed import Client
from dask import datasets
import random

client = Client(processes=True)

dtypes = {f"col{i}": random.choice([float, int]) for i in range(170)}
df = datasets.timeseries(dtypes=dtypes)
# succeeds
df[df.columns[:100]].compute()
# fails
df.compute()

Running this in Google Colab with a local runtime in case that makes a difference...
EDIT: tested on standard Jupyter notebook running at localhost:8888 with the same result.

pip freeze:

appnope==0.1.2
argon2-cffi==20.1.0
async-generator==1.10
attrs==20.3.0
backcall==0.2.0
bleach==3.3.0
bokeh==2.3.0
Brotli==1.0.9
cffi==1.14.5
click==7.1.2
cloudpickle==1.6.0
dash-core-components==1.15.0
dash-html-components==1.1.2
dash-renderer==1.9.0
dash-table==4.11.2
dask==2021.4.0
decorator==5.0.5
defusedxml==0.7.1
distributed==2021.4.0
entrypoints==0.3
Flask==1.1.2
Flask-Compress==1.9.0
fsspec==0.8.7
future==0.18.2
HeapDict==1.0.1
ipykernel==5.5.3
ipython==7.22.0
ipython-genutils==0.2.0
ipywidgets==7.6.3
itsdangerous==1.1.0
jedi==0.18.0
Jinja2==2.11.3
jsonschema==3.2.0
jupyter==1.0.0
jupyter-client==6.1.12
jupyter-console==6.4.0
jupyter-core==4.7.1
jupyter-http-over-ws==0.0.8
jupyterlab-pygments==0.1.2
jupyterlab-widgets==1.0.0
locket==0.2.1
MarkupSafe==1.1.1
mistune==0.8.4
msgpack==1.0.2
nbclient==0.5.3
nbconvert==6.0.7
nbformat==5.1.3
nest-asyncio==1.5.1
notebook==6.3.0
numpy==1.20.2
packaging==20.9
pandas==1.2.3
pandocfilters==1.4.3
parso==0.8.2
partd==1.1.0
pexpect==4.8.0
pickleshare==0.7.5
Pillow==8.2.0
plotly==4.14.3
prometheus-client==0.10.0
prompt-toolkit==3.0.18
psutil==5.8.0
ptyprocess==0.7.0
pycparser==2.20
Pygments==2.8.1
pyparsing==2.4.7
pyrsistent==0.17.3
python-dateutil==2.8.1
pytz==2021.1
PyYAML==5.4.1
pyzmq==22.0.3
qtconsole==5.0.3
QtPy==1.9.0
retrying==1.3.3
Send2Trash==1.5.0
six==1.15.0
sortedcontainers==2.3.0
tblib==1.7.0
terminado==0.9.4
testpath==0.4.4
toolz==0.11.1
tornado==6.1
traitlets==5.0.5
typing-extensions==3.7.4.3
wcwidth==0.2.5
webencodings==0.5.1
Werkzeug==1.0.1
widgetsnbextension==3.5.1
zict==2.0.0