From 346ab1752a70f6857e35eb30491dd8f14ff4c1e4 Mon Sep 17 00:00:00 2001 From: Gabe Joseph Date: Fri, 18 Jun 2021 11:31:52 -0600 Subject: [PATCH] Update `many_independent_leaves` test --- distributed/tests/test_scheduler.py | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/distributed/tests/test_scheduler.py b/distributed/tests/test_scheduler.py index 91964650663..132778ef9ff 100644 --- a/distributed/tests/test_scheduler.py +++ b/distributed/tests/test_scheduler.py @@ -100,14 +100,16 @@ async def test_recompute_released_results(c, s, a, b): assert result == 1 -@gen_cluster(client=True) +@gen_cluster(client=True, config={"distributed.scheduler.bandwidth": "1mb"}) async def test_decide_worker_with_many_independent_leaves(c, s, a, b): + # Make data large to penalize scheduling dependent tasks on other workers + ballast = b"\0" * int(s.bandwidth) xs = await asyncio.gather( - c.scatter(list(range(0, 100, 2)), workers=a.address), - c.scatter(list(range(1, 100, 2)), workers=b.address), + c.scatter([bytes(i) + ballast for i in range(0, 100, 2)], workers=a.address), + c.scatter([bytes(i) + ballast for i in range(1, 100, 2)], workers=b.address), ) xs = list(concat(zip(*xs))) - ys = [delayed(inc)(x) for x in xs] + ys = [delayed(lambda s: s[0])(x) for x in xs] y2s = c.persist(ys) await wait(y2s)