dask / dask/distributed

Cannot spin up Client in a subprocess if a Client/LocalCluster was initialized in the parent process

Open
#5,947 4 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

**What happened**:

I have a setup that reduces down to the following:
- Parent process initializes a LocalCluster
- It then starts a subprocess that initializes a Client that happens to talk to the LocalCluster.

But after about a minute's delay, the following stacktrace occurs:
```
Traceback (most recent call last):
File "/home/ubuntu/strap/.venv/lib/python3.8/site-packages/distributed/comm/core.py", line 289, in connect
comm = await asyncio.wait_for(
File "/usr/lib/python3.8/asyncio/tasks.py", line 501, in wait_for
raise exceptions.TimeoutError()
asyncio.exceptions.TimeoutError

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
File "/usr/lib/python3.8/multiprocessing/process.py", line 315, in _bootstrap
self.run()
File "/usr/lib/python3.8/multiprocessing/process.py", line 108, in run
self._target(*self._args, **self._kwargs)
File "test.py", line 6, in inner_function
inner_client = Client('tcp://127.0.0.1:9999')
File "/home/ubuntu/strap/.venv/lib/python3.8/site-packages/distributed/client.py", line 933, in __init__
self.start(timeout=timeout)
File "/home/ubuntu/strap/.venv/lib/python3.8/site-packages/distributed/client.py", line 1091, in start
sync(self.loop, self._start, **kwargs)
File "/home/ubuntu/strap/.venv/lib/python3.8/site-packages/distributed/utils.py", line 378, in sync
raise exc.with_traceback(tb)
File "/home/ubuntu/strap/.venv/lib/python3.8/site-packages/distributed/utils.py", line 351, in f
result = yield future
File "/home/ubuntu/strap/.venv/lib/python3.8/site-packages/tornado/gen.py", line 735, in run
value = future.result()
File "/home/ubuntu/strap/.venv/lib/python3.8/site-packages/distributed/client.py", line 1183, in _start
await self._ensure_connected(timeout=timeout)
File "/home/ubuntu/strap/.venv/lib/python3.8/site-packages/distributed/client.py", line 1245, in _ensure_connected
comm = await connect(
File "/home/ubuntu/strap/.venv/lib/python3.8/site-packages/distributed/comm/core.py", line 315, in connect
raise OSError(
OSError: Timed out trying to connect to tcp://127.0.0.1:9999 after 30 s
```

**What you expected to happen**:

The subprocess client to succeed without error

**Minimal Complete Verifiable Example**:

```python
from dask.distributed import Client, LocalCluster
from dask import delayed
import multiprocessing as mp

def inner_function():
inner_client = Client('tcp://127.0.0.1:9999')
inner_fut = inner_client.compute(delayed(sum)([1,2]))
print(inner_fut.result())

if __name__ == '__main__':
cluster = LocalCluster(scheduler_port=9999)

# This works
inner_function()

# This works
spawn_ctx = mp.get_context('spawn')
p = spawn_ctx.Process(target=inner_function)
p.start()
p.join()

# This errors out
fork_ctx = mp.get_context('fork')
p = fork_ctx.Process(target=inner_function)
p.start()
p.join()

```

**Anything else we need to know?**:

My best guess is that this involves asyncio and forking, wanted to see if there's a known solution for this kind of problem

**Environment**:

- Dask version: 2022.2.1, but this affects older versions as well
- Python version: 3.8.8
- Operating System: Ubuntu 18.04.3
- Install method (conda, pip, source): pip

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.