Test timeout test_target_duration

[fjetter]

The test distributed/deploy/tests/test_adaptive.py::test_target_duration is frequently hanging and is torn down by pytest-timeout which aborts the entire test run. Cause is unclear and it is not reliably reproducible.

Example run https://github.com/dask/distributed/runs/2683878847?check_suite_focus=true

distributed/deploy/tests/test_adaptive.py::test_target_duration 
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ Captured stderr ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
distributed.comm.inproc - WARNING - Closing dangling queue in <InProc  local=inproc://10.79.10.98/3099/328 remote=inproc://10.79.10.98/3099/370>

~~~~~~~~~~~~~~~~~~~~~~ Stack of Profile (123145649283072) ~~~~~~~~~~~~~~~~~~~~~~
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/threading.py", line 890, in _bootstrap
    self._bootstrap_inner()
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/threading.py", line 932, in _bootstrap_inner
    self.run()
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/threading.py", line 870, in run
    self._target(*self._args, **self._kwargs)
  File "/Users/runner/work/distributed/distributed/distributed/profile.py", line 269, in _watch
    sleep(interval)

 Stack of AsyncProcess Dask Worker process (from Nanny) watch message queue (123145632493568) 
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/threading.py", line 890, in _bootstrap
    self._bootstrap_inner()
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/threading.py", line 932, in _bootstrap_inner
    self.run()
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/threading.py", line 870, in run
    self._target(*self._args, **self._kwargs)
  File "/Users/runner/work/distributed/distributed/distributed/process.py", line 218, in _watch_message_queue
    msg = q.get()
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/queue.py", line 170, in get
    self.not_empty.wait()
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/threading.py", line 302, in wait
    waiter.acquire()

~~~~~~~~~~~~~~~~~~~~~~ Stack of Profile (123145615704064) ~~~~~~~~~~~~~~~~~~~~~~
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/threading.py", line 890, in _bootstrap
    self._bootstrap_inner()
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/threading.py", line 932, in _bootstrap_inner
    self.run()
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/threading.py", line 870, in run
    self._target(*self._args, **self._kwargs)
  File "/Users/runner/work/distributed/distributed/distributed/profile.py", line 269, in _watch
    sleep(interval)

~~~~~~~~~~~~~~~~ Stack of TCP-Executor-3099-1 (123145598914560) ~~~~~~~~~~~~~~~~
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/threading.py", line 890, in _bootstrap
    self._bootstrap_inner()
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/threading.py", line 932, in _bootstrap_inner
    self.run()
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/threading.py", line 870, in run
    self._target(*self._args, **self._kwargs)
  File "/Users/runner/work/distributed/distributed/distributed/threadpoolexecutor.py", line 51, in _worker
    task = work_queue.get(timeout=1)
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/queue.py", line 179, in get
    self.not_empty.wait(remaining)
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/threading.py", line 306, in wait
    gotit = waiter.acquire(True, timeout)

~~~~~~~~~~~~~~~~ Stack of TCP-Executor-3099-0 (123145582125056) ~~~~~~~~~~~~~~~~
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/threading.py", line 890, in _bootstrap
    self._bootstrap_inner()
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/threading.py", line 932, in _bootstrap_inner
    self.run()
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/threading.py", line 870, in run
    self._target(*self._args, **self._kwargs)
  File "/Users/runner/work/distributed/distributed/distributed/threadpoolexecutor.py", line 51, in _worker
    task = work_queue.get(timeout=1)
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/queue.py", line 179, in get
    self.not_empty.wait(remaining)
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/threading.py", line 306, in wait
    gotit = waiter.acquire(True, timeout)

~~~~~~~~~~~~~~~~~~ Stack of Dask-Offload_0 (123145531756544) ~~~~~~~~~~~~~~~~~~~
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/threading.py", line 890, in _bootstrap
    self._bootstrap_inner()
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/threading.py", line 932, in _bootstrap_inner
    self.run()
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/threading.py", line 870, in run
    self._target(*self._args, **self._kwargs)
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/concurrent/futures/thread.py", line 78, in _worker
    work_item = work_queue.get(block=True)

~~~~~~~~~~~~~~~~~~~~~~~ Stack of MainThread (4661456320) ~~~~~~~~~~~~~~~~~~~~~~~
  File "/Users/runner/miniconda3/envs/dask-distributed/bin/pytest", line 11, in <module>
    sys.exit(console_main())
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/site-packages/_pytest/config/__init__.py", line 185, in console_main
    code = main()
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/site-packages/_pytest/config/__init__.py", line 162, in main
    ret: Union[ExitCode, int] = config.hook.pytest_cmdline_main(
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/site-packages/pluggy/hooks.py", line 286, in __call__
    return self._hookexec(self, self.get_hookimpls(), kwargs)
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/site-packages/pluggy/manager.py", line 93, in _hookexec
    return self._inner_hookexec(hook, methods, kwargs)
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/site-packages/pluggy/manager.py", line 84, in <lambda>
    self._inner_hookexec = lambda hook, methods, kwargs: hook.multicall(
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/site-packages/pluggy/callers.py", line 187, in _multicall
    res = hook_impl.function(*args)
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/site-packages/_pytest/main.py", line 316, in pytest_cmdline_main
    return wrap_session(config, _main)
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/site-packages/_pytest/main.py", line 269, in wrap_session
    session.exitstatus = doit(config, session) or 0
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/site-packages/_pytest/main.py", line 323, in _main
    config.hook.pytest_runtestloop(session=session)
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/site-packages/pluggy/hooks.py", line 286, in __call__
    return self._hookexec(self, self.get_hookimpls(), kwargs)
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/site-packages/pluggy/manager.py", line 93, in _hookexec
    return self._inner_hookexec(hook, methods, kwargs)
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/site-packages/pluggy/manager.py", line 84, in <lambda>
    self._inner_hookexec = lambda hook, methods, kwargs: hook.multicall(
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/site-packages/pluggy/callers.py", line 187, in _multicall
    res = hook_impl.function(*args)
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/site-packages/_pytest/main.py", line 348, in pytest_runtestloop
    item.config.hook.pytest_runtest_protocol(item=item, nextitem=nextitem)
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/site-packages/pluggy/hooks.py", line 286, in __call__
    return self._hookexec(self, self.get_hookimpls(), kwargs)
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/site-packages/pluggy/manager.py", line 93, in _hookexec
    return self._inner_hookexec(hook, methods, kwargs)
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/site-packages/pluggy/manager.py", line 84, in <lambda>
    self._inner_hookexec = lambda hook, methods, kwargs: hook.multicall(
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/site-packages/pluggy/callers.py", line 187, in _multicall
    res = hook_impl.function(*args)
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/site-packages/_pytest/runner.py", line 109, in pytest_runtest_protocol
    runtestprotocol(item, nextitem=nextitem)
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/site-packages/_pytest/runner.py", line 126, in runtestprotocol
    reports.append(call_and_report(item, "call", log))
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/site-packages/_pytest/runner.py", line 215, in call_and_report
    call = call_runtest_hook(item, when, **kwds)
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/site-packages/_pytest/runner.py", line 254, in call_runtest_hook
    return CallInfo.from_call(
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/site-packages/_pytest/runner.py", line 311, in from_call
    result: Optional[TResult] = func()
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/site-packages/_pytest/runner.py", line 255, in <lambda>
    lambda: ihook(item=item, **kwds), when=when, reraise=reraise
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/site-packages/pluggy/hooks.py", line 286, in __call__
    return self._hookexec(self, self.get_hookimpls(), kwargs)
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/site-packages/pluggy/manager.py", line 93, in _hookexec
    return self._inner_hookexec(hook, methods, kwargs)
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/site-packages/pluggy/manager.py", line 84, in <lambda>
    self._inner_hookexec = lambda hook, methods, kwargs: hook.multicall(
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/site-packages/pluggy/callers.py", line 187, in _multicall
    res = hook_impl.function(*args)
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/site-packages/_pytest/runner.py", line 162, in pytest_runtest_call
    item.runtest()
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/site-packages/_pytest/python.py", line 1641, in runtest
    self.ihook.pytest_pyfunc_call(pyfuncitem=self)
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/site-packages/pluggy/hooks.py", line 286, in __call__
    return self._hookexec(self, self.get_hookimpls(), kwargs)
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/site-packages/pluggy/manager.py", line 93, in _hookexec
    return self._inner_hookexec(hook, methods, kwargs)
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/site-packages/pluggy/manager.py", line 84, in <lambda>
    self._inner_hookexec = lambda hook, methods, kwargs: hook.multicall(
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/site-packages/pluggy/callers.py", line 187, in _multicall
    res = hook_impl.function(*args)
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/site-packages/_pytest/python.py", line 183, in pytest_pyfunc_call
    result = testfunction(**testargs)
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/site-packages/pytest_asyncio/plugin.py", line 155, in inner
    loop.run_until_complete(task)
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/asyncio/base_events.py", line 603, in run_until_complete
    self.run_forever()
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/asyncio/base_events.py", line 570, in run_forever
    self._run_once()
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/asyncio/base_events.py", line 1823, in _run_once
    event_list = self._selector.select(timeout)
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.8/selectors.py", line 558, in select
    kev_list = self._selector.control(None, max_ev, timeout)

+++++++++++++++++++++++++++++++++++ Timeout ++++++++++++++++++++++++++++++++++++

[fjetter]

The test stimulates an upscaling to 20 workers. I could track this down to a race condition where the cluster close would hang in the SpecCluster._lock while trying to upscale the 20 workers here

distributed/distributed/deploy/spec.py

Line 369 in 6b08926

await asyncio.wait(workers)

while simulateneously trying to close. The close call tries to scale down to zero workers and issues another correct_internal but that will hit the lock.

Haven't found out if this is resolved given enough time or if this is an actual deadlock

I suspected a connection to a different problem and will need to postpone a fix. if anybody else is interested to pick this up, I'm happy to provide assistance.

[crusaderky]

This is a purely windows-specific issue; I've never observed it anywhere else.
#5022 prevents pytest-timeout from kicking in by introducing proper in-test timeout. The test still times out though, it just does not tear down the whole test suite anymore. This also gives us a proper log: https://github.com/crusaderky/distributed/runs/3024124327?check_suite_focus=true
The same PR marks this test as xfail on Windows.