diff --git a/distributed/deploy/tests/test_local.py b/distributed/deploy/tests/test_local.py index 520996f64a0..ebe2d7ec99e 100644 --- a/distributed/deploy/tests/test_local.py +++ b/distributed/deploy/tests/test_local.py @@ -14,7 +14,7 @@ from tornado import gen import pytest -from distributed import Client, Worker, Nanny +from distributed import Client, Worker, Nanny, get_client from distributed.deploy.local import LocalCluster, nprocesses_nthreads from distributed.metrics import time from distributed.utils_test import ( @@ -831,3 +831,26 @@ def test_starts_up_sync(loop): assert len(cluster.scheduler.workers) == 2 finally: cluster.close() + + +def test_dont_select_closed_worker(): + # Make sure distributed does not try to reuse a client from a + # closed cluster (https://github.com/dask/distributed/issues/2840). + with clean(threads=False): + cluster = LocalCluster(n_workers=0) + c = Client(cluster) + cluster.scale(2) + assert c == get_client() + + c.close() + cluster.close() + + cluster2 = LocalCluster(n_workers=0) + c2 = Client(cluster2) + cluster2.scale(2) + + current_client = get_client() + assert c2 == current_client + + cluster2.close() + c2.close() diff --git a/distributed/worker.py b/distributed/worker.py index 4ce564f1b7b..a3fc2322cb9 100644 --- a/distributed/worker.py +++ b/distributed/worker.py @@ -2857,7 +2857,7 @@ def get_worker(): return thread_state.execution_state["worker"] except AttributeError: try: - return first(Worker._instances) + return first(w for w in Worker._instances if w.status == "running") except StopIteration: raise ValueError("No workers found")