Flaky test_compute_per_key

[gjoseph92]

macOS 3.10:

____________________________ test_adapt_then_manual ____________________________
loop = <tornado.platform.asyncio.AsyncIOLoop object at 0x13c5dbe80>
deftest_adapt_then_manual(loop):
"""We can revert from adaptive, back to manual"""
with LocalCluster(
            silence_logs=False,
            loop=loop,
            dashboard_address=":0",
            processes=False,
            n_workers=8,
        ) as cluster:
            cluster.adapt(minimum=0, maximum=4, interval="10ms")
            start = time()
while cluster.scheduler.workers or cluster.workers:
                sleep(0.01)
assert time() < start + 30
assertnot cluster.workers
with Client(cluster) as client:
                futures = client.map(slowinc, range(1000), delay=0.1)
                sleep(0.2)
                cluster._adaptive.stop()
                sleep(0.2)
                cluster.scale(2)
                start = time()
whilelen(cluster.scheduler.workers) != 2:
                    sleep(0.01)
>                   assert time() < start + 30
E                   assert 1658211621.203754 < (1658211591.147773 + 30)
E                    +  where 1658211621.203754 = time()
distributed/deploy/tests/test_local.py:782: AssertionError
------------------------------ Captured log call -------------------------------
ERROR    asyncio:base_events.py:1744 Future exception was never retrieved
future: <Future finished exception=TimeoutError('Timeout')>
tornado.util.TimeoutError: Timeout

https://github.com/dask/distributed/runs/7403340779?check_suite_focus=true#step:11:1933

ubuntu 3.8:

____________________________ test_adapt_then_manual ____________________________
cls = <class '_pytest.runner.CallInfo'>
func = <function call_runtest_hook.<locals>.<lambda> at 0x7f1885a8f430>
when = 'call'
reraise = (<class '_pytest.outcomes.Exit'>, <class 'KeyboardInterrupt'>)
@classmethod
deffrom_call(
cls,
        func: "Callable[[], TResult]",
        when: "Literal['collect', 'setup', 'call', 'teardown']",
        reraise: Optional[
            Union[Type[BaseException], Tuple[Type[BaseException], ...]]
        ] = None,
    ) -> "CallInfo[TResult]":
"""Call func, wrapping the result in a CallInfo.
    :param func:
        The function to call. Called without arguments.
    :param when:
        The phase in which the function is called.
    :param reraise:
        Exception or exceptions that shall propagate if raised by the
        function, instead of being wrapped in the CallInfo.
    """
        excinfo = None
        start = timing.time()
        precise_start = timing.perf_counter()
try:
>           result: Optional[TResult] = func()
/usr/share/miniconda3/envs/dask-distributed/lib/python3.8/site-packages/_pytest/runner.py:338: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
/usr/share/miniconda3/envs/dask-distributed/lib/python3.8/site-packages/_pytest/runner.py:259: in <lambda>
lambda: ihook(item=item, **kwds), when=when, reraise=reraise
/usr/share/miniconda3/envs/dask-distributed/lib/python3.8/site-packages/pluggy/_hooks.py:265: in __call__
returnself._hookexec(self.name, self.get_hookimpls(), kwargs, firstresult)
/usr/share/miniconda3/envs/dask-distributed/lib/python3.8/site-packages/pluggy/_manager.py:80: in _hookexec
returnself._inner_hookexec(hook_name, methods, kwargs, firstresult)
/usr/share/miniconda3/envs/dask-distributed/lib/python3.8/site-packages/_pytest/unraisableexception.py:88: in pytest_runtest_call
yield from unraisable_exception_runtest_hook()
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
defunraisable_exception_runtest_hook() -> Generator[None, None, None]:
with catch_unraisable_exception() as cm:
yield
if cm.unraisable:
if cm.unraisable.err_msg isnotNone:
                    err_msg = cm.unraisable.err_msg
else:
                    err_msg = "Exception ignored in"
                msg = f"{err_msg}: {cm.unraisable.object!r}\n\n"
                msg += "".join(
                    traceback.format_exception(
                        cm.unraisable.exc_type,
                        cm.unraisable.exc_value,
                        cm.unraisable.exc_traceback,
                    )
                )
>               warnings.warn(pytest.PytestUnraisableExceptionWarning(msg))
E               pytest.PytestUnraisableExceptionWarning: Exception ignored in: <coroutine object InProc.write at 0x7f188637cac0>
E               
E               Traceback (most recent call last):
E                 File "/usr/share/miniconda3/envs/dask-distributed/lib/python3.8/warnings.py", line 506, in _warn_unawaited_coroutine
E                   warn(msg, category=RuntimeWarning, stacklevel=2, source=coro)
E               RuntimeWarning: coroutine 'InProc.write' was never awaited
/usr/share/miniconda3/envs/dask-distributed/lib/python3.8/site-packages/_pytest/unraisableexception.py:78: PytestUnraisableExceptionWarning
------------------------------ Captured log call -------------------------------
ERROR    asyncio.events:utils.py:804 
Traceback (most recent call last):
  File "/home/runner/work/distributed/distributed/distributed/utils.py", line 778, in wrapper
    return await func(*args, **kwargs)
  File "/home/runner/work/distributed/distributed/distributed/worker.py", line 1443, in close
    await self.finished()
  File "/home/runner/work/distributed/distributed/distributed/core.py", line 447, in finished
    await self._event_finished.wait()
  File "/usr/share/miniconda3/envs/dask-distributed/lib/python3.8/asyncio/locks.py", line 309, in wait
    await fut
asyncio.exceptions.CancelledError
ERROR    asyncio.events:utils.py:804 
Traceback (most recent call last):
  File "/home/runner/work/distributed/distributed/distributed/utils.py", line 778, in wrapper
    return await func(*args, **kwargs)
  File "/home/runner/work/distributed/distributed/distributed/worker.py", line 1443, in close
    await self.finished()
  File "/home/runner/work/distributed/distributed/distributed/core.py", line 447, in finished
    await self._event_finished.wait()
  File "/usr/share/miniconda3/envs/dask-distributed/lib/python3.8/asyncio/locks.py", line 309, in wait
    await fut
asyncio.exceptions.CancelledError
ERROR    asyncio.events:utils.py:804 
Traceback (most recent call last):
  File "/home/runner/work/distributed/distributed/distributed/utils.py", line 778, in wrapper
    return await func(*args, **kwargs)
  File "/home/runner/work/distributed/distributed/distributed/worker.py", line 1443, in close
    await self.finished()
  File "/home/runner/work/distributed/distributed/distributed/core.py", line 447, in finished
    await self._event_finished.wait()
  File "/usr/share/miniconda3/envs/dask-distributed/lib/python3.8/asyncio/locks.py", line 309, in wait
    await fut
asyncio.exceptions.CancelledError
ERROR    asyncio.events:utils.py:804 
Traceback (most recent call last):
  File "/home/runner/work/distributed/distributed/distributed/utils.py", line 778, in wrapper
    return await func(*args, **kwargs)
  File "/home/runner/work/distributed/distributed/distributed/worker.py", line 1443, in close
    await self.finished()
  File "/home/runner/work/distributed/distributed/distributed/core.py", line 447, in finished
    await self._event_finished.wait()
  File "/usr/share/miniconda3/envs/dask-distributed/lib/python3.8/asyncio/locks.py", line 309, in wait
    await fut
asyncio.exceptions.CancelledError
ERROR    asyncio.events:utils.py:804 
Traceback (most recent call last):
  File "/home/runner/work/distributed/distributed/distributed/utils.py", line 778, in wrapper
    return await func(*args, **kwargs)
  File "/home/runner/work/distributed/distributed/distributed/worker.py", line 1443, in close
    await self.finished()
  File "/home/runner/work/distributed/distributed/distributed/core.py", line 447, in finished
    await self._event_finished.wait()
  File "/usr/share/miniconda3/envs/dask-distributed/lib/python3.8/asyncio/locks.py", line 309, in wait
    await fut
asyncio.exceptions.CancelledError
ERROR    asyncio.events:utils.py:804 
Traceback (most recent call last):
  File "/home/runner/work/distributed/distributed/distributed/utils.py", line 778, in wrapper
    return await func(*args, **kwargs)
  File "/home/runner/work/distributed/distributed/distributed/worker.py", line 1443, in close
    await self.finished()
  File "/home/runner/work/distributed/distributed/distributed/core.py", line 447, in finished
    await self._event_finished.wait()
  File "/usr/share/miniconda3/envs/dask-distributed/lib/python3.8/asyncio/locks.py", line 309, in wait
    await fut
asyncio.exceptions.CancelledError
ERROR    asyncio.events:utils.py:804 
Traceback (most recent call last):
  File "/home/runner/work/distributed/distributed/distributed/utils.py", line 778, in wrapper
    return await func(*args, **kwargs)
  File "/home/runner/work/distributed/distributed/distributed/worker.py", line 1443, in close
    await self.finished()
  File "/home/runner/work/distributed/distributed/distributed/core.py", line 447, in finished
    await self._event_finished.wait()
  File "/usr/share/miniconda3/envs/dask-distributed/lib/python3.8/asyncio/locks.py", line 309, in wait
    await fut
asyncio.exceptions.CancelledError
ERROR    asyncio.events:utils.py:804 
Traceback (most recent call last):
  File "/home/runner/work/distributed/distributed/distributed/utils.py", line 778, in wrapper
    return await func(*args, **kwargs)
  File "/home/runner/work/distributed/distributed/distributed/worker.py", line 1443, in close
    await self.finished()
  File "/home/runner/work/distributed/distributed/distributed/core.py", line 447, in finished
    await self._event_finished.wait()
  File "/usr/share/miniconda3/envs/dask-distributed/lib/python3.8/asyncio/locks.py", line 309, in wait
    await fut
asyncio.exceptions.CancelledError
ERROR    asyncio:base_events.py:1707 Task was destroyed but it is pending!
task: <Task pending name='Task-36123' coro=<InProc.write() running at /home/runner/work/distributed/distributed/distributed/comm/inproc.py:215> cb=[IOLoop.add_future.<locals>.<lambda>() at /usr/share/miniconda3/envs/dask-distributed/lib/python3.8/site-packages/tornado/ioloop.py:688]>
ERROR    asyncio:base_events.py:1707 Task was destroyed but it is pending!
task: <Task pending name='Task-37000' coro=<InProc.write() running at /home/runner/work/distributed/distributed/distributed/comm/inproc.py:215> cb=[IOLoop.add_future.<locals>.<lambda>() at /usr/share/miniconda3/envs/dask-distributed/lib/python3.8/site-packages/tornado/ioloop.py:688]>

https://github.com/dask/distributed/runs/7256210760?check_suite_focus=true#step:11:1876

[gjoseph92]

Duplicate of #4955; accidentally opened and pasted in the wrong thing.