一块缓冲的一生

主线位置：两条主线都在。读缓冲从分配、写入到 split_to、reserve 走的是构建主线（BytesMut）；切出去的请求头和 body freeze 之后的 slice、drop 走的是共享主线（Bytes）。本页用一个最小程序把两条线连起来跑一遍，每一步只点出落在哪个函数上，机制留给 02、03 两部分。

sequenceDiagram
  participant Sock as socket
  participant RB as read_buf BytesMut
  participant Hd as head Bytes
  participant Uri as uri Bytes
  participant Bd as body Bytes
  RB->>RB: with_capacity 0
  RB->>RB: reserve 8192，分配
  Sock->>RB: 写入 65 字节
  RB->>Hd: split_to 54，建 Shared
  Hd->>Hd: freeze，换 vtable
  Hd->>Uri: slice 5..12
  RB->>Bd: split_to 11 再 freeze
  Uri-->>Uri: drop
  Hd-->>Hd: drop
  Bd-->>Bd: drop
  RB->>RB: reserve 8192，原地复用

hyper 的 HTTP/1 服务端读一个请求，对读缓冲只做这几件事（见 hyper 卷的读写缓冲、解析与编码和 body 的解码与编码）：

Buffered::new 里读缓冲是 BytesMut::with_capacity(0)，不分配。
poll_read_from_io 在剩余容量不够下一次读的大小（初始 8 KiB）时 reserve，然后把 chunk_mut() 交给 socket 填，填多少 advance_mut 多少。
请求头解析完，Server::parse 用 buf.split_to(len).freeze() 把头那一段切走，URI 和每个头的值都是这个 Bytes 的 slice。
body 由 read_mem 交出，也是 split_to(n).freeze()。
上层用完这些 Bytes 后 drop；连接上的下一次读又回到第 2 步。

下面的程序照这个顺序做一遍。为了不引入 socket，第 2 步的「socket 填数据」用 put_slice 代替。

程序

common.rs 是上一页的计数分配器和打印函数（见 bytes 解决什么问题）。mut_state / bytes_state 打印指针相对缓冲起点 base 的偏移，不在最初那块 8 KiB 里就打印 elsewhere；BytesMut 没有公开的 is_unique，所以只对 Bytes 打印它。

bytes = "=1.12.1"

use bytes::{BufMut, BytesMut};

#[path = "../common.rs"]
mod common;
use common::*;

const REQ: &[u8] = b"POST /upload HTTP/1.1\r\nHost: a\r\nContent-Length: 11\r\n\r\nhello world";

fn main() {
  let head_len = REQ.len() - 11;
  let mut s = snap();

  // 1. hyper 的 Buffered::new：容量 0，不分配
  let mut read_buf = BytesMut::with_capacity(0);
  let d = delta(s);
  s = report("with_capacity(0)", d, format!("read_buf: len={} cap={}", read_buf.len(), read_buf.capacity()));

  // 2. 第一次读之前 reserve(8192)
  read_buf.reserve(8192);
  let base = read_buf.as_ptr() as usize;
  let d = delta(s);
  s = report("reserve(8192)", d, mut_state("read_buf", &read_buf, base));

  // 3. socket 读进来（这里用 put_slice 代替 chunk_mut + advance_mut）
  read_buf.put_slice(REQ);
  let d = delta(s);
  s = report("put_slice(65)", d, mut_state("read_buf", &read_buf, base));

  // 4. 解析完请求头，把头那一段切走
  let head = read_buf.split_to(head_len);
  let d = delta(s);
  s = report("split_to(head)", d, format!("{} | {}", mut_state("head", &head, base), mut_state("read_buf", &read_buf, base)));

  // 5. 冻结成 Bytes
  let head = head.freeze();
  let d = delta(s);
  s = report("head.freeze()", d, bytes_state("head", &head, base));

  // 6. 取出 URI 那一段
  let uri = head.slice(5..12);
  let d = delta(s);
  s = report("head.slice(5..12)", d, format!("{} {:?}", bytes_state("uri", &uri, base), std::str::from_utf8(&uri).unwrap()));

  // 7. body：Length 解码器 split_to(n).freeze()
  let body = read_buf.split_to(11).freeze();
  let d = delta(s);
  s = report("split_to(11).freeze()", d, format!("{} | {}", bytes_state("body", &body, base), mut_state("read_buf", &read_buf, base)));

  // 8. 上层用完，逐个 drop
  drop(uri);
  let d = delta(s);
  s = report("drop(uri)", d, bytes_state("head", &head, base));
  drop(head);
  let d = delta(s);
  s = report("drop(head)", d, bytes_state("body", &body, base));
  drop(body);
  let d = delta(s);
  s = report("drop(body)", d, mut_state("read_buf", &read_buf, base));

  // 9. 下一次读之前又 reserve(8192)
  read_buf.reserve(8192);
  let d = delta(s);
  s = report("reserve(8192)", d, mut_state("read_buf", &read_buf, base));

  drop(read_buf);
  let d = delta(s);
  report("drop(read_buf)", d, String::new());
}

输出

M2 Max、rustc 1.99.0，cargo run --release --bin lifecycle，跑 3 次，输出相同：

with_capacity(0)       alloc+0 (0B) free+0  read_buf: len=0 cap=0
reserve(8192)          alloc+1 (8192B) free+0  read_buf: ptr=base+0 len=0 cap=8192
put_slice(65)          alloc+0 (0B) free+0  read_buf: ptr=base+0 len=65 cap=8192
split_to(head)         alloc+1 (40B) free+0  head: ptr=base+0 len=54 cap=54 | read_buf: ptr=base+54 len=11 cap=8138
head.freeze()          alloc+0 (0B) free+0  head: ptr=base+0 len=54 unique=false
head.slice(5..12)      alloc+0 (0B) free+0  uri: ptr=base+5 len=7 unique=false "/upload"
split_to(11).freeze()  alloc+0 (0B) free+0  body: ptr=base+54 len=11 unique=false | read_buf: ptr=base+65 len=0 cap=8127
drop(uri)              alloc+0 (0B) free+0  head: ptr=base+0 len=54 unique=false
drop(head)             alloc+0 (0B) free+0  body: ptr=base+54 len=11 unique=false
drop(body)             alloc+0 (0B) free+0  read_buf: ptr=base+65 len=0 cap=8127
reserve(8192)          alloc+0 (0B) free+0  read_buf: ptr=base+0 len=0 cap=8192
drop(read_buf)         alloc+0 (0B) free+2

一个请求从进来到处理完，只分配了两次：8192 字节的缓冲，和第一次 split_to 时 40 字节的 Shared。之后的 freeze、slice、第二次 split_to、所有的 drop 和最后的 reserve 都是 0。所有指针都是 base+N，没有一个字节被搬到别处。下面按步骤看每一行落在哪里。

分配与写入：还是一个 Vec

with_capacity 只是包了一个 Vec：

    pub fn with_capacity(capacity: usize) -> BytesMut {
        BytesMut::from_vec(Vec::with_capacity(capacity))
    }

Vec::with_capacity(0) 不分配。from_vec 把 Vec 拆成 ptr、len、cap 三个字段，data 不是真的指针，而是一组标记位：最低位 KIND_VEC 表示「这还是一个独占的 Vec」，其余位记原始容量和前面被切掉了多少。这时 BytesMut 和 Vec<u8> 的差别只在这几位上。

reserve(8192) 发现剩余容量 0 不够，进 reserve_inner 的 KIND_VEC 分支，把三个字段拼回一个 Vec 再调 Vec::reserve，分配 8192 字节。put_slice(65) 只是往容量里写、len 加 65，没有分配。hyper 真正用的是 chunk_mut() 加 advance_mut(n)，效果一样，区别是它把未初始化的那段直接交给 socket 去填，见 BufMut 与 UninitSlice。

data 的位怎么排、为什么能塞进一个指针，在 data 字段的编码。

split_to：从 Vec 变成共享

BytesMut::split_to：

    pub fn split_to(&mut self, at: usize) -> BytesMut {
        assert!(
            at <= self.len(),
            "split_to out of bounds: {:?} <= {:?}",
            at,
            self.len(),
        );

        unsafe {
            // ...
            let mut other = self.shallow_clone();
            // ...
            self.advance_unchecked(at);
            other.cap = at;
            other.len = at;
            other
        }
    }

shallow_clone 复制出一个和自己一模一样的句柄，然后两边各改各的：self 往后挪 at 个字节，other 的 len 和 cap 都截成 at。输出里正是这样：head 是 base+0、len=54、cap=54；read_buf 是 base+54、len=11（还没被解析的 body）、cap=8138（8192 − 54）。head 的 cap 被截成 54，所以它再怎么写也写不到 read_buf 那一段去，两个可写的句柄指向同一块内存却不会互相覆盖。

两个句柄共用一块内存，就得有引用计数。shallow_clone 在 KIND_VEC 时调用 promote_to_shared(2)：分配一个 Shared，把原来的 Vec 放进去，计数直接从 2 开始，data 改成指向它，最低位变成 0（KIND_ARC）。这就是那 40 字节。已经是 KIND_ARC 时只把计数加一，所以第二次 split_to（第 7 步）没有分配。提升的细节和 split_off、unsplit 在 split 与 freeze。

freeze 与 slice：换一张 vtable

freeze：

    pub fn freeze(self) -> Bytes {
        let bytes = ManuallyDrop::new(self);
        if bytes.kind() == KIND_VEC {
            // Just re-use `Bytes` internal Vec vtable
            unsafe {
                let off = bytes.get_vec_pos();
                let vec = rebuild_vec(bytes.ptr.as_ptr(), bytes.len, bytes.cap, off);
                let mut b: Bytes = vec.into();
                b.advance(off);
                b
            }
        } else {
            debug_assert_eq!(bytes.kind(), KIND_ARC);

            let ptr = bytes.ptr.as_ptr();
            let len = bytes.len;
            let data = AtomicPtr::new(bytes.data.cast());
            unsafe { Bytes::with_vtable(ptr, len, data, &SHARED_VTABLE) }
        }
    }

head 已经是 KIND_ARC，走 else 分支：ptr、len 原样搬过去，data 还是那个 Shared 的地址，再配上 bytes_mut.rs 里的 SHARED_VTABLE。不分配，不动计数，BytesMut 的那份计数直接变成了 Bytes 的。Bytes 有好几张 vtable，这是其中专门认识 BytesMut 的 Shared 的那一张，见五种存储。

KIND_VEC 分支是从没 split 过的 BytesMut 冻结时走的：拼回 Vec，交给 From<Vec<u8>>。上一页看过，这一步在长度不等于容量时要分配一个 24 字节的 Shared。用同一个 common.rs 量一下：

use bytes::{BufMut, BytesMut};

#[path = "../common.rs"]
mod common;
use common::*;

fn main() {
  // 没有 split 过（KIND_VEC），容量比长度大
  let mut a = BytesMut::with_capacity(8192);
  a.put_slice(&[7u8; 65]);
  let base = a.as_ptr() as usize;
  let s = snap();
  let a = a.freeze();
  let d = delta(s);
  report("cap 8192 freeze()", d, bytes_state("a", &a, base));

  // 没有 split 过，容量正好等于长度
  let mut b = BytesMut::with_capacity(65);
  b.put_slice(&[7u8; 65]);
  let base = b.as_ptr() as usize;
  let s = snap();
  let b = b.freeze();
  let d = delta(s);
  let s = report("cap 65 freeze()", d, bytes_state("b", &b, base));
  let b2 = b.clone();
  let d = delta(s);
  report("b.clone()", d, bytes_state("b2", &b2, base));
}

跑 3 次，输出相同：

cap 8192 freeze()      alloc+1 (24B) free+0  a: ptr=base+0 len=65 unique=true
cap 65 freeze()        alloc+0 (0B) free+0  b: ptr=base+0 len=65 unique=true
b.clone()              alloc+1 (24B) free+0  b2: ptr=base+0 len=65 unique=false

容量 8192、只用了 65 字节的缓冲，freeze 分配 24 字节，数据留在原地（base+0）；容量正好 65 时 freeze 不分配，到第一次 clone 才分配。两种情况都没有拷贝那 65 字节。

📝NOTE

冷知识：freeze 曾经会把整块缓冲重新分配一遍

freeze 的 KIND_VEC 分支把 Vec 交给 From<Vec<u8>>。在 bytes 1.4.0 之前，From<Vec<u8>> 就是 vec.into_boxed_slice() 再转 Bytes。PR #592（「Avoid large reallocations when freezing BytesMut」）指出：into_boxed_slice「reallocates to a slice of the same length as the Vev if the length and the capacity are not equal」，「This can pose a performance problem if the Vec is large or if this happens many times in a loop」。上面第一个例子在当时就是：为了 65 字节的内容，把 8192 字节的缓冲重新分配成 65 字节。

改法是先比较长度和容量，相等时照旧 into_boxed_slice，不等时改用能记住独立容量的共享 vtable：「While this does not avoid allocation altogether, it performs a fixed size allocation and avoids any need to memcpy.」这个「固定大小的分配」就是输出里那 24 字节。CHANGELOG 记在 1.4.0：「Avoid large reallocations when freezing BytesMut (#592)」。

回到主程序。Bytes::slice 先 clone 一次（经 vtable，Shared 的计数加一），再改新句柄的 ptr 和 len：

    pub fn slice(&self, range: impl RangeBounds<usize>) -> Self {
        let (begin, end) = crate::range(range, self.len());

        if end == begin {
            return Bytes::new_empty_with_ptr(self.ptr.wrapping_add(begin));
        }

        let mut ret = self.clone();

        ret.len = end - begin;
        ret.ptr = unsafe { ret.ptr.add(begin) };

        ret
    }

所以 uri 是 base+5、len=7，内容是 "/upload"。空范围是个例外，不碰计数，直接返回一个空的 Bytes。hyper 里每个头的值、URI 都是这样从同一个 head 上切出来的。slice、slice_ref、split_to 的边界情况在切片。

第 7 步的 read_buf.split_to(11).freeze() 和 hyper 的 read_mem 一模一样：read_buf 已经是 KIND_ARC，split_to 只加计数，freeze 只换 vtable。之后 read_buf 是 base+65、len=0、cap=8127。

drop：只减计数

这一路上谁持有那个 Shared 的计数，可以照源码数出来（计数本身是私有字段，程序里看不到，能看到的是 is_unique）：

步骤	持有者	计数
`split_to(head)`	`read_buf`、`head`	2
`head.freeze()`	`read_buf`、`head`	2
`head.slice(5..12)`	`read_buf`、`head`、`uri`	3
`split_to(11).freeze()`	`read_buf`、`head`、`uri`、`body`	4
`drop(uri)`、`drop(head)`、`drop(body)`	依次减到 1，只剩 `read_buf`	1
`drop(read_buf)`	无	0，释放

前三次 drop 的 free+0 对应「减了但没到零」；最后 drop(read_buf) 计数归零，free+2 是缓冲和 Shared 两块。

📝NOTE

没有蠢问题

问： drop(head) 之后只剩 body 一个 Bytes 了，为什么 body.is_unique() 还是 false？

答：因为 read_buf 也算一个。is_unique 看的是那个 Shared 的计数是不是 1，而 read_buf 虽然是 BytesMut，持有的是同一个 Shared。只要 hyper 的读缓冲还在（连接还开着），从它切出去的任何 Bytes 都不是唯一的，try_into_mut 也会返回 Err。这不是浪费：正是这一份计数让 read_buf 能在下一步判断「别人都走了，整块归我」。

再 reserve：整块收回

最后 read_buf 的 len=0、cap=8127，比 hyper 下一次要读的 8192 少，于是 reserve(8192)。这次进的是 reserve_inner 的 KIND_ARC 分支：

        unsafe {
            // First, try to reclaim the buffer. This is possible if the current
            // handle is the only outstanding handle pointing to the buffer.
            if (*shared).is_unique() {
                // ...
                let v = &mut (*shared).vec;

                let v_capacity = v.capacity();
                let ptr = v.as_mut_ptr();

                let offset = self.ptr.as_ptr().offset_from(ptr) as usize;
                // ...
                if v_capacity >= new_cap_plus_offset {
                    self.cap = new_cap;
                    // no copy is necessary
                } else if v_capacity >= new_cap && offset >= len {
                    // The capacity is sufficient, and copying is not too much
                    // overhead: reclaim the buffer!

                    // `offset >= len` means: no overlap
                    ptr::copy_nonoverlapping(self.ptr.as_ptr(), ptr, len);

                    self.ptr = vptr(ptr);
                    self.cap = v.capacity();
                } else {
                    // ...

计数是 1，read_buf 是唯一的持有者。new_cap 是 0 + 8192，offset 是 65：第一个条件 8192 ≥ 8192 + 65 不成立；第二个条件 8192 ≥ 8192 且 65 ≥ 0 成立，把 0 个字节拷回开头，ptr 回到 base+0，cap 恢复成整块的 8192。输出里这一步 alloc+0，ptr=base+0 cap=8192：上一个请求用过的内存原封不动地给了下一个请求。

reserve 的其他结果（不唯一时分配新块、新容量怎么定、KIND_VEC 时的原地回收）在 reserve。

❗IMPORTANT

动脑筋：上层还拿着 body 时，hyper 准备下一次读会怎样？

平台 M2 Max、rustc 1.99.0，common.rs 同上。和主程序一样走到 body 切出去，head 已经 drop，但 body 还活着时就 reserve(8192)。这一步分配几次、多少字节？read_buf 的指针还在不在原来那块里？之后 body.is_unique() 是什么，drop(body) 释放几块？

use bytes::{BufMut, BytesMut};

#[path = "../common.rs"]
mod common;
use common::*;

const REQ: &[u8] = b"POST /upload HTTP/1.1\r\nHost: a\r\nContent-Length: 11\r\n\r\nhello world";

fn main() {
  let head_len = REQ.len() - 11;
  let mut read_buf = BytesMut::with_capacity(0);
  read_buf.reserve(8192);
  let base = read_buf.as_ptr() as usize;
  read_buf.put_slice(REQ);
  let head = read_buf.split_to(head_len).freeze();
  let body = read_buf.split_to(11).freeze();
  drop(head);
  let mut s = snap();

  // body 还在上层手里，hyper 准备下一次读
  read_buf.reserve(8192);
  let d = delta(s);
  s = report("reserve(8192)", d, format!("{} | {}", mut_state("read_buf", &read_buf, base), bytes_state("body", &body, base)));

  drop(body);
  let d = delta(s);
  report("drop(body)", d, String::new());
}

动脑筋答案