Tornado Already reading errors

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

E.g. in
- https://github.com/dask/distributed/pull/4727
- https://github.com/dask/distributed/pull/4729

CI runs

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

```python
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 =

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

```