Intra-worker compression errors should be handled

[dwragge]

Describe the issue:

We were seeing some our jobs enter a state where not all of the tasks had completed, but the workers were sitting idle. This seemed to be happening fairly consistently on a particular set of jobs we were running.

Looking in the worker logs, I saw the following error:

  File "/.pyenv/versions/3.8.10/lib/python3.8/site-packages/distributed/protocol/core.py", line 100, in _encode_default
    frames.extend(create_serialized_sub_frames(obj))
  File "/.pyenv/versions/3.8.10/lib/python3.8/site-packages/distributed/protocol/core.py", line 67, in create_serialized_sub_frames
    _inplace_compress_frames(sub_header, sub_frames)
  File "/.pyenv/versions/3.8.10/lib/python3.8/site-packages/distributed/protocol/core.py", line 50, in _inplace_compress_frames
    compression[i], frames[i] = maybe_compress(
  File "/.pyenv/versions/3.8.10/lib/python3.8/site-packages/distributed/protocol/compression.py", line 188, in maybe_compress
    compressed = compress(mv)
_block.LZ4BlockError: Compression failed

It seems that at this point, Dask had attempted to transfer data between workers, failed, and then gotten into a bit of a stalemate state. Looking into the LZ4 code, I believe we were hitting this case where the input data compresses so poorly it exceeds the original size. Why this doesn't get hit in the "subset test", I'm not sure.

From the perspective of the library, it seems reasonable that this case should raise an error (indeed, it did, technically, fail to compress, but there was nothing inherently wrong or corrupt with the data).

What are your thoughts on catching compression errors and returning the uncompressed data as a fall back? I can see issues with a) what to catch, we don't want to catch errors that highlight a bad system state and b) handling errors from all the different libraries, however I feel like some of this is mitigated by the fact that this code is in full control of the compress-decompress lifecycle, and it can be reasonably expected that compression should fail. Decompression errors should definitely not be handled.

Thoughts here?

Environment:

• Dask version: 2022.11.1
• Python version: 3.8.10
• Operating System: Ubuntu 16.04.7 LTS
• Install method (conda, pip, source): Pip

[fjetter]

From the perspective of the library, it seems reasonable that this case should raise an error

Is it possible to run lz4 s.t. this is not raised as an error? We have our own logic that rejects the compressed bytes if they are not at least 10% smaller

I assume there are many different cases where LZ4BlockError is raised and not just this poor compression.
I'm generally open to being more forgiving but would still want the user to know that something went wrong, e.g. a log/warning. After all, if this happens frequently this is something we want to know.

[dwragge]

It does not look like it (ref: https://github.com/python-lz4/python-lz4/blob/master/lz4/block/_block.c#L242) where the library treats anything <= 0 as an error (although AFAIK only strictly less is an error per se).

I definitely agree with you that this error should be loud and there are a myriad of reasons why compression would fail. I unfortunately don't have much more information on the exact data that's failing, other than it should just be a dataframe (and thus compress reasonably well).

I can follow up with LZ4 if you think it's better suited to be handled there in your opinion!

[gjoseph92]

I feel like the broader issue here is that compression errors cause a deadlock? Ideally, I imagine we'd treat the task as failed (so at least it would be in a terminal state). Haven't looked, but I imagine this is a similar code path to #7060.