Worker memory not being freed when tasks complete

I'm still investigating, but in the meantime I wanted to get this issue started.

I'm noticing that after executing a task graph with large inputs and a small output, my worker memory stays high. In the example below we

1. Generate data (large byte strings)
2. filter data (slice)
3. reduce many tasks (sum)

So the final result returned to the client is small, a single Python int. The only large objects should be the initially generated bytestrings.

The console output below is

1. per-worker memory usage before the computation (~30 MB)
2. per-worker memory usage right after the computation (~ 230 MB)
3. per-worker memory usage 5 seconds after, in case things take some time to settle down. (~ 230 MB)

```
Memory usage [before]
{'tcp://192.168.7.20:50533': '30.92 MB', 'tcp://192.168.7.20:50534': '30.95 MB'}
running
Memory usage [after]
{'tcp://192.168.7.20:50533': '231.97 MB',
'tcp://192.168.7.20:50534': '232.63 MB'}
Memory usage [after]
{'tcp://192.168.7.20:50533': '232.05 MB',
'tcp://192.168.7.20:50534': '232.63 MB'}
```

In an effort to test whether the scheduler or worker is holding a reference to the data, I submit a bunch of tiny `inc` tasks to one of the worker. I notice that the memory on that worker does settle down

```
Memory usage [final]
{'tcp://192.168.7.20:52114': '232.77 MB',
'tcp://192.168.7.20:52115': '49.73 MB'}
```

That's at least consistent with the worker or scheduler holding a reference to the data, but there could be many other causes. I'm still debugging.

The number of `inc` tasks, 2731, seems to be significant. With 2730 `inc` tasks, I don't see any memory reduction on that worker.

```python
import time
from dask.utils import parse_bytes, format_bytes
import pprint
import string
import toolz
from distributed import Client, wait

N = parse_bytes("100 Mb")
I = 20

def inc(x):
return x + 1

def f(x, n=N):
time.sleep(0.05)
return string.ascii_letters[x % 52].encode() * n

def g(x):
time.sleep(0.02)
return x[:5]

def h(*args):
return sum(x[0] for x in args)

def get_mem(dask_worker):
return dask_worker.monitor.proc.memory_info().rss

def main():
dsk = {}
for i in range(I):
dsk[f'a-{i}'] = (f, i, N)
dsk[f'b-{i}'] = (g, f'a-{i}')
dsk['c-0'] = (h,) + tuple(f'b-{i}' for i in range(I))

with Client(n_workers=2, threads_per_worker=1, memory_limit='500Mb', processes=True) as client:
print("Memory usage [before]")
pprint.pprint(toolz.valmap(format_bytes, client.run(get_mem)))

print("running")
client.get(dsk, keys=["c-0"])
time.sleep(2) # let things settle

print("Memory usage [after]")
pprint.pprint(toolz.valmap(format_bytes, client.run(get_mem)))

time.sleep(5) # settle some more?
print("Memory usage [after]")
pprint.pprint(toolz.valmap(format_bytes, client.run(get_mem)))

print("clear things?")
futures = client.map(inc, range(2731), pure=False)
wait(futures)
del futures

print("Memory usage [final]")
pprint.pprint(toolz.valmap(format_bytes, client.run(get_mem)))

if __name__ == '__main__':
main()

```