perf(h1): reuse one pipe per connection for spliced bodies

h1/src/body.rs

@@ -15,7 +15,7 @@ use std::{
     task::{Context, Poll, ready},
 };
 
-use async_channel::{Receiver, Sender, TrySendError};
+use async_channel::{Receiver, Sender};
 use futures_lite::Stream;
 use http_kit::{BodyError, BodyFrame, FrameSource, PipeOffer};
 
@@ -25,7 +25,9 @@ pub(crate) enum Message {
     /// A frame, with the consumer's offer when the driver did not use its
     /// pipe.
     Frame(BodyFrame, Option<PipeOffer>),
-    End,
+    /// The end, with the consumer's offer when it answers a demand that
+    /// carried one.
+    End(Option<PipeOffer>),
     Error(Error),
 }
 
@@ -37,11 +39,12 @@ pub(crate) fn channel() -> (BodySender, IncomingBody) {
     (
         BodySender {
             frames: frames_tx,
-            demand: demand_rx,
+            demand: demand_rx.clone(),
         },
         IncomingBody {
             frames: Box::pin(frames_rx),
             demand: demand_tx,
+            unanswered: Box::new(demand_rx),
             waiting: false,
             done: false,
         },
@@ -72,6 +75,11 @@ pub(crate) struct IncomingBody {
     // Pinned: the receiver's stream keeps a listener that must not move.
     frames: Pin<Box<Receiver<Message>>>,
     demand: Sender<Option<PipeOffer>>,
+    /// The demand channel's other end, to take back a demand the driver never
+    /// received: the end of a body is queued behind its last frame, so the
+    /// demand that finds it went unread, offer and pipe with it. Boxed, since
+    /// a receiver is not `Unpin`; only `try_recv` is called on it.
+    unanswered: Box<Receiver<Option<PipeOffer>>>,
     waiting: bool,
     done: bool,
 }
@@ -88,40 +96,50 @@ impl FrameSource for IncomingBody {
         }
         if !this.waiting {
             let demand = offer.as_deref_mut().map(PipeOffer::detach);
-            match this.demand.try_send(demand) {
-                // A closed demand channel means the driver is gone; the frame
-                // channel reports how it ended.
-                Ok(()) | Err(TrySendError::Closed(_)) => this.waiting = true,
-                Err(TrySendError::Full(_)) => {
-                    unreachable!("one demand is outstanding at a time")
-                }
+            // The channel stays open while this end holds `unanswered`, also
+            // once the driver is gone; the frame channel reports how it ended.
+            if this.demand.try_send(demand).is_err() {
+                unreachable!("one demand is outstanding at a time");
             }
+            this.waiting = true;
         }
         let message = ready!(this.frames.as_mut().poll_next(cx));
         this.waiting = false;
-        Poll::Ready(match message {
+        let (polled, detached) = match message {
             Some(Message::Frame(frame, detached)) => {
-                let offered = offer.is_some();
-                if let (Some(offer), Some(detached)) = (offer, detached) {
-                    offer.reattach(detached);
-                }
-                Some(if offered { Ok(frame) } else { read_out(frame) })
+                let frame = if offer.is_some() {
+                    Ok(frame)
+                } else {
+                    read_out(frame)
+                };
+                (Some(frame), detached)
             }
-            Some(Message::End) => {
+            Some(Message::End(detached)) => {
                 this.done = true;
-                None
+                (None, detached)
             }
             Some(Message::Error(error)) => {
                 this.done = true;
-                Some(Err(error.into()))
+                (Some(Err(error.into())), None)
             }
             None => {
                 // The driver stopped without ending the body: never pass a
                 // truncated body off as a complete one.
                 this.done = true;
-                Some(Err(Error::Parse(ParseError::Incomplete).into()))
+                (Some(Err(Error::Parse(ParseError::Incomplete).into())), None)
             }
-        })
+        };
+        if let Some(offer) = offer {
+            if let Some(detached) = detached {
+                offer.reattach(detached);
+            }
+            if this.done
+                && let Ok(Some(detached)) = this.unanswered.try_recv()
+            {
+                offer.reattach(detached);
+            }
+        }
+        Poll::Ready(polled)
     }
 }
 

h1/src/client.rs

@@ -250,6 +250,8 @@ pub struct Connection<R, W, T> {
     timer: T,
     exchanges: Queue,
     config: ClientConfig,
+    /// Queued writes, and the pipe request bodies reuse.
+    outbox: Outbox,
 }
 
 impl<R, W, T> fmt::Debug for Connection<R, W, T> {
@@ -283,6 +285,7 @@ where
             timer,
             exchanges: Queue(exchanges),
             config,
+            outbox: Outbox::default(),
         },
     )
 }
@@ -388,9 +391,9 @@ where
         let delivered = Cell::new(false);
 
         let writer = &mut self.writer;
+        let outbox = &mut self.outbox;
         let send = async {
             let _written = written_tx;
-            let mut outbox = Outbox::default();
             outbox.push(head.freeze());
             if expect_continue {
                 outbox.flush(writer).await?;
@@ -403,7 +406,7 @@ where
                     None => tracing::debug!("no 100 Continue in time, sending the body"),
                 }
             }
-            send_message(writer, &mut outbox, Some(body), &mut plan.encoder).await?;
+            send_message(writer, outbox, Some(body), &mut plan.encoder).await?;
             Ok::<_, Error>(true)
         };
 

h1/src/conn.rs

@@ -4,17 +4,14 @@
 use std::{
     future::{Future, pending, poll_fn},
     io, mem,
-    pin::pin,
+    pin::{Pin, pin},
     task::Poll,
     time::{Duration, Instant},
 };
 
 use bytes::{Bytes, BytesMut};
-use futures_lite::{
-    StreamExt as _,
-    future::{or, poll_once},
-};
-use http_kit::{Body, BodyFrame, PipeOffer};
+use futures_lite::future::{or, poll_once};
+use http_kit::{Body, BodyError, BodyFrame, PipeOffer};
 use http_kit_io::{
     BufResult, IntoInner as _, IoBuf as _, ReadOwned, Received, SendFile, SpliceRead, SpliceWrite,
     Timer, Write,
@@ -117,10 +114,13 @@ pub(crate) async fn until<T: Timer, F: Future>(
     or(done, expired).await
 }
 
-/// Buffers queued for one vectored write.
+/// Buffers queued for one vectored write, and the empty pipe the last pipe
+/// frame sent left behind, which the next body sent is offered.
 #[derive(Default)]
 pub(crate) struct Outbox {
     bufs: Vec<Bytes>,
+    #[cfg(all(feature = "pipe", any(target_os = "linux", target_os = "android")))]
+    pipe: Option<http_kit_io::splice::Pipe>,
 }
 
 impl Outbox {
@@ -168,19 +168,24 @@ impl Outbox {
 /// frame ready, what is queued goes out first, so a slow body never holds
 /// back its head. A file region goes to [`SendFile`] untouched, and a pipe
 /// to [`SpliceWrite`], each framed on the writer side.
+///
+/// A writer that can splice polls the body with a [`PipeOffer`] carrying the
+/// pipe the last pipe frame left, so a body forwarded from a socket comes as
+/// pipe frames that all reuse the one pipe the connection keeps.
 pub(crate) async fn send_message<W: Write + SendFile + SpliceWrite>(
     writer: &mut W,
     outbox: &mut Outbox,
     body: Option<Body>,
     encoder: &mut BodyEncoder,
 ) -> Result<(), Error> {
     if let Some(mut body) = body.filter(|_| encoder.sends_body()) {
+        let mut offer = writer_offer(writer, outbox);
         loop {
-            let frame = match poll_once(body.next()).await {
+            let frame = match poll_once(next_sent_frame(&mut body, offer.as_mut())).await {
                 Some(frame) => frame,
                 None => {
                     outbox.flush(writer).await?;
-                    body.next().await
+                    next_sent_frame(&mut body, offer.as_mut()).await
                 }
             };
             match frame {
@@ -190,12 +195,26 @@ pub(crate) async fn send_message<W: Write + SendFile + SpliceWrite>(
                 Some(Ok(BodyFrame::Trailers(trailers))) => {
                     outbox.push(encoder.trailers(&trailers)?);
                 }
+                #[cfg(all(feature = "pipe", any(target_os = "linux", target_os = "android")))]
+                Some(Ok(BodyFrame::Pipe(filled))) => {
+                    let pipe = send_pipe(writer, outbox, encoder, filled).await?;
+                    // The emptied pipe is offered for the next frame, unless
+                    // the offer still holds the one it had: the frame was a
+                    // pipe its source made without taking the offered one.
+                    if let Some(offer) = &mut offer {
+                        offer.reattach(PipeOffer::with_pipe(offer.max(), pipe));
+                    }
+                }
                 Some(Ok(other)) => send_other_frame(writer, outbox, encoder, other).await?,
             }
             if outbox.is_full() {
                 outbox.flush(writer).await?;
             }
         }
+        #[cfg(all(feature = "pipe", any(target_os = "linux", target_os = "android")))]
+        {
+            outbox.pipe = offer.and_then(|mut offer| offer.take_pipe());
+        }
     }
     if let Some(end) = encoder.finish()? {
         outbox.push(end);
@@ -205,15 +224,35 @@ pub(crate) async fn send_message<W: Write + SendFile + SpliceWrite>(
     Ok(())
 }
 
-/// Sends a frame that is neither data nor trailers: a file region, or a pipe,
-/// each with the features it needs.
-#[cfg_attr(
-    not(any(
-        feature = "fs",
-        all(feature = "pipe", any(target_os = "linux", target_os = "android"))
-    )),
-    allow(unused_variables)
-)]
+/// What a writer offers the body it sends: pipe frames when its socket can
+/// splice, of any size, since [`send_pipe`] writes a whole pipe however full,
+/// carrying the pipe the connection kept from its last pipe frame.
+#[cfg(all(feature = "pipe", any(target_os = "linux", target_os = "android")))]
+fn writer_offer<W: SpliceWrite>(writer: &W, outbox: &mut Outbox) -> Option<PipeOffer> {
+    SpliceWrite::can_splice(writer).then(|| match outbox.pipe.take() {
+        Some(pipe) => PipeOffer::with_pipe(usize::MAX, pipe),
+        None => PipeOffer::new(usize::MAX),
+    })
+}
+
+/// Where pipes do not exist, or the feature is off, a writer offers nothing.
+#[cfg(not(all(feature = "pipe", any(target_os = "linux", target_os = "android"))))]
+const fn writer_offer<W>(_writer: &W, _outbox: &mut Outbox) -> Option<PipeOffer> {
+    None
+}
+
+/// The next frame of a body being sent, polled with `offer` when the writer
+/// made one.
+async fn next_sent_frame(
+    body: &mut Body,
+    mut offer: Option<&mut PipeOffer>,
+) -> Option<Result<BodyFrame, BodyError>> {
+    poll_fn(|cx| Pin::new(&mut *body).poll_frame_offering(cx, offer.as_deref_mut())).await
+}
+
+/// Sends a frame that is neither data, trailers nor a pipe: a file region,
+/// with the feature it needs.
+#[cfg_attr(not(feature = "fs"), allow(unused_variables))]
 async fn send_other_frame<W: Write + SendFile + SpliceWrite>(
     writer: &mut W,
     outbox: &mut Outbox,
@@ -223,33 +262,32 @@ async fn send_other_frame<W: Write + SendFile + SpliceWrite>(
     match frame {
         #[cfg(feature = "fs")]
         BodyFrame::File(region) => send_region(writer, outbox, encoder, region).await,
-        #[cfg(all(feature = "pipe", any(target_os = "linux", target_os = "android")))]
-        BodyFrame::Pipe(filled) => send_pipe(writer, outbox, encoder, filled).await,
         _ => Err(Error::UnsupportedFrame),
     }
 }
 
 /// Sends the bytes of a pipe: spliced into the socket when it can take them,
 /// framed like a file region, and otherwise read out and sent as data.
+/// Returns the emptied pipe.
 #[cfg(all(feature = "pipe", any(target_os = "linux", target_os = "android")))]
 async fn send_pipe<W: Write + SendFile + SpliceWrite>(
     writer: &mut W,
     outbox: &mut Outbox,
     encoder: &mut BodyEncoder,
     filled: http_kit_io::splice::FilledPipe,
-) -> Result<(), Error> {
+) -> Result<http_kit_io::splice::Pipe, Error> {
     if !SpliceWrite::can_splice(writer) {
-        let (data, _pipe) = filled.read_out()?;
+        let (data, pipe) = filled.read_out()?;
         encoder.data(data, &mut outbox.bufs)?;
-        return Ok(());
+        return Ok(pipe);
     }
     let len = filled.len() as u64;
     outbox.extend(encoder.region_start(len)?);
     outbox.flush(writer).await?;
     tracing::trace!(len, "sending a pipe");
-    writer.splice_from_pipe(filled).await?;
+    let pipe = writer.splice_from_pipe(filled).await?;
     outbox.extend(encoder.region_end());
-    Ok(())
+    Ok(pipe)
 }
 
 #[cfg(feature = "fs")]
@@ -324,13 +362,14 @@ pub(crate) async fn pump_body<R: ReadOwned + SpliceRead, T: Timer, F: Future<Out
         };
         // A send to a dropped body fails; the next demand then reports it.
         if let Some(frame) = frame {
-            sender.send(Message::Frame(frame, offer)).await;
+            sender.send(Message::Frame(frame, offer.take())).await;
         }
         // The end is queued behind the last frame without waiting for a
         // demand: the connection is past the body as soon as it is decoded,
-        // however slowly the consumer reads.
+        // however slowly the consumer reads. It carries the offer of a demand
+        // the end itself answered.
         if decoder.is_done() {
-            sender.send(Message::End).await;
+            sender.send(Message::End(offer)).await;
             return Ok(Pumped::Complete);
         }
     }

h1/src/lib.rs

@@ -28,7 +28,11 @@
 //! with nothing buffered ahead of them; framing, trailers and everything else
 //! stay data and trailers frames, and a consumer that offers nothing receives
 //! only those. A pipe frame in a body being sent is spliced into the socket
-//! like a file region, with the chunk framing written around it.
+//! like a file region, with the chunk framing written around it. A
+//! connection whose socket can splice polls each body it sends with an
+//! offer that carries the pipe it emptied last, so a body forwarded from
+//! another connection goes through one pipe for the whole connection, and
+//! a forwarder between the two need not offer anything itself.
 //!
 //! # Request smuggling
 //!

h1/src/splice.rs

@@ -24,7 +24,8 @@ pub(crate) enum Attempt {
 ///
 /// The splice takes at most the run and the offer's `max`, so it never
 /// reaches past the body into the next message's bytes, which stay in the
-/// socket. A stream that received ahead of its reader puts those bytes into
+/// socket. It fills the offer's pipe, and makes one only when the offer
+/// supplies none. A stream that received ahead of its reader puts those bytes into
 /// the pipe before it splices from the socket.
 #[cfg(all(feature = "pipe", any(target_os = "linux", target_os = "android")))]
 pub(crate) async fn try_splice<R: SpliceRead, T: Timer>(
@@ -65,7 +66,9 @@ pub(crate) async fn try_splice<R: SpliceRead, T: Timer>(
             decoder.consume_raw(filled.len() as u64);
             Ok(Attempt::Frame(http_kit::BodyFrame::Pipe(filled)))
         }
-        Ok(Spliced::End(_)) => {
+        Ok(Spliced::End(pipe)) => {
+            // The pipe is still empty: it goes back to the consumer.
+            *offer = PipeOffer::with_pipe(offer.max(), pipe);
             // Nothing is buffered, so this ends a close-delimited body and
             // fails any other that is incomplete.
             decoder.decode_eof(&mut recv.buf, limits)?;