We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Cannot serialize datetime64/timedelta64 numpy arrays
```
distributed.core - INFO - Unserializable Message: {'finalize-c29ed21fd1e79d8eb6b3ff89cf2ae354': <Serialize: ['2010-01-01T00:00:00.000000000' '2010-01-02T00:00:00.000000000'
 '2010-01-03T00:00:00.000000000']>}
distributed.core - ERROR - cannot include dtype 'M' in a buffer
Traceback (most recent call last):
  File "/Users/shoyer/dev/distributed/distributed/core.py", line 228, in write
    frames = protocol.dumps(msg)
  File "/Users/shoyer/dev/distributed/distributed/protocol/core.py", line 43, in dumps
    for key, value in data.items()
  File "/Users/shoyer/dev/distributed/distributed/protocol/core.py", line 44, in <dictcomp>
    if type(value) is Serialize}
  File "/Users/shoyer/dev/distributed/distributed/protocol/serialize.py", line 103, in serialize
    header, frames = serializers[name](x)
  File "/Users/shoyer/dev/distributed/distributed/protocol/numpy.py", line 58, in serialize_numpy_ndarray
    frames = [x.data]
ValueError: cannot include dtype 'M' in a buffer
```

The buffer protocol fails on these (sigh). You can view as int64, but then you'll need to keep track of the original type (including time units) separately.

This came up in some of my integration tests with xarray.

--- 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 from __future__ import print_function, division, absolute_import
2 
3 import sys
4 
5 import numpy as np
6 
7 try:
8     import blosc
9     n = blosc.set_nthreads(2)
10 except ImportError:
11     blosc = False
12 
13 from .utils import frame_split_size
14 from .serialize import register_serialization
15 from . import pickle
16 
17 from ..utils import log_errors, ensure_bytes
18 
19 
20 def itemsize(dt):
21     """ Itemsize of dtype
22 
23     Try to return the itemsize of the base element, return 8 as a fallback
24     """
25     result = dt.base.itemsize
26     if result > 255:
27         result = 8
28     return result
29 
30 
31 def serialize_numpy_ndarray(x):
32     if x.dtype.hasobject:
33         header = {'pickle': True}
34         frames = [pickle.dumps(x)]
35         return header, frames
36 
37     size = itemsize(x.dtype)
38 
39     if x.dtype.kind == 'V':
40         dt = x.dtype.descr
41     else:
42         dt = x.dtype.str
43 
44     x = np.ascontiguousarray(x)  # np.frombuffer requires this
45 
46     header = {'dtype': dt,
47               'strides': x.strides,
48               'shape': x.shape}
49 
50     if blosc:
51         frames = frame_split_size([x.data])
52         if sys.version_info.major == 2:
53             frames = [ensure_bytes(frame) for frame in frames]
54         frames = [blosc.compress(frame, typesize=size,
55                                  cname='lz4', clevel=5) for frame in frames]
56         header['compression'] = ['blosc'] * len(frames)
57     else:
58         frames = [x.data]
59 
60     header['lengths'] = [x.nbytes]
61 
62     return header, frames
63 
64 
65 def deserialize_numpy_ndarray(header, frames):
66     with log_errors():
67         assert len(frames) == 1
68 
69         if header.get('pickle'):
70             return pickle.loads(frames[0])
71 
72         dt = header['dtype']
73         if isinstance(dt, tuple):
74             dt = list(dt)
75         dt = np.dtype(dt)
76 
77         buffer = frames[0]
78 
79         x = np.frombuffer(buffer, dt)
80         x = np.lib.stride_tricks.as_strided(x, header['shape'], header['strides'])
81 
82         return x
83 
84 
85 register_serialization(np.ndarray, serialize_numpy_ndarray, deserialize_numpy_ndarray)
86
```
--- 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
@@ -47,15 +47,17 @@
               'strides': x.strides,
               'shape': x.shape}
 
+    data = x.view('u1').data
+
     if blosc:
-        frames = frame_split_size([x.data])
+        frames = frame_split_size([data])
         if sys.version_info.major == 2:
             frames = [ensure_bytes(frame) for frame in frames]
         frames = [blosc.compress(frame, typesize=size,
                                  cname='lz4', clevel=5) for frame in frames]
         header['compression'] = ['blosc'] * len(frames)
     else:
-        frames = [x.data]
+        frames = [data]
 
     header['lengths'] = [x.nbytes]

{"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@@ -47,15 +47,17 @@\n               'strides': x.strides,\n               'shape': x.shape}\n \n+    data = x.view('u1').data\n+\n     if blosc:\n-        frames = frame_split_size([x.data])\n+        frames = frame_split_size([data])\n         if sys.version_info.major == 2:\n             frames = [ensure_bytes(frame) for frame in frames]\n         frames = [blosc.compress(frame, typesize=size,\n                                  cname='lz4', clevel=5) for frame in frames]\n         header['compression'] = ['blosc'] * len(frames)\n     else:\n-        frames = [x.data]\n+        frames = [data]\n \n     header['lengths'] = [x.nbytes]\n", "issue": "Cannot serialize datetime64/timedelta64 numpy arrays\n```\r\ndistributed.core - INFO - Unserializable Message: {'finalize-c29ed21fd1e79d8eb6b3ff89cf2ae354': <Serialize: ['2010-01-01T00:00:00.000000000' '2010-01-02T00:00:00.000000000'\r\n '2010-01-03T00:00:00.000000000']>}\r\ndistributed.core - ERROR - cannot include dtype 'M' in a buffer\r\nTraceback (most recent call last):\r\n  File \"/Users/shoyer/dev/distributed/distributed/core.py\", line 228, in write\r\n    frames = protocol.dumps(msg)\r\n  File \"/Users/shoyer/dev/distributed/distributed/protocol/core.py\", line 43, in dumps\r\n    for key, value in data.items()\r\n  File \"/Users/shoyer/dev/distributed/distributed/protocol/core.py\", line 44, in <dictcomp>\r\n    if type(value) is Serialize}\r\n  File \"/Users/shoyer/dev/distributed/distributed/protocol/serialize.py\", line 103, in serialize\r\n    header, frames = serializers[name](x)\r\n  File \"/Users/shoyer/dev/distributed/distributed/protocol/numpy.py\", line 58, in serialize_numpy_ndarray\r\n    frames = [x.data]\r\nValueError: cannot include dtype 'M' in a buffer\r\n```\r\n\r\nThe buffer protocol fails on these (sigh). You can view as int64, but then you'll need to keep track of the original type (including time units) separately.\r\n\r\nThis came up in some of my integration tests with xarray.\n", "before_files": [{"content": "from __future__ import print_function, division, absolute_import\n\nimport sys\n\nimport numpy as np\n\ntry:\n    import blosc\n    n = blosc.set_nthreads(2)\nexcept ImportError:\n    blosc = False\n\nfrom .utils import frame_split_size\nfrom .serialize import register_serialization\nfrom . import pickle\n\nfrom ..utils import log_errors, ensure_bytes\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\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    size = itemsize(x.dtype)\n\n    if x.dtype.kind == 'V':\n        dt = x.dtype.descr\n    else:\n        dt = x.dtype.str\n\n    x = np.ascontiguousarray(x)  # np.frombuffer requires this\n\n    header = {'dtype': dt,\n              'strides': x.strides,\n              'shape': x.shape}\n\n    if blosc:\n        frames = frame_split_size([x.data])\n        if sys.version_info.major == 2:\n            frames = [ensure_bytes(frame) for frame in frames]\n        frames = [blosc.compress(frame, typesize=size,\n                                 cname='lz4', clevel=5) for frame in frames]\n        header['compression'] = ['blosc'] * len(frames)\n    else:\n        frames = [x.data]\n\n    header['lengths'] = [x.nbytes]\n\n    return header, frames\n\n\ndef deserialize_numpy_ndarray(header, frames):\n    with log_errors():\n        assert len(frames) == 1\n\n        if header.get('pickle'):\n            return pickle.loads(frames[0])\n\n        dt = header['dtype']\n        if isinstance(dt, tuple):\n            dt = list(dt)\n        dt = np.dtype(dt)\n\n        buffer = frames[0]\n\n        x = np.frombuffer(buffer, dt)\n        x = np.lib.stride_tricks.as_strided(x, header['shape'], header['strides'])\n\n        return x\n\n\nregister_serialization(np.ndarray, serialize_numpy_ndarray, deserialize_numpy_ndarray)\n", "path": "distributed/protocol/numpy.py"}], "after_files": [{"content": "from __future__ import print_function, division, absolute_import\n\nimport sys\n\nimport numpy as np\n\ntry:\n    import blosc\n    n = blosc.set_nthreads(2)\nexcept ImportError:\n    blosc = False\n\nfrom .utils import frame_split_size\nfrom .serialize import register_serialization\nfrom . import pickle\n\nfrom ..utils import log_errors, ensure_bytes\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\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    size = itemsize(x.dtype)\n\n    if x.dtype.kind == 'V':\n        dt = x.dtype.descr\n    else:\n        dt = x.dtype.str\n\n    x = np.ascontiguousarray(x)  # np.frombuffer requires this\n\n    header = {'dtype': dt,\n              'strides': x.strides,\n              'shape': x.shape}\n\n    data = x.view('u1').data\n\n    if blosc:\n        frames = frame_split_size([data])\n        if sys.version_info.major == 2:\n            frames = [ensure_bytes(frame) for frame in frames]\n        frames = [blosc.compress(frame, typesize=size,\n                                 cname='lz4', clevel=5) for frame in frames]\n        header['compression'] = ['blosc'] * len(frames)\n    else:\n        frames = [data]\n\n    header['lengths'] = [x.nbytes]\n\n    return header, frames\n\n\ndef deserialize_numpy_ndarray(header, frames):\n    with log_errors():\n        assert len(frames) == 1\n\n        if header.get('pickle'):\n            return pickle.loads(frames[0])\n\n        dt = header['dtype']\n        if isinstance(dt, tuple):\n            dt = list(dt)\n        dt = np.dtype(dt)\n\n        buffer = frames[0]\n\n        x = np.frombuffer(buffer, dt)\n        x = np.lib.stride_tricks.as_strided(x, header['shape'], header['strides'])\n\n        return x\n\n\nregister_serialization(np.ndarray, serialize_numpy_ndarray, deserialize_numpy_ndarray)\n", "path": "distributed/protocol/numpy.py"}]}