Issue with obtaining large output from worker machines

This problem is a bit hard to explain, as it encompasses various experiments I did with Dask, but I will try to go step by step.

I have been using Dask for a while now along with the distributed library. My current setup is a cluster with one master and 9 homogeneous worker machines. I've set up dask workers in each machines without trouble, and so far I've done experiments on various large scale workloads, including Word Count and Sort.

A key characteristics of the above mentioned experiments is that, while the input data given to mappers can be gigantic (I've tested upto 32 gigs of input), the output provided by each task is quite small. Recently, I've been attempting to parallelize a bioinformatics application. For that implementation, the input is quite small (around 2-4 gigs). However, each task is supposed to output a pandas dataframe of around 1000 rows and 11 columns. So the output of each task for this application is bigger than my previous experiments.

First, I ran the application using a single machine with multiple cores acting as task handlers. From that run, I determined the total output size of around 160 tasks to be 141 megabytes, so on average each task provides an output of 0.88 megabytes.

However, when I try to run the same application on Dask, the execution stalls. I used dask bag to parallelize the input, then provide the mapper with the mapper function. After I call the compute function, the terminal halts. No output is shown.

I checked the memory of master and worker machines while executing. When computation starts, master machine's memory quickly starts to fill up. Before execution, around 4GB of memory is full (out of 32GB). When the used memory is around 12GB, the filling process stops. After that, no activity is recorded for the master machine memory. As for the workers, I did not detect any such memory or CPU surge while computation was ongoing.

Along with the memory problem, I encountered another problem which is similar to issue #1012. While the computation was running, all the workers were removed with the following general error messages -
```
Worker 'tcp://X.X.X.X:40028' failed from closed comm: Stream is closed
Remove worker tcp://X.X.X.X:40028
```
This happened with almost all the workers. After being removed, the connections were established again, although the computation was still stuck.

I thought this might be a problem with python queue, as python queue can get stuck if too much data is inserted into it. so I used [Data Streams with Queues](https://github.com/dask/distributed/blob/master/docs/source/queues.rst) to insert input manually into a queue, send the queue to workers and then tried to get output from the stream. I was unsuccessful with this method also; the computation was still stuck.

So, the main problem here is that *dask distributed computation is getting stuck when working with tasks which are supposed to produce large outputs. Workers streams are being closed and removed, and master machine memory continues to fill up without explanation*.

I should mention here that due to bokeh/tornado incompability (which I learned from issue #1057), I am using tornado 4.4 instead of the latest versions. All other packages are updated up to 7 days ago as of this writing.

I realize I may not have been eloquent with my description of the problem, so please let me know if I need to explain anything further about any individual detail. If possible, I want to solve this problem myself, so a bit of hint about where the problem may lie will also be helpful. All these experiments are related to my masters thesis, and I might have a publication in near future should I solve this problem ;) Thanks for your time.