We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
LZ4 compression fails on very large frames
This causes a complete halt of the system.  We could consider framing or punting.


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `distributed/protocol.py`
Content:
```
1 """
2 The distributed message protocol consists of the following parts:
3 
4 1.  The length of the header, stored as a uint32
5 2.  The header, stored as msgpack.
6     If there are no fields in the header then we skip it entirely.
7 3.  The payload, stored as possibly compressed msgpack
8 4.  A sentinel value
9 
10 **Header**
11 
12 The Header contains the following fields:
13 
14 * **compression**: string, optional
15     One of the following: ``'snappy', 'lz4', 'zlib'`` or missing for None
16 
17 **Payload**
18 
19 The payload is any msgpack serializable value.  It may be compressed based
20 on the header.
21 
22 **Sentinel**
23 
24 We often terminate each message with a sentinel value.  This happens
25 outside of this module though and is not baked in.
26 """
27 from __future__ import print_function, division, absolute_import
28 
29 import random
30 import struct
31 
32 try:
33     import pandas.msgpack as msgpack
34 except ImportError:
35     import msgpack
36 
37 from toolz import first, keymap, identity, merge
38 
39 from .utils import ignoring
40 from .compatibility import unicode
41 
42 
43 compressions = {None: {'compress': identity,
44                        'decompress': identity}}
45 
46 default_compression = None
47 
48 
49 with ignoring(ImportError):
50     import zlib
51     compressions['zlib'] = {'compress': zlib.compress,
52                             'decompress': zlib.decompress}
53 
54 with ignoring(ImportError):
55     import snappy
56     compressions['snappy'] = {'compress': snappy.compress,
57                               'decompress': snappy.decompress}
58     default_compression = 'snappy'
59 
60 with ignoring(ImportError):
61     import lz4
62     compressions['lz4'] = {'compress': lz4.LZ4_compress,
63                            'decompress': lz4.LZ4_uncompress}
64     default_compression = 'lz4'
65 
66 
67 def dumps(msg):
68     """ Transform Python value to bytestream suitable for communication """
69     small_header = {}
70 
71     if isinstance(msg, dict):
72         big = {k: v for k, v in msg.items()
73                     if isinstance(v, bytes) and len(v) > 1e6}
74     else:
75         big = False
76     if big:
77         small = {k: v for k, v in msg.items() if k not in big}
78     else:
79         small = msg
80 
81     frames = dumps_msgpack(small)
82     if big:
83         frames += dumps_big_byte_dict(big)
84 
85     return frames
86 
87 
88 def loads(frames):
89     """ Transform bytestream back into Python value """
90     header, payload, frames = frames[0], frames[1], frames[2:]
91     msg = loads_msgpack(header, payload)
92 
93     if frames:
94         big = loads_big_byte_dict(*frames)
95         msg.update(big)
96 
97     return msg
98 
99 
100 def byte_sample(b, size, n):
101     """ Sample a bytestring from many locations """
102     starts = [random.randint(0, len(b) - size) for j in range(n)]
103     ends = []
104     for i, start in enumerate(starts[:-1]):
105         ends.append(min(start + size, starts[i + 1]))
106     ends.append(starts[-1] + size)
107 
108     return b''.join([b[start:end] for start, end in zip(starts, ends)])
109 
110 
111 def maybe_compress(payload, compression=default_compression, min_size=1e4,
112         sample_size=1e4, nsamples=5):
113     """ Maybe compress payload
114 
115     1.  We don't compress small messages
116     2.  We sample the payload in a few spots, compress that, and if it doesn't
117         do any good we return the original
118     3.  We then compress the full original, it it doesn't compress well then we
119         return the original
120     4.  We return the compressed result
121     """
122     if not compression:
123         return None, payload
124     if len(payload) < min_size:
125         return None, payload
126 
127     min_size = int(min_size)
128     sample_size = int(sample_size)
129 
130     compress = compressions[compression]['compress']
131 
132     # Compress a sample, return original if not very compressed
133     sample = byte_sample(payload, sample_size, nsamples)
134     if len(compress(sample)) > 0.9 * len(sample):  # not very compressible
135         return None, payload
136 
137     compressed = compress(payload)
138     if len(compressed) > 0.9 * len(payload):  # not very compressible
139         return None, payload
140 
141     return compression, compress(payload)
142 
143 
144 def dumps_msgpack(msg):
145     """ Dump msg into header and payload, both bytestrings
146 
147     All of the message must be msgpack encodable
148 
149     See Also:
150         loads_msgpack
151     """
152     header = {}
153     payload = msgpack.dumps(msg, use_bin_type=True)
154 
155     fmt, payload = maybe_compress(payload)
156     if fmt:
157         header['compression'] = fmt
158 
159     if header:
160         header_bytes = msgpack.dumps(header, use_bin_type=True)
161     else:
162         header_bytes = b''
163 
164     return [header_bytes, payload]
165 
166 
167 def loads_msgpack(header, payload):
168     """ Read msgpack header and payload back to Python object
169 
170     See Also:
171         dumps_msgpack
172     """
173     if header:
174         header = msgpack.loads(header, encoding='utf8')
175     else:
176         header = {}
177 
178     if header.get('compression'):
179         try:
180             decompress = compressions[header['compression']]['decompress']
181             payload = decompress(payload)
182         except KeyError:
183             raise ValueError("Data is compressed as %s but we don't have this"
184                     " installed" % header['compression'].decode())
185 
186     return msgpack.loads(payload, encoding='utf8')
187 
188 
189 def dumps_big_byte_dict(d):
190     """ Serialize large byte dictionary to sequence of frames
191 
192     The input must be a dictionary and all values of that dictionary must be
193     bytestrings.  These should probably be large.
194 
195     Returns a sequence of frames, one header followed by each of the values
196 
197     See Also:
198         loads_big_byte_dict
199     """
200     assert isinstance(d, dict) and all(isinstance(v, bytes) for v in d.values())
201     keys, values = zip(*d.items())
202 
203     compress = compressions[default_compression]['compress']
204     compression = []
205     values2 = []
206     for v in values:
207         fmt, vv = maybe_compress(v)
208         compression.append(fmt)
209         values2.append(vv)
210 
211     header = {'encoding': 'big-byte-dict',
212               'keys': keys,
213               'compression': compression}
214 
215     return [msgpack.dumps(header, use_bin_type=True)] + values2
216 
217 
218 def loads_big_byte_dict(header, *values):
219     """ Deserialize big-byte frames to large byte dictionary
220 
221     See Also:
222         dumps_big_byte_dict
223     """
224     header = msgpack.loads(header, encoding='utf8')
225 
226     values2 = [compressions[c]['decompress'](v)
227                for c, v in zip(header['compression'], values)]
228     return dict(zip(header['keys'], values2))
229
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/distributed/protocol.py b/distributed/protocol.py
--- a/distributed/protocol.py
+++ b/distributed/protocol.py
@@ -198,6 +198,17 @@
         loads_big_byte_dict
     """
     assert isinstance(d, dict) and all(isinstance(v, bytes) for v in d.values())
+    shards = {}
+    for k, v in list(d.items()):
+        if len(v) >= 2**31:
+            L = []
+            for i, j in enumerate(range(0, len(v), 2**30)):
+                key = '.shard-%d-%s' % (i, k)
+                d[key] = v[j: j + 2**30]
+                L.append(key)
+            del d[k]
+            shards[k] = L
+
     keys, values = zip(*d.items())
 
     compress = compressions[default_compression]['compress']
@@ -211,6 +222,8 @@
     header = {'encoding': 'big-byte-dict',
               'keys': keys,
               'compression': compression}
+    if shards:
+        header['shards'] = shards
 
     return [msgpack.dumps(header, use_bin_type=True)] + values2
 
@@ -225,4 +238,8 @@
 
     values2 = [compressions[c]['decompress'](v)
                for c, v in zip(header['compression'], values)]
-    return dict(zip(header['keys'], values2))
+    result = dict(zip(header['keys'], values2))
+
+    for k, keys in header.get('shards', {}).items():
+        result[k] = b''.join(result.pop(kk) for kk in keys)
+    return result

{"golden_diff": "diff --git a/distributed/protocol.py b/distributed/protocol.py\n--- a/distributed/protocol.py\n+++ b/distributed/protocol.py\n@@ -198,6 +198,17 @@\n         loads_big_byte_dict\n     \"\"\"\n     assert isinstance(d, dict) and all(isinstance(v, bytes) for v in d.values())\n+    shards = {}\n+    for k, v in list(d.items()):\n+        if len(v) >= 2**31:\n+            L = []\n+            for i, j in enumerate(range(0, len(v), 2**30)):\n+                key = '.shard-%d-%s' % (i, k)\n+                d[key] = v[j: j + 2**30]\n+                L.append(key)\n+            del d[k]\n+            shards[k] = L\n+\n     keys, values = zip(*d.items())\n \n     compress = compressions[default_compression]['compress']\n@@ -211,6 +222,8 @@\n     header = {'encoding': 'big-byte-dict',\n               'keys': keys,\n               'compression': compression}\n+    if shards:\n+        header['shards'] = shards\n \n     return [msgpack.dumps(header, use_bin_type=True)] + values2\n \n@@ -225,4 +238,8 @@\n \n     values2 = [compressions[c]['decompress'](v)\n                for c, v in zip(header['compression'], values)]\n-    return dict(zip(header['keys'], values2))\n+    result = dict(zip(header['keys'], values2))\n+\n+    for k, keys in header.get('shards', {}).items():\n+        result[k] = b''.join(result.pop(kk) for kk in keys)\n+    return result\n", "issue": "LZ4 compression fails on very large frames\nThis causes a complete halt of the system.  We could consider framing or punting.\n\n", "before_files": [{"content": "\"\"\"\nThe distributed message protocol consists of the following parts:\n\n1.  The length of the header, stored as a uint32\n2.  The header, stored as msgpack.\n    If there are no fields in the header then we skip it entirely.\n3.  The payload, stored as possibly compressed msgpack\n4.  A sentinel value\n\n**Header**\n\nThe Header contains the following fields:\n\n* **compression**: string, optional\n    One of the following: ``'snappy', 'lz4', 'zlib'`` or missing for None\n\n**Payload**\n\nThe payload is any msgpack serializable value.  It may be compressed based\non the header.\n\n**Sentinel**\n\nWe often terminate each message with a sentinel value.  This happens\noutside of this module though and is not baked in.\n\"\"\"\nfrom __future__ import print_function, division, absolute_import\n\nimport random\nimport struct\n\ntry:\n    import pandas.msgpack as msgpack\nexcept ImportError:\n    import msgpack\n\nfrom toolz import first, keymap, identity, merge\n\nfrom .utils import ignoring\nfrom .compatibility import unicode\n\n\ncompressions = {None: {'compress': identity,\n                       'decompress': identity}}\n\ndefault_compression = None\n\n\nwith ignoring(ImportError):\n    import zlib\n    compressions['zlib'] = {'compress': zlib.compress,\n                            'decompress': zlib.decompress}\n\nwith ignoring(ImportError):\n    import snappy\n    compressions['snappy'] = {'compress': snappy.compress,\n                              'decompress': snappy.decompress}\n    default_compression = 'snappy'\n\nwith ignoring(ImportError):\n    import lz4\n    compressions['lz4'] = {'compress': lz4.LZ4_compress,\n                           'decompress': lz4.LZ4_uncompress}\n    default_compression = 'lz4'\n\n\ndef dumps(msg):\n    \"\"\" Transform Python value to bytestream suitable for communication \"\"\"\n    small_header = {}\n\n    if isinstance(msg, dict):\n        big = {k: v for k, v in msg.items()\n                    if isinstance(v, bytes) and len(v) > 1e6}\n    else:\n        big = False\n    if big:\n        small = {k: v for k, v in msg.items() if k not in big}\n    else:\n        small = msg\n\n    frames = dumps_msgpack(small)\n    if big:\n        frames += dumps_big_byte_dict(big)\n\n    return frames\n\n\ndef loads(frames):\n    \"\"\" Transform bytestream back into Python value \"\"\"\n    header, payload, frames = frames[0], frames[1], frames[2:]\n    msg = loads_msgpack(header, payload)\n\n    if frames:\n        big = loads_big_byte_dict(*frames)\n        msg.update(big)\n\n    return msg\n\n\ndef byte_sample(b, size, n):\n    \"\"\" Sample a bytestring from many locations \"\"\"\n    starts = [random.randint(0, len(b) - size) for j in range(n)]\n    ends = []\n    for i, start in enumerate(starts[:-1]):\n        ends.append(min(start + size, starts[i + 1]))\n    ends.append(starts[-1] + size)\n\n    return b''.join([b[start:end] for start, end in zip(starts, ends)])\n\n\ndef maybe_compress(payload, compression=default_compression, min_size=1e4,\n        sample_size=1e4, nsamples=5):\n    \"\"\" Maybe compress payload\n\n    1.  We don't compress small messages\n    2.  We sample the payload in a few spots, compress that, and if it doesn't\n        do any good we return the original\n    3.  We then compress the full original, it it doesn't compress well then we\n        return the original\n    4.  We return the compressed result\n    \"\"\"\n    if not compression:\n        return None, payload\n    if len(payload) < min_size:\n        return None, payload\n\n    min_size = int(min_size)\n    sample_size = int(sample_size)\n\n    compress = compressions[compression]['compress']\n\n    # Compress a sample, return original if not very compressed\n    sample = byte_sample(payload, sample_size, nsamples)\n    if len(compress(sample)) > 0.9 * len(sample):  # not very compressible\n        return None, payload\n\n    compressed = compress(payload)\n    if len(compressed) > 0.9 * len(payload):  # not very compressible\n        return None, payload\n\n    return compression, compress(payload)\n\n\ndef dumps_msgpack(msg):\n    \"\"\" Dump msg into header and payload, both bytestrings\n\n    All of the message must be msgpack encodable\n\n    See Also:\n        loads_msgpack\n    \"\"\"\n    header = {}\n    payload = msgpack.dumps(msg, use_bin_type=True)\n\n    fmt, payload = maybe_compress(payload)\n    if fmt:\n        header['compression'] = fmt\n\n    if header:\n        header_bytes = msgpack.dumps(header, use_bin_type=True)\n    else:\n        header_bytes = b''\n\n    return [header_bytes, payload]\n\n\ndef loads_msgpack(header, payload):\n    \"\"\" Read msgpack header and payload back to Python object\n\n    See Also:\n        dumps_msgpack\n    \"\"\"\n    if header:\n        header = msgpack.loads(header, encoding='utf8')\n    else:\n        header = {}\n\n    if header.get('compression'):\n        try:\n            decompress = compressions[header['compression']]['decompress']\n            payload = decompress(payload)\n        except KeyError:\n            raise ValueError(\"Data is compressed as %s but we don't have this\"\n                    \" installed\" % header['compression'].decode())\n\n    return msgpack.loads(payload, encoding='utf8')\n\n\ndef dumps_big_byte_dict(d):\n    \"\"\" Serialize large byte dictionary to sequence of frames\n\n    The input must be a dictionary and all values of that dictionary must be\n    bytestrings.  These should probably be large.\n\n    Returns a sequence of frames, one header followed by each of the values\n\n    See Also:\n        loads_big_byte_dict\n    \"\"\"\n    assert isinstance(d, dict) and all(isinstance(v, bytes) for v in d.values())\n    keys, values = zip(*d.items())\n\n    compress = compressions[default_compression]['compress']\n    compression = []\n    values2 = []\n    for v in values:\n        fmt, vv = maybe_compress(v)\n        compression.append(fmt)\n        values2.append(vv)\n\n    header = {'encoding': 'big-byte-dict',\n              'keys': keys,\n              'compression': compression}\n\n    return [msgpack.dumps(header, use_bin_type=True)] + values2\n\n\ndef loads_big_byte_dict(header, *values):\n    \"\"\" Deserialize big-byte frames to large byte dictionary\n\n    See Also:\n        dumps_big_byte_dict\n    \"\"\"\n    header = msgpack.loads(header, encoding='utf8')\n\n    values2 = [compressions[c]['decompress'](v)\n               for c, v in zip(header['compression'], values)]\n    return dict(zip(header['keys'], values2))\n", "path": "distributed/protocol.py"}], "after_files": [{"content": "\"\"\"\nThe distributed message protocol consists of the following parts:\n\n1.  The length of the header, stored as a uint32\n2.  The header, stored as msgpack.\n    If there are no fields in the header then we skip it entirely.\n3.  The payload, stored as possibly compressed msgpack\n4.  A sentinel value\n\n**Header**\n\nThe Header contains the following fields:\n\n* **compression**: string, optional\n    One of the following: ``'snappy', 'lz4', 'zlib'`` or missing for None\n\n**Payload**\n\nThe payload is any msgpack serializable value.  It may be compressed based\non the header.\n\n**Sentinel**\n\nWe often terminate each message with a sentinel value.  This happens\noutside of this module though and is not baked in.\n\"\"\"\nfrom __future__ import print_function, division, absolute_import\n\nimport random\nimport struct\n\ntry:\n    import pandas.msgpack as msgpack\nexcept ImportError:\n    import msgpack\n\nfrom toolz import first, keymap, identity, merge\n\nfrom .utils import ignoring\nfrom .compatibility import unicode\n\n\ncompressions = {None: {'compress': identity,\n                       'decompress': identity}}\n\ndefault_compression = None\n\n\nwith ignoring(ImportError):\n    import zlib\n    compressions['zlib'] = {'compress': zlib.compress,\n                            'decompress': zlib.decompress}\n\nwith ignoring(ImportError):\n    import snappy\n    compressions['snappy'] = {'compress': snappy.compress,\n                              'decompress': snappy.decompress}\n    default_compression = 'snappy'\n\nwith ignoring(ImportError):\n    import lz4\n    compressions['lz4'] = {'compress': lz4.LZ4_compress,\n                           'decompress': lz4.LZ4_uncompress}\n    default_compression = 'lz4'\n\n\ndef dumps(msg):\n    \"\"\" Transform Python value to bytestream suitable for communication \"\"\"\n    small_header = {}\n\n    if isinstance(msg, dict):\n        big = {k: v for k, v in msg.items()\n                    if isinstance(v, bytes) and len(v) > 1e6}\n    else:\n        big = False\n    if big:\n        small = {k: v for k, v in msg.items() if k not in big}\n    else:\n        small = msg\n\n    frames = dumps_msgpack(small)\n    if big:\n        frames += dumps_big_byte_dict(big)\n\n    return frames\n\n\ndef loads(frames):\n    \"\"\" Transform bytestream back into Python value \"\"\"\n    header, payload, frames = frames[0], frames[1], frames[2:]\n    msg = loads_msgpack(header, payload)\n\n    if frames:\n        big = loads_big_byte_dict(*frames)\n        msg.update(big)\n\n    return msg\n\n\ndef byte_sample(b, size, n):\n    \"\"\" Sample a bytestring from many locations \"\"\"\n    starts = [random.randint(0, len(b) - size) for j in range(n)]\n    ends = []\n    for i, start in enumerate(starts[:-1]):\n        ends.append(min(start + size, starts[i + 1]))\n    ends.append(starts[-1] + size)\n\n    return b''.join([b[start:end] for start, end in zip(starts, ends)])\n\n\ndef maybe_compress(payload, compression=default_compression, min_size=1e4,\n        sample_size=1e4, nsamples=5):\n    \"\"\" Maybe compress payload\n\n    1.  We don't compress small messages\n    2.  We sample the payload in a few spots, compress that, and if it doesn't\n        do any good we return the original\n    3.  We then compress the full original, it it doesn't compress well then we\n        return the original\n    4.  We return the compressed result\n    \"\"\"\n    if not compression:\n        return None, payload\n    if len(payload) < min_size:\n        return None, payload\n\n    min_size = int(min_size)\n    sample_size = int(sample_size)\n\n    compress = compressions[compression]['compress']\n\n    # Compress a sample, return original if not very compressed\n    sample = byte_sample(payload, sample_size, nsamples)\n    if len(compress(sample)) > 0.9 * len(sample):  # not very compressible\n        return None, payload\n\n    compressed = compress(payload)\n    if len(compressed) > 0.9 * len(payload):  # not very compressible\n        return None, payload\n\n    return compression, compress(payload)\n\n\ndef dumps_msgpack(msg):\n    \"\"\" Dump msg into header and payload, both bytestrings\n\n    All of the message must be msgpack encodable\n\n    See Also:\n        loads_msgpack\n    \"\"\"\n    header = {}\n    payload = msgpack.dumps(msg, use_bin_type=True)\n\n    fmt, payload = maybe_compress(payload)\n    if fmt:\n        header['compression'] = fmt\n\n    if header:\n        header_bytes = msgpack.dumps(header, use_bin_type=True)\n    else:\n        header_bytes = b''\n\n    return [header_bytes, payload]\n\n\ndef loads_msgpack(header, payload):\n    \"\"\" Read msgpack header and payload back to Python object\n\n    See Also:\n        dumps_msgpack\n    \"\"\"\n    if header:\n        header = msgpack.loads(header, encoding='utf8')\n    else:\n        header = {}\n\n    if header.get('compression'):\n        try:\n            decompress = compressions[header['compression']]['decompress']\n            payload = decompress(payload)\n        except KeyError:\n            raise ValueError(\"Data is compressed as %s but we don't have this\"\n                    \" installed\" % header['compression'].decode())\n\n    return msgpack.loads(payload, encoding='utf8')\n\n\ndef dumps_big_byte_dict(d):\n    \"\"\" Serialize large byte dictionary to sequence of frames\n\n    The input must be a dictionary and all values of that dictionary must be\n    bytestrings.  These should probably be large.\n\n    Returns a sequence of frames, one header followed by each of the values\n\n    See Also:\n        loads_big_byte_dict\n    \"\"\"\n    assert isinstance(d, dict) and all(isinstance(v, bytes) for v in d.values())\n    shards = {}\n    for k, v in list(d.items()):\n        if len(v) >= 2**31:\n            L = []\n            for i, j in enumerate(range(0, len(v), 2**30)):\n                key = '.shard-%d-%s' % (i, k)\n                d[key] = v[j: j + 2**30]\n                L.append(key)\n            del d[k]\n            shards[k] = L\n\n    keys, values = zip(*d.items())\n\n    compress = compressions[default_compression]['compress']\n    compression = []\n    values2 = []\n    for v in values:\n        fmt, vv = maybe_compress(v)\n        compression.append(fmt)\n        values2.append(vv)\n\n    header = {'encoding': 'big-byte-dict',\n              'keys': keys,\n              'compression': compression}\n    if shards:\n        header['shards'] = shards\n\n    return [msgpack.dumps(header, use_bin_type=True)] + values2\n\n\ndef loads_big_byte_dict(header, *values):\n    \"\"\" Deserialize big-byte frames to large byte dictionary\n\n    See Also:\n        dumps_big_byte_dict\n    \"\"\"\n    header = msgpack.loads(header, encoding='utf8')\n\n    values2 = [compressions[c]['decompress'](v)\n               for c, v in zip(header['compression'], values)]\n    result = dict(zip(header['keys'], values2))\n\n    for k, keys in header.get('shards', {}).items():\n        result[k] = b''.join(result.pop(kk) for kk in keys)\n    return result\n", "path": "distributed/protocol.py"}]}