try_into_mut 返回 Err

bytes = "=1.12.1"

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

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

fn report(name: &str, b: Bytes) {
  let ptr = b.as_ptr();
  let (r, d) = count(|| b.try_into_mut());
  match r {
    Ok(m) => println!("{name:<30} Ok  same_ptr={:<5} {d}", m.as_ptr() == ptr),
    Err(b) => {
      // 退路：From<Bytes> for BytesMut，唯一时复用，否则拷贝
      let (m, d2) = count(|| BytesMut::from(b));
      println!("{name:<30} Err {d} | From: same_ptr={:<5} {d2}", m.as_ptr() == ptr);
    }
  }
}

fn main() {
  // 1. 还有一个 clone 活着
  let b = Bytes::from(vec![1u8; 1024]);
  let keep = b.clone();
  report("clone alive", b);
  drop(keep);

  // 2. 自己是从别的 Bytes 切出来的，原来那个还在
  let whole = Bytes::from(vec![1u8; 1024]);
  let part = whole.slice(0..16);
  report("slice, parent alive", part);
  drop(whole);

  // 3. 从 BytesMut 切出来再 freeze，剩下的那个 BytesMut 还在（哪怕它是空的）
  let mut buf = BytesMut::with_capacity(1024);
  buf.put_bytes(1, 16);
  let frozen = buf.split().freeze();
  println!("{:<30} buf.len()={} buf.capacity()={}", "  (remaining BytesMut)", buf.len(), buf.capacity());
  report("split().freeze(), buf alive", frozen);

  // 4. 同上，但先把剩下的 BytesMut drop 掉
  let mut buf = BytesMut::with_capacity(1024);
  buf.put_bytes(1, 16);
  let frozen = buf.split().freeze();
  drop(buf);
  report("split().freeze(), buf dropped", frozen);

  // 5. 存储本身不支持：只有一个句柄也 Err
  report("from_static", Bytes::from_static(&[1u8; 1024]));
  report("from_owner(Vec)", Bytes::from_owner(vec![1u8; 1024]));
}

clone alive                    Err alloc+0 realloc+0 (0B) free+0 | From: same_ptr=false alloc+1 realloc+0 (1024B) free+0
slice, parent alive            Err alloc+0 realloc+0 (0B) free+0 | From: same_ptr=false alloc+1 realloc+0 (16B) free+0
  (remaining BytesMut)         buf.len()=0 buf.capacity()=1008
split().freeze(), buf alive    Err alloc+0 realloc+0 (0B) free+0 | From: same_ptr=false alloc+1 realloc+0 (16B) free+0
split().freeze(), buf dropped  Ok  same_ptr=true  alloc+0 realloc+0 (0B) free+0
from_static                    Err alloc+0 realloc+0 (0B) free+0 | From: same_ptr=false alloc+1 realloc+0 (1024B) free+0
from_owner(Vec)                Err alloc+0 realloc+0 (0B) free+0 | From: same_ptr=false alloc+1 realloc+0 (1024B) free+2

    pub fn try_into_mut(self) -> Result<BytesMut, Bytes> {
        if self.is_unique() {
            Ok(self.into())
        } else {
            Err(self)
        }
    }

bytes = "=1.12.1"
http-body-util = "=0.1.5"
hyper = { version = "=1.12.0", features = ["server", "http1"] }
hyper-util = { version = "=0.1.21", features = ["tokio"] }
tokio = { version = "=1.54.1", features = ["full"] }

use std::convert::Infallible;

use bytes::Bytes;
use http_body_util::{BodyExt, Full};
use hyper::body::Incoming;
use hyper::server::conn::http1;
use hyper::service::service_fn;
use hyper::{Request, Response};
use hyper_util::rt::TokioIo;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};

// 处理函数：把 body 收集成 Bytes，看能不能原地转成 BytesMut
async fn handle(req: Request<Incoming>) -> Result<Response<Full<Bytes>>, Infallible> {
  let body = req.into_body().collect().await.unwrap().to_bytes();
  let len = body.len();
  let unique = body.is_unique();
  let r = match body.try_into_mut() {
    Ok(m) => format!("Ok cap={}", m.capacity()),
    Err(_) => "Err".to_string(),
  };
  println!("body len={len:<6} is_unique={unique:<5} try_into_mut={r}");
  Ok(Response::new(Full::new(Bytes::from_static(b"ok"))))
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
  let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
  let addr = listener.local_addr().unwrap();
  tokio::spawn(async move {
    let (tcp, _) = listener.accept().await.unwrap();
    http1::Builder::new().serve_connection(TokioIo::new(tcp), service_fn(handle)).await.unwrap();
  });

  let mut c = TcpStream::connect(addr).await.unwrap();
  for n in [10usize, 1000, 100_000] {
    let head = format!("POST / HTTP/1.1\r\nhost: a\r\ncontent-length: {n}\r\n\r\n");
    c.write_all(head.as_bytes()).await.unwrap();
    c.write_all(&vec![b'x'; n]).await.unwrap();
    // 读完响应（"ok"，响应很短，一次 read 足够）
    let mut resp = [0u8; 1024];
    let _ = c.read(&mut resp).await.unwrap();
  }
}

body len=10     is_unique=true  try_into_mut=Ok cap=8144
body len=1000   is_unique=true  try_into_mut=Ok cap=8192
body len=100000 is_unique=true  try_into_mut=Ok cap=100000

use bytes::{BufMut, BytesMut};

fn main() {
  let mut buf = BytesMut::with_capacity(1024);
  buf.put_bytes(1, 16);
  let b = buf.freeze();
  println!("freeze() is_unique={} try_into_mut ok={}", b.is_unique(), b.try_into_mut().is_ok());
}

freeze() is_unique=true try_into_mut ok=true