use crate::alloc::alloc::{self, Layout, LayoutError};
use std::sync::atomic::Ordering::{Acquire, Relaxed, Release};
use std::sync::atomic::{self, AtomicU32};
use std::{cmp, mem, ptr, ptr::NonNull, slice};
use crate::storage::{INLINE_CAP, MIN_CAPACITY, Storage};
#[cfg(feature = "tracker")]
use crate::tracker::{BytesTracker, TrackerInner};
use crate::{BytePageSize, buf::UninitSlice, cache};
#[derive(Debug)]
pub(crate) struct Header {
pub(crate) offset: u32,
pub(crate) len: u32,
pub(crate) capacity: u32,
pub(crate) ref_count: AtomicU32,
#[cfg(feature = "tracker")]
pub(crate) tracker: *const TrackerInner,
#[cfg(all(feature = "tracker", target_pointer_width = "64"))]
_pad: u64,
}
#[derive(Debug)]
pub(crate) struct StorageVec(pub(crate) NonNull<Header>);
pub const METADATA_SIZE: usize = mem::size_of::<Header>();
#[cfg(not(feature = "tracker"))]
const _: () = assert!(METADATA_SIZE == 16 && mem::align_of::<Header>() == 4);
#[cfg(all(feature = "tracker", target_pointer_width = "64"))]
const _: () = assert!(METADATA_SIZE == 32 && mem::align_of::<Header>() == 8);
#[cfg(all(feature = "tracker", target_pointer_width = "32"))]
const _: () = assert!(METADATA_SIZE == 20 && mem::align_of::<Header>() == 4);
const METADATA_SIZE_U32: u32 = METADATA_SIZE as u32;
pub(crate) const MAX_CAPACITY: usize = u32::MAX as usize - METADATA_SIZE;
const MAX_MORE_STEP: usize = 64 * 1024;
impl StorageVec {
pub(crate) fn with_capacity(capacity: usize) -> StorageVec {
StorageVec(Header::create(capacity, &[]))
}
pub(crate) fn sized(size: BytePageSize) -> StorageVec {
cache::take_page(size).unwrap_or_else(|| StorageVec(Header::create(size.capacity(), &[])))
}
#[cfg(feature = "tracker")]
pub(crate) fn sized_in(size: BytePageSize, tracker: &BytesTracker) -> StorageVec {
let st = StorageVec::sized(size);
unsafe { Header::attach(st.0.as_ptr(), tracker.as_inner()) };
st
}
#[cfg(feature = "tracker")]
pub(crate) fn tracker(&self) -> Option<&BytesTracker> {
let ptr = self.0.as_ptr();
unsafe {
if Header::tracker(ptr).is_null() {
None
} else {
Some(&*ptr::addr_of!((*ptr).tracker).cast::<BytesTracker>())
}
}
}
#[cfg(feature = "tracker")]
pub(crate) fn set_tracker(&mut self, tracker: &BytesTracker) {
let ptr = self.0.as_ptr();
let new = tracker.as_inner();
unsafe {
let current = Header::tracker(ptr);
if current != new {
let size = Header::capacity(ptr) + METADATA_SIZE;
(*new).charge(size);
ptr::addr_of_mut!((*ptr).tracker).write(new);
if !current.is_null() {
TrackerInner::uncharge(current, size);
}
}
}
}
#[inline]
#[cfg_attr(not(feature = "tracker"), allow(unused_variables, clippy::unused_self))]
fn inherit_tracker(&self, st: &StorageVec) {
#[cfg(feature = "tracker")]
unsafe {
Header::attach(st.0.as_ptr(), Header::tracker(self.0.as_ptr()));
}
}
pub(crate) unsafe fn recycle(ptr: *mut Header) -> StorageVec {
unsafe {
(*ptr).len = 0;
(*ptr).offset = METADATA_SIZE_U32;
(*ptr).ref_count = AtomicU32::new(1);
#[cfg(feature = "tracker")]
{
(*ptr).tracker = ptr::null();
}
StorageVec(NonNull::new_unchecked(ptr))
}
}
pub(crate) fn from_slice(capacity: usize, src: &[u8]) -> StorageVec {
StorageVec(Header::create(capacity, src))
}
pub(crate) fn as_ref(&self) -> &[u8] {
unsafe { slice::from_raw_parts(self.as_ptr(), self.len()) }
}
pub(crate) fn as_mut(&mut self) -> &mut [u8] {
unsafe { slice::from_raw_parts_mut(self.as_ptr(), self.len()) }
}
pub(crate) unsafe fn as_raw(&mut self) -> &mut [u8] {
slice::from_raw_parts_mut(self.as_ptr(), self.capacity())
}
pub(crate) unsafe fn as_ptr(&self) -> *mut u8 {
(self.0.as_ptr().cast::<u8>()).add((*self.0.as_ptr()).offset as usize)
}
fn as_inner(&mut self) -> *mut Header {
self.0.as_ptr()
}
pub(crate) fn put_u8(&mut self, n: u8) {
let len = self.len();
unsafe {
let inner = self.as_inner();
(*inner).len += 1;
*self.as_ptr().add(len) = n;
}
}
#[inline]
pub(crate) fn spare_mut(&mut self) -> &mut UninitSlice {
unsafe { UninitSlice::from_raw_parts_mut(self.as_ptr().add(self.len()), self.remaining()) }
}
#[inline]
pub(crate) fn put_slice_partial(&mut self, src: &[u8]) -> usize {
let cnt = cmp::min(src.len(), self.remaining());
self.spare_mut()[..cnt].copy_from_slice(&src[..cnt]);
unsafe { self.set_len(self.len() + cnt) };
cnt
}
pub(crate) fn len(&self) -> usize {
unsafe { (*self.0.as_ptr()).len as usize }
}
pub(crate) fn capacity(&self) -> usize {
unsafe {
let inner = self.0.as_ref();
(inner.capacity + METADATA_SIZE_U32 - inner.offset) as usize
}
}
#[inline]
pub(crate) fn remaining(&self) -> usize {
unsafe {
let inner = self.0.as_ref();
(inner.capacity + METADATA_SIZE_U32 - inner.offset - inner.len) as usize
}
}
pub(crate) fn is_full(&self) -> bool {
self.remaining() == 0
}
pub(crate) fn is_unique(&self) -> bool {
unsafe { Header::is_unique(self.0.as_ptr()) }
}
pub(crate) unsafe fn from_unique_view(
ptr: *mut Header,
offset: usize,
len: usize,
) -> Option<StorageVec> {
if !Header::is_unique(ptr) {
return None;
}
(*ptr).offset = offset as u32;
(*ptr).len = len as u32;
Some(StorageVec(NonNull::new_unchecked(ptr)))
}
pub(crate) fn shallow_freeze(&self) -> Storage {
unsafe {
if self.len() <= INLINE_CAP {
Storage::from_ptr_inline(self.as_ptr(), self.len())
} else {
let inner = self.0.as_ref();
let ref_cnt = inner.ref_count.fetch_add(1, Relaxed);
if ref_cnt == u32::MAX {
abort();
}
Storage::from_shared_vec(self.0, inner.offset as usize, self.len())
}
}
}
pub(crate) fn freeze(self) -> Storage {
unsafe {
if self.len() <= INLINE_CAP {
Storage::from_ptr_inline(self.as_ptr(), self.len())
} else {
let offset = self.0.as_ref().offset as usize;
let inner = Storage::from_shared_vec(self.0, offset, self.len());
mem::forget(self);
inner
}
}
}
pub(crate) fn split_to(&mut self, at: usize) -> Storage {
unsafe {
let ptr = self.as_ptr();
let other = if at <= INLINE_CAP {
Storage::from_ptr_inline(ptr, at)
} else {
let inner = self.as_inner();
let ref_cnt = (*inner).ref_count.fetch_add(1, Relaxed);
if ref_cnt == u32::MAX {
abort();
}
Storage::from_shared_vec(self.0, (*inner).offset as usize, at)
};
self.set_start(at);
other
}
}
pub(crate) fn truncate(&mut self, len: usize) {
unsafe {
if len == 0 {
let inner = self.as_inner();
if Header::is_unique(inner) && (*inner).offset != METADATA_SIZE_U32 {
(*inner).len = 0;
(*inner).offset = METADATA_SIZE_U32;
return;
}
}
if len < self.len() {
self.set_len(len);
}
}
}
pub(crate) fn resize(&mut self, new_len: usize, value: u8) {
let len = self.len();
if new_len > len {
let additional = new_len - len;
self.reserve(additional);
unsafe {
let dst = self.as_raw()[len..].as_mut_ptr();
ptr::write_bytes(dst, value, additional);
self.set_len(new_len);
}
} else {
self.truncate(new_len);
}
}
#[inline]
pub(crate) fn reserve(&mut self, additional: usize) {
if additional <= self.remaining() {
return;
}
self.reserve_inner(additional, false);
}
#[inline]
pub(crate) fn reserve_exact(&mut self, additional: usize) {
if additional <= self.remaining() {
return;
}
self.reserve_inner(additional, true);
}
pub(crate) fn reserve_more(&mut self) {
let size = self.page_size();
if self.remaining() >= size.low() {
return;
}
let len = self.len();
let half = size.half_capacity();
if len + half <= unsafe { Header::capacity(self.0.as_ptr()) } {
if self.is_unique() {
self.reserve_inner(half, true);
return;
}
if size != BytePageSize::Unset {
self.move_to_page(size);
return;
}
}
let next = size.next();
if next == BytePageSize::Unset {
let cap = self.capacity();
let new_cap = cap.saturating_add(cap.clamp(MIN_CAPACITY, MAX_MORE_STEP));
self.reserve_inner(new_cap - len, true);
} else {
self.move_to_page(next);
}
}
fn move_to_page(&mut self, size: BytePageSize) {
let len = self.len();
let mut st = StorageVec::sized(size);
unsafe {
ptr::copy_nonoverlapping(self.as_ptr(), st.as_ptr(), len);
st.set_len(len);
}
self.inherit_tracker(&st);
*self = st;
}
fn reserve_inner(&mut self, additional: usize, exact: bool) {
unsafe {
let inner = self.as_inner();
let len = (*inner).len as usize;
let new_cap = len
.checked_add(additional)
.expect("buffer capacity overflow");
let grow_cap = if exact {
new_cap
} else {
grown_capacity(len, new_cap)
};
let size = self.page_size();
if Header::is_unique(inner) {
let capacity = (*inner).capacity as usize;
if capacity >= new_cap {
let offset = (*inner).offset;
(*inner).offset = METADATA_SIZE_U32;
if len != 0 {
let ptr = self.0.as_ptr().cast::<u8>();
ptr::copy(ptr.add(offset as usize), ptr.add(METADATA_SIZE), len);
}
return;
}
if exact || size == BytePageSize::Unset {
self.realloc(len, capacity, grow_cap);
return;
}
}
if !exact && size != BytePageSize::Unset {
let new_size = BytePageSize::for_capacity(grow_cap);
if new_size != BytePageSize::Unset {
let new_size = if (new_size as usize) < (size as usize) {
size
} else {
new_size
};
self.move_to_page(new_size);
return;
}
}
let st = StorageVec(Header::create(grow_cap, self.as_ref()));
self.inherit_tracker(&st);
*self = st;
}
}
pub(crate) fn page_size(&self) -> BytePageSize {
unsafe { BytePageSize::from_alloc_size(Header::capacity(self.0.as_ptr()) + METADATA_SIZE) }
}
unsafe fn realloc(&mut self, len: usize, capacity: usize, new_cap: usize) {
assert!(
new_cap <= MAX_CAPACITY,
"buffer capacity {new_cap} exceeds maximum {MAX_CAPACITY}"
);
let old_layout = shared_vec_layout(capacity).unwrap();
let new_layout = shared_vec_layout(new_cap).unwrap();
unsafe {
let ptr = self.0.as_ptr();
let offset = (*ptr).offset as usize;
if offset != METADATA_SIZE {
if len != 0 {
let data = ptr.cast::<u8>();
ptr::copy(data.add(offset), data.add(METADATA_SIZE), len);
}
(*ptr).offset = METADATA_SIZE_U32;
}
let new_ptr = alloc::realloc(ptr.cast(), old_layout, new_layout.size());
if new_ptr.is_null() {
alloc::handle_alloc_error(new_layout);
}
#[allow(clippy::cast_ptr_alignment)]
let inner = new_ptr.cast::<Header>();
(*inner).capacity = (new_layout.size() - METADATA_SIZE) as u32;
self.0 = NonNull::new_unchecked(inner);
#[cfg(feature = "tracker")]
{
let tracker = (*inner).tracker;
if !tracker.is_null() {
(*tracker).resize(old_layout.size(), new_layout.size());
}
}
}
}
#[inline]
pub(crate) unsafe fn set_len(&mut self, len: usize) {
let capacity = self.capacity();
assert!(len <= capacity);
(*self.as_inner()).len = len as u32;
}
pub(crate) unsafe fn set_start(&mut self, start: usize) {
if start != 0 {
let inner = self.as_inner();
assert!(
start <= (*inner).len as usize,
"cannot advance past the end of the buffer, cnt:{start} len:{}",
(*inner).len,
);
let start = start as u32;
(*inner).offset += start;
(*inner).len -= start;
}
}
}
unsafe impl Send for StorageVec {}
unsafe impl Sync for StorageVec {}
impl Drop for StorageVec {
fn drop(&mut self) {
release_shared_vec(self.0.as_ptr());
}
}
impl Header {
pub(crate) fn create(cap: usize, src: &[u8]) -> NonNull<Header> {
assert!(
cap >= src.len(),
"Header capacity {cap} is smaller than data length {}",
src.len()
);
let ptr = Self::alloc_with_capacity(cap, src.len() as u32);
unsafe {
let dst = ptr.add(METADATA_SIZE);
let sl = slice::from_raw_parts_mut(dst, src.len());
sl.copy_from_slice(src);
#[allow(clippy::cast_ptr_alignment)]
NonNull::new_unchecked(ptr.cast::<Header>())
}
}
fn alloc_with_capacity(cap: usize, len: u32) -> *mut u8 {
assert!(
cap <= MAX_CAPACITY,
"buffer capacity {cap} exceeds maximum {MAX_CAPACITY}"
);
let layout = shared_vec_layout(cap).unwrap();
unsafe {
let ptr = alloc::alloc(layout);
if ptr.is_null() {
alloc::handle_alloc_error(layout);
}
let capacity = (layout.size() - METADATA_SIZE) as u32;
#[allow(clippy::cast_ptr_alignment)]
ptr::write(
ptr.cast::<Header>(),
Header {
len,
capacity,
offset: METADATA_SIZE_U32,
ref_count: AtomicU32::new(1),
#[cfg(feature = "tracker")]
tracker: ptr::null(),
#[cfg(all(feature = "tracker", target_pointer_width = "64"))]
_pad: 0,
},
);
ptr
}
}
unsafe fn is_unique(ptr: *const Header) -> bool {
unsafe { (*ptr).ref_count.load(Acquire) == 1 }
}
#[cfg(feature = "tracker")]
pub(crate) unsafe fn tracker(ptr: *const Header) -> *const TrackerInner {
unsafe { ptr::addr_of!((*ptr).tracker).read() }
}
#[cfg(feature = "tracker")]
pub(crate) unsafe fn attach(ptr: *mut Header, tracker: *const TrackerInner) {
if !tracker.is_null() {
unsafe {
debug_assert!((*ptr).tracker.is_null());
(*tracker).charge((*ptr).capacity as usize + METADATA_SIZE);
(*ptr).tracker = tracker;
}
}
}
pub(crate) unsafe fn capacity(ptr: *const Header) -> usize {
ptr::addr_of!((*ptr).capacity).read() as usize
}
}
pub(crate) fn release_shared_vec(ptr: *mut Header) {
unsafe {
if (*ptr).ref_count.fetch_sub(1, Release) != 1 {
return;
}
atomic::fence(Acquire);
let capacity = (*ptr).capacity;
#[cfg(feature = "tracker")]
{
let tracker = ptr::replace(ptr::addr_of_mut!((*ptr).tracker), ptr::null());
if !tracker.is_null() {
TrackerInner::uncharge(tracker, capacity as usize + METADATA_SIZE);
}
}
let size = BytePageSize::from_alloc_size(capacity as usize + METADATA_SIZE);
if size != BytePageSize::Unset && cache::put_page(ptr, size) {
return;
}
ptr::drop_in_place(ptr);
let layout = shared_vec_layout(capacity as usize).unwrap();
alloc::dealloc(ptr.cast(), layout);
}
}
fn grown_capacity(len: usize, required: usize) -> usize {
cmp::max(required, cmp::min(len.saturating_mul(2), MAX_CAPACITY))
}
const fn shared_vec_layout(cap: usize) -> Result<Layout, LayoutError> {
let s_layout = match Layout::from_size_align(cap, Layout::new::<u8>().align()) {
Ok(l) => l,
Err(e) => return Err(e),
};
match Layout::new::<Header>().pad_to_align().extend(s_layout) {
Ok((l, _)) => Ok(l),
Err(err) => Err(err),
}
}
#[inline(never)]
#[cold]
pub(crate) fn abort() -> ! {
std::process::abort()
}
#[cfg(test)]
#[allow(clippy::assert_is_empty)]
mod tests {
use super::*;
use crate::*;
#[test]
fn page_allocation_is_category_size() {
for (size, alloc) in [
(BytePageSize::Size4, 4 * 1024),
(BytePageSize::Size16, 16 * 1024),
(BytePageSize::Size64, 64 * 1024),
(BytePageSize::Size128, 128 * 1024),
(BytePageSize::Size256, 256 * 1024),
] {
let st = StorageVec::sized(size);
assert_eq!(st.capacity(), size.capacity());
let layout = shared_vec_layout(st.capacity()).unwrap();
assert_eq!(layout.size(), alloc, "{size:?}");
}
}
#[test]
fn is_unique_synchronizes_with_release() {
let mut st = StorageVec::with_capacity(64);
for _ in 0..=INLINE_CAP {
st.put_u8(1);
}
let other = st.shallow_freeze();
assert!(!other.is_inline());
let handle = std::thread::spawn(move || {
let val = other.as_ref()[0];
drop(other);
val
});
while !st.is_unique() {
std::thread::yield_now();
}
st.as_mut()[0] = 2;
assert_eq!(handle.join().unwrap(), 1);
}
fn pattern(len: usize) -> Vec<u8> {
(0..len).map(|i| i as u8).collect()
}
#[test]
fn reserve_grows_unique_buffer() {
let data = pattern(100);
let mut st = StorageVec::from_slice(100, &data);
st.reserve(1000);
assert_eq!(st.as_ref(), &data[..]);
assert!(st.capacity() >= 1100);
assert_eq!(st.remaining(), st.capacity() - st.len());
let mut st = StorageVec::from_slice(100, &data);
unsafe { st.set_start(30) };
st.reserve(1000);
assert_eq!(st.as_ref(), &data[30..]);
assert!(st.capacity() >= 1070);
assert_eq!(st.remaining(), st.capacity() - st.len());
for i in 0..1000 {
st.put_u8(i as u8);
}
assert_eq!(&st.as_ref()[..70], &data[30..]);
assert_eq!(st.len(), 1070);
let mut st = StorageVec::from_slice(100, &data);
unsafe { st.set_start(100) };
st.reserve(1000);
assert!(st.as_ref().is_empty());
assert_eq!(st.remaining(), st.capacity());
}
#[test]
fn reserve_keeps_shared_views() {
let data = pattern(100);
let mut st = StorageVec::from_slice(100, &data);
let view = st.shallow_freeze();
assert!(!view.is_inline());
st.reserve(1000);
st.as_mut()[0] = 0xff;
assert_eq!(view.as_ref(), &data[..]);
assert_eq!(&st.as_ref()[1..], &data[1..]);
}
#[test]
fn reserve_pooled_page_grows_to_next_page_size() {
cache::reset();
let mut st = StorageVec::sized(BytePageSize::Size8);
let page = st.0;
let data = pattern(st.capacity());
for b in &data {
st.put_u8(*b);
}
st.reserve(1);
assert_eq!(st.page_size(), BytePageSize::Size16);
assert_eq!(st.capacity(), BytePageSize::Size16.capacity());
assert_ne!(st.0, page);
assert_eq!(st.as_ref(), &data[..]);
assert_eq!(cache::cached_pages(BytePageSize::Size8), 1);
let st2 = StorageVec::sized(BytePageSize::Size8);
assert_eq!(st2.0, page);
drop(st);
assert_eq!(cache::cached_pages(BytePageSize::Size16), 1);
}
#[test]
fn reserve_pooled_page_skips_page_sizes() {
cache::reset();
let data = pattern(100);
let mut st = StorageVec::sized(BytePageSize::Size4);
assert_eq!(st.put_slice_partial(&data), data.len());
st.reserve(40 * 1024);
assert_eq!(st.page_size(), BytePageSize::Size48);
assert_eq!(st.as_ref(), &data[..]);
let mut st = StorageVec::sized(BytePageSize::Size32);
let full = pattern(st.capacity());
assert_eq!(st.put_slice_partial(&full), full.len());
unsafe { st.set_start(full.len() - 100) };
let view = st.shallow_freeze();
assert_eq!(st.remaining(), 0);
st.reserve(10);
assert_eq!(st.page_size(), BytePageSize::Size32);
assert_eq!(st.as_ref(), &full[full.len() - 100..]);
assert_eq!(view.as_ref(), &full[full.len() - 100..]);
}
#[test]
fn reserve_shared_pooled_page() {
cache::reset();
let mut st = StorageVec::sized(BytePageSize::Size4);
let data = pattern(st.capacity());
assert_eq!(st.put_slice_partial(&data), data.len());
let view = st.shallow_freeze();
st.reserve(1);
assert_eq!(st.page_size(), BytePageSize::Size8);
st.as_mut()[0] = 0xff;
assert_eq!(view.as_ref(), &data[..]);
assert_eq!(&st.as_ref()[1..], &data[1..]);
assert_eq!(cache::cached_pages(BytePageSize::Size4), 0);
drop(view);
assert_eq!(cache::cached_pages(BytePageSize::Size4), 1);
}
#[test]
fn reserve_pooled_page_above_largest_page_size() {
cache::reset();
let data = pattern(1000);
let mut st = StorageVec::sized(BytePageSize::Size256);
assert_eq!(st.put_slice_partial(&data), data.len());
st.reserve(BytePageSize::Size256.capacity());
assert_eq!(st.page_size(), BytePageSize::Unset);
assert_eq!(st.capacity(), 1000 + BytePageSize::Size256.capacity());
assert_eq!(st.as_ref(), &data[..]);
assert_eq!(cache::cached_pages(BytePageSize::Size256), 1);
drop(st);
for size in crate::size::PAGE_SIZES {
assert_eq!(
cache::cached_pages(size),
usize::from(size == BytePageSize::Size256)
);
}
}
#[test]
fn reserve_exact_pooled_page() {
cache::reset();
let data = pattern(100);
let mut st = StorageVec::sized(BytePageSize::Size4);
assert_eq!(st.put_slice_partial(&data), data.len());
st.reserve_exact(5000);
assert_eq!(st.capacity(), 5100);
assert_eq!(st.page_size(), BytePageSize::Unset);
assert_eq!(st.as_ref(), &data[..]);
assert_eq!(cache::cached_pages(BytePageSize::Size4), 0);
let mut st = StorageVec::sized(BytePageSize::Size4);
assert_eq!(st.put_slice_partial(&data), data.len());
let view = st.shallow_freeze();
st.reserve_exact(5000);
assert_eq!(st.capacity(), 5100);
assert_eq!(st.page_size(), BytePageSize::Unset);
assert_eq!(st.as_ref(), &data[..]);
assert_eq!(view.as_ref(), &data[..]);
drop(view);
assert_eq!(cache::cached_pages(BytePageSize::Size4), 1);
let mut st = StorageVec::sized(BytePageSize::Size4);
assert_eq!(st.put_slice_partial(&data), data.len());
st.reserve_exact(BytePageSize::Size8.capacity() - 100);
assert_eq!(st.page_size(), BytePageSize::Size8);
assert_eq!(st.as_ref(), &data[..]);
}
#[test]
fn reserve_unsized_buffer_keeps_page_size() {
cache::reset();
let mut st = StorageVec::with_capacity(100);
assert_eq!(st.put_slice_partial(&pattern(100)), 100);
st.reserve_exact(4000);
assert_eq!(st.page_size(), BytePageSize::Unset);
assert_eq!(st.capacity(), 4100);
let view = st.shallow_freeze();
st.reserve(st.remaining() + 1);
assert_eq!(st.page_size(), BytePageSize::Unset);
drop(view);
let st = StorageVec::with_capacity(BytePageSize::Size4.capacity());
assert_eq!(st.capacity(), BytePageSize::Size4.capacity());
assert_eq!(st.page_size(), BytePageSize::Size4);
drop(st);
assert_eq!(cache::cached_pages(BytePageSize::Size4), 1);
let st = StorageVec::from_slice(BytePageSize::Size8.capacity(), b"hello");
assert_eq!(st.page_size(), BytePageSize::Size8);
drop(st);
assert_eq!(cache::cached_pages(BytePageSize::Size8), 1);
let mut st = StorageVec::sized(BytePageSize::Size8);
assert_eq!(st.len(), 0);
assert_eq!(st.remaining(), BytePageSize::Size8.capacity());
assert_eq!(st.put_slice_partial(b"world"), 5);
assert_eq!(st.as_ref(), b"world");
}
#[test]
fn remaining_is_derived() {
let mut st = StorageVec::with_capacity(100);
let cap = st.capacity();
assert_eq!(st.remaining(), cap);
assert!(!st.is_full());
st.put_u8(1);
assert_eq!(st.put_slice_partial(&pattern(50)), 50);
assert_eq!(st.remaining(), cap - 51);
unsafe { st.set_start(10) };
assert_eq!(st.len(), 41);
assert_eq!(st.capacity(), cap - 10);
assert_eq!(st.remaining(), cap - 51);
st.truncate(20);
assert_eq!(st.remaining(), cap - 30);
st.truncate(0);
assert_eq!(st.capacity(), cap);
assert_eq!(st.remaining(), cap);
assert_eq!(st.put_slice_partial(&pattern(cap + 10)), cap);
assert_eq!(st.remaining(), 0);
assert!(st.is_full());
let mut b = BytesMut::with_capacity(300);
b.extend_from_slice(&pattern(200));
let cap = b.capacity();
b.advance(50);
let ptr = b.as_ptr();
let mut b = BytesMut::from(b.freeze());
assert_eq!(b.as_ptr(), ptr);
assert_eq!(b.len(), 150);
assert_eq!(BufMut::remaining_mut(&b), cap - 200);
b.extend_from_slice(&pattern(cap - 200));
assert_eq!(BufMut::remaining_mut(&b), 0);
}
#[test]
fn truncate_reclaims_unique_buffer() {
let mut b = BytesMut::with_capacity(128);
let cap = b.capacity();
b.extend_from_slice(&[1; 64]);
b.advance(0);
b.advance(32);
assert_eq!(b.capacity(), cap - 32);
b.truncate(0);
assert_eq!(b.capacity(), cap);
b.extend_from_slice(&[1; 64]);
b.advance(32);
let other = b.split_to(30);
b.truncate(0);
assert_eq!(b.capacity(), cap - 62);
drop(other);
}
#[test]
fn resize_shrinks() {
let mut b = BytesMut::copy_from_slice(b"hello world");
b.resize(5, 0);
assert_eq!(&b[..], b"hello");
b.resize(7, b'!');
assert_eq!(&b[..], b"hello!!");
}
#[test]
fn reserve_reclaims_front_space() {
let mut b = BytesMut::with_capacity(128);
let cap = b.capacity();
b.extend_from_slice(&[1; 40]);
b.extend_from_slice(&[2; 10]);
b.advance(40);
let spare = b.capacity() - b.len();
b.reserve(spare + 1);
assert_eq!(&b[..], &[2; 10]);
assert_eq!(b.capacity(), cap);
assert!(b.is_unique());
}
#[test]
fn reserve_exact() {
let mut b = BytesMut::with_capacity(64);
b.extend_from_slice(&[1; 64]);
let cap = b.capacity();
b.reserve_exact(0);
assert_eq!(b.capacity(), cap);
b.reserve_exact(1000);
assert_eq!(&b[..], &[1; 64]);
assert!(b.capacity() >= 1064);
assert!(b.capacity() < 1064 + 2 * METADATA_SIZE);
let other = b.split_to(32);
b.reserve_exact(2000);
assert_eq!(&b[..], &[1; 32]);
assert!(b.capacity() >= 2032);
assert!(b.capacity() < 2032 + 2 * METADATA_SIZE);
assert!(b.is_unique());
assert_eq!(&other[..], &[1; 32]);
}
}
#[cfg(all(test, feature = "tracker"))]
mod tracker_tests {
use super::*;
use crate::*;
const PAGE4: usize = 4 * 1024;
const PAGE8: usize = 8 * 1024;
fn snap(tracker: &BytesTracker) -> (usize, usize) {
let s = tracker.snapshot();
(s.live_bytes, s.live_allocations)
}
fn alloc_size(b: &BytesMut) -> usize {
unsafe { Header::capacity(b.storage.0.as_ptr()) + METADATA_SIZE }
}
fn is_tracked(b: &BytesMut) -> bool {
unsafe { !Header::tracker(b.storage.0.as_ptr()).is_null() }
}
#[test]
fn page_charged_until_released() {
cache::reset();
let tracker = BytesTracker::new();
let b = BytesMut::with_page_size_in(BytePageSize::Size4, &tracker);
assert_eq!(b.page_size(), BytePageSize::Size4);
assert_eq!(snap(&tracker), (PAGE4, 1));
let ptr = b.as_ptr();
drop(b);
assert_eq!(snap(&tracker), (0, 0));
assert_eq!(cache::cached_pages(BytePageSize::Size4), 1);
let b = BytesMut::with_page_size(BytePageSize::Size4);
assert_eq!(b.as_ptr(), ptr);
assert!(!is_tracked(&b));
drop(b);
let b = BytesMut::with_page_size_in(BytePageSize::Unset, &tracker);
assert_eq!(b.page_size(), BytePageSize::Size64);
assert_eq!(snap(&tracker), (64 * 1024, 1));
let b2 = BytesMut::with_page_size_in(BytePageSize::Size4, &tracker);
assert_eq!(b2.as_ptr(), ptr);
assert_eq!(snap(&tracker), (64 * 1024 + PAGE4, 2));
drop((b, b2));
assert_eq!(snap(&tracker), (0, 0));
cache::reset();
}
#[test]
fn handles_share_the_charge() {
cache::reset();
let tracker = BytesTracker::new();
let mut b = BytesMut::with_page_size_in(BytePageSize::Size4, &tracker);
b.extend_from_slice(&[1; 200]);
let head = b.split_to(100);
let view = b.split_to(50);
assert_eq!(snap(&tracker), (PAGE4, 1));
drop(b);
drop(view);
assert_eq!(snap(&tracker), (PAGE4, 1));
let b = BytesMut::from(head);
assert!(is_tracked(&b));
assert_eq!(snap(&tracker), (PAGE4, 1));
drop(b);
assert_eq!(snap(&tracker), (0, 0));
cache::reset();
}
#[test]
fn shared_copy_from_bytes_is_untracked() {
cache::reset();
let tracker = BytesTracker::new();
let mut b = BytesMut::with_page_size_in(BytePageSize::Size4, &tracker);
b.extend_from_slice(&[1; 200]);
let bytes = b.freeze();
let other = bytes.clone();
let b = BytesMut::from(bytes);
assert!(!is_tracked(&b));
assert_eq!(snap(&tracker), (PAGE4, 1));
drop(other);
assert_eq!(snap(&tracker), (0, 0));
drop(b);
cache::reset();
}
#[test]
fn realloc_resizes_charge() {
cache::reset();
let tracker = BytesTracker::new();
let mut b = BytesMut::with_page_size_in(BytePageSize::Size4, &tracker);
b.extend_from_slice(&[1; 100]);
b.reserve_exact(10_000);
assert!(alloc_size(&b) > PAGE4);
assert_eq!(snap(&tracker), (alloc_size(&b), 1));
b.extend_from_slice(&[2; 10_000]);
b.reserve_exact(50_000);
assert_eq!(snap(&tracker), (alloc_size(&b), 1));
assert_eq!(&b[..100], &[1; 100]);
drop(b);
assert_eq!(snap(&tracker), (0, 0));
}
#[test]
fn move_to_page_carries_tracker() {
cache::reset();
let tracker = BytesTracker::new();
let mut b = BytesMut::with_page_size_in(BytePageSize::Size4, &tracker);
b.extend_from_slice(&[1; 5000]);
assert_eq!(b.page_size(), BytePageSize::Size8);
assert_eq!(snap(&tracker), (PAGE8, 1));
assert_eq!(cache::cached_pages(BytePageSize::Size4), 1);
let head = b.split_to(1000);
let ptr = b.as_ptr();
b.extend_from_slice(&[2; 4000]);
assert_eq!(b.page_size(), BytePageSize::Size8);
assert_ne!(b.as_ptr(), ptr);
assert_eq!(snap(&tracker), (2 * PAGE8, 2));
drop(head);
assert_eq!(snap(&tracker), (PAGE8, 1));
drop(b);
assert_eq!(snap(&tracker), (0, 0));
cache::reset();
}
#[test]
fn new_allocation_carries_tracker() {
cache::reset();
let tracker = BytesTracker::new();
let mut b = BytesMut::with_page_size_in(BytePageSize::Size4, &tracker);
b.extend_from_slice(&[1; 200]);
let head = b.split_to(100);
b.reserve_exact(10_000);
assert_eq!(b.page_size(), BytePageSize::Unset);
assert_eq!(alloc_size(&b), 10_100 + METADATA_SIZE);
assert_eq!(snap(&tracker), (PAGE4 + alloc_size(&b), 2));
drop(head);
assert_eq!(snap(&tracker), (alloc_size(&b), 1));
drop(b);
let mut b = BytesMut::with_page_size_in(BytePageSize::Size4, &tracker);
b.extend_from_slice(&[1; 200]);
let head = b.split_to(100);
b.reserve(1024 * 1024);
assert_eq!(b.page_size(), BytePageSize::Unset);
assert_eq!(snap(&tracker), (PAGE4 + alloc_size(&b), 2));
drop(head);
assert_eq!(snap(&tracker), (alloc_size(&b), 1));
drop(b);
assert_eq!(snap(&tracker), (0, 0));
cache::reset();
}
#[test]
fn trimdown_drops_tracker() {
cache::reset();
let tracker = BytesTracker::new();
let mut b = BytesMut::with_page_size_in(BytePageSize::Size4, &tracker);
b.extend_from_slice(&[1; 400]);
let mut head = b.split_to(128);
let mut tail = b.freeze();
head.trimdown();
assert_eq!(snap(&tracker), (4096, 1));
tail.trimdown();
assert_eq!(snap(&tracker), (0, 0));
let mut b = BytesMut::with_page_size_in(BytePageSize::Size4, &tracker);
b.extend_from_slice(&[1; 10]);
let mut bytes = b.freeze();
bytes.trimdown();
assert_eq!(snap(&tracker), (0, 0));
drop((head, tail, bytes));
cache::reset();
}
#[test]
fn untracked_copies() {
let tracker = BytesTracker::new();
let mut b = BytesMut::with_page_size_in(BytePageSize::Size4, &tracker);
b.extend_from_slice(&[1; 200]);
let copies = (b.clone(), BytesMut::copy_from_slice(&b[..]));
assert!(!is_tracked(&copies.0));
assert!(!is_tracked(&copies.1));
drop(b);
assert_eq!(snap(&tracker), (0, 0));
}
#[test]
fn tracker_outlives_handles() {
let tracker = BytesTracker::new();
let mut b = BytesMut::with_page_size_in(BytePageSize::Size4, &tracker);
drop(tracker);
b.extend_from_slice(&[1; 5000]);
let head = b.split_to(10);
drop(b);
drop(head);
}
#[test]
fn release_on_other_thread() {
let tracker = BytesTracker::new();
let mut b = BytesMut::with_page_size_in(BytePageSize::Size4, &tracker);
b.extend_from_slice(&[1; 200]);
let head = b.split_to(100);
let t1 = std::thread::spawn(move || drop(head));
let t2 = std::thread::spawn(move || drop(b));
t1.join().unwrap();
t2.join().unwrap();
assert_eq!(snap(&tracker), (0, 0));
}
#[test]
fn set_tracker_adopts_allocation() {
let tracker = BytesTracker::new();
let mut b = BytesMut::with_capacity(100);
b.extend_from_slice(&[1; 50]);
assert!(b.tracker().is_none());
b.set_tracker(&tracker);
assert!(b.tracker().unwrap().ptr_eq(&tracker));
assert_eq!(snap(&tracker), (alloc_size(&b), 1));
b.extend_from_slice(&[1; 1000]);
assert_eq!(snap(&tracker), (alloc_size(&b), 1));
let frozen = b.freeze();
assert_eq!(snap(&tracker).1, 1);
drop(frozen);
assert_eq!(snap(&tracker), (0, 0));
}
#[test]
fn set_tracker_reattributes() {
let tracker = BytesTracker::new();
let other = BytesTracker::new();
let mut b = BytesMut::with_capacity(200);
b.extend_from_slice(&[1; 150]);
let size = alloc_size(&b);
let head = b.split_to(64);
b.set_tracker(&tracker);
assert_eq!(snap(&tracker), (size, 1));
b.set_tracker(&tracker);
b.set_tracker(&tracker.clone());
assert_eq!(snap(&tracker), (size, 1));
b.set_tracker(&other);
assert!(b.tracker().unwrap().ptr_eq(&other));
assert_eq!(snap(&tracker), (0, 0));
assert_eq!(snap(&other), (size, 1));
drop(head);
b.set_tracker(&tracker);
assert_eq!(snap(&tracker), (size, 1));
assert_eq!(snap(&other), (0, 0));
drop(b);
assert_eq!(snap(&tracker), (0, 0));
}
#[test]
fn set_tracker_releases_last_tracker_handle() {
let mut b = BytesMut::with_capacity(200);
b.set_tracker(&BytesTracker::new());
let other = BytesTracker::new();
b.set_tracker(&other);
drop(other);
let tracker = b.tracker().unwrap().clone();
assert_eq!(snap(&tracker), (alloc_size(&b), 1));
drop(b);
assert_eq!(snap(&tracker), (0, 0));
}
#[test]
fn set_tracker_while_shared_across_threads() {
let tracker = BytesTracker::new();
let other = BytesTracker::new();
let mut b = BytesMut::with_capacity(512);
b.extend_from_slice(&[1; 400]);
b.set_tracker(&tracker);
let size = alloc_size(&b);
let mut head = b.split_to(128);
let handle = std::thread::spawn(move || {
head.trimdown();
head
});
b.set_tracker(&other);
let head = handle.join().unwrap();
assert_eq!(head.len(), 128);
assert_eq!(snap(&tracker).1, 0);
assert_eq!(snap(&other), (size, 1));
drop(b);
assert_eq!(snap(&other), (0, 0));
}
#[test]
fn set_tracker_on_page() {
cache::reset();
let tracker = BytesTracker::new();
let mut b = BytesMut::with_page_size(BytePageSize::Size4);
b.set_tracker(&tracker);
assert_eq!(snap(&tracker), (PAGE4, 1));
drop(b);
assert_eq!(snap(&tracker), (0, 0));
assert_eq!(cache::cached_pages(BytePageSize::Size4), 1);
let b = BytesMut::with_page_size(BytePageSize::Size4);
assert!(b.tracker().is_none());
}
#[test]
fn concurrent_releases() {
const THREADS: usize = 4;
const BUFS: usize = if cfg!(miri) { 4 } else { 64 };
let tracker = BytesTracker::new();
let mut heads = Vec::new();
let mut tails = Vec::new();
for _ in 0..THREADS {
let (mut h, mut t) = (Vec::new(), Vec::new());
for _ in 0..BUFS {
let mut b = BytesMut::with_page_size_in(BytePageSize::Size4, &tracker);
b.extend_from_slice(&[1; 200]);
h.push(b.split_to(100));
t.push(b);
}
heads.push(h);
tails.push(t);
}
assert_eq!(snap(&tracker), (THREADS * BUFS * PAGE4, THREADS * BUFS));
let barrier = std::sync::Arc::new(std::sync::Barrier::new(THREADS * 2));
let mut threads = Vec::new();
for (h, t) in heads.into_iter().zip(tails) {
let b1 = barrier.clone();
let b2 = barrier.clone();
threads.push(std::thread::spawn(move || {
b1.wait();
drop(h);
}));
threads.push(std::thread::spawn(move || {
b2.wait();
drop(t);
}));
}
for t in threads {
t.join().unwrap();
}
assert_eq!(snap(&tracker), (0, 0));
}
#[test]
fn cache_eviction_does_not_uncharge() {
cache::reset();
let freed = BytesTracker::new();
let tracker = BytesTracker::new();
let pages = [
BytesMut::with_page_size_in(BytePageSize::Size4, &freed),
BytesMut::with_page_size_in(BytePageSize::Size4, &tracker),
BytesMut::with_page_size_in(BytePageSize::Size4, &tracker),
];
let live = BytesMut::with_page_size_in(BytePageSize::Size8, &tracker);
drop(pages);
assert_eq!(cache::cached_pages(BytePageSize::Size4), 3);
assert_eq!(snap(&tracker), (PAGE8, 1));
drop(freed);
crate::set_page_cache_size(BytePageSize::Size4, 1);
assert_eq!(cache::cached_pages(BytePageSize::Size4), 1);
cache::reset();
assert_eq!(snap(&tracker), (PAGE8, 1));
drop(live);
assert_eq!(snap(&tracker), (0, 0));
}
#[test]
fn cache_reuse_by_other_tracker() {
cache::reset();
let first = BytesTracker::new();
let second = BytesTracker::new();
let page = BytesMut::with_page_size_in(BytePageSize::Size4, &first);
let ptr = page.as_ptr();
drop(page);
assert_eq!(snap(&first), (0, 0));
drop(first);
let page = BytesMut::with_page_size_in(BytePageSize::Size4, &second);
assert_eq!(page.as_ptr(), ptr);
assert!(page.tracker().unwrap().ptr_eq(&second));
assert_eq!(snap(&second), (PAGE4, 1));
drop(page);
assert_eq!(snap(&second), (0, 0));
}
#[test]
fn borrowed_tracker_without_handles() {
let mut b = BytesMut::with_capacity(100);
b.set_tracker(&BytesTracker::new());
let tracker = b.tracker().unwrap().clone();
assert_eq!(snap(&tracker), (alloc_size(&b), 1));
drop(b);
assert_eq!(snap(&tracker), (0, 0));
}
}