From e96476eaeb5b42f2db79db101008d8d001b97e1e Mon Sep 17 00:00:00 2001 From: Pierre Glaser Date: Mon, 15 Jul 2019 18:13:09 +0200 Subject: [PATCH 1/6] discard closed workers in get_worker --- distributed/worker.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/distributed/worker.py b/distributed/worker.py index 4ce564f1b7b..56c2da1305f 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 != "closed") except StopIteration: raise ValueError("No workers found") From 9d4cbc9f12e9fc8b379c406d5bfac4cae752a1d7 Mon Sep 17 00:00:00 2001 From: Pierre Glaser Date: Mon, 15 Jul 2019 18:13:43 +0200 Subject: [PATCH 2/6] TST test get_client does not return closed client --- distributed/deploy/tests/test_local.py | 19 ++++++++++++++++++- 1 file changed, 18 insertions(+), 1 deletion(-) diff --git a/distributed/deploy/tests/test_local.py b/distributed/deploy/tests/test_local.py index 520996f64a0..841bd02e588 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,20 @@ def test_starts_up_sync(loop): assert len(cluster.scheduler.workers) == 2 finally: cluster.close() + + +def test_dont_select_closed_worker(): + cluster = LocalCluster(n_workers=0, processes=False, threads_per_worker=2) + c = Client(cluster) + cluster.scale(2) + assert c == get_client() + + c.close() + cluster.close() + + cluster2 = LocalCluster(n_workers=0, processes=False, threads_per_worker=2) + c2 = Client(cluster2) + cluster2.scale(2) + + current_client = get_client() + assert c2 == current_client From b384942d3d5d1afd6462d00e147b3c03b9b3fea5 Mon Sep 17 00:00:00 2001 From: Pierre Glaser Date: Tue, 16 Jul 2019 00:52:27 +0200 Subject: [PATCH 3/6] CLN, TST ensure proper resource cleanup --- distributed/deploy/tests/test_local.py | 26 +++++++++++++++----------- 1 file changed, 15 insertions(+), 11 deletions(-) diff --git a/distributed/deploy/tests/test_local.py b/distributed/deploy/tests/test_local.py index 841bd02e588..b090b250cd0 100644 --- a/distributed/deploy/tests/test_local.py +++ b/distributed/deploy/tests/test_local.py @@ -834,17 +834,21 @@ def test_starts_up_sync(loop): def test_dont_select_closed_worker(): - cluster = LocalCluster(n_workers=0, processes=False, threads_per_worker=2) - c = Client(cluster) - cluster.scale(2) - assert c == get_client() + with clean(threads=False): + cluster = LocalCluster(n_workers=0) + c = Client(cluster) + cluster.scale(2) + assert c == get_client() - c.close() - cluster.close() + c.close() + cluster.close() + + cluster2 = LocalCluster(n_workers=0) + c2 = Client(cluster2) + cluster2.scale(2) - cluster2 = LocalCluster(n_workers=0, processes=False, threads_per_worker=2) - c2 = Client(cluster2) - cluster2.scale(2) + current_client = get_client() + assert c2 == current_client - current_client = get_client() - assert c2 == current_client + cluster2.close() + c2.close() From 1a997ef4bfff87e710f8a677857b92bbab2f61f7 Mon Sep 17 00:00:00 2001 From: Pierre Glaser Date: Tue, 16 Jul 2019 00:54:20 +0200 Subject: [PATCH 4/6] DOC, TST add a small description --- distributed/deploy/tests/test_local.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/distributed/deploy/tests/test_local.py b/distributed/deploy/tests/test_local.py index b090b250cd0..ebe2d7ec99e 100644 --- a/distributed/deploy/tests/test_local.py +++ b/distributed/deploy/tests/test_local.py @@ -834,6 +834,8 @@ def test_starts_up_sync(loop): 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) From 360553ed7dbfc6557bf9a75030a528a1d3e41dd8 Mon Sep 17 00:00:00 2001 From: Pierre Glaser Date: Tue, 16 Jul 2019 16:57:59 +0200 Subject: [PATCH 5/6] FIX only select running workers --- distributed/worker.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/distributed/worker.py b/distributed/worker.py index 56c2da1305f..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(w for w in Worker._instances if w.status != "closed") + return first(w for w in Worker._instances if w.status == "running") except StopIteration: raise ValueError("No workers found") From 249e2dad4212bd1ce4c772d034d09907f75319ab Mon Sep 17 00:00:00 2001 From: Pierre Glaser Date: Wed, 17 Jul 2019 10:10:24 +0200 Subject: [PATCH 6/6] CI trigger