bytes 解决什么问题

bytes = "=1.12.1"

#![allow(dead_code)]
use std::alloc::{GlobalAlloc, Layout, System};
use std::sync::atomic::{AtomicUsize, Ordering::Relaxed};

use bytes::{Bytes, BytesMut};

// 计数分配器：只数次数和字节，真正的分配交给 System
struct Counting;
static ALLOCS: AtomicUsize = AtomicUsize::new(0);
static ALLOC_BYTES: AtomicUsize = AtomicUsize::new(0);
static FREES: AtomicUsize = AtomicUsize::new(0);

unsafe impl GlobalAlloc for Counting {
  unsafe fn alloc(&self, layout: Layout) -> *mut u8 {
    ALLOCS.fetch_add(1, Relaxed);
    ALLOC_BYTES.fetch_add(layout.size(), Relaxed);
    unsafe { System.alloc(layout) }
  }
  unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) {
    FREES.fetch_add(1, Relaxed);
    unsafe { System.dealloc(ptr, layout) }
  }
}

#[global_allocator]
static GLOBAL: Counting = Counting;

#[derive(Clone, Copy)]
pub struct Snap(pub usize, pub usize, pub usize);

pub fn snap() -> Snap {
  Snap(ALLOCS.load(Relaxed), ALLOC_BYTES.load(Relaxed), FREES.load(Relaxed))
}

// 这一步的分配增量。操作一结束就取，后面拼字符串、打印的分配不算进去
pub fn delta(before: Snap) -> Snap {
  let now = snap();
  Snap(now.0 - before.0, now.1 - before.1, now.2 - before.2)
}

// 打印，并返回打印之后的快照作为下一步的起点
pub fn report(step: &str, d: Snap, state: String) -> Snap {
  println!("{step:<22} alloc+{} ({}B) free+{}  {state}", d.0, d.1, d.2);
  drop(state);
  snap()
}

// 指针落在最初那块 8 KiB 里就打印偏移，否则是别的分配
fn loc(p: *const u8, base: usize) -> String {
  let p = p as usize;
  if base <= p && p <= base + 8192 { format!("base+{}", p - base) } else { "elsewhere".to_string() }
}

pub fn mut_state(name: &str, b: &BytesMut, base: usize) -> String {
  format!("{name}: ptr={} len={} cap={}", loc(b.as_ptr(), base), b.len(), b.capacity())
}

pub fn bytes_state(name: &str, b: &Bytes, base: usize) -> String {
  format!("{name}: ptr={} len={} unique={}", loc(b.as_ptr(), base), b.len(), b.is_unique())
}

use std::ops::Range;
use std::sync::Arc;

use bytes::{Bytes, BytesMut};

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

const SIZE: usize = 64 * 1024;
const FRAME: usize = 656; // 65536 / 656 向上取整 = 100 帧

// 一次「从 socket 读到的」64 KiB，长度等于容量
fn input() -> Vec<u8> {
  let mut v = Vec::with_capacity(SIZE);
  v.extend((0..SIZE).map(|i| i as u8));
  v
}

// data 是切分前的缓冲，lo 是它的起始地址；只数 split 这一段的分配
fn measure<D, T>(
  name: &str,
  data: D,
  lo: usize,
  split: impl FnOnce(D, &mut Vec<T>),
  view: impl Fn(&T) -> &[u8],
) {
  let mut frames = Vec::with_capacity(100);
  let s = snap();
  split(data, &mut frames);
  let d = delta(s);
  // 帧的字节不在原缓冲 [lo, lo + SIZE) 里，就是被拷贝过的
  let copied: usize = frames
    .iter()
    .map(&view)
    .filter(|f| !(lo..lo + SIZE).contains(&(f.as_ptr() as usize)))
    .map(|f| f.len())
    .sum();
  let n = frames.len();
  let s = snap();
  frames.clear();
  let freed = delta(s);
  println!("{name:<26}{n:>7}{:>7}{:>12}{copied:>8}{:>15}", d.0, d.1, freed.2);
}

fn main() {
  println!("{:<26}{:>7}{:>7}{:>12}{:>8}{:>15}", "way", "frames", "allocs", "alloc_bytes", "copied", "frees_on_drop");

  // 1. 每个所有者一份自己的 Vec
  let data = input();
  let lo = data.as_ptr() as usize;
  measure("Vec<u8> to_vec", data, lo, |data, out| {
    out.extend(data.chunks(FRAME).map(|c| c.to_vec()));
  }, |f: &Vec<u8>| &f[..]);

  // 2. 每帧一个 Arc<[u8]>
  let data = input();
  let lo = data.as_ptr() as usize;
  measure("Arc<[u8]> per frame", data, lo, |data, out| {
    out.extend(data.chunks(FRAME).map(Arc::<[u8]>::from));
  }, |f: &Arc<[u8]>| &f[..]);

  // 3. 整块转成一个 Arc<[u8]>，每帧共享它、自己记范围
  let data = input();
  let lo = data.as_ptr() as usize;
  measure("Arc<[u8]> + Range", data, lo, |data, out| {
    let whole: Arc<[u8]> = Arc::from(data);
    out.extend((0..SIZE).step_by(FRAME).map(|i| (whole.clone(), i..(i + FRAME).min(SIZE))));
  }, |(a, r): &(Arc<[u8]>, Range<usize>)| &a[r.clone()]);

  // 4. Vec 转成 Bytes，再一段段 split_to
  let data = input();
  let lo = data.as_ptr() as usize;
  measure("Bytes split_to", data, lo, |data, out| {
    let mut rest = Bytes::from(data);
    while !rest.is_empty() {
      out.push(rest.split_to(FRAME.min(rest.len())));
    }
  }, |f: &Bytes| &f[..]);

  // 5. 直接读进 BytesMut，split_to + freeze
  let mut buf = BytesMut::with_capacity(SIZE);
  buf.extend_from_slice(&input());
  let lo = buf.as_ptr() as usize;
  measure("BytesMut split_to+freeze", buf, lo, |mut buf, out| {
    while !buf.is_empty() {
      out.push(buf.split_to(FRAME.min(buf.len())).freeze());
    }
  }, |f: &Bytes| &f[..]);
}

way                        frames allocs alloc_bytes  copied  frees_on_drop
Vec<u8> to_vec                100    100       65536   65536            100
Arc<[u8]> per frame           100    100       67136   65536            100
Arc<[u8]> + Range             100      1       65552   65536              1
Bytes split_to                100      1          24       0              2
BytesMut split_to+freeze      100      1          40       0              2

impl From<Vec<u8>> for Bytes {
    fn from(vec: Vec<u8>) -> Bytes {
        // Avoid an extra allocation if possible.
        if vec.len() == vec.capacity() {
            return Bytes::from(vec.into_boxed_slice());
        }

        let shared = Box::new(MaybeUninit::<Shared>::uninit());
        // ...

pub struct Bytes {
    ptr: *const u8,
    len: usize,
    // inlined "trait object"
    data: AtomicPtr<()>,
    vtable: &'static Vtable,
}

pub struct BytesMut {
    ptr: NonNull<u8>,
    len: usize,
    cap: usize,
    data: *mut Shared,
}

use bytes::{Buf, BufMut, Bytes, BytesMut};

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

// 长度前缀帧：2 字节大端长度 + 内容。对任何 Buf 都能用
fn parse<B: Buf>(label: &str, mut buf: B) -> Vec<Bytes> {
  let mut out = Vec::with_capacity(4);
  let s = snap();
  while buf.remaining() >= 2 {
    let n = buf.get_u16() as usize; // 读 2 字节，游标前进 2
    out.push(buf.copy_to_bytes(n)); // 取 n 字节，游标前进 n
  }
  let d = delta(s);
  report(label, d, format!("frames={:?}", out));
  out
}

fn main() {
  let mut msg = BytesMut::with_capacity(64);
  msg.put_u16(5);
  msg.put_slice(b"hello");
  msg.put_u16(3);
  msg.put_slice(b"abc");
  let wire = msg.freeze();

  let raw: Vec<u8> = wire.to_vec();
  parse("&[u8]", &raw[..]);
  parse("Bytes", wire.clone());
}

&[u8]                  alloc+2 (8B) free+0  frames=[b"hello", b"abc"]
Bytes                  alloc+0 (0B) free+0  frames=[b"hello", b"abc"]

    fn copy_to_bytes(&mut self, len: usize) -> Self {
        self.split_to(len)
    }

cap=65536 Bytes::from  alloc+0 (0B) free+0  
split_to #1            alloc+1 (24B) free+0  
split_to #2            alloc+0 (0B) free+0  same buffer: true
cap=65537 Bytes::from  alloc+1 (24B) free+0  
split_to #1            alloc+0 (0B) free+0  
split_to #2            alloc+0 (0B) free+0  same buffer: true