Nodelay

aiohttp/parsers.py

@@ -59,6 +59,7 @@ def set_parser(self, parser):
 import asyncio
 import asyncio.streams
 import inspect
+import socket
 from . import errors
 from .streams import FlowControlDataQueue, EofStream
 
@@ -224,6 +225,22 @@ def __init__(self, transport, protocol, reader, loop):
         self._protocol = protocol
         self._reader = reader
         self._loop = loop
+        self._tcp_nodelay = False
+        self._socket = transport.get_extra_info('socket')
+
+    @property
+    def tcp_nodelay(self):
+        return self._tcp_nodelay
+
+    def set_tcp_nodelay(self, value):
+        if self._tcp_nodelay == value:
+            return
+        self._tcp_nodelay = value
+        if self._socket is None:
+            return
+        if self._socket.family not in (socket.AF_INET, socket.AF_INET6):
+            return
+        self._socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, value)
 
 
 class StreamProtocol(asyncio.streams.FlowControlMixin, asyncio.Protocol):

aiohttp/web_reqrep.py

@@ -421,6 +421,7 @@ def __init__(self, *, status=200, reason=None, headers=None):
         self._req = None
         self._resp_impl = None
         self._eof_sent = False
+        self._tcp_nodelay = True
 
         if headers is not None:
             self._headers.extend(headers)
@@ -604,6 +605,16 @@ def last_modified(self, value):
         elif isinstance(value, str):
             self.headers[hdrs.LAST_MODIFIED] = value
 
+    @property
+    def tcp_nodelay(self):
+        return self._tcp_nodelay
+
+    def set_tcp_nodelay(self, value):
+        self._tcp_nodelay = value
+        if self._resp_impl is None:
+            return
+        self._resp_impl.transport.set_tcp_nodelay(value)
+
     def _generate_content_type_header(self, CONTENT_TYPE=hdrs.CONTENT_TYPE):
         params = '; '.join("%s=%s" % i for i in self._content_dict.items())
         if params:
@@ -669,6 +680,7 @@ def _start(self, request):
             request.version,
             not keep_alive,
             self._reason)
+        resp_impl.transport.set_tcp_nodelay(self._tcp_nodelay)
 
         self._copy_cookies()
 

tests/test_stream_writer.py

@@ -0,0 +1,88 @@
+import pytest
+import socket
+from aiohttp.parsers import StreamWriter
+from unittest import mock
+
+
+def test_nodelay_default(loop):
+    transport = mock.Mock()
+    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    transport.get_extra_info.return_value = s
+    proto = mock.Mock()
+    reader = mock.Mock()
+    writer = StreamWriter(transport, proto, reader, loop)
+    assert not writer.tcp_nodelay
+    assert not s.getsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY)
+
+
+def test_set_nodelay_no_change(loop):
+    transport = mock.Mock()
+    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    transport.get_extra_info.return_value = s
+    proto = mock.Mock()
+    reader = mock.Mock()
+    writer = StreamWriter(transport, proto, reader, loop)
+    writer.set_tcp_nodelay(False)
+    assert not writer.tcp_nodelay
+    assert not s.getsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY)
+
+
+def test_set_nodelay_enable(loop):
+    transport = mock.Mock()
+    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    transport.get_extra_info.return_value = s
+    proto = mock.Mock()
+    reader = mock.Mock()
+    writer = StreamWriter(transport, proto, reader, loop)
+    writer.set_tcp_nodelay(True)
+    assert writer.tcp_nodelay
+    assert s.getsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY)
+
+
+def test_set_nodelay_enable_and_disable(loop):
+    transport = mock.Mock()
+    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    transport.get_extra_info.return_value = s
+    proto = mock.Mock()
+    reader = mock.Mock()
+    writer = StreamWriter(transport, proto, reader, loop)
+    writer.set_tcp_nodelay(True)
+    writer.set_tcp_nodelay(False)
+    assert not writer.tcp_nodelay
+    assert not s.getsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY)
+
+
+def test_set_nodelay_enable_ipv6(loop):
+    transport = mock.Mock()
+    s = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
+    transport.get_extra_info.return_value = s
+    proto = mock.Mock()
+    reader = mock.Mock()
+    writer = StreamWriter(transport, proto, reader, loop)
+    writer.set_tcp_nodelay(True)
+    assert writer.tcp_nodelay
+    assert s.getsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY)
+
+
+@pytest.mark.skipif(not hasattr(socket, 'AF_UNIX'),
+                    reason="requires unix sockets")
+def test_set_nodelay_enable_unix(loop):
+    transport = mock.Mock()
+    s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+    transport.get_extra_info.return_value = s
+    proto = mock.Mock()
+    reader = mock.Mock()
+    writer = StreamWriter(transport, proto, reader, loop)
+    writer.set_tcp_nodelay(True)
+    assert writer.tcp_nodelay
+
+
+def test_set_nodelay_enable_no_socket(loop):
+    transport = mock.Mock()
+    transport.get_extra_info.return_value = None
+    proto = mock.Mock()
+    reader = mock.Mock()
+    writer = StreamWriter(transport, proto, reader, loop)
+    writer.set_tcp_nodelay(True)
+    assert writer.tcp_nodelay
+    assert writer._socket is None

tests/test_web_response.py

@@ -591,6 +591,48 @@ def test_prepare_calls_signal():
     sig.assert_called_with(req, resp)
 
 
+def test_default_nodelay():
+    resp = StreamResponse()
+    assert resp.tcp_nodelay
+
+
+def test_set_tcp_nodelay_before_start():
+    resp = StreamResponse()
+    resp.set_tcp_nodelay(False)
+    assert not resp.tcp_nodelay
+    resp.set_tcp_nodelay(True)
+    assert resp.tcp_nodelay
+
+
+@pytest.mark.run_loop
+def test_set_tcp_nodelay_on_start():
+    req = make_request('GET', '/')
+    resp = StreamResponse()
+
+    with mock.patch('aiohttp.web_reqrep.ResponseImpl'):
+        resp_impl = yield from resp.prepare(req)
+    resp_impl.transport.set_tcp_nodelay.assert_called_with(True)
+
+
+@pytest.mark.run_loop
+def test_set_tcp_nodelay_after_start():
+    req = make_request('GET', '/')
+    resp = StreamResponse()
+
+    with mock.patch('aiohttp.web_reqrep.ResponseImpl'):
+        resp_impl = yield from resp.prepare(req)
+    resp_impl.transport.set_tcp_nodelay.assert_called_with(True)
+    resp.set_tcp_nodelay(False)
+    assert not resp.tcp_nodelay
+    resp_impl.transport.set_tcp_nodelay.assert_called_with(False)
+    resp.set_tcp_nodelay(True)
+    assert resp.tcp_nodelay
+    resp_impl.transport.set_tcp_nodelay.assert_called_with(True)
+
+
+# Response class
+
+
 def test_response_ctor():
     resp = Response()