dask / dask/distributed

pickle cache of submitted functions generate object collusion

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

Description

TLDR: `distributed.worker` uses a `pickle` cache when loading `submit`'ed functions. This cache is hit even for impure, stateful callables whose execution will have side effects on themselves.

Here is a simple example illustrating this:
```python
import threading

from distributed import Client, LocalCluster

class Func:
def __init__(self):
self._thread_idents = []

# Uncomment this line and the test passes (reason: cache_loads)
# self._data = b'\x00' * int(1e5)

def __call__(self):
self._thread_idents.append(threading.get_ident()) # side effect # noqa
return self

if __name__ == "__main__":
# note how each f in fs is a different object.
fs = [Func() for _ in range(10)]

cluster = LocalCluster(n_workers=1, threads_per_worker=4)
client = Client(cluster)

# setting pure=False because of anticipated side effect of f.__call__
futures = [client.submit(f, pure=False) for f in fs]
results = [f.result() for f in futures]

# r._thread_indents should be incremented only one time for each r in
# results.
assert all(len(r._thread_idents) == 1 for r in results)
```

### Typical real life use-case
Using `scikit-learn`: calling `[client.submit(clone(estimator).fit, X, y) for _ in range(5)]`.
Because each submitted task has the same exact `pickle` representation, `distributed` will use its cache and treat all 5 estimators as one. (provided there is only 1 `distributed` worker).

Do you consider that to be a bug? I would have guessed using `pure=False` in `client.submit` would solve this, but it does not.

cc'ing @ogrisel @tomMoral.
Cheers!

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.