dask / dask/distributed

Infinite loop after FargateCluster shutdown fails

Open
#4,950 0 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

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

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.