Flaky `test_host_uses_scheduler_protocol`

Failed on macOS, 3.8. The timeout error is remarkably inscrutable, it's not clear where in the test it's timing out. I would assume it's something about opening connections as usual, probably the worker trying to connect to the scheduler.

https://github.com/dask/distributed/blob/1bb0ce74ad93c1403684d0819f8dba7486a2fdba/distributed/tests/test_worker.py#L1446-L1463

```
______________________ test_host_uses_scheduler_protocol _______________________

args = ()
kwds = {'monkeypatch': <_pytest.monkeypatch.MonkeyPatch object at 0x7fb435861040>}

@wraps(func)
def inner(*args, **kwds):
with self._recreate_cm():
> return func(*args, **kwds)

../../../miniconda3/envs/dask-distributed/lib/python3.8/contextlib.py:75:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
../../../miniconda3/envs/dask-distributed/lib/python3.8/contextlib.py:75: in inner
return func(*args, **kwds)
distributed/utils_test.py:728: in test_func
return _run_and_close_tornado(async_fn_outer, func, *args, **kwargs)
distributed/utils_test.py:376: in _run_and_close_tornado
return asyncio.run(inner_fn())
../../../miniconda3/envs/dask-distributed/lib/python3.8/asyncio/runners.py:44: in run
return loop.run_until_complete(main)
../../../miniconda3/envs/dask-distributed/lib/python3.8/asyncio/base_events.py:616: in run_until_complete
return future.result()
distributed/utils_test.py:373: in inner_fn
return await async_fn(*args, **kwargs)
distributed/utils_test.py:716: in async_fn_outer
return await asyncio.wait_for(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

fut = >
timeout = 30

async def wait_for(fut, timeout, *, loop=None):
"""Wait for the single Future or coroutine to complete, with timeout.

Coroutine will be wrapped in Task.

Returns result of the Future or coroutine. When a timeout occurs,
it cancels the task and raises TimeoutError. To avoid the task
cancellation, wrap it in shield().

If the wait is cancelled, the task is also cancelled.

This function is a coroutine.
"""
if loop is None:
loop = events.get_running_loop()
else:
warnings.warn("The loop argument is deprecated since Python 3.8, "
"and scheduled for removal in Python 3.10.",
DeprecationWarning, stacklevel=2)

if timeout is None:
return await fut

if timeout <= 0:
fut = ensure_future(fut, loop=loop)

if fut.done():
return fut.result()

await _cancel_and_wait(fut, loop=loop)
try:
fut.result()
except exceptions.CancelledError as exc:
raise exceptions.TimeoutError() from exc
else:
raise exceptions.TimeoutError()

waiter = loop.create_future()
timeout_handle = loop.call_later(timeout, _release_waiter, waiter)
cb = functools.partial(_release_waiter, waiter)

fut = ensure_future(fut, loop=loop)
fut.add_done_callback(cb)

try:
# wait until the future completes or the timeout
try:
await waiter
except exceptions.CancelledError:
if fut.done():
return fut.result()
else:
fut.remove_done_callback(cb)
# We must ensure that the task is not running
# after wait_for() returns.
# See https://bugs.python.org/issue32751
await _cancel_and_wait(fut, loop=loop)
raise

if fut.done():
return fut.result()
else:
fut.remove_done_callback(cb)
# We must ensure that the task is not running
# after wait_for() returns.
# See https://bugs.python.org/issue32751
await _cancel_and_wait(fut, loop=loop)
> raise exceptions.TimeoutError()
E asyncio.exceptions.TimeoutError

../../../miniconda3/envs/dask-distributed/lib/python3.8/asyncio/tasks.py:501: TimeoutError
```
https://github.com/dask/distributed/runs/7999878381?check_suite_focus=true#step:11:1718