Lack of backpressure #641

It appears that we can get streams to lose data if we push a large amount of data without receiving. We can trigger this in the following test if we crank up the size of the sent data an order of magnitude:

distributed/tests/test_batched.py::test_sending_traffic_jam

I was able to reproduce this without the BatchedSend object, instead using standard distributed.core.read/write on normal Tornado streams.

This issue has only come up in stress testing, not in the wild, but is somewhat concerning.