dask / dask/distributed

The buffer of embedded numpy variables is deep-copied in client->scheduler comms

Open
#8,608 2 comments 0 reactions 0 assignees View on GitHub
p3
Dominant language
Python
Stars
1.7k
Forks
778
Avg merge
2h 50m
Merged PRs (30d)
3

Description

```python

import distributed
import numpy as np

if __name__ == "__main__":
with distributed.Client(n_workers=2, processes=False, protocol="tcp") as client:
a, b = client.has_what()
x = client.submit(np.random.random, 1024, key="x", workers=[a])
y = client.submit(lambda x: x, x, key="y", workers=[b])
y.result()
```

When the buffer reaches `distributed.protocol.serialize.pickle_loads`, `buffers[0]` is a bytes object.
This causes pickle_loads to deep-copy the buffer in order to honour the writeable flag of the original.

To verify, add at the top of `pickle_loads`:
```python
print(header["writeable"], [ensure_memoryview(b).readonly for b in buffers])
```

What's causing me a migraine is:
- if you replace
```python
x = client.submit(np.random.random, 1024, key="x", workers=[a])
```
with
```python
x = client.submit(lambda: np.random.random(1024), key="x", workers=[a])
```
then the numpy object is no longer deserialized by `distributed.protocol.serialize.pickle_loads`, but it's instead processed by `distributed.protocol.numpy.deserialize_numpy_array`, which receives a writeable buffer
- if you replace the submit API with dask.array:
```python
x = da.random.random(1024).persist(workers=[a])
y = x.map_blocks(lambda x: x).persist(workers=[b])
y.compute()
```
then we are using again `distributed.protocol.serialize.pickle_loads`, which receives a read-only buffer but this time the writeable flag is False so no deep copy happens.

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.