From: Ben Darnell <ben@bendarnell.com>
Date: Sun, 16 Nov 2014 00:08:21 +0000 (-0500)
Subject: Fix IOStream flow control under kqueue.
X-Git-Tag: v4.1.0b1~55
X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=41ff81b2375720dafe986a9a343daf49ae6253f8;p=thirdparty%2Ftornado.git

Fix IOStream flow control under kqueue.

An early attempt to detect closed connections in kqueue led
us to always read from the socket when data was available,
defeating any higher-level flow control.

Add a test to verify this behavior.
---

diff --git a/tornado/platform/kqueue.py b/tornado/platform/kqueue.py
index de8c046d3..f8f3e4a61 100644
--- a/tornado/platform/kqueue.py
+++ b/tornado/platform/kqueue.py
@@ -54,8 +54,7 @@ class _KQueue(object):
         if events & IOLoop.WRITE:
             kevents.append(select.kevent(
                 fd, filter=select.KQ_FILTER_WRITE, flags=flags))
-        if events & IOLoop.READ or not kevents:
-            # Always read when there is not a write
+        if events & IOLoop.READ:
             kevents.append(select.kevent(
                 fd, filter=select.KQ_FILTER_READ, flags=flags))
         # Even though control() takes a list, it seems to return EINVAL
diff --git a/tornado/test/iostream_test.py b/tornado/test/iostream_test.py
index fa8590c83..f54eed6bb 100644
--- a/tornado/test/iostream_test.py
+++ b/tornado/test/iostream_test.py
@@ -725,6 +725,26 @@ class TestIOStreamMixin(object):
             server.close()
             client.close()
 
+    def test_flow_control(self):
+        MB = 1024 * 1024
+        server, client = self.make_iostream_pair(max_buffer_size=5 * MB)
+        try:
+            # Client writes more than the server will accept.
+            client.write(b"a" * 10 * MB)
+            # The server pauses while reading.
+            server.read_bytes(MB, self.stop)
+            self.wait()
+            self.io_loop.call_later(0.1, self.stop)
+            self.wait()
+            # The client's writes have been blocked; the server can
+            # continue to read gradually.
+            for i in range(9):
+                server.read_bytes(MB, self.stop)
+                self.wait()
+        finally:
+            server.close()
+            client.close()
+
 
 class TestIOStreamWebHTTP(TestIOStreamWebMixin, AsyncHTTPTestCase):
     def _make_client_iostream(self):