From ca9c4532b0167bb216b16092d25fac1e0cb8644b Mon Sep 17 00:00:00 2001 From: ljluestc <63439129+ljluestc@users.noreply.github.com> Date: Sat, 26 Sep 2026 09:19:28 -0700 Subject: [PATCH] perf(http1): pick write strategy per buffer with a size heuristic The HTTP/1 write strategy was chosen once per connection, purely from `AsyncWrite::is_write_vectored()`: vectored IO always queued, everything else always flattened. Vectored writes avoid copying the body, but each extra `IoSlice` costs bookkeeping in `BufList`, in `chunks_vectored`, and in the kernel's gather loop, so for small bodies queueing is a net loss against a single `memcpy` plus one `write`. Add a `WriteStrategy::Auto` variant, now the default when the IO supports vectored writes, that decides per buffer: flatten anything below `AUTO_FLATTEN_LIMIT`, queue the rest. Once something is queued, later buffers must queue too, or they would be written ahead of the bytes that came first. `AUTO_FLATTEN_LIMIT` is 16KB, from `benches/h1_writev.rs` (added here), which serves a fixed-size response body over loopback TCP with each strategy forced on. On x86_64 Linux the crossover sits around 20-24KB; 16KB is the nearest power of two below it. No public API change. `http1::Builder::writev` still overrides the default in either direction. Closes #2910 --- Cargo.toml | 5 ++ benches/h1_writev.rs | 118 +++++++++++++++++++++++++++ src/proto/h1/io.rs | 190 +++++++++++++++++++++++++++++++++++-------- 3 files changed, 280 insertions(+), 33 deletions(-) create mode 100644 benches/h1_writev.rs diff --git a/Cargo.toml b/Cargo.toml index 2c98449797..f76b975b0b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -312,6 +312,11 @@ name = "end_to_end" path = "benches/end_to_end.rs" required-features = ["full"] +[[bench]] +name = "h1_writev" +path = "benches/h1_writev.rs" +required-features = ["full"] + [[bench]] name = "pipeline" path = "benches/pipeline.rs" diff --git a/benches/h1_writev.rs b/benches/h1_writev.rs new file mode 100644 index 0000000000..f580e36687 --- /dev/null +++ b/benches/h1_writev.rs @@ -0,0 +1,118 @@ +#![feature(test)] +#![deny(warnings)] + +//! Compares the HTTP/1 write strategies at a range of response body sizes. +//! `flatten` and `queue` force one strategy on via `http1::Builder::writev`; +//! `auto` leaves the default in place. Used to pick `AUTO_FLATTEN_LIMIT` in +//! `src/proto/h1/io.rs`, and to check that `auto` tracks whichever of the two +//! is faster at each size. + +extern crate test; +mod support; + +use std::convert::Infallible; +use std::net::SocketAddr; + +use http_body_util::{BodyExt, Full}; +use hyper::service::service_fn; +use hyper::{Request, Response}; + +fn spawn_server( + rt: &tokio::runtime::Runtime, + writev: Option, + body: &'static [u8], +) -> SocketAddr { + use tokio::net::TcpListener; + let addr = "127.0.0.1:0".parse::().unwrap(); + let listener = rt.block_on(async { TcpListener::bind(&addr).await.unwrap() }); + let addr = listener.local_addr().unwrap(); + rt.spawn(async move { + while let Ok((sock, _)) = listener.accept().await { + let io = support::TokioIo::new(sock); + let mut builder = hyper::server::conn::http1::Builder::new(); + if let Some(w) = writev { + builder.writev(w); + } + tokio::spawn(builder.serve_connection( + io, + service_fn(move |req: Request| async move { + let mut req_body = req.into_body(); + while let Some(_chunk) = req_body.frame().await {} + Ok::<_, Infallible>(Response::new(Full::::from(body))) + }), + )); + } + }); + addr +} + +fn bench(b: &mut test::Bencher, writev: Option, body: &'static [u8]) { + let rt = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("rt build"); + + b.bytes = body.len() as u64; + + let addr = spawn_server(&rt, writev, body); + + let mut client = rt.block_on(async { + let tcp = tokio::net::TcpStream::connect(&addr).await.unwrap(); + let io = support::TokioIo::new(tcp); + let (tx, conn) = hyper::client::conn::http1::Builder::new() + .handshake::<_, http_body_util::Empty>(io) + .await + .unwrap(); + tokio::spawn(conn); + tx + }); + + let url: hyper::Uri = format!("http://{}/hello", addr).parse().unwrap(); + + b.iter(|| { + let mut req = Request::new(http_body_util::Empty::::new()); + *req.uri_mut() = url.clone(); + rt.block_on(async { + let res = client.send_request(req).await.expect("client wait"); + let mut body = res.into_body(); + while let Some(_chunk) = body.frame().await {} + }); + }); +} + +macro_rules! sizes { + ($($n:ident: $sz:expr,)*) => { + $( + mod $n { + const BODY: &[u8] = &[b'x'; $sz]; + #[bench] + fn auto(b: &mut ::test::Bencher) { + super::bench(b, None, BODY) + } + #[bench] + fn flatten(b: &mut ::test::Bencher) { + super::bench(b, Some(false), BODY) + } + #[bench] + fn queue(b: &mut ::test::Bencher) { + super::bench(b, Some(true), BODY) + } + } + )* + }; +} + +sizes! { + b_00064: 64, + b_00256: 256, + b_01024: 1024, + b_04096: 4096, + b_08192: 8192, + b_16384: 16384, + b_20480: 20480, + b_24576: 24576, + b_32768: 32768, + b_65536: 65536, + b_262144: 262144, + b_1048576: 1048576, +} diff --git a/src/proto/h1/io.rs b/src/proto/h1/io.rs index 386f1be35b..1aeaf6cc81 100644 --- a/src/proto/h1/io.rs +++ b/src/proto/h1/io.rs @@ -29,6 +29,25 @@ pub(crate) const DEFAULT_MAX_BUFFER_SIZE: usize = 8192 + 4096 * 100; /// forces a flush if the queue gets this big. const MAX_BUF_LIST_BUFFERS: usize = 16; +/// `WriteStrategy::Auto` copies ("flattens") a buffer smaller than this into +/// the headers buffer, and queues anything this size or larger for a vectored +/// write. +/// +/// Vectored writes avoid copying the body, but they are not free: each extra +/// `IoSlice` costs bookkeeping in `BufList`, in `chunks_vectored`, and in the +/// kernel's `writev` gather loop. For a small body that cost is larger than +/// the `memcpy` it saves, and flattening also lets `poll_flush` use the +/// specialized single-`write` path. +/// +/// This number comes from `benches/h1_writev.rs`, which serves a single +/// response body of a fixed size over a loopback TCP connection with +/// `writev` forced on and forced off. On an `x86_64` Linux machine, flattening +/// was consistently ~1-3% faster up to 20KB, the two were even from 20KB to +/// 24KB, and queueing pulled ahead beyond that (+3% at 64KB, +13% at 256KB, +/// +19% at 1MB). 16KB is the nearest power of two below that crossover, so it +/// stays on the conservative side of it. +const AUTO_FLATTEN_LIMIT: usize = 16 * 1024; + pub(crate) struct Buffered { flush_pipeline: bool, io: T, @@ -58,7 +77,7 @@ where { pub(crate) fn new(io: T) -> Buffered { let strategy = if io.is_write_vectored() { - WriteStrategy::Queue + WriteStrategy::Auto } else { WriteStrategy::Flatten }; @@ -274,7 +293,7 @@ where } else if self.write_buf.remaining() == 0 { Pin::new(&mut self.io).poll_flush(cx) } else { - if let WriteStrategy::Flatten = self.write_buf.strategy { + if self.write_buf.is_flattened() { return self.poll_flush_flattened(cx); } @@ -544,47 +563,70 @@ where self.strategy = strategy; } - pub(super) fn buffer>(&mut self, mut buf: BB) { - debug_assert!(buf.has_remaining()); + /// Returns true if everything buffered so far lives in the headers + /// buffer, and so can be flushed with a single plain `write`. + fn is_flattened(&self) -> bool { match self.strategy { - WriteStrategy::Flatten => { - let head = self.headers_mut(); + WriteStrategy::Flatten => true, + WriteStrategy::Auto => !self.queue.has_remaining(), + WriteStrategy::Queue => false, + } + } - head.maybe_unshift(buf.remaining()); - trace!( - self.len = head.remaining(), - buf.len = buf.remaining(), - "buffer.flatten" - ); - //perf: This is a little faster than >::put, - //but accomplishes the same result. - loop { - let adv = { - let slice = buf.chunk(); - if slice.is_empty() { - return; - } - head.bytes.extend_from_slice(slice); - slice.len() - }; - buf.advance(adv); - } + /// Whether a buffer about to be stored should be flattened into the + /// headers buffer rather than queued. + /// + /// Anything already queued has to be written before `buf`, and the + /// headers buffer is always flushed first, so once the queue is + /// non-empty, queueing is the only way to keep the bytes in order. + fn should_flatten(&self, buf: &BB) -> bool { + match self.strategy { + WriteStrategy::Flatten => true, + WriteStrategy::Auto => { + !self.queue.has_remaining() && buf.remaining() < AUTO_FLATTEN_LIMIT } - WriteStrategy::Queue => { - trace!( - self.len = self.remaining(), - buf.len = buf.remaining(), - "buffer.queue" - ); - self.queue.push(buf.into()); + WriteStrategy::Queue => false, + } + } + + pub(super) fn buffer>(&mut self, mut buf: BB) { + debug_assert!(buf.has_remaining()); + if self.should_flatten(&buf) { + let head = self.headers_mut(); + + head.maybe_unshift(buf.remaining()); + trace!( + self.len = head.remaining(), + buf.len = buf.remaining(), + "buffer.flatten" + ); + //perf: This is a little faster than >::put, + //but accomplishes the same result. + loop { + let adv = { + let slice = buf.chunk(); + if slice.is_empty() { + return; + } + head.bytes.extend_from_slice(slice); + slice.len() + }; + buf.advance(adv); } + } else { + trace!( + self.len = self.remaining(), + buf.len = buf.remaining(), + "buffer.queue" + ); + self.queue.push(buf.into()); } } fn can_buffer(&self) -> bool { match self.strategy { WriteStrategy::Flatten => self.remaining() < self.max_buf_size, - WriteStrategy::Queue => { + WriteStrategy::Auto | WriteStrategy::Queue => { self.queue.bufs_cnt() < MAX_BUF_LIST_BUFFERS && self.remaining() < self.max_buf_size } } @@ -645,6 +687,8 @@ impl Buf for WriteBuf { #[derive(Debug)] enum WriteStrategy { + /// Pick `Flatten` or `Queue` per buffer, based on its size. + Auto, Flatten, Queue, } @@ -945,6 +989,86 @@ mod tests { assert_eq!(buffered.write_buf.queue.bufs_cnt(), 0); } + #[test] + fn write_buf_auto_flattens_small_bufs() { + let _ = pretty_env_logger::try_init(); + + let b = |s: &str| Cursor::new(s.as_bytes().to_vec()); + + let mut write_buf = WriteBuf::>>::new(WriteStrategy::Auto); + + write_buf.buffer(b("hello ")); + write_buf.buffer(b("world, ")); + write_buf.buffer(b("it's hyper!")); + + // all under the limit, so nothing was queued + assert_eq!(write_buf.queue.bufs_cnt(), 0); + assert!(write_buf.is_flattened()); + assert_eq!(write_buf.chunk(), b"hello world, it's hyper!"); + } + + #[test] + fn write_buf_auto_queues_big_bufs() { + let _ = pretty_env_logger::try_init(); + + let mut write_buf = WriteBuf::>>::new(WriteStrategy::Auto); + + write_buf.buffer(Cursor::new(vec![b'X'; AUTO_FLATTEN_LIMIT])); + + assert_eq!(write_buf.queue.bufs_cnt(), 1); + assert!(!write_buf.is_flattened()); + assert_eq!(write_buf.headers.remaining(), 0); + assert_eq!(write_buf.remaining(), AUTO_FLATTEN_LIMIT); + } + + #[test] + fn write_buf_auto_keeps_order_after_queueing() { + let _ = pretty_env_logger::try_init(); + + let b = |s: &str| Cursor::new(s.as_bytes().to_vec()); + + let mut write_buf = WriteBuf::>>::new(WriteStrategy::Auto); + + // headers are small, so they flatten + write_buf.buffer(b("hello ")); + assert_eq!(write_buf.queue.bufs_cnt(), 0); + + // this one is too big, so it queues + write_buf.buffer(Cursor::new(vec![b'X'; AUTO_FLATTEN_LIMIT])); + assert_eq!(write_buf.queue.bufs_cnt(), 1); + + // a small buf after a queued one must *also* queue, or it would be + // written before the bytes that came first + write_buf.buffer(b("bye!")); + assert_eq!(write_buf.queue.bufs_cnt(), 2); + assert!(!write_buf.is_flattened()); + + assert_eq!(write_buf.chunk(), b"hello "); + write_buf.advance(6 + AUTO_FLATTEN_LIMIT); + assert_eq!(write_buf.chunk(), b"bye!"); + } + + #[cfg(not(miri))] + #[tokio::test] + async fn write_buf_auto_flatten_writes_once() { + let _ = pretty_env_logger::try_init(); + + // `tokio_test::io` is not vectored, so a queued flush would show up + // here as separate writes. + let mock = Mock::new().write(b"hello world, it's hyper!").build(); + + let mut buffered = Buffered::<_, Cursor>>::new(Compat::new(mock)); + buffered.write_buf.set_strategy(WriteStrategy::Auto); + + buffered.headers_buf().extend(b"hello "); + buffered.buffer(Cursor::new(b"world, ".to_vec())); + buffered.buffer(Cursor::new(b"it's ".to_vec())); + buffered.buffer(Cursor::new(b"hyper!".to_vec())); + assert_eq!(buffered.write_buf.queue.bufs_cnt(), 0); + + buffered.flush().await.expect("flush"); + } + // #[cfg(feature = "nightly")] // #[bench] // fn bench_write_buf_flatten_buffer_chunk(b: &mut Bencher) {