flush SO_SNDBUF data on every send and deprecate send_bytesso_sndbuf

8 files changed, 65 insertions, 20 deletions

diff --git a/CHANGES.txt b/CHANGES.txt
index a0352ba..5a3b0c7 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,7 +1,14 @@
 unreleased
 ----------
 
-Bugfixes
+Deprecations
+~~~~~~~~~~~~
+
+- The ``send_bytes`` adjustment now defaults to ``1`` and is deprecated
+  pending removal in a future release.
+  and https://github.com/Pylons/waitress/pull/246
+
+Features
 ~~~~~~~~
 
 - Stop early and close the ``app_iter`` when attempting to write to a closed
@@ -11,6 +18,17 @@ Bugfixes
   and https://github.com/Pylons/waitress/pull/240
   and https://github.com/Pylons/waitress/pull/241
 
+- Adjust the flush to output ``SO_SNDBUF`` bytes instead of whatever was
+  set in the ``send_bytes`` adjustment. ``send_bytes`` now only controls how
+  much waitress will buffer internally before flushing to the kernel, whereas
+  previously it used to also throttle how much data was sent to the kernel.
+  This change enables a streaming ``app_iter`` containing small chunks to
+  still be flushed efficiently.
+  See https://github.com/Pylons/waitress/pull/246
+
+Bugfixes
+~~~~~~~~
+
 - When a client closes a socket unexpectedly there was potential for memory
   leaks in which data was written to the buffers after they were closed,
   causing them to reopen.
diff --git a/docs/api.rst b/docs/api.rst
index 5e0a523..70c174c 100644
--- a/docs/api.rst
+++ b/docs/api.rst
@@ -5,6 +5,6 @@
 
 .. module:: waitress
 
-.. function:: serve(app, listen='0.0.0.0:8080', unix_socket=None, unix_socket_perms='600', threads=4, url_scheme='http', url_prefix='', ident='waitress', backlog=1204, recv_bytes=8192, send_bytes=18000, outbuf_overflow=104856, inbuf_overflow=52488, connection_limit=1000, cleanup_interval=30, channel_timeout=120, log_socket_errors=True, max_request_header_size=262144, max_request_body_size=1073741824, expose_tracebacks=False)
+.. function:: serve(app, listen='0.0.0.0:8080', unix_socket=None, unix_socket_perms='600', threads=4, url_scheme='http', url_prefix='', ident='waitress', backlog=1204, recv_bytes=8192, send_bytes=1, outbuf_overflow=104856, inbuf_overflow=52488, connection_limit=1000, cleanup_interval=30, channel_timeout=120, log_socket_errors=True, max_request_header_size=262144, max_request_body_size=1073741824, expose_tracebacks=False)
 
      See :ref:`arguments` for more information.
diff --git a/docs/arguments.rst b/docs/arguments.rst
index 2665087..8bacc44 100644
--- a/docs/arguments.rst
+++ b/docs/arguments.rst
@@ -190,7 +190,9 @@ send_bytes
     Linux, ``/proc/sys/net/ipv4/tcp_wmem`` controls the minimum, default, and
     maximum sizes of TCP write buffers.
 
-    Default: ``18000``
+    Default: ``1``
+
+    .. deprecated:: 1.3
 
 outbuf_overflow
     A tempfile should be created if the pending output is larger than
diff --git a/docs/runner.rst b/docs/runner.rst
index fc424ad..0b61307 100644
--- a/docs/runner.rst
+++ b/docs/runner.rst
@@ -143,9 +143,11 @@ Tuning options:
     8192.
 
 ``--send-bytes=INT``
-    Number of bytes to send to socket.send(). Default is 18000.
+    Number of bytes to send to socket.send(). Default is 1.
     Multiples of 9000 should avoid partly-filled TCP packets.
 
+    .. deprecated:: 1.3
+
 ``--outbuf-overflow=INT``
     A temporary file should be created if the pending output is larger than
     this. Default is 1048576 (1MB).
diff --git a/waitress/adjustments.py b/waitress/adjustments.py
index aa55c73..3b0b364 100644
--- a/waitress/adjustments.py
+++ b/waitress/adjustments.py
@@ -195,11 +195,9 @@ class Adjustments(object):
     # recv_bytes is the argument to pass to socket.recv().
     recv_bytes = 8192
 
-    # send_bytes is the number of bytes to send to socket.send().  Multiples
-    # of 9000 should avoid partly-filled packets, but don't set this larger
-    # than the TCP write buffer size.  In Linux, /proc/sys/net/ipv4/tcp_wmem
-    # controls the minimum, default, and maximum sizes of TCP write buffers.
-    send_bytes = 18000
+    # deprecated setting controls how many bytes will be buffered before
+    # being flushed to the socket
+    send_bytes = 1
 
     # A tempfile should be created if the pending output is larger than
     # outbuf_overflow, which is measured in bytes. The default is 1MB.  This
@@ -289,6 +287,12 @@ class Adjustments(object):
         if 'unix_socket' in kw and 'listen' in kw:
             raise ValueError('unix_socket may not be set if listen is set')
 
+        if 'send_bytes' in kw:
+            warnings.warn(
+                'send_bytes will be removed in a future release',
+                DeprecationWarning,
+            )
+
         for k, v in kw.items():
             if k not in self._param_map:
                 raise ValueError('Unknown adjustment %r' % k)
diff --git a/waitress/channel.py b/waitress/channel.py
index 5af549e..8fee59c 100644
--- a/waitress/channel.py
+++ b/waitress/channel.py
@@ -71,6 +71,7 @@ class HTTPChannel(wasyncore.dispatcher, object):
         self.outbufs = [OverflowableBuffer(adj.outbuf_overflow)]
         self.total_outbufs_len = 0
         self.creation_time = self.last_activity = time.time()
+        self.sendbuf_len = sock.getsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF)
 
         # task_lock used to push/pop requests
         self.task_lock = threading.Lock()
@@ -82,14 +83,11 @@ class HTTPChannel(wasyncore.dispatcher, object):
         # Don't let wasyncore.dispatcher throttle self.addr on us.
         self.addr = addr
 
-    def any_outbuf_has_data(self):
-        return self.total_outbufs_len > 0
-
     def writable(self):
         # if there's data in the out buffer or we've been instructed to close
         # the channel (possibly by our server maintenance logic), run
         # handle_write
-        return self.any_outbuf_has_data() or self.will_close
+        return self.total_outbufs_len or self.will_close
 
     def handle_write(self):
         # Precondition: there's data in the out buffer to be sent, or
@@ -106,7 +104,7 @@ class HTTPChannel(wasyncore.dispatcher, object):
             #    because it's either data left over from task output
             #    or a 100 Continue line sent within "received".
             flush = self._flush_some
-        elif (self.total_outbufs_len >= self.adj.send_bytes):
+        elif self.total_outbufs_len >= self.adj.send_bytes:
             # 1. There's a running task, so we need to try to lock
             #    the outbuf before sending
             # 2. Only try to send if the data in the out buffer is larger
@@ -128,7 +126,7 @@ class HTTPChannel(wasyncore.dispatcher, object):
                 self.logger.exception('Unexpected exception when flushing')
                 self.will_close = True
 
-        if self.close_when_flushed and not self.any_outbuf_has_data():
+        if self.close_when_flushed and not self.total_outbufs_len:
             self.close_when_flushed = False
             self.will_close = True
 
@@ -141,8 +139,7 @@ class HTTPChannel(wasyncore.dispatcher, object):
         # 2. There's no already currently running task(s).
         # 3. There's no data in the output buffer that needs to be sent
         #    before we potentially create a new task.
-        return not (self.will_close or self.requests or
-                    self.any_outbuf_has_data())
+        return not (self.will_close or self.requests or self.total_outbufs_len)
 
     def handle_read(self):
         try:
@@ -238,7 +235,7 @@ class HTTPChannel(wasyncore.dispatcher, object):
                     dobreak = True
 
             while outbuflen > 0:
-                chunk = outbuf.get(self.adj.send_bytes)
+                chunk = outbuf.get(self.sendbuf_len)
                 num_sent = self.send(chunk)
                 if num_sent:
                     outbuf.skip(num_sent, True)
diff --git a/waitress/tests/test_adjustments.py b/waitress/tests/test_adjustments.py
index e35fdaf..00b0353 100644
--- a/waitress/tests/test_adjustments.py
+++ b/waitress/tests/test_adjustments.py
@@ -361,6 +361,16 @@ class TestAdjustments(unittest.TestCase):
             self.assertTrue(issubclass(w[0].category, DeprecationWarning))
             self.assertIn("clear_untrusted_proxy_headers will be set to True", str(w[0]))
 
+    def test_deprecated_send_bytes(self):
+        with warnings.catch_warnings(record=True) as w:
+            warnings.resetwarnings()
+            warnings.simplefilter("always")
+            self._makeOne(send_bytes=1)
+
+            self.assertGreaterEqual(len(w), 1)
+            self.assertTrue(issubclass(w[0].category, DeprecationWarning))
+            self.assertIn("send_bytes", str(w[0]))
+
     def test_badvar(self):
         self.assertRaises(ValueError, self._makeOne, nope=True)
 
diff --git a/waitress/tests/test_channel.py b/waitress/tests/test_channel.py
index 3915af9..f66766b 100644
--- a/waitress/tests/test_channel.py
+++ b/waitress/tests/test_channel.py
@@ -20,6 +20,7 @@ class TestHTTPChannel(unittest.TestCase):
     def test_ctor(self):
         inst, _, map = self._makeOneWithMap()
         self.assertEqual(inst.addr, '127.0.0.1')
+        self.assertEqual(inst.sendbuf_len, 2048)
         self.assertEqual(map[100], inst)
 
     def test_total_outbufs_len_an_outbuf_size_gt_sys_maxint(self):
@@ -70,6 +71,7 @@ class TestHTTPChannel(unittest.TestCase):
         inst, sock, map = self._makeOneWithMap()
         inst.requests = []
         inst.outbufs = [DummyBuffer(b'abc')]
+        inst.total_outbufs_len = len(inst.outbufs[0])
         inst.last_activity = 0
         result = inst.handle_write()
         self.assertEqual(result, None)
@@ -82,6 +84,7 @@ class TestHTTPChannel(unittest.TestCase):
         inst.requests = []
         outbuf = DummyBuffer(b'abc', socket.error)
         inst.outbufs = [outbuf]
+        inst.total_outbufs_len = len(outbuf)
         inst.last_activity = 0
         inst.logger = DummyLogger()
         result = inst.handle_write()
@@ -96,6 +99,7 @@ class TestHTTPChannel(unittest.TestCase):
         inst.requests = []
         outbuf = DummyBuffer(b'abc', IOError)
         inst.outbufs = [outbuf]
+        inst.total_outbufs_len = len(outbuf)
         inst.last_activity = 0
         inst.logger = DummyLogger()
         result = inst.handle_write()
@@ -123,7 +127,7 @@ class TestHTTPChannel(unittest.TestCase):
         inst, sock, map = self._makeOneWithMap()
         inst.requests = [True]
         inst.outbufs = [DummyBuffer(b'abc')]
-        inst.total_outbufs_len = 3
+        inst.total_outbufs_len = len(inst.outbufs[0])
         inst.adj.send_bytes = 2
         inst.will_close = False
         inst.last_activity = 0
@@ -137,6 +141,7 @@ class TestHTTPChannel(unittest.TestCase):
         inst, sock, map = self._makeOneWithMap()
         outbuf = DummyBuffer(b'abc')
         inst.outbufs = [outbuf]
+        inst.total_outbufs_len = len(outbuf)
         inst.will_close = False
         inst.close_when_flushed = True
         inst.last_activity = 0
@@ -231,6 +236,7 @@ class TestHTTPChannel(unittest.TestCase):
     def test__flush_some_full_outbuf_socket_returns_nonzero(self):
         inst, sock, map = self._makeOneWithMap()
         inst.outbufs[0].append(b'abc')
+        inst.total_outbufs_len = sum(len(x) for x in inst.outbufs)
         result = inst._flush_some()
         self.assertEqual(result, True)
 
@@ -238,6 +244,7 @@ class TestHTTPChannel(unittest.TestCase):
         inst, sock, map = self._makeOneWithMap()
         sock.send = lambda x: False
         inst.outbufs[0].append(b'abc')
+        inst.total_outbufs_len = sum(len(x) for x in inst.outbufs)
         result = inst._flush_some()
         self.assertEqual(result, False)
 
@@ -246,6 +253,7 @@ class TestHTTPChannel(unittest.TestCase):
         sock.send = lambda x: len(x)
         buffer = DummyBuffer(b'abc')
         inst.outbufs.append(buffer)
+        inst.total_outbufs_len = sum(len(x) for x in inst.outbufs)
         result = inst._flush_some()
         self.assertEqual(result, True)
         self.assertEqual(buffer.skipped, 3)
@@ -256,6 +264,7 @@ class TestHTTPChannel(unittest.TestCase):
         sock.send = lambda x: len(x)
         buffer = DummyBuffer(b'abc')
         inst.outbufs.append(buffer)
+        inst.total_outbufs_len = sum(len(x) for x in inst.outbufs)
         inst.logger = DummyLogger()
         def doraise():
             raise NotImplementedError
@@ -632,6 +641,9 @@ class DummySock(object):
     def getpeername(self):
         return '127.0.0.1'
 
+    def getsockopt(self, level, option):
+        return 2048
+
     def close(self):
         self.closed = True
 
@@ -685,11 +697,11 @@ class DummyAdjustments(object):
     outbuf_overflow = 1048576
     inbuf_overflow = 512000
     cleanup_interval = 900
-    send_bytes = 9000
     url_scheme = 'http'
     channel_timeout = 300
     log_socket_errors = True
     recv_bytes = 8192
+    send_bytes = 1
     expose_tracebacks = True
     ident = 'waitress'
     max_request_header_size = 10000