feat(io)!: multishot receive into the runtime's provided buffer ring

h1/src/client.rs

@@ -15,7 +15,7 @@ use async_channel::{Receiver, Sender};
 use bytes::BytesMut;
 use http::{Method, StatusCode};
 use http_kit::{Body, Deadlines, Request, Response};
-use http_kit_io::{Read, SendFile, SpliceRead, SpliceWrite, Split, Timer, Write};
+use http_kit_io::{ReadOwned, SendFile, SpliceRead, SpliceWrite, Split, Timer, Write};
 
 use crate::{
     ClientConfig, Error, ParseError, Timeout, Upgraded, body,
@@ -270,7 +270,7 @@ pub fn handshake<R, W, T>(
     config: ClientConfig,
 ) -> (SendRequest, Connection<R, W, T>)
 where
-    R: Read + SpliceRead,
+    R: ReadOwned + SpliceRead,
     W: Write + SendFile + SpliceWrite,
     T: Timer,
 {
@@ -296,7 +296,7 @@ pub fn handshake_stream<S, T>(
 ) -> (SendRequest, Connection<S::ReadHalf, S::WriteHalf, T>)
 where
     S: Split,
-    S::ReadHalf: SpliceRead,
+    S::ReadHalf: ReadOwned + SpliceRead,
     S::WriteHalf: SendFile + SpliceWrite,
     T: Timer,
 {
@@ -312,7 +312,7 @@ enum Next {
 
 impl<R, W, T> Connection<R, W, T>
 where
-    R: Read + SpliceRead,
+    R: ReadOwned + SpliceRead,
     W: Write + SendFile + SpliceWrite,
     T: Timer,
 {
@@ -499,7 +499,7 @@ enum Received {
 
 /// Reads one response head, every wait for bytes bounded by the same
 /// `deadline`, so a head that dribbles in cannot extend it.
-async fn read_head<R: Read, T: Timer>(
+async fn read_head<R: ReadOwned, T: Timer>(
     recv: &mut Recv<R>,
     limits: &crate::Limits,
     method: &Method,

h1/src/conn.rs

@@ -16,7 +16,8 @@ use futures_lite::{
 };
 use http_kit::{Body, BodyFrame, PipeOffer};
 use http_kit_io::{
-    BufResult, IntoInner as _, IoBuf as _, Read, SendFile, SpliceRead, SpliceWrite, Timer, Write,
+    BufResult, IntoInner as _, IoBuf as _, ReadOwned, Received, SendFile, SpliceRead, SpliceWrite,
+    Timer, Write,
 };
 
 use crate::{
@@ -39,7 +40,7 @@ pub(crate) struct Recv<R> {
     read_size: usize,
 }
 
-impl<R: Read> Recv<R> {
+impl<R: ReadOwned> Recv<R> {
     pub(crate) fn new(reader: R, read_size: usize) -> Self {
         Self {
             reader,
@@ -51,11 +52,26 @@ impl<R: Read> Recv<R> {
     /// Reads more bytes onto the end of the buffer, returning how many; `0`
     /// is the end of the stream.
     ///
-    /// The read lands in the buffer's spare capacity, and the bytes are
-    /// later split off that same allocation, so nothing received is copied.
-    /// If this future is dropped mid-read, the buffer goes with it; the
-    /// connection is unusable then and must close.
+    /// A stream that lends its own receive buffer has the bytes copied onto
+    /// the end of this one and gets its buffer back at once, so the
+    /// connection reserves room only for bytes that arrived. Otherwise the
+    /// read lands in the buffer's spare capacity, and the bytes are later
+    /// split off that same allocation, so nothing received is copied; if this
+    /// future is dropped mid-read then, the buffer goes with it, and the
+    /// connection is unusable and must close.
     pub(crate) async fn fill(&mut self) -> Result<usize, Error> {
+        loop {
+            match self.reader.read_owned().await {
+                Ok(Received::Lent(bytes)) => {
+                    self.buf.extend_from_slice(&bytes);
+                    return Ok(bytes.len());
+                }
+                Ok(Received::End) => return Ok(0),
+                Ok(Received::ReadInstead) => break,
+                Err(error) if error.kind() == io::ErrorKind::Interrupted => {}
+                Err(error) => return Err(error.into()),
+            }
+        }
         self.buf.reserve(self.read_size.max(1));
         let mut buf = mem::take(&mut self.buf);
         loop {
@@ -305,7 +321,7 @@ pub(crate) enum Pumped {
 /// A demand that carries a [`PipeOffer`] is answered with a pipe frame when
 /// the socket can splice, nothing is buffered and the decoder is in a raw
 /// run; every other frame is data or trailers, whatever was offered.
-pub(crate) async fn pump_body<R: Read + SpliceRead, T: Timer, F: Future<Output = ()>>(
+pub(crate) async fn pump_body<R: ReadOwned + SpliceRead, T: Timer, F: Future<Output = ()>>(
     recv: &mut Recv<R>,
     mut decoder: BodyDecoder,
     limits: &Limits,
@@ -349,7 +365,7 @@ pub(crate) async fn pump_body<R: Read + SpliceRead, T: Timer, F: Future<Output =
 /// Each time the decoder needs bytes it first offers the socket to the
 /// consumer's pipe: a chunk's data, for one, starts right after the chunk-size
 /// line that was just decoded, with nothing buffered.
-async fn next_frame<R: Read + SpliceRead, T: Timer>(
+async fn next_frame<R: ReadOwned + SpliceRead, T: Timer>(
     recv: &mut Recv<R>,
     decoder: &mut BodyDecoder,
     limits: &Limits,

h1/src/server.rs

@@ -23,7 +23,7 @@ use bytes::{Bytes, BytesMut};
 use futures_lite::future::{or, poll_once};
 use http::{HeaderValue, Method, StatusCode, Version, response};
 use http_kit::{Body, Request, Response};
-use http_kit_io::{Read, SendFile, SpliceRead, SpliceWrite, Split, Timer, Write};
+use http_kit_io::{ReadOwned, SendFile, SpliceRead, SpliceWrite, Split, Timer, Write};
 
 use crate::{
     Error, ParseError, ServerConfig, Timeout, Upgraded, body,
@@ -109,7 +109,7 @@ impl<R, W, T> std::fmt::Debug for ServerConnection<R, W, T> {
 
 impl<R, W, T> ServerConnection<R, W, T>
 where
-    R: Read + SpliceRead,
+    R: ReadOwned + SpliceRead,
     W: Write + SendFile + SpliceWrite,
     T: Timer,
 {
@@ -424,7 +424,7 @@ where
 
 impl<R, W, T> ServerConnection<R, W, T>
 where
-    R: Read + SpliceRead,
+    R: ReadOwned + SpliceRead,
     W: Write + SendFile + SpliceWrite,
     T: Timer,
 {

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.
+/// socket. 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>(
     recv: &mut Recv<R>,

h1/tests/splice.rs

@@ -31,8 +31,8 @@ use http_kit_h1::{
     ClientConfig, ClientOutcome, ServerConfig, ServerConnection, ServerOutcome, Service, handshake,
 };
 use http_kit_io::{
-    BufResult, Dial, IoBuf, IoBufMut, IoVectoredBuf, Listen, Listener as _, Read, SendFile,
-    SpliceRead, SpliceWrite, Split, Timer, Write,
+    BufResult, Dial, IoBuf, IoBufMut, IoVectoredBuf, Listen, Listener as _, Read, ReadOwned,
+    Received, SendFile, SpliceRead, SpliceWrite, Split, Timer, Write,
     splice::{FilledPipe, Pipe, Spliced},
 };
 
@@ -70,6 +70,14 @@ impl<R: Read> Read for SpyRead<R> {
     }
 }
 
+impl<R: ReadOwned> ReadOwned for SpyRead<R> {
+    type Buf = R::Buf;
+
+    fn read_owned(&mut self) -> impl Future<Output = io::Result<Received<R::Buf>>> {
+        self.inner.read_owned()
+    }
+}
+
 impl<R: SpliceRead> SpliceRead for SpyRead<R> {
     fn can_splice(&self) -> bool {
         self.inner.can_splice()

h1/tests/support/mod.rs

@@ -18,8 +18,8 @@ use std::{
 
 use async_channel::{Receiver, Sender};
 use http_kit_io::{
-    BufResult, IoBuf, IoBufMut, IoVectoredBuf, Read, SendFile, SpliceRead, SpliceWrite, Timer,
-    Write,
+    BufResult, IoBuf, IoBufMut, IoVectoredBuf, NoBuf, Read, ReadOwned, Received, SendFile,
+    SpliceRead, SpliceWrite, Timer, Write,
 };
 
 /// A timer whose deadlines never pass, for tests that run no timeouts.
@@ -193,6 +193,16 @@ impl Read for PipeReader {
     }
 }
 
+/// The reads land in the caller's buffer, which is what the reads it records
+/// are about.
+impl ReadOwned for PipeReader {
+    type Buf = NoBuf;
+
+    async fn read_owned(&mut self) -> io::Result<Received<NoBuf>> {
+        Ok(Received::ReadInstead)
+    }
+}
+
 impl Write for PipeWriter {
     async fn write<B: IoBuf>(&mut self, buf: B) -> BufResult<usize, B> {
         let result = self.push(buf.as_init().to_vec());