LocalCluster raises "Stream is Closed" error on exit if initialized with too many threads per worker
- Dominant language
- Python
- Stars
- 1.7k
- Forks
- 778
- Avg merge
- 2h 50m
- Merged PRs (30d)
- 3
Description
**Describe the issue**:
In trying to scale up a distributed process to take advantage of all the compute resources on a machine, I noticed that I started getting error messages like `2022-10-04 21:14:06,420 - distributed.worker - WARNING - Heartbeat to scheduler failed` and `tornado.iostream.StreamClosedError: Stream is closed`. After reading through some of the other issues (e.g. #1674, #2368, #6087, #1969, #3129, #1688) I was able to determine that the errors are only raised when my program exits the context manager, and the my Client/LocalCluster close. The behavior I'm seeing, however, differs slightly in that the issue only appears once the number of threads per worker specified in the `Client()` goes above a certain level.
In the code below, I run the same distributed process four times, but with different numbers of threads per worker specified each time.
**Minimal Complete Verifiable Example**:
```python
import time
from dask.distributed import Client, progress
def sq_x(x):
x = min(x, 1000)
return x**2
if __name__ == "__main__":
n_workers = 24
for threads_per in [1, 2, 3]:
print("Running with {0} threads per worker.".format(threads_per))
with Client(
n_workers=n_workers, threads_per_worker=threads_per, timeout="600s"
) as client:
print("Submitting jobs...")
sttm = time.time()
futures = client.map(sq_x, range(100000))
print("Done. Took {0} s".format(round(time.time() - sttm)))
sttm = time.time()
foo = client.gather(futures)
print("Gathering results...")
print("Done. Took {0} s".format(round(time.time() - sttm)))
print("Client closed!")
```
As shown below in the output, the error starts to show up only after `threads_per` is greater than 2:
```
Running with 1 threads per worker.
Submitting jobs...
Done. Took 4 s
Gathering results...
Done. Took 39 s
Client closed!
Running with 2 threads per worker.
Submitting jobs...
Done. Took 5 s
Gathering results...
Done. Took 39 s
2022-10-04 22:12:39,117 - distributed.worker - WARNING - Heartbeat to scheduler failed
Traceback (most recent call last):
File "/home/max.gardner/anaconda3/envs/main/lib/python3.10/site-packages/distributed/comm/tcp.py", line 225, in read
frames_nbytes = await stream.read_bytes(fmt_size)
tornado.iostream.StreamClosedError: Stream is closed
The above exception was the direct cause of the following exception:
Traceback (most recent call last):
File "/home/max.gardner/anaconda3/envs/main/lib/python3.10/site-packages/distributed/worker.py", line 1193, in heartbeat
response = await retry_operation(
File "/home/max.gardner/anaconda3/envs/main/lib/python3.10/site-packages/distributed/utils_comm.py", line 383, in retry_operation
return await retry(
File "/home/max.gardner/anaconda3/envs/main/lib/python3.10/site-packages/distributed/utils_comm.py", line 368, in retry
return await coro()
File "/home/max.gardner/anaconda3/envs/main/lib/python3.10/site-packages/distributed/core.py", line 1154, in send_recv_from_rpc
return await send_recv(comm=comm, op=key, **kwargs)
File "/home/max.gardner/anaconda3/envs/main/lib/python3.10/site-packages/distributed/core.py", line 919, in send_recv
response = await comm.read(deserializers=deserializers)
File "/home/max.gardner/anaconda3/envs/main/lib/python3.10/site-packages/distributed/comm/tcp.py", line 241, in read
convert_stream_closed_error(self, e)
File "/home/max.gardner/anaconda3/envs/main/lib/python3.10/site-packages/distributed/comm/tcp.py", line 144, in convert_stream_closed_error
raise CommClosedError(f"in {obj}: {exc}") from exc
distributed.comm.core.CommClosedError: in : Stream is closed
Client closed!
Running with 3 threads per worker.
Submitting jobs...
Done. Took 5 s
Gathering results...
Done. Took 38 s
2022-10-04 22:13:41,149 - distributed.worker - WARNING - Heartbeat to scheduler failed
Traceback (most recent call last):
File "/home/max.gardner/anaconda3/envs/main/lib/python3.10/site-packages/distributed/comm/tcp.py", line 225, in read
frames_nbytes = await stream.read_bytes(fmt_size)
tornado.iostream.StreamClosedError: Stream is closed
The above exception was the direct cause of the following exception:
Traceback (most recent call last):
File "/home/max.gardner/anaconda3/envs/main/lib/python3.10/site-packages/distributed/worker.py", line 1193, in heartbeat
response = await retry_operation(
File "/home/max.gardner/anaconda3/envs/main/lib/python3.10/site-packages/distributed/utils_comm.py", line 383, in retry_operation
return await retry(
File "/home/max.gardner/anaconda3/envs/main/lib/python3.10/site-packages/distributed/utils_comm.py", line 368, in retry
return await coro()
File "/home/max.gardner/anaconda3/envs/main/lib/python3.10/site-packages/distributed/core.py", line 1154, in send_recv_from_rpc
return await send_recv(comm=comm, op=key, **kwargs)
File "/home/max.gardner/anaconda3/envs/main/lib/python3.10/site-packages/distributed/core.py", line 919, in send_recv
response = await comm.read(deserializers=deserializers)
File "/home/max.gardner/anaconda3/envs/main/lib/python3.10/site-packages/distributed/comm/tcp.py", line 241, in read
convert_stream_closed_error(self, e)
File "/home/max.gardner/anaconda3/envs/main/lib/python3.10/site-packages/distributed/comm/tcp.py", line 144, in convert_stream_closed_error
raise CommClosedError(f"in {obj}: {exc}") from exc
distributed.comm.core.CommClosedError: in : Stream is closed
Client closed!
```
Querying the worker logs, there are no errors I can find. And the output of the `gather()` looks exactly as it should. Seems like the error message can be ignored? Any way to turn it off? Other dask config settings that might help? Thank you!
**Environment**:
- Dask version: 2022.9.2
- Python version: 3.10.6
- Operating System: Ubuntu 22.04
- Install method (conda, pip, source): conda (conda-forge)
Contributor guide
Assessment
This issue has not been assessed yet.