Series.values.compute() leads to "TypeError: can't concat buffer to bytearray"

[bluenote10]

Running a local dask-scheduler + dask-worker pair, the following code leads to a crash of the worker:

import dask.dataframe as dd
from dask.distributed import Executor
e = Executor('127.0.0.1:8786', set_as_default=True)
df = pd.DataFrame({"A": [1, 2, 3] * 10})
ddf = dd.from_pandas(df, npartitions=3)
ddf["A"].values.compute()

The worker crashes with:

Traceback (most recent call last):
  File "/home/PHI-TPS/fkeller/.virtualenvs/dask/local/lib/python2.7/site-packages/distributed/core.py", line 259, in handle_comm
    result = yield result
  File "/home/PHI-TPS/fkeller/.virtualenvs/dask/local/lib/python2.7/site-packages/tornado/gen.py", line 1055, in run
    value = future.result()
  File "/home/PHI-TPS/fkeller/.virtualenvs/dask/local/lib/python2.7/site-packages/tornado/concurrent.py", line 238, in result
    raise_exc_info(self._exc_info)
  File "/home/PHI-TPS/fkeller/.virtualenvs/dask/local/lib/python2.7/site-packages/tornado/gen.py", line 1063, in run
    yielded = self.gen.throw(*exc_info)
  File "/home/PHI-TPS/fkeller/.virtualenvs/dask/local/lib/python2.7/site-packages/distributed/worker.py", line 439, in get_data
    compressed = yield comm.write(msg)
  File "/home/PHI-TPS/fkeller/.virtualenvs/dask/local/lib/python2.7/site-packages/tornado/gen.py", line 1055, in run
    value = future.result()
  File "/home/PHI-TPS/fkeller/.virtualenvs/dask/local/lib/python2.7/site-packages/tornado/concurrent.py", line 238, in result
    raise_exc_info(self._exc_info)
  File "/home/PHI-TPS/fkeller/.virtualenvs/dask/local/lib/python2.7/site-packages/tornado/gen.py", line 292, in wrapper
    result = func(*args, **kwargs)
  File "/home/PHI-TPS/fkeller/.virtualenvs/dask/local/lib/python2.7/site-packages/distributed/comm/tcp.py", line 196, in write
    stream.write(frame)
  File "/home/PHI-TPS/fkeller/.virtualenvs/dask/local/lib/python2.7/site-packages/tornado/iostream.py", line 395, in write
    self._write_buffer += data
TypeError: can't concat buffer to bytearray

Other computations like ddf["A"].compute() or even ddf.values.compute() work fine though.

[bluenote10]

I noticed that I can compute ddf["A"].values.compute() if I initialize with client = Client('127.0.0.1:8786') instead of using Executor. However then I get the worker crash on computing ddf["A"].unique().compute(), but only after successfully computing ddf["A"].values.compute() once. I can't really put my finger on whether it is related to the initialization, these specific operations, or even the order of operations.

[mrocklin]

To be clear, Executor is just an alias to Client

Executor = Client

This runs fine for me. My first guess is that you have a version mismatch. Can you verify that the following does not err:

client.get_versions(check=True)

[bluenote10]

That's the output I get:

{'client': {'host': [('python', '2.7.3.final.0'),
   ('python-bits', 64),
   ('OS', 'Linux'),
   ('OS-release', '3.13.0-100-generic'),
   ('machine', 'x86_64'),
   ('processor', 'x86_64'),
   ('byteorder', 'little'),
   ('LC_ALL', 'None'),
   ('LANG', 'en_US.UTF-8'),
   ('LOCALE', 'en_US.UTF-8')],
  'packages': {'optional': [('numpy', '1.13.0'), ('pandas', u'0.20.2')],
   'required': [('dask', u'0.15.0'),
    ('distributed', u'1.17.0'),
    ('msgpack', '0.4.8'),
    ('cloudpickle', '0.3.1'),
    ('toolz', '0.8.2')]}},
 'scheduler': {'host': [['python', '2.7.3.final.0'],
   ['python-bits', 64],
   ['OS', 'Linux'],
   ['OS-release', '3.13.0-100-generic'],
   ['machine', 'x86_64'],
   ['processor', 'x86_64'],
   ['byteorder', 'little'],
   ['LC_ALL', 'None'],
   ['LANG', 'en_US.UTF-8'],
   ['LOCALE', 'None.None']],
  'packages': {'optional': [['numpy', '1.13.0'], ['pandas', u'0.20.2']],
   'required': [['dask', u'0.15.0'],
    ['distributed', u'1.17.0'],
    ['msgpack', '0.4.8'],
    ['cloudpickle', '0.3.1'],
    ['toolz', '0.8.2']]}},
 'workers': {'tcp://10.128.4.209:41229': {'host': [('python', '2.7.3.final.0'),
    ('python-bits', 64),
    ('OS', 'Linux'),
    ('OS-release', '3.13.0-100-generic'),
    ('machine', 'x86_64'),
    ('processor', 'x86_64'),
    ('byteorder', 'little'),
    ('LC_ALL', 'None'),
    ('LANG', 'en_US.UTF-8'),
    ('LOCALE', 'None.None')],
   'packages': {'optional': [('numpy', '1.13.0'), ('pandas', u'0.20.2')],
    'required': [('dask', u'0.15.0'),
     ('distributed', u'1.17.0'),
     ('msgpack', '0.4.8'),
     ('cloudpickle', '0.3.1'),
     ('toolz', '0.8.2')]}}}}

The installation should be up-to-date versions of both dask + distributed. Client + scheduler + worker are all running from the same venv on the same host.

[mrocklin]

I tried reproducing this with an environment like the following and unfortunately everything worked.

conda create -n gh-1179 python=2.7 distributed=1.17.0 numpy=1.13.0 pandas ipython

Any recommendations on what to check that might differ between our environments?

[bluenote10]

Maybe it's related to tornado, since that's where the error is thrown? My tornado version is 4.5.1.

[mrocklin]

Same

[mrocklin]

There was also a micro-release shortly after 1.17.1. This resolved some things with moving around memoryviews. I don't think that this is likely to affect you but you might try updating and see if that has an effect.

[bluenote10]

Updated to 1.17.1 but the worker still throws the exception with slightly modified behavior: I did get the computation result back in the client on the first attempt although the worker raised the exception. Repeating the command a second time hangs in the client. I feel that the issue manifests itself in a somewhat erratic way.

Right now I have no idea what could be wrong...

[mrocklin]

Lets wait for @pitrou to take a look at the error. He is more familiar with the networking stack than I am (I think he wrote the bytearray code in tornado) and may have thoughts. Unfortunately I think he's on vacation until Monday.

[bluenote10]

Adding a few more observations: Currently the frame in the the loop for frame in frames (distributed/comm/tcp.py line 192) are typically of type str, which is fine to pass on to Tornado's stream.write(frame). For some reason the frames list sometimes contains raw buffer objects, which Tornado can't handle. I can avoid the issue by a dirty fix like:

if "buffer" in type(frame).__name__:
    stream.write(b"{}".format(frame))
else:
    stream.write(frame)

But maybe that is not the right place to fix the issue and probably we still want to understand why the problem isn't easily reproducible.

Update: Maybe this isn't really a fix. It doesn't crash any more, but now my computation results differ from running with the single machine schedulers.

[mrocklin]

If I recall correctly we recently started allowing buffers and memoryviews through, but only if the stream supported them. Perhaps we're making this judgment incorrectly. This is definitely a situation where @pitrou would know more.

[bluenote10]

Ah, I think I found what is causing the issue: It might be related to allocators. I'm encountering the issue in a setting where the driver is using jemalloc while the scheduler/worker are using glibc. If I use glibc everywhere the issue disappears.

[xhochy]

Same as in the other issue: I cannot reproduce the failures here with a newer jemalloc version / on OSX. I rather suspect that there is some usage of unitialised memory. This will probably not be a bug in distributed but one of the libraries it depends on that call native code.

[bluenote10]

Dang, I'm getting the TypeError: can't concat buffer to bytearray now in a pure glibc setting as well :(. But it takes more complex code now to trigger the issue, the simple example from above works. This is a tough issue.