一个小切片让整块缓冲一直不释放

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

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

fn show(step: &str, base: usize) {
  println!("{step:<44} live = {:>8} bytes", live() - base);
}

fn main() {
  println!("(stdout warmed up)");
  let base = live();

  let big = Bytes::from(vec![b'x'; 1 << 20]);
  let small = big.slice(100..110);
  drop(big);
  show("Vec -> Bytes, slice(100..110), drop(big)", base);
  println!("small.len() = {}, small.is_unique() = {}", small.len(), small.is_unique());
  let copy = Bytes::copy_from_slice(&small);
  drop(small);
  show("copy_from_slice, drop(small)", base);
  drop(copy);

  // clear() on the kept slice
  let mut small = Bytes::from(vec![b'x'; 1 << 20]).slice(100..110);
  small.clear();
  show("Vec -> Bytes slice, clear()", base);
  drop(small);

  let mut m = BytesMut::with_capacity(1 << 20);
  m.put_bytes(b'x', 1 << 20);
  let mut small = m.freeze().slice(100..110);
  show("BytesMut -> freeze, slice(100..110)", base);
  small.clear();
  show("  then clear()", base);
  small = Bytes::new();
  show("  then assign Bytes::new()", base);
  drop(small);
}

(stdout warmed up)
Vec -> Bytes, slice(100..110), drop(big)     live =  1048600 bytes
small.len() = 10, small.is_unique() = true
copy_from_slice, drop(small)                 live =       10 bytes
Vec -> Bytes slice, clear()                  live =  1048600 bytes
BytesMut -> freeze, slice(100..110)          live =  1048600 bytes
  then clear()                               live =  1048600 bytes
  then assign Bytes::new()                   live =        0 bytes

        let slice = buf.split_to(len).freeze();
        // ...
            let value = header_value!(slice.slice(header.value.0..header.value.1));

use std::convert::Infallible;
use std::sync::Mutex;

use bytes::Bytes;
use exp_bytes_pit1::live;
use http_body_util::Full;
use hyper::header::HeaderValue;
use hyper::{Request, Response};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};

const N: usize = 1000;
static STORE: Mutex<Vec<HeaderValue>> = Mutex::new(Vec::new());

fn request(i: usize) -> Vec<u8> {
  // about 600 bytes of request head, one 32-byte session id
  format!(
    "GET /item/{i} HTTP/1.1\r\nHost: example.com\r\nx-session-id: {i:032x}\r\n\
     user-agent: {}\r\naccept: */*\r\n\r\n",
    "m".repeat(480)
  )
  .into_bytes()
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
  let mode = std::env::args().nth(1).unwrap_or_default();
  STORE.lock().unwrap().reserve_exact(N);
  let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
  let addr = listener.local_addr().unwrap();
  let label = mode.clone();
  let base = live();

  let server = tokio::spawn(async move {
    let (sock, _) = listener.accept().await.unwrap();
    let svc = hyper::service::service_fn(move |req: Request<hyper::body::Incoming>| {
      let mode = mode.clone();
      async move {
        let v = req.headers().get("x-session-id").unwrap();
        let kept = match mode.as_str() {
          "keep" => Some(v.clone()),
          "copy" => Some(HeaderValue::from_bytes(v.as_bytes()).unwrap()),
          _ => None,
        };
        if let Some(kept) = kept {
          STORE.lock().unwrap().push(kept);
        }
        Ok::<_, Infallible>(Response::new(Full::new(Bytes::from_static(b"ok"))))
      }
    });
    hyper::server::conn::http1::Builder::new()
      .serve_connection(hyper_util::rt::TokioIo::new(sock), svc)
      .await
      .unwrap();
  });

  let mut c = TcpStream::connect(addr).await.unwrap();
  let mut buf = vec![0u8; 4096];
  for i in 0..N {
    c.write_all(&request(i)).await.unwrap();
    let mut got = Vec::new();
    while !got.ends_with(b"\r\n\r\nok") {
      let n = c.read(&mut buf).await.unwrap();
      got.extend_from_slice(&buf[..n]);
    }
  }
  drop(c);
  server.await.unwrap();
  drop(buf);

  let kept = STORE.lock().unwrap().len();
  let extra = live() as isize - base as isize;
  println!("mode={label:<5} kept={kept:>4}  live after connection closed = {extra:>8} bytes");
}

mode=none  kept=   0  live after connection closed =     -388 bytes
mode=keep  kept=1000  live after connection closed =  8231612 bytes
mode=copy  kept=1000  live after connection closed =    31612 bytes

        if self.read_buf_remaining_mut() < next {
            self.read_buf.reserve(next);
        }

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

const N: usize = 1000;

fn request(i: usize) -> Vec<u8> {
  format!("GET /item/{i} HTTP/1.1\r\nx-session-id: {i:032x}\r\nuser-agent: {}\r\n\r\n", "m".repeat(520))
    .into_bytes()
}

// hyper-like read loop: reserve 8 KiB when spare < 8 KiB, read one request, split it off, freeze
fn run(name: &str, keep_every: usize, copy: bool) {
  let mut store: Vec<Bytes> = Vec::with_capacity(N);
  let reqs: Vec<Vec<u8>> = (0..N).map(request).collect();
  let base = live();
  let mut read_buf = BytesMut::with_capacity(0);
  for (i, req) in reqs.iter().enumerate() {
    if read_buf.capacity() - read_buf.len() < 8192 {
      read_buf.reserve(8192);
    }
    read_buf.put_slice(req); // stands in for socket.read
    let head = read_buf.split_to(read_buf.len()).freeze();
    let start = head.windows(14).position(|w| w == b"x-session-id: ").unwrap() + 14;
    let value = head.slice(start..start + 32);
    if keep_every > 0 && i % keep_every == 0 {
      store.push(if copy { Bytes::copy_from_slice(&value) } else { value });
    }
  }
  let during = live() as isize - base as isize;
  drop(read_buf);
  let after = live() as isize - base as isize;
  println!(
    "{name:<28} kept={:>4} ({:>5} data bytes)  live with read_buf = {during:>8}  after drop(read_buf) = {after:>8}",
    store.len(),
    store.iter().map(|b| b.len()).sum::<usize>()
  );
}

fn main() {
  println!("request head = {} bytes", request(0).len());
  run("keep none", 0, false);
  run("keep every slice", 1, false);
  run("keep 1 in 100 slices", 100, false);
  run("keep every copy_from_slice", 1, true);
}

request head = 606 bytes
keep none                    kept=   0 (    0 data bytes)  live with read_buf =     8232  after drop(read_buf) =        0
keep every slice             kept=1000 (32000 data bytes)  live with read_buf =  8232000  after drop(read_buf) =  8232000
keep 1 in 100 slices         kept=  10 (  320 data bytes)  live with read_buf =    90552  after drop(read_buf) =    82320
keep every copy_from_slice   kept=1000 (32000 data bytes)  live with read_buf =    40232  after drop(read_buf) =    32000

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

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

const HEAD: usize = 600;
const BODY: usize = 1 << 20;

// returns the 32-byte id kept from the head, and how many bytes stay alive with it
fn run(name: &str, split_first: bool) {
  let base = live();
  let mut buf = BytesMut::with_capacity(8192);
  buf.put_bytes(b'h', HEAD); // the head has arrived
  let mut head = None;
  if split_first {
    head = Some(buf.split_to(HEAD).freeze());
  }
  buf.reserve(BODY); // make room for a 1 MiB body
  buf.put_bytes(b'b', BODY);
  let head = head.unwrap_or_else(|| buf.split_to(HEAD).freeze());
  let id: Bytes = head.slice(10..42);
  drop(head);
  let body = buf.split().freeze();
  drop((buf, body)); // the message is done; only `id` is kept
  println!("{name:<32} id.len()={}  live = {:>8} bytes", id.len(), live() - base);
}

fn main() {
  run("reserve body, then split head", false);
  run("split head, then reserve body", true);
}

reserve body, then split head    id.len()=32  live =  1049216 bytes
split head, then reserve body    id.len()=32  live =     8232 bytes