We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
`distributed.protocol.dumps` does not handle objects not serializable via `msgpack.dumps` as suggested by docstring
When using `distributed.protocol.dumps` to serialize arbitrary data, it fails and raises a `TypeError`. The docstring suggests it should be able to handle this.

**Minimal Complete Verifiable Example**:

```python
import numpy as np
from distributed.protocol import dumps

dumps(np.int64(123))
```

raises

```python
2024-01-09 19:55:04,572 - distributed.protocol.core - CRITICAL - Failed to Serialize
Traceback (most recent call last):
  File "/Users/hendrikmakait/projects/dask/distributed/distributed/protocol/core.py", line 108, in dumps
    frames[0] = msgpack.dumps(msg, default=_encode_default, use_bin_type=True)
                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/homebrew/Caskroom/mambaforge/base/envs/dask-distributed/lib/python3.12/site-packages/msgpack/__init__.py", line 36, in packb
    return Packer(**kwargs).pack(o)
           ^^^^^^^^^^^^^^^^^^^^^^^^
  File "msgpack/_packer.pyx", line 294, in msgpack._cmsgpack.Packer.pack
  File "msgpack/_packer.pyx", line 300, in msgpack._cmsgpack.Packer.pack
  File "msgpack/_packer.pyx", line 297, in msgpack._cmsgpack.Packer.pack
  File "msgpack/_packer.pyx", line 291, in msgpack._cmsgpack.Packer._pack
TypeError: can not serialize 'numpy.int64' object
Traceback (most recent call last):
  File "/Users/hendrikmakait/projects/dask/distributed/sandbox.py", line 5, in <module>
    dumps(np.int64(123))
  File "/Users/hendrikmakait/projects/dask/distributed/distributed/protocol/core.py", line 108, in dumps
    frames[0] = msgpack.dumps(msg, default=_encode_default, use_bin_type=True)
                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/homebrew/Caskroom/mambaforge/base/envs/dask-distributed/lib/python3.12/site-packages/msgpack/__init__.py", line 36, in packb
    return Packer(**kwargs).pack(o)
           ^^^^^^^^^^^^^^^^^^^^^^^^
  File "msgpack/_packer.pyx", line 294, in msgpack._cmsgpack.Packer.pack
  File "msgpack/_packer.pyx", line 300, in msgpack._cmsgpack.Packer.pack
  File "msgpack/_packer.pyx", line 297, in msgpack._cmsgpack.Packer.pack
  File "msgpack/_packer.pyx", line 291, in msgpack._cmsgpack.Packer._pack
TypeError: can not serialize 'numpy.int64' object
```

Relevant section of `dumps`'s docstring: https://github.com/dask/distributed/blob/7562f9c566978de4f3f5b73920a24ea1813d6e28/distributed/protocol/core.py#L30-L35

--- 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/core.py`
Content:
```
1 from __future__ import annotations
2 
3 import logging
4 
5 import msgpack
6 
7 from distributed.protocol import pickle
8 from distributed.protocol.compression import decompress, maybe_compress
9 from distributed.protocol.serialize import (
10     Pickled,
11     Serialize,
12     Serialized,
13     ToPickle,
14     merge_and_deserialize,
15     msgpack_decode_default,
16     msgpack_encode_default,
17     serialize_and_split,
18 )
19 from distributed.protocol.utils import msgpack_opts
20 from distributed.utils import ensure_memoryview
21 
22 logger = logging.getLogger(__name__)
23 
24 
25 def dumps(  # type: ignore[no-untyped-def]
26     msg, serializers=None, on_error="message", context=None, frame_split_size=None
27 ) -> list:
28     """Transform Python message to bytestream suitable for communication
29 
30     Developer Notes
31     ---------------
32     The approach here is to use `msgpack.dumps()` to serialize `msg` and
33     write the result to the first output frame. If `msgpack.dumps()`
34     encounters an object it cannot serialize like a NumPy array, it is handled
35     out-of-band by `_encode_default()` and appended to the output frame list.
36     """
37     try:
38         if context and "compression" in context:
39             compress_opts = {"compression": context["compression"]}
40         else:
41             compress_opts = {}
42 
43         def _inplace_compress_frames(header, frames):
44             compression = list(header.get("compression", [None] * len(frames)))
45 
46             for i in range(len(frames)):
47                 if compression[i] is None:
48                     compression[i], frames[i] = maybe_compress(
49                         frames[i], **compress_opts
50                     )
51 
52             header["compression"] = tuple(compression)
53 
54         def create_serialized_sub_frames(obj: Serialized | Serialize) -> list:
55             if isinstance(obj, Serialized):
56                 sub_header, sub_frames = obj.header, obj.frames
57             else:
58                 sub_header, sub_frames = serialize_and_split(
59                     obj,
60                     serializers=serializers,
61                     on_error=on_error,
62                     context=context,
63                     size=frame_split_size,
64                 )
65                 _inplace_compress_frames(sub_header, sub_frames)
66             sub_header["num-sub-frames"] = len(sub_frames)
67             sub_header = msgpack.dumps(
68                 sub_header, default=msgpack_encode_default, use_bin_type=True
69             )
70             return [sub_header] + sub_frames
71 
72         def create_pickled_sub_frames(obj: Pickled | ToPickle) -> list:
73             if isinstance(obj, Pickled):
74                 sub_header, sub_frames = obj.header, obj.frames
75             else:
76                 sub_frames = []
77                 sub_header = {
78                     "pickled-obj": pickle.dumps(
79                         obj.data,
80                         # In to support len() and slicing, we convert `PickleBuffer`
81                         # objects to memoryviews of bytes.
82                         buffer_callback=lambda x: sub_frames.append(
83                             ensure_memoryview(x)
84                         ),
85                     )
86                 }
87                 _inplace_compress_frames(sub_header, sub_frames)
88 
89             sub_header["num-sub-frames"] = len(sub_frames)
90             sub_header = msgpack.dumps(sub_header)
91             return [sub_header] + sub_frames
92 
93         frames = [None]
94 
95         def _encode_default(obj):
96             if isinstance(obj, (Serialize, Serialized)):
97                 offset = len(frames)
98                 frames.extend(create_serialized_sub_frames(obj))
99                 return {"__Serialized__": offset}
100             elif isinstance(obj, (ToPickle, Pickled)):
101                 offset = len(frames)
102                 frames.extend(create_pickled_sub_frames(obj))
103                 return {"__Pickled__": offset}
104             else:
105                 return msgpack_encode_default(obj)
106 
107         frames[0] = msgpack.dumps(msg, default=_encode_default, use_bin_type=True)
108         return frames
109 
110     except Exception:
111         logger.critical("Failed to Serialize", exc_info=True)
112         raise
113 
114 
115 def loads(frames, deserialize=True, deserializers=None):
116     """Transform bytestream back into Python value"""
117 
118     try:
119 
120         def _decode_default(obj):
121             offset = obj.get("__Serialized__", 0)
122             if offset > 0:
123                 sub_header = msgpack.loads(
124                     frames[offset],
125                     object_hook=msgpack_decode_default,
126                     use_list=False,
127                     **msgpack_opts,
128                 )
129                 offset += 1
130                 sub_frames = frames[offset : offset + sub_header["num-sub-frames"]]
131                 if deserialize:
132                     if "compression" in sub_header:
133                         sub_frames = decompress(sub_header, sub_frames)
134                     return merge_and_deserialize(
135                         sub_header, sub_frames, deserializers=deserializers
136                     )
137                 else:
138                     return Serialized(sub_header, sub_frames)
139 
140             offset = obj.get("__Pickled__", 0)
141             if offset > 0:
142                 sub_header = msgpack.loads(frames[offset])
143                 offset += 1
144                 sub_frames = frames[offset : offset + sub_header["num-sub-frames"]]
145                 if "compression" in sub_header:
146                     sub_frames = decompress(sub_header, sub_frames)
147                 return pickle.loads(sub_header["pickled-obj"], buffers=sub_frames)
148             return msgpack_decode_default(obj)
149 
150         return msgpack.loads(
151             frames[0], object_hook=_decode_default, use_list=False, **msgpack_opts
152         )
153 
154     except Exception:
155         logger.critical("Failed to deserialize", exc_info=True)
156         raise
157
```
--- 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/core.py b/distributed/protocol/core.py
--- a/distributed/protocol/core.py
+++ b/distributed/protocol/core.py
@@ -11,6 +11,7 @@
     Serialize,
     Serialized,
     ToPickle,
+    _is_msgpack_serializable,
     merge_and_deserialize,
     msgpack_decode_default,
     msgpack_encode_default,
@@ -104,7 +105,31 @@
             else:
                 return msgpack_encode_default(obj)
 
-        frames[0] = msgpack.dumps(msg, default=_encode_default, use_bin_type=True)
+        try:
+            frames[0] = msgpack.dumps(msg, default=_encode_default, use_bin_type=True)
+        except TypeError as e:
+            logger.info(
+                f"Failed to serialize ({e}); falling back to pickle. "
+                "Be aware that this may degrade performance."
+            )
+
+            def _encode_default_safe(obj):
+                encoded = _encode_default(obj)
+                if encoded is not obj or _is_msgpack_serializable(obj):
+                    return encoded
+
+                obj = ToPickle(obj)
+                offset = len(frames)
+                frames.extend(create_pickled_sub_frames(obj))
+                return {"__Pickled__": offset}
+
+            # If possible, we want to avoid the performance penalty from the checks
+            # implemented in _encode_default_safe to fall back to pickle, so we
+            # try to serialize the data without the fallback first assuming that
+            # this succeeds in the overwhelming majority of cases.
+            frames[0] = msgpack.dumps(
+                msg, default=_encode_default_safe, use_bin_type=True
+            )
         return frames
 
     except Exception:

{"golden_diff": "diff --git a/distributed/protocol/core.py b/distributed/protocol/core.py\n--- a/distributed/protocol/core.py\n+++ b/distributed/protocol/core.py\n@@ -11,6 +11,7 @@\n     Serialize,\n     Serialized,\n     ToPickle,\n+    _is_msgpack_serializable,\n     merge_and_deserialize,\n     msgpack_decode_default,\n     msgpack_encode_default,\n@@ -104,7 +105,31 @@\n             else:\n                 return msgpack_encode_default(obj)\n \n-        frames[0] = msgpack.dumps(msg, default=_encode_default, use_bin_type=True)\n+        try:\n+            frames[0] = msgpack.dumps(msg, default=_encode_default, use_bin_type=True)\n+        except TypeError as e:\n+            logger.info(\n+                f\"Failed to serialize ({e}); falling back to pickle. \"\n+                \"Be aware that this may degrade performance.\"\n+            )\n+\n+            def _encode_default_safe(obj):\n+                encoded = _encode_default(obj)\n+                if encoded is not obj or _is_msgpack_serializable(obj):\n+                    return encoded\n+\n+                obj = ToPickle(obj)\n+                offset = len(frames)\n+                frames.extend(create_pickled_sub_frames(obj))\n+                return {\"__Pickled__\": offset}\n+\n+            # If possible, we want to avoid the performance penalty from the checks\n+            # implemented in _encode_default_safe to fall back to pickle, so we\n+            # try to serialize the data without the fallback first assuming that\n+            # this succeeds in the overwhelming majority of cases.\n+            frames[0] = msgpack.dumps(\n+                msg, default=_encode_default_safe, use_bin_type=True\n+            )\n         return frames\n \n     except Exception:\n", "issue": "`distributed.protocol.dumps` does not handle objects not serializable via `msgpack.dumps` as suggested by docstring\nWhen using `distributed.protocol.dumps` to serialize arbitrary data, it fails and raises a `TypeError`. The docstring suggests it should be able to handle this.\r\n\r\n**Minimal Complete Verifiable Example**:\r\n\r\n```python\r\nimport numpy as np\r\nfrom distributed.protocol import dumps\r\n\r\ndumps(np.int64(123))\r\n```\r\n\r\nraises\r\n\r\n```python\r\n2024-01-09 19:55:04,572 - distributed.protocol.core - CRITICAL - Failed to Serialize\r\nTraceback (most recent call last):\r\n  File \"/Users/hendrikmakait/projects/dask/distributed/distributed/protocol/core.py\", line 108, in dumps\r\n    frames[0] = msgpack.dumps(msg, default=_encode_default, use_bin_type=True)\r\n                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\r\n  File \"/opt/homebrew/Caskroom/mambaforge/base/envs/dask-distributed/lib/python3.12/site-packages/msgpack/__init__.py\", line 36, in packb\r\n    return Packer(**kwargs).pack(o)\r\n           ^^^^^^^^^^^^^^^^^^^^^^^^\r\n  File \"msgpack/_packer.pyx\", line 294, in msgpack._cmsgpack.Packer.pack\r\n  File \"msgpack/_packer.pyx\", line 300, in msgpack._cmsgpack.Packer.pack\r\n  File \"msgpack/_packer.pyx\", line 297, in msgpack._cmsgpack.Packer.pack\r\n  File \"msgpack/_packer.pyx\", line 291, in msgpack._cmsgpack.Packer._pack\r\nTypeError: can not serialize 'numpy.int64' object\r\nTraceback (most recent call last):\r\n  File \"/Users/hendrikmakait/projects/dask/distributed/sandbox.py\", line 5, in <module>\r\n    dumps(np.int64(123))\r\n  File \"/Users/hendrikmakait/projects/dask/distributed/distributed/protocol/core.py\", line 108, in dumps\r\n    frames[0] = msgpack.dumps(msg, default=_encode_default, use_bin_type=True)\r\n                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\r\n  File \"/opt/homebrew/Caskroom/mambaforge/base/envs/dask-distributed/lib/python3.12/site-packages/msgpack/__init__.py\", line 36, in packb\r\n    return Packer(**kwargs).pack(o)\r\n           ^^^^^^^^^^^^^^^^^^^^^^^^\r\n  File \"msgpack/_packer.pyx\", line 294, in msgpack._cmsgpack.Packer.pack\r\n  File \"msgpack/_packer.pyx\", line 300, in msgpack._cmsgpack.Packer.pack\r\n  File \"msgpack/_packer.pyx\", line 297, in msgpack._cmsgpack.Packer.pack\r\n  File \"msgpack/_packer.pyx\", line 291, in msgpack._cmsgpack.Packer._pack\r\nTypeError: can not serialize 'numpy.int64' object\r\n```\r\n\r\nRelevant section of `dumps`'s docstring: https://github.com/dask/distributed/blob/7562f9c566978de4f3f5b73920a24ea1813d6e28/distributed/protocol/core.py#L30-L35\n", "before_files": [{"content": "from __future__ import annotations\n\nimport logging\n\nimport msgpack\n\nfrom distributed.protocol import pickle\nfrom distributed.protocol.compression import decompress, maybe_compress\nfrom distributed.protocol.serialize import (\n    Pickled,\n    Serialize,\n    Serialized,\n    ToPickle,\n    merge_and_deserialize,\n    msgpack_decode_default,\n    msgpack_encode_default,\n    serialize_and_split,\n)\nfrom distributed.protocol.utils import msgpack_opts\nfrom distributed.utils import ensure_memoryview\n\nlogger = logging.getLogger(__name__)\n\n\ndef dumps(  # type: ignore[no-untyped-def]\n    msg, serializers=None, on_error=\"message\", context=None, frame_split_size=None\n) -> list:\n    \"\"\"Transform Python message to bytestream suitable for communication\n\n    Developer Notes\n    ---------------\n    The approach here is to use `msgpack.dumps()` to serialize `msg` and\n    write the result to the first output frame. If `msgpack.dumps()`\n    encounters an object it cannot serialize like a NumPy array, it is handled\n    out-of-band by `_encode_default()` and appended to the output frame list.\n    \"\"\"\n    try:\n        if context and \"compression\" in context:\n            compress_opts = {\"compression\": context[\"compression\"]}\n        else:\n            compress_opts = {}\n\n        def _inplace_compress_frames(header, frames):\n            compression = list(header.get(\"compression\", [None] * len(frames)))\n\n            for i in range(len(frames)):\n                if compression[i] is None:\n                    compression[i], frames[i] = maybe_compress(\n                        frames[i], **compress_opts\n                    )\n\n            header[\"compression\"] = tuple(compression)\n\n        def create_serialized_sub_frames(obj: Serialized | Serialize) -> list:\n            if isinstance(obj, Serialized):\n                sub_header, sub_frames = obj.header, obj.frames\n            else:\n                sub_header, sub_frames = serialize_and_split(\n                    obj,\n                    serializers=serializers,\n                    on_error=on_error,\n                    context=context,\n                    size=frame_split_size,\n                )\n                _inplace_compress_frames(sub_header, sub_frames)\n            sub_header[\"num-sub-frames\"] = len(sub_frames)\n            sub_header = msgpack.dumps(\n                sub_header, default=msgpack_encode_default, use_bin_type=True\n            )\n            return [sub_header] + sub_frames\n\n        def create_pickled_sub_frames(obj: Pickled | ToPickle) -> list:\n            if isinstance(obj, Pickled):\n                sub_header, sub_frames = obj.header, obj.frames\n            else:\n                sub_frames = []\n                sub_header = {\n                    \"pickled-obj\": pickle.dumps(\n                        obj.data,\n                        # In to support len() and slicing, we convert `PickleBuffer`\n                        # objects to memoryviews of bytes.\n                        buffer_callback=lambda x: sub_frames.append(\n                            ensure_memoryview(x)\n                        ),\n                    )\n                }\n                _inplace_compress_frames(sub_header, sub_frames)\n\n            sub_header[\"num-sub-frames\"] = len(sub_frames)\n            sub_header = msgpack.dumps(sub_header)\n            return [sub_header] + sub_frames\n\n        frames = [None]\n\n        def _encode_default(obj):\n            if isinstance(obj, (Serialize, Serialized)):\n                offset = len(frames)\n                frames.extend(create_serialized_sub_frames(obj))\n                return {\"__Serialized__\": offset}\n            elif isinstance(obj, (ToPickle, Pickled)):\n                offset = len(frames)\n                frames.extend(create_pickled_sub_frames(obj))\n                return {\"__Pickled__\": offset}\n            else:\n                return msgpack_encode_default(obj)\n\n        frames[0] = msgpack.dumps(msg, default=_encode_default, use_bin_type=True)\n        return frames\n\n    except Exception:\n        logger.critical(\"Failed to Serialize\", exc_info=True)\n        raise\n\n\ndef loads(frames, deserialize=True, deserializers=None):\n    \"\"\"Transform bytestream back into Python value\"\"\"\n\n    try:\n\n        def _decode_default(obj):\n            offset = obj.get(\"__Serialized__\", 0)\n            if offset > 0:\n                sub_header = msgpack.loads(\n                    frames[offset],\n                    object_hook=msgpack_decode_default,\n                    use_list=False,\n                    **msgpack_opts,\n                )\n                offset += 1\n                sub_frames = frames[offset : offset + sub_header[\"num-sub-frames\"]]\n                if deserialize:\n                    if \"compression\" in sub_header:\n                        sub_frames = decompress(sub_header, sub_frames)\n                    return merge_and_deserialize(\n                        sub_header, sub_frames, deserializers=deserializers\n                    )\n                else:\n                    return Serialized(sub_header, sub_frames)\n\n            offset = obj.get(\"__Pickled__\", 0)\n            if offset > 0:\n                sub_header = msgpack.loads(frames[offset])\n                offset += 1\n                sub_frames = frames[offset : offset + sub_header[\"num-sub-frames\"]]\n                if \"compression\" in sub_header:\n                    sub_frames = decompress(sub_header, sub_frames)\n                return pickle.loads(sub_header[\"pickled-obj\"], buffers=sub_frames)\n            return msgpack_decode_default(obj)\n\n        return msgpack.loads(\n            frames[0], object_hook=_decode_default, use_list=False, **msgpack_opts\n        )\n\n    except Exception:\n        logger.critical(\"Failed to deserialize\", exc_info=True)\n        raise\n", "path": "distributed/protocol/core.py"}], "after_files": [{"content": "from __future__ import annotations\n\nimport logging\n\nimport msgpack\n\nfrom distributed.protocol import pickle\nfrom distributed.protocol.compression import decompress, maybe_compress\nfrom distributed.protocol.serialize import (\n    Pickled,\n    Serialize,\n    Serialized,\n    ToPickle,\n    _is_msgpack_serializable,\n    merge_and_deserialize,\n    msgpack_decode_default,\n    msgpack_encode_default,\n    serialize_and_split,\n)\nfrom distributed.protocol.utils import msgpack_opts\nfrom distributed.utils import ensure_memoryview\n\nlogger = logging.getLogger(__name__)\n\n\ndef dumps(  # type: ignore[no-untyped-def]\n    msg, serializers=None, on_error=\"message\", context=None, frame_split_size=None\n) -> list:\n    \"\"\"Transform Python message to bytestream suitable for communication\n\n    Developer Notes\n    ---------------\n    The approach here is to use `msgpack.dumps()` to serialize `msg` and\n    write the result to the first output frame. If `msgpack.dumps()`\n    encounters an object it cannot serialize like a NumPy array, it is handled\n    out-of-band by `_encode_default()` and appended to the output frame list.\n    \"\"\"\n    try:\n        if context and \"compression\" in context:\n            compress_opts = {\"compression\": context[\"compression\"]}\n        else:\n            compress_opts = {}\n\n        def _inplace_compress_frames(header, frames):\n            compression = list(header.get(\"compression\", [None] * len(frames)))\n\n            for i in range(len(frames)):\n                if compression[i] is None:\n                    compression[i], frames[i] = maybe_compress(\n                        frames[i], **compress_opts\n                    )\n\n            header[\"compression\"] = tuple(compression)\n\n        def create_serialized_sub_frames(obj: Serialized | Serialize) -> list:\n            if isinstance(obj, Serialized):\n                sub_header, sub_frames = obj.header, obj.frames\n            else:\n                sub_header, sub_frames = serialize_and_split(\n                    obj,\n                    serializers=serializers,\n                    on_error=on_error,\n                    context=context,\n                    size=frame_split_size,\n                )\n                _inplace_compress_frames(sub_header, sub_frames)\n            sub_header[\"num-sub-frames\"] = len(sub_frames)\n            sub_header = msgpack.dumps(\n                sub_header, default=msgpack_encode_default, use_bin_type=True\n            )\n            return [sub_header] + sub_frames\n\n        def create_pickled_sub_frames(obj: Pickled | ToPickle) -> list:\n            if isinstance(obj, Pickled):\n                sub_header, sub_frames = obj.header, obj.frames\n            else:\n                sub_frames = []\n                sub_header = {\n                    \"pickled-obj\": pickle.dumps(\n                        obj.data,\n                        # In to support len() and slicing, we convert `PickleBuffer`\n                        # objects to memoryviews of bytes.\n                        buffer_callback=lambda x: sub_frames.append(\n                            ensure_memoryview(x)\n                        ),\n                    )\n                }\n                _inplace_compress_frames(sub_header, sub_frames)\n\n            sub_header[\"num-sub-frames\"] = len(sub_frames)\n            sub_header = msgpack.dumps(sub_header)\n            return [sub_header] + sub_frames\n\n        frames = [None]\n\n        def _encode_default(obj):\n            if isinstance(obj, (Serialize, Serialized)):\n                offset = len(frames)\n                frames.extend(create_serialized_sub_frames(obj))\n                return {\"__Serialized__\": offset}\n            elif isinstance(obj, (ToPickle, Pickled)):\n                offset = len(frames)\n                frames.extend(create_pickled_sub_frames(obj))\n                return {\"__Pickled__\": offset}\n            else:\n                return msgpack_encode_default(obj)\n\n        try:\n            frames[0] = msgpack.dumps(msg, default=_encode_default, use_bin_type=True)\n        except TypeError as e:\n            logger.info(\n                f\"Failed to serialize ({e}); falling back to pickle. \"\n                \"Be aware that this may degrade performance.\"\n            )\n\n            def _encode_default_safe(obj):\n                encoded = _encode_default(obj)\n                if encoded is not obj or _is_msgpack_serializable(obj):\n                    return encoded\n\n                obj = ToPickle(obj)\n                offset = len(frames)\n                frames.extend(create_pickled_sub_frames(obj))\n                return {\"__Pickled__\": offset}\n\n            # If possible, we want to avoid the performance penalty from the checks\n            # implemented in _encode_default_safe to fall back to pickle, so we\n            # try to serialize the data without the fallback first assuming that\n            # this succeeds in the overwhelming majority of cases.\n            frames[0] = msgpack.dumps(\n                msg, default=_encode_default_safe, use_bin_type=True\n            )\n        return frames\n\n    except Exception:\n        logger.critical(\"Failed to Serialize\", exc_info=True)\n        raise\n\n\ndef loads(frames, deserialize=True, deserializers=None):\n    \"\"\"Transform bytestream back into Python value\"\"\"\n\n    try:\n\n        def _decode_default(obj):\n            offset = obj.get(\"__Serialized__\", 0)\n            if offset > 0:\n                sub_header = msgpack.loads(\n                    frames[offset],\n                    object_hook=msgpack_decode_default,\n                    use_list=False,\n                    **msgpack_opts,\n                )\n                offset += 1\n                sub_frames = frames[offset : offset + sub_header[\"num-sub-frames\"]]\n                if deserialize:\n                    if \"compression\" in sub_header:\n                        sub_frames = decompress(sub_header, sub_frames)\n                    return merge_and_deserialize(\n                        sub_header, sub_frames, deserializers=deserializers\n                    )\n                else:\n                    return Serialized(sub_header, sub_frames)\n\n            offset = obj.get(\"__Pickled__\", 0)\n            if offset > 0:\n                sub_header = msgpack.loads(frames[offset])\n                offset += 1\n                sub_frames = frames[offset : offset + sub_header[\"num-sub-frames\"]]\n                if \"compression\" in sub_header:\n                    sub_frames = decompress(sub_header, sub_frames)\n                return pickle.loads(sub_header[\"pickled-obj\"], buffers=sub_frames)\n            return msgpack_decode_default(obj)\n\n        return msgpack.loads(\n            frames[0], object_hook=_decode_default, use_list=False, **msgpack_opts\n        )\n\n    except Exception:\n        logger.critical(\"Failed to deserialize\", exc_info=True)\n        raise\n", "path": "distributed/protocol/core.py"}]}