ntex-bytes 1.15.1

//! # Data storage modes
//!
//! The goal of `bytes` is to be as efficient as possible for server networking
//! workloads. As such, `bytes` needs to handle buffers that are never shared,
//! shared on a single thread, and shared across many threads. `bytes` also needs
//! to handle both tiny buffers as well as very large buffers.
//!
//! To achieve high performance in these various situations, `Bytes` and
//! `BytesMut` use different strategies for storing the buffer depending on the
//! usage pattern.
//!
//! ## Shared vec buffer
//!
//! `BytesMut` always allocates and is backed by a `Header`. A `Header` is
//! similar to `Vec<u8>`, except that it stores the vector parameters on the heap
//! and includes a reference counter for shared buffer support. `BytesMut` owns
//! the tail of the buffer; the buffer can be modified only via `BytesMut`. The
//! head of the buffer may be owned by `Bytes`.
//!
//! ## Inlining small buffers
//!
//! The `Bytes` struct requires three pointer-sized fields. On 64-bit systems,
//! this ends up being 24 bytes, which is a significant amount of storage for
//! cases where `Bytes` is used to represent small byte strings, such as HTTP
//! header names and values.
//!
//! To avoid any allocation in these cases, `Bytes` uses the struct itself to
//! store the buffer, reserving one byte for metadata. This means that, on 64-bit
//! systems, buffers of up to 23 bytes require no allocation at all.
//!
//! The metadata byte stores a 2-bit flag indicating that the buffer is stored
//! inline, as well as 6 bits for tracking the buffer length (the return value of
//! `Bytes::len`).
//!
//! ## Static buffers
//!
//! `Bytes` can also represent a static buffer, which is created with
//! `Bytes::from_static`. No copying or allocations are required for static
//! buffers. A pointer to the `&'static [u8]`, the length, and a flag indicating
//! that the `Bytes` instance represents a static buffer are stored directly in
//! the `Bytes` struct.
//!
//! # Struct layout
//!
//! `Bytes` is a wrapper around `Storage`, which provides the data fields as well
//! as all function implementations.
//!
//! The `Storage` struct contains the following fields:
//!
//! * `ptr: *mut u8`
//! * `len: usize`
//! * `offset: usize`
//!
//! ## `ptr: *mut u8`
//!
//! A pointer to the start of the handle’s buffer view. When backed by a
//! `Header`, this pointer is shifted to point somewhere inside the buffer.
//!
//! When in inline mode, `ptr` is used as part of the inlined buffer.
//!
//! ## `len: usize`
//!
//! The length of the handle’s buffer view. When backed by a `Header`, this is
//! the length of the buffer slice. The slice represented by `ptr` and `len`
//! always points to initialized memory.
//!
//! When in inline mode, `len` is used as part of the inlined buffer.
//!
//! ## `offset: usize`
//!
//! The lower two bits of `offset` are used to track the storage mode of
//! `Storage`. `0b00` indicates external storage, `0b01` indicates shared
//! storage, `0b10` indicates inline storage, and `0b11` indicates static
//! storage.
//!
//! When storage is backed by a `Header`, the upper bits of `offset` hold the
//! address of the `Header` header, which is at least 4-byte aligned. The
//! header pointer is recovered from `ptr` with that address, and the offset of
//! the view is `ptr - header`.
//!
//! When in inline mode, `offset` is used as part of the inlined buffer.
//!
//! On little-endian platforms, the `offset` field must be the first field in the
//! struct. On big-endian platforms, the `offset` field must be the last field in
//! the struct. Since a deterministic struct layout is required, `Storage` is
//! annotated with `#[repr(C)]`.
//!
//! # Thread safety
//!
//! `Bytes::clone()` returns a new `Bytes` handle without copying. This is done by
//! incrementing the buffer’s reference count and returning a new struct pointing
//! to the same buffer.
//!
//! Care is taken to minimize the need for synchronization. Most operations do
//! not require any synchronization.
//!
#![allow(clippy::assert_is_empty)]
use std::{cmp, mem, num::NonZeroUsize, ptr, ptr::NonNull, slice, sync::atomic::Ordering::Relaxed};

use crate::{info::Info, info::Kind, stext::StorageVTable, stvec};

#[cfg(target_endian = "little")]
#[repr(C)]
pub(crate) struct Storage {
    pub(crate) offset: NonZeroUsize,
    pub(crate) ptr: *mut u8,
    pub(crate) len: usize,
}

#[cfg(target_endian = "big")]
#[repr(C)]
pub(crate) struct Storage {
    pub(crate) len: usize,
    pub(crate) ptr: *mut u8,
    pub(crate) offset: NonZeroUsize,
}

// Buffer storage strategy flags.
const KIND_STEXT: usize = 0b00;
const KIND_VEC: usize = 0b01;
const KIND_INLINE: usize = 0b10;
const KIND_STATIC: usize = 0b11;
const KIND_MASK: usize = 0b11;
const KIND_OFFSET_BITS: usize = 2;

pub(crate) const MIN_CAPACITY: usize = 128 - crate::METADATA_SIZE;

// Mask of the inline length bits in the low byte of the `offset` field.
const INLINE_LEN_MASK: usize = 0b1111_1100;

// Byte offset from the start of `Storage` to where the inline buffer data
// starts. The low byte of `offset` holds the kind and the inline length. On
// little endian platforms `offset` is the first field and its low byte is the
// first byte of the struct, so the data is shifted by a byte. On big endian
// platforms `offset` is the last field and its low byte is the last byte of the
// struct, so the data starts at the beginning of the struct.
#[cfg(target_endian = "little")]
const INLINE_DATA_OFFSET: isize = 1;
#[cfg(target_endian = "big")]
const INLINE_DATA_OFFSET: isize = 0;

// Inline buffer capacity. This is the size of `Storage` minus 1 byte for the
// metadata.
#[cfg(target_pointer_width = "64")]
pub(crate) const INLINE_CAP: usize = 3 * 8 - 1;
#[cfg(target_pointer_width = "32")]
pub(crate) const INLINE_CAP: usize = 3 * 4 - 1;

// Inline storage
const PTR_INLINE: NonZeroUsize = NonZeroUsize::new(KIND_INLINE).unwrap();
// Static storage
const PTR_STATIC: NonZeroUsize = NonZeroUsize::new(KIND_STATIC).unwrap();

// the kind bits of vec storage are stored in the low bits of the header address
const _: () = assert!(mem::align_of::<stvec::Header>() > KIND_MASK);

/// Builds inline storage in registers with fixed-size loads, instead of a
/// variable-length `memmove` into a stack value that is then read back.
#[cfg(all(target_endian = "little", target_pointer_width = "64"))]
mod inline_le64 {
    use std::{num::NonZeroUsize, ptr};

    use super::{INLINE_CAP, KIND_INLINE, KIND_OFFSET_BITS, Storage};

    const _: () = assert!(INLINE_CAP == 23 && super::INLINE_DATA_OFFSET == 1);

    #[inline]
    unsafe fn rd64(src: *const u8) -> u64 {
        src.cast::<u64>().read_unaligned()
    }

    #[inline]
    unsafe fn rd32(src: *const u8) -> u64 {
        u64::from(src.cast::<u32>().read_unaligned())
    }

    /// Loads `len` bytes as three little-endian words, zero-filled past `len`,
    /// reading only `src..src + len`.
    #[inline]
    pub(super) unsafe fn load_exact(src: *const u8, len: usize) -> [u64; 3] {
        match len {
            0 => [0, 0, 0],
            1..=3 => {
                let mid = len / 2;
                let w = u64::from(*src)
                    | u64::from(*src.add(mid)) << (mid * 8)
                    | u64::from(*src.add(len - 1)) << ((len - 1) * 8);
                [w, 0, 0]
            }
            // overlapping loads repeat identical bytes, so `|` merges them
            4..=8 => [rd32(src) | rd32(src.add(len - 4)) << ((len - 4) * 8), 0, 0],
            9..=16 => [rd64(src), rd64(src.add(len - 8)) >> ((16 - len) * 8), 0],
            _ => [
                rd64(src),
                rd64(src.add(8)),
                rd64(src.add(len - 8)) >> ((24 - len) * 8),
            ],
        }
    }

    /// Loads `len` bytes as three little-endian words, zero-filled past `len`,
    /// reading `src..src + INLINE_CAP`.
    #[inline]
    pub(super) unsafe fn load_window(src: *const u8, len: usize) -> [u64; 3] {
        [
            rd64(src) & mask(len),
            rd64(src.add(8)) & mask(len.saturating_sub(8)),
            (rd64(src.add(15)) >> 8) & mask(len.saturating_sub(16)),
        ]
    }

    /// Mask of the low `min(n, 8)` bytes.
    #[inline]
    fn mask(n: usize) -> u64 {
        ((1u128 << (n.min(8) * 8)) - 1) as u64
    }

    /// Packs the kind, the length and 23 data bytes into the `Storage` fields.
    /// Data starts at byte 1, after the kind and length byte of `offset`.
    #[inline]
    pub(super) fn build(d0: u64, d1: u64, d2: u64, len: usize) -> Storage {
        let w0 = (KIND_INLINE | (len << KIND_OFFSET_BITS)) as u64 | (d0 << 8);
        let w1 = (d0 >> 56) | (d1 << 8);
        let w2 = (d1 >> 56) | (d2 << 8);
        Storage {
            // SAFETY: the inline kind bits are set
            offset: unsafe { NonZeroUsize::new_unchecked(w0 as usize) },
            ptr: ptr::without_provenance_mut(w1 as usize),
            len: w2 as usize,
        }
    }
}

/*
 *
 * ===== Storage =====
 *
 */

impl Storage {
    #[inline]
    pub(crate) const fn empty() -> Storage {
        Storage {
            ptr: ptr::null_mut(),
            len: 0,
            offset: PTR_INLINE,
        }
    }

    #[inline]
    pub(crate) const fn from_static(bytes: &'static [u8]) -> Storage {
        let ptr = bytes.as_ptr().cast_mut();

        Storage {
            ptr,
            len: bytes.len(),
            offset: PTR_STATIC,
        }
    }

    #[inline]
    pub(crate) fn from_stext(addr: *const u8, len: usize, vtable: *const StorageVTable) -> Storage {
        Storage {
            len,
            ptr: addr.cast_mut(),
            // SAFETY: All callers supply a non-null static `StorageVTable` reference.
            offset: unsafe { NonZeroUsize::new_unchecked(vtable.expose_provenance()) },
        }
    }

    #[inline]
    pub(crate) fn from_slice(src: &[u8]) -> Storage {
        if src.len() <= INLINE_CAP {
            // SAFETY: The source is readable for `src.len()` bytes, which fits the inline capacity
            // checked above.
            unsafe { Storage::from_ptr_inline(src.as_ptr(), src.len()) }
        } else {
            Storage::from_slice_with_capacity(src.len(), src)
        }
    }

    #[inline]
    fn from_slice_with_capacity(cap: usize, src: &[u8]) -> Storage {
        let shared = stvec::Header::create(cap, src);
        // SAFETY: the data follows the `Header` header in the allocation,
        // the new allocation holds one reference
        unsafe { Storage::from_shared_vec(shared, stvec::METADATA_SIZE, src.len()) }
    }

    /// Creates a view of `len` bytes at `offset` from the start of
    /// a `Header` allocation.
    ///
    /// # Safety
    ///
    /// The caller transfers one reference of the allocation to the view,
    /// `offset..offset + len` must be initialized data within the allocation.
    #[inline]
    pub(crate) unsafe fn from_shared_vec(
        shared: NonNull<stvec::Header>,
        offset: usize,
        len: usize,
    ) -> Storage {
        Storage {
            len,
            ptr: shared.as_ptr().cast::<u8>().add(offset),
            offset: NonZeroUsize::new_unchecked(shared.as_ptr().addr() | KIND_VEC),
        }
    }

    /// Copies `len` bytes at `src` into a new inline storage.
    ///
    /// # Safety
    ///
    /// `src` must be readable for `len` initialized bytes and `len <= INLINE_CAP`.
    #[inline]
    pub(crate) unsafe fn from_ptr_inline(src: *const u8, len: usize) -> Storage {
        #[cfg(all(target_endian = "little", target_pointer_width = "64"))]
        {
            let [d0, d1, d2] = inline_le64::load_exact(src, len);
            inline_le64::build(d0, d1, d2, len)
        }
        #[cfg(not(all(target_endian = "little", target_pointer_width = "64")))]
        {
            let mut st = Storage {
                ptr: ptr::null_mut(),
                len: 0,
                offset: PTR_INLINE,
            };

            let dst = st.inline_ptr();
            ptr::copy(src, dst, len);
            st.set_inline_len(len);
            st
        }
    }

    /// Copies the first `len` of `avail` bytes at `src` into a new inline storage.
    ///
    /// Reads a whole `INLINE_CAP` window when `avail` allows it, which avoids
    /// branching on `len`.
    ///
    /// # Safety
    ///
    /// `src` must be readable for `avail` initialized bytes and
    /// `len <= min(avail, INLINE_CAP)`.
    #[inline]
    pub(crate) unsafe fn from_ptr_inline_prefix(
        src: *const u8,
        avail: usize,
        len: usize,
    ) -> Storage {
        debug_assert!(len <= avail && len <= INLINE_CAP);
        #[cfg(all(target_endian = "little", target_pointer_width = "64"))]
        if avail >= INLINE_CAP {
            let [d0, d1, d2] = inline_le64::load_window(src, len);
            return inline_le64::build(d0, d1, d2, len);
        }
        let _ = avail;
        Storage::from_ptr_inline(src, len)
    }

    /// Return a slice for the handle's view into the shared buffer
    #[inline]
    pub(crate) fn as_ref(&self) -> &[u8] {
        match self.kind() {
            // SAFETY: the inline buffer holds `inline_len` initialized bytes
            KIND_INLINE => unsafe {
                slice::from_raw_parts(self.inline_ptr_ro(), self.inline_len())
            },
            // SAFETY: `StorageExt` guarantees the vtable pointer and length
            KIND_STEXT => unsafe { slice::from_raw_parts(self.as_ptr(), self.len()) },
            // SAFETY: vec and static storage point to `len` initialized bytes
            _ => unsafe { slice::from_raw_parts(self.ptr, self.len) },
        }
    }

    /// Return a raw pointer to data
    #[inline]
    pub(crate) unsafe fn as_ptr(&self) -> *const u8 {
        match self.kind() {
            KIND_INLINE => self.inline_ptr_ro(),
            KIND_STEXT => ((*self.st_vtable()).as_ptr)(self.st_addr(), self.st_len()),
            _ => self.ptr,
        }
    }

    #[inline]
    pub(crate) fn len(&self) -> usize {
        match self.kind() {
            // SAFETY: External storage retains the live vtable and handle pair provided by
            // `StorageExt::create`.
            KIND_STEXT => unsafe { ((*self.st_vtable()).len)(self.st_addr(), self.st_len()) },
            KIND_INLINE => self.inline_len(),
            _ => self.len,
        }
    }

    pub(crate) fn get_u8(&mut self) -> u8 {
        let data = self.as_ref();
        assert!(!data.is_empty());
        let ret = data[0];
        // SAFETY: the view holds at least one byte
        unsafe { self.set_start(1) };
        ret
    }

    /// Pointer to the start of the inline buffer
    #[inline]
    unsafe fn inline_ptr(&mut self) -> *mut u8 {
        (ptr::from_mut::<Storage>(self).cast::<u8>()).offset(INLINE_DATA_OFFSET)
    }

    /// Pointer to the start of the inline buffer
    #[inline]
    unsafe fn inline_ptr_ro(&self) -> *const u8 {
        ptr::from_ref::<Storage>(self)
            .cast::<u8>()
            .offset(INLINE_DATA_OFFSET)
    }

    #[inline]
    fn inline_len(&self) -> usize {
        (self.offset.get() & INLINE_LEN_MASK) >> KIND_OFFSET_BITS
    }

    #[inline]
    pub(crate) fn is_empty(&self) -> bool {
        self.len() == 0
    }

    #[inline]
    pub(crate) fn capacity(&self) -> usize {
        let kind = self.kind();
        match kind {
            // capacity from the start of this view to the end of the allocation
            // SAFETY: The vec kind owns a reference to the header; capacity is immutable while
            // shared.
            KIND_VEC => unsafe {
                stvec::Header::capacity(self.shared_vec()) + stvec::METADATA_SIZE
                    - self.vec_offset()
            },
            // SAFETY: External storage retains the live vtable and handle pair provided by
            // `StorageExt::create`.
            KIND_STEXT => unsafe { ((*self.st_vtable()).len)(self.st_addr(), self.st_len()) },
            KIND_INLINE => INLINE_CAP,
            _ => self.len,
        }
    }

    #[inline]
    pub(crate) fn split_off(&mut self, at: usize, create_inline: bool) -> Storage {
        // SAFETY: Callers ensure `at <= len`; copied tails fit inline and shared tails retain a
        // reference.
        let other = unsafe {
            if create_inline && self.len() - at <= INLINE_CAP {
                Storage::from_ptr_inline(self.as_ptr().add(at), self.len() - at)
            } else {
                let mut other = self.shallow_clone();
                other.set_start(at);
                other
            }
        };
        // SAFETY: The prefix ends at the caller-checked `at`; an inline copy is made only when it
        // fits.
        unsafe {
            if create_inline && at <= INLINE_CAP {
                *self = Storage::from_ptr_inline_prefix(self.as_ptr(), self.len(), at);
            } else {
                self.set_end(at);
            }
        }

        other
    }

    /// Returns a handle to the `begin..end` range of the view.
    ///
    /// `begin <= end <= self.len()` must hold.
    pub(crate) fn slice(&self, begin: usize, end: usize) -> Storage {
        debug_assert!(begin <= end && end <= self.len());
        if end - begin <= INLINE_CAP || (self.kind() == KIND_STEXT && end - begin < self.len()) {
            // external storage cannot share a range, only the range is copied
            Storage::from_slice(&self.as_ref()[begin..end])
        } else {
            // SAFETY: The checked range lies within initialized data and the cloned view retains
            // its allocation reference.
            unsafe {
                let mut other = self.shallow_clone();
                other.set_end(end);
                other.set_start(begin);
                other
            }
        }
    }

    #[inline]
    pub(crate) fn split_to(&mut self, at: usize) -> Storage {
        // SAFETY: Callers ensure `at <= len`; the initialized prefix is copied inline or retains a
        // shared reference.
        let other = unsafe {
            if at <= INLINE_CAP {
                Storage::from_ptr_inline_prefix(self.as_ptr(), self.len(), at)
            } else {
                let mut other = self.shallow_clone();
                other.set_end(at);
                other
            }
        };
        // SAFETY: The caller-checked `at` lies within the view, and the returned prefix is no
        // longer mutable here.
        unsafe {
            self.set_start(at);
        }

        other
    }

    pub(crate) fn truncate(&mut self, len: usize) {
        // SAFETY: Only a prefix of initialized bytes is retained, and inline copies are bounded by
        // `INLINE_CAP`.
        unsafe {
            if len <= self.len() {
                if len < INLINE_CAP {
                    *self = Storage::from_ptr_inline_prefix(self.as_ptr(), self.len(), len);
                } else {
                    self.set_len(len);
                }
            }
        }
    }

    pub(crate) fn trimdown(&mut self) {
        let kind = self.kind();

        // inline and static buffers are left as is. Data that fits inline is
        // moved inline, otherwise the data is copied into an exact-sized
        // buffer if at least 64 bytes of capacity are unused
        if !(kind == KIND_INLINE || kind == KIND_STATIC) {
            // a shared view retains the whole allocation, including
            // the space before the start of the view
            let capacity = if kind == KIND_VEC {
                // SAFETY: The vec kind owns a reference to the live header; capacity is immutable
                // while shared.
                unsafe { stvec::Header::capacity(self.shared_vec()) }
            } else {
                self.capacity()
            };

            if self.len() <= INLINE_CAP {
                // SAFETY: The initialized view fits inline, and the old storage stays alive until
                // the copy completes.
                *self = unsafe { Storage::from_ptr_inline(self.as_ptr(), self.len()) };
            } else if capacity - self.len() >= 64 {
                // the copy is not attributed to a tracker
                *self = Storage::from_slice_with_capacity(self.len(), self.as_ref());
            }
        }
    }

    #[inline]
    pub(crate) unsafe fn set_len(&mut self, len: usize) {
        let kind = self.kind();
        match kind {
            KIND_VEC => {
                // other handles may use the memory past the end of this view
                assert!(len <= self.len);
                self.len = len;
            }
            KIND_STEXT => {
                if len != self.len() {
                    assert!(len <= self.capacity());
                    *self = Storage::from_slice(&self.as_ref()[..len]);
                }
            }
            KIND_INLINE => self.set_inline_len(len),
            _ => {
                assert!(len <= self.len);
                self.len = len;
            }
        }
    }

    /// Set the length of the inline buffer. This is done by writing to the
    /// inline length bits in the least significant byte of the `offset` field.
    #[inline]
    fn set_inline_len(&mut self, len: usize) {
        debug_assert!(len <= INLINE_CAP);
        // SAFETY: Inline kind bits remain set, so replacing only the length bits cannot produce
        // zero.
        self.offset = unsafe {
            NonZeroUsize::new_unchecked(
                self.offset.get() & !INLINE_LEN_MASK | (len << KIND_OFFSET_BITS),
            )
        };
    }

    pub(crate) unsafe fn set_start(&mut self, start: usize) {
        // Setting the start to 0 is a no-op, so return early if this is the
        // case.
        if start == 0 {
            return;
        }

        assert!(
            start <= self.len(),
            "cannot advance past the end of the buffer, cnt:{start} len:{}",
            self.len()
        );

        match self.kind() {
            KIND_VEC => {
                // the header address in `offset` stays the same
                self.ptr = self.ptr.add(start);
                self.len -= start;
            }
            KIND_INLINE => {
                let len = self.inline_len();
                if len <= start {
                    self.set_inline_len(0);
                } else {
                    // `set_start` is essentially shifting data off the front of the
                    // view. Inlined buffers only track the length of the slice.
                    // So, to update the start, the data at the new starting point
                    // is copied to the beginning of the buffer.
                    let new_len = len - start;

                    let dst = self.inline_ptr();
                    let src = (dst.cast_const()).add(start);

                    ptr::copy(src, dst, new_len);

                    self.set_inline_len(new_len);
                }
            }
            KIND_STEXT => {
                *self = Storage::from_slice(&self.as_ref()[start..]);
            }
            _ => {
                // advance the start of static storage
                self.len -= start;
                self.ptr = self.ptr.add(start);
            }
        }
    }

    pub(crate) unsafe fn set_end(&mut self, end: usize) {
        match self.kind() {
            KIND_VEC => {
                self.len = cmp::min(self.len, end);
            }
            KIND_INLINE => {
                assert!(end <= INLINE_CAP);
                let new_len = cmp::min(self.inline_len(), end);
                self.set_inline_len(new_len);
            }
            KIND_STEXT => {
                if end < self.len() {
                    *self = Storage::from_slice(&self.as_ref()[..end]);
                }
            }
            _ => {
                // set len for static storage
                assert!(end <= self.len);
                self.len = end;
            }
        }
    }

    #[inline]
    /// Increments the ref count. This should only be done if it is known that
    /// it can be done safely. As such, this fn is not public, instead other
    /// fns will use this one while maintaining the guarantees.
    ///
    /// "Safely" is defined as not exposing two `BytesMut` values that point to
    /// the same byte window.
    ///
    /// This function is thread safe.
    unsafe fn shallow_clone(&self) -> Storage {
        // Check the kind first, if the handle is using inline data storage,
        // the `ptr` and `len` fields hold data and must not be dereferenced.
        //
        // Inline and static storage own no shared state, a bitwise copy is
        // a valid clone. The kind of a handle never changes while it is
        // borrowed, so branching on it is safe.
        let kind = self.kind();

        if kind == KIND_INLINE || kind == KIND_STATIC {
            // In this case, a shallow_clone still involves copying the data.
            Storage { ..*self }
        } else if kind == KIND_VEC {
            // ptr points to Header
            let shared = self.shared_vec();
            let ref_cnt = (*shared).ref_count.fetch_add(1, Relaxed);
            if ref_cnt == u32::MAX {
                stvec::abort();
            }

            Storage { ..*self }
        } else {
            // ext storage
            if let Some((addr, len)) = ((*self.st_vtable()).clone)(self.st_addr(), self.st_len()) {
                Storage {
                    len,
                    ptr: addr.cast_mut(),
                    offset: self.offset,
                }
            } else {
                Storage::from_slice(self.as_ref())
            }
        }
    }

    /// Returns true if the buffer is stored inline
    #[inline]
    pub(crate) fn is_inline(&self) -> bool {
        self.kind() == KIND_INLINE
    }

    /// Converts into a `StorageVec` without copying if this is the only
    /// reference to a shared vec.
    pub(crate) fn try_into_vec(self) -> Result<stvec::StorageVec, Storage> {
        if self.kind() == KIND_VEC {
            let offset = self.vec_offset();
            let st =
                // SAFETY: This vec view owns a reference and its bounds are initialized; success
                // transfers that reference.
                unsafe { stvec::StorageVec::from_unique_view(self.shared_vec(), offset, self.len) };
            if let Some(st) = st {
                mem::forget(self);
                return Ok(st);
            }
        }
        Err(self)
    }

    #[inline]
    fn shared_vec(&self) -> *mut stvec::Header {
        // `ptr` is derived from the header pointer and keeps the provenance
        // of the whole allocation
        #[allow(clippy::cast_ptr_alignment)]
        self.ptr
            .with_addr(self.offset.get() & !KIND_MASK)
            .cast::<stvec::Header>()
    }

    /// Offset of the view from the start of the `Header` allocation.
    #[inline]
    fn vec_offset(&self) -> usize {
        self.ptr.addr() - (self.offset.get() & !KIND_MASK)
    }

    #[inline]
    fn st_vtable(&self) -> *const StorageVTable {
        ptr::with_exposed_provenance::<StorageVTable>(self.offset.get())
    }

    #[inline]
    fn st_addr(&self) -> *const u8 {
        self.ptr.cast_const()
    }

    #[inline]
    fn st_len(&self) -> usize {
        self.len
    }

    #[inline]
    fn kind(&self) -> usize {
        // The kind is stored in the low bits of `offset` for every storage kind,
        // on both little- and big-endian targets.
        self.offset.get() & KIND_MASK
    }

    pub(crate) fn info(&self) -> Info {
        let kind = self.kind();

        // SAFETY: The vec-kind check guards header access; the handle keeps it alive and only
        // shared-safe fields are read.
        let (id, refs, capacity) = unsafe {
            if kind == KIND_VEC {
                // the other header fields belong to the `BytesMut` handle
                // and can be modified concurrently
                let ptr = self.shared_vec();
                (
                    ptr as usize,
                    (*ptr).ref_count.load(Relaxed),
                    stvec::Header::capacity(ptr) + stvec::METADATA_SIZE,
                )
            } else {
                (0, 0, 0)
            }
        };

        Info {
            id,
            refs,
            capacity,
            kind: Kind::from_raw(kind),
        }
    }
}

// SAFETY: Views are immutable, shared vec ownership uses atomics, and external storage requires
// `Send + Sync`.
unsafe impl Send for Storage {}
// SAFETY: Shared views only expose immutable bytes; vec refcounts are atomic and external storage
// requires `Sync`.
unsafe impl Sync for Storage {}

impl Clone for Storage {
    #[inline]
    fn clone(&self) -> Storage {
        // SAFETY: Cloning preserves immutable views and retains their backing storage through its
        // reference-count or vtable.
        unsafe { self.shallow_clone() }
    }
}

impl Drop for Storage {
    #[inline]
    fn drop(&mut self) {
        match self.kind() {
            KIND_VEC => {
                stvec::release_shared_vec(self.shared_vec());
            }
            // SAFETY: This handle owns the external pair, which its live vtable releases exactly
            // once.
            KIND_STEXT => unsafe {
                ((*self.st_vtable()).drop)(self.st_addr(), self.st_len());
            },
            _ => {}
        }
    }
}

impl Kind {
    fn from_raw(n: usize) -> Kind {
        match n {
            KIND_STEXT => Kind::StExt,
            KIND_INLINE => Kind::Inline,
            KIND_STATIC => Kind::Static,
            _ => Kind::Vec,
        }
    }
}

#[cfg(test)]
mod tests {
    use crate::*;
    use std::ptr;

    use super::{KIND_INLINE, Storage};

    const LONG: &[u8] =
        b"mary had a little lamb, little lamb, little lamb, little lamb, little lamb, little lamb \
        mary had a little lamb, little lamb, little lamb, little lamb, little lamb, little lamb \
        mary had a little lamb, little lamb, little lamb, little lamb, little lamb, little lamb";

    /// STATEMENT: Large vector-backed views retain valid offsets and shared storage on 32-bit targets.
    ///
    /// HOW:
    /// - Split a `Bytes` allocation after one GiB, check lengths, storage identity and tail capacity, then clone and drop the tail.
    #[test]
    #[cfg(target_pointer_width = "32")]
    fn vec_view_offset_over_1gib() {
        const GIB: usize = 1 << 30;

        let mut buf = BytesMut::with_capacity(GIB + 256);
        unsafe { buf.advance_mut(GIB + 128) };
        let mut b = buf.freeze();
        let head = b.split_to(GIB);
        assert_eq!(head.len(), GIB);
        assert_eq!(b.len(), 128);
        assert_eq!(b.storage.info().id, head.storage.info().id);
        assert_eq!(
            b.storage.capacity(),
            b.storage.info().capacity - stvec::METADATA_SIZE - GIB
        );
        drop(head);
        let b2 = b.clone();
        drop(b);
        drop(b2);
    }

    /// STATEMENT: `trimdown` shrinks storage or makes short data inline without changing bytes.
    ///
    /// HOW:
    /// - Shorten `LONG`, trim to 68 bytes and `INLINE_CAP`, and check contents and storage size.
    /// - Trim a 40-byte tail view and verify its exact capacity and independent allocation.
    #[test]
    fn trimdown() {
        let mut b = Bytes::from(LONG.to_vec());
        assert_eq!(b.storage.capacity(), 263);
        // SAFETY: The initialized `LONG` view contains the 68-byte prefix being retained.
        unsafe { b.storage.set_len(68) };
        assert_eq!(b.len(), 68);
        assert_eq!(&b[..], &LONG[..68]);
        assert_eq!(b.storage.capacity(), 263);
        b.trimdown();
        assert_eq!(&b[..], &LONG[..68]);
        assert_eq!(b.storage.capacity(), 68);

        // SAFETY: The initialized 68-byte view contains the `INLINE_CAP`-byte prefix being
        // retained.
        unsafe { b.storage.set_len(super::INLINE_CAP) };
        assert_eq!(&b[..], &LONG[..super::INLINE_CAP]);
        b.trimdown();
        assert!(b.is_inline());

        // a view at the end of a larger allocation
        let src = Bytes::from(LONG.to_vec());
        let mut b = src.slice(LONG.len() - 40..);
        assert!(b.storage.capacity() - b.len() < 64);
        b.trimdown();
        assert_eq!(&b[..], &LONG[LONG.len() - 40..]);
        assert_eq!(b.storage.capacity(), 40);
        assert!(!src.as_ref().as_ptr_range().contains(&b.as_ptr()));
    }

    /// STATEMENT: Mutable storage preserves data through growth, advancement, and allocation reuse.
    ///
    /// HOW:
    /// - Copy empty and nonempty slices, reserve exact and amortized capacity, and check lengths and appended bytes.
    /// - Convert to `Bytes`, then consume and reserve a 48-byte buffer and verify its pointer is reused.
    #[test]
    #[allow(clippy::unnecessary_fallible_conversions)]
    fn bytes_mut() {
        let bv = BytesMut::copy_from_slice(LONG);
        assert_eq!(bv.capacity(), 263);
        assert_eq!(bv.len(), 263);
        assert_eq!(bv.as_ref().len(), 263);
        assert_eq!(bv.as_ref(), LONG);
        assert_eq!(&bv[..], LONG);

        let sl: &[u8] = &[];
        let bv = BytesMut::copy_from_slice(sl);
        assert_eq!(bv.capacity(), 0);
        assert_eq!(bv.len(), 0);
        assert_eq!(bv.as_ref().len(), 0);
        assert_eq!(bv.as_ref(), sl);
        assert_eq!(&bv[..], sl);

        let mut bv = BytesMut::copy_from_slice(&b"hello"[..]);
        assert_eq!(bv.capacity(), 5);
        bv.reserve_exact(123);
        assert_eq!(bv.capacity(), 128);
        assert_eq!(bv.len(), 5);
        assert_eq!(bv.as_ref(), &b"hello"[..]);

        let mut bv = BytesMut::copy_from_slice(&b"hello"[..]);
        assert_eq!(bv.capacity(), 5);
        assert_eq!(bv.len(), 5);
        assert_eq!(bv.as_ref().len(), 5);
        assert_eq!(bv.as_ref()[0], b"h"[0]);
        assert_eq!(bv.remaining_mut(), 0);
        // growth is at least twice the length
        bv.reserve(1);
        assert_eq!(bv.capacity(), 10);
        assert_eq!(bv.remaining_mut(), 5);
        bv.put_u8(b" "[0]);
        assert_eq!(bv.as_ref(), &b"hello "[..]);
        assert_eq!(bv.remaining_mut(), 4);
        bv.reserve(5);
        assert_eq!(bv.capacity(), 12);
        assert_eq!(bv.remaining_mut(), 6);
        bv.put_slice(b"world");
        assert_eq!(bv, "hello world");
        bv.advance_to(6);
        assert_eq!(bv, "world");
        assert_eq!(bv.remaining_mut(), 1);

        let bv = BytesMut::copy_from_slice(&b"hello world"[..]);
        let b = Bytes::from(bv);
        assert_eq!(b, "hello world");

        // does not re-alloc
        let mut bv = BytesMut::with_capacity(0);
        bv.extend_from_slice(b"hello world.");
        bv.extend_from_slice(b"hello world.");
        bv.extend_from_slice(b"hello world.");
        bv.extend_from_slice(b"hello world.");
        // SAFETY: The live buffer owns the allocation; its pointer is only recorded, not
        // dereferenced.
        let p1 = unsafe { bv.storage.as_ptr() as usize };

        bv.advance(48);
        assert!(bv.is_empty());
        assert_eq!(bv.capacity(), 0);
        bv.reserve(48);
        assert!(bv.is_empty());
        assert_eq!(bv.capacity(), 48);
        // SAFETY: The live buffer owns the allocation; its pointer is only recorded, not
        // dereferenced.
        let p2 = unsafe { bv.storage.as_ptr() as usize };
        assert_eq!(p1, p2);
    }

    /// STATEMENT: External storage is shared for full views but copied for partial views and truncation.
    ///
    /// HOW:
    /// - Wrap `LONG` in `Arc<str>` and verify no-op trimming, truncation, and advancement retain `StExt`.
    /// - Slice full, tail, and middle ranges and truncate to 100 bytes, checking contents, kinds, and capacities.
    #[test]
    fn ext_storage() {
        let long: std::sync::Arc<str> = std::str::from_utf8(LONG).unwrap().into();
        let mut b = Bytes::from_ext(long);
        assert_eq!(b.info().kind, info::Kind::StExt);
        assert_eq!(b.storage.capacity(), LONG.len());

        // trimdown and truncating past the end keep the external storage
        b.trimdown();
        b.truncate(LONG.len() + 10);
        b.advance(0);
        assert_eq!(b.info().kind, info::Kind::StExt);
        assert_eq!(&b[..], LONG);

        // a full slice shares the external storage, a partial one copies
        // only its range
        let full = b.slice(..);
        assert_eq!(full.info().kind, info::Kind::StExt);
        assert_eq!(&full[..], LONG);
        let len = LONG.len();
        let tail = b.slice(len - 40..);
        assert_eq!(tail.info().kind, info::Kind::Vec);
        assert_eq!(&tail[..], &LONG[len - 40..]);
        assert!(tail.info().capacity < 100, "{}", tail.info().capacity);
        let mid = b.slice(10..len - 10);
        assert_eq!(&mid[..], &LONG[10..len - 10]);
        assert!(
            mid.info().capacity < len + stvec::METADATA_SIZE,
            "{}",
            mid.info().capacity
        );
        assert_eq!(b.info().kind, info::Kind::StExt);

        // truncating copies the data
        b.truncate(100);
        assert_eq!(&b[..], &LONG[..100]);
        assert_eq!(b.info().kind, info::Kind::Vec);
    }

    /// STATEMENT: Truncating static bytes keeps their static storage kind.
    ///
    /// HOW:
    /// - Construct `Bytes` from static `LONG`, check capacity, truncate to 100 bytes, and check the prefix and `Static` kind.
    #[test]
    fn static_storage() {
        let mut b = Bytes::from_static(LONG);
        assert_eq!(b.info().kind, info::Kind::Static);
        assert_eq!(b.storage.capacity(), LONG.len());
        b.truncate(100);
        assert_eq!(&b[..], &LONG[..100]);
        assert_eq!(b.info().kind, info::Kind::Static);
    }

    /// STATEMENT: Inline storage cannot expose data again by extending its end after shortening.
    ///
    /// HOW:
    /// - Copy `INLINE_CAP` bytes inline, set the end to five and then back to capacity, and check only five bytes remain visible.
    #[test]
    fn inline_storage() {
        let cap = super::INLINE_CAP;
        let mut b = Bytes::copy_from_slice(&LONG[..cap]);
        assert!(b.is_inline());
        b.truncate(cap);
        assert_eq!(&b[..], &LONG[..cap]);

        // SAFETY: The inline view contains at least five initialized bytes.
        unsafe { b.storage.set_end(5) };
        assert_eq!(&b[..], &LONG[..5]);
        // SAFETY: `cap` is the inline capacity and `set_end` clamps the result to the current
        // initialized length.
        unsafe { b.storage.set_end(cap) };
        assert_eq!(&b[..], &LONG[..5]);
    }

    /// STATEMENT: Inline storage built from a pointer matches a byte-by-byte copy for every length, whether the source is exact or a wider window.
    ///
    /// HOW:
    /// - For each length up to `INLINE_CAP` and each source offset in a buffer of distinct bytes, build with `from_ptr_inline` from an exact copy and with `from_ptr_inline_prefix` from both the exact copy and the full buffer.
    /// - Compare the raw fields with a reference that copies bytes into a zeroed inline storage, and check the visible contents.
    #[test]
    fn inline_from_ptr_matches_reference() {
        fn reference(src: &[u8]) -> Storage {
            let mut st = Storage::empty();
            // SAFETY: `src` fits the inline buffer of `st`.
            unsafe { ptr::copy_nonoverlapping(src.as_ptr(), st.inline_ptr(), src.len()) };
            st.set_inline_len(src.len());
            st
        }
        fn fields(st: &Storage) -> (usize, usize, usize) {
            (st.offset.get(), st.ptr.addr(), st.len)
        }

        let buf: Vec<u8> = (1..=64).collect();
        for len in 0..=super::INLINE_CAP {
            for start in 0..=8 {
                let src = &buf[start..start + len];
                let exact = src.to_vec();
                let expected = fields(&reference(src));
                // SAFETY: each source is readable and initialized for the declared bytes.
                let built = unsafe {
                    [
                        Storage::from_ptr_inline(exact.as_ptr(), len),
                        Storage::from_ptr_inline_prefix(exact.as_ptr(), len, len),
                        Storage::from_ptr_inline_prefix(
                            buf[start..].as_ptr(),
                            buf.len() - start,
                            len,
                        ),
                    ]
                };
                for st in &built {
                    assert_eq!(st.kind(), KIND_INLINE);
                    assert_eq!(st.as_ref(), src, "len {len}, start {start}");
                    assert_eq!(fields(st), expected, "len {len}, start {start}");
                }
            }
        }
    }
}