Performance degradation of repeated aggregations with distributed scheduler and client.persist() on a single machine

The following code illustrates a performance issue with dask's distributed scheduler when combined the client.persist(df) functionality on a single machine:

```
from dask import distributed
import multiprocessing
cluster = distributed.LocalCluster(n_workers=multiprocessing.cpu_count(), threads_per_worker=1)
dask_client = distributed.Client(cluster)
dask_df = dd.from_pandas(df, npartitions=multiprocessing.cpu_count())
...
cvs = datashader.Canvas(...)
agg = cvs.points(dask_df, ...)
```

When we insert the following line before `agg = cvs.points(dask_df, ...)`:
```
dask_df = dask_client.persist(dask_df)
```

The runtime performance of `cvs.points()` gets much worse than when `persist` was not used.