diff --git a/h1/src/body.rs b/h1/src/body.rs index 25dee10..79c0f2c 100644 --- a/h1/src/body.rs +++ b/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), - End, + /// The end, with the consumer's offer when it answers a demand that + /// carried one. + End(Option), 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>>, demand: Sender>, + /// 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>>, 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) } } diff --git a/h1/src/client.rs b/h1/src/client.rs index d895212..236eb7f 100644 --- a/h1/src/client.rs +++ b/h1/src/client.rs @@ -250,6 +250,8 @@ pub struct Connection { timer: T, exchanges: Queue, config: ClientConfig, + /// Queued writes, and the pipe request bodies reuse. + outbox: Outbox, } impl fmt::Debug for Connection { @@ -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) }; diff --git a/h1/src/conn.rs b/h1/src/conn.rs index 94b0103..41f8240 100644 --- a/h1/src/conn.rs +++ b/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( 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, + #[cfg(all(feature = "pipe", any(target_os = "linux", target_os = "android")))] + pipe: Option, } impl Outbox { @@ -168,6 +168,10 @@ 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( writer: &mut W, outbox: &mut Outbox, @@ -175,12 +179,13 @@ pub(crate) async fn send_message( 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( 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( 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(writer: &W, outbox: &mut Outbox) -> Option { + 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(_writer: &W, _outbox: &mut Outbox) -> Option { + 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> { + 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( writer: &mut W, outbox: &mut Outbox, @@ -223,33 +262,32 @@ async fn send_other_frame( 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( writer: &mut W, outbox: &mut Outbox, encoder: &mut BodyEncoder, filled: http_kit_io::splice::FilledPipe, -) -> Result<(), Error> { +) -> Result { 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( @@ -65,7 +66,9 @@ pub(crate) async fn try_splice( 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)?; diff --git a/h1/tests/splice.rs b/h1/tests/splice.rs index 8f9ef1a..1c2b6d5 100644 --- a/h1/tests/splice.rs +++ b/h1/tests/splice.rs @@ -13,12 +13,19 @@ mod support; use std::{ cell::{Cell, RefCell}, + collections::{HashSet, VecDeque}, fs::File, future::{Future, poll_fn}, io::{self, Read as _, Write as _}, net::{SocketAddr, TcpListener, TcpStream}, + os::unix::fs::MetadataExt as _, pin::Pin, rc::Rc, + sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, + }, + task::{Context, Poll}, thread::{self, JoinHandle}, time::Duration, }; @@ -26,7 +33,7 @@ use std::{ use bytes::Bytes; use futures_lite::{StreamExt as _, future::zip, stream}; use http::{HeaderMap, HeaderValue, Method}; -use http_kit::{Body, BodyFrame, PipeOffer, Request, Response}; +use http_kit::{Body, BodyError, BodyFrame, FrameSource, PipeOffer, Request, Response}; use http_kit_h1::{ ClientConfig, ClientOutcome, ServerConfig, ServerConnection, ServerOutcome, Service, handshake, }; @@ -48,12 +55,34 @@ const PEER_TIMEOUT: Duration = Duration::from_secs(20); /// The most bytes one pipe frame may hold in these tests. const OFFER: usize = 64 * 1024; -/// Bytes that went through each path of a socket, per side. +/// Bytes that went through each path of a socket, per side, and the pipes +/// the socket was spliced into. #[derive(Debug, Default)] struct Counts { spliced_in: Cell, spliced_out: Cell, written: Cell, + /// How many splices into a pipe brought bytes. + fills: Cell, + /// The inode of every pipe a splice went into. + pipes: RefCell>, +} + +/// The inode of `pipe`, which tells pipes apart where descriptor numbers do +/// not: a closed pipe's descriptors are the first a new pipe gets. +fn pipe_inode(pipe: &Pipe) -> u64 { + // `Pipe` shows its descriptors only through `Debug`, as + // `Pipe { read: OwnedFd { fd: N }, .. }`. + let shown = format!("{pipe:?}"); + let fd: u32 = shown + .split("fd: ") + .nth(1) + .and_then(|rest| rest.split(|c: char| !c.is_ascii_digit()).next()) + .and_then(|digits| digits.parse().ok()) + .unwrap_or_else(|| panic!("no descriptor in {shown}")); + std::fs::metadata(format!("/proc/self/fd/{fd}")) + .expect("the pipe's metadata") + .ino() } type Spy = Rc; @@ -88,11 +117,13 @@ impl SpliceRead for SpyRead { pipe: Pipe, max: usize, ) -> Result { + self.counts.pipes.borrow_mut().insert(pipe_inode(&pipe)); let spliced = self.inner.splice_to_pipe(pipe, max).await?; if let Spliced::Filled(filled) = &spliced { self.counts .spliced_in .set(self.counts.spliced_in.get() + filled.len()); + self.counts.fills.set(self.counts.fills.get() + 1); } Ok(spliced) } @@ -193,8 +224,8 @@ fn read_until(stream: &mut TcpStream, needle: &[u8]) -> Vec { read } -/// A response with a `Content-Length`: its head and its body. -fn read_response(stream: &mut TcpStream) -> (String, Vec) { +/// A message with a `Content-Length`: its head and its body. +fn read_message(stream: &mut TcpStream) -> (String, Vec) { let head = String::from_utf8(read_until(stream, b"\r\n\r\n")).expect("a UTF-8 head"); let len = head .lines() @@ -243,7 +274,7 @@ async fn collect_offering(mut body: Body, max: usize) -> Seen { let mut seen = Seen::default(); let mut offer = PipeOffer::with_pipe(max, Pipe::new().expect("pipe")); while let Some(frame) = - poll_fn(|cx| Pin::new(&mut body).poll_frame_offering(cx, &mut offer)).await + poll_fn(|cx| Pin::new(&mut body).poll_frame_offering(cx, Some(&mut offer))).await { match frame.expect("body frame") { BodyFrame::Pipe(filled) => { @@ -355,7 +386,7 @@ async fn content_length_request(io: I) { serve_peer(io, recording_service(&seen, true), move |mut stream| { send_continued_head(&mut stream, &format!("Content-Length: {}", sent.len())); stream.write_all(&sent).expect("send the body"); - read_response(&mut stream) + read_message(&mut stream) }) .await; assert!(head.starts_with("HTTP/1.1 200"), "{head}"); @@ -389,7 +420,7 @@ async fn chunked_request_with_trailers(io: I) { stream .write_all(b"\r\n0\r\nx-checksum: abc123\r\n\r\n") .expect("send the end of the body"); - read_response(&mut stream) + read_message(&mut stream) }) .await; assert!(head.starts_with("HTTP/1.1 200"), "{head}"); @@ -418,8 +449,8 @@ async fn pipelined_request_stays_in_the_socket(io: I) stream .write_all(&both) .expect("send the body and a request"); - let first = read_response(&mut stream).1; - let second = read_response(&mut stream).1; + let first = read_message(&mut stream).1; + let second = read_message(&mut stream).1; (first, second) }) .await; @@ -442,8 +473,8 @@ async fn pipelined_requests_in_one_write_are_read_as_data(io: serve_peer(io, recording_service(&seen, false), move |mut stream| { send_continued_head(&mut stream, &format!("Content-Length: {}", sent.len())); stream.write_all(&sent).expect("send the body"); - read_response(&mut stream) + read_message(&mut stream) }) .await; assert_eq!(reply, b"/up"); @@ -487,10 +518,11 @@ async fn a_dropped_request_body_closes_the_connection async move { let mut body = request.into_body(); let mut offer = PipeOffer::with_pipe(OFFER, Pipe::new().expect("pipe")); - let frame = poll_fn(|cx| Pin::new(&mut body).poll_frame_offering(cx, &mut offer)) - .await - .expect("a frame") - .expect("body frame"); + let frame = + poll_fn(|cx| Pin::new(&mut body).poll_frame_offering(cx, Some(&mut offer))) + .await + .expect("a frame") + .expect("body frame"); let BodyFrame::Pipe(filled) = frame else { panic!("expected a pipe frame, got {frame:?}"); }; @@ -505,7 +537,7 @@ async fn a_dropped_request_body_closes_the_connection send_continued_head(&mut stream, &format!("Content-Length: {announced}")); stream.write_all(&sent).expect("send the first part"); // The peer holds the rest back, so the server has nothing to reset. - let reply = read_response(&mut stream).1; + let reply = read_message(&mut stream).1; (reply, read_to_end(&mut stream)) }) .await; @@ -699,7 +731,7 @@ async fn a_dropped_response_body_closes_the_connection( |response| async move { let mut body = response.into_body(); let mut offer = PipeOffer::with_pipe(OFFER, Pipe::new().expect("pipe")); - let frame = poll_fn(|cx| Pin::new(&mut body).poll_frame_offering(cx, &mut offer)) + let frame = poll_fn(|cx| Pin::new(&mut body).poll_frame_offering(cx, Some(&mut offer))) .await .expect("a frame") .expect("body frame"); @@ -835,7 +867,7 @@ async fn pipes_forward_from_request_to_response(io: I let mut offer = PipeOffer::with_pipe(OFFER, Pipe::new().expect("pipe")); let mut frames = Vec::new(); while let Some(frame) = - poll_fn(|cx| Pin::new(&mut body).poll_frame_offering(cx, &mut offer)).await + poll_fn(|cx| Pin::new(&mut body).poll_frame_offering(cx, Some(&mut offer))).await { frames.push(Ok::<_, io::Error>(frame.expect("body frame"))); } @@ -852,7 +884,7 @@ async fn pipes_forward_from_request_to_response(io: I let (counts, (head, reply)) = serve_peer(io, service, move |mut stream| { send_continued_head(&mut stream, &format!("Content-Length: {}", sent.len())); stream.write_all(&sent).expect("send the body"); - read_response(&mut stream) + read_message(&mut stream) }) .await; assert!(head.starts_with("HTTP/1.1 200"), "{head}"); @@ -866,6 +898,207 @@ async fn pipes_forward_from_request_to_response(io: I ); } +/// A body passed on as it is after a first frame its forwarder already +/// took, with the writer's offer passed through: what a proxy hands a +/// writer. +struct Forwarded { + first: Option, + rest: Body, +} + +impl FrameSource for Forwarded { + fn poll_next_frame( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + offer: Option<&mut PipeOffer>, + ) -> Poll>> { + let this = &mut *self; + if let Some(first) = this.first.take() { + return Poll::Ready(Some(Ok(first))); + } + Pin::new(&mut this.rest).poll_frame_offering(cx, offer) + } +} + +/// A request body sent back as the response body, for two requests on one +/// connection: the writer's offers reach the request body's connection, and +/// every frame the writer asks for goes through the one pipe the connection +/// keeps. +async fn a_forwarded_body_reuses_one_pipe_per_connection(io: I) { + let bodies = [payload(4000), payload(3000)]; + let sent = bodies.clone(); + let service = |request: Request| async move { + let length = request.headers()["content-length"].clone(); + let mut body = request.into_body(); + // Read before answering, so the connection knows the body was asked + // for and stays open. Nothing was offered, so it is data. + let first = body.next().await.expect("a frame").expect("body frame"); + assert!(first.as_data().is_some(), "{first:?}"); + let mut response = Response::new(Body::from_source(Forwarded { + first: Some(first), + rest: body, + })); + response.headers_mut().insert("content-length", length); + response + }; + let (counts, replies) = serve_peer(io, service, move |mut stream| { + sent.into_iter() + .map(|body| { + write!( + stream, + "POST /echo HTTP/1.1\r\nHost: localhost\r\nContent-Length: {}\r\n\r\n", + body.len() + ) + .expect("send the head"); + // The body arrives once the head was read, so it waits in the + // socket; it is sent while the echo is read, so neither side + // stalls on a full socket buffer. + thread::sleep(PAUSE); + let mut writer = stream.try_clone().expect("clone the stream"); + let sending = + thread::spawn(move || writer.write_all(&body).expect("send the body")); + let reply = read_message(&mut stream); + sending.join().expect("the sending thread"); + reply + }) + .collect::>() + }) + .await; + for ((head, reply), body) in replies.iter().zip(&bodies) { + assert!(head.starts_with("HTTP/1.1 200"), "{head}"); + assert!( + !head.to_ascii_lowercase().contains("connection: close"), + "{head}" + ); + assert!(reply == body, "the body came back intact"); + } + assert!( + counts.fills.get() > 2 * bodies.len(), + "{} splices, not enough to tell reuse", + counts.fills.get() + ); + assert_eq!( + counts.pipes.borrow().len(), + 1, + "one pipe for every pipe frame of both bodies" + ); + assert_eq!(counts.spliced_out.get(), counts.spliced_in.get()); +} + +/// A request body that answers an offer with its chunks in pipes, filled +/// from memory, and counts the pipes it makes: one for every offer that +/// carries none. +struct Filling { + chunks: VecDeque, + made: Arc, +} + +impl FrameSource for Filling { + fn poll_next_frame( + mut self: Pin<&mut Self>, + _cx: &mut Context<'_>, + offer: Option<&mut PipeOffer>, + ) -> Poll>> { + let Some(chunk) = self.chunks.pop_front() else { + return Poll::Ready(None); + }; + let Some(offer) = offer else { + return Poll::Ready(Some(Ok(BodyFrame::Data(chunk)))); + }; + let pipe = offer.take_pipe().unwrap_or_else(|| { + self.made.fetch_add(1, Ordering::Relaxed); + Pipe::new().expect("pipe") + }); + let filled = pipe + .fill(&chunk) + .unwrap_or_else(|(error, _)| panic!("fill: {error}")); + assert_eq!(filled.len(), chunk.len(), "a chunk fits one pipe"); + Poll::Ready(Some(Ok(BodyFrame::Pipe(filled)))) + } +} + +/// Two requests on one client connection, each body several pipe frames: the +/// client offers the pipe it emptied last, for the next request too. +async fn a_client_offers_its_emptied_pipe(io: I) { + const REQUESTS: usize = 2; + const CHUNKS: usize = 3; + let chunk = Bytes::from(payload(200)); + let listener = TcpListener::bind(loopback()).expect("bind"); + let addr = listener.local_addr().expect("local address"); + let server = thread::spawn(move || { + let (mut stream, _) = listener.accept().expect("accept"); + stream + .set_read_timeout(Some(PEER_TIMEOUT)) + .expect("read timeout"); + let bodies = (0..REQUESTS) + .map(|_| { + let (_, body) = read_message(&mut stream); + stream + .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n") + .expect("send the response"); + body + }) + .collect::>(); + read_to_end(&mut stream); + bodies + }); + let counts = Spy::default(); + let made = Arc::new(AtomicUsize::new(0)); + let stream = io.connect(addr).await.expect("connect"); + let (reader, writer) = stream.split(); + let (send, connection) = handshake( + SpyRead { + inner: reader, + counts: counts.clone(), + }, + SpyWrite { + inner: writer, + counts: counts.clone(), + }, + io.clone(), + ClientConfig::default(), + ); + let requests = { + let chunk = chunk.clone(); + let made = made.clone(); + async move { + for _ in 0..REQUESTS { + let mut request = Request::new(Body::from_source(Filling { + chunks: std::iter::repeat_n(chunk.clone(), CHUNKS).collect(), + made: made.clone(), + })); + *request.method_mut() = Method::POST; + *request.uri_mut() = format!("http://{addr}/up").parse().unwrap(); + request + .headers_mut() + .insert("content-length", HeaderValue::from(chunk.len() * CHUNKS)); + let response = send.send(request).await.expect("response"); + assert_eq!(response.status(), 200); + response + .into_body() + .into_bytes() + .await + .expect("response body"); + } + } + }; + let (ran, ()) = zip(connection.run(), requests).await; + assert!(matches!( + ran.expect("client connection"), + ClientOutcome::Closed + )); + let bodies = server.join().expect("the server thread"); + let expected = chunk.repeat(CHUNKS); + assert!(bodies.iter().all(|body| *body == expected), "bodies intact"); + assert_eq!( + made.load(Ordering::Relaxed), + 1, + "one pipe for all {} frames", + REQUESTS * CHUNKS + ); + assert_eq!(counts.spliced_out.get(), REQUESTS * CHUNKS * chunk.len()); +} + macro_rules! suite { ($io:expr, $run:expr) => { #[test] @@ -937,6 +1170,16 @@ macro_rules! suite { fn pipes_forward_from_request_to_response() { $run(super::pipes_forward_from_request_to_response($io)); } + + #[test] + fn a_forwarded_body_reuses_one_pipe_per_connection() { + $run(super::a_forwarded_body_reuses_one_pipe_per_connection($io)); + } + + #[test] + fn a_client_offers_its_emptied_pipe() { + $run(super::a_client_offers_its_emptied_pipe($io)); + } }; } diff --git a/src/body/mod.rs b/src/body/mod.rs index e91d5e6..3d6e457 100644 --- a/src/body/mod.rs +++ b/src/body/mod.rs @@ -1209,8 +1209,10 @@ impl GenericBody { .map(|frame| frame.map(|frame| frame.map(into_http_frame))) } - /// Polls the next frame, telling the source the consumer would take a - /// pipe frame of up to [`PipeOffer::max`] bytes. + /// Polls the next frame, telling the source, when `offer` is `Some`, that + /// the consumer would take a pipe frame of up to [`PipeOffer::max`] bytes; + /// with `None` it is [`poll_next`](Stream::poll_next), so a forwarder + /// passes its own consumer's offer, or its absence, straight through. /// /// The frame is a `BodyFrame::Pipe` only if the source holds bytes in /// the kernel and accepted the offer; otherwise it is whatever @@ -1223,9 +1225,9 @@ impl GenericBody { pub fn poll_frame_offering( self: Pin<&mut Self>, cx: &mut Context<'_>, - offer: &mut PipeOffer, + offer: Option<&mut PipeOffer>, ) -> Poll>> { - self.get_mut().poll_frames(cx, Some(offer)) + self.get_mut().poll_frames(cx, offer) } fn poll_frames( diff --git a/src/body/source.rs b/src/body/source.rs index 97a0407..8bf7f1c 100644 --- a/src/body/source.rs +++ b/src/body/source.rs @@ -24,7 +24,8 @@ use http_kit_io::splice::Pipe; /// /// The offer can carry an empty `Pipe`, so a consumer reuses one pipe for a /// whole body; a source that takes it with `PipeOffer::take_pipe` -/// fills it, and the consumer gets it back from the frame. Offers exist on +/// fills it, and the consumer gets it back from the frame. A wrapper between +/// a source and its consumer passes the consumer's offer on as it is. Offers exist on /// every platform, so code that passes them needs no `cfg`; only a platform /// with pipes has sources that accept them. #[derive(Debug)] @@ -209,7 +210,7 @@ mod tests { assert_eq!(plain.as_data().unwrap(), CJK_TEXT); let mut offer = PipeOffer::new(64 * 1024); - let offered = poll_fn(|cx| Pin::new(&mut body).poll_frame_offering(cx, &mut offer)) + let offered = poll_fn(|cx| Pin::new(&mut body).poll_frame_offering(cx, Some(&mut offer))) .await .unwrap() .unwrap(); @@ -241,7 +242,7 @@ mod tests { let mut frames = Vec::new(); while let Some(frame) = - poll_fn(|cx| Pin::new(&mut body).poll_frame_offering(cx, &mut offer)).await + poll_fn(|cx| Pin::new(&mut body).poll_frame_offering(cx, Some(&mut offer))).await { frames.push(frame.unwrap()); } @@ -260,7 +261,7 @@ mod tests { Body::from_bytes(CJK_TEXT), Body::from_stream(stream::iter(vec![Ok::<_, Error>(Bytes::from(CJK_TEXT))])), ] { - let frame = poll_fn(|cx| Pin::new(&mut body).poll_frame_offering(cx, &mut offer)) + let frame = poll_fn(|cx| Pin::new(&mut body).poll_frame_offering(cx, Some(&mut offer))) .await .unwrap() .unwrap(); @@ -429,7 +430,7 @@ mod pipe_tests { async fn an_offered_pipe_is_answered_with_a_pipe_frame() { let mut body = Body::from_source(Taking(Some(filled_pipe(CJK_TEXT).await))); let mut offer = PipeOffer::with_pipe(64 * 1024, Pipe::new().unwrap()); - let frame = poll_fn(|cx| Pin::new(&mut body).poll_frame_offering(cx, &mut offer)) + let frame = poll_fn(|cx| Pin::new(&mut body).poll_frame_offering(cx, Some(&mut offer))) .await .unwrap() .unwrap(); @@ -472,7 +473,7 @@ mod pipe_tests { Bytes::from(CJK_TEXT), ))])); let mut offer = PipeOffer::with_pipe(4096, Pipe::new().unwrap()); - let frame = poll_fn(|cx| Pin::new(&mut body).poll_frame_offering(cx, &mut offer)) + let frame = poll_fn(|cx| Pin::new(&mut body).poll_frame_offering(cx, Some(&mut offer))) .await .unwrap() .unwrap();