Rewrite test_reconnect to use subproc to kill scheduler reliably
@@ -21,7 +21,7 @@ import weakref
import zipfile
from collections import deque
from collections.abc import Generator
-from contextlib import contextmanager, nullcontext
+from contextlib import ExitStack, contextmanager, nullcontext
from functools import partial
from operator import add
from threading import Semaphore
@@ -71,7 +71,7 @@ from distributed.client import (
from distributed.cluster_dump import load_cluster_dump
from distributed.comm import CommClosedError
from distributed.compatibility import LINUX, WINDOWS
-from distributed.core import Server, Status
+from distributed.core import Status
from distributed.metrics import time
from distributed.scheduler import CollectTaskMetaDataPlugin, KilledWorker, Scheduler
from distributed.sizeof import sizeof
@@ -3592,27 +3592,16 @@ async def test_scatter_raises_if_no_workers(c, s):
async def test_reconnect():
port = open_port()
- async def hard_stop(s):
- for pc in s.periodic_callbacks.values():
- pc.stop()
-
- s.stop_services()
- for comm in list(s.stream_comms.values()):
- comm.abort()
- for comm in list(s.client_comms.values()):
- comm.abort()
-
- await s.rpc.close()
- s.stop()
- await Server.close(s)
-
- async with Scheduler(port=port) as s:
- async with Client(f"127.0.0.1:{port}", asynchronous=True) as c:
- async with Worker(f"127.0.0.1:{port}") as w:
+ stack = ExitStack()
+ proc = popen(["dask-scheduler", "--no-dashboard", f"--port={port}"])
+ stack.enter_context(proc)
+ async with Client(f"127.0.0.1:{port}", asynchronous=True) as c, Worker(
+ f"127.0.0.1:{port}"
+ ) as w:
await c.wait_for_workers(1, timeout=10)
x = c.submit(inc, 1)
assert (await x) == 2
- await hard_stop(s)
+ stack.close()
start = time()
while c.status != "connecting":
@@ -3623,7 +3612,7 @@ async def test_reconnect():
with pytest.raises(CancelledError):
await x
- async with Scheduler(port=port) as s2:
+ with popen(["dask-scheduler", "--no-dashboard", f"--port={port}"]):
start = time()
while c.status != "running":
await asyncio.sleep(0.1)
@@ -3638,7 +3627,6 @@ async def test_reconnect():
x = c.submit(inc, 1)
assert (await x) == 2
- await hard_stop(s2)
start = time()
while True:
