读写缓冲
主线位置：两条主线的最底层。服务端主线上，socket 读到的每个字节先进读缓冲，解析与编码和 body 的解码都从这里取；响应的头和 body 先放进写缓冲，再由 flush 写到 socket。客户端主线完全一样，只是读的是响应、写的是请求。
flowchart TB io["IO (rt::Read + rt::Write)"] -->|poll_read| rfi["poll_read_from_io"] rfi --> rbuf["read_buf: BytesMut"] rbuf --> parse["parse：解析消息头"] rbuf --> mem["read_mem：body 字节"] head["Conn 编码的头"] --> hdrs["headers: Vec u8"] bodyb["EncodedBuf"] -->|Flatten 拷贝| hdrs bodyb -->|Queue| queue["queue: BufList"] hdrs --> flush["poll_flush"] queue --> flush flush -->|"write 或 writev"| io
Conn 不直接碰 IO，它持有的是一个 Buffered：
pub(crate) struct Buffered<T, B> {
flush_pipeline: bool,
io: T,
partial_len: Option<usize>,
read_blocked: bool,
read_buf: BytesMut,
read_buf_strategy: ReadStrategy,
write_buf: WriteBuf<B>,
}
T 是 IO 类型，B 是要写的 body 块的类型（在 Conn 里是 EncodedBuf<_>）。这一层回答两个问题：每次从 socket 读多少、写出去的字节先在哪攒着。
读：每次读多少
pub(crate) fn poll_read_from_io(&mut self, cx: &mut Context<'_>) -> Poll<io::Result<usize>> {
self.read_blocked = false;
// Get the next amount to allocate, but make sure we don't go over
// the max read buf size configured.
let next = cmp::min(
self.read_buf_strategy.next(),
self.read_buf_strategy
.max()
.saturating_sub(self.read_buf.len()),
);
if self.read_buf_remaining_mut() < next {
self.read_buf.reserve(next);
}
// SAFETY: ReadBuf and poll_read promise not to set any uninitialized
// bytes onto `dst`.
let dst = unsafe { self.read_buf.chunk_mut().as_uninit_slice_mut() };
let mut buf = ReadBuf::uninit(dst);
match Pin::new(&mut self.io).poll_read(cx, buf.unfilled()) {
Poll::Ready(Ok(())) => {
let n = buf.filled().len();
trace!("received {} bytes", n);
// Safety: we just read that many bytes into the
// uninitialized part of the buffer, so this is okay.
// @tokio pls give me back `poll_read_buf` thanks
unsafe {
self.read_buf.advance_mut(n);
}
self.read_buf_strategy.record(n);
Poll::Ready(Ok(n))
}
Poll::Pending => {
self.read_blocked = true;
Poll::Pending
}
Poll::Ready(Err(e)) => Poll::Ready(Err(e)),
}
}
三点：
- 读缓冲一开始容量是 0（
Buffered::new里BytesMut::with_capacity(0)），第一次读时才reserve。只有剩余容量不够next时才扩，所以交给 IO 的空间（chunk_mut的长度）至少是next，可能更大。 - 交给 IO 的是
BytesMut未初始化的那部分，包成ReadBuf::uninit，读多少advance_mut多少，不先清零。ReadBuf为什么能安全地交出未初始化内存，见 IO trait。IO 底下是 tokio 的TcpStream时，一次poll_read怎么走到read(2)或挂起，见从 TcpStream::read 到挂起。 - 每次读完都把读到的字节数交给
read_buf_strategy.record(n)，决定下一次的next。
ReadStrategy::Adaptive
enum ReadStrategy {
Adaptive {
decrease_now: bool,
next: usize,
max: usize,
},
#[cfg(feature = "client")]
Exact(usize),
}
默认是 Adaptive，next 从 INIT_BUFFER_SIZE（8192）开始，max 是 DEFAULT_MAX_BUFFER_SIZE：
/// The default maximum read buffer size. If the buffer gets this big and
/// a message is still not complete, a `TooLarge` error is triggered.
// Note: if this changes, update server::conn::Http::max_buf_size docs.
pub(crate) const DEFAULT_MAX_BUFFER_SIZE: usize = 8192 + 4096 * 100;
8192 + 409600 = 417792 字节，408 KiB，Builder::max_buf_size 的文档写作「~400kb」。它不是 2 的幂。
fn record(&mut self, bytes_read: usize) {
match self {
ReadStrategy::Adaptive {
decrease_now,
next,
max,
..
} => {
if bytes_read >= *next {
*next = cmp::min(incr_power_of_two(*next), *max);
*decrease_now = false;
} else {
let decr_to = prev_power_of_two(*next);
if bytes_read < decr_to {
if *decrease_now {
*next = cmp::max(decr_to, INIT_BUFFER_SIZE);
*decrease_now = false;
} else {
// Decreasing is a two "record" process.
*decrease_now = true;
}
} else {
// A read within the current range should cancel
// a potential decrease, since we just saw proof
// that we still need this size.
*decrease_now = false;
}
}
}
#[cfg(feature = "client")]
ReadStrategy::Exact(_) => (),
}
}
规则：
- 读满了就翻倍：这次读到的不少于
next，说明 socket 里可能还有，下次多要一倍，但不超过max。 - 连续两次读得少才减半：读到的不足
prev_power_of_two(next)（当next是 2 的幂时就是它的一半），第一次只记一笔decrease_now，第二次才真的减；中间只要有一次读到了「一半到next之间」，这笔记录就作废。最低不低于 8192。 - 增长一次一步、减少要两次确认，是为了不在突发流量上来回抖：一次小读可能只是对端刚好暂停。
next 一路翻倍时都是 2 的幂，直到碰到 max。默认 max 是 417792，262144 再翻倍会被截成 417792。从 417792 往下减，prev_power_of_two 算的是：
fn prev_power_of_two(n: usize) -> usize {
// Only way this shift can underflow is if n is less than 4.
// (Which would means `usize::MAX >> 64` and underflowed!)
debug_assert!(n >= 4);
(usize::MAX >> (n.leading_zeros() + 2)) + 1
}
417792 的最高位是第 18 位（2^18 = 262144），leading_zeros 是 45，usize::MAX >> 47 再加 1 是 2^17 = 131072。所以从上限往下减，一步就回到 131072，而不是 262144。io.rs 的单元测试 read_strategy_adaptive_max_fuzz 用一串不是 2 的幂的 max 检查「decrement should be powers of two」，保证的就是这一点：减下来的值总是 2 的幂。
客户端还可以用 read_buf_exact_size 换成 Exact(sz)：每次都读 sz，max() 也是 sz。它和 max_buf_size 互斥，设一个会清掉另一个。
实验：上传 1 MiB 时每次读多少
用一个包装类型记下 hyper 对 IO 的每一次读写：room 是交给 poll_read 的空间（ReadBufCursor::remaining()），got 是实际读到的字节数。
hyper = { version = "=1.12.0", features = ["server", "http1"] }
http-body-util = "=0.1.5"
bytes = "1"
futures = "0.3"
tokio = { version = "=1.54.1", features = ["rt", "net", "macros", "io-util", "time", "sync"] }
#[path = "../common.rs"]
mod common;
use std::convert::Infallible;
use std::io;
use std::pin::Pin;
use std::sync::{Arc, Mutex};
use std::task::{Context, Poll};
use std::time::Duration;
use bytes::Bytes;
use http_body_util::{combinators::BoxBody, BodyExt, Full, StreamBody};
use hyper::body::{Frame, Incoming};
use hyper::rt::{Read, ReadBufCursor, Write};
use hyper::server::conn::http1;
use hyper::service::service_fn;
use hyper::{Request, Response};
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};
type Log = Arc<Mutex<Vec<String>>>;
// Wraps a TcpStream and records every read/write hyper makes.
struct Spy {
inner: TcpStream,
vectored: bool,
log: Log,
}
impl Read for Spy {
fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, mut buf: ReadBufCursor<'_>) -> Poll<io::Result<()>> {
let room = buf.remaining();
let mut tmp = vec![0u8; room];
let mut rb = tokio::io::ReadBuf::new(&mut tmp);
match Pin::new(&mut self.inner).poll_read(cx, &mut rb) {
Poll::Ready(Ok(())) => {
let n = rb.filled().len();
self.log.lock().unwrap().push(format!("read room={room:>6} got={n:>6}"));
buf.put_slice(&tmp[..n]);
Poll::Ready(Ok(()))
}
other => other,
}
}
}
impl Write for Spy {
fn poll_write(mut self: Pin<&mut Self>, cx: &mut Context<'_>, b: &[u8]) -> Poll<io::Result<usize>> {
let r = Pin::new(&mut self.inner).poll_write(cx, b);
if let Poll::Ready(Ok(n)) = r {
self.log.lock().unwrap().push(format!("write {n}B"));
}
r
}
fn poll_write_vectored(mut self: Pin<&mut Self>, cx: &mut Context<'_>, bufs: &[io::IoSlice<'_>]) -> Poll<io::Result<usize>> {
let r = Pin::new(&mut self.inner).poll_write_vectored(cx, bufs);
if let Poll::Ready(Ok(n)) = r {
let lens: Vec<usize> = bufs.iter().map(|b| b.len()).collect();
self.log.lock().unwrap().push(format!("writev {n}B iovecs={lens:?}"));
}
r
}
fn is_write_vectored(&self) -> bool {
self.vectored
}
fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
Pin::new(&mut self.inner).poll_flush(cx)
}
fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
Pin::new(&mut self.inner).poll_shutdown(cx)
}
}
type B = BoxBody<Bytes, Infallible>;
async fn handle(req: Request<Incoming>) -> Result<Response<B>, Infallible> {
match req.uri().path() {
"/stream" => {
let frames = vec![Ok::<_, Infallible>(Frame::data(Bytes::from("abc"))), Ok(Frame::data(Bytes::from("defg")))];
Ok(Response::new(StreamBody::new(futures::stream::iter(frames)).boxed()))
}
"/upload" => {
let n = req.into_body().collect().await.unwrap().to_bytes().len();
Ok(Response::new(Full::new(Bytes::from(n.to_string())).boxed()))
}
_ => Ok(Response::new(Full::new(Bytes::from("hello")).boxed())),
}
}
async fn run(name: &str, vectored: bool, cfg: impl Fn(&mut http1::Builder), parts: Vec<Vec<u8>>) {
println!("== {name}");
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
let log: Log = Default::default();
let mut builder = http1::Builder::new();
cfg(&mut builder);
let server_log = log.clone();
let server = tokio::spawn(async move {
let (stream, _) = listener.accept().await.unwrap();
let io = Spy { inner: stream, vectored, log: server_log };
builder.serve_connection(io, service_fn(handle)).await.map_err(|e| common::err_chain(&e))
});
let mut tcp = TcpStream::connect(addr).await.unwrap();
for (i, p) in parts.iter().enumerate() {
if i > 0 {
tokio::time::sleep(Duration::from_millis(50)).await;
}
if tcp.write_all(p).await.is_err() {
break;
}
}
let mut got = Vec::new();
let mut buf = vec![0u8; 64 * 1024];
loop {
match tokio::time::timeout(Duration::from_millis(300), tcp.read(&mut buf)).await {
Ok(Ok(n)) if n > 0 => got.extend_from_slice(&buf[..n]),
_ => break,
}
}
drop(tcp);
let result = server.await.unwrap();
for line in log.lock().unwrap().iter() {
println!("{line}");
}
let first_line = String::from_utf8_lossy(&got).lines().next().unwrap_or("").to_string();
println!("client got {} bytes, first line {first_line:?}", got.len());
println!("server: {result:?}");
}
#[tokio::main(flavor = "current_thread")]
async fn main() {
// 1. Read sizes while a 1 MiB body streams in.
let mut upload = b"POST /upload HTTP/1.1\r\ncontent-length: 1048576\r\n\r\n".to_vec();
upload.extend(std::iter::repeat_n(b'x', 1 << 20));
run("upload 1 MiB", true, |_| {}, vec![upload]).await;
// 2. Write strategy for a chunked response.
let get_stream = b"GET /stream HTTP/1.1\r\n\r\n".to_vec();
run("stream, queue (vectored io)", true, |_| {}, vec![get_stream.clone()]).await;
run("stream, flatten (io not vectored)", false, |_| {}, vec![get_stream.clone()]).await;
run("stream, vectored io but writev(false)", true, |b| { b.writev(false); }, vec![get_stream]).await;
// 3. Two pipelined requests in one write.
let two = b"GET /a HTTP/1.1\r\n\r\nGET /b HTTP/1.1\r\n\r\n".to_vec();
run("pipelined, default", false, |_| {}, vec![two.clone()]).await;
run("pipelined, pipeline_flush(true)", false, |b| { b.pipeline_flush(true); }, vec![two]).await;
// 4. max_buf_size(8192), a 12 KiB head split over two writes (issue #4081's shape).
let mut p1 = b"GET / HTTP/1.1\r\nHost: x\r\nX: ".to_vec();
p1.extend(std::iter::repeat_n(b'a', 7000));
let mut p2 = vec![b'a'; 5000];
p2.extend_from_slice(b"\r\n\r\n");
run("max_buf_size(8192), 12 KiB head", true, |b| { b.max_buf_size(8192); }, vec![p1, p2]).await;
}
common.rs 只用到其中的 err_chain，见解析与编码页。M2 Max、rustc 1.99.0、--release。第一项依赖 socket 里攒了多少数据，每次运行不同；跑了 5 次，前 6 次读每次都一样，后面各不相同。两次运行的输出：
== upload 1 MiB
read room= 8192 got= 8192
read room= 16384 got= 16384
read room= 32768 got= 32768
read room= 65536 got= 65536
read room=131072 got=131072
read room=262144 got= 73260
read room=262144 got=262144
read room=417792 got=130604
read room=417792 got= 90112
read room=327680 got= 32664
read room=295016 got=130656
read room=164360 got= 75234
writev 82B iovecs=[75, 7]
read room=131072 got= 0
client got 82 bytes, first line "HTTP/1.1 200 OK"
server: Ok(())
== upload 1 MiB
read room= 8192 got= 8192
read room= 16384 got= 16384
read room= 32768 got= 32768
read room= 65536 got= 65536
read room=131072 got=131072
read room=262144 got= 73260
read room=262144 got=146988
read room=262144 got= 81660
read room=262144 got= 81660
read room=180484 got= 16332
read room=164152 got=164152
read room=262144 got=230622
writev 82B iovecs=[75, 7]
read room=262144 got= 0
client got 82 bytes, first line "HTTP/1.1 200 OK"
server: Ok(())
按第一次运行逐行对照 record：
- 前五次都读满，
next从 8192 翻到 262144。 - 第 6 次只读到 73260，小于
prev_power_of_two(262144)= 131072，记一笔decrease_now。第 7 次读满 262144，翻倍并清掉这笔记录，next=min(524288, 417792)= 417792，第 8 行的room正是它。 - 第 8 次读到 130604，小于 131072，记一笔；第 9 次 90112，又小于，
next减到 131072。之后的room（327680、295016、164360）比next大，是因为BytesMut里还有剩余容量，只有不够next时才reserve。 - 1 MiB 的 body 比
max_buf_size大一倍多，照样读完：body 的字节被read_mem切走交给Incoming，读缓冲不会累积。
头部太大：max_buf_size 与 431
Buffered::parse 是一个循环：解析、没收全就检查长度、再读一次。
loop {
if let Some(msg) = super::role::parse_headers::<S>(
&mut self.read_buf,
self.partial_len,
ParseContext {
// ...
},
)? {
debug!("parsed {} headers", msg.head.headers.len());
self.partial_len = None;
return Poll::Ready(Ok(msg));
} else {
let curr_len = self.read_buf.len();
if let Some(max_header_size) = parse_ctx.h1_max_header_size {
if curr_len >= max_header_size {
debug!("max_header_size ({}) reached, closing", max_header_size);
return Poll::Ready(Err(crate::Error::new_too_large()));
}
}
let max = self.read_buf_strategy.max();
if curr_len >= max {
debug!("max_buf_size ({}) reached, closing", max);
return Poll::Ready(Err(crate::Error::new_too_large()));
}
if curr_len > 0 {
trace!("partial headers; {} bytes so far", curr_len);
self.partial_len = Some(curr_len);
} else {
// 1xx gobled some bytes
self.partial_len = None;
}
}
if ready!(self.poll_read_from_io(cx)).map_err(crate::Error::new_io)? == 0 {
trace!("parse eof");
return Poll::Ready(Err(crate::Error::new_incomplete()));
}
}
缓冲涨到 max_buf_size 还没解析出完整的头，就是「message head is too large」，服务端回 431（on_error 见解析与编码）。partial_len 就是解析与编码页里 is_complete_fast 用的 prev_len。读到 0 字节（对端关了）是 IncompleteMessage。
这个检查只在「头没收全」的分支里做。头一旦完整，就不再看长度。所以要保证缓冲永远不会在头完整之前超过 max，靠的是 poll_read_from_io 里 next 的那个 min：每次最多只读到 max - read_buf.len()，缓冲恰好填到 max 时还不完整，下一轮循环就报错。
issue #4081 报告：「In HTTP/1, Buffered::parse, the check only runs when parse_headers returns None (incomplete headers).」当时 poll_read_from_io 的第一行是 let next = self.read_buf_strategy.next();，没有和 max - read_buf.len() 取小。issue 给的复现是 max_buf_size(8192)，先发 7000 多字节的头，停一下，再发 5000 字节和结尾的空行：第二次读的空间仍是 8192，一次就把剩下的全读进来，头完整了，长度检查被跳过，「The server happily responds HTTP/1.1 200 OK and the handler sees the oversized X header.」
#4093（首次发布于 v1.11.0）加上了上面那个 cmp::min，同时在 tests/server.rs 里加了回归测试。下面的实验用的就是 issue 里的形状。
实验第 4 项，连跑 3 次输出相同：
== max_buf_size(8192), 12 KiB head
read room= 8192 got= 7028
read room= 1164 got= 1164
writev 123B iovecs=[123]
client got 123 bytes, first line "HTTP/1.1 431 Request Header Fields Too Large"
server: Err("message head is too large")
第一次读到 7028 字节，没收全；第二次的空间是 8192 − 7028 = 1164，恰好把缓冲填到 8192，仍然没收全，回 431。剩下的近 4000 字节根本没读。
几个相关的值：
max_buf_size不能小于 8192（MINIMUM_MAX_BUFFER_SIZE，等于INIT_BUFFER_SIZE），否则 builder 直接 panic：「the max_buf_size cannot be smaller than the minimum that h1 specifies.」max_header_size（1.12.0 新增，默认None）是一个更直接的头部上限，在同一个循环里先检查；解析成功后Server::parse还会再对比一次完整头部的长度。max_buf_size同时也是写缓冲的上限，见下一节。
写：先攒再写
pub(super) struct WriteBuf<B> {
/// Re-usable buffer that holds message headers.
headers: Cursor<Vec<u8>>,
max_buf_size: usize,
/// Deque of user buffers if strategy is `Queue`.
queue: BufList<B>,
strategy: WriteStrategy,
}
headers 是一个可复用的 Vec<u8>，初始容量 8192，Conn::encode_head 把消息头直接写进去（Buffered::headers_buf）。body 的块（EncodedBuf）怎么放，看 WriteStrategy。WriteBuf::buffer：
match self.strategy {
WriteStrategy::Flatten => {
let head = self.headers_mut();
head.maybe_unshift(buf.remaining());
// ...
loop {
let adv = {
let slice = buf.chunk();
if slice.is_empty() {
return;
}
head.bytes.extend_from_slice(slice);
slice.len()
};
buf.advance(adv);
}
}
WriteStrategy::Queue => {
// ...
self.queue.push(buf.into());
}
}
- Flatten：把 body 拷进
headers这个Vec后面，flush 时一次poll_write写一整段。maybe_unshift在空间不够时先把已写出的前缀挪掉，复用内存。 - Queue：body 的 buf 原样放进一个
VecDeque（BufList），flush 时用chunks_vectored收集成最多 64 个IoSlice（MAX_WRITEV_BUFS），一次poll_write_vectored。EncodedBuf里的Chain每一段都是一个IoSlice，用户数据不拷贝。
选哪个，在 Buffered::new 里按 IO 自己说的定：
pub(crate) fn new(io: T) -> Buffered<T, B> {
let strategy = if io.is_write_vectored() {
WriteStrategy::Queue
} else {
WriteStrategy::Flatten
};
这就是 writev 选项文档里的「Default is auto」。hyper-util 的 TokioIo 把 is_write_vectored 转给 tokio，TcpStream 返回 true，所以裸 TCP 默认是 Queue。显式设置会覆盖：writev(true) 强制 Queue，writev(false) 强制 Flatten。文档说关掉它「may also improve performance when an IO transport doesn't support vectored writes well, such as most TLS implementations」：TLS 层要把明文加密成记录，多个小 IoSlice 未必比一整段好处理。
实验第 2 项，同一个两帧（abc、defg）的 chunked 响应，连跑 3 次输出相同：
== stream, queue (vectored io)
read room= 8192 got= 24
writev 106B iovecs=[84, 3, 3, 2, 3, 4, 2, 5]
read room= 8192 got= 0
client got 106 bytes, first line "HTTP/1.1 200 OK"
server: Ok(())
== stream, flatten (io not vectored)
read room= 8192 got= 24
write 106B
read room= 8192 got= 0
client got 106 bytes, first line "HTTP/1.1 200 OK"
server: Ok(())
== stream, vectored io but writev(false)
read room= 8192 got= 24
write 106B
read room= 8192 got= 0
client got 106 bytes, first line "HTTP/1.1 200 OK"
server: Ok(())
Queue 下 8 个 IoSlice：84 字节的响应头，3\r\n、abc、\r\n，4\r\n、defg、\r\n，最后 0\r\n\r\n。两帧都立即就绪，Dispatcher 在 flush 前把整个响应都放进了写缓冲，所以只有一次系统调用。Flatten 下同样是一次，只是 106 字节先拷进了一个 Vec。上面上传实验的响应是 Full，iovecs=[75, 7]：头和 "1048576" 两段。
写缓冲的上限与顺序
fn can_buffer(&self) -> bool {
match self.strategy {
WriteStrategy::Flatten => self.remaining() < self.max_buf_size,
WriteStrategy::Queue => {
self.queue.bufs_cnt() < MAX_BUF_LIST_BUFFERS && self.remaining() < self.max_buf_size
}
}
}
写缓冲里攒的字节不能超过 max_buf_size，Queue 下 buf 的个数也不能超过 MAX_BUF_LIST_BUFFERS（16）。不能再攒时，Dispatcher 的 poll_write 先 flush，flush 不动就返回 Pending，不再 poll_frame 响应 body。socket 写不出去，body 就不会被继续拉取，这是写方向的背压。
Queue 还有一个顺序上的约束，can_headers_buf：
/// Return whether we can append to the headers buffer.
///
/// Reasons we can't:
/// - The write buf is in queue mode, and some of the past body is still
/// needing to be flushed.
pub(crate) fn can_headers_buf(&self) -> bool {
!self.write_buf.queue.has_remaining()
}
WriteBuf 作为 Buf 时永远先吐 headers 再吐 queue。如果上一个响应的 body 还在 queue 里，下一个响应的头就写进 headers，它会跑到上一个 body 的前面。所以 Conn::can_write_head 在 Writing::Init 时还要问一句 can_headers_buf，queue 没写完就不写下一个头。Flatten 没有这个问题，所有东西都按顺序拷在同一个 Vec 里。
pipeline_flush
HTTP/1.1 允许客户端不等响应就连发几个请求（pipelining）。Dispatcher 仍然一个一个处理，但每处理完一个，poll_loop 都会 flush 一次。pipeline_flush(true) 改的是 Buffered::poll_flush 的第一个分支：
pub(crate) fn poll_flush(&mut self, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
if self.flush_pipeline && !self.read_buf.is_empty() {
Poll::Ready(Ok(()))
} else if self.write_buf.remaining() == 0 {
Pin::new(&mut self.io).poll_flush(cx)
} else {
if let WriteStrategy::Flatten = self.write_buf.strategy {
return self.poll_flush_flattened(cx);
}
// ...
读缓冲里还有没处理的字节（多半是下一个流水线请求），就假装 flush 完了，先不写。等读缓冲空了再一起写。打开它时 set_flush_pipeline 还会把写策略切成 Flatten。另外 Buffered::can_buffer 是 self.flush_pipeline || self.write_buf.can_buffer()：打开后上一节的写缓冲上限不再生效。文档把它标为「Experimental, may have bugs」，默认关。
交还 IO
Buffered::into_inner 返回 (io, read_buf.freeze())：IO 本身，加上读缓冲里已经读进来、还没被解析的字节。写缓冲里的字节不在返回值里。Connection::into_parts 交给用户的 read_buf 和协议升级后 Upgraded 前面那段 Rewind 的字节，都来自这里，见服务端 Builder 与 Connection 和协议升级。
实验第 3 项：一次 write_all 发出两个请求，GET /a 和 GET /b，每个响应都是 Full::new("hello")。IO 不支持向量写（Flatten）。默认配置下，服务端调了几次 poll_write？打开 pipeline_flush(true) 呢？
动脑筋答案
M2 Max、rustc 1.99.0、--release，连跑 3 次输出相同：
== pipelined, default
read room= 8192 got= 38
write 80B
write 80B
read room= 8192 got= 0
client got 160 bytes, first line "HTTP/1.1 200 OK"
server: Ok(())
== pipelined, pipeline_flush(true)
read room= 8192 got= 38
write 160B
read room= 8192 got= 0
client got 160 bytes, first line "HTTP/1.1 200 OK"
server: Ok(())
- 两个请求共 38 字节，一次读进读缓冲。
- 默认：处理
/a、写好响应后，poll_loop调poll_flush，80 字节写出去；再处理/b，又写一次。两次poll_write。 pipeline_flush(true)：写好/a的响应时，读缓冲里还躺着GET /b的 19 字节，poll_flush直接返回Ready，什么都不写；处理完/b，读缓冲空了，两个响应 160 字节一次写出。一次poll_write。- 每个响应 80 字节：
HTTP/1.1 200 OK\r\n、content-length: 5\r\n、date:加 29 字节的日期和\r\n\r\n，再加hello。
小结
Buffered持有 IO、一个BytesMut读缓冲和一个WriteBuf写缓冲；Conn的所有 IO 都经过它。- 每次读的大小由
ReadStrategy::Adaptive决定：从 8192 起，读满就翻倍，连续两次读得不到一半才减半，最低 8192，最高max_buf_size（默认 8192 + 4096×100 = 417792）。客户端可以用read_buf_exact_size固定大小。 - 解析消息头时，读缓冲到
max_buf_size（或max_header_size）还不完整就是「message head is too large」，服务端回 431。每次读最多读到max - len，保证头完整前不会越过上限（1.11.0 的 #4093 起）。body 不受这个上限影响。 - 写缓冲按
is_write_vectored自动选：Queue 把EncodedBuf排进队列，用writev一次写出，不拷贝；Flatten 拷进一个Vec，一次write。writev(bool)可以覆盖，TLS 下关掉可能更好。 - 写缓冲同样以
max_buf_size为上限（Queue 下还限 16 个 buf），满了就先 flush，flush 不动就不再拉 body。Queue 下上一个 body 没写完不写下一个头，保证顺序。 pipeline_flush在读缓冲非空时推迟 flush，把流水线请求的响应攒成一次写；会强制 Flatten 并取消写缓冲上限，是实验性选项。into_inner交还 IO 和读缓冲里没解析的字节，into_parts和协议升级都靠它。