pubsub: not all put() operations are completed

When using the pub/sub pattern in dask distributed the object put in the publish method is not deeply copied to the buffer. Instead it is somewhat loosly coupled. Thinking about message queueing I expected the object to be moved to the queue.

If the call to ` p.put(copy.copy(mydict))` would be done without the copy method, the number `n` in the dictionary would be referenced to the last given number (`9` in this case). This is not the behaviour I expected.

When using larger cycle numbers in n (`range(1000)`) n stays the same for about 10-20 messages and then jumps to the next reference. This is certainly unpredictable behaviour.

It would be great if the message handling would create copies of the object to ensure predictable behaviour. What about adding an flag to the method for optionally not copying the message?

```
import time
from dask.distributed import LocalCluster, Client, Pub, Sub,wait
import copy

def subscriber_function():
print("Calling my loop listening to Test123")
s=Sub('Test123')
n=0
#while True:
for msg in s:
n = n + 1
print("Waiting for notification, n: {}\n".format(str(n)))
print(msg)

def publish_function():
mydict= {
'name' : 'DEADBEEF',
'handler' : 1234
}
p = Pub('Test123')
for n in range(10):
mydict['n'] = str(n)
# A deep copy of the dictionary is needed to ensure a valid transaction
p.put(copy.copy(mydict))

if __name__=='__main__':
cluster=LocalCluster(scheduler_port=12345,processes=False)
myscheduler = Client(cluster)
myscheduler.submit(subscriber_function)
a = myscheduler.submit(publish_function)
wait(a)
```