feat(http1): send response bodies from files with Write::poll_write_file

Cargo.toml

@@ -342,6 +342,11 @@ name = "server"
 path = "tests/server.rs"
 required-features = ["full"]
 
+[[test]]
+name = "h1_send_file"
+path = "tests/h1_send_file.rs"
+required-features = ["full"]
+
 [[test]]
 name = "h1_shutdown_while_buffered"
 path = "tests/h1_shutdown_while_buffered.rs"

src/common/io/rewind.rs

@@ -107,6 +107,20 @@ where
     fn is_write_vectored(&self) -> bool {
         self.inner.is_write_vectored()
     }
+
+    fn supports_write_file(&self) -> bool {
+        self.inner.supports_write_file()
+    }
+
+    fn poll_write_file(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        file: &std::fs::File,
+        offset: u64,
+        len: usize,
+    ) -> Poll<io::Result<usize>> {
+        Pin::new(&mut self.inner).poll_write_file(cx, file, offset, len)
+    }
 }
 
 #[cfg(all(

src/ext/mod.rs

@@ -27,6 +27,7 @@
 //! The extensions in this module can be grouped as follows:
 //!
 //! - **HTTP/1 Reason Phrase**: [`ReasonPhrase`] — Access non-canonical reason phrases in HTTP/1 responses.
+//! - **HTTP/1 File Bodies**: [`SendFile`] — Send a response body straight from a file, with zero-copy IO where the connection supports it.
 //! - **Informational Responses**: [`on_informational`] — Register callbacks for 1xx HTTP/1 responses on the client.
 //! - **Header Case Tracking**: Internal types for tracking the original casing and order of headers as received.
 //! - **HTTP/2 Protocol Extensions**: [`Protocol`] — Access the `:protocol` pseudo-header for Extended CONNECT in HTTP/2.
@@ -54,6 +55,11 @@ mod h1_reason_phrase;
 #[cfg(any(feature = "http1", feature = "ffi"))]
 pub use h1_reason_phrase::ReasonPhrase;
 
+#[cfg(all(feature = "http1", feature = "server"))]
+mod send_file;
+#[cfg(all(feature = "http1", feature = "server"))]
+pub use send_file::SendFile;
+
 #[cfg(all(feature = "http1", feature = "client"))]
 mod informational;
 #[cfg(all(feature = "http1", feature = "client"))]

src/ext/send_file.rs

@@ -0,0 +1,80 @@
+use std::fs::File;
+use std::sync::Arc;
+
+/// A range of a file to send as an HTTP/1 response body without copying it
+/// through the process.
+///
+/// Insert a `SendFile` into the extensions of an `http::Response`, and hyper's
+/// HTTP/1 server will write that range of the file to the connection with
+/// [`Write::poll_write_file`](crate::rt::Write::poll_write_file) instead of
+/// polling the response body. On a plain TCP socket this lets the IO use a
+/// zero-copy system call like Linux's `sendfile(2)`, which hands the file's
+/// pages from the kernel's page cache to the socket.
+///
+/// The response body is still required, and must hold the same bytes. It is
+/// the fallback, used whenever the file can't be sent directly:
+///
+/// - the connection's IO doesn't support it (its
+///   [`supports_write_file`](crate::rt::Write::supports_write_file) returns
+///   `false`, as it does for TLS streams and by default),
+/// - the connection is HTTP/2,
+/// - or the response isn't framed with a `Content-Length` equal to `len`:
+///   it is chunked, or has no body at all (a reply to `HEAD`, a `304`).
+///
+/// When the file is sent, the body is dropped without being polled, so a body
+/// that reads the file lazily costs nothing.
+///
+/// The head is flushed on its own before the file is handed to the IO, so a
+/// response sent this way costs at least one more write than a buffered one.
+/// For small bodies that can outweigh the copy it saves; it pays off for
+/// large ones.
+///
+/// The file must not be modified while it is being sent. hyper reads it by
+/// offset and never moves its cursor, so one `File` can be shared by any
+/// number of responses at once.
+///
+/// # Example
+///
+/// ```
+/// # fn doc(file: std::sync::Arc<std::fs::File>, bytes: bytes::Bytes) {
+/// use http_body_util::Full;
+/// use hyper::ext::SendFile;
+///
+/// let len = bytes.len() as u64;
+/// let mut response = http::Response::new(Full::new(bytes));
+/// response.extensions_mut().insert(SendFile::new(file, 0, len));
+/// # }
+/// ```
+#[derive(Clone, Debug)]
+pub struct SendFile {
+    pub(crate) file: Arc<File>,
+    pub(crate) offset: u64,
+    pub(crate) len: u64,
+}
+
+impl SendFile {
+    /// Send `len` bytes of `file`, starting at `offset`.
+    pub fn new(file: Arc<File>, offset: u64, len: u64) -> Self {
+        SendFile { file, offset, len }
+    }
+
+    /// The file to send from.
+    pub fn file(&self) -> &Arc<File> {
+        &self.file
+    }
+
+    /// Where in the file the body starts.
+    pub fn offset(&self) -> u64 {
+        self.offset
+    }
+
+    /// How many bytes of the file make up the body.
+    pub fn len(&self) -> u64 {
+        self.len
+    }
+
+    /// Whether the body is empty.
+    pub fn is_empty(&self) -> bool {
+        self.len == 0
+    }
+}

src/proto/h1/conn.rs

@@ -78,6 +78,8 @@ where
                 notify_read: false,
                 reading: Reading::Init,
                 writing: Writing::Init,
+                #[cfg(feature = "server")]
+                send_file: None,
                 upgrade: None,
                 // We assume a modern world where the remote speaks HTTP/1.1.
                 // If they tell us otherwise, we'll downgrade in `read_head`.
@@ -822,6 +824,81 @@ where
         }
     }
 
+    /// Take a response's `SendFile`, if the IO can write it.
+    #[cfg(feature = "server")]
+    pub(crate) fn take_send_file(
+        &mut self,
+        head: &mut MessageHead<T::Outgoing>,
+    ) -> Option<crate::ext::SendFile> {
+        let file = head.extensions.remove::<crate::ext::SendFile>()?;
+        if T::is_server() && self.io.supports_write_file() {
+            Some(file)
+        } else {
+            None
+        }
+    }
+
+    /// Send the body just started by `write_head` from `file`, if its framing
+    /// allows: a `Content-Length` that is exactly the file's range. Anything
+    /// else (chunked, or no body at all) leaves the body to be polled.
+    #[cfg(feature = "server")]
+    pub(crate) fn start_send_file(&mut self, file: crate::ext::SendFile) -> bool {
+        match &self.state.writing {
+            Writing::Body(encoder) if encoder.remaining_length() == Some(file.len) => {
+                self.state.send_file = Some(file);
+                true
+            }
+            _ => false,
+        }
+    }
+
+    #[cfg(feature = "server")]
+    pub(crate) fn is_sending_file(&self) -> bool {
+        self.state.send_file.is_some() && matches!(self.state.writing, Writing::Body(_))
+    }
+
+    /// Write the file started by `start_send_file` until it is done or the IO
+    /// would block.
+    #[cfg(feature = "server")]
+    pub(crate) fn poll_send_file(&mut self, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        loop {
+            let (file, encoder) = if let (Some(file), Writing::Body(encoder)) =
+                (&mut self.state.send_file, &mut self.state.writing)
+            {
+                (file, encoder)
+            } else {
+                self.state.send_file = None;
+                return Poll::Ready(Ok(()));
+            };
+
+            let remaining = encoder.remaining_length().unwrap_or(0);
+            let len = usize::try_from(remaining).unwrap_or(usize::MAX);
+            let n = ready!(self
+                .io
+                .poll_write_file(cx, &file.file, file.offset, len))?;
+            if n == 0 {
+                // Either the peer stopped accepting bytes, or the file is
+                // shorter than the Content-Length already sent for it.
+                self.state.send_file = None;
+                self.state.writing = Writing::Closed;
+                return Poll::Ready(Err(io::ErrorKind::WriteZero.into()));
+            }
+            trace!("sent {} bytes from file", n);
+            file.offset += n as u64;
+            encoder.wrote_length(n as u64);
+
+            if encoder.is_eof() {
+                self.state.writing = if encoder.is_last() {
+                    Writing::Closed
+                } else {
+                    Writing::KeepAlive
+                };
+                self.state.send_file = None;
+                return Poll::Ready(Ok(()));
+            }
+        }
+    }
+
     // When we get a parse error, depending on what side we are, we might be able
     // to write a response before closing the connection.
     //
@@ -973,6 +1050,10 @@ struct State {
     reading: Reading,
     /// State of allowed writes.
     writing: Writing,
+    /// The file a `Writing::Body` is being sent from, when the IO writes
+    /// it directly rather than the body being polled.
+    #[cfg(feature = "server")]
+    send_file: Option<crate::ext::SendFile>,
     /// An expected pending HTTP upgrade.
     upgrade: Option<crate::upgrade::Pending>,
     /// Either HTTP/1.0 or 1.1 connection.

src/proto/h1/dispatch.rs

@@ -362,7 +362,10 @@ where
                 && self.dispatch.should_poll()
             {
                 if let Some(msg) = ready!(Pin::new(&mut self.dispatch).poll_msg(cx)) {
-                    let (head, body) = msg.map_err(crate::Error::new_user_service)?;
+                    #[cfg_attr(not(feature = "server"), allow(unused_mut))]
+                    let (mut head, body) = msg.map_err(crate::Error::new_user_service)?;
+                    #[cfg(feature = "server")]
+                    let send_file = self.conn.take_send_file(&mut head);
 
                     let body_type = if body.is_end_stream() {
                         self.body_rx.set(None);
@@ -377,10 +380,22 @@ where
                         btype
                     };
                     self.conn.write_head(head, body_type);
+
+                    // The body stays the fallback until the head's framing is
+                    // known to match the file; once it does, it's never read.
+                    #[cfg(feature = "server")]
+                    if let Some(file) = send_file {
+                        if self.conn.start_send_file(file) {
+                            self.body_rx.set(None);
+                        }
+                    }
                 } else {
                     self.close();
                     return Poll::Ready(Ok(()));
                 }
+            } else if self.is_sending_file() {
+                #[cfg(feature = "server")]
+                ready!(self.conn.poll_send_file(cx)).map_err(crate::Error::new_body_write)?;
             } else if !self.conn.can_buffer_body() {
                 ready!(self.poll_flush(cx))?;
             } else {
@@ -466,6 +481,14 @@ where
         !self.is_closing && self.body_rx.is_some() && self.conn.can_write_body()
     }
 
+    /// Whether a body is being sent from a file, rather than from `body_rx`.
+    fn is_sending_file(&self) -> bool {
+        #[cfg(feature = "server")]
+        return self.conn.is_sending_file();
+        #[cfg(not(feature = "server"))]
+        return false;
+    }
+
     fn is_done(&self) -> bool {
         if self.is_closing {
             return true;
@@ -478,7 +501,9 @@ where
             true
         } else {
             let write_done = self.conn.is_write_closed()
-                || (!self.dispatch.should_poll() && self.body_rx.is_none());
+                || (!self.dispatch.should_poll()
+                    && self.body_rx.is_none()
+                    && !self.is_sending_file());
             read_done && write_done
         }
     }

src/proto/h1/encode.rs

@@ -113,6 +113,28 @@ impl Encoder {
         matches!(self.kind, Kind::Chunked(_))
     }
 
+    /// The bytes still owed to a body framed by `Content-Length`, or `None`
+    /// for any other framing.
+    #[cfg(feature = "server")]
+    pub(crate) fn remaining_length(&self) -> Option<u64> {
+        match self.kind {
+            Kind::Length(remaining) => Some(remaining),
+            _ => None,
+        }
+    }
+
+    /// Count `n` body bytes the IO wrote on its own, without `encode`.
+    #[cfg(feature = "server")]
+    pub(crate) fn wrote_length(&mut self, n: u64) {
+        match &mut self.kind {
+            Kind::Length(remaining) => {
+                debug_assert!(n <= *remaining, "wrote past Content-Length");
+                *remaining -= n;
+            }
+            _ => unreachable!("wrote_length on a body not framed by Content-Length"),
+        }
+    }
+
     pub(crate) fn end<B>(&self) -> Result<Option<EncodedBuf<B>>, NotEof> {
         match self.kind {
             Kind::Length(0) => Ok(None),

src/proto/h1/io.rs

@@ -327,6 +327,31 @@ where
         Pin::new(&mut self.io).poll_flush(cx)
     }
 
+    #[cfg(feature = "server")]
+    pub(crate) fn supports_write_file(&self) -> bool {
+        self.io.supports_write_file()
+    }
+
+    /// Write part of a file straight to the IO, after everything buffered.
+    #[cfg(feature = "server")]
+    pub(crate) fn poll_write_file(
+        &mut self,
+        cx: &mut Context<'_>,
+        file: &std::fs::File,
+        offset: u64,
+        len: usize,
+    ) -> Poll<io::Result<usize>> {
+        // The head goes out first, since the file's bytes follow it on the
+        // wire. That holds even when pipelined flushes are being deferred.
+        if self.write_buf.remaining() > 0 {
+            let flush_pipeline = std::mem::replace(&mut self.flush_pipeline, false);
+            let flushed = self.poll_flush(cx);
+            self.flush_pipeline = flush_pipeline;
+            ready!(flushed)?;
+        }
+        Pin::new(&mut self.io).poll_write_file(cx, file, offset, len)
+    }
+
     pub(crate) fn poll_shutdown(&mut self, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
         ready!(self.poll_flush(cx))?;
         Pin::new(&mut self.io).poll_shutdown(cx)