Tornado Already reading errors

[fjetter]

We've seen lately a few failing tests with low level tornado assert errors

E.g. in

• Unit test for metrics["memory"]=None #4727
• Avoid active_threads changing size during iteration #4729

CI runs

• https://github.com/dask/distributed/pull/4729/checks?check_run_id=2412174289

distributed/cfexecutor.py:146: in result_iterator
    self._client.cancel(remaining)
distributed/client.py:2212: in cancel
    return self.sync(self._cancel, futures, asynchronous=asynchronous, force=force)
distributed/client.py:843: in sync
    return sync(
distributed/utils.py:353: in sync
    raise exc.with_traceback(tb)
distributed/utils.py:336: in f
    result[0] = yield future
../../../miniconda3/envs/dask-distributed/lib/python3.8/site-packages/tornado/gen.py:762: in run
    value = future.result()
distributed/client.py:2192: in _cancel
    await self.scheduler.cancel(keys=keys, client=self.id, force=force)
distributed/core.py:862: in send_recv_from_rpc
    result = await send_recv(comm=comm, op=key, **kwargs)
distributed/core.py:645: in send_recv
    response = await comm.read(deserializers=deserializers)
distributed/comm/tcp.py:196: in read
    frames_nbytes = await stream.read_bytes(fmt_size)
../../../miniconda3/envs/dask-distributed/lib/python3.8/site-packages/tornado/iostream.py:421: in read_bytes
    future = self._start_read()
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tornado.iostream.IOStream object at 0x7fa73fc71130>

    def _start_read(self) -> Future:
        if self._read_future is not None:
            # It is an error to start a read while a prior read is unresolved.
            # However, if the prior read is unresolved because the stream was
            # closed without satisfying it, it's better to raise
            # StreamClosedError instead of AssertionError. In particular, this
            # situation occurs in harmless situations in http1connection.py and
            # an AssertionError would be logged noisily.
            #
            # On the other hand, it is legal to start a new read while the
            # stream is closed, in case the read can be satisfied from the
            # read buffer. So we only want to check the closed status of the
            # stream if we need to decide what kind of error to raise for
            # "already reading".
            #
            # These conditions have proven difficult to test; we have no
            # unittests that reliably verify this behavior so be careful
            # when making changes here. See #2651 and #2719.
            self._check_closed()
>           assert self._read_future is None, "Already reading"
E           AssertionError: Already reading

../../../miniconda3/envs/dask-distributed/lib/python3.8/site-packages/tornado/iostream.py:809: AssertionError