We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
cannot set WRITEABLE flag to True of this array
My colleague is seeing the following error on distirbuted 2.8.0. It looks related to #3180 and this set of lines

https://github.com/dask/distributed/blob/35551998d7350cd5ae6a5c24970d8437fd8d521d/distributed/protocol/numpy.py#L110-L114


```
distributed.worker - INFO - Start worker at: tcp://10.12.205.19:34307

distributed.worker - INFO - Listening to: tcp://10.12.205.19:34307

distributed.worker - INFO - dashboard at: 10.12.205.19:34094

distributed.worker - INFO - Waiting to connect to: tcp://128.117.181.211:37309

distributed.worker - INFO - -------------------------------------------------

distributed.worker - INFO - Threads: 1

distributed.worker - INFO - Memory: 25.00 GB

distributed.worker - INFO - Local Directory: /glade/scratch/deppenme/dask-tmp/worker-k05sxxku

distributed.worker - INFO - -------------------------------------------------

distributed.worker - INFO - Registered to: tcp://128.117.181.211:37309

distributed.worker - INFO - -------------------------------------------------

distributed.worker - ERROR - cannot set WRITEABLE flag to True of this array 

Traceback (most recent call last): File "/glade/work/deppenme/miniconda3/envs/analysis/lib/python3.7/site-packages/distributed/worker.py", line 894, in handle_scheduler comm, every_cycle=[self.ensure_communicating, self.ensure_computing] File "/glade/work/deppenme/miniconda3/envs/analysis/lib/python3.7/site-packages/distributed/core.py", line 447, in handle_stream msgs = await comm.read() File "/glade/work/deppenme/miniconda3/envs/analysis/lib/python3.7/site-packages/distributed/comm/tcp.py", line 208, in read frames, deserialize=self.deserialize, deserializers=deserializers 
File "/glade/work/deppenme/miniconda3/envs/analysis/lib/python3.7/site-packages/distributed/comm/utils.py", line 63, in from_frames res = await offload(_from_frames) File "/glade/work/deppenme/miniconda3/envs/analysis/lib/python3.7/site-packages/distributed/utils.py", line 1371, in offload return await loop.run_in_executor(_offload_executor, fn, *args, **kwargs) File "/glade/work/deppenme/miniconda3/envs/analysis/lib/python3.7/concurrent/futures/thread.py", line 57, in run result = self.fn(*self.args, **self.kwargs) 
File "/glade/work/deppenme/miniconda3/envs/analysis/lib/python3.7/site-packages/distributed/comm/utils.py", line 51, in _from_frames frames, deserialize=deserialize, deserializers=deserializers File "/glade/work/deppenme/miniconda3/envs/analysis/lib/python3.7/site-packages/distributed/protocol/core.py", line 124, in loads value = _deserialize(head, fs, deserializers=deserializers) 
File "/glade/work/deppenme/miniconda3/envs/analysis/lib/python3.7/site-packages/distributed/protocol/serialize.py", line 268, in deserialize return loads(header, frames) 
File "/glade/work/deppenme/miniconda3/envs/analysis/lib/python3.7/site-packages/distributed/protocol/serialize.py", line 54, in dask_loads return loads(header, frames) 
File "/glade/work/deppenme/miniconda3/envs/analysis/lib/python3.7/site-packages/distributed/protocol/numpy.py",
 line 113, in deserialize_numpy_ndarray x.setflags(write=writeable) ValueError: cannot set WRITEABLE flag to True of this array
```

This is coming out of a complicated analysis pipeline with xarray, zarr and dask so we don't have a minimal example yet. We could work to find one if you have some pointers on what to look for.

--- 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/numpy.py`
Content:
```
1 import math
2 import numpy as np
3 
4 from .utils import frame_split_size, merge_frames
5 from .serialize import dask_serialize, dask_deserialize
6 from . import pickle
7 
8 from ..utils import log_errors
9 
10 
11 def itemsize(dt):
12     """ Itemsize of dtype
13 
14     Try to return the itemsize of the base element, return 8 as a fallback
15     """
16     result = dt.base.itemsize
17     if result > 255:
18         result = 8
19     return result
20 
21 
22 @dask_serialize.register(np.ndarray)
23 def serialize_numpy_ndarray(x):
24     if x.dtype.hasobject:
25         header = {"pickle": True}
26         frames = [pickle.dumps(x)]
27         return header, frames
28 
29     # We cannot blindly pickle the dtype as some may fail pickling,
30     # so we have a mixture of strategies.
31     if x.dtype.kind == "V":
32         # Preserving all the information works best when pickling
33         try:
34             # Only use stdlib pickle as cloudpickle is slow when failing
35             # (microseconds instead of nanoseconds)
36             dt = (1, pickle.pickle.dumps(x.dtype))
37             pickle.loads(dt[1])  # does it unpickle fine?
38         except Exception:
39             # dtype fails pickling => fall back on the descr if reasonable.
40             if x.dtype.type is not np.void or x.dtype.alignment != 1:
41                 raise
42             else:
43                 dt = (0, x.dtype.descr)
44     else:
45         dt = (0, x.dtype.str)
46 
47     # Only serialize non-broadcasted data for arrays with zero strided axes
48     if 0 in x.strides:
49         broadcast_to = (x.shape, x.flags.writeable)
50         x = x[tuple(slice(None) if s != 0 else slice(1) for s in x.strides)]
51     else:
52         broadcast_to = None
53 
54     if not x.shape:
55         # 0d array
56         strides = x.strides
57         data = x.ravel()
58     elif x.flags.c_contiguous or x.flags.f_contiguous:
59         # Avoid a copy and respect order when unserializing
60         strides = x.strides
61         data = x.ravel(order="K")
62     else:
63         x = np.ascontiguousarray(x)
64         strides = x.strides
65         data = x.ravel()
66 
67     if data.dtype.fields or data.dtype.itemsize > 8:
68         data = data.view("u%d" % math.gcd(x.dtype.itemsize, 8))
69 
70     try:
71         data = data.data
72     except ValueError:
73         # "ValueError: cannot include dtype 'M' in a buffer"
74         data = data.view("u%d" % math.gcd(x.dtype.itemsize, 8)).data
75 
76     header = {"dtype": dt, "shape": x.shape, "strides": strides}
77 
78     if broadcast_to is not None:
79         header["broadcast_to"] = broadcast_to
80 
81     if x.nbytes > 1e5:
82         frames = frame_split_size([data])
83     else:
84         frames = [data]
85 
86     header["lengths"] = [x.nbytes]
87 
88     return header, frames
89 
90 
91 @dask_deserialize.register(np.ndarray)
92 def deserialize_numpy_ndarray(header, frames):
93     with log_errors():
94         if len(frames) > 1:
95             frames = merge_frames(header, frames)
96 
97         if header.get("pickle"):
98             return pickle.loads(frames[0])
99 
100         is_custom, dt = header["dtype"]
101         if is_custom:
102             dt = pickle.loads(dt)
103         else:
104             dt = np.dtype(dt)
105 
106         x = np.ndarray(
107             header["shape"], dtype=dt, buffer=frames[0], strides=header["strides"]
108         )
109 
110         if header.get("broadcast_to"):
111             shape, writeable = header["broadcast_to"]
112             x = np.broadcast_to(x, shape)
113             x.setflags(write=writeable)
114 
115         return x
116 
117 
118 @dask_serialize.register(np.ma.core.MaskedConstant)
119 def serialize_numpy_ma_masked(x):
120     return {}, []
121 
122 
123 @dask_deserialize.register(np.ma.core.MaskedConstant)
124 def deserialize_numpy_ma_masked(header, frames):
125     return np.ma.masked
126 
127 
128 @dask_serialize.register(np.ma.core.MaskedArray)
129 def serialize_numpy_maskedarray(x):
130     data_header, frames = serialize_numpy_ndarray(x.data)
131     header = {"data-header": data_header, "nframes": len(frames)}
132 
133     # Serialize mask if present
134     if x.mask is not np.ma.nomask:
135         mask_header, mask_frames = serialize_numpy_ndarray(x.mask)
136         header["mask-header"] = mask_header
137         frames += mask_frames
138 
139     # Only a few dtypes have python equivalents msgpack can serialize
140     if isinstance(x.fill_value, (np.integer, np.floating, np.bool_)):
141         serialized_fill_value = (False, x.fill_value.item())
142     else:
143         serialized_fill_value = (True, pickle.dumps(x.fill_value))
144     header["fill-value"] = serialized_fill_value
145 
146     return header, frames
147 
148 
149 @dask_deserialize.register(np.ma.core.MaskedArray)
150 def deserialize_numpy_maskedarray(header, frames):
151     data_header = header["data-header"]
152     data_frames = frames[: header["nframes"]]
153     data = deserialize_numpy_ndarray(data_header, data_frames)
154 
155     if "mask-header" in header:
156         mask_header = header["mask-header"]
157         mask_frames = frames[header["nframes"] :]
158         mask = deserialize_numpy_ndarray(mask_header, mask_frames)
159     else:
160         mask = np.ma.nomask
161 
162     pickled_fv, fill_value = header["fill-value"]
163     if pickled_fv:
164         fill_value = pickle.loads(fill_value)
165 
166     return np.ma.masked_array(data, mask=mask, fill_value=fill_value)
167
```
--- 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/numpy.py b/distributed/protocol/numpy.py
--- a/distributed/protocol/numpy.py
+++ b/distributed/protocol/numpy.py
@@ -46,7 +46,7 @@
 
     # Only serialize non-broadcasted data for arrays with zero strided axes
     if 0 in x.strides:
-        broadcast_to = (x.shape, x.flags.writeable)
+        broadcast_to = x.shape
         x = x[tuple(slice(None) if s != 0 else slice(1) for s in x.strides)]
     else:
         broadcast_to = None
@@ -103,14 +103,12 @@
         else:
             dt = np.dtype(dt)
 
-        x = np.ndarray(
-            header["shape"], dtype=dt, buffer=frames[0], strides=header["strides"]
-        )
-
         if header.get("broadcast_to"):
-            shape, writeable = header["broadcast_to"]
-            x = np.broadcast_to(x, shape)
-            x.setflags(write=writeable)
+            shape = header["broadcast_to"]
+        else:
+            shape = header["shape"]
+
+        x = np.ndarray(shape, dtype=dt, buffer=frames[0], strides=header["strides"])
 
         return x

{"golden_diff": "diff --git a/distributed/protocol/numpy.py b/distributed/protocol/numpy.py\n--- a/distributed/protocol/numpy.py\n+++ b/distributed/protocol/numpy.py\n@@ -46,7 +46,7 @@\n \n     # Only serialize non-broadcasted data for arrays with zero strided axes\n     if 0 in x.strides:\n-        broadcast_to = (x.shape, x.flags.writeable)\n+        broadcast_to = x.shape\n         x = x[tuple(slice(None) if s != 0 else slice(1) for s in x.strides)]\n     else:\n         broadcast_to = None\n@@ -103,14 +103,12 @@\n         else:\n             dt = np.dtype(dt)\n \n-        x = np.ndarray(\n-            header[\"shape\"], dtype=dt, buffer=frames[0], strides=header[\"strides\"]\n-        )\n-\n         if header.get(\"broadcast_to\"):\n-            shape, writeable = header[\"broadcast_to\"]\n-            x = np.broadcast_to(x, shape)\n-            x.setflags(write=writeable)\n+            shape = header[\"broadcast_to\"]\n+        else:\n+            shape = header[\"shape\"]\n+\n+        x = np.ndarray(shape, dtype=dt, buffer=frames[0], strides=header[\"strides\"])\n \n         return x\n", "issue": "cannot set WRITEABLE flag to True of this array\nMy colleague is seeing the following error on distirbuted 2.8.0. It looks related to #3180 and this set of lines\r\n\r\nhttps://github.com/dask/distributed/blob/35551998d7350cd5ae6a5c24970d8437fd8d521d/distributed/protocol/numpy.py#L110-L114\r\n\r\n\r\n```\r\ndistributed.worker - INFO - Start worker at: tcp://10.12.205.19:34307\r\n\r\ndistributed.worker - INFO - Listening to: tcp://10.12.205.19:34307\r\n\r\ndistributed.worker - INFO - dashboard at: 10.12.205.19:34094\r\n\r\ndistributed.worker - INFO - Waiting to connect to: tcp://128.117.181.211:37309\r\n\r\ndistributed.worker - INFO - -------------------------------------------------\r\n\r\ndistributed.worker - INFO - Threads: 1\r\n\r\ndistributed.worker - INFO - Memory: 25.00 GB\r\n\r\ndistributed.worker - INFO - Local Directory: /glade/scratch/deppenme/dask-tmp/worker-k05sxxku\r\n\r\ndistributed.worker - INFO - -------------------------------------------------\r\n\r\ndistributed.worker - INFO - Registered to: tcp://128.117.181.211:37309\r\n\r\ndistributed.worker - INFO - -------------------------------------------------\r\n\r\ndistributed.worker - ERROR - cannot set WRITEABLE flag to True of this array \r\n\r\nTraceback (most recent call last): File \"/glade/work/deppenme/miniconda3/envs/analysis/lib/python3.7/site-packages/distributed/worker.py\", line 894, in handle_scheduler comm, every_cycle=[self.ensure_communicating, self.ensure_computing] File \"/glade/work/deppenme/miniconda3/envs/analysis/lib/python3.7/site-packages/distributed/core.py\", line 447, in handle_stream msgs = await comm.read() File \"/glade/work/deppenme/miniconda3/envs/analysis/lib/python3.7/site-packages/distributed/comm/tcp.py\", line 208, in read frames, deserialize=self.deserialize, deserializers=deserializers \r\nFile \"/glade/work/deppenme/miniconda3/envs/analysis/lib/python3.7/site-packages/distributed/comm/utils.py\", line 63, in from_frames res = await offload(_from_frames) File \"/glade/work/deppenme/miniconda3/envs/analysis/lib/python3.7/site-packages/distributed/utils.py\", line 1371, in offload return await loop.run_in_executor(_offload_executor, fn, *args, **kwargs) File \"/glade/work/deppenme/miniconda3/envs/analysis/lib/python3.7/concurrent/futures/thread.py\", line 57, in run result = self.fn(*self.args, **self.kwargs) \r\nFile \"/glade/work/deppenme/miniconda3/envs/analysis/lib/python3.7/site-packages/distributed/comm/utils.py\", line 51, in _from_frames frames, deserialize=deserialize, deserializers=deserializers File \"/glade/work/deppenme/miniconda3/envs/analysis/lib/python3.7/site-packages/distributed/protocol/core.py\", line 124, in loads value = _deserialize(head, fs, deserializers=deserializers) \r\nFile \"/glade/work/deppenme/miniconda3/envs/analysis/lib/python3.7/site-packages/distributed/protocol/serialize.py\", line 268, in deserialize return loads(header, frames) \r\nFile \"/glade/work/deppenme/miniconda3/envs/analysis/lib/python3.7/site-packages/distributed/protocol/serialize.py\", line 54, in dask_loads return loads(header, frames) \r\nFile \"/glade/work/deppenme/miniconda3/envs/analysis/lib/python3.7/site-packages/distributed/protocol/numpy.py\",\r\n line 113, in deserialize_numpy_ndarray x.setflags(write=writeable) ValueError: cannot set WRITEABLE flag to True of this array\r\n```\r\n\r\nThis is coming out of a complicated analysis pipeline with xarray, zarr and dask so we don't have a minimal example yet. We could work to find one if you have some pointers on what to look for.\n", "before_files": [{"content": "import math\nimport numpy as np\n\nfrom .utils import frame_split_size, merge_frames\nfrom .serialize import dask_serialize, dask_deserialize\nfrom . import pickle\n\nfrom ..utils import log_errors\n\n\ndef itemsize(dt):\n    \"\"\" Itemsize of dtype\n\n    Try to return the itemsize of the base element, return 8 as a fallback\n    \"\"\"\n    result = dt.base.itemsize\n    if result > 255:\n        result = 8\n    return result\n\n\n@dask_serialize.register(np.ndarray)\ndef serialize_numpy_ndarray(x):\n    if x.dtype.hasobject:\n        header = {\"pickle\": True}\n        frames = [pickle.dumps(x)]\n        return header, frames\n\n    # We cannot blindly pickle the dtype as some may fail pickling,\n    # so we have a mixture of strategies.\n    if x.dtype.kind == \"V\":\n        # Preserving all the information works best when pickling\n        try:\n            # Only use stdlib pickle as cloudpickle is slow when failing\n            # (microseconds instead of nanoseconds)\n            dt = (1, pickle.pickle.dumps(x.dtype))\n            pickle.loads(dt[1])  # does it unpickle fine?\n        except Exception:\n            # dtype fails pickling => fall back on the descr if reasonable.\n            if x.dtype.type is not np.void or x.dtype.alignment != 1:\n                raise\n            else:\n                dt = (0, x.dtype.descr)\n    else:\n        dt = (0, x.dtype.str)\n\n    # Only serialize non-broadcasted data for arrays with zero strided axes\n    if 0 in x.strides:\n        broadcast_to = (x.shape, x.flags.writeable)\n        x = x[tuple(slice(None) if s != 0 else slice(1) for s in x.strides)]\n    else:\n        broadcast_to = None\n\n    if not x.shape:\n        # 0d array\n        strides = x.strides\n        data = x.ravel()\n    elif x.flags.c_contiguous or x.flags.f_contiguous:\n        # Avoid a copy and respect order when unserializing\n        strides = x.strides\n        data = x.ravel(order=\"K\")\n    else:\n        x = np.ascontiguousarray(x)\n        strides = x.strides\n        data = x.ravel()\n\n    if data.dtype.fields or data.dtype.itemsize > 8:\n        data = data.view(\"u%d\" % math.gcd(x.dtype.itemsize, 8))\n\n    try:\n        data = data.data\n    except ValueError:\n        # \"ValueError: cannot include dtype 'M' in a buffer\"\n        data = data.view(\"u%d\" % math.gcd(x.dtype.itemsize, 8)).data\n\n    header = {\"dtype\": dt, \"shape\": x.shape, \"strides\": strides}\n\n    if broadcast_to is not None:\n        header[\"broadcast_to\"] = broadcast_to\n\n    if x.nbytes > 1e5:\n        frames = frame_split_size([data])\n    else:\n        frames = [data]\n\n    header[\"lengths\"] = [x.nbytes]\n\n    return header, frames\n\n\n@dask_deserialize.register(np.ndarray)\ndef deserialize_numpy_ndarray(header, frames):\n    with log_errors():\n        if len(frames) > 1:\n            frames = merge_frames(header, frames)\n\n        if header.get(\"pickle\"):\n            return pickle.loads(frames[0])\n\n        is_custom, dt = header[\"dtype\"]\n        if is_custom:\n            dt = pickle.loads(dt)\n        else:\n            dt = np.dtype(dt)\n\n        x = np.ndarray(\n            header[\"shape\"], dtype=dt, buffer=frames[0], strides=header[\"strides\"]\n        )\n\n        if header.get(\"broadcast_to\"):\n            shape, writeable = header[\"broadcast_to\"]\n            x = np.broadcast_to(x, shape)\n            x.setflags(write=writeable)\n\n        return x\n\n\n@dask_serialize.register(np.ma.core.MaskedConstant)\ndef serialize_numpy_ma_masked(x):\n    return {}, []\n\n\n@dask_deserialize.register(np.ma.core.MaskedConstant)\ndef deserialize_numpy_ma_masked(header, frames):\n    return np.ma.masked\n\n\n@dask_serialize.register(np.ma.core.MaskedArray)\ndef serialize_numpy_maskedarray(x):\n    data_header, frames = serialize_numpy_ndarray(x.data)\n    header = {\"data-header\": data_header, \"nframes\": len(frames)}\n\n    # Serialize mask if present\n    if x.mask is not np.ma.nomask:\n        mask_header, mask_frames = serialize_numpy_ndarray(x.mask)\n        header[\"mask-header\"] = mask_header\n        frames += mask_frames\n\n    # Only a few dtypes have python equivalents msgpack can serialize\n    if isinstance(x.fill_value, (np.integer, np.floating, np.bool_)):\n        serialized_fill_value = (False, x.fill_value.item())\n    else:\n        serialized_fill_value = (True, pickle.dumps(x.fill_value))\n    header[\"fill-value\"] = serialized_fill_value\n\n    return header, frames\n\n\n@dask_deserialize.register(np.ma.core.MaskedArray)\ndef deserialize_numpy_maskedarray(header, frames):\n    data_header = header[\"data-header\"]\n    data_frames = frames[: header[\"nframes\"]]\n    data = deserialize_numpy_ndarray(data_header, data_frames)\n\n    if \"mask-header\" in header:\n        mask_header = header[\"mask-header\"]\n        mask_frames = frames[header[\"nframes\"] :]\n        mask = deserialize_numpy_ndarray(mask_header, mask_frames)\n    else:\n        mask = np.ma.nomask\n\n    pickled_fv, fill_value = header[\"fill-value\"]\n    if pickled_fv:\n        fill_value = pickle.loads(fill_value)\n\n    return np.ma.masked_array(data, mask=mask, fill_value=fill_value)\n", "path": "distributed/protocol/numpy.py"}], "after_files": [{"content": "import math\nimport numpy as np\n\nfrom .utils import frame_split_size, merge_frames\nfrom .serialize import dask_serialize, dask_deserialize\nfrom . import pickle\n\nfrom ..utils import log_errors\n\n\ndef itemsize(dt):\n    \"\"\" Itemsize of dtype\n\n    Try to return the itemsize of the base element, return 8 as a fallback\n    \"\"\"\n    result = dt.base.itemsize\n    if result > 255:\n        result = 8\n    return result\n\n\n@dask_serialize.register(np.ndarray)\ndef serialize_numpy_ndarray(x):\n    if x.dtype.hasobject:\n        header = {\"pickle\": True}\n        frames = [pickle.dumps(x)]\n        return header, frames\n\n    # We cannot blindly pickle the dtype as some may fail pickling,\n    # so we have a mixture of strategies.\n    if x.dtype.kind == \"V\":\n        # Preserving all the information works best when pickling\n        try:\n            # Only use stdlib pickle as cloudpickle is slow when failing\n            # (microseconds instead of nanoseconds)\n            dt = (1, pickle.pickle.dumps(x.dtype))\n            pickle.loads(dt[1])  # does it unpickle fine?\n        except Exception:\n            # dtype fails pickling => fall back on the descr if reasonable.\n            if x.dtype.type is not np.void or x.dtype.alignment != 1:\n                raise\n            else:\n                dt = (0, x.dtype.descr)\n    else:\n        dt = (0, x.dtype.str)\n\n    # Only serialize non-broadcasted data for arrays with zero strided axes\n    if 0 in x.strides:\n        broadcast_to = x.shape\n        x = x[tuple(slice(None) if s != 0 else slice(1) for s in x.strides)]\n    else:\n        broadcast_to = None\n\n    if not x.shape:\n        # 0d array\n        strides = x.strides\n        data = x.ravel()\n    elif x.flags.c_contiguous or x.flags.f_contiguous:\n        # Avoid a copy and respect order when unserializing\n        strides = x.strides\n        data = x.ravel(order=\"K\")\n    else:\n        x = np.ascontiguousarray(x)\n        strides = x.strides\n        data = x.ravel()\n\n    if data.dtype.fields or data.dtype.itemsize > 8:\n        data = data.view(\"u%d\" % math.gcd(x.dtype.itemsize, 8))\n\n    try:\n        data = data.data\n    except ValueError:\n        # \"ValueError: cannot include dtype 'M' in a buffer\"\n        data = data.view(\"u%d\" % math.gcd(x.dtype.itemsize, 8)).data\n\n    header = {\"dtype\": dt, \"shape\": x.shape, \"strides\": strides}\n\n    if broadcast_to is not None:\n        header[\"broadcast_to\"] = broadcast_to\n\n    if x.nbytes > 1e5:\n        frames = frame_split_size([data])\n    else:\n        frames = [data]\n\n    header[\"lengths\"] = [x.nbytes]\n\n    return header, frames\n\n\n@dask_deserialize.register(np.ndarray)\ndef deserialize_numpy_ndarray(header, frames):\n    with log_errors():\n        if len(frames) > 1:\n            frames = merge_frames(header, frames)\n\n        if header.get(\"pickle\"):\n            return pickle.loads(frames[0])\n\n        is_custom, dt = header[\"dtype\"]\n        if is_custom:\n            dt = pickle.loads(dt)\n        else:\n            dt = np.dtype(dt)\n\n        if header.get(\"broadcast_to\"):\n            shape = header[\"broadcast_to\"]\n        else:\n            shape = header[\"shape\"]\n\n        x = np.ndarray(shape, dtype=dt, buffer=frames[0], strides=header[\"strides\"])\n\n        return x\n\n\n@dask_serialize.register(np.ma.core.MaskedConstant)\ndef serialize_numpy_ma_masked(x):\n    return {}, []\n\n\n@dask_deserialize.register(np.ma.core.MaskedConstant)\ndef deserialize_numpy_ma_masked(header, frames):\n    return np.ma.masked\n\n\n@dask_serialize.register(np.ma.core.MaskedArray)\ndef serialize_numpy_maskedarray(x):\n    data_header, frames = serialize_numpy_ndarray(x.data)\n    header = {\"data-header\": data_header, \"nframes\": len(frames)}\n\n    # Serialize mask if present\n    if x.mask is not np.ma.nomask:\n        mask_header, mask_frames = serialize_numpy_ndarray(x.mask)\n        header[\"mask-header\"] = mask_header\n        frames += mask_frames\n\n    # Only a few dtypes have python equivalents msgpack can serialize\n    if isinstance(x.fill_value, (np.integer, np.floating, np.bool_)):\n        serialized_fill_value = (False, x.fill_value.item())\n    else:\n        serialized_fill_value = (True, pickle.dumps(x.fill_value))\n    header[\"fill-value\"] = serialized_fill_value\n\n    return header, frames\n\n\n@dask_deserialize.register(np.ma.core.MaskedArray)\ndef deserialize_numpy_maskedarray(header, frames):\n    data_header = header[\"data-header\"]\n    data_frames = frames[: header[\"nframes\"]]\n    data = deserialize_numpy_ndarray(data_header, data_frames)\n\n    if \"mask-header\" in header:\n        mask_header = header[\"mask-header\"]\n        mask_frames = frames[header[\"nframes\"] :]\n        mask = deserialize_numpy_ndarray(mask_header, mask_frames)\n    else:\n        mask = np.ma.nomask\n\n    pickled_fv, fill_value = header[\"fill-value\"]\n    if pickled_fv:\n        fill_value = pickle.loads(fill_value)\n\n    return np.ma.masked_array(data, mask=mask, fill_value=fill_value)\n", "path": "distributed/protocol/numpy.py"}]}