Support zerocopy send

docs/server-behavior.md

@@ -56,6 +56,12 @@ Applications should generally treat `HEAD` requests in the same manner as `GET`
 
 One exception to this might be if your application serves large file downloads, in which case you might wish to only generate the response headers.
 
+### Zero copy send
+
+Uvicorn supports [ASGI Zero Copy Send Extension](https://asgi.readthedocs.io/en/latest/extensions.html#zero-copy-send). Using the method specified by the extension, you can call the zero-copy interface of the operating system to send files.
+
+Note: This is currently only available on Python3.7+. And it is temporarily unavailable with uvloop because [uvloop#228](https://github.com/MagicStack/uvloop/issues/228).
+
 ---
 
 ## Timeouts

scripts/coverage

@@ -8,4 +8,4 @@ export SOURCE_FILES="uvicorn tests"
 
 set -x
 
-${PREFIX}coverage report --show-missing --skip-covered --fail-under=96.93
+${PREFIX}coverage report --show-missing --skip-covered --fail-under=96.69

tests/protocols/test_http.py

@@ -1,10 +1,14 @@
 import asyncio
 import contextlib
 import logging
+import os
+import sys
 
+import httpx
 import pytest
 
 from tests.response import Response
+from tests.utils import run_server
 from uvicorn.config import Config
 from uvicorn.main import ServerState
 from uvicorn.protocols.http.h11_impl import H11Protocol
@@ -123,6 +127,9 @@ def clear_buffer(self):
     def set_protocol(self, protocol):
         pass
 
+    def set_write_buffer_limits(self, high=None, low=None):
+        pass
+
 
 class MockLoop(asyncio.AbstractEventLoop):
     def __init__(self, event_loop):
@@ -744,3 +751,43 @@ def test_invalid_http_request(request_line, protocol_cls, caplog, event_loop):
         protocol.data_received(request)
         assert not protocol.transport.buffer
         assert "Invalid HTTP request received." in caplog.messages
+
+
+@pytest.mark.skipif(
+    sys.version_info[:2] < (3, 7),
+    not hasattr(os, "sendfile"),
+    reason="Sendfile only available in python3.7+",
+)
+@pytest.mark.parametrize("http", ["h11", "httptools"])
+@pytest.mark.parametrize("loop", ["asyncio"])
+@pytest.mark.asyncio
+async def test_sendfile(http, loop):
+    async def app(scope, receive, send):
+        with open("./README.md", "rb") as file:
+            content_length = len(file.read())
+            file.seek(0, 0)
+            await send(
+                {
+                    "type": "http.response.start",
+                    "status": 200,
+                    "headers": [
+                        (b"Content-Length", str(content_length).encode("ascii")),
+                        (b"Content-Type", b"text/plain; charset=utf8"),
+                    ],
+                }
+            )
+            await send(
+                {
+                    "type": "http.response.zerocopysend",
+                    "file": file.fileno(),
+                }
+            )
+
+    config = Config(app=app, http=http, loop=loop, limit_max_requests=1)
+    async with run_server(config):
+        with open("./README.md", "rb") as file:
+            file_content = file.read()
+
+        async with httpx.AsyncClient() as client:
+            response = await client.get("http://127.0.0.1:8000")
+            assert response.content == file_content

uvicorn/protocols/http/h11_impl.py

@@ -1,6 +1,7 @@
 import asyncio
 import http
 import logging
+import os
 from urllib.parse import unquote
 
 import h11
@@ -12,6 +13,7 @@
     FlowControl,
     service_unavailable,
 )
+from uvicorn.protocols.http.sendfile import can_sendfile
 from uvicorn.protocols.utils import (
     get_client_addr,
     get_local_addr,
@@ -76,6 +78,7 @@ def connection_made(self, transport):
         self.connections.add(self)
 
         self.transport = transport
+        self.allow_sendfile = can_sendfile(self.loop) and not is_ssl(self.transport)
         self.flow = FlowControl(transport)
         self.server = get_local_addr(transport)
         self.client = get_remote_addr(transport)
@@ -165,6 +168,9 @@ def handle_events(self):
                     "query_string": query_string,
                     "headers": self.headers,
                 }
+                if self.allow_sendfile:  # pragma: no cover
+                    extensions = self.scope.setdefault("extensions", {})
+                    extensions["http.response.zerocopysend"] = {}
 
                 for name, value in self.headers:
                     if name == b"connection":
@@ -185,6 +191,8 @@ def handle_events(self):
                     app = self.app
 
                 self.cycle = RequestResponseCycle(
+                    loop=self.loop,
+                    allow_sendfile=self.allow_sendfile,
                     scope=self.scope,
                     conn=self.conn,
                     transport=self.transport,
@@ -323,6 +331,8 @@ def timeout_keep_alive_handler(self):
 class RequestResponseCycle:
     def __init__(
         self,
+        loop,
+        allow_sendfile,
         scope,
         conn,
         transport,
@@ -334,6 +344,8 @@ def __init__(
         message_event,
         on_response,
     ):
+        self.loop = loop
+        self.allow_sendfile = allow_sendfile
         self.scope = scope
         self.conn = conn
         self.transport = transport
@@ -358,6 +370,11 @@ def __init__(
         self.response_started = False
         self.response_complete = False
 
+        # Sendfile
+        if self.allow_sendfile:  # pragma: no cover
+            # Set the buffer to 0 to avoid the problem of sending file before headers.
+            transport.set_write_buffer_limits(0)
+
     # ASGI exception wrapper
     async def run_asgi(self, app):
         try:
@@ -445,20 +462,55 @@ async def send(self, message):
 
         elif not self.response_complete:
             # Sending response body
-            if message_type != "http.response.body":
-                msg = "Expected ASGI message 'http.response.body', but got '%s'."
-                raise RuntimeError(msg % message_type)
-
-            body = message.get("body", b"")
-            more_body = message.get("more_body", False)
+            use_sendfile = False
+            if message_type == "http.response.body":
+                body = message.get("body", b"")
+                more_body = message.get("more_body", False)
+            elif (
+                self.allow_sendfile and message_type == "http.response.zerocopysend"
+            ):  # pragma: no cover
+                file_fd = message["file"]
+                sendfile_offset = message.get("offset", None)
+                if sendfile_offset is None:
+                    sendfile_offset = os.lseek(file_fd, 0, os.SEEK_CUR)
+                sendfile_count = message.get("count", None)
+                if sendfile_count is None:
+                    sendfile_count = os.stat(file_fd).st_size - sendfile_offset
+                more_body = message.get("more_body", False)
+                use_sendfile = True
+            else:
+                if self.allow_sendfile:  # pragma: no cover
+                    expect_message_types = (
+                        "http.response.body",
+                        "http.response.zerocopysend",
+                    )
+                else:
+                    expect_message_types = ("http.response.body",)
+                msg = "Expected ASGI message %s, but got '%s'."
+                raise RuntimeError(msg % expect_message_types, message_type)
 
             # Write response body
             if self.scope["method"] == "HEAD":
                 event = h11.Data(data=b"")
+                output = self.conn.send(event)
+                self.transport.write(output)
+            elif use_sendfile:  # pragma: no cover
+                with os.fdopen(os.dup(file_fd), "rb") as file:
+                    event = SendfileData(file, sendfile_offset, sendfile_count)
+                    for data in self.conn.send_with_data_passthrough(
+                        h11.Data(data=event)
+                    ):
+                        if isinstance(data, SendfileData):
+                            await self.flow.drain()
+                            await self.loop.sendfile(
+                                self.transport, data.file, data.offset, data.count
+                            )
+                        else:
+                            self.transport.write(data)
             else:
                 event = h11.Data(data=body)
-            output = self.conn.send(event)
-            self.transport.write(output)
+                output = self.conn.send(event)
+                self.transport.write(output)
 
             # Handle response completion
             if not more_body:
@@ -505,3 +557,13 @@ async def receive(self):
             self.body = b""
 
         return message
+
+
+class SendfileData:
+    def __init__(self, file, offset, count):
+        self.file = file
+        self.offset = offset
+        self.count = count
+
+    def __len__(self):
+        return self.count

uvicorn/protocols/http/httptools_impl.py

@@ -1,6 +1,7 @@
 import asyncio
 import http
 import logging
+import os
 import re
 import urllib
 from collections import deque
@@ -14,6 +15,7 @@
     FlowControl,
     service_unavailable,
 )
+from uvicorn.protocols.http.sendfile import can_sendfile
 from uvicorn.protocols.utils import (
     get_client_addr,
     get_local_addr,
@@ -85,6 +87,7 @@ def connection_made(self, transport):
         self.connections.add(self)
 
         self.transport = transport
+        self.allow_sendfile = can_sendfile(self.loop) and not is_ssl(self.transport)
         self.flow = FlowControl(transport)
         self.server = get_local_addr(transport)
         self.client = get_remote_addr(transport)
@@ -204,6 +207,9 @@ def on_url(self, url):
             "query_string": parsed_url.query if parsed_url.query else b"",
             "headers": self.headers,
         }
+        if self.allow_sendfile:  # pragma: no cover
+            extensions = self.scope.setdefault("extensions", {})
+            extensions["http.response.zerocopysend"] = {}
 
     def on_header(self, name: bytes, value: bytes):
         name = name.lower()
@@ -231,6 +237,8 @@ def on_headers_complete(self):
 
         existing_cycle = self.cycle
         self.cycle = RequestResponseCycle(
+            loop=self.loop,
+            allow_sendfile=self.allow_sendfile,
             scope=self.scope,
             transport=self.transport,
             flow=self.flow,
@@ -324,6 +332,8 @@ def timeout_keep_alive_handler(self):
 class RequestResponseCycle:
     def __init__(
         self,
+        loop,
+        allow_sendfile,
         scope,
         transport,
         flow,
@@ -336,6 +346,8 @@ def __init__(
         keep_alive,
         on_response,
     ):
+        self.loop = loop
+        self.allow_sendfile = allow_sendfile
         self.scope = scope
         self.transport = transport
         self.flow = flow
@@ -361,6 +373,11 @@ def __init__(
         self.chunked_encoding = None
         self.expected_content_length = 0
 
+        # Sendfile
+        if self.allow_sendfile:  # pragma: no cover
+            # Set the buffer to 0 to avoid the problem of sending file before headers.
+            transport.set_write_buffer_limits(0)
+
     # ASGI exception wrapper
     async def run_asgi(self, app):
         try:
@@ -472,31 +489,72 @@ async def send(self, message):
 
         elif not self.response_complete:
             # Sending response body
-            if message_type != "http.response.body":
-                msg = "Expected ASGI message 'http.response.body', but got '%s'."
-                raise RuntimeError(msg % message_type)
-
-            body = message.get("body", b"")
-            more_body = message.get("more_body", False)
+            use_sendfile = False
+            if message_type == "http.response.body":
+                body = message.get("body", b"")
+                more_body = message.get("more_body", False)
+            elif (
+                self.allow_sendfile and message_type == "http.response.zerocopysend"
+            ):  # pragma: no cover
+                file_fd = message["file"]
+                sendfile_offset = message.get("offset", None)
+                if sendfile_offset is None:
+                    sendfile_offset = os.lseek(file_fd, 0, os.SEEK_CUR)
+                sendfile_count = message.get("count", None)
+                if sendfile_count is None:
+                    sendfile_count = os.stat(file_fd).st_size - sendfile_offset
+                more_body = message.get("more_body", False)
+                use_sendfile = True
+            else:
+                if self.allow_sendfile:  # pragma: no cover
+                    expect_message_types = (
+                        "http.response.body",
+                        "http.response.zerocopysend",
+                    )
+                else:
+                    expect_message_types = ("http.response.body",)
+                msg = "Expected ASGI message %s, but got '%s'."
+                raise RuntimeError(msg % expect_message_types, message_type)
 
             # Write response body
             if self.scope["method"] == "HEAD":
                 self.expected_content_length = 0
             elif self.chunked_encoding:
-                if body:
-                    content = [b"%x\r\n" % len(body), body, b"\r\n"]
-                else:
-                    content = []
-                if not more_body:
-                    content.append(b"0\r\n\r\n")
-                self.transport.write(b"".join(content))
+                if not use_sendfile:
+                    if body:
+                        content = [b"%x\r\n" % len(body), body, b"\r\n"]
+                    else:
+                        content = []
+                    if not more_body:
+                        content.append(b"0\r\n\r\n")
+                    self.transport.write(b"".join(content))
+                else:  # pragma: no cover
+                    self.transport.write(b"%x\r\n" % sendfile_count)
+                    await self.flow.drain()
+                    with os.fdopen(os.dup(file_fd), "rb") as file:
+                        await self.loop.sendfile(
+                            self.transport, file, sendfile_offset, sendfile_count
+                        )
+                    if more_body:
+                        self.transport.write(b"\r\n")
+                    else:
+                        self.transport.write(b"\r\n0\r\n\r\n")
             else:
-                num_bytes = len(body)
+                if not use_sendfile:
+                    num_bytes = len(body)
+                    self.transport.write(body)
+                else:  # pragma: no cover
+                    num_bytes = sendfile_count
+                    await self.flow.drain()
+                    with os.fdopen(os.dup(file_fd), "rb") as file:
+                        await self.loop.sendfile(
+                            self.transport, file, sendfile_offset, sendfile_count
+                        )
+
                 if num_bytes > self.expected_content_length:
                     raise RuntimeError("Response content longer than Content-Length")
                 else:
                     self.expected_content_length -= num_bytes
-                self.transport.write(body)
 
             # Handle response completion
             if not more_body: