flaky `test_quiet_close_process[True]`

```
_______________________ test_quiet_close_process[True] ________________________

processes = True
tmp_path = WindowsPath('C:/Users/runneradmin/AppData/Local/Temp/pytest-of-runneradmin/pytest-0/test_quiet_close_process_True_0')

@pytest.mark.slow
@pytest.mark.parametrize("processes", [True, False])
def test_quiet_close_process(processes, tmp_path):
with open(tmp_path / "script.py", mode="w") as f:
f.write(client_script % processes)

with popen([sys.executable, tmp_path / "script.py"], capture_output=True) as proc:
> out, err = proc.communicate(timeout=10)

distributed\tests\test_client.py:7554:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
C:\Miniconda3\envs\dask-distributed\lib\subprocess.py:1134: in communicate
stdout, stderr = self._communicate(input, endtime, timeout)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

self =
input = None, endtime = 1556.765, orig_timeout = 10

def _communicate(self, input, endtime, orig_timeout):
# Start reader threads feeding into a list hanging off of this
# object, unless they've already been started.
if self.stdout and not hasattr(self, "_stdout_buff"):
self._stdout_buff = []
self.stdout_thread = \
threading.Thread(target=self._readerthread,
args=(self.stdout, self._stdout_buff))
self.stdout_thread.daemon = True
self.stdout_thread.start()
if self.stderr and not hasattr(self, "_stderr_buff"):
self._stderr_buff = []
self.stderr_thread = \
threading.Thread(target=self._readerthread,
args=(self.stderr, self._stderr_buff))
self.stderr_thread.daemon = True
self.stderr_thread.start()

if self.stdin:
self._stdin_write(input)

# Wait for the reader threads, or time out. If we time out, the
# threads remain reading and the fds left open in case the user
# calls communicate again.
if self.stdout is not None:
self.stdout_thread.join(self._remaining_time(endtime))
if self.stdout_thread.is_alive():
> raise TimeoutExpired(self.args, orig_timeout)
E subprocess.TimeoutExpired: Command '['C:\\Miniconda3\\envs\\dask-distributed\\python.exe', WindowsPath('C:/Users/runneradmin/AppData/Local/Temp/pytest-of-runneradmin/pytest-0/test_quiet_close_process_True_0/script.py')]' timed out after 10 seconds

C:\Miniconda3\envs\dask-distributed\lib\subprocess.py:1510: TimeoutExpired
---------------------------- Captured stdout call -----------------------------
------ stdout: returncode 3221225786, ['C:\\Miniconda3\\envs\\dask-distributed\\python.exe', WindowsPath('C:/Users/runneradmin/AppData/Local/Temp/pytest-of-runneradmin/pytest-0/test_quiet_close_process_True_0/script.py')] ------
2022-08-09 06:19:33,073 - tornado.application - ERROR - Exception in callback functools.partial(>, exception=RuntimeError('cannot schedule new futures after shutdown')>)

Traceback (most recent call last):

File "C:\Miniconda3\envs\dask-distributed\lib\site-packages\tornado\ioloop.py", line 741, in _run_callback

ret = callback()

File "C:\Miniconda3\envs\dask-distributed\lib\site-packages\tornado\ioloop.py", line 765, in _discard_future_result

future.result()

File "d:\a\distributed\distributed\distributed\deploy\spec.py", line 423, in _close

await self._correct_state()

File "d:\a\distributed\distributed\distributed\deploy\spec.py", line 337, in _correct_state_internal

await self.scheduler_comm.retire_workers(workers=list(to_close))

File "d:\a\distributed\distributed\distributed\core.py", line 1071, in send_recv_from_rpc

comm = await self.live_comm()

File "d:\a\distributed\distributed\distributed\core.py", line 1030, in live_comm

comm = await connect(

File "d:\a\distributed\distributed\distributed\comm\core.py", line 291, in connect

comm = await asyncio.wait_for(

File "C:\Miniconda3\envs\dask-distributed\lib\asyncio\tasks.py", line 479, in wait_for

return fut.result()

File "d:\a\distributed\distributed\distributed\comm\tcp.py", line 451, in connect

stream = await self.client.connect(

File "C:\Miniconda3\envs\dask-distributed\lib\site-packages\tornado\tcpclient.py", line 265, in connect

addrinfo = await self.resolver.resolve(host, port, af)

File "d:\a\distributed\distributed\distributed\comm\tcp.py", line 436, in resolve

for fam, _, _, _, address in await asyncio.get_running_loop().getaddrinfo(

File "C:\Miniconda3\envs\dask-distributed\lib\asyncio\base_events.py", line 861, in getaddrinfo

return await self.run_in_executor(

File "C:\Miniconda3\envs\dask-distributed\lib\asyncio\base_events.py", line 819, in run_in_executor

executor.submit(func, *args), loop=self)

File "C:\Miniconda3\envs\dask-distributed\lib\concurrent\futures\thread.py", line 167, in submit

raise RuntimeError('cannot schedule new futures after shutdown')

RuntimeError: cannot schedule new futures after shutdown
```

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