dask / dask/distributed

Future.result() on a worker's client unnecessarily pickles and duplicates data

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

Description

In this task:
```python
def f():
c = get_client()
c.gather(some_future)
```
If the data for the future is already on the worker, the network stack is completely sidestepped and the client just gets a reference to the python object:

In `Worker.gather`:
https://github.com/dask/distributed/blob/9b8f3b88400e81a46a2e04fc742bf68ff838501e/distributed/client.py#L2400-L2411
In `Worker._gather`:
https://github.com/dask/distributed/blob/9b8f3b88400e81a46a2e04fc742bf68ff838501e/distributed/client.py#L2277-L2281

However, if you do
```python
def f():
c = get_client()
some_future.result()
```
then this special case is not dealt with:
In `Future._result`:
https://github.com/dask/distributed/blob/9b8f3b88400e81a46a2e04fc742bf68ff838501e/distributed/client.py#L349

# Reproducer
```python
import distributed

class C:
def __reduce__(self):
assert False

def f():
c = distributed.get_client()
f = c.submit(C)
return str(c.gather(f))

def g():
c = distributed.get_client()
f = c.submit(C)
return str(f.result())

with distributed.Client(n_workers=1) as client:
print("f", client.submit(f).result())
print("g", client.submit(g).result())
```
Output:
```
f <__main__.C object at 0x7fcc840212e0>

2023-08-09 12:20:47,316 - distributed.protocol.pickle - ERROR - Failed to serialize <__main__.C object at 0x7fcc84038100>.
Traceback (most recent call last):
File "distributed/protocol/pickle.py", line 63, in dumps
result = pickle.dumps(x, **dump_kwargs)
File "/tmp/ipykernel_889930/1792512223.py", line 5, in __reduce__
AssertionError
```

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.