A possible bug in the pub/sub system

Hi, I'm using dask 2.6.0 on Linux with Jupyter. See the following code:

```
from distributed import Client, LocalCluster, Pub, Sub
cluster = LocalCluster(n_workers=4, threads_per_worker=1)
client = Client(cluster)

def run(key):
sub = Sub(key)
def worker(i):
pub = Pub(key)
pub.put(['finished', i])
client.map(worker, range(4))
finished = [False, False, False, False]
for msg in sub:
if msg[0] == 'finished':
finished[msg[1]] = True
if all(finished):
break
print(['finished', key])

run('1')
```

The motivation is to use the pub/sub system to monitor the progress in the worker. The code above works well with the following output:

```
['finished', '1']
```

And it also works well if I execute `run('1')` again in a new cell:

```
['finished', '1']
```

But if I execute `run('2')` in a new cell, I get:

```
['finished', '2']

distributed.core - ERROR - 'Client-c9530ed2-112b-11ea-8d56-44a842272021'
Traceback (most recent call last):
File "/global/homes/h/hejia/.conda/envs/hejia@cori-2/lib/python3.6/site-packages/distributed/core.py", line 464, in handle_stream
handler(**merge(extra, msg))
File "/global/homes/h/hejia/.conda/envs/hejia@cori-2/lib/python3.6/site-packages/distributed/pubsub.py", line 88, in remove_subscriber
self.client_subscribers[name].remove(client)
KeyError: 'Client-c9530ed2-112b-11ea-8d56-44a842272021'
distributed.core - ERROR - 'Client-c9530ed2-112b-11ea-8d56-44a842272021'
Traceback (most recent call last):
File "/global/homes/h/hejia/.conda/envs/hejia@cori-2/lib/python3.6/site-packages/distributed/core.py", line 403, in handle_comm
result = await result
File "/global/homes/h/hejia/.conda/envs/hejia@cori-2/lib/python3.6/site-packages/distributed/scheduler.py", line 2253, in add_client
await self.handle_stream(comm=comm, extra={"client": client})
File "/global/homes/h/hejia/.conda/envs/hejia@cori-2/lib/python3.6/site-packages/distributed/core.py", line 464, in handle_stream
handler(**merge(extra, msg))
File "/global/homes/h/hejia/.conda/envs/hejia@cori-2/lib/python3.6/site-packages/distributed/pubsub.py", line 88, in remove_subscriber
self.client_subscribers[name].remove(client)
KeyError: 'Client-c9530ed2-112b-11ea-8d56-44a842272021'
```

While if I execute them together (two `run('1')` and one `run('2')`) in a single cell, it gets stuck after giving the first output:

```
['finished', '1']
```

So could you help me look into this issue? Is it because I'm not using the pub/sub system correctly?