minimal example to reproduce the "memory problem" of dask.distributed

The 'memory leak' (maybe not exact) problem of dask messed me up for a long time and it's much urgent for me these days. Therefore, I tried to make a minimal example here for more help from the great developer.

The code below is simple: 1) generate a list of dataframe, 2）do a two stage calculation on these dataframe. 3) wait for these tasks to be finished. **The key point is that if I make the datatype of the dataframe to be np.float32 before scattering them onto the cluster, when the programme finished, the memory will be not released(It's easy to see from htop). If we do not change the datatype(comment out that line of code:`df_list = [df.astype(np.float32) for df in df_list] `, the memory will be ok(released) aftering the code is finished.**

the whole code is listed as below(dask_debug.py)
[dask_debug.zip](https://github.com/dask/distributed/files/3669331/dask_debug.zip)
```
import pandas as pd
import numpy as np
from dask.distributed import Client
import time

client = Client('tcp://10.18.8.2:8786')
base_n = 50

df_list = [pd.DataFrame(np.random.rand(2188, 3563), index= range(2188), columns = range(3563)) for _ in range(base_n)] #generate base data

df_list = [df.astype(np.float32) for df in df_list] # the key of the memory problem !!!! if this line is commented out , when the code finishing, the memory is released. If this line is used, when the code finishing, the memroy is not released.

df_list_future = [client.scatter(df, broadcast=True) for df in df_list] #scatter the base data on the workers

#define two simple function standing for my calculation: add two dataframe, then, get the sum of all the element
def add_df(df1, df2):
return df1 + df2
def get_sum(df):
return df.sum().sum()

#do the calcuation between every pair of the base data
result_list = [] # store the result
for future1 in df_list_future:
for future2 in df_list_future:
#two stage calculation
r_add = client.submit(add_df, future1, future2, workers='127.0.0.2') #worker 1
r = client.submit(get_sum, r_add, workers='127.0.0.3' ) #worker 2
result_list.append((r_add, r))

#pop each result and wait for them to comlete
while result_list:
result_one = result_list.pop(0)
while not (result_one[1].status == 'finished'):
time.sleep(1)
del result_one
print(len(result_list))
```

To run the above code, I used the cmdline to launch one scheduler and two workers:
```
$ dask-scheduler
$ dask-worker tcp://10.18.8.2:8786 --host 127.0.0.2 --nprocs 5 --nthreads 1 --memory-limit 50e9
$ dask-worker tcp://10.18.8.2:8786 --host 127.0.0.3 --nprocs 3 --nthreads 1 --memory-limit 30e9
```
and then:
`python dask_debug.py`

It should be easy to be reproduced(I confirmed it many times on my machine). Great great thanks for any suggestions. @mrocklin