五种存储

const STATIC_VTABLE: Vtable = Vtable {
    clone: static_clone,
    into_vec: static_to_vec,
    into_mut: static_to_mut,
    is_unique: static_is_unique,
    drop: static_drop,
};

unsafe fn static_clone(_: &AtomicPtr<()>, ptr: *const u8, len: usize) -> Bytes {
    let slice = slice::from_raw_parts(ptr, len);
    Bytes::from_static(slice)
}

unsafe fn static_to_vec(_: *mut (), ptr: *const u8, len: usize) -> Vec<u8> {
    let slice = slice::from_raw_parts(ptr, len);
    slice.to_vec()
}

unsafe fn static_to_mut(_: *mut (), ptr: *const u8, len: usize) -> BytesMut {
    let slice = slice::from_raw_parts(ptr, len);
    BytesMut::from(slice)
}

fn static_is_unique(_: &AtomicPtr<()>) -> bool {
    false
}

unsafe fn static_drop(_: *mut (), _: *const u8, _: usize) {
    // nothing to drop for &'static [u8]
}

impl From<&'static str> for Bytes {
    fn from(slice: &'static str) -> Bytes {
        Bytes::from_static(slice.as_bytes())
    }
}

impl From<Box<[u8]>> for Bytes {
    fn from(slice: Box<[u8]>) -> Bytes {
        // Box<[u8]> doesn't contain a heap allocation for empty slices,
        // so the pointer isn't aligned enough for the KIND_VEC stashing to
        // work.
        if slice.is_empty() {
            return Bytes::new();
        }

        let len = slice.len();
        let ptr = Box::into_raw(slice) as *mut u8;

        if ptr as usize & 0x1 == 0 {
            let data = ptr_map(ptr, |addr| addr | KIND_VEC);
            Bytes {
                ptr,
                len,
                data: AtomicPtr::new(data.cast()),
                vtable: &PROMOTABLE_EVEN_VTABLE,
            }
        } else {
            Bytes {
                ptr,
                len,
                data: AtomicPtr::new(ptr.cast()),
                vtable: &PROMOTABLE_ODD_VTABLE,
            }
        }
    }
}

const KIND_ARC: usize = 0b0;
const KIND_VEC: usize = 0b1;
const KIND_MASK: usize = 0b1;

const _: [(); 0 - mem::align_of::<Shared>() % 2] = []; // Assert that the alignment of `Shared` is divisible by 2.

unsafe fn promotable_even_to_vec(shared: *mut (), ptr: *const u8, len: usize) -> Vec<u8> {
    promotable_to_vec(shared, ptr, len, |shared| {
        ptr_map(shared.cast(), |addr| addr & !KIND_MASK)
    })
}

unsafe fn promotable_odd_to_vec(shared: *mut (), ptr: *const u8, len: usize) -> Vec<u8> {
    promotable_to_vec(shared, ptr, len, |shared| shared.cast())
}

        let buf = f(shared);

        let cap = ptr.offset_from(buf) as usize + len;

        // Copy back buffer
        ptr::copy(ptr, buf, len);

        Vec::from_raw_parts(buf, len, cap)

unsafe impl GlobalAlloc for Odd {
    unsafe fn alloc(&self, layout: Layout) -> *mut u8 {
        if layout.align() == 1 && layout.size() > 0 {
            // Allocate slightly bigger so that we can offset the pointer by 1
            let size = layout.size() + 1;
            let new_layout = match Layout::from_size_align(size, 1) {
                Ok(layout) => layout,
                Err(_err) => return ptr::null_mut(),
            };
            let ptr = System.alloc(new_layout);
            if !ptr.is_null() {
                ptr.offset(1)
            } else {
                ptr
            }
        } else {
            System.alloc(layout)
        }
    }
    // dealloc 对称地 offset(-1) 再释放
    // ...
}

cargo test --test test_bytes_odd_alloc -- --test-threads=1

running 8 tests
test sanity_check_odd_allocator ... ok
test test_bytes_clone_drop ... ok
test test_bytes_from_vec_drop ... ok
test test_bytes_into_vec ... ok
test test_bytesmut_from_bytes_arc_1 ... ok
test test_bytesmut_from_bytes_arc_2 ... ok
test test_bytesmut_from_bytes_arc_offset ... ok
test test_bytesmut_from_bytes_vec ... ok

test result: ok. 8 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.00s

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());
        let mut vec = ManuallyDrop::new(vec);
        let ptr = vec.as_mut_ptr();
        let len = vec.len();
        let cap = vec.capacity();

        let shared = Shared::init_to_raw(
            shared,
            Shared {
                buf: ptr,
                cap,
                ref_cnt: AtomicUsize::new(1),
            },
        );

        // ...
        Bytes {
            ptr,
            len,
            data: AtomicPtr::new(shared as _),
            vtable: &SHARED_VTABLE,
        }
    }
}

struct Shared {
    // Holds arguments to dealloc upon Drop, but otherwise doesn't use them
    buf: *mut u8,
    cap: usize,
    ref_cnt: AtomicUsize,
}

impl Drop for Shared {
    fn drop(&mut self) {
        unsafe { dealloc(self.buf, Layout::from_size_align(self.cap, 1).unwrap()) }
    }
}

#[repr(C)]
struct Owned<T> {
    ref_cnt: AtomicUsize,
    owner: T,
}

impl<T> Owned<T> {
    const VTABLE: Vtable = Vtable {
        clone: owned_clone::<T>,
        into_vec: owned_to_vec::<T>,
        into_mut: owned_to_mut::<T>,
        is_unique: owned_is_unique,
        drop: owned_drop::<T>,
    };
}

    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) }
        }
    }

bytes = "=1.12.1"

use std::alloc::{GlobalAlloc, Layout, System};
use std::mem::size_of;
use std::sync::Mutex;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering::Relaxed};

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

// ---- 计数分配器；ODD 打开时，对齐为 1 的分配返回奇数地址（同 tests/test_bytes_odd_alloc.rs）----
struct Counting;
static ALLOCS: AtomicUsize = AtomicUsize::new(0);
static ODD: AtomicBool = AtomicBool::new(false);

unsafe impl GlobalAlloc for Counting {
  unsafe fn alloc(&self, layout: Layout) -> *mut u8 {
    ALLOCS.fetch_add(1, Relaxed);
    if ODD.load(Relaxed) && layout.align() == 1 && layout.size() > 0 {
      let l = Layout::from_size_align(layout.size() + 1, 1).unwrap();
      let p = unsafe { System.alloc(l) };
      return if p.is_null() { p } else { unsafe { p.add(1) } };
    }
    unsafe { System.alloc(layout) }
  }
  unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) {
    // 奇数地址一定来自上面的分支（System 给的地址至少按 8 对齐）
    if layout.align() == 1 && layout.size() > 0 && (ptr as usize) & 1 == 1 {
      let l = Layout::from_size_align(layout.size() + 1, 1).unwrap();
      unsafe { System.dealloc(ptr.sub(1), l) }
    } else {
      unsafe { System.dealloc(ptr, layout) }
    }
  }
}

#[global_allocator]
static A: Counting = Counting;

// ---- 只用于观察：把 Bytes 当成 4 个 usize 读（依赖私有布局）----
// 字段顺序由编译器决定，所以先用 from_static 定位：ptr、len 用公开 API 对上，
// 剩下两个字里 from_static 为 0 的是 data，另一个是 vtable。
struct Slots {
  ptr: usize,
  len: usize,
  data: usize,
  vtable: usize,
}

fn words(b: &Bytes) -> [usize; 4] {
  assert_eq!(size_of::<[usize; 4]>(), size_of::<Bytes>());
  unsafe { *(b as *const Bytes as *const [usize; 4]) }
}

fn locate() -> Slots {
  let b = Bytes::from_static(b"locate me");
  let w = words(&b);
  let ptr = w.iter().position(|&x| x == b.as_ptr() as usize).unwrap();
  let len = w.iter().position(|&x| x == b.len()).unwrap();
  let data = (0..4).find(|&i| i != ptr && i != len && w[i] == 0).unwrap();
  let vtable = (0..4).find(|&i| i != ptr && i != len && i != data).unwrap();
  println!("layout: ptr=word{ptr} len=word{len} data=word{data} vtable=word{vtable}");
  Slots { ptr, len, data, vtable }
}

static SLOTS: Mutex<Option<Slots>> = Mutex::new(None);
static SEEN: Mutex<Vec<usize>> = Mutex::new(Vec::new());

fn peek(b: &Bytes) -> (char, usize) {
  let w = words(b);
  let guard = SLOTS.lock().unwrap();
  let s = guard.as_ref().unwrap();
  // 每次都校验：ptr/len 必须与公开 API 一致
  assert_eq!(w[s.ptr], b.as_ptr() as usize);
  assert_eq!(w[s.len], b.len());
  let addr = w[s.vtable];
  let data = w[s.data];
  let mut seen = SEEN.lock().unwrap();
  let idx = match seen.iter().position(|&a| a == addr) {
    Some(i) => i,
    None => {
      seen.push(addr);
      seen.len() - 1
    }
  };
  (char::from(b'A' + idx as u8), data)
}

fn describe_data(b: &Bytes, data: usize) -> String {
  if data == 0 {
    "null".into()
  } else if data == b.as_ptr() as usize {
    "=ptr".into()
  } else if data == (b.as_ptr() as usize | 1) {
    "ptr|1".into()
  } else {
    format!("other,lsb={}", data & 1)
  }
}

fn allocs<R>(f: impl FnOnce() -> R) -> (R, usize) {
  let before = ALLOCS.load(Relaxed);
  let r = f();
  (r, ALLOCS.load(Relaxed) - before)
}

fn row<T>(name: &str, prep: impl Fn() -> T, conv: impl Fn(T) -> Bytes) {
  // 只把「转成 Bytes」这一步算进 new_allocs，准备输入的分配不算
  let input = prep();
  let (b, n) = allocs(|| conv(input));
  let (vt, data) = peek(&b);
  let unique = b.is_unique();
  let (c, n_clone) = allocs(|| b.clone());
  let (cvt, _) = peek(&c);
  let (after, data_after) = peek(&b);
  drop(c);
  let ok = conv(prep()).try_into_mut().is_ok();
  println!(
    "{name:<26} vt={vt} data={:<5} new={n} unique={unique:<5} clone={n_clone} clone_vt={cvt} after: vt={after} data={:<5} try_into_mut={}",
    describe_data(&b, data),
    describe_data(&b, data_after),
    if ok { "Ok" } else { "Err" },
  );
}

fn main() {
  *SLOTS.lock().unwrap() = Some(locate());
  row("from_static", || (), |()| Bytes::from_static(b"hello"));
  row("new()", || (), |()| Bytes::new());
  row("From<&'static str>", || "hello", Bytes::from);
  row("From<Vec> len==cap", || vec![7u8; 64], Bytes::from);
  row(
    "From<Vec> len<cap",
    || {
      let mut v = Vec::with_capacity(64);
      v.extend_from_slice(b"hello");
      v
    },
    Bytes::from,
  );
  row("From<Vec> empty cap=0", Vec::new, Bytes::from);
  row("From<Box<[u8]>>", || vec![7u8; 64].into_boxed_slice(), Bytes::from);
  row("from_owner(Vec)", || vec![7u8; 64], Bytes::from_owner);
  row(
    "freeze KIND_VEC len<cap",
    || {
      let mut m = BytesMut::with_capacity(64);
      m.put_slice(b"hello");
      m
    },
    BytesMut::freeze,
  );
  row("freeze KIND_VEC len==cap", || BytesMut::from(&[7u8; 64][..]), BytesMut::freeze);
  row(
    "freeze KIND_ARC",
    || {
      let mut m = BytesMut::with_capacity(64);
      m.put_slice(b"hello world");
      drop(m.split_to(6));
      m
    },
    BytesMut::freeze,
  );

  ODD.store(true, Relaxed);
  row("ODD From<Vec> len==cap", || vec![7u8; 64], Bytes::from);
  ODD.store(false, Relaxed);

  // A（在本 crate 里内联出来的 from_static）和 D（bytes 内部 Bytes::new）的 5 个函数指针
  let seen = SEEN.lock().unwrap().clone();
  let body = |a: usize| unsafe { *(a as *const [usize; 5]) };
  let same = body(seen[0]).iter().zip(body(seen[3]).iter()).filter(|(x, y)| x == y).count();
  println!("A 与 D 的 5 个函数指针里相同的有 {same} 个");
}

layout: ptr=word1 len=word2 data=word3 vtable=word0
from_static                vt=A data=null  new=0 unique=false clone=0 clone_vt=A after: vt=A data=null  try_into_mut=Err
new()                      vt=A data=null  new=0 unique=false clone=0 clone_vt=A after: vt=A data=null  try_into_mut=Err
From<&'static str>         vt=A data=null  new=0 unique=false clone=0 clone_vt=A after: vt=A data=null  try_into_mut=Err
From<Vec> len==cap         vt=B data=ptr|1 new=0 unique=true  clone=1 clone_vt=C after: vt=B data=other,lsb=0 try_into_mut=Ok
From<Vec> len<cap          vt=C data=other,lsb=0 new=1 unique=true  clone=0 clone_vt=C after: vt=C data=other,lsb=0 try_into_mut=Ok
From<Vec> empty cap=0      vt=D data=null  new=0 unique=false clone=0 clone_vt=D after: vt=D data=null  try_into_mut=Err
From<Box<[u8]>>            vt=B data=ptr|1 new=0 unique=true  clone=1 clone_vt=C after: vt=B data=other,lsb=0 try_into_mut=Ok
from_owner(Vec)            vt=E data=other,lsb=0 new=1 unique=false clone=0 clone_vt=E after: vt=E data=other,lsb=0 try_into_mut=Err
freeze KIND_VEC len<cap    vt=C data=other,lsb=0 new=1 unique=true  clone=0 clone_vt=C after: vt=C data=other,lsb=0 try_into_mut=Ok
freeze KIND_VEC len==cap   vt=B data=ptr|1 new=0 unique=true  clone=1 clone_vt=C after: vt=B data=other,lsb=0 try_into_mut=Ok
freeze KIND_ARC            vt=F data=other,lsb=0 new=0 unique=true  clone=0 clone_vt=F after: vt=F data=other,lsb=0 try_into_mut=Ok
ODD From<Vec> len==cap     vt=G data==ptr  new=0 unique=true  clone=1 clone_vt=C after: vt=G data=other,lsb=0 try_into_mut=Ok
A 与 D 的 5 个函数指针里相同的有 2 个

Vec::new()            -> allocs=0 len=0 is_unique=false
Vec::with_capacity(10) -> allocs=1 len=0 is_unique=true