scatter to empty adaptive cluster

[crusaderky]

Client.scatter hangs until there is at least one worker on the cluster.
This is a problem for adaptive clusters that scale down to zero.

Expected behaviour

scatter() causes the cluster to scale up to 1 worker

Actual behaviour

scatter() hangs forever

[hendrikmakait]

Instead of hanging indefinitely, scatter() times out with

2022-07-15 19:39:12,046 - distributed.core - ERROR - Exception while handling op scatter
Traceback (most recent call last):
  File "/Users/hendrikmakait/projects/dask/distributed/distributed/core.py", line 769, in _handle_comm
    result = await result
  File "/Users/hendrikmakait/projects/dask/distributed/distributed/scheduler.py", line 5041, in scatter
    raise TimeoutError("No valid workers found")
asyncio.exceptions.TimeoutError: No valid workers found

thanks to

distributed/distributed/scheduler.py

Lines 5030 to 5042 in 930d3dc

	while True:
	if workers is None:
	wss = self.running
	else:
	workers = [self.coerce_address(w) for w in workers]
	wss = {self.workers[w] for w in workers}
	wss = {ws for ws in wss if ws.status == Status.running}
	if wss:
	break
	if time() > start + timeout:
	raise TimeoutError("No valid workers found")
	await asyncio.sleep(0.1)

Reproducer:

from __future__ import annotations

from distributed import Client, LocalCluster
from distributed.utils_test import gen_test


@gen_test()
async def test_scatter_scales_up_empty_cluster():
    async with LocalCluster(
        n_workers=0,
        silence_logs=False,
        processes=False,
        dashboard_address=":0",
        asynchronous=True,
    ) as cluster:
        cluster.adapt(minimum=0, maximum=4, interval="10 ms")
        async with Client(cluster, asynchronous=True) as client:
            await client.scatter([0])