dask / dask/distributed

LocalCluster raises "Stream is Closed" error on exit if initialized with too many threads per worker

Open
#7,105 3 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

**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

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.