Rust in Everything
基于 bytes 1.12.1

unsplit 之后前后两段对调了

主线位置:构建主线(BytesMut)上 split 的反方向。split_to / split_off 把一块分配切成几个可写句柄,unsplit 想把它们并回去;并不回去时,它不报错,而是退回拷贝。

flowchart TB
  u["self.unsplit(other)"] --> e{"self 为空"}
  e -->|是| rep["换成 other"]
  e -->|否| t["try_unsplit"]
  t --> c{"self 末尾接上 other"}
  c -->|是| ok["只改 len 和 cap"]
  c -->|否| cp["extend_from_slice"]
  cp --> app["other 追加在后面"]

unsplit 和 try_unsplit 的四个合并条件、各种不相邻情况的分配次数,在 split 与 freeze 里逐条跑过。这一页把会让结果出乎意料的几种用法放在一起:顺序反了、切点越过长度、切出去的那段扩过容、拿 try_unsplit 的 Ok 当证据,以及空缓冲上的 split 也会分配。

本页实验在 M2 Max、rustc 1.99.0、--release 下运行,依赖:

bytes = "=1.12.1"

复现

common.rs 是一个包住 System 的计数分配器,count 返回闭包执行期间的 alloc、realloc、free 次数和新申请的字节数。

use bytes::{BufMut, BytesMut};

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

fn fresh() -> BytesMut {
  let mut buf = BytesMut::with_capacity(64);
  buf.put_slice(b"aaabbbcccddd");
  buf
}

fn show(name: &str, buf: &BytesMut, d: Delta) {
  println!("{name:<28} {:?} len={} cap={}  {d}", std::str::from_utf8(buf).unwrap(), buf.len(), buf.capacity());
}

fn main() {
  // 1. 顺序对:前一段 unsplit 后一段
  let mut head = fresh();
  let tail = head.split_off(6);
  let (_, d) = count(|| head.unsplit(tail));
  show("head.unsplit(tail)", &head, d);

  // 2. 顺序反了:后一段 unsplit 前一段
  let mut tail = fresh();
  let head = tail.split_to(6);
  let (_, d) = count(|| tail.unsplit(head));
  show("tail.unsplit(head)", &tail, d);

  // 3. 顺序反了,用 try_unsplit:失败交回来,由调用方决定
  let mut tail = fresh();
  let head = tail.split_to(6);
  let (r, d) = count(|| tail.try_unsplit(head));
  match r {
    Ok(()) => show("tail.try_unsplit(head) Ok", &tail, d),
    Err(mut head) => {
      let (r2, d2) = count(|| head.try_unsplit(tail));
      println!("{:<28} Err {d}; swap -> {:?}", "tail.try_unsplit(head)", r2.is_ok());
      show("  head.try_unsplit(tail)", &head, d2);
    }
  }

  // 4. split_off 的位置超过 len,再并回来:后面那段容量丢了
  let mut buf = BytesMut::with_capacity(64);
  buf.put_slice(b"hello");
  let tail = buf.split_off(16);
  let (_, d) = count(|| buf.unsplit(tail));
  show("split_off(16) past len=5", &buf, d);

  // 5. 切出去的那段自己扩容过(#532):不再相邻,退回拷贝
  let mut head = fresh();
  let mut tail = head.split_off(head.len() - 1); // tail: len=1 cap=53
  tail.put_bytes(b'z', 60); // 超过 53,tail reserve 到新的一块
  let (_, d) = count(|| head.unsplit(tail));
  println!("{:<28} len={} cap={}  {d}", "tail grew, then unsplit", head.len(), head.capacity());

  // 6. try_unsplit 对容量为 0 的 other 直接 Ok,不管它从哪来
  let mut buf = fresh();
  let (r, d) = count(|| buf.try_unsplit(BytesMut::new()));
  println!("{:<28} {:?}  {d}", "try_unsplit(BytesMut::new())", r.map_err(|_| ()));

  // 7. 对空的 BytesMut split,也会提升,分配 40 字节的 Shared
  let mut empty = BytesMut::new();
  let (part, d) = count(|| empty.split());
  println!("{:<28} part.len={} part.cap={}  {d}", "BytesMut::new().split()", part.len(), part.capacity());
  let mut buf = fresh();
  let (part, d) = count(|| buf.split_to(0));
  println!("{:<28} part.len={} part.cap={}  {d}", "split_to(0)", part.len(), part.capacity());
}

跑 3 次,输出相同:

head.unsplit(tail)           "aaabbbcccddd" len=12 cap=64  alloc+0 realloc+0 (0B) free+0
tail.unsplit(head)           "cccdddaaabbb" len=12 cap=58  alloc+0 realloc+0 (0B) free+0
tail.try_unsplit(head)       Err alloc+0 realloc+0 (0B) free+0; swap -> true
  head.try_unsplit(tail)     "aaabbbcccddd" len=12 cap=64  alloc+0 realloc+0 (0B) free+0
split_off(16) past len=5     "hello" len=5 cap=16  alloc+0 realloc+0 (0B) free+0
tail grew, then unsplit      len=72 cap=128  alloc+0 realloc+1 (128B) free+1
try_unsplit(BytesMut::new()) Ok(())  alloc+0 realloc+0 (0B) free+0
BytesMut::new().split()      part.len=0 part.cap=0  alloc+1 realloc+0 (40B) free+0
split_to(0)                  part.len=0 part.cap=0  alloc+1 realloc+0 (40B) free+0

前两行是本页的标题:同样两段,head.unsplit(tail) 得到 aaabbbcccddd,tail.unsplit(head) 得到 cccdddaaabbb。第二行没有 panic、没有分配,长度也对,只是内容顺序错了;后面接着解析的代码读到的就是错的数据。

为什么

unsplit 先试 try_unsplit,失败就把 other 的字节追加到 self 后面:

    pub fn unsplit(&mut self, other: BytesMut) {
        if self.is_empty() {
            *self = other;
            return;
        }

        if let Err(other) = self.try_unsplit(other) {
            self.extend_from_slice(other.as_ref());
        }
    }

try_unsplit 能合并的第一个条件是 self.ptr + self.len == other.ptr:self 的数据末尾正好是 other 的开头。tail.unsplit(head) 里 tail 在后、head 在前,这个等式不可能成立,于是走 extend_from_slice,head 的 aaabbb 被拷到 tail 的 cccddd 后面。tail 后面还有 52 字节空闲,所以连分配都没有,cap 从 64 变成 58 是因为 tail 本来就从第 6 字节开始。unsplit 的文档说的是「otherwise appends its bytes to this BytesMut」:「追加」不看两段原来谁在前。

其余几行也都是这个条件不成立:

  • split_off(16) 越过 len(第 5 行):buf 是 len=5 cap=16,数据末尾在 5,tail 从 16 开始,中间隔着 11 字节没写的容量。合并失败,追加 0 字节,tail 被 drop,它的 48 字节容量从 buf 的视角消失了,cap 停在 16。内存没有释放(free+0),只是没有句柄再看得到。split 与 freeze 的动脑筋是同一个情况。
  • tail 扩过容(第 6 行):tail 写 61 字节超过了自己的 53,reserve 时 head 还活着,只能另起一块新的 Vec,tail 从此是独占新块的 KIND_VEC(reserve)。两段不再相邻,unsplit 退回拷贝:head 此时是那块旧分配的唯一句柄,reserve(61) 把它扩到 128(一次 realloc),再拷 61 字节;free+1 是 tail 那块新 Vec 随 tail 一起释放。
📝NOTE
冷知识:「没有被重新分配过」这半句是 1.2.0 才补进文档的

issue #532(「Unclear BytesMut::unsplit documentation」)引用当时的文档「If the two BytesMut objects were previously contiguous, i.e., if other was created by calling split_off on this BytesMut, then this is an O(1) operation」,说「I don't think this is true. It looks like other must also not be mutated in a way that requires an allocation.」复现正是上面第 6 行的形状:split_off(buf.len() - 1) 切出一段,往里写一个 u64 让它扩容,再 unsplit,在调试器里看到调用了 extend_from_slice。

PR #535(「Clarify BytesMut::unsplit docs」,首次发布于 1.2.0)只改文档,在条件里加了半句:「If the two BytesMut objects were previously contiguous and not mutated in a way that causes re-allocation」。代码没有变。1.12.0 公开 try_unsplit 时(#746),它的文档用的还是 #535 之前的那句「If the two BytesMut objects were previously contiguous, i.e., if other was created by calling split_off on this BytesMut, then this is an O(1) operation」,1.12.1 里仍是这样;两个函数的合并条件其实是同一个。

修法

  • 永远用前一段去 unsplit 后一段。 split_off 之后是 buf.unsplit(tail);split_to 之后是 head.unsplit(buf),不是 buf.unsplit(head)。
  • 想知道有没有合并,用 try_unsplit。 它失败时不拷贝,把 other 原样放进 Err 交回来(第 3、4 行:0 分配,交回的 head 换个方向再试就合上了)。在「一定应该相邻」的代码里,把 Err 当成 bug 处理,比让 unsplit 静默拷贝更容易发现问题。
  • 要并回容量,切点就别越过 len。 split_off(at) 的 at 可以落在没写的容量里(split 与 freeze 的边界检查表),但这样切出去的容量并不回来。
  • 切出去打算并回来的那段,不要写到超过它的容量。 一扩容就换了块,unsplit 只能拷贝。

另外两个意外

try_unsplit 的 Ok 不说明来自同一块缓冲

第 7 行:一个和 buf 毫无关系的 BytesMut::new(),try_unsplit 也返回 Ok。它的第一行就是:

        if other.capacity() == 0 {
            return Ok(());
        }

容量为 0 的句柄没有字节,并不并都一样,所以直接算成功。Ok 的意思是「other 已经被吸收,或者本来就没东西」。如果你用 try_unsplit(x).is_ok() 来判断 x 是不是从 self 上切下来的,容量为 0 的 x 会让判断出错。先排除 x.capacity() == 0,剩下的 Ok 才是真的合并:那时四个条件都查过,包括 self.data == other.data(同一个 Shared)。只比指针是否首尾相接不够,两块不同的分配也可能恰好相邻。

空缓冲上的 split 也分配

最后两行:BytesMut::new().split() 切出来的东西长度和容量都是 0,split_to(0) 也一样,照样各分配了 40 字节。split / split_to / split_off 不看切出来的是不是空的,一律先 shallow_clone,KIND_VEC 就提升成 KIND_ARC,分配一个 Shared(split 与 freeze、data 字段的编码)。Bytes::split_to 在切出空段时有捷径、不碰计数,BytesMut 没有。

只发生在每块缓冲第一次 split 时,之后已经是 KIND_ARC,只加计数。在意这一次分配的话,split 之前先判断 if !buf.is_empty()。

怎么确认自己没踩

  • 搜 unsplit(:调用者是不是内存里靠前的那一段?如果两个句柄的先后不是一眼能看出来的,换成 try_unsplit 并处理 Err。
  • 测试里断言合并后的内容,而不只断言长度:顺序反了的时候长度是对的。
  • 想确认没有发生拷贝,用计数分配器量 unsplit 那一步,或者比较合并前后 as_ptr() 是否不变、capacity() 是否回到切之前的值。

小结

  • unsplit 只在 self 的数据末尾正好接上 other 时 O(1) 合并;否则不报错,把 other 追加到 self 后面。
  • 用后一段去 unsplit 前一段,内容静默变成 cccdddaaabbb,长度不变,常常也不分配。
  • split_off 越过 len 切出去的容量并不回来;切出去的那段扩过容,就只能拷贝。
  • try_unsplit 失败时不拷贝、把 other 还回来,适合「必须相邻」的代码;它对容量为 0 的 other 恒为 Ok,不能拿来证明同源。
  • BytesMut 的 split 没有空结果的捷径,new().split()、split_to(0) 第一次也会分配 40 字节的 Shared。

💬 关联讨论

发起讨论
加载中...