dask / dask/distributed

Too many file descriptors issue from own code instantiating too many Clients.

Open
#1,472 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

I have been using `dask`+`distributed` for a while: I have a python script running every morning that launches a `LocalCluster(n_workers=4)`, load some data, process it a bit, persist and publish as a dataset. From there, several clients (either launched from the same computer or remotely) will connect and do some querying on that same dataset.

Occasionally, the cluster would stop being responsive. I could never pinpoint the cause (the way I launch the cluster prevents me to easily see `stderr` and I hadn't bother fixing that yet) and a relaunch would fix it.

Recently however, I got a really weird issue: after a few issues like above in the same day and a few cluster relaunch, I couldn't launch a `LocalCluster` anymore. If launched outside the process that masks `stderr`, I could see `ValueError: too many file descriptors in select()` and a bunch of `tornado.iostream.StreamClosedError: Stream is closed` and `distributed.comm.core.CommClosedError: in : Stream is closed`.

But I could still launch `dask-scheduler` + `dask-worker`s without issue from the command line...
Very surprisingly, even after a pc restart, this situation (no `LocalCluster` working but command-line launch ok) persists.

The next day, working from manually launched scheduler and workers, I was getting logs. The cluster stopped being responsive again but I finally understood the issue: my code is leaking `Client` instances.

A typical computing function I use would do something akin to:

```python
def compute_thing(*args, **kwargs):
dask_client = Client(...) # here's there's some internal configuration so it knows which ip/port to use
df = dask_client.get_dataset('dataset')
return _compute_thing(df) # <- this function does nothing re. scheduling. Apart from the metadata, it works as well on a pandas or dask df
```
And it looks like the `dask_client` connections don't get severed automatically (totally possible python gc would choke on that, fair enough, that's on me). And of course, I would call 10s if not 100s such function throughout the day and at some point that's too much:

```
distributed.scheduler - INFO - Receive client connection: DaskClient-f889ffda-afeb-11e7-abd8-64006a6df23b
distributed.scheduler - INFO - Receive client connection: DaskClient-f8cc46dc-afeb-11e7-abd8-64006a6df23b
distributed.scheduler - INFO - Receive client connection: DaskClient-f90b0ac0-afeb-11e7-abd8-64006a6df23b
distributed.scheduler - INFO - Receive client connection: DaskClient-f948e288-afeb-11e7-abd8-64006a6df23b
distributed.scheduler - INFO - Receive client connection: DaskClient-f989ef36-afeb-11e7-abd8-64006a6df23b
distributed.scheduler - INFO - Receive client connection: DaskClient-f9ce03dc-afeb-11e7-abd8-64006a6df23b
distributed.scheduler - INFO - End scheduler at 'tcp://:8786'
Traceback (most recent call last):
File "C:\Users\...\Scripts\dask-scheduler-script.py", line 5, in
sys.exit(distributed.cli.dask_scheduler.go())
File "C:\Users\...\lib\site-packages\distributed\cli\dask_scheduler.py", line 153, in go
main()
File "C:\Users\...\lib\site-packages\click\core.py", line 722, in __call__
return self.main(*args, **kwargs)
File "C:\Users\...\lib\site-packages\click\core.py", line 697, in main
rv = self.invoke(ctx)
File "C:\Users\...\lib\site-packages\click\core.py", line 895, in invoke
return ctx.invoke(self.callback, **ctx.params)
File "C:\Users\...\lib\site-packages\click\core.py", line 535, in invoke
return callback(*args, **kwargs)
File "C:\Users\...\lib\site-packages\distributed\cli\dask_scheduler.py", line 140, in main
loop.start()
File "C:\Users\...\lib\site-packages\tornado\ioloop.py", line 863, in start
event_pairs = self._impl.poll(poll_timeout)
File "C:\Users\...\lib\site-packages\tornado\platform\select.py", line 63, in poll
self.read_fds, self.write_fds, self.error_fds, timeout)
ValueError: too many file descriptors in select()
```

For what it's worth, there were 260 client connections, it failed after receiving the 260th or on receiving the 261st. The scheduler was running with 4x4 workers (by the way, I was surprised that `LocalClient(n_workers=4)` means 4 processes but launching `dask-worker` 4 times with default parameters ended up with 16 processes. Note: the pc has 4 cores).

So questions/notes:
* the fix on my side is easy enough: make `Client` (actually I use a subclass that provides configuration tooling) a singleton or something like that. That would fix my use case because there won't be 10s of interpreters running at the same time.
* shouldn't the client go out of scope and disconnect when the function returns. If that's not possible, then probably a note in the docs could be useful (point me where to make that note and I'd be happy to contribute it)
* more important in my opinion, is there a way for the scheduler to gracefully refuse connections instead of just quitting and if that's not possible because that's an OS action, could we have a parameter on the scheduler to refuse more than n connections?
* that leaves the outstanding point of not being able to launch `LocalClient`. If anyone has an idea of where to look, what to monitor for this, I'm all ears. Otherwise, I guess this is going to start working again without me understanding why it happened.

End. Sorry wall of text.

system: Windows 10
`dask` 0.15.2 and 0.15.4
`distributed` 1.18.2 and 1.19.2

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.