Infinite loop after FargateCluster shutdown fails
- Dominant language
- Python
- Stars
- 1.7k
- Forks
- 778
- Avg merge
- 2h 50m
- Merged PRs (30d)
- 3
Description
We are running AWS Fargate Clusters in adaptive mode. We observe that the clusters sometimes fail to shut down after long running (~1hr) jobs due to unhandled ThrottlingExceptions, but we handle this in code, as by that time the AWS-based processing is complete and we can continue. However, at the end of execution, the program never exits. When the program is terminated via a KeyboardInterrupt we see the following output:
```
Error in atexit._run_exitfuncs:
Traceback (most recent call last):
File "/home/ukhouser/git/UKHO/gsfpy-cleanse/.venv/lib/python3.8/site-packages/distributed/deploy/spec.py", line 671, in close_clusters
cluster.close(timeout=10)
File "/home/ukhouser/git/UKHO/gsfpy-cleanse/.venv/lib/python3.8/site-packages/distributed/deploy/cluster.py", line 114, in close
return self.sync(self._close, callback_timeout=timeout)
File "/home/ukhouser/git/UKHO/gsfpy-cleanse/.venv/lib/python3.8/site-packages/distributed/deploy/cluster.py", line 193, in sync
return sync(self.loop, func, *args, **kwargs)
File "/home/ukhouser/git/UKHO/gsfpy-cleanse/.venv/lib/python3.8/site-packages/distributed/utils.py", line 335, in sync
e.wait(10)
File "/home/ukhouser/.pyenv/versions/3.8.6/lib/python3.8/threading.py", line 558, in wait
signaled = self._cond.wait(timeout)
File "/home/ukhouser/.pyenv/versions/3.8.6/lib/python3.8/threading.py", line 306, in wait
gotit = waiter.acquire(True, timeout)
KeyboardInterrupt
```
It seems that, because the cluster no longer exists, but has not been marked as closed, the following code in `distributed/spec.py`:
```
@atexit.register
def close_clusters():
for cluster in list(SpecCluster._instances):
if cluster.shutdown_on_close:
with suppress(gen.TimeoutError, TimeoutError):
if cluster.status != Status.closed:
cluster.close(timeout=10)
```
eventually causes an infinite loop in `distributed/utils.py` (line 334):
```
...
while not e.is_set():
e.wait(10)
```
-->
**What happened**:
FargateCluster shutdown throws exception, which is subsequently handled. Program continues executing but never exits.
**What you expected to happen**:
Program exits after completion.
**Minimal Complete Verifiable Example**:
```python
try:
# FargateCluster client
client.shutdown()
except botocore.exceptions.ClientError as clex:
if clex.response.["Error"]["Code"] == "ThrottlingException":
ecs = botocore.session.get_session().create_client("ecs")
# ... deregister container instances and stop tasks using the ecs client before deleting the cluster itself
ecs.delete_cluster(cluster=cluster.cluster_arn)
client.close()
# ... program continues to run until end of processing but never exits
```
**Environment**:
- Dask version: `dask`: 2021.6.1, `distributed`: 2021.6.1, `dask-cloudprovider`: 2021.6.0
- Python version: 3.8.6
- Operating System: Ubuntu 20.04 LTS
- Install method (conda, pip, source): pip
Contributor guide
Assessment
This issue has not been assessed yet.