Simplify test_worker_heartbeat_after_cancel
@@ -22,7 +22,7 @@ from dask.utils import apply, parse_timedelta, stringify
from distributed import Client, Nanny, Worker, fire_and_forget, wait
from distributed.comm import Comm
-from distributed.compatibility import LINUX, MACOS, WINDOWS
+from distributed.compatibility import LINUX, WINDOWS
from distributed.core import ConnectionPool, Status, connect, rpc
from distributed.metrics import time
from distributed.protocol.pickle import dumps
@@ -3102,13 +3102,9 @@ async def test_transition_counter(c, s, a, b):
assert s.transition_counter > 1
-@pytest.mark.skipif(MACOS and sys.version_info < (3, 9), reason="GH#5056")
-@pytest.mark.slow
@gen_cluster(
client=True,
nthreads=[("127.0.0.1", 1) for _ in range(10)],
- # typical runtime just 2-3s but on CI this may increase significantly
- timeout=60,
)
async def test_worker_heartbeat_after_cancel(c, s, *workers):
"""This test is intended to ensure that after cancellation of a graph, the
@@ -3118,30 +3114,17 @@ async def test_worker_heartbeat_after_cancel(c, s, *workers):
computation where the worker returns metrics about duration, type, etc. and
the scheduler doesn't handle the forgotten task gracefully.
- Failures are not triggered reliably since the race conditions for this error
- case are very hard to produce. Likelihood of failure increases with the
- number of workers.
-
See also https://github.com/dask/distributed/issues/4587
"""
- da = pytest.importorskip("dask.array")
for w in workers:
w.periodic_callbacks["heartbeat"].stop()
- x = da.random.random((2000000, 100), chunks=(10000, None))
- svd = da.linalg.svd(x)
- futs = c.compute(svd)
+ futs = c.map(slowinc, range(100), delay=0.1)
- while not s.tasks:
- await asyncio.sleep(0.001)
-
- while sum(w.executing_count for w in workers) < len(workers) / 2:
+ while sum(w.executing_count for w in workers) < len(workers):
await asyncio.sleep(0.001)
await c.cancel(futs)
- while s.tasks:
- await asyncio.sleep(0.001)
-
while any(w.tasks for w in workers):
await asyncio.gather(*[w.heartbeat() for w in workers])
