Remove leaking reference to `workers` from `gen_cluster`
@@ -1003,14 +1003,15 @@ def gen_cluster(
@functools.wraps(func)
def test_func(*outer_args, **kwargs):
result = None
- workers = []
with clean(timeout=active_rpc_timeout, **clean_kwargs) as loop:
async def coro():
with tempfile.TemporaryDirectory() as tmpdir:
config2 = merge({"temporary-directory": tmpdir}, config)
with dask.config.set(config2):
+ workers = []
s = False
+
for _ in range(60):
try:
s, ws = await start_cluster(
@@ -1139,12 +1140,6 @@ def gen_cluster(
Comm._instances.clear()
_global_clients.clear()
- return result
-
- result = loop.run_sync(
- coro, timeout=timeout * 2 if timeout else timeout
- )
-
for w in workers:
if getattr(w, "data", None):
try:
@@ -1156,6 +1151,12 @@ def gen_cluster(
return result
+ result = loop.run_sync(
+ coro, timeout=timeout * 2 if timeout else timeout
+ )
+
+ return result
+
# Patch the signature so pytest can inject fixtures
orig_sig = inspect.signature(func)
args = [None] * (1 + len(nthreads)) # scheduler, *workers
