Scheduler deadlocks after asynchronous `client.who_has` call

**What happened**:

Dear `dask-distributed` developers,

First of all, thank you for this wonderful project!
We are using `dask-distributed` for our local HTCondor computing cluster.
In our use-case we periodically kill and spawn `dask-workers` in HTCondorJobs, such that other HTCondorJobs from other users can slide in between our computing runs.
We also need to work with heavy input, which we need to distribute to the `dask-workers` beforehand using `client.scatter`. Of course we want to replicate this as soon as new `dask-worker` are spawned. Thus we added an asynchronous periodic callback to the `client`'s IOLoop, which takes care of this replication.
Unfortunately we noticed that the `client.who_has(..., asynchronous=True)` call deadlocks our scheduler (unfortunately without a stack trace). Any connection to the scheduler results then in a timeout.

**What you expected to happen**:

We expected that we can add a asynchronous callback, which uses `client.who_has(..., asynchronous=True)`, to the `client`'s IOLoop without deadlocking the scheduler.

**Minimal Complete Verifiable Example**:

This is a minimal reproducible example, which shows the above-mentioned problem. Since it also happens on a `LocalCluster` the problem seems to be batch-system-agnostic.

```python
# coding: utf-8

from tornado.ioloop import PeriodicCallback
from dask.distributed import LocalCluster, Client
import numpy as np

class Heavy:
def __init__(self):
self.state = np.arange(1000000)

def __call__(self, arg):
return np.sum(self.state * arg)

heavy = Heavy()

async def replicate() -> None:
print("In callback")
workers = client.ncores(asynchronous=True)
print("Workers", await workers)
avail = client.who_has([func], asynchronous=True) # <--- This seems to deadlock the scheduler
print("Avail", await avail)
missing = set((await workers).keys()) - set((await avail)[func.key])
print("Missing", missing)
if missing:
try:
await client.replicate(func, asynchronous=True)
print("replicate: success!")
except:
print("replicate: failed!")

if __name__ == "__main__":
cluster = LocalCluster()
client = Client(cluster)

func = client.scatter(heavy, broadcast=True, hash=False, direct=None)

pc = PeriodicCallback(replicate, callback_time=100)
client.loop.add_callback(pc.start)
```

The output (only once!):

```bash
In callback
Workers {'tls://127.0.0.1:17693': 8, 'tls://127.0.0.1:17747': 8, 'tls://127.0.0.1:18651': 8, 'tls://127.0.0.1:20033': 8, 'tls://127.0.0.1:25061': 8, 'tls://127.0.0.1:4257': 8, 'tls://127.0.0.1:6413': 8, 'tls://127.0.0.1:9189': 8}
```
Afterwards the scheduler is stuck.

**Anything else we need to know?**:

\-

**Environment**:

- Dask version: '2021.06.2'
- Tornado version: '6.2.dev1'
- Python version: '3.8.10'
- Operating System: 'Ubuntu 20.04.2 LTS'
- Install method (conda, pip, source): With pip into a conda environment

`client.get_versions(check=True)` does not throw an error and outputs:

```bash
'client': {'host': {'python': '3.8.10.final.0',
'python-bits': 64,
'OS': 'Linux',
'OS-release': '5.4.0-80-generic',
'machine': 'x86_64',
'processor': 'x86_64',
'byteorder': 'little',
'LC_ALL': 'en_US.UTF-8',
'LANG': 'en_US.UTF-8'},
'packages': {'python': '3.8.10.final.0',
'dask': '2021.06.2',
'distributed': '2021.06.2',
'msgpack': '1.0.2',
'cloudpickle': '1.6.0',
'tornado': '6.2.dev1',
'toolz': '0.11.1',
'numpy': '1.19.5',
'lz4': '3.1.3',
'blosc': '1.10.4'}}}
```

Thank you very much in advance for your input and help!

Best, Peter